diff --git a/openadapt_flow/ir.py b/openadapt_flow/ir.py index ce98d4c..c28a570 100644 --- a/openadapt_flow/ir.py +++ b/openadapt_flow/ir.py @@ -20,7 +20,7 @@ from datetime import datetime, timezone from enum import Enum from pathlib import Path -from typing import TYPE_CHECKING, Literal, Optional +from typing import TYPE_CHECKING, Any, Literal, Optional from pydantic import BaseModel, Field @@ -233,6 +233,88 @@ class Postcondition(BaseModel): ) +class ApiBinding(BaseModel): + """A declarative API/tool call that performs a step's write WITHOUT the GUI. + + The TOP of the capability ladder (RFC ``docs/design/WORKFLOW_PROGRAM_IR.md`` + section 4, the ``api`` implementation of a ``TransitionContract``): where the + target app exposes a real API, driving its GUI to make the same write is the + wrong tool. When a step carries an ``ApiBinding`` AND the run configures an + :class:`~openadapt_flow.runtime.actuators.ApiActuator`, the runtime performs + the write by CALLING the API deterministically (``$0``, zero model calls), + confirms it with the same + :class:`~openadapt_flow.runtime.effects.EffectVerifier` that gates a GUI + write, and SKIPS the GUI resolution/act for that step. This is the + ``api`` leaf of the same contract the structural rung realizes as ``dom_uia`` + and the visual ladder realizes as ``vision_rdp`` -- one semantic effect, + backend-specific implementation. + + ADDITIVE and back-compatible: the field is optional and defaults absent, so a + bundle carrying no binding replays EXACTLY as today (GUI actuation). A binding + present with no actuator configured also falls through to the GUI ladder -- + the API tier is an OPTIMIZATION whose safe fallback is the GUI, never a gate + that can block a runnable step. + + Fields are REST/JSON-first but shaped so a FHIR / MCP / tool binding fits the + same model (``kind`` selects the substrate; a FHIR resource POST, an MCP tool + invocation, and a plain REST write all reduce to method + endpoint + body + + the expected effect). Placeholders ``{param}`` in the URL / query / body are + substituted from the run's typed params (``Workflow.params`` overlaid by the + caller's values) at actuation time. + """ + + kind: Literal["rest", "fhir", "mcp", "tool"] = Field( + default="rest", + description="Substrate: 'rest'/'fhir' HTTP, or an 'mcp'/'tool' call", + ) + method: str = Field( + default="POST", + description="HTTP verb (REST/FHIR) or logical operation name (mcp/tool)", + ) + url_template: str = Field( + description=( + "Endpoint template; absolute (http...) or relative to the" + " actuator's base_url. `{param}` placeholders are substituted from" + " the run's typed params." + ), + ) + body_template: dict[str, Any] = Field( + default_factory=dict, + description=( + "JSON request body template; string leaves may carry `{param}`" + " placeholders substituted from the run's params." + ), + ) + query: dict[str, str] = Field( + default_factory=dict, + description="Query-string params; values may carry `{param}` too", + ) + headers: dict[str, str] = Field( + default_factory=dict, + description="Extra request headers; values may carry `{param}`", + ) + expected_status: list[int] = Field( + default_factory=list, + description=( + "Explicit acceptable HTTP status codes; empty means any 2xx is" + " success (anything else is treated as an attempted write of" + " unknown effect and HALTs -- never GUI-retried)." + ), + ) + timeout_s: float = Field( + default=5.0, description="Per-request timeout in seconds" + ) + effects: list["Effect"] = Field( + default_factory=list, + description=( + "The system-of-record effect(s) this call is expected to produce." + " Used to CONFIRM the API write via the run's EffectVerifier when" + " the step itself declares no `effects` (an API write must be" + " confirmable, exactly as a GUI write with declared effects is)." + ), + ) + + # -- Workflow-program IR, Phase 1 (RFC docs/design/WORKFLOW_PROGRAM_IR.md §6) -- # # Additive, backward-compatible first step toward the parameterized workflow @@ -394,6 +476,16 @@ class Step(BaseModel): # before, and a declared effect with no verifier configured is a # deployment error that HALTS (never a silent unverifiable write). effects: list["Effect"] = Field(default_factory=list) + # API/tool binding (RFC section 4, the `api` implementation of the + # transition contract): a declarative description of the API call that + # performs THIS step's write. When present AND the run configures an + # ApiActuator, the runtime performs the write via the API (deterministic, + # $0, no model), confirms it with the EffectVerifier, and SKIPS the GUI + # resolve/act for this step (see openadapt_flow.runtime.replayer). Additive + # and back-compatible: None (default) means the step actuates through the + # GUI resolution ladder EXACTLY as today; a binding present with no actuator + # configured also falls through to the GUI (the API tier's safe fallback). + api_binding: Optional[ApiBinding] = None risk: Literal["reversible", "irreversible"] = "reversible" # Workflow-program IR, Phase 1 (RFC §6) -- both OPTIONAL and additive; a # step with neither replays EXACTLY as a v0 step. Orthogonal to effects / @@ -584,6 +676,11 @@ class StepResult(BaseModel): # calls on this path — effect verification reads the system of record. effect_verified: Optional[bool] = None effect_results: list[str] = Field(default_factory=list) + # How this step's write was PERFORMED: "api" when actuated via an + # ApiBinding (GUI resolve/act skipped), None when it went through the GUI + # resolution ladder (the default). Diagnostic/audit — lets an operator see + # which steps ran on the deterministic API tier vs the visual floor. + actuation: Optional[str] = None # Drift-oracle: postconditions that deterministically FAILED but were # confirmed by the optional on-prem VLM state-verifier under render drift # (recorded for audit; empty unless an appliance is configured). @@ -645,5 +742,6 @@ def save(self, run_dir: Path | str) -> Path: # the forward reference; bundles with no effects are unaffected. from openadapt_flow.runtime.effects.effect import Effect # noqa: E402,F401 +ApiBinding.model_rebuild() Step.model_rebuild() Workflow.model_rebuild() diff --git a/openadapt_flow/runtime/actuators/__init__.py b/openadapt_flow/runtime/actuators/__init__.py new file mode 100644 index 0000000..1b04fc0 --- /dev/null +++ b/openadapt_flow/runtime/actuators/__init__.py @@ -0,0 +1,38 @@ +"""Actuators: PERFORM a step's write through a non-GUI channel. + +The runtime's default actuator is the GUI: it resolves the recorded target on +the live screen (:mod:`openadapt_flow.runtime.resolver`) and clicks / types +through the :class:`~openadapt_flow.backend.Backend`. That is the FLOOR -- it +works on any pixel surface (RDP/Citrix/canvas) -- but it is also the *weakest* +and most expensive way to effect a change: where the target app exposes a real +API, driving its GUI to make the same write is the wrong tool. + +This package adds the TOP of the capability ladder (RFC +``docs/design/WORKFLOW_PROGRAM_IR.md`` section 4, the ``api`` implementation of +a ``TransitionContract``): when a step carries an +:class:`~openadapt_flow.ir.ApiBinding`, perform the write by CALLING the API +deterministically -- $0, zero model calls -- and confirm it with the same +:class:`~openadapt_flow.runtime.effects.EffectVerifier` that gates a GUI write. +The GUI resolution ladder is then SKIPPED for that step. A step with no binding +(or with no actuator configured) behaves EXACTLY as before -- the API tier is +additive and falls through to the structural -> visual ladder. + +Public surface: + +- :class:`ApiActuator` -- the REST/JSON actuator (and the shape a FHIR/MCP/tool + actuator slots into). +- :class:`ApiActuationResult`, :class:`ActuationStatus` -- the fail-safe + outcome of an actuation attempt (the no-double-write contract). +""" + +from openadapt_flow.runtime.actuators.api import ( # noqa: F401 + ActuationStatus, + ApiActuationResult, + ApiActuator, +) + +__all__ = [ + "ApiActuator", + "ApiActuationResult", + "ActuationStatus", +] diff --git a/openadapt_flow/runtime/actuators/api.py b/openadapt_flow/runtime/actuators/api.py new file mode 100644 index 0000000..f45ce99 --- /dev/null +++ b/openadapt_flow/runtime/actuators/api.py @@ -0,0 +1,278 @@ +"""REST/JSON :class:`ApiActuator` -- perform a step's write via its API binding. + +This is the ``api`` implementation tier of the RFC's transition contract +(``docs/design/WORKFLOW_PROGRAM_IR.md`` section 4: "call the app's API / DB +write; effect probed against the system of record"). Given a step's +:class:`~openadapt_flow.ir.ApiBinding` and the run's typed params, it +substitutes the params into the endpoint / query / body templates and issues a +single deterministic HTTP request -- no pixel matching, no model call, ``$0``. + +The hard requirement is **never double-write the same effect**. A naive +"try the API, and on any failure fall back to GUI-clicking the same save" +would, on a request that the server actually PROCESSED (a read-timeout after +commit, a 5xx after a partial write), perform the write TWICE. So the actuator +classifies every attempt into exactly one of three fail-safe outcomes +(:class:`ActuationStatus`), keyed on *whether the request could have reached +the server*: + +- :attr:`ActuationStatus.UNAVAILABLE` -- the request was **never sent** (the + TCP connection was never established: connection refused, DNS failure, + connect-timeout) or the binding could not even be built (a param the URL/body + needs was not supplied). Nothing was written, so it is SAFE for the caller to + fall through to the GUI ladder for this step. This is the "reachable + ApiBinding" gate: an unreachable endpoint simply is not actuated. +- :attr:`ActuationStatus.ACTUATED` -- the request was sent and the server + returned success (2xx, or an explicitly-allowed status). The write was + performed; the caller MUST now confirm it with the EffectVerifier and MUST + skip the GUI (never re-do the write). +- :attr:`ActuationStatus.HALT` -- the request WAS sent but its outcome is + unknown or a rejection (read-timeout after the bytes went out, a non-2xx + response, any post-send transport error). The write MAY have landed, so the + caller must NEITHER accept it as success NOR GUI-write it again -- it HALTs + (the same refuse-rather-than-guess posture as the EffectVerifier's + INDETERMINATE verdict). + +The connect-phase / read-phase split is exact in ``requests``: +``ConnectTimeout`` subclasses ``ConnectionError`` (nothing sent -> UNAVAILABLE) +while ``ReadTimeout`` does not (bytes sent -> HALT), so catching +``ConnectionError`` before ``Timeout`` gives the right classification. + +Import-light: ``requests`` is imported lazily so importing this module (and the +runtime package) stays cheap and model-free. +""" + +from __future__ import annotations + +from enum import Enum +from typing import Any, Optional + +from pydantic import BaseModel + +from openadapt_flow.ir import ApiBinding + + +class ActuationStatus(str, Enum): + """The fail-safe outcome of an API actuation attempt (no-double-write).""" + + #: The write was performed and the server acknowledged success -> the + #: caller confirms it with the EffectVerifier and SKIPS the GUI. + ACTUATED = "actuated" + #: The request was never sent (endpoint unreachable, or the binding could + #: not be built) -> nothing was written; SAFE to fall through to the GUI + #: ladder for this step. + UNAVAILABLE = "unavailable" + #: The request WAS sent but its outcome is unknown or a rejection -> the + #: write may have landed; HALT (never accept, never GUI-write it again). + HALT = "halt" + + +class ApiActuationResult(BaseModel): + """Outcome of one :meth:`ApiActuator.actuate` call.""" + + status: ActuationStatus + substrate: str = "rest" + #: Human-readable reason (audit trail / error surface). Contains the + #: UNSUBSTITUTED method + endpoint template only -- never the substituted + #: values -- so it is safe to log without leaking PHI-bearing params. + reason: str = "" + #: HTTP status code when a response was received; None otherwise. + http_status: Optional[int] = None + #: ``"METHOD url_template"`` (unsubstituted) for the audit line. + request_summary: str = "" + + @property + def actuated(self) -> bool: + return self.status is ActuationStatus.ACTUATED + + @property + def should_fall_through(self) -> bool: + """True when the caller may safely fall through to the GUI ladder + (the request was never sent, so nothing was written).""" + return self.status is ActuationStatus.UNAVAILABLE + + @property + def should_halt(self) -> bool: + return self.status is ActuationStatus.HALT + + +class _MissingParam(KeyError): + """A template referenced a param the run did not supply.""" + + +class _StrictMap(dict): + def __missing__(self, key: str) -> Any: # noqa: D401 + raise _MissingParam(key) + + +def _fill(template: str, params: dict[str, str]) -> str: + """Substitute ``{param}`` placeholders in ``template`` from ``params``. + + Raises :class:`_MissingParam` when the template references a key that is + not in ``params`` -- the binding cannot be built, so the actuator reports + UNAVAILABLE (a before-send problem: nothing is written, GUI fallback is + safe) rather than sending a half-formed request. + """ + return template.format_map(_StrictMap(params)) + + +def _fill_body(node: Any, params: dict[str, str]) -> Any: + """Recursively substitute ``{param}`` in every string leaf of a JSON body.""" + if isinstance(node, str): + return _fill(node, params) + if isinstance(node, dict): + return {k: _fill_body(v, params) for k, v in node.items()} + if isinstance(node, list): + return [_fill_body(v, params) for v in node] + return node + + +class ApiActuator: + """Perform a step's write via its :class:`~openadapt_flow.ir.ApiBinding`. + + Bound to a deployment's API base URL (and an optional injected session for + auth headers / tests), mirroring + :class:`~openadapt_flow.runtime.effects.rest.RestRecordVerifier`. The + binding's ``url_template`` may be absolute (``http...``) or relative to + ``base_url``; params from the run substitute into the URL, query, and body + templates. A single request is issued and classified into the fail-safe + :class:`ActuationStatus` outcome. Makes ZERO model calls. + + Args: + base_url: Deployment API base URL (used for relative ``url_template``s; + trailing slash optional). May be empty when every binding is + absolute. + session: Optional ``requests``-style session (auth headers / test + injection); a module-level default is created lazily when omitted. + timeout_s: Default per-request timeout when the binding sets none. + """ + + substrate = "rest" + + def __init__( + self, + base_url: str = "", + *, + session: Any = None, + timeout_s: float = 5.0, + ) -> None: + self.base_url = base_url.rstrip("/") + self.default_timeout_s = timeout_s + self._session = session + + def _get_session(self) -> Any: + if self._session is None: + import requests # lazy: keep module import light and model-free + + self._session = requests.Session() + return self._session + + def _resolve_url(self, url_template: str) -> str: + if url_template.startswith(("http://", "https://")): + return url_template + return f"{self.base_url}{url_template}" + + def actuate( + self, binding: ApiBinding, params: dict[str, str] + ) -> ApiActuationResult: + """Perform ``binding``'s write, substituting ``params``; classify safely. + + Returns an :class:`ApiActuationResult` whose :attr:`status` tells the + caller exactly one safe next move: confirm-and-skip-GUI (ACTUATED), + fall-through-to-GUI (UNAVAILABLE, nothing was written), or HALT + (attempted, outcome unknown -- never double-write). Never raises. + """ + summary = f"{binding.method} {binding.url_template}" + + # -- build the request (a before-send problem is UNAVAILABLE) --------- + try: + url = self._resolve_url(_fill(binding.url_template, params)) + query = {k: _fill(v, params) for k, v in binding.query.items()} + body = _fill_body(binding.body_template, params) + headers = {k: _fill(v, params) for k, v in binding.headers.items()} + except _MissingParam as exc: + return ApiActuationResult( + status=ActuationStatus.UNAVAILABLE, + substrate=self.substrate, + reason=( + f"binding for {summary} references param {exc} not supplied " + "by the run -- API tier unavailable, falling through to GUI" + ), + request_summary=summary, + ) + + import requests # lazy; hierarchy: ConnectTimeout is a ConnectionError + + timeout = binding.timeout_s or self.default_timeout_s + try: + resp = self._get_session().request( + binding.method.upper(), + url, + params=query or None, + json=body if body else None, + headers=headers or None, + timeout=timeout, + ) + except requests.exceptions.ConnectionError as exc: + # Connection never established (refused / DNS / connect-timeout): + # the request was NEVER sent, so nothing was written -> it is safe + # to fall through to the GUI ladder for this step. + return ApiActuationResult( + status=ActuationStatus.UNAVAILABLE, + substrate=self.substrate, + reason=( + f"endpoint unreachable ({type(exc).__name__}) -- request " + "not sent, API tier unavailable, falling through to GUI" + ), + request_summary=summary, + ) + except requests.exceptions.Timeout as exc: + # Read-timeout: the bytes WENT OUT and the server may have processed + # the write. Outcome unknown -> HALT (never GUI-write it again). + return ApiActuationResult( + status=ActuationStatus.HALT, + substrate=self.substrate, + reason=( + f"request sent but timed out awaiting the response " + f"({type(exc).__name__}) -- the write may have landed; HALT " + "(never double-write via the GUI)" + ), + request_summary=summary, + ) + except Exception as exc: # noqa: BLE001 + # Any other transport error after the request left the client is of + # unknown effect on the server -> HALT rather than risk a duplicate. + return ApiActuationResult( + status=ActuationStatus.HALT, + substrate=self.substrate, + reason=( + f"request failed after being sent ({type(exc).__name__}) -- " + "outcome unknown; HALT (never double-write via the GUI)" + ), + request_summary=summary, + ) + + allowed = binding.expected_status or list(range(200, 300)) + ok = resp.status_code in allowed or ( + not binding.expected_status and resp.status_code // 100 == 2 + ) + if ok: + return ApiActuationResult( + status=ActuationStatus.ACTUATED, + substrate=self.substrate, + reason=f"{summary} -> {resp.status_code}", + http_status=resp.status_code, + request_summary=summary, + ) + # Non-success response: the request was PROCESSED by the server. Even a + # clean rejection is ambiguous about what (if anything) persisted, and + # re-driving the same write through the GUI risks a duplicate -> HALT. + return ApiActuationResult( + status=ActuationStatus.HALT, + substrate=self.substrate, + reason=( + f"{summary} returned {resp.status_code} (not success) -- the " + "write was attempted; HALT (never double-write via the GUI)" + ), + http_status=resp.status_code, + request_summary=summary, + ) diff --git a/openadapt_flow/runtime/replayer.py b/openadapt_flow/runtime/replayer.py index 4433052..9990ca9 100644 --- a/openadapt_flow/runtime/replayer.py +++ b/openadapt_flow/runtime/replayer.py @@ -60,6 +60,7 @@ from openadapt_flow.runtime import healing as healing_mod from openadapt_flow.runtime import identity as identity_mod from openadapt_flow.runtime.effects import ( + Effect, EffectState, EffectVerifier, reconcile_or_escalate, @@ -148,6 +149,15 @@ class Replayer: is REFUTED as a duplicate — deletes the extras and re-verifies (RECONCILED continues, else ESCALATE halts). None (default) means every refuted/indeterminate effect simply halts. + api_actuator: Optional ``ApiActuator`` (``runtime.actuators``) bound to + the deployment's API. It is the TOP of the capability ladder: when a + step carries an ``ir.Step.api_binding`` and this actuator is set, + the step's write is PERFORMED via the API (deterministic, $0, no + model), CONFIRMED by the ``effect_verifier``, and the GUI + resolve/act is SKIPPED. None (default) disables the API tier -- a + step with a binding then actuates through the GUI ladder exactly as + today (the API tier is an optimization whose safe fallback IS the + GUI). Never makes a model call. poll_interval_s: Postcondition polling interval in seconds. """ @@ -161,6 +171,7 @@ def __init__( state_verifier: Optional[Any] = None, effect_verifier: Optional[EffectVerifier] = None, effect_compensator: Optional[Any] = None, + api_actuator: Optional[Any] = None, poll_interval_s: float = 0.05, use_structural: bool = True, ) -> None: @@ -184,6 +195,21 @@ def __init__( # makes a model call — the $0 runtime guarantee is preserved. self.effect_verifier = effect_verifier self.effect_compensator = effect_compensator + # API/tool actuator -- the TOP of the capability ladder (RFC section 4 + # `api` tier). When set, a step carrying an `api_binding` has its write + # performed via the API and confirmed by the effect_verifier, SKIPPING + # the GUI resolve/act. None (default) = the API tier is off and such a + # step falls through to the GUI ladder unchanged. Deterministic; makes + # no model call (the $0 runtime guarantee holds on this path too). + self.api_actuator = api_actuator + # Whether the deterministic structural ACTION rung (top of the ladder) + # may run. True (default) = product behavior: on a structure-bearing + # backend, resolve the recorded target as a DOM/UIA element. Set False + # to force the VISUAL fallback floor even on a structure-bearing + # backend -- used to characterize the pixel-only substrate path + # (RDP/Citrix/VDI) on a Playwright surface (see the e2e visual-floor + # drift/heal suites). Never disables the visual ladder itself. + self.use_structural = use_structural # Optional local-VLM identity veto (tier 3 of the identity ladder), # OFF by default like the grounder: an IdentityVLM verifier or None. # When None the ladder runs structured-text + pixel-compare + OCR + @@ -300,6 +326,12 @@ def run( report.rung_counts[rung] = report.rung_counts.get(rung, 0) + 1 if rung == "grounder": report.model_calls += 1 + elif result.ok and result.actuation == "api": + # API-tier actuation has no visual resolution rung; count it + # under "api" so the run report shows the deterministic top of + # the ladder in the same place as the visual rungs. Zero model + # calls (the $0 guarantee holds on the API path too). + report.rung_counts["api"] = report.rung_counts.get("api", 0) + 1 # Drift-oracle state-verifier calls are model calls too (honest # accounting: a rescued run is not a zero-model run). report.model_calls += result.drift_oracle_calls @@ -376,6 +408,20 @@ def _run_step( t0 = time.monotonic() result = StepResult(step_id=step.id, intent=step.intent, ok=False) + # API/tool tier -- the TOP of the capability ladder. When the step + # carries an api_binding and an ApiActuator is configured, PERFORM the + # write via the API (deterministic, $0, no GUI), CONFIRM it with the + # EffectVerifier, and SKIP the GUI resolve/act entirely. Returns True + # when the API tier took responsibility (actuated+verified, or HALTed); + # returns False (API tier unavailable -- endpoint unreachable / no + # binding param) to fall through to the GUI ladder below with NO write + # yet performed (the no-double-write contract). See + # openadapt_flow.runtime.actuators. + if self.api_actuator is not None and step.api_binding is not None: + if self._try_api_tier(step, params, result): + result.elapsed_ms = (time.monotonic() - t0) * 1000.0 + return result + # Settle before the pre-action screenshot. before_png = self.vision.wait_settled(self.backend) result.before_png = self._save_step_png(run_dir, step.id, "before", before_png) @@ -617,6 +663,115 @@ def _run_step( result.elapsed_ms = (time.monotonic() - t0) * 1000.0 return result + # -- API/tool actuation tier (top of the capability ladder) ----------------- + + def _try_api_tier( + self, + step: Step, + params: dict[str, str], + result: StepResult, + ) -> bool: + """Perform ``step.api_binding``'s write via the API and confirm it. + + The TOP of the capability ladder (RFC section 4 ``api`` tier): a + deterministic, ``$0``, model-free write, gated by the SAME + ``EffectVerifier`` as a GUI write. For an API write the target + "identity" is the explicit API parameter -- stronger than a resolved + pixel band -- so no identity gate is weakened by skipping the GUI. + + Fail-safe ordering (the no-double-write contract): + + - refuse BEFORE any request when the write could not be confirmed (no + effect contract, or no verifier) -- an unverifiable consequential + write never proceeds; + - snapshot the system of record, then actuate ONCE; + - UNAVAILABLE (request never sent) -> return False so the caller falls + through to the GUI ladder; nothing was written, so no double-write; + - HALT (request sent, outcome unknown / rejected) -> stop the run; the + write may have landed, so it is NEVER re-done through the GUI; + - ACTUATED (2xx) -> CONFIRM with the EffectVerifier; a non-CONFIRMED + verdict HALTs exactly as it would for a GUI write. + + Returns True when the API tier took responsibility for the step (the + result is final -- actuated+verified, HALTed, or a config-error HALT), + False when the API tier is UNAVAILABLE and the caller must fall through + to the GUI resolution ladder. + """ + binding = step.api_binding + assert binding is not None # guaranteed by the caller + + # An API write MUST be confirmable against the system of record -- + # exactly as a GUI write that declares effects must be. The binding may + # carry its own effect contract; the step's own effects take precedence. + effects = step.effects or binding.effects + if not effects: + result.effect_verified = False + result.effect_results.append( + "API binding declares no effect to confirm the write (neither " + "step.effects nor api_binding.effects) -- refusing an " + "unverifiable API write (fail-safe HALT)" + ) + result.ok = False + result.error = ( + f"Step '{step.id}' ({step.intent}) has an API binding but no " + "system-of-record effect to CONFIRM the write -- an API write " + "must be verifiable; refusing to actuate " + "(deployment/configuration error); run aborted" + ) + return True + if self.effect_verifier is None: + result.effect_verified = False + result.effect_results.append( + "API binding present but no EffectVerifier is configured to " + "confirm the write (fail-safe HALT)" + ) + result.ok = False + result.error = ( + f"Step '{step.id}' ({step.intent}) would actuate via the API " + "but no EffectVerifier is configured to confirm the write -- " + "refusing an unverifiable consequential write; run aborted" + ) + return True + + # Snapshot the system of record BEFORE the write so the verifier counts + # only what THIS actuation wrote (delta / at-most-once / collateral + # loss), then actuate exactly once. + before = self.effect_verifier.capture_pre_state() + outcome = self.api_actuator.actuate(binding, params) + + from openadapt_flow.runtime.actuators import ActuationStatus + + if outcome.status == ActuationStatus.UNAVAILABLE: + # The request was NEVER sent -- nothing was written. Fall through to + # the GUI ladder for this step (no double-write risk). The GUI path + # populates the result; leave only an audit breadcrumb here. + result.effect_results.append(f"[api] {outcome.reason}") + return False + + # From here the request WAS attempted -- this step is API-tier and is + # NEVER also GUI-written (the no-double-write contract). + result.actuation = "api" + + if outcome.status == ActuationStatus.HALT: + result.effect_verified = False + result.effect_results.append(f"[api] {outcome.reason}") + result.ok = False + result.error = ( + f"API actuation HALTED step '{step.id}' ({step.intent}): " + f"{outcome.reason} -- run aborted" + ) + return True + + # ACTUATED (2xx): confirm the write against the system of record with + # the same EffectVerifier that gates a GUI write. A non-CONFIRMED + # verdict HALTs. The GUI resolve/act (and screen postconditions) are + # SKIPPED -- the record, not the screen, is the oracle for an API write. + result.effect_results.append(f"[api] actuated {outcome.reason}") + error = self._verify_effects(step, before, result, effects=effects) + result.ok = error is None + result.error = error + return True + # -- system-of-record effect verification ----------------------------------- def _verify_effects( @@ -624,25 +779,37 @@ def _verify_effects( step: Step, before: EffectState, result: StepResult, + effects: Optional[list["Effect"]] = None, ) -> Optional[str]: """Verify each declared Effect against the system of record; HALT on any non-CONFIRMED verdict. - Runs AFTER the action and its screen postconditions. For each - ``step.effects`` entry the configured verifier rules CONFIRMED / - REFUTED / INDETERMINATE against the REAL record (an API/DB read, never - the screen). A CONFIRMED effect proceeds; a non-CONFIRMED effect halts, - except that an IRREVERSIBLE effect is first routed through - ``reconcile_or_escalate`` -- a duplicate the configured compensator can - undo is RECONCILED (proceed), everything else ESCALATEs (halt). Every - verdict is recorded on ``result.effect_results`` for the audit trail. - Makes ZERO model calls (the $0 runtime guarantee). + Runs AFTER the action and its screen postconditions (for a GUI write), + or immediately after an API actuation. For each effect the configured + verifier rules CONFIRMED / REFUTED / INDETERMINATE against the REAL + record (an API/DB read, never the screen). A CONFIRMED effect proceeds; + a non-CONFIRMED effect halts, except that an IRREVERSIBLE effect is + first routed through ``reconcile_or_escalate`` -- a duplicate the + configured compensator can undo is RECONCILED (proceed), everything else + ESCALATEs (halt). Every verdict is recorded on ``result.effect_results`` + for the audit trail. Makes ZERO model calls (the $0 runtime guarantee). + + Args: + step: The step being verified (for the audit/error text). + before: The pre-action snapshot of the system of record. + result: The step result to record verdicts on. + effects: The effects to verify; defaults to ``step.effects`` (the + GUI path). The API tier passes ``step.effects or + api_binding.effects`` so a self-contained binding carries its + own confirmation contract. Returns an error string (HALT) or None (all effects confirmed or reconciled). """ assert self.effect_verifier is not None # guaranteed by the caller - for effect in step.effects: + if effects is None: + effects = step.effects + for effect in effects: # A compiler-mined PLACEHOLDER effect (binding not derivable from # the demonstration — see compiler.effect_mining) carries a # sentinel selector, NOT a real one. Never verify it against the diff --git a/tests/test_replayer_api_actuator.py b/tests/test_replayer_api_actuator.py new file mode 100644 index 0000000..24cd66a --- /dev/null +++ b/tests/test_replayer_api_actuator.py @@ -0,0 +1,487 @@ +"""Live-local tests for the API/tool actuator tier (top of the capability ladder). + +These drive the REAL Replayer wired with a REAL +:class:`~openadapt_flow.runtime.actuators.ApiActuator` against a REAL system of +record (the in-process MockMed ``fault_server``), confirmed by the same +:class:`~openadapt_flow.runtime.effects.RestRecordVerifier` the GUI-write +effects tests use. No model calls, no network beyond localhost -- runs in CI. + +The theses these pin (RFC ``docs/design/WORKFLOW_PROGRAM_IR.md`` section 4, the +``api`` implementation of a transition contract): + +- a step with a REACHABLE ``ApiBinding`` performs its write via the API, the + EffectVerifier CONFIRMS it against the record, and the GUI actuation is + SKIPPED entirely -- ``$0``, zero model calls; +- an UNREACHABLE API falls through to the GUI ladder CLEANLY, with NO + double-write (the request never left the client, so nothing was written); +- a step with NO binding replays byte-identically to today (back-compat); +- a REFUTED effect after an API write HALTS (the record, not the screen, is the + oracle); +- an API write whose outcome is unknown / rejected HALTs (never GUI-retried). +""" + +from __future__ import annotations + +import requests +from openadapt_flow.runtime.actuators import ActuationStatus, ApiActuator +from openadapt_flow.runtime.effects import ( + Effect, + EffectKind, + RestRecordVerifier, +) + +# Reuse the scripted fakes from the main replayer unit tests (pytest's prepend +# import mode puts tests/ on sys.path). +from test_replayer import FakeBackend, FakeVision, Match + +from openadapt_flow.ir import ( + ActionKind, + ApiBinding, + Postcondition, + PostconditionKind, + Step, + Workflow, +) +from openadapt_flow.mockmed.fault_server import serve as fault_serve +from openadapt_flow.runtime.replayer import Replayer + +TARGET = {"patient_id": "p1", "type": "Triage"} + + +class GuiWritingBackend(FakeBackend): + """A GUI backend whose ``press`` writes to the system of record. + + Models the CURRENT (GUI) path: the consequential keypress makes the app + POST an encounter. Used to prove the API tier SKIPS the GUI (no press + lands) on the actuated path, and to prove the fall-through path DOES + GUI-write when the API tier is unavailable. ``record_presses`` records + every ``press`` so a test can assert the GUI was or was not driven. + """ + + def __init__(self, sor_url, *, viewport=(300, 200)): + super().__init__(viewport=viewport) + self.sor_url = sor_url.rstrip("/") + + def press(self, key): + super().press(key) + requests.post( + f"{self.sor_url}/api/encounter", + json={"patient_id": "p1", "type": "Triage", "note": "gui"}, + timeout=5, + ) + + +def _fault_server(): + url, db, stop = fault_serve() + return url.rstrip("/"), db, stop + + +def _api_save_workflow( + *, url_template="/api/encounter", effects, risk="reversible", effects_on="step" +): + """A one-step workflow: press Enter (the GUI action), but carrying an + ApiBinding so the API tier performs the write instead. The screen + postcondition PASSES -- the point is that the API tier bypasses it. + + ``effects_on`` places the effect contract on the ``"step"`` (the canonical + location, verified on BOTH the API tier and a GUI fall-through) or on the + ``"binding"`` (the self-contained-binding case, used by the API tier when + the step declares none).""" + step_effects = effects if effects_on == "step" else [] + binding_effects = effects if effects_on == "binding" else [] + return Workflow( + name="api-save", + steps=[ + Step( + id="save", + intent="save encounter", + action=ActionKind.KEY, + key="Enter", + expect=[ + Postcondition( + kind=PostconditionKind.TEXT_PRESENT, + text="Saved", + timeout_s=0.2, + ) + ], + risk=risk, + effects=step_effects, + api_binding=ApiBinding( + method="POST", + url_template=url_template, + body_template={ + "patient_id": "p1", + "type": "Triage", + "note": "{note}", + }, + effects=binding_effects, + timeout_s=2.0, + ), + ) + ], + params={"note": "charted via API"}, + ) + + +def _vision_that_confirms_saved(): + vision = FakeVision() + vision.text_results = { + "Saved": Match(point=(50, 10), region=(30, 5, 40, 10), confidence=0.9) + } + return vision + + +def _dirs(tmp_path): + bundle = tmp_path / "bundle" + (bundle / "templates").mkdir(parents=True) + return bundle, tmp_path / "run" + + +def _record_written(**over): + kw = dict(kind=EffectKind.RECORD_WRITTEN, match=TARGET, expected_count=1, + timeout_s=2.0) + kw.update(over) + return Effect(**kw) + + +# -- ACTUATED + CONFIRMED: API performs the write, GUI is skipped ----------- + + +def test_api_binding_actuates_and_confirms_skipping_gui(tmp_path): + url, db, stop = _fault_server() + try: + backend = GuiWritingBackend(url) + vision = _vision_that_confirms_saved() + # The effect contract rides on the BINDING itself (self-contained) to + # prove the API tier confirms even when the step declares no effects. + workflow = _api_save_workflow( + effects=[_record_written()], effects_on="binding" + ) + bundle, run_dir = _dirs(tmp_path) + replayer = Replayer( + backend, + vision=vision, + effect_verifier=RestRecordVerifier(url), + api_actuator=ApiActuator(url), + poll_interval_s=0.01, + ) + report = replayer.run(workflow, bundle_dir=bundle, run_dir=run_dir) + + assert report.success is True + r = report.results[0] + # The write landed via the API and was CONFIRMED against the record. + assert r.actuation == "api" + assert r.effect_verified is True + assert any("CONFIRMED" in line for line in r.effect_results) + assert any("actuated" in line for line in r.effect_results) + # The GUI was SKIPPED: no click/type/press was ever issued. + assert backend.actions == [] + # Exactly one record, written by the API with the run's param value. + records = db.snapshot()["records"] + assert len(records) == 1 + assert records[0]["note"] == "charted via API" + # $0 guarantee: the API path makes no model calls. + assert report.model_calls == 0 + assert report.est_model_cost_usd == 0.0 + # Audit: the run report counts the deterministic top-of-ladder tier. + assert report.rung_counts.get("api") == 1 + finally: + stop() + + +# -- UNREACHABLE API falls through to the GUI ladder cleanly (no double-write) -- + + +def test_unreachable_api_falls_through_to_gui_no_double_write(tmp_path): + url, db, stop = _fault_server() + try: + # The actuator points at a DEAD endpoint (connection refused): the + # request is never sent, so the API tier is UNAVAILABLE and the step + # falls through to the GUI, which performs the write exactly once. + dead = "http://127.0.0.1:1" # nothing listens here -> ConnectionError + backend = GuiWritingBackend(url) + vision = _vision_that_confirms_saved() + workflow = _api_save_workflow(effects=[_record_written()]) + bundle, run_dir = _dirs(tmp_path) + replayer = Replayer( + backend, + vision=vision, + effect_verifier=RestRecordVerifier(url), + api_actuator=ApiActuator(dead, timeout_s=1.0), + poll_interval_s=0.01, + ) + report = replayer.run(workflow, bundle_dir=bundle, run_dir=run_dir) + + assert report.success is True + r = report.results[0] + # Fell through to the GUI: actuation is NOT "api"; the keypress fired. + assert r.actuation is None + assert ("press", "Enter") in backend.actions + # An audit breadcrumb records the API tier was unavailable. + assert any("unavailable" in line.lower() for line in r.effect_results) + # The write happened EXACTLY ONCE (no double-write): the GUI wrote it, + # the dead API did not, and the effect check CONFIRMS the single row. + assert len(db.snapshot()["records"]) == 1 + assert r.effect_verified is True + assert report.model_calls == 0 + finally: + stop() + + +# -- REFUTED effect after an API write HALTS --------------------------------- + + +def test_api_write_refuted_by_record_halts(tmp_path): + url, db, stop = _fault_server() + try: + # The API write lands ONE row, but the effect asserts a record for a + # DIFFERENT patient (p2) -- the record refutes it (0 found, expected 1) + # -> HALT, even though the API returned 2xx. + backend = GuiWritingBackend(url) + vision = _vision_that_confirms_saved() + workflow = _api_save_workflow( + effects=[_record_written(match={"patient_id": "p2", "type": "Triage"})] + ) + bundle, run_dir = _dirs(tmp_path) + replayer = Replayer( + backend, + vision=vision, + effect_verifier=RestRecordVerifier(url), + api_actuator=ApiActuator(url), + poll_interval_s=0.01, + ) + report = replayer.run(workflow, bundle_dir=bundle, run_dir=run_dir) + + assert report.success is False + r = report.results[0] + assert r.actuation == "api" + assert r.effect_verified is False + assert r.ok is False + assert "refuted" in (r.error or "").lower() + assert "system of record" in (r.error or "") + # The API DID write one (p1) row -- the write was performed; it is the + # RECORD check that refused it, and the GUI never ran (no double-write). + assert backend.actions == [] + assert len(db.snapshot()["records"]) == 1 + finally: + stop() + + +# -- ATTEMPTED-but-rejected API write HALTS (never GUI-retried) -------------- + + +def test_api_non_2xx_halts_never_double_writes(tmp_path): + url, db, stop = _fault_server() + try: + # ?fault=session makes /api/encounter return 401: the request WAS sent + # (nothing persisted here, but that is not knowable in general), so the + # actuator must HALT rather than fall through and GUI-write a possible + # duplicate. + backend = GuiWritingBackend(url) + vision = _vision_that_confirms_saved() + workflow = _api_save_workflow( + url_template="/api/encounter?fault=session", effects=[_record_written()] + ) + bundle, run_dir = _dirs(tmp_path) + replayer = Replayer( + backend, + vision=vision, + effect_verifier=RestRecordVerifier(url), + api_actuator=ApiActuator(url), + poll_interval_s=0.01, + ) + report = replayer.run(workflow, bundle_dir=bundle, run_dir=run_dir) + + assert report.success is False + r = report.results[0] + assert r.actuation == "api" + assert r.ok is False + assert "halted" in (r.error or "").lower() + # The GUI was NEVER driven -- the attempted API write is not re-done. + assert backend.actions == [] + assert db.snapshot()["records"] == [] + finally: + stop() + + +# -- FAIL-SAFE: an API binding with no effect to confirm the write -> HALT ---- + + +def test_api_binding_without_effects_is_config_error_halt(tmp_path): + url, db, stop = _fault_server() + try: + backend = GuiWritingBackend(url) + vision = _vision_that_confirms_saved() + # An ApiBinding with NO effects (neither on the step nor the binding): + # the write could not be confirmed, so it must be refused BEFORE any + # request is sent. + workflow = _api_save_workflow(effects=[]) + bundle, run_dir = _dirs(tmp_path) + replayer = Replayer( + backend, + vision=vision, + effect_verifier=RestRecordVerifier(url), + api_actuator=ApiActuator(url), + poll_interval_s=0.01, + ) + report = replayer.run(workflow, bundle_dir=bundle, run_dir=run_dir) + + assert report.success is False + r = report.results[0] + assert r.ok is False + assert "must be verifiable" in (r.error or "") + # Nothing was written and the GUI never ran -- refused before actuating. + assert backend.actions == [] + assert db.snapshot()["records"] == [] + finally: + stop() + + +# -- FAIL-SAFE: an API binding but no EffectVerifier configured -> HALT ------- + + +def test_api_binding_without_verifier_halts(tmp_path): + url, db, stop = _fault_server() + try: + backend = GuiWritingBackend(url) + vision = _vision_that_confirms_saved() + workflow = _api_save_workflow(effects=[_record_written()]) + bundle, run_dir = _dirs(tmp_path) + # An ApiActuator but NO EffectVerifier: an API write we cannot confirm. + replayer = Replayer( + backend, + vision=vision, + api_actuator=ApiActuator(url), + poll_interval_s=0.01, + ) + report = replayer.run(workflow, bundle_dir=bundle, run_dir=run_dir) + + assert report.success is False + r = report.results[0] + assert r.ok is False + assert "no EffectVerifier" in (r.error or "") + assert backend.actions == [] + assert db.snapshot()["records"] == [] + finally: + stop() + + +# -- BACK-COMPAT: a binding present but NO actuator configured -> GUI path ---- + + +def test_binding_present_but_no_actuator_uses_gui_unchanged(tmp_path): + url, db, stop = _fault_server() + try: + backend = GuiWritingBackend(url) + vision = _vision_that_confirms_saved() + workflow = _api_save_workflow(effects=[_record_written()]) + bundle, run_dir = _dirs(tmp_path) + # NO api_actuator -> the API tier is OFF; the step actuates via the GUI + # exactly as today (the binding is inert, its declared effects are still + # verified by the normal GUI effect path). + replayer = Replayer( + backend, + vision=vision, + effect_verifier=RestRecordVerifier(url), + poll_interval_s=0.01, + ) + report = replayer.run(workflow, bundle_dir=bundle, run_dir=run_dir) + + assert report.success is True + r = report.results[0] + assert r.actuation is None + assert ("press", "Enter") in backend.actions + assert r.effect_verified is True + assert len(db.snapshot()["records"]) == 1 + finally: + stop() + + +# -- BACK-COMPAT: a no-binding, no-effects bundle replays byte-identically ---- + + +def test_no_binding_bundle_replays_unchanged(tmp_path): + # A plain step with no api_binding and no effects: no API machinery engages, + # the GUI runs, and the result carries no actuation marker. + backend = FakeBackend() + vision = _vision_that_confirms_saved() + workflow = Workflow( + name="plain", + steps=[ + Step( + id="save", + intent="save", + action=ActionKind.KEY, + key="Enter", + expect=[ + Postcondition( + kind=PostconditionKind.TEXT_PRESENT, text="Saved", + timeout_s=0.2, + ) + ], + ) + ], + ) + bundle, run_dir = _dirs(tmp_path) + replayer = Replayer( + backend, + vision=vision, + api_actuator=ApiActuator("http://127.0.0.1:1"), # present but unused + poll_interval_s=0.01, + ) + report = replayer.run(workflow, bundle_dir=bundle, run_dir=run_dir) + + assert report.success is True + r = report.results[0] + assert r.actuation is None + assert r.effect_verified is None + assert r.effect_results == [] + assert ("press", "Enter") in backend.actions + + +# -- Unit: the actuator's fail-safe classification (no double-write contract) -- + + +def test_actuator_unavailable_on_connection_refused(): + binding = ApiBinding(url_template="http://127.0.0.1:1/api/encounter", + body_template={"patient_id": "p1"}, timeout_s=1.0) + res = ApiActuator().actuate(binding, {}) + assert res.status is ActuationStatus.UNAVAILABLE + assert res.should_fall_through is True + + +def test_actuator_unavailable_on_missing_param(): + # A URL/body that references a param the run did not supply cannot be built + # -> UNAVAILABLE (before-send, nothing written, safe to fall through). + binding = ApiBinding(url_template="/api/encounter", + body_template={"note": "{missing}"}) + res = ApiActuator("http://127.0.0.1:9").actuate(binding, {}) + assert res.status is ActuationStatus.UNAVAILABLE + assert "missing" in res.reason + + +def test_actuator_actuated_on_2xx(): + url, db, stop = _fault_server() + try: + binding = ApiBinding(url_template="/api/encounter", + body_template={"patient_id": "p1", "type": "Triage", + "note": "n"}, timeout_s=2.0) + res = ApiActuator(url).actuate(binding, {}) + assert res.status is ActuationStatus.ACTUATED + assert res.http_status == 200 + assert len(db.snapshot()["records"]) == 1 + finally: + stop() + + +def test_actuator_halts_on_non_2xx(): + url, _db, stop = _fault_server() + try: + binding = ApiBinding(url_template="/api/encounter?fault=session", + body_template={"patient_id": "p1"}, timeout_s=2.0) + res = ApiActuator(url).actuate(binding, {}) + assert res.status is ActuationStatus.HALT + assert res.http_status == 401 + assert res.should_halt is True + finally: + stop()