From c28d9f4aab4d998e85cd36b7077ef4d8837285e0 Mon Sep 17 00:00:00 2001 From: abrichr Date: Tue, 3 Nov 2026 17:31:30 -0500 Subject: [PATCH 1/2] API span: read the checks again after step writes; keep an earlier halt's reason The checks before saving ran in prepare(), before the span's step writes went out in send(). The GUI checks its empty-result guard after those writes, right before the save. When the span has step writes, send() now reads the checks again after the last write and halts without sending the save if one no longer passes. The end-of-run read of a still-armed wire check replaced the last step's error and category. It now adds the late refusal to an existing error and keeps an existing failure_category. --- .../runtime/actuators/web_form.py | 70 +++++++++++--- .../flow/openadapt_flow/runtime/replayer.py | 9 +- .../flow/tests/test_api_step_page_script.py | 82 ++++++++++++++++ packages/flow/tests/test_create_form_wire.py | 95 +++++++++++++++++++ 4 files changed, 240 insertions(+), 16 deletions(-) diff --git a/packages/flow/openadapt_flow/runtime/actuators/web_form.py b/packages/flow/openadapt_flow/runtime/actuators/web_form.py index 03dfe9e1f3..272839e957 100644 --- a/packages/flow/openadapt_flow/runtime/actuators/web_form.py +++ b/packages/flow/openadapt_flow/runtime/actuators/web_form.py @@ -603,6 +603,11 @@ class PreparedSave: session_fields: frozenset[str] = frozenset() #: The span's ``step`` writes, sent in this order before the save. writes: tuple["PreparedWrite", ...] = () + #: The checks before saving, as reads (address, required text, label). + #: When the span has step writes, :meth:`WebFormActuator.send` reads them + #: again after the last write and right before the save, as the GUI + #: checks its guard after those steps ran. Reads only: not in the digest. + rechecks: tuple[tuple[str, tuple[str, ...], str], ...] = () def encoded(self) -> str: return up.urlencode(list(self.body)) @@ -848,26 +853,21 @@ def unavailable(reason: str) -> ApiActuationResult: ) if problem: return unavailable(problem) + rechecks = [] for check in spec.prechecks: query = [(k, _fill(v, params)) for k, v in check.query] qs = "&".join( up.quote(k) + ("=" + up.quote(v) if v else "") for k, v in query ) - resp = self.transport.get(check.path + (f"?{qs}" if qs else "")) - missing = [t for t in check.require_text if t not in resp.text] - self.log.append( - { - "stage": "precheck", - "path": check.path, - "status": resp.status, - "missing": missing, - } + read = ( + check.path + (f"?{qs}" if qs else ""), + tuple(check.require_text), + check.from_step or check.path, ) - if resp.status != 200 or missing: - return unavailable( - f"the check before saving ({check.from_step or check.path}) did not pass: " - f"missing {missing!r}" - ) + problem = self._precheck(read) + if problem: + return unavailable(problem) + rechecks.append(read) except KeyError as exc: return unavailable(f"the run has no value for {exc}") except Exception as exc: # noqa: BLE001 - only reads were sent @@ -881,7 +881,28 @@ def unavailable(reason: str) -> ApiActuationResult: f.name for f in spec.fields if f.source == "session" ), writes=tuple(writes), + rechecks=tuple(rechecks), + ) + + def _precheck(self, read: tuple[str, tuple[str, ...], str]) -> Optional[str]: + """Read one check before saving; why it fails, or None.""" + + url, require_text, label = read + resp = self.transport.get(url) + missing = [t for t in require_text if t not in resp.text] + self.log.append( + { + "stage": "precheck", + "path": up.urlsplit(url).path, + "status": resp.status, + "missing": missing, + } ) + if resp.status != 200 or missing: + return ( + f"the check before saving ({label}) did not pass: missing {missing!r}" + ) + return None def _run_page_script( self, @@ -1005,6 +1026,27 @@ def send(self, binding: Any, prepared: "PreparedSave") -> ApiActuationResult: http_status=wresp.status, request_summary=summary, ) + if prepared.writes: + # The step writes ran; the GUI checks its guard (for example the + # duplicate search's "No matches were found.") after them, right + # before the save. Read the checks again here, in that order. + sent = len(prepared.writes) + for read in prepared.rechecks: + try: + problem = self._precheck(read) + except Exception as exc: # noqa: BLE001 - a read; the save is not sent + problem = f"the check before saving ({read[2]}) failed ({type(exc).__name__})" + if problem: + return ApiActuationResult( + status=ActuationStatus.HALT, + substrate=self.substrate, + halt_kind=ApiHaltKind.RESPONSE_REJECTED, + reason=( + f"after the {sent} step write(s) were accepted, {problem}; " + "the save was not sent and nothing is retried" + ), + request_summary=summary, + ) try: resp = self.transport.post_form( prepared.path, prepared.encoded(), referer=prepared.referer diff --git a/packages/flow/openadapt_flow/runtime/replayer.py b/packages/flow/openadapt_flow/runtime/replayer.py index 828f296d7d..f79e998b4c 100644 --- a/packages/flow/openadapt_flow/runtime/replayer.py +++ b/packages/flow/openadapt_flow/runtime/replayer.py @@ -13369,8 +13369,13 @@ def _halt_on_late_form_submit(self, report: RunReport) -> None: last = report.results[-1] last.ok = False last.safety_halt = True - last.failure_category = "safety_halt" - last.error = f"after an earlier save, {late} — run aborted" + # Keep an earlier halt's reason and category (for example a reconcile + # or a refuted record check): add the late refusal, don't replace it. + last.failure_category = last.failure_category or "safety_halt" + if not last.error: + last.error = f"after an earlier save, {late} — run aborted" + elif late not in last.error: + last.error = f"{last.error}; also after an earlier save, {late}" report.success = False def _act_closing_surface( diff --git a/packages/flow/tests/test_api_step_page_script.py b/packages/flow/tests/test_api_step_page_script.py index 3170fc6173..f10a1ca485 100644 --- a/packages/flow/tests/test_api_step_page_script.py +++ b/packages/flow/tests/test_api_step_page_script.py @@ -627,3 +627,85 @@ def test_a_form_submit_in_another_frame_is_seen(app_page): result = _evaluate(app, page, submitter_requests=("GET /side.php",)) assert result.status == "unevaluable" and "form submit in" in result.reason assert app.writes == [] + + +class _DupSearch(dict): + """Pages, where the duplicate search finds the record once the page's + draft write has reached the server.""" + + def __init__(self, app: _App) -> None: + super().__init__(app.pages) + self.app = app + + def __contains__(self, path: object) -> bool: + return path == "/dup.php" or super().__contains__(path) + + def __getitem__(self, path: str) -> str: + if path != "/dup.php": + return super().__getitem__(path) + if "/draft.php" in self.app.writes: + return "

