flow: pipeline an API save's read-back; keep self-repair probes on the GUI

CAPABILITIES.md

@@ -184,6 +184,7 @@ improvising or retrying when something is off.
 | Approve an API step's exact request under per-item approval: one digest binds the request bytes, the record and the record check, and a changed request is not sent | In review ([#1355](https://github.com/OpenAdaptAI/OpenAdapt/pull/1355)) | Synthetic | [API_TIER.md](packages/flow/docs/API_TIER.md#under-an-approval-mode) |
 | Run an API step in a durable run: checkpoint the span after the save is sent, never resend or redo a save that may have landed, and reconcile it by reading the record back | In review ([#1358](https://github.com/OpenAdaptAI/OpenAdapt/pull/1358)) | Synthetic | [API_TIER.md](packages/flow/docs/API_TIER.md#in-a-durable-run) |
 | Run an admitted API step under a governed run authorization and the Standard or Regulated profile: the authorization names the API step by digest, the person who admits it names the right-record parameter, and the outcome reads `VERIFIED` | In review ([#1361](https://github.com/OpenAdaptAI/OpenAdapt/pull/1361)) | Synthetic | [API_TIER.md](packages/flow/docs/API_TIER.md#under-governance) |
+| Defer an API save's read-back with pipelined read-back, as for a GUI save, and keep self-repair's probes and resumed run on the repaired GUI step | In review ([#1362](https://github.com/OpenAdaptAI/OpenAdapt/pull/1362)) | Synthetic | [API_TIER.md](packages/flow/docs/API_TIER.md#with-parallel-workers-pipelined-read-back-and-self-repair) |
 | Run a worklist on several workers at once, one signed-in session each, each item claimed once, with a stop on any worker stopping all | In review ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic patients on a local OpenEMR test copy | `--workers`, [THROUGHPUT.md](packages/flow/docs/THROUGHPUT.md) |
 | Read a save back while the next item runs (pipelined read-back, REST, FHIR, GraphQL or SQL record checks) | In review ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic | `--readback-window`, [THROUGHPUT.md](packages/flow/docs/THROUGHPUT.md) |
 | A faster speed profile that keeps every check (one insert per plain text field, shorter page-quiet wait) | In review ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic | `runtime.speed: fast`, [THROUGHPUT.md](packages/flow/docs/THROUGHPUT.md) |

llms-full.txt

@@ -540,6 +540,7 @@ improvising or retrying when something is off.
 | Approve an API step's exact request under per-item approval: one digest binds the request bytes, the record and the record check, and a changed request is not sent | In review ([#1355](https://github.com/OpenAdaptAI/OpenAdapt/pull/1355)) | Synthetic | [API_TIER.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/API_TIER.md#under-an-approval-mode) |
 | Run an API step in a durable run: checkpoint the span after the save is sent, never resend or redo a save that may have landed, and reconcile it by reading the record back | In review ([#1358](https://github.com/OpenAdaptAI/OpenAdapt/pull/1358)) | Synthetic | [API_TIER.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/API_TIER.md#in-a-durable-run) |
 | Run an admitted API step under a governed run authorization and the Standard or Regulated profile: the authorization names the API step by digest, the person who admits it names the right-record parameter, and the outcome reads `VERIFIED` | In review ([#1361](https://github.com/OpenAdaptAI/OpenAdapt/pull/1361)) | Synthetic | [API_TIER.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/API_TIER.md#under-governance) |
+| Defer an API save's read-back with pipelined read-back, as for a GUI save, and keep self-repair's probes and resumed run on the repaired GUI step | In review ([#1362](https://github.com/OpenAdaptAI/OpenAdapt/pull/1362)) | Synthetic | [API_TIER.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/API_TIER.md#with-parallel-workers-pipelined-read-back-and-self-repair) |
 | Run a worklist on several workers at once, one signed-in session each, each item claimed once, with a stop on any worker stopping all | In review ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic patients on a local OpenEMR test copy | `--workers`, [THROUGHPUT.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/THROUGHPUT.md) |
 | Read a save back while the next item runs (pipelined read-back, REST, FHIR, GraphQL or SQL record checks) | In review ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic | `--readback-window`, [THROUGHPUT.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/THROUGHPUT.md) |
 | A faster speed profile that keeps every check (one insert per plain text field, shorter page-quiet wait) | In review ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic | `runtime.speed: fast`, [THROUGHPUT.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/THROUGHPUT.md) |

packages/flow/docs/API_TIER.md

@@ -209,6 +209,17 @@ Not yet covered, so the GUI steps run with that reason:
 - **Managed (Cloud) dispatch.** Cloud's run authorization doesn't list API
   steps yet.
 
+## With parallel workers, pipelined read-back and self-repair
+
+- **Workers** (`--workers`): each item has its own run, authorization and
+  approval gate, so the API step runs in each, as in a single run.
+- **Pipelined read-back** (`--readback-window`): an API save's record check
+  is deferred like a GUI save's (same check, later timing). Durable runs
+  verify inline and say why ([THROUGHPUT.md](THROUGHPUT.md)).
+- **Self-repair**: the first run can use the API step. A repaired bundle's
+  probes and resumed run use the GUI steps, because the repair and the
+  person's approval are about the GUI step ([SELF_REPAIR.md](SELF_REPAIR.md)).
+
 ## When it doesn't run
 
 The API step runs only in a plain linear run. It doesn't run, and the GUI
@@ -223,7 +234,8 @@ steps run instead, when:
   save's record check changed since it was proposed or admitted;
 - the save checks the right record on screen and the API step names no
   right-record contract;
-- there's no live browser session (`--backend web` only).
+- there's no live browser session (`--backend web` only);
+- it is a self-repair probe or the resumed run of a repaired bundle.
 
 `Replayer(api_steps=False)` keeps a run on the GUI.
 

packages/flow/docs/SELF_REPAIR.md

@@ -248,7 +248,17 @@ original bundle:
 - a qualification case;
 - `--pin-digest`.
 
-In those cases the approved bundle needs its own admission first. Under the
+In those cases the approved bundle needs its own admission first: a
+repaired bundle has a new digest, and each of these admits only the bundle
+it names.
+
+The first run is the ordinary run, so an admitted API step
+([API_TIER.md](API_TIER.md)) can run there. The probes and the resumed run of
+a repaired bundle run the GUI steps only, and the run report says why: the
+repair, the fault probes and the person's approval are all about the GUI
+step, so an API step in its place would save without exercising the repair. The
+same holds for a standard change's run of the repaired bundle on the test
+record. Under the
 CLI, the Standard profile requires a signed qualification admission. So a CLI
 self-repair runs without a named profile or under Demo, and the record check
 still runs. The Python API can run the Standard profile contract directly, as

packages/flow/docs/THROUGHPUT.md

@@ -90,6 +90,7 @@ starts.
   doesn't use the screen (a REST, FHIR or database read). If the item runs
   any further step, the engine waits for the read-back first and halts on a
   failure, as it would inline.
+- An API step's save ([API_TIER.md](API_TIER.md)) is deferred the same way.
 - The item's `report.json` is rewritten with the verdict when the read-back
   finishes, and its outcome is classified again from the completed evidence.
 - If a read-back fails, the run stops on every worker. The report names the
@@ -98,6 +99,11 @@ starts.
 - It needs `items_independent: true`. Durable and idempotency-ledger runs
   verify inline, because their terminal state is written once; the Standard
   and Regulated profiles are durable, so for them the window has no effect.
+  The worker's item detail records why (`readback: inline (durable runs
+  verify inline)`). Deferring a durable save would mean committing its
+  checkpoint and the run's terminal state from the side thread, after the
+  next item has taken the browser, which the durable runtime doesn't
+  support. Workers still run in parallel under those profiles.
 
 ## One entry, faster
 

packages/flow/openadapt_flow/__main__.py

@@ -840,8 +840,12 @@ def _build_and_run_replayer(
     deployment_surface: Optional[str] = None,
     readback_pipeline=None,
     secret_env_overrides: Optional[Mapping[str, str]] = None,
+    api_steps_off_reason: Optional[str] = None,
 ):
-    """Build the shared Replayer configuration and execute one workflow."""
+    """Build the shared Replayer configuration and execute one workflow.
+
+    ``api_steps_off_reason`` turns API steps off for this run and says why.
+    """
     replayer = _configured_replayer(
         backend,
         workflow=workflow,
@@ -867,6 +871,9 @@ def _build_and_run_replayer(
     # and per-account secret sources. Both default to the serial behavior.
     replayer.readback_pipeline = readback_pipeline
     replayer.secret_env_overrides = dict(secret_env_overrides or {})
+    if api_steps_off_reason is not None:
+        replayer.api_steps = False
+        replayer.api_steps_off_reason = api_steps_off_reason
     return replayer.run(
         workflow,
         params=params,

packages/flow/openadapt_flow/cli_self_repair.py

@@ -203,10 +203,14 @@ def execute(
         run_dir: Path,
         *,
         approval_gate: Any = None,
+        gui_only: bool = False,
     ) -> Any:
         from urllib.parse import urlsplit
 
-        from openadapt_flow.repair.self_repair import SelfRepairRefused
+        from openadapt_flow.repair.self_repair import (
+            REPAIR_GUI_ONLY,
+            SelfRepairRefused,
+        )
         from openadapt_flow.run_gate import (
             build_runtime_authorization,
             evaluate_run_gate,
@@ -258,6 +262,11 @@ def execute(
             execution_entry_url=page_url,
             approval_gate=gate,
             deployment_surface=self.backend_cfg.deployment_surface,
+            # A probe or the resumed run of a repaired bundle runs the GUI
+            # step the person approved (repair.self_repair.keep_gui_for_repair).
+            api_steps_off_reason=(
+                None if original and not gui_only else REPAIR_GUI_ONLY
+            ),
         )
 
 

packages/flow/openadapt_flow/repair/self_repair.py

@@ -235,7 +235,9 @@ class SelfRepairTarget(Protocol):
     fault name opens a session that presents that fault (for the fault
     probes). ``execute`` admits ``workflow`` through the run gate and runs it
     with a fresh authorization; it must raise :class:`SelfRepairRefused` when
-    the gate refuses.
+    the gate refuses. ``gui_only`` (and any ``approval_gate``, which only
+    probes and resumed runs pass) runs the GUI steps only
+    (:func:`keep_gui_for_repair`).
     """
 
     def open(self, fault: Optional[str] = None) -> AppSession: ...
@@ -248,6 +250,7 @@ def execute(
         run_dir: Path,
         *,
         approval_gate: Optional[ApprovalGate] = None,
+        gui_only: bool = False,
     ) -> RunReport: ...
 
 
@@ -307,6 +310,7 @@ def execute(
         run_dir: Path,
         *,
         approval_gate: Optional[ApprovalGate] = None,
+        gui_only: bool = False,
     ) -> RunReport:
         from openadapt_flow.deployment import DeploymentConfig, PolicySection
         from openadapt_flow.execution_profiles import (
@@ -350,6 +354,8 @@ def execute(
             approval_gate=approval_gate,
             **self.replayer_kwargs,
         )
+        if approval_gate is not None or gui_only:
+            keep_gui_for_repair(replayer)
         return replayer.run(
             workflow.model_copy(deep=True),
             params=self.params,
@@ -362,6 +368,25 @@ def execute(
         )
 
 
+#: Why a probe or the resumed run of a repaired bundle doesn't use API steps.
+REPAIR_GUI_ONLY = (
+    "self-repair's probes and resumed run exercise the repaired GUI step, and "
+    "the person approved that step, so its API step doesn't run"
+)
+
+
+def keep_gui_for_repair(replayer: Any) -> None:
+    """Run a repaired bundle's probe or resumed run on the GUI steps only.
+
+    The repair, the fault probes and the person's approval are all about the
+    GUI step. An API step in its place would save without exercising the
+    repair, so the canary would prove nothing about it.
+    """
+
+    replayer.api_steps = False
+    replayer.api_steps_off_reason = REPAIR_GUI_ONLY
+
+
 # -- the probe gate --------------------------------------------------------------
 
 

packages/flow/openadapt_flow/repair/standard_change.py

@@ -595,9 +595,17 @@ def _verified(report: RunReport) -> bool:
 
 
 def run_preflight(
-    test_record: TestRecord, bundle_dir: Path | str, run_dir: Path | str
+    test_record: TestRecord,
+    bundle_dir: Path | str,
+    run_dir: Path | str,
+    *,
+    gui_only: bool = False,
 ) -> PreflightResult:
-    """Run ``bundle_dir`` once on the test record, with the full read-back."""
+    """Run ``bundle_dir`` once on the test record, with the full read-back.
+
+    ``gui_only`` runs the GUI steps only: a repaired bundle's test-record run
+    must exercise the repaired GUI step, not an API step in its place.
+    """
 
     from openadapt_flow.repair.self_repair import _close
 
@@ -606,7 +614,9 @@ def run_preflight(
     try:
         workflow = Workflow.load(bundle_dir)
         session = test_record.target.open()
-        report = test_record.target.execute(workflow, bundle_dir, session, run_dir)
+        report = test_record.target.execute(
+            workflow, bundle_dir, session, run_dir, gui_only=gui_only
+        )
     except Exception as exc:  # noqa: BLE001 - a run that can't start fails
         return PreflightResult(
             test_record.record_id,
@@ -916,7 +926,7 @@ def to_person(why: str) -> None:
             to_person(str(exc))
             return None
         preflight = run_preflight(
-            self.test_record, patch.bundle_dir, work / "preflight"
+            self.test_record, patch.bundle_dir, work / "preflight", gui_only=True
         )
         self.last_preflight = preflight
         if not preflight.passed:

packages/flow/openadapt_flow/runtime/replayer.py

@@ -1044,6 +1044,8 @@ def __init__(
         # ``api_steps=False`` keeps every run on the GUI. ``web_form_actuator``
         # replaces the live-session actuator (tests, other transports).
         self.api_steps = api_steps
+        #: Why ``api_steps`` is off, when a caller says (for the report).
+        self.api_steps_off_reason: Optional[str] = None
         self.web_form_actuator = web_form_actuator
         self._web_form_span_active = False
         self._web_form_fallback_reason: Optional[str] = None
@@ -2431,7 +2433,7 @@ def _web_form_span_at(
                 continue
             ids = [s.id for s in workflow.steps[step_index : end_index + 1]]
             if not self.api_steps:
-                reason = "API steps are off for this run"
+                reason = self.api_steps_off_reason or "API steps are off for this run"
             elif getattr(self, "_surface_router", None) is not None:
                 reason = "a run across several windows or tabs uses the GUI steps"
             elif self._approval_gated() and not callable(
@@ -9106,6 +9108,16 @@ def _try_api_tier(
             current_step=overlay_current,
             total_steps=overlay_total,
         )
+        if self._may_pipeline_readback(effect_verifier):
+            # Pipelined read-back (runtime/readback_pipeline.py), as for a GUI
+            # save: the same record check on the side thread. The item's
+            # report stays unsuccessful until it settles with the verdict.
+            self._defer_readback(
+                step, before, result, effects=effects, verifier=effect_verifier
+            )
+            result.ok = True
+            result.error = None
+            return True
         try:
             error = self._verify_effects(
                 step,