API span: re-read the checks after step writes; keep an earlier halt reason

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

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(

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 "<html><body><p>Marsh 1979-06-14</p></body></html>"
+        return "<html><body><p>No matches were found.</p></body></html>"
+
+
+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"]

packages/flow/tests/test_create_form_wire.py

@@ -1195,3 +1195,122 @@ 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.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
+    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()
+            replayer = Replayer(
+                backend,
+                web_form_actuator=actuator,
+                effect_verifier=RestRecordVerifier(sor_url.rstrip("/")),
+                poll_interval_s=0.01,
+            )
+            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()
+    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
+    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 "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}"
+    )