Marsh 1979-06-14

" + return "

No matches were found.

" + + +def test_a_step_write_that_changes_the_search_result_stops_the_save(app_page): + """The span sends a step write (the page's draft) after the duplicate + search was read and before the save. The GUI checks its empty-result + guard after that write, right before the save; so must the API step.""" + + from openadapt_flow.ir import ApiBinding, WebFormPrecheck, WebFormWrite + from openadapt_flow.runtime.actuators import ActuationStatus + + app, page = app_page + app.form( + script="document.forms[0].lname.addEventListener('change', e => fetch('/draft.php', {" + " method: 'POST', headers: {'Content-Type': 'application/x-www-form-urlencoded'}," + " body: 'lname=' + encodeURIComponent(e.target.value) }));" + ) + app.pages = _DupSearch(app) + html = app.pages["/new.php"].replace("{n}", "1") + names = ["csrf", "lname", "dob", "state"] + spec = WebFormSpec( + form_path="/new.php", + fields=[ + WebFormField(name="csrf", source="session"), + WebFormField(name="lname", source="param", param="lname"), + WebFormField(name="dob", source="param", param="dob"), + WebFormField(name="state", source="select_label", param="state_label"), + ], + controls=sorted(names), + client_scripts=page_scripts(html, "/save.php", names, None), + submitter="button[type=submit]", + prechecks=[ + WebFormPrecheck( + path="/dup.php", + query=[("lname", "{lname}")], + require_text=["No matches were found."], + from_step="s0", + ) + ], + writes=[ + WebFormWrite( + role="step", + method="POST", + path="/draft.php", + from_step="s1", + fields=[WebFormField(name="lname", source="param", param="lname")], + expected_status=[200], + ) + ], + replaces_steps=["s0", "s1", "s2"], + folded_checks_digest="sha256:" + "0" * 64, + ) + binding = ApiBinding( + kind="web_form", method="POST", url_template="/save.php", web_form=spec + ) + actuator = WebFormActuator(PlaywrightSessionTransport(page)) + out = actuator.actuate(binding, RUN) + # The draft went out, as under the GUI steps; the search then finds the + # record, so the save never leaves. + assert app.writes == ["/draft.php"], app.writes + assert out.status is ActuationStatus.HALT, out.reason + assert "after the 1 step write(s) were accepted" in out.reason + assert "No matches were found." in out.reason + assert [e["stage"] for e in actuator.log][-3:] == ["precheck", "write", "precheck"] diff --git a/packages/flow/tests/test_create_form_wire.py b/packages/flow/tests/test_create_form_wire.py index ef5cb23bec..b09c46bb35 100644 --- a/packages/flow/tests/test_create_form_wire.py +++ b/packages/flow/tests/test_create_form_wire.py @@ -1195,3 +1195,98 @@ def test_an_api_span_after_a_late_refusal_halts_before_sending( assert "after the previous save" in (halt.error or "") or ( "after an earlier save" in (halt.error or "") ), halt.error + + +class _LateSubmitThenRejectedActuator: + """An API step whose save the server rejects, while the live page sends + a stray submit that the earlier save's wire check, still armed, aborts.""" + + def __init__(self, page: Any) -> None: + self.page, self.calls = page, 0 + + def actuate(self, binding: Any, params: Any) -> Any: + from openadapt_flow.runtime.actuators import ( + ActuationStatus, + ApiActuationResult, + ApiHaltKind, + ) + + self.calls += 1 + # After the span's first read of the check: the stray submit goes out + # now, and the wire check aborts it. + self.page.evaluate("() => window.__sendLate()") + self.page.wait_for_timeout(500) + return ApiActuationResult( + status=ActuationStatus.HALT, + halt_kind=ApiHaltKind.RESPONSE_REJECTED, + http_status=500, + reason="POST /save.php -> 500 without the recorded success response", + ) + + +def test_a_late_refusal_adds_to_an_earlier_halt_instead_of_replacing_it( + server: _Server, tmp_path_factory: pytest.TempPathFactory +) -> None: + """The span halts for its own reason (its save was rejected), and a + stray submit is aborted during the span. The end-of-run read keeps the + span's message and category and adds the late refusal.""" + + from openadapt_flow.backends.playwright_backend import ( + PlaywrightBackend, + prepare_ws_guard, + ) + from openadapt_flow.mockmed.fault_server import serve as fault_serve + from openadapt_flow.runtime import Replayer + from openadapt_flow.runtime.effects import RestRecordVerifier + from tests.test_api_tier_web_form import _span_workflow + + bundle = _bundle(server, "fetch", tmp_path_factory.mktemp("rec")) + sor_url, _db, stop_sor = fault_serve() + with sync_api.sync_playwright() as playwright: + browser = playwright.chromium.launch(headless=True) + try: + page = browser.new_context( + service_workers="block", viewport={"width": 1280, "height": 800} + ).new_page() + prepare_ws_guard(page.context) + page.goto(f"{server.url}?v=fetchlate") + page.wait_for_selector("[name=field_summary]", state="visible") + backend = PlaywrightBackend(page) + saved = Replayer(backend).run( + Workflow.load(bundle), + params={ + "field_summary": "Scanner offline on floor 3", + "field_description": "Since this morning", + }, + bundle_dir=bundle, + run_dir=tmp_path_factory.mktemp("run"), + ) + assert backend.form_submit_guard_refusal() is None + actuator = _LateSubmitThenRejectedActuator(page) + span_bundle = tmp_path_factory.mktemp("span") + (span_bundle / "templates").mkdir() + report = Replayer( + backend, + web_form_actuator=actuator, + effect_verifier=RestRecordVerifier(sor_url.rstrip("/")), + poll_interval_s=0.01, + ).run( + _span_workflow(), + bundle_dir=span_bundle, + run_dir=tmp_path_factory.mktemp("run"), + ) + late = backend.form_submit_guard_refusal() + finally: + browser.close() + stop_sor() + assert saved.success, [(r.step_id, r.error) for r in saved.results] + assert actuator.calls == 1 and late is not None + assert not report.success + halt = report.results[-1] + assert halt.actuation == "api" and halt.delivery_attempted + # The span's own reason and category, then the late refusal. + print("CATEGORY", halt.failure_category, "ERROR", halt.error) + assert halt.failure_category not in (None, "safety_halt"), halt.failure_category + own, _, added = (halt.error or "").partition("; also after an earlier save, ") + assert "500" in own, halt.error + assert added == late, halt.error From af0043bf6bd67aea83e02606f68f881e044665ed Mon Sep 17 00:00:00 2001 From: abrichr Date: Tue, 3 Nov 2026 18:13:41 -0500 Subject: [PATCH 2/2] Test B: assert the span's own message, and pin category retention on a governed halt --- packages/flow/tests/test_create_form_wire.py | 36 ++++++++++++++++---- 1 file changed, 30 insertions(+), 6 deletions(-) diff --git a/packages/flow/tests/test_create_form_wire.py b/packages/flow/tests/test_create_form_wire.py index b09c46bb35..f0265811af 100644 --- a/packages/flow/tests/test_create_form_wire.py +++ b/packages/flow/tests/test_create_form_wire.py @@ -1235,6 +1235,7 @@ def test_a_late_refusal_adds_to_an_earlier_halt_instead_of_replacing_it( PlaywrightBackend, prepare_ws_guard, ) + from openadapt_flow.ir import RunReport, StepResult from openadapt_flow.mockmed.fault_server import serve as fault_serve from openadapt_flow.runtime import Replayer from openadapt_flow.runtime.effects import RestRecordVerifier @@ -1265,17 +1266,35 @@ def test_a_late_refusal_adds_to_an_earlier_halt_instead_of_replacing_it( actuator = _LateSubmitThenRejectedActuator(page) span_bundle = tmp_path_factory.mktemp("span") (span_bundle / "templates").mkdir() - report = Replayer( + replayer = Replayer( backend, web_form_actuator=actuator, effect_verifier=RestRecordVerifier(sor_url.rstrip("/")), poll_interval_s=0.01, - ).run( + ) + report = replayer.run( _span_workflow(), bundle_dir=span_bundle, run_dir=tmp_path_factory.mktemp("run"), ) late = backend.form_submit_guard_refusal() + # The same end-of-run read on a run whose last step halted with + # another category: the category is kept. + governed = RunReport( + workflow_name="span", + started_at="t", + success=False, + results=[ + StepResult( + step_id="step_001", + intent="save", + ok=False, + failure_category="governed_refusal", + error="the record check was refuted; reconcile", + ) + ], + ) + replayer._halt_on_late_form_submit(governed) finally: browser.close() stop_sor() @@ -1284,9 +1303,14 @@ def test_a_late_refusal_adds_to_an_earlier_halt_instead_of_replacing_it( assert not report.success halt = report.results[-1] assert halt.actuation == "api" and halt.delivery_attempted - # The span's own reason and category, then the late refusal. - print("CATEGORY", halt.failure_category, "ERROR", halt.error) - assert halt.failure_category not in (None, "safety_halt"), halt.failure_category + assert halt.safety_halt and halt.failure_category == "safety_halt" + # The span's own reason first, then the late refusal. own, _, added = (halt.error or "").partition("; also after an earlier save, ") - assert "500" in own, halt.error + assert "outside the binding's accepted contract" in own, halt.error assert added == late, halt.error + kept = governed.results[0] + assert not governed.success and not kept.ok and kept.safety_halt + assert kept.failure_category == "governed_refusal" + assert kept.error == ( + f"the record check was refuted; reconcile; also after an earlier save, {late}" + )