diff --git a/CAPABILITIES.md b/CAPABILITIES.md index 41c2670cb..b29347101 100644 --- a/CAPABILITIES.md +++ b/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) | diff --git a/llms-full.txt b/llms-full.txt index 4c747535c..2fcd25df9 100644 --- a/llms-full.txt +++ b/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) | diff --git a/packages/flow/docs/API_TIER.md b/packages/flow/docs/API_TIER.md index 8753108fc..029781876 100644 --- a/packages/flow/docs/API_TIER.md +++ b/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. diff --git a/packages/flow/docs/SELF_REPAIR.md b/packages/flow/docs/SELF_REPAIR.md index aa7fd5138..fd5c9e0ba 100644 --- a/packages/flow/docs/SELF_REPAIR.md +++ b/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 diff --git a/packages/flow/docs/THROUGHPUT.md b/packages/flow/docs/THROUGHPUT.md index 46929238f..3852046db 100644 --- a/packages/flow/docs/THROUGHPUT.md +++ b/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 diff --git a/packages/flow/openadapt_flow/__main__.py b/packages/flow/openadapt_flow/__main__.py index 76d68058c..46023e918 100644 --- a/packages/flow/openadapt_flow/__main__.py +++ b/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, diff --git a/packages/flow/openadapt_flow/cli_self_repair.py b/packages/flow/openadapt_flow/cli_self_repair.py index eb0a5e0aa..7646ecf00 100644 --- a/packages/flow/openadapt_flow/cli_self_repair.py +++ b/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 + ), ) diff --git a/packages/flow/openadapt_flow/repair/self_repair.py b/packages/flow/openadapt_flow/repair/self_repair.py index 41e6485dc..abc2d329e 100644 --- a/packages/flow/openadapt_flow/repair/self_repair.py +++ b/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 -------------------------------------------------------------- diff --git a/packages/flow/openadapt_flow/repair/standard_change.py b/packages/flow/openadapt_flow/repair/standard_change.py index 10a1fb09a..f7cefd792 100644 --- a/packages/flow/openadapt_flow/repair/standard_change.py +++ b/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: diff --git a/packages/flow/openadapt_flow/runtime/replayer.py b/packages/flow/openadapt_flow/runtime/replayer.py index 5a3baebd0..079a128bb 100644 --- a/packages/flow/openadapt_flow/runtime/replayer.py +++ b/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, diff --git a/packages/flow/tests/test_api_step_readback_repair.py b/packages/flow/tests/test_api_step_readback_repair.py new file mode 100644 index 000000000..8ba9fc207 --- /dev/null +++ b/packages/flow/tests/test_api_step_readback_repair.py @@ -0,0 +1,163 @@ +"""The API step with pipelined read-back and with self-repair. + +- Pipelined read-back (docs/THROUGHPUT.md) defers an API save's record check + the same way it defers a GUI save's: same check, later timing, and the + item's report stays unsuccessful until it settles. A durable run verifies + inline and says why. +- Self-repair's probes and its resumed run of a repaired bundle run the GUI + steps: the repair and the person's approval are about the GUI step, so an + API step there would save without exercising the repair. + +The system of record is the MockMed fault server; the values are made up. +""" + +from __future__ import annotations + +import json + +from openadapt_flow.mockmed.fault_server import serve as fault_serve +from openadapt_flow.repair.self_repair import ( + REPAIR_GUI_ONLY, + ReplayTarget, + keep_gui_for_repair, +) +from openadapt_flow.runtime.actuators import ActuationStatus +from openadapt_flow.runtime.effects import RestRecordVerifier +from openadapt_flow.runtime.readback_pipeline import ReadbackPipeline +from openadapt_flow.runtime.replayer import Replayer +from tests.test_api_tier_web_form import ( + EnterWritesBackend, + ScriptedActuator, + _span_workflow, +) +from tests.test_execution_profiles import _sealed +from tests.test_replayer import FakeVision + + +def _run(tmp_path, *, pipeline, durable=False, verifier_url=None): + url, db, stop = fault_serve() + url = url.rstrip("/") + try: + workflow, bundle = _sealed(tmp_path, _span_workflow(), encrypted=False) + actuator = ScriptedActuator(url, ActuationStatus.ACTUATED) + report = Replayer( + EnterWritesBackend(url), + vision=FakeVision(), + effect_verifier=RestRecordVerifier(verifier_url or url), + web_form_actuator=actuator, + poll_interval_s=0.01, + readback_pipeline=pipeline, + durable=durable, + ).run(workflow, bundle_dir=bundle, run_dir=tmp_path / "run") + handoff = pipeline.take() + settled = handoff.settled.result(timeout=30) if handoff is not None else None + return report, settled, actuator, db.snapshot()["records"] + finally: + stop() + + +def test_an_api_save_reads_back_while_the_next_item_runs(tmp_path): + pipeline = ReadbackPipeline() + try: + report, settled, actuator, records = _run(tmp_path, pipeline=pipeline) + finally: + pipeline.close() + + assert actuator.calls == 1 and [r["note"] for r in records] == ["api"] + assert settled is report and report.success is True + save = report.results[-1] + assert save.actuation == "api" and save.effect_verified is True + assert not any(line.startswith("[pipelined]") for line in save.effect_results) + saved = json.loads((tmp_path / "run" / "report.json").read_text()) + assert saved["success"] is True + + +def test_a_durable_api_save_reads_back_inline_and_says_why(tmp_path): + pipeline = ReadbackPipeline() + try: + report, settled, actuator, records = _run( + tmp_path, pipeline=pipeline, durable=True + ) + finally: + pipeline.close() + + assert settled is None and report.success is True + assert report.results[-1].effect_verified is True + assert pipeline.inline_reason == "durable runs verify inline" + + +def test_self_repair_runs_a_repaired_bundle_on_the_gui(tmp_path): + replayer = Replayer(EnterWritesBackend("http://127.0.0.1:9"), vision=FakeVision()) + keep_gui_for_repair(replayer) + span = replayer._web_form_span_at(_span_workflow(), 0, durable=True) + assert span is not None and span[2] == REPAIR_GUI_ONLY + + +class _Session: + def __init__(self, backend) -> None: + self.backend = backend + + +def test_a_probe_or_resumed_run_never_sends_the_api_step(tmp_path): + from openadapt_flow.approval_mode.gate import ApprovalGate + + url, db, stop = fault_serve() + url = url.rstrip("/") + try: + workflow, bundle = _sealed(tmp_path, _span_workflow(), encrypted=False) + actuator = ScriptedActuator(url, ActuationStatus.ACTUATED) + backend = EnterWritesBackend(url) + target = ReplayTarget( + lambda _fault: _Session(backend), + effect_verifier=RestRecordVerifier(url), + policy="permissive", + profile="demo", + replayer_kwargs={ + "vision": FakeVision(), + "web_form_actuator": actuator, + "poll_interval_s": 0.01, + }, + ) + report = target.execute( + workflow, + bundle, + _Session(backend), + tmp_path / "probe", + approval_gate=ApprovalGate("not_gated"), + ) + finally: + stop() + + assert actuator.calls == 0 + assert report.results[0].api_fallback == REPAIR_GUI_ONLY + + +def test_a_standard_change_test_record_run_is_gui_only(tmp_path): + from openadapt_flow.repair.standard_change import TestRecord, run_preflight + + url, db, stop = fault_serve() + url = url.rstrip("/") + try: + workflow, bundle = _sealed(tmp_path, _span_workflow(), encrypted=False) + actuator = ScriptedActuator(url, ActuationStatus.ACTUATED) + backend = EnterWritesBackend(url) + target = ReplayTarget( + lambda _fault: _Session(backend), + effect_verifier=RestRecordVerifier(url), + policy="permissive", + profile="demo", + replayer_kwargs={ + "vision": FakeVision(), + "web_form_actuator": actuator, + "poll_interval_s": 0.01, + }, + ) + record = TestRecord("TEST-001", target) + result = run_preflight(record, bundle, tmp_path / "preflight", gui_only=True) + assert actuator.calls == 0 + assert result.report.results[0].api_fallback == REPAIR_GUI_ONLY + # The pre-flight of the bundle the items will run may use the API step. + run_preflight(record, bundle, tmp_path / "preflight-original") + assert actuator.calls == 1 + finally: + stop()