From 3c2e621b3b582cb6dcfe371c6212098484edccd8 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Sat, 18 Jul 2026 21:12:32 -0400 Subject: [PATCH 1/4] Add governed attended actions --- docs/EXCEPTION_INBOX.md | 84 +- openadapt_flow/__main__.py | 250 +++- openadapt_flow/console/__init__.py | 7 +- openadapt_flow/console/app.py | 82 +- openadapt_flow/console/attention.py | 17 +- openadapt_flow/console/server.py | 5 + openadapt_flow/console/static/console.js | 75 +- openadapt_flow/runtime/durable/__init__.py | 24 + openadapt_flow/runtime/durable/attended.py | 1345 ++++++++++++++++++ openadapt_flow/runtime/durable/checkpoint.py | 4 + openadapt_flow/runtime/durable/controller.py | 100 +- openadapt_flow/runtime/durable/resume.py | 4 + openadapt_flow/runtime/replayer.py | 298 +++- tests/test_attended_actions.py | 1131 +++++++++++++++ tests/test_console.py | 198 ++- 15 files changed, 3451 insertions(+), 173 deletions(-) create mode 100644 openadapt_flow/runtime/durable/attended.py create mode 100644 tests/test_attended_actions.py diff --git a/docs/EXCEPTION_INBOX.md b/docs/EXCEPTION_INBOX.md index 7c26708d..7468a40c 100644 --- a/docs/EXCEPTION_INBOX.md +++ b/docs/EXCEPTION_INBOX.md @@ -1,6 +1,6 @@ # Local Needs Attention Queue -The staff-attended queue is the local review surface for workflows that halt +The staff-attended queue is the local operator surface for workflows that halt instead of guessing: ```bash @@ -8,9 +8,62 @@ pip install 'openadapt-flow[console]' openadapt-flow console --runs ./runs --bundles ./bundles --attend ``` -`--attend` opens the Needs Attention view first and is always read-only. -`--allow-actions` is ignored for this mode. The attended browser surface cannot -approve, resume, retry, teach, promote, roll back, certify, or run a workflow. +`--attend` opens Needs Attention first and remains read-only by default. To +operate a live browser deployment from the queue: + +```bash +openadapt-flow console \ + --runs ./runs \ + --bundles ./bundles \ + --attend --allow-actions \ + --config deployment.yaml \ + --headed +``` + +The deployment selects the exact browser, Windows, macOS, or RDP target and its +effect-verification/API wiring. A web target must also declare `backend.url` +(or pass `--url`) and remain headed so the staff member and executor share the +same visible session. The console owns that live session until it exits and +closes it afterward. + +An engine-issued pause exposes only the governed decisions supported by that +step's semantics: + +- **I fixed it — Continue:** observe fresh live state, verify the paused + postconditions and independent effects, prove the expected next target and + armed identity, checkpoint the human-completed step without actuating it, + then resume after it. +- **Skip / disposition:** resume only when the workflow already declares a + non-consequential `on_unmet: skip` path and its guard is currently unmet. + Otherwise the decision remains a recorded non-success disposition. +- **Teach the fix:** enter the existing governed teach/revision path. Regression + and identity/effect/risk gates report accepted, banked progress, or refused; + identity-evidence changes are never silently promoted. +- **Needs more time:** record an escalation while preserving the durable pause. + +Normal console and CLI capabilities remain available. Attended mode is an +additional operator workflow, not a reduced product mode. + +## Action contract + +Every attended mutation is bound to an engine-issued capability covering: + +- a random run-instance identity; +- the exact bundle revision and workflow; +- the paused step/state and verified resume point; +- the expected next transition; +- keyed digests of any pre-human URL/title plus the numeric browser-page count + needed to prove a login/CAPTCHA redirect, never the raw URL or title; +- the exact semantically permitted actions for that paused step; +- an expiry and canonical capability digest. + +An atomic single-flight lease and idempotency key serialize decisions. The +audit journal persists `prepared` before work and `delivery_started` immediately +before live verification/resume. A crash after that point records +`delivery_uncertain`; automatic retry is refused until a person reconciles the +live application and system of record. A successful Continue resumes from the +new verified checkpoint, so neither prior confirmed work nor the +human-completed step is actuated again. ## Security boundary @@ -19,27 +72,26 @@ approve, resume, retry, teach, promote, roll back, certify, or run a workflow. capability. - Browser mutations also require a valid Host, same-origin request, `application/json`, and the session CSRF token. +- The authenticated local OS account is recorded as the operator. - Queue records use opaque IDs and category-derived copy. Workflow names, parameters, raw halt reasons, observed text, local paths, and raw reports do not enter the DTO. - Artifact lookup accepts only report-referenced PNG IDs and never follows a symlink. - The notification endpoint returns only a count and `#/attention`; it is the - PHI-free seam for a future desktop/tray OS notification. Flow does not pop a - system notification or inject input. + PHI-free seam for a desktop/tray OS notification. ## Human-required interruptions CAPTCHA, MFA, one-time codes, and expired authentication can appear as `human_required` halts. OpenAdapt never answers, solves, clicks, retries, or -sends those challenges to a model. The person at the workstation may complete -the challenge in the live application, but the queue only displays redacted -local evidence. It does not record approval or continue the run. - -## Deliberately deferred +sends those challenges to a model. The person at the workstation completes the +challenge in the live application. The action payload accepts no answer, code, +or raw local path. -The first secure slice does not approve or resume runs, collect free-text staff -notes, automate teaching, upload local evidence, send cloud notifications, or -implement CAPTCHA/MFA solvers. Approval and resume require an immutable -pause/action-delivery binding and single-flight execution semantics before they -can be added to an attended browser surface. +Afterward, Continue treats action delivery and outcome verification as separate +facts. A common URL/title change or newly opened tab can be confirmed against +the signed PHI-safe pause baseline. Unverifiable effects, relative +postconditions without that baseline, ambiguous targets, identity mismatch, +expired capabilities, changed bundles, stale pages, and uncertain prior +delivery all refuse rather than report false success. diff --git a/openadapt_flow/__main__.py b/openadapt_flow/__main__.py index a54a3304..4358b9c3 100644 --- a/openadapt_flow/__main__.py +++ b/openadapt_flow/__main__.py @@ -40,8 +40,9 @@ import argparse import sys +from contextlib import contextmanager from pathlib import Path -from typing import TYPE_CHECKING, Optional, Sequence +from typing import TYPE_CHECKING, Any, Iterator, Optional, Sequence from urllib.parse import urlsplit if TYPE_CHECKING: # pragma: no cover @@ -179,27 +180,9 @@ def _resolve_worklists( return worklists -def _deployment_runtime(args: argparse.Namespace, params: dict[str, str] | None = None): - """Resolve the deployment wiring for a replay/run from ``--config`` + flags. - - Returns ``(cfg, effect_verifier, api_actuator, durable, allow_egress)``. - A ``--config`` deployment YAML supplies the full surface (records paths, - FHIR search params, ...); direct flags override the common fields. With - neither, everything is default: no verifier, no actuator, non-durable, and - egress only if ``--allow-model-grounding`` was passed (fully back-compatible). - - ``params`` (the governed ``--params-file`` / ``--param`` values) binds an - effect-verifier config's explicit ``{param: ...}`` references - (``effects.path_params`` / ``search_param_exprs`` / ``sql_query_params``) - at construction — see ``docs/EFFECT_KIT.md``. A config with no references - ignores it. - """ - from openadapt_flow.deployment import ( - DeploymentConfig, - build_api_actuator, - build_effect_verifier, - load_deployment, - ) +def _deployment_sections(args: argparse.Namespace): + """Snapshot deployment config plus direct effects/actuation overrides.""" + from openadapt_flow.deployment import DeploymentConfig, load_deployment cfg = ( load_deployment(args.config) @@ -222,7 +205,27 @@ def _deployment_runtime(args: argparse.Namespace, params: dict[str, str] | None ) elif getattr(args, "api_actuator", False): actuation = actuation.model_copy(update={"api": True}) + return cfg, effects, actuation + +def _deployment_runtime(args: argparse.Namespace, params: dict[str, str] | None = None): + """Resolve the deployment wiring for a replay/run from ``--config`` + flags. + + Returns ``(cfg, effect_verifier, api_actuator, durable, allow_egress)``. + A ``--config`` deployment YAML supplies the full surface (records paths, + FHIR search params, ...); direct flags override the common fields. With + neither, everything is default: no verifier, no actuator, non-durable, and + egress only if ``--allow-model-grounding`` was passed (fully back-compatible). + + ``params`` (the governed ``--params-file`` / ``--param`` values) binds an + effect-verifier config's explicit ``{param: ...}`` references + (``effects.path_params`` / ``search_param_exprs`` / ``sql_query_params``) + at construction — see ``docs/EFFECT_KIT.md``. A config with no references + ignores it. + """ + from openadapt_flow.deployment import build_api_actuator, build_effect_verifier + + cfg, effects, actuation = _deployment_sections(args) try: effect_verifier = build_effect_verifier(effects, params=params) api_actuator = build_api_actuator(actuation) @@ -271,32 +274,22 @@ def _resolve_backend_config(args: argparse.Namespace, cfg): return backend -def _build_and_run_replayer( +def _configured_replayer( backend, *, - workflow, - params: dict[str, str], - worklists: dict[str, list[dict[str, str]]], - bundle: Path, - run_dir: Path, - save_healed_to: Optional[Path], allow_egress: bool, effect_verifier, api_actuator, durable: bool, use_structural: bool, governed_authorization=None, - execution_origin: Optional[str] = None, - execution_entry_url: Optional[str] = None, ): - """Wire the grounding / identity-VLM ladder and run the replayer. + """Wire the grounding, verification, and actuation layers into a Replayer. Backend-agnostic: the on-prem VLM appliance (opt-in, egress-guarded), the OCR grounding rung, and the deployment wiring (effect verifier / API - actuator / durable runtime) are identical whether the backend is the - browser, the Windows agent, or an RDP/remote-display session. Returns the - run report. Extracted verbatim from the historical inline web path so the - web behavior is unchanged and every backend shares one code path. + actuator / durable runtime) are identical for browser, Windows, macOS, and + RDP/remote-display sessions. The caller owns the backend lifecycle. """ import os @@ -351,6 +344,36 @@ def _build_and_run_replayer( durable=durable, use_structural=use_structural, governed_authorization=governed_authorization, + ) + + +def _build_and_run_replayer( + backend, + *, + workflow, + params: dict[str, str], + worklists: dict[str, list[dict[str, str]]], + bundle: Path, + run_dir: Path, + save_healed_to: Optional[Path], + allow_egress: bool, + effect_verifier, + api_actuator, + durable: bool, + use_structural: bool, + governed_authorization=None, + execution_origin: Optional[str] = None, + execution_entry_url: Optional[str] = None, +): + """Build the shared Replayer configuration and execute one workflow.""" + return _configured_replayer( + backend, + allow_egress=allow_egress, + effect_verifier=effect_verifier, + api_actuator=api_actuator, + durable=durable, + use_structural=use_structural, + governed_authorization=governed_authorization, ).run( workflow, params=params, @@ -2698,15 +2721,146 @@ def build_parser() -> argparse.ArgumentParser: "--attend", action="store_true", help=( - "Open the authenticated local Needs Attention queue in a hard " - "read-only mode. --allow-actions is ignored when --attend is set" + "Open the authenticated local Needs Attention queue first. It is " + "read-only unless --allow-actions is also supplied; attended " + "mutations additionally require an engine-issued pause capability " + "and qualified deployment executor" + ), + ) + p.add_argument( + "--url", + default=None, + help=( + "Live browser URL for attended verification/continuation. Required " + "for --backend web unless backend.url is set in --config." + ), + ) + p.add_argument( + "--headed", + action="store_true", + help=( + "Keep the attended browser visible for the operator. The web " + "attended-action path requires a headed session." + ), + ) + p.add_argument( + "--allow-model-grounding", + action="store_true", + help=( + "EGRESS OPT-IN: permit configured off-box model components during " + "fresh attended verification and deterministic continuation" ), ) + _add_backend_flags(p) + _add_deployment_flags(p) p.set_defaults(func=_cmd_console) return parser +@contextmanager +def _attended_executor_from_args(args: argparse.Namespace) -> Iterator[Any]: + """Own one qualified live backend for the blocking attended console. + + Ordinary console modes do not inspect deployment flags or construct a + backend. With both ``--attend`` and ``--allow-actions``, the operator must + explicitly select a deployment/backend. The same browser, desktop, or + remote-display session is then reused for fresh verification and the + deterministic continuation, and is closed exactly once when the console + exits. + """ + if not (args.attend and args.allow_actions): + yield None + return + if not (getattr(args, "config", None) or getattr(args, "backend", None)): + raise SystemExit( + "console --attend --allow-actions requires an explicit --config " + "or --backend target; refusing to attach mutations to a demo or " + "implicit default" + ) + + from openadapt_flow.backends.factory import _normalize_kind, build_backend + from openadapt_flow.deployment import build_api_actuator, build_effect_verifier + from openadapt_flow.runtime.durable.attended import ( + AttendedActionRefused, + BoundAttendedExecutor, + ) + + try: + cfg, effects_cfg, actuation_cfg = _deployment_sections(args) + backend_cfg = _resolve_backend_config(args, cfg) + except (FileNotFoundError, ValueError) as exc: + raise SystemExit(str(exc)) from exc + allow_egress = bool( + cfg.runtime.allow_model_grounding + or getattr(args, "allow_model_grounding", False) + ) + + def _executor(backend: Any) -> BoundAttendedExecutor: + def _replayer_for_manifest(manifest: Any): + try: + effect_verifier = build_effect_verifier( + effects_cfg, params=dict(manifest.params) + ) + api_actuator = build_api_actuator(actuation_cfg) + except ValueError as exc: + raise AttendedActionRefused(str(exc)) from exc + return _configured_replayer( + backend, + allow_egress=allow_egress, + effect_verifier=effect_verifier, + api_actuator=api_actuator, + durable=True, + use_structural=True, + governed_authorization=manifest.governed_authorization, + ) + + return BoundAttendedExecutor(_replayer_for_manifest) + + if _normalize_kind(backend_cfg.kind) == "web": + target_url = getattr(args, "url", None) or backend_cfg.url + if not target_url: + raise SystemExit( + "attended web actions require --url or backend.url in --config" + ) + headed = bool(getattr(args, "headed", False) or backend_cfg.headed) + if not headed: + raise SystemExit( + "attended web actions require a visible live session; pass " + "--headed or set backend.headed: true" + ) + + from playwright.sync_api import sync_playwright + + from openadapt_flow._browser_setup import ensure_chromium_installed + + ensure_chromium_installed() + with sync_playwright() as playwright: + browser = playwright.chromium.launch(headless=False) + try: + page = browser.new_page(viewport=_VIEWPORT) + page.goto(target_url) + try: + backend = build_backend(backend_cfg, page=page) + except ValueError as exc: + raise SystemExit(str(exc)) from exc + yield _executor(backend) + finally: + browser.close() + return + + try: + backend = build_backend(backend_cfg) + except ValueError as exc: + raise SystemExit(str(exc)) from exc + try: + yield _executor(backend) + finally: + close = getattr(backend, "close", None) + if callable(close): + close() + + def _cmd_console(args: argparse.Namespace) -> int: # find_spec first so "extra not installed" is distinguishable from "the # console package itself is broken" (a wiring bug must surface, not hide @@ -2721,7 +2875,7 @@ def _cmd_console(args: argparse.Namespace) -> int: ) from openadapt_flow.console.server import LOOPBACK_HOST, serve - allow_actions = args.allow_actions and not args.attend + allow_actions = args.allow_actions mode = "ACTIONS ENABLED (confirm-gated)" if allow_actions else "read-only" if args.attend: mode += "; attention-first" @@ -2730,14 +2884,16 @@ def _cmd_console(args: argparse.Namespace) -> int: f" bundles: {Path(args.bundles).resolve()}\n" f" runs: {Path(args.runs).resolve()}" ) - serve( - args.bundles, - args.runs, - args.skills, - allow_actions=allow_actions, - attend=args.attend, - port=args.port, - ) + with _attended_executor_from_args(args) as attended_executor: + serve( + args.bundles, + args.runs, + args.skills, + allow_actions=allow_actions, + attend=args.attend, + attended_executor=attended_executor, + port=args.port, + ) return 0 diff --git a/openadapt_flow/console/__init__.py b/openadapt_flow/console/__init__.py index 0903c856..0cceb4e0 100644 --- a/openadapt_flow/console/__init__.py +++ b/openadapt_flow/console/__init__.py @@ -16,9 +16,10 @@ refuse unless the operator opted in with ``--allow-actions``. APIs and artifacts require an unguessable fragment-delivered bearer capability; mutations additionally require same-origin JSON and a session CSRF token. -- ``console --attend`` opens the redacted Needs Attention queue under a hard - read-only capability profile. It does not expose or execute governance - actions, even if ``--allow-actions`` is also supplied. +- ``console --attend`` opens the redacted Needs Attention queue first. It + remains read-only by default; explicit action enablement still requires an + exact engine-issued pause capability and a deployment-bound executor before + an attended decision can cross a delivery boundary. Serve it with ``openadapt-flow console`` (requires the ``console`` extra: ``pip install 'openadapt-flow[console]'``). diff --git a/openadapt_flow/console/app.py b/openadapt_flow/console/app.py index f5081c39..e18ea8f2 100644 --- a/openadapt_flow/console/app.py +++ b/openadapt_flow/console/app.py @@ -6,8 +6,9 @@ mutations additionally require same-origin JSON and a session-bound CSRF token. With ``allow_actions=False`` (the default), mutating endpoints refuse with a browser-safe placeholder command the operator can copy. ``attend=True`` -is a hard read-only profile: it suppresses action catalogs and refuses every -generic mutation route even if ``allow_actions=True`` was also requested. +makes the secure Needs Attention queue the first view; it does not weaken or +remove the normal console. Attended mutations additionally require an exact +engine-issued pause capability and a deployment-bound executor. Browser DTOs use opaque ids and explicit projections; protected workflow labels, parameters, identity evidence, local paths, and raw reports do not @@ -34,6 +35,13 @@ from openadapt_flow.console import actions as actions_mod from openadapt_flow.console import attention as attention_mod from openadapt_flow.console import data +from openadapt_flow.runtime.durable.approval import ResumeRefused +from openadapt_flow.runtime.durable.attended import ( + AttendedActionExecutor, + AttendedActionRefused, + AttendedActionRequest, + execute_attended_action, +) _STATIC_DIR = Path(__file__).parent / "static" _MAX_BODY_BYTES = 16 * 1024 @@ -271,11 +279,8 @@ def create_app( attend: bool = False, access_token: Optional[str] = None, csrf_token: Optional[str] = None, + attended_executor: Optional[AttendedActionExecutor] = None, ) -> FastAPI: - # ``attend`` is a deliberately narrow read-only capability profile. Keep - # this defense in the app factory (not only the CLI) so embedders and tests - # cannot combine attend=True with a mutation-enabled server. - allow_actions = allow_actions and not attend bundles = _validated_root(bundles_root, label="bundles") runs = _validated_root(runs_root, label="runs") skills = _validated_root(skills_root, label="skills") if skills_root else None @@ -381,6 +386,10 @@ def session() -> dict[str, Any]: "version": __version__, "read_only": not allow_actions, "attend": attend, + "attended_decisions_ready": bool(attend and allow_actions), + "attended_actions_ready": bool( + attend and allow_actions and attended_executor is not None + ), "csrf_token": csrf, } @@ -391,6 +400,10 @@ def health() -> dict[str, Any]: "version": __version__, "read_only": not allow_actions, "attend": attend, + "attended_decisions_ready": bool(attend and allow_actions), + "attended_actions_ready": bool( + attend and allow_actions and attended_executor is not None + ), } # -- workflows ---------------------------------------------------------- @@ -415,8 +428,6 @@ def workflow_diff(bundle_id: str, other_id: str) -> dict[str, Any]: @app.get("/api/workflows/{bundle_id:path}/actions") def workflow_actions(bundle_id: str) -> list[dict[str, Any]]: - if attend: - return [] path = _resolve_bundle(bundles, bundle_id) summary = data.bundle_summary(bundles, path) return [ @@ -461,8 +472,6 @@ def run_artifact( @app.get("/api/runs/{run_id:path}/actions") def run_actions(run_id: str) -> list[dict[str, Any]]: - if attend: - return [] run_dir = _resolve_run(runs, run_id) summary = data.run_summary(runs, run_dir) bundle = _guess_bundle_for_run(bundles, runs, run_dir) @@ -503,6 +512,44 @@ def attention_detail(run_id: str) -> dict[str, Any]: raise HTTPException(status_code=404, detail="no open attention item") return item.model_dump() + @app.post("/api/attention/{run_id:path}/actions/{action_id}") + def execute_attention_action( + run_id: str, + action_id: str, + payload: AttendedActionRequest, + ) -> dict[str, Any]: + if not attend: + raise HTTPException(status_code=404, detail="no such action") + _refuse_or_none( + allow_actions, + "start `openadapt-flow console --attend --allow-actions` with " + "a qualified deployment executor", + ) + if action_id != payload.action: + raise HTTPException( + status_code=400, + detail="action path and capability request do not match", + ) + run_dir = _resolve_run(runs, run_id) + try: + decision = execute_attended_action( + run_dir, + payload, + operator=operator, + executor=attended_executor, + ) + except (AttendedActionRefused, ResumeRefused) as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + # The durable audit keeps operator, request, capability, and transition + # bindings. The browser needs only the decision outcome; never reflect + # those protected identifiers back across the presentation boundary. + return { + "action": decision.action, + "status": decision.status, + "message": decision.message, + "report_success": decision.report_success, + } + # -- skills ------------------------------------------------------------- @app.get("/api/skills") @@ -518,11 +565,6 @@ def skills_list() -> list[dict[str, Any]]: def execute_run_action( run_id: str, action_id: str, payload: dict[str, Any] | None = None ) -> JSONResponse: - if attend: - raise HTTPException( - status_code=404, - detail="actions are unavailable in attended read-only mode", - ) run_dir = _resolve_run(runs, run_id) summary = data.run_summary(runs, run_dir) bundle = _guess_bundle_for_run(bundles, runs, run_dir) @@ -570,11 +612,6 @@ def execute_run_action( def execute_bundle_action( bundle_id: str, action_id: str, payload: dict[str, Any] | None = None ) -> JSONResponse: - if attend: - raise HTTPException( - status_code=404, - detail="actions are unavailable in attended read-only mode", - ) path = _resolve_bundle(bundles, bundle_id) summary = data.bundle_summary(bundles, path) specs = { @@ -622,11 +659,6 @@ def execute_bundle_action( def execute_skill_action( skill_id: str, action_id: str, payload: dict[str, Any] ) -> JSONResponse: - if attend: - raise HTTPException( - status_code=404, - detail="actions are unavailable in attended read-only mode", - ) library_id = payload.get("library") version = payload.get("version") if not isinstance(library_id, str) or not isinstance(version, int): diff --git a/openadapt_flow/console/attention.py b/openadapt_flow/console/attention.py index 419b8ef6..31b05a94 100644 --- a/openadapt_flow/console/attention.py +++ b/openadapt_flow/console/attention.py @@ -6,10 +6,9 @@ and existing opaque screenshot ids. Protected screenshots remain behind the console's authenticated, symlink-safe artifact endpoint. -This module is projection-only. The attended surface does not approve, resume, -retry, solve, click, type, or otherwise mutate a run. Human-intervention and -durable-resume semantics remain deployment-bound CLI/runtime concerns until -they can be tied to an immutable pause and explicit action-delivery phase. +This module remains projection-only. Engine-owned attended mutations live in +``runtime.durable.attended`` and appear only when an exact pause capability and +deployment-bound executor are present. """ from __future__ import annotations @@ -20,6 +19,7 @@ from pydantic import BaseModel from openadapt_flow.console import data +from openadapt_flow.runtime.durable.attended import attended_capability_summary from openadapt_flow.runtime.durable.controller import looks_like_human_required _KNOWN_CATEGORIES = { @@ -41,7 +41,8 @@ "human_required": ( "The application needs a person before the workflow can continue.", "Complete the challenge, sign-in, or verification in the live " - "application. OpenAdapt never answers it; this queue remains read-only.", + "application. OpenAdapt never answers it; Continue only verifies the " + "resulting live state before deterministic resume.", ), "identity": ( "The target identity could not be certified.", @@ -109,6 +110,7 @@ class AttentionItem(BaseModel): completed_intent_count: int = 0 before_artifact_id: Optional[str] = None after_artifact_id: Optional[str] = None + capability: Optional[dict[str, Any]] = None class AttentionNotification(BaseModel): @@ -124,8 +126,8 @@ def _normal_category(raw: Any, *protected_texts: Any) -> str: value = raw if isinstance(raw, str) else "" # Preserve a runtime-produced typed category. An observed phrase such as # "session expired" must not re-label an effect/identity/postcondition halt. - # The queue is read-only, so an imported/tampered category can at worst - # affect fixed explanatory copy; it cannot authorize an action. + # A category never authorizes an action. Mutations require a separately + # signed engine capability tied to the exact pause. if value in _KNOWN_CATEGORIES: return value marker_present = looks_like_human_required( @@ -233,6 +235,7 @@ def attention_item(root: Path, path: Path) -> Optional[AttentionItem]: completed_intent_count=completed_count, before_artifact_id=before_id, after_artifact_id=after_id, + capability=attended_capability_summary(path), ) diff --git a/openadapt_flow/console/server.py b/openadapt_flow/console/server.py index f08d565a..74c44642 100644 --- a/openadapt_flow/console/server.py +++ b/openadapt_flow/console/server.py @@ -9,6 +9,9 @@ import secrets from pathlib import Path +from typing import Optional + +from openadapt_flow.runtime.durable.attended import AttendedActionExecutor #: The only address the console ever binds. Not configurable. LOOPBACK_HOST = "127.0.0.1" @@ -23,6 +26,7 @@ def serve( *, allow_actions: bool = False, attend: bool = False, + attended_executor: Optional[AttendedActionExecutor] = None, port: int = DEFAULT_PORT, ) -> None: """Build the app and serve it on ``http://127.0.0.1:`` (blocking).""" @@ -38,6 +42,7 @@ def serve( allow_actions=allow_actions, attend=attend, access_token=access_token, + attended_executor=attended_executor, ) # URL fragments are consumed entirely by the browser and are never sent in # HTTP requests or uvicorn access logs. The UI removes the fragment before diff --git a/openadapt_flow/console/static/console.js b/openadapt_flow/console/static/console.js index 1ffa46ac..c404f6a4 100644 --- a/openadapt_flow/console/static/console.js +++ b/openadapt_flow/console/static/console.js @@ -284,9 +284,10 @@ async function viewAttention() { } return `

Needs Attention ${safeNumber(items.length, "0")}

Protected values and paths stay in the local run artifacts. - This attended view is read-only; it never approves or resumes a run. Only the person at this computer should complete CAPTCHA, MFA, or sign-in - in the live application.

+ in the live application. ${HEALTH.read_only + ? "Start with explicit action enablement to make governed decisions." + : "Continue verifies live outcomes before deterministic resume; it never answers the challenge."}

${items.map((item) => `
@@ -305,10 +306,71 @@ async function viewAttention() {

Review protected evidence + ${HEALTH.attended_decisions_ready && item.capability ? ` + ${HEALTH.attended_actions_ready && + item.capability.allowed_actions.includes("continue") ? ` + ` : ""} + ${HEALTH.attended_actions_ready && + item.capability.allowed_actions.includes("skip") ? ` + ` : ""} + ${item.capability.allowed_actions.includes("teach") ? ` + ` : ""} + ${item.capability.allowed_actions.includes("escalate") ? ` + ` : ""} + ` : ""}
+

`).join("")}
`; } +async function attendedAction(button) { + const card = button.closest(".attention-card"); + const output = card ? card.querySelector(".attention-result") : null; + const action = button.dataset.attendedAction; + const runId = button.dataset.runId; + const capability = button.dataset.capability; + if (!action || !runId || !capability) return; + const siblings = card ? [...card.querySelectorAll("[data-attended-action]")] : [button]; + siblings.forEach((item) => { item.disabled = true; }); + if (output) output.textContent = "Submitting governed decision…"; + const disposition = { + continue: "completed_by_operator", + skip: "not_applicable", + teach: "teach_requested", + escalate: "needs_assistance", + }[action]; + try { + const result = await api( + `/api/attention/${enc(runId)}/actions/${enc(action)}`, + { + method: "POST", + headers: {"Content-Type": "application/json"}, + body: JSON.stringify({ + capability_digest: capability, + idempotency_key: crypto.randomUUID().replaceAll("-", ""), + action, + disposition, + }), + }, + ); + if (output) output.textContent = result.message || result.status; + if (result.status === "completed" || result.status === "halted") { + await route(); + } + } catch (error) { + const detail = error.body && error.body.detail; + if (output) output.textContent = typeof detail === "string" + ? detail + : "Decision refused; reload the current pause and review live state."; + } finally { + siblings.forEach((item) => { item.disabled = false; }); + } +} + function shot(runId, artifactId, caption) { if (!artifactId) return ""; const artifactUrl = `/api/runs/${enc(runId)}/artifact?id=${enc(artifactId)}`; @@ -417,7 +479,7 @@ async function viewSkills() { VersionStatusScoreProvenanceNoteActions ${skill.versions.map((version) => { const actions = []; - if (!HEALTH.attend && version.status === "candidate") { + if (version.status === "candidate") { const index = currentSkillActions.push({ library: library.id, skillId: skill.id, @@ -426,7 +488,7 @@ async function viewSkills() { }) - 1; actions.push(``); } - if (!HEALTH.attend && version.status !== "rolled_back") { + if (version.status !== "rolled_back") { const index = currentSkillActions.push({ library: library.id, skillId: skill.id, @@ -636,6 +698,11 @@ document.addEventListener("click", async (event) => { if (spec) await skillAction(spec); return; } + const attendedTarget = event.target.closest("[data-attended-action]"); + if (attendedTarget) { + await attendedAction(attendedTarget); + return; + } }); $("#modal-cancel").addEventListener("click", closeModal); diff --git a/openadapt_flow/runtime/durable/__init__.py b/openadapt_flow/runtime/durable/__init__.py index 98d6830c..412201a8 100644 --- a/openadapt_flow/runtime/durable/__init__.py +++ b/openadapt_flow/runtime/durable/__init__.py @@ -34,6 +34,19 @@ StateDiverged, enforce_resume_authorization, ) +from openadapt_flow.runtime.durable.attended import ( # noqa: F401 + AttendedActionRefused, + AttendedActionRequest, + AttendedActionStore, + AttendedDecision, + AttendedExecutionResult, + AttendedPauseCapability, + BoundAttendedExecutor, + SignedTransitionBaseline, + TransitionObservation, + execute_attended_action, + issue_attended_capability, +) from openadapt_flow.runtime.durable.checkpoint import ( # noqa: F401 CheckpointStore, PendingEscalation, @@ -73,6 +86,17 @@ "PauseExpired", "ResumeRefused", "StateDiverged", + "AttendedActionRefused", + "AttendedActionRequest", + "AttendedActionStore", + "AttendedDecision", + "AttendedExecutionResult", + "AttendedPauseCapability", + "BoundAttendedExecutor", + "SignedTransitionBaseline", + "TransitionObservation", + "execute_attended_action", + "issue_attended_capability", "enforce_resume_authorization", "DurableRun", "classify_halt", diff --git a/openadapt_flow/runtime/durable/attended.py b/openadapt_flow/runtime/durable/attended.py new file mode 100644 index 00000000..18c0f386 --- /dev/null +++ b/openadapt_flow/runtime/durable/attended.py @@ -0,0 +1,1345 @@ +"""Governed actions for a durably paused, staff-attended run. + +The browser console is only a presentation surface. This module owns the +engine contract that makes an attended action safe: + +* an engine-issued capability is bound to the exact run, bundle revision, + paused step/state, resume point, and expected next transition; +* an authenticated operator must present that exact capability before acting; +* one filesystem lease serializes decisions and one idempotency key makes a + retried HTTP request replay its recorded result instead of acting twice; +* a human-completed step is *observed and verified*, never actuated again; +* delivery evidence is never accepted as outcome evidence; +* every accepted, refused, deferred, and escalated decision is auditable. + +CAPTCHA, MFA, re-authentication, and other human-presence challenges are +deliberately outside the automation path. A person completes them in the live +application and then asks OpenAdapt to verify the resulting state. +""" + +from __future__ import annotations + +import hashlib +import hmac +import json +import os +import secrets +import threading +from contextlib import contextmanager +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any, Callable, Iterator, Literal, Optional, Protocol + +from pydantic import BaseModel, ConfigDict, Field + +from openadapt_flow.ir import StepResult, Workflow +from openadapt_flow.runtime.durable.approval import ( + ApprovalRecord, + ApprovalRequired, + BundleMismatch, + PauseExpired, + ResumeRefused, + pause_is_expired, +) +from openadapt_flow.runtime.durable.checkpoint import ( + CheckpointStore, + PendingEscalation, + RunCheckpoint, +) +from openadapt_flow.runtime.durable.program_checkpoint import bundle_version + +CAPABILITY_FILENAME = "attended_capability.json" +CAPABILITY_HISTORY_FILENAME = "attended_capability_history.json" +CAPABILITY_KEY_FILENAME = ".attended_capability.key" +DECISIONS_FILENAME = "attended_decisions.json" +LEASE_FILENAME = ".attended_action.lease" +DEFAULT_CAPABILITY_TTL_S = 24 * 3600.0 +DEFAULT_LEASE_TTL_S = 15 * 60.0 + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +def _iso(value: datetime) -> str: + return value.astimezone(timezone.utc).isoformat() + + +def _parse(value: str) -> datetime: + parsed = datetime.fromisoformat(value) + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) + + +def _canonical(payload: Any) -> bytes: + if isinstance(payload, BaseModel): + payload = payload.model_dump(mode="json") + return json.dumps( + payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False + ).encode("utf-8") + + +def _digest(payload: Any) -> str: + return "sha256:" + hashlib.sha256(_canonical(payload)).hexdigest() + + +class AttendedActionRefused(ResumeRefused): + """An attended mutation was refused before any workflow actuation.""" + + +class AttendedActionBusy(AttendedActionRefused): + """Another operator request currently owns the run's single-flight lease.""" + + +class AttendedActionExecutor(Protocol): + """Deployment-bound bridge used only after the engine admits a decision.""" + + def continue_run( + self, + run_dir: Path, + capability: "AttendedPauseCapability", + approval: ApprovalRecord, + ) -> "AttendedExecutionResult": + """Verify the human-completed outcome and resume deterministically.""" + + def skip_run( + self, + run_dir: Path, + capability: "AttendedPauseCapability", + approval: ApprovalRecord, + ) -> "AttendedExecutionResult": + """Apply declared skip semantics and resume, or refuse.""" + + +class TransitionObservation(BaseModel): + """Ephemeral pre-human browser state; never serialized to the run.""" + + model_config = ConfigDict(extra="forbid") + + url: Optional[str] = None + page_title: Optional[str] = None + page_count: Optional[int] = Field(default=None, ge=0) + + +class SignedTransitionBaseline(BaseModel): + """PHI-safe structural baseline bound into a signed pause capability.""" + + schema_version: int = 1 + url_digest: Optional[str] = None + title_digest: Optional[str] = None + page_count: Optional[int] = Field(default=None, ge=0) + + +class AttendedPauseCapability(BaseModel): + """Exact authority the engine grants for one durable pause.""" + + schema_version: int = 1 + pause_id: str + run_id: str + workflow_name: str + bundle_version: str + step_index: int + step_id: str + state_id: Optional[str] = None + resume_from_index: int + resume_from_step_id: Optional[str] = None + pause_digest: str + expected_next_transition: Optional[str] = None + expected_transition_digest: str + transition_baseline: SignedTransitionBaseline = Field( + default_factory=SignedTransitionBaseline + ) + delivery_state: Literal["not_delivered", "delivered", "unknown"] = "unknown" + issued_at: str + expires_at: str + allowed_actions: tuple[Literal["continue", "skip", "teach", "escalate"], ...] = ( + "teach", + "escalate", + ) + signature: str = "" + + def unsigned(self) -> dict[str, Any]: + return self.model_dump(exclude={"signature"}, mode="json") + + @property + def digest(self) -> str: + """Public, non-authorizing fingerprint used for stale-UI binding.""" + return _digest(self.unsigned()) + + +class AttendedActionRequest(BaseModel): + """One browser decision, bound to a capability and retry key.""" + + model_config = ConfigDict(extra="forbid") + + capability_digest: str = Field(pattern=r"^sha256:[0-9a-f]{64}$") + idempotency_key: str = Field( + min_length=16, + max_length=200, + pattern=r"^[A-Za-z0-9._:-]+$", + ) + action: Literal["continue", "skip", "teach", "escalate"] + disposition: Optional[ + Literal[ + "completed_by_operator", + "not_applicable", + "cannot_complete", + "needs_assistance", + "teach_requested", + ] + ] = None + + +class AttendedExecutionResult(BaseModel): + """Outcome returned by a deployment-bound continue/skip executor.""" + + status: Literal["completed", "refused", "halted"] + message: str + report_success: Optional[bool] = None + resumed_from: Optional[str] = None + next_transition: Optional[str] = None + + +class AttendedDecision(BaseModel): + """Append-only audit record for an admitted or refused operator decision.""" + + schema_version: int = 1 + decision_id: str = Field(default_factory=lambda: secrets.token_hex(16)) + pause_id: str + capability_digest: str + request_digest: str + idempotency_key: str + action: Literal["continue", "skip", "teach", "escalate"] + operator: str + disposition: Optional[str] = None + status: Literal[ + "prepared", + "delivery_started", + "delivery_uncertain", + "completed", + "refused", + "halted", + "needs_demonstration", + "escalated", + ] + message: str + created_at: str = Field(default_factory=lambda: _iso(_now())) + report_success: Optional[bool] = None + next_transition: Optional[str] = None + + +class AttendedDecisionLog(BaseModel): + schema_version: int = 1 + decisions: list[AttendedDecision] = Field(default_factory=list) + + +def _delivery_state( + result: StepResult, +) -> Literal["not_delivered", "delivered", "unknown"]: + """Classify delivery without ever implying outcome success.""" + error = (result.error or "").lower() + if result.delivery_receipt is not None or result.actuation == "api": + return "delivered" + if ( + result.identity is not None + or "could not resolve" in error + or "refusing to act" in error + or "precondition" in error + or "guard" in error + ): + return "not_delivered" + return "unknown" + + +def _expected_transition( + workflow: Workflow, pending: PendingEscalation +) -> Optional[str]: + if pending.program: + # The complete interpreter frame is held in the last program + # checkpoint. The paused state is the only transition identity the + # controller can safely name without re-evaluating guarded edges. + return pending.state_id or pending.step_id + next_index = pending.step_index + 1 + if 0 <= next_index < len(workflow.steps): + return workflow.steps[next_index].id + return "" + + +def _transition_payload( + *, + run_id: str, + workflow_name: str, + bundle_revision: str, + pending: PendingEscalation, + expected_next_transition: Optional[str], +) -> dict[str, Any]: + return { + "run_id": run_id, + "workflow_name": workflow_name, + "bundle_version": bundle_revision, + "step_index": pending.step_index, + "step_id": pending.step_id, + "state_id": pending.state_id, + "resume_from_index": pending.resume_from_index, + "resume_from_step_id": pending.resume_from_step_id, + "expected_next_transition": expected_next_transition, + } + + +def _relative_postcondition_kinds(step: Any) -> set[str]: + return { + pc.kind.value if hasattr(pc.kind, "value") else str(pc.kind) + for pc in step.expect + if (pc.kind.value if hasattr(pc.kind, "value") else str(pc.kind)) + in {"url_changed", "title_changed", "new_tab_opened"} + } + + +def _allowed_actions( + workflow: Workflow, + pending: PendingEscalation, + baseline: SignedTransitionBaseline, +) -> tuple[Literal["continue", "skip", "teach", "escalate"], ...]: + """Derive mutation authority from the exact workflow step semantics.""" + actions: list[Literal["continue", "skip", "teach", "escalate"]] = [ + "teach", + "escalate", + ] + if pending.program or not 0 <= pending.step_index < len(workflow.steps): + return tuple(actions) + step = workflow.steps[pending.step_index] + if step.id != pending.step_id: + return tuple(actions) + + relative = _relative_postcondition_kinds(step) + has_relative_baseline = ( + ("url_changed" not in relative or baseline.url_digest is not None) + and ("title_changed" not in relative or baseline.title_digest is not None) + and ("new_tab_opened" not in relative or baseline.page_count is not None) + ) + has_unsupported_effect = any( + effect.needs_operator_confirmation + or effect.count_new_only + or effect.forbid_collateral_loss + for effect in step.effects + ) + if ( + bool(step.expect or step.effects) + and has_relative_baseline + and not has_unsupported_effect + ): + actions.insert(0, "continue") + + if ( + step.risk != "irreversible" + and not step.effects + and step.guard is not None + and step.guard.on_unmet == "skip" + ): + actions.insert(1 if actions[0] == "continue" else 0, "skip") + return tuple(actions) + + +class AttendedActionStore: + """Capability, single-flight lease, and append-only decision persistence.""" + + def __init__(self, run_dir: Path | str) -> None: + self.run_dir = Path(run_dir) + self.capability_path = self.run_dir / CAPABILITY_FILENAME + self.capability_history_path = self.run_dir / CAPABILITY_HISTORY_FILENAME + self.key_path = self.run_dir / CAPABILITY_KEY_FILENAME + self.decisions_path = self.run_dir / DECISIONS_FILENAME + self.lease_path = self.run_dir / LEASE_FILENAME + + @staticmethod + def _fsync_parent(path: Path) -> None: + """Persist a replace/create directory entry on POSIX.""" + if os.name == "nt": + return + fd = os.open(path.parent, os.O_RDONLY) + try: + os.fsync(fd) + finally: + os.close(fd) + + @staticmethod + def _atomic_write(path: Path, payload: bytes, *, mode: int = 0o600) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_name(f".{path.name}.{secrets.token_hex(8)}.tmp") + fd = os.open(tmp, os.O_CREAT | os.O_EXCL | os.O_WRONLY, mode) + try: + with os.fdopen(fd, "wb") as handle: + handle.write(payload) + handle.flush() + os.fsync(handle.fileno()) + os.replace(tmp, path) + AttendedActionStore._fsync_parent(path) + finally: + try: + tmp.unlink() + except FileNotFoundError: + pass + + def _key(self, *, create: bool) -> bytes: + try: + key = self.key_path.read_bytes() + except FileNotFoundError: + if not create: + raise AttendedActionRefused( + "the pause capability key is missing; refusing an " + "unverifiable operator action" + ) from None + self.run_dir.mkdir(parents=True, exist_ok=True) + key = secrets.token_bytes(32) + try: + fd = os.open(self.key_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) + except FileExistsError: + return self._key(create=False) + with os.fdopen(fd, "wb") as handle: + handle.write(key) + handle.flush() + os.fsync(handle.fileno()) + self._fsync_parent(self.key_path) + if os.name != "nt" and self.key_path.stat().st_mode & 0o077: + raise AttendedActionRefused( + "the pause capability key permissions are too broad; refusing" + ) + if len(key) != 32: + raise AttendedActionRefused( + "the pause capability key has an invalid length; refusing" + ) + return key + + def _sign(self, capability: AttendedPauseCapability, *, create_key: bool) -> str: + return ( + "hmac-sha256:" + + hmac.new( + self._key(create=create_key), + _canonical(capability.unsigned()), + hashlib.sha256, + ).hexdigest() + ) + + def transition_value_digest(self, field: str, value: str) -> str: + """Keyed digest for one transient URL/title observation.""" + if field not in {"url", "page_title"}: + raise ValueError("transition digest field must be url or page_title") + payload = f"openadapt-attended-transition-v1:{field}:".encode() + value.encode( + "utf-8" + ) + return ( + "hmac-sha256:" + + hmac.new(self._key(create=False), payload, hashlib.sha256).hexdigest() + ) + + def _transition_baseline( + self, observation: Optional[TransitionObservation] + ) -> SignedTransitionBaseline: + observation = observation or TransitionObservation() + return SignedTransitionBaseline( + url_digest=( + self.transition_value_digest("url", observation.url) + if observation.url is not None + else None + ), + title_digest=( + self.transition_value_digest("page_title", observation.page_title) + if observation.page_title is not None + else None + ), + page_count=observation.page_count, + ) + + def issue( + self, + *, + manifest: Any, + pending: PendingEscalation, + workflow: Workflow, + result: StepResult, + transition_observation: Optional[TransitionObservation] = None, + ttl_s: float = DEFAULT_CAPABILITY_TTL_S, + ) -> AttendedPauseCapability: + """Issue once for a new pause; re-reads an existing valid capability.""" + revision = bundle_version(manifest.bundle_dir) + expected = _expected_transition(workflow, pending) + pause_digest = _digest(pending) + # Creating the HMAC key before digesting transition values gives the + # baseline and capability signature one stable per-run trust root. + self._key(create=True) + baseline = self._transition_baseline(transition_observation) + if self.capability_path.is_file(): + existing = self.read() + if ( + existing.pause_digest == pause_digest + and existing.step_id == pending.step_id + and existing.step_index == pending.step_index + and existing.state_id == pending.state_id + and existing.resume_from_index == pending.resume_from_index + and existing.resume_from_step_id == pending.resume_from_step_id + and existing.expected_next_transition == expected + and existing.bundle_version == revision + and existing.run_id == manifest.run_id + and existing.workflow_name == pending.workflow_name + and existing.transition_baseline == baseline + ): + return existing + # A resumed run may halt again before the first request's terminal + # HTTP response is written. Preserve the old signed capability in + # an append-only history and let the engine issue the new pause; + # browser callers still present the exact current digest. + history: list[dict[str, Any]] = [] + if self.capability_history_path.is_file(): + try: + raw_history = json.loads(self.capability_history_path.read_text()) + if isinstance(raw_history, list): + history = [ + item for item in raw_history if isinstance(item, dict) + ] + except (OSError, ValueError): + raise AttendedActionRefused( + "the attended capability history is invalid" + ) from None + history.append(existing.model_dump(mode="json")) + self._atomic_write( + self.capability_history_path, + json.dumps(history, indent=2, sort_keys=True).encode("utf-8"), + ) + now = _now() + transition = _transition_payload( + run_id=manifest.run_id, + workflow_name=pending.workflow_name, + bundle_revision=revision, + pending=pending, + expected_next_transition=expected, + ) + capability = AttendedPauseCapability( + pause_id=secrets.token_hex(16), + run_id=manifest.run_id, + workflow_name=pending.workflow_name, + bundle_version=transition["bundle_version"], + step_index=pending.step_index, + step_id=pending.step_id, + state_id=pending.state_id, + resume_from_index=pending.resume_from_index, + resume_from_step_id=pending.resume_from_step_id, + pause_digest=pause_digest, + expected_next_transition=expected, + expected_transition_digest=_digest(transition), + transition_baseline=baseline, + delivery_state=_delivery_state(result), + issued_at=_iso(now), + expires_at=_iso(now + timedelta(seconds=max(1.0, ttl_s))), + allowed_actions=_allowed_actions(workflow, pending, baseline), + ) + capability.signature = self._sign(capability, create_key=False) + self._atomic_write( + self.capability_path, + capability.model_dump_json(indent=2).encode("utf-8"), + ) + return capability + + def read(self) -> AttendedPauseCapability: + try: + capability = AttendedPauseCapability.model_validate_json( + self.capability_path.read_text() + ) + except (FileNotFoundError, ValueError) as exc: + raise AttendedActionRefused( + "the run has no valid engine-issued attended capability" + ) from exc + expected = self._sign(capability, create_key=False) + if not hmac.compare_digest(capability.signature, expected): + raise AttendedActionRefused( + "the attended capability signature does not verify" + ) + return capability + + def validate( + self, + request: AttendedActionRequest, + *, + pending: PendingEscalation, + manifest: Any, + now: Optional[datetime] = None, + ) -> AttendedPauseCapability: + capability = self.read() + now = now or _now() + if request.capability_digest != capability.digest: + raise AttendedActionRefused( + "the operator page is stale or the pause capability changed" + ) + if request.action not in capability.allowed_actions: + raise AttendedActionRefused("the capability does not allow this action") + if _parse(capability.expires_at) < now or pause_is_expired(pending, now): + raise PauseExpired( + "the attended pause expired; reload and re-qualify live state" + ) + live_version = bundle_version(manifest.bundle_dir) + if live_version != capability.bundle_version: + raise BundleMismatch("the bundle revision changed after the attended pause") + if _digest(pending) != capability.pause_digest: + raise AttendedActionRefused( + "the exact durable pause changed after capability issuance" + ) + transition = _transition_payload( + run_id=manifest.run_id, + workflow_name=pending.workflow_name, + bundle_revision=live_version, + pending=pending, + expected_next_transition=capability.expected_next_transition, + ) + if _digest(transition) != capability.expected_transition_digest: + raise AttendedActionRefused( + "the expected attended transition binding no longer verifies" + ) + if ( + pending.step_id != capability.step_id + or pending.step_index != capability.step_index + or pending.resume_from_index != capability.resume_from_index + or pending.resume_from_step_id != capability.resume_from_step_id + or manifest.run_id != capability.run_id + or manifest.workflow_name != capability.workflow_name + ): + raise AttendedActionRefused( + "the durable pause no longer matches its issued capability" + ) + return capability + + def _read_log(self) -> AttendedDecisionLog: + if not self.decisions_path.is_file(): + return AttendedDecisionLog() + try: + return AttendedDecisionLog.model_validate_json( + self.decisions_path.read_text() + ) + except ValueError as exc: + raise AttendedActionRefused( + "the attended decision audit log is invalid" + ) from exc + + def prior(self, request: AttendedActionRequest) -> Optional[AttendedDecision]: + request_digest = _digest(request) + for decision in reversed(self._read_log().decisions): + if decision.idempotency_key != request.idempotency_key: + continue + if decision.request_digest != request_digest: + raise AttendedActionRefused( + "the idempotency key was already used for a different request" + ) + return decision + return None + + def unresolved_delivery(self, pause_id: str) -> Optional[AttendedDecision]: + """Return a request whose latest journal state crossed delivery. + + This is pause-wide, not merely idempotency-key-wide. A caller must not + bypass an uncertain delivery by generating a fresh browser retry key. + """ + latest: dict[str, AttendedDecision] = {} + for decision in self._read_log().decisions: + if decision.pause_id == pause_id: + latest[decision.request_digest] = decision + for decision in reversed(list(latest.values())): + if decision.status in {"delivery_started", "delivery_uncertain"}: + return decision + return None + + def append(self, decision: AttendedDecision) -> None: + log = self._read_log() + log.decisions.append(decision) + self._atomic_write( + self.decisions_path, log.model_dump_json(indent=2).encode("utf-8") + ) + + @contextmanager + def lease( + self, + request: AttendedActionRequest, + *, + ttl_s: float = DEFAULT_LEASE_TTL_S, + now: Optional[datetime] = None, + ) -> Iterator[None]: + """Acquire one per-run action lease with no silent stale takeover.""" + now = now or _now() + lease = { + "request_digest": _digest(request), + "idempotency_key": request.idempotency_key, + "acquired_at": _iso(now), + "expires_at": _iso(now + timedelta(seconds=max(1.0, ttl_s))), + } + self.run_dir.mkdir(parents=True, exist_ok=True) + try: + fd = os.open(self.lease_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) + except FileExistsError: + try: + existing = json.loads(self.lease_path.read_text()) + expired = _parse(str(existing["expires_at"])) < now + except (OSError, ValueError, KeyError): + expired = False + if expired: + raise AttendedActionBusy( + "a prior action lease expired without a recorded outcome; " + "delivery is uncertain and must be reconciled before retry" + ) from None + raise AttendedActionBusy( + "another attended action is already in progress" + ) from None + try: + with os.fdopen(fd, "w") as handle: + json.dump(lease, handle, sort_keys=True) + handle.flush() + os.fsync(handle.fileno()) + self._fsync_parent(self.lease_path) + yield + finally: + try: + self.lease_path.unlink() + self._fsync_parent(self.lease_path) + except FileNotFoundError: + pass + + +def issue_attended_capability( + run_dir: Path | str, + *, + store: CheckpointStore, + pending: PendingEscalation, + workflow: Workflow, + result: StepResult, + transition_observation: Optional[TransitionObservation] = None, +) -> AttendedPauseCapability: + manifest = store.read_manifest() + if manifest is None or not manifest.run_id: + raise AttendedActionRefused( + "the durable manifest has no stable run identity; cannot issue " + "an attended mutation capability" + ) + return AttendedActionStore(run_dir).issue( + manifest=manifest, + pending=pending, + workflow=workflow, + result=result, + transition_observation=transition_observation, + ) + + +def attended_capability_summary( + run_dir: Path | str, +) -> Optional[dict[str, Any]]: + """Browser-safe capability metadata; the HMAC and local paths stay private.""" + try: + capability = AttendedActionStore(run_dir).read() + except AttendedActionRefused: + return None + return { + "digest": capability.digest, + "expires_at": capability.expires_at, + "allowed_actions": list(capability.allowed_actions), + "delivery_state": capability.delivery_state, + } + + +def _approval( + capability: AttendedPauseCapability, + *, + operator: str, + resolution: str, + run_dir: Path, +) -> ApprovalRecord: + if not operator.strip(): + raise ApprovalRequired("attended actions require an authenticated operator") + return ApprovalRecord( + approver=operator, + resolution=resolution, + bundle_version=capability.bundle_version, + workflow_name=capability.workflow_name, + run_dir=str(run_dir), + ) + + +def execute_attended_action( + run_dir: Path | str, + request: AttendedActionRequest, + *, + operator: str, + executor: Optional[AttendedActionExecutor] = None, + key: Optional[str] = None, + now: Optional[datetime] = None, +) -> AttendedDecision: + """Admit and execute one attended decision under exact binding.""" + from openadapt_flow import crypto as _crypto + + key = _crypto.resolve_key(key) + run_dir = Path(run_dir) + if not operator.strip(): + raise ApprovalRequired("attended actions require an authenticated operator") + expected_dispositions = { + "continue": {None, "completed_by_operator"}, + "skip": {None, "not_applicable"}, + "teach": {None, "teach_requested"}, + "escalate": {None, "cannot_complete", "needs_assistance"}, + } + if request.disposition not in expected_dispositions[request.action]: + raise AttendedActionRefused( + "the disposition does not match the requested attended action" + ) + actions = AttendedActionStore(run_dir) + prior = actions.prior(request) + if prior is not None: + if prior.status in {"delivery_started", "delivery_uncertain"}: + raise AttendedActionRefused( + "the prior request may have crossed the delivery boundary; " + "automatic retry is refused until an audited reconciliation" + ) + if prior.status != "prepared": + return prior + + checkpoints = CheckpointStore(run_dir, key=key) + pending = checkpoints.read_pending() + manifest = checkpoints.read_manifest() + if pending is None or manifest is None: + raise AttendedActionRefused("the run is not durably paused") + capability = actions.validate(request, pending=pending, manifest=manifest, now=now) + + with actions.lease(request, now=now): + # Repeat the complete validation under the lease: the bundle/pause may + # have changed between page load and lock acquisition. + pending = checkpoints.read_pending() + manifest = checkpoints.read_manifest() + if pending is None or manifest is None: + raise AttendedActionRefused("the run is no longer durably paused") + capability = actions.validate( + request, pending=pending, manifest=manifest, now=now + ) + prior = actions.prior(request) + if prior is not None: + if prior.status in {"delivery_started", "delivery_uncertain"}: + raise AttendedActionRefused( + "the prior request may have crossed the delivery boundary; " + "automatic retry is refused until reconciliation" + ) + if prior.status != "prepared": + return prior + request_digest = _digest(request) + unresolved = actions.unresolved_delivery(capability.pause_id) + if unresolved is not None and request.action in {"continue", "skip"}: + raise AttendedActionRefused( + "another request for this pause may have crossed the delivery " + "boundary; reconcile its live state before continuing or skipping" + ) + + if request.action == "teach": + decision = AttendedDecision( + pause_id=capability.pause_id, + capability_digest=capability.digest, + request_digest=request_digest, + idempotency_key=request.idempotency_key, + action=request.action, + operator=operator, + disposition=request.disposition or "teach_requested", + status="needs_demonstration", + message=( + "Record the corrective demonstration, then run the existing " + "governed teach command. Its regression/revision gate decides " + "accepted, banked-progress, or refused; identity-evidence " + "changes are never auto-promoted." + ), + next_transition=capability.expected_next_transition, + ) + actions.append(decision) + return decision + + if request.action == "escalate": + decision = AttendedDecision( + pause_id=capability.pause_id, + capability_digest=capability.digest, + request_digest=request_digest, + idempotency_key=request.idempotency_key, + action=request.action, + operator=operator, + disposition=request.disposition or "needs_assistance", + status="escalated", + message=( + "Escalation recorded. The durable pause remains intact and " + "can be continued after a qualified operator resolves it." + ), + next_transition=capability.expected_next_transition, + ) + actions.append(decision) + return decision + + if executor is None: + raise AttendedActionRefused( + "this console has no deployment-bound attended executor; start " + "it with the qualified backend/effect configuration" + ) + + resolution = ( + "operator completed the live-app task; verify and continue" + if request.action == "continue" + else "operator requested policy-scoped skip" + ) + approval = _approval( + capability, operator=operator, resolution=resolution, run_dir=run_dir + ) + prepared = AttendedDecision( + pause_id=capability.pause_id, + capability_digest=capability.digest, + request_digest=request_digest, + idempotency_key=request.idempotency_key, + action=request.action, + operator=operator, + disposition=request.disposition, + status="prepared", + message="request admitted; no delivery attempted", + next_transition=capability.expected_next_transition, + ) + if prior is None: + actions.append(prepared) + started = prepared.model_copy( + update={ + "decision_id": secrets.token_hex(16), + "status": "delivery_started", + "message": ( + "deployment-bound verification/resume started; a crash " + "after this record makes delivery uncertain" + ), + "created_at": _iso(_now()), + } + ) + actions.append(started) + try: + result = ( + executor.continue_run(run_dir, capability, approval) + if request.action == "continue" + else executor.skip_run(run_dir, capability, approval) + ) + except Exception: + uncertain = started.model_copy( + update={ + "decision_id": secrets.token_hex(16), + "status": "delivery_uncertain", + "message": ( + "the deployment-bound action did not return a terminal " + "receipt; reconcile live state before any retry" + ), + "created_at": _iso(_now()), + } + ) + actions.append(uncertain) + raise + decision = AttendedDecision( + pause_id=capability.pause_id, + capability_digest=capability.digest, + request_digest=request_digest, + idempotency_key=request.idempotency_key, + action=request.action, + operator=operator, + disposition=request.disposition, + status=result.status, + message=result.message, + report_success=result.report_success, + next_transition=result.next_transition, + ) + actions.append(decision) + return decision + + +def checkpoint_human_completed_step( + run_dir: Path | str, + *, + capability: AttendedPauseCapability, + result: StepResult, + params: dict[str, str], + key: Optional[str] = None, +) -> RunCheckpoint: + """Advance a linear resume point after outcome verification, without acting.""" + if not result.ok or result.postconditions_ok is False: + raise AttendedActionRefused( + "the human-completed step did not pass outcome verification" + ) + if result.effect_verified is False: + raise AttendedActionRefused( + "the human-completed step's independent effect was not confirmed" + ) + checkpoint = RunCheckpoint( + workflow_name=capability.workflow_name, + step_index=capability.step_index, + step_id=capability.step_id, + intent=result.intent, + next_step_index=capability.step_index + 1, + params=dict(params), + effect_verified=result.effect_verified, + effect_approved_unverified=result.effect_approved_unverified, + effect_contract_hashes=list(result.effect_contract_hashes), + postconditions_ok=result.postconditions_ok, + skipped=False, + actuation="human_attended", + ) + CheckpointStore(run_dir, key=key).write_checkpoint(checkpoint) + return checkpoint + + +class BoundAttendedExecutor: + """Real engine executor constructed from a deployment-bound Replayer factory. + + The factory must return a fresh Replayer wired to the qualified live backend, + effect verifier, policy authorization, and egress posture. The console + never accepts backend credentials or challenge answers in an HTTP payload. + """ + + def __init__( + self, + replayer_factory: Callable[[Any], Any], + *, + key: Optional[str] = None, + ) -> None: + from openadapt_flow import crypto as _crypto + + self.replayer_factory = replayer_factory + self.key = _crypto.resolve_key(key) + # Per-run filesystem leases prevent duplicate decisions for one pause. + # The executor additionally owns one shared live backend/session, so + # actions for different runs must not observe or drive it concurrently. + self._live_session_lock = threading.Lock() + + @contextmanager + def _exclusive_live_session(self) -> Iterator[None]: + if not self._live_session_lock.acquire(blocking=False): + raise AttendedActionBusy( + "the qualified live application session is serving another " + "attended action; reload after that decision completes" + ) + try: + yield + finally: + self._live_session_lock.release() + + @staticmethod + def _expected(workflow: Workflow, step_index: int) -> str: + next_index = step_index + 1 + return ( + workflow.steps[next_index].id + if next_index < len(workflow.steps) + else "" + ) + + def _load( + self, run_dir: Path, capability: AttendedPauseCapability + ) -> tuple[CheckpointStore, Any, Workflow]: + store = CheckpointStore(run_dir, key=self.key) + manifest = store.read_manifest() + if manifest is None: + raise AttendedActionRefused("durable manifest missing") + if manifest.run_id != capability.run_id: + raise AttendedActionRefused("run identity changed after pause") + if bundle_version(manifest.bundle_dir) != capability.bundle_version: + raise BundleMismatch("bundle changed after attended capability issuance") + workflow = Workflow.load(manifest.bundle_dir, key=self.key) + if workflow.name != capability.workflow_name: + raise AttendedActionRefused( + "workflow identity changed after attended capability issuance" + ) + if workflow.program is not None: + raise AttendedActionRefused( + "program-graph attended completion requires an exact interpreter " + "transition receipt; generic program continuation is refused" + ) + if ( + not 0 <= capability.step_index < len(workflow.steps) + or workflow.steps[capability.step_index].id != capability.step_id + ): + raise AttendedActionRefused( + "paused step identity no longer matches the qualified workflow" + ) + if self._expected(workflow, capability.step_index) != ( + capability.expected_next_transition + ): + raise AttendedActionRefused( + "the expected next transition no longer matches the workflow" + ) + return store, manifest, workflow + + @staticmethod + def _bind_authorization(replayer: Any, manifest: Any) -> None: + if manifest.governed_authorization is not None: + existing = getattr(replayer, "governed_authorization", None) + if existing is not None and existing != manifest.governed_authorization: + raise BundleMismatch( + "attended Replayer carries a different governed authorization" + ) + replayer.governed_authorization = manifest.governed_authorization + replayer.governed_continuation = True + + def _resume( + self, + *, + run_dir: Path, + store: CheckpointStore, + manifest: Any, + workflow: Workflow, + capability: AttendedPauseCapability, + approval: ApprovalRecord, + result: StepResult, + skipped: bool, + resume_replayer: Any, + ) -> AttendedExecutionResult: + checkpoint = RunCheckpoint( + workflow_name=capability.workflow_name, + step_index=capability.step_index, + step_id=capability.step_id, + intent=result.intent, + next_step_index=capability.step_index + 1, + params=dict(manifest.params), + effect_verified=result.effect_verified, + effect_approved_unverified=result.effect_approved_unverified, + effect_contract_hashes=list(result.effect_contract_hashes), + governed_authorization_id=( + manifest.governed_authorization.authorization_id + if manifest.governed_authorization is not None + else None + ), + governed_approval_source=( + manifest.governed_authorization.approval_source + if manifest.governed_authorization is not None + else None + ), + postconditions_ok=result.postconditions_ok, + skipped=skipped, + actuation="human_attended_skip" if skipped else "human_attended", + ) + store.write_checkpoint(checkpoint) + store.write_approval(approval) + pending = store.read_pending() + if pending is None: + raise AttendedActionRefused("pause disappeared before resume") + store.write_pending(pending.model_copy(update={"status": "approved"})) + + # Import lazily to avoid a durable-module cycle. + from openadapt_flow.runtime.durable.resume import resume + + resumed = resume( + run_dir, + resume_replayer, + approval=approval, + key=self.key, + ) + return AttendedExecutionResult( + status="completed" if resumed.success else "halted", + message=( + "Human-completed outcome verified; resumed after the attended " + "step without re-actuating it." + if resumed.success and not skipped + else ( + "Declared optional step skipped; resumed without actuation." + if resumed.success + else "The deterministic continuation halted and remains auditable." + ) + ), + report_success=resumed.success, + resumed_from=capability.step_id, + next_transition=capability.expected_next_transition, + ) + + def continue_run( + self, + run_dir: Path, + capability: AttendedPauseCapability, + approval: ApprovalRecord, + ) -> AttendedExecutionResult: + try: + with self._exclusive_live_session(): + return self._continue_run_locked(run_dir, capability, approval) + except AttendedActionBusy as exc: + return AttendedExecutionResult( + status="refused", + message=str(exc), + report_success=False, + resumed_from=capability.step_id, + next_transition=capability.expected_next_transition, + ) + + def _continue_run_locked( + self, + run_dir: Path, + capability: AttendedPauseCapability, + approval: ApprovalRecord, + ) -> AttendedExecutionResult: + try: + store, manifest, workflow = self._load(run_dir, capability) + replayer = self.replayer_factory(manifest) + self._bind_authorization(replayer, manifest) + attended_store = AttendedActionStore(run_dir) + result = replayer.revalidate_attended_completion( + workflow, + step_index=capability.step_index, + params=dict(manifest.params), + bundle_dir=Path(manifest.bundle_dir), + run_dir=run_dir, + run_id=manifest.run_id, + transition_baseline=capability.transition_baseline, + transition_digest=attended_store.transition_value_digest, + ) + if not result.ok: + return AttendedExecutionResult( + status="refused", + message=result.error or "attended outcome verification refused", + report_success=False, + resumed_from=capability.step_id, + next_transition=capability.expected_next_transition, + ) + except ResumeRefused as exc: + return AttendedExecutionResult( + status="refused", + message=str(exc), + report_success=False, + resumed_from=capability.step_id, + next_transition=capability.expected_next_transition, + ) + except Exception: + # Loading, attaching to the live session, and fresh verification + # are observation-only. A failure here cannot be outcome evidence, + # but it also has not mutated workflow state. + return AttendedExecutionResult( + status="refused", + message=( + "fresh attended verification was unavailable before " + "resume; no workflow continuation was admitted" + ), + report_success=False, + resumed_from=capability.step_id, + next_transition=capability.expected_next_transition, + ) + try: + return self._resume( + run_dir=run_dir, + store=store, + manifest=manifest, + workflow=workflow, + capability=capability, + approval=approval, + result=result, + skipped=False, + resume_replayer=replayer, + ) + except ResumeRefused as exc: + return AttendedExecutionResult( + status="refused", + message=str(exc), + report_success=False, + resumed_from=capability.step_id, + next_transition=capability.expected_next_transition, + ) + + def skip_run( + self, + run_dir: Path, + capability: AttendedPauseCapability, + approval: ApprovalRecord, + ) -> AttendedExecutionResult: + try: + with self._exclusive_live_session(): + return self._skip_run_locked(run_dir, capability, approval) + except AttendedActionBusy as exc: + return AttendedExecutionResult( + status="refused", + message=str(exc), + report_success=False, + resumed_from=capability.step_id, + next_transition=capability.expected_next_transition, + ) + + def _skip_run_locked( + self, + run_dir: Path, + capability: AttendedPauseCapability, + approval: ApprovalRecord, + ) -> AttendedExecutionResult: + try: + store, manifest, workflow = self._load(run_dir, capability) + step = workflow.steps[capability.step_index] + if ( + step.risk == "irreversible" + or step.effects + or step.guard is None + or step.guard.on_unmet != "skip" + ): + return AttendedExecutionResult( + status="refused", + message=( + "Skip is not declared by this workflow, or the step is " + "consequential/effectful. A non-success disposition may " + "be escalated, but it cannot be turned into success." + ), + report_success=False, + resumed_from=capability.step_id, + next_transition=capability.expected_next_transition, + ) + replayer = self.replayer_factory(manifest) + self._bind_authorization(replayer, manifest) + frame = replayer.vision.wait_settled(replayer.backend) + if replayer._predicate_holds( + step.guard.predicate, + frame, + Path(manifest.bundle_dir), + dict(manifest.params), + ): + return AttendedExecutionResult( + status="refused", + message=( + "The declared skip guard currently holds, so normal " + "workflow semantics require executing this step." + ), + report_success=False, + resumed_from=capability.step_id, + next_transition=capability.expected_next_transition, + ) + result = StepResult( + step_id=step.id, + intent=step.intent, + ok=True, + skipped=True, + postconditions_ok=None, + actuation="human_attended_skip", + ) + except ResumeRefused as exc: + return AttendedExecutionResult( + status="refused", + message=str(exc), + report_success=False, + resumed_from=capability.step_id, + next_transition=capability.expected_next_transition, + ) + except Exception: + return AttendedExecutionResult( + status="refused", + message=( + "fresh skip-policy validation was unavailable before " + "resume; no workflow continuation was admitted" + ), + report_success=False, + resumed_from=capability.step_id, + next_transition=capability.expected_next_transition, + ) + try: + return self._resume( + run_dir=run_dir, + store=store, + manifest=manifest, + workflow=workflow, + capability=capability, + approval=approval, + result=result, + skipped=True, + resume_replayer=replayer, + ) + except ResumeRefused as exc: + return AttendedExecutionResult( + status="refused", + message=str(exc), + report_success=False, + resumed_from=capability.step_id, + next_transition=capability.expected_next_transition, + ) diff --git a/openadapt_flow/runtime/durable/checkpoint.py b/openadapt_flow/runtime/durable/checkpoint.py index fc4287c0..11519b08 100644 --- a/openadapt_flow/runtime/durable/checkpoint.py +++ b/openadapt_flow/runtime/durable/checkpoint.py @@ -74,6 +74,10 @@ class RunManifest(BaseModel): """ schema_version: int = 1 + #: Random run-instance identity, distinct from workflow/bundle identity. + #: Attended capabilities bind to this so a capability copied between two + #: runs of the same bundle is refused. + run_id: str = "" workflow_name: str #: The workflow bundle directory (absolute), source of ``workflow.json`` #: and the template crops. diff --git a/openadapt_flow/runtime/durable/controller.py b/openadapt_flow/runtime/durable/controller.py index fc4b7aef..b30b6ba8 100644 --- a/openadapt_flow/runtime/durable/controller.py +++ b/openadapt_flow/runtime/durable/controller.py @@ -30,7 +30,7 @@ import re from pathlib import Path -from typing import Optional +from typing import TYPE_CHECKING, Optional from openadapt_flow.ir import Step, StepResult, Workflow from openadapt_flow.runtime.authorization import GovernedRunAuthorization @@ -42,6 +42,9 @@ ) from openadapt_flow.runtime.durable.program_checkpoint import ProgramCheckpoint +if TYPE_CHECKING: + from openadapt_flow.runtime.durable.attended import TransitionObservation + HUMAN_REQUIRED_MARKERS: tuple[str, ...] = ( "captcha", "verify you are human", @@ -225,6 +228,7 @@ def __init__( self, run_dir: Path | str, *, + run_id: str, workflow_name: str, bundle_dir: Path | str, params: dict[str, str], @@ -236,12 +240,15 @@ def __init__( # ``key`` (None by default) opts the durable artifacts into AES-256-GCM # encryption-at-rest; unset => plaintext, exactly as before. self.store = CheckpointStore(run_dir, key=key) + self.run_id = run_id self.workflow_name = workflow_name + self.bundle_dir = Path(bundle_dir).resolve() self.governed_authorization = governed_authorization self.store.write_manifest( RunManifest( + run_id=run_id, workflow_name=workflow_name, - bundle_dir=str(Path(bundle_dir).resolve()), + bundle_dir=str(self.bundle_dir), params=dict(params), worklists=worklists, governed_authorization=governed_authorization, @@ -255,6 +262,9 @@ def record( step: Step, result: StepResult, params: dict[str, str], + *, + workflow: Optional[Workflow] = None, + transition_observation: Optional["TransitionObservation"] = None, ) -> None: """Persist the durable artifact for one completed step. @@ -297,21 +307,33 @@ def record( last = self.store.last_checkpoint() resume_from = last.next_step_index if last is not None else 0 category, options = classify_halt(step, result) - self.store.write_pending( - PendingEscalation( - workflow_name=self.workflow_name, - step_index=step_index, - step_id=step.id, - intent=step.intent, - category=category, - reason=result.error or "", - detail=list(result.effect_results or []), - proposed_options=options, - resume_from_index=resume_from, - resume_from_step_id=(last.step_id if last is not None else None), - params=dict(params), - ) + pending = PendingEscalation( + workflow_name=self.workflow_name, + step_index=step_index, + step_id=step.id, + intent=step.intent, + category=category, + reason=result.error or "", + detail=list(result.effect_results or []), + proposed_options=options, + resume_from_index=resume_from, + resume_from_step_id=(last.step_id if last is not None else None), + params=dict(params), ) + self.store.write_pending(pending) + if workflow is not None: + from openadapt_flow.runtime.durable.attended import ( + issue_attended_capability, + ) + + issue_attended_capability( + self.store.run_dir, + store=self.store, + pending=pending, + workflow=workflow, + result=result, + transition_observation=transition_observation, + ) # -- Phase-2 program (state-machine) durability -------------------------- @@ -332,6 +354,8 @@ def record_program_halt( intent: str, result: StepResult, params: dict[str, str], + workflow: Optional[Workflow] = None, + transition_observation: Optional["TransitionObservation"] = None, ) -> None: """Persist a durable PROGRAM pause (the interpreter HALTED for a human). @@ -344,24 +368,34 @@ def record_program_halt( marks the pause as a state-machine pause.""" last = self.store.last_program_checkpoint() category, options = classify_halt(None, result) - self.store.write_pending( - PendingEscalation( - workflow_name=self.workflow_name, - step_index=0, - step_id=state_id, - intent=intent, - state_id=state_id, - category=category, - reason=result.error or "", - detail=list(result.effect_results or []), - proposed_options=options, - resume_from_step_id=( - last.verified_state_id if last is not None else None - ), - params=dict(params), - program=True, - ) + pending = PendingEscalation( + workflow_name=self.workflow_name, + step_index=0, + step_id=state_id, + intent=intent, + state_id=state_id, + category=category, + reason=result.error or "", + detail=list(result.effect_results or []), + proposed_options=options, + resume_from_step_id=(last.verified_state_id if last is not None else None), + params=dict(params), + program=True, ) + self.store.write_pending(pending) + if workflow is not None: + from openadapt_flow.runtime.durable.attended import ( + issue_attended_capability, + ) + + issue_attended_capability( + self.store.run_dir, + store=self.store, + pending=pending, + workflow=workflow, + result=result, + transition_observation=transition_observation, + ) def resumed_step_results( diff --git a/openadapt_flow/runtime/durable/resume.py b/openadapt_flow/runtime/durable/resume.py index 854af374..0c4f3b48 100644 --- a/openadapt_flow/runtime/durable/resume.py +++ b/openadapt_flow/runtime/durable/resume.py @@ -175,6 +175,7 @@ def resume( worklists=resolved_worklists, save_healed_to=resolved_healed, live_bundle_version=live_bundle_version, + run_id=(manifest.run_id if manifest is not None else None), ) # -- linear resume (unchanged control flow; now gated by approval) -------- @@ -188,6 +189,7 @@ def resume( run_dir=run_dir, save_healed_to=(Path(resolved_healed) if resolved_healed else None), resume_from=start_index, + run_id=(manifest.run_id if manifest is not None else None), ) @@ -202,6 +204,7 @@ def _resume_program( worklists: dict[str, list[dict[str, str]]], save_healed_to: Optional[Path | str], live_bundle_version: str, + run_id: Optional[str], ) -> RunReport: """Restore and continue a Phase-2 PROGRAM run from its interpreter checkpoint. @@ -235,4 +238,5 @@ def _resume_program( run_dir=store.run_dir, save_healed_to=(Path(save_healed_to) if save_healed_to else None), resume_program=checkpoint, + run_id=run_id, ) diff --git a/openadapt_flow/runtime/replayer.py b/openadapt_flow/runtime/replayer.py index 2ff36197..5d21ba03 100644 --- a/openadapt_flow/runtime/replayer.py +++ b/openadapt_flow/runtime/replayer.py @@ -40,7 +40,7 @@ import uuid from datetime import date, datetime, timezone from pathlib import Path -from typing import Any, Optional +from typing import Any, Callable, Optional from openadapt_flow.backend import Backend, StructuralResolutionRefused from openadapt_flow.bundle_validation import compute_parameter_schema_digest @@ -413,6 +413,7 @@ def run( save_healed_to: Optional[Path] = None, resume_from: Optional[int] = None, resume_program: Optional[ProgramCheckpoint] = None, + run_id: Optional[str] = None, execution_origin: Optional[str] = None, execution_entry_url: Optional[str] = None, ) -> RunReport: @@ -457,6 +458,10 @@ def run( paused state -- never from the graph entry, never re-performing an already-confirmed write. Supplied by ``openadapt_flow.runtime.durable.resume``, not by hand. + run_id: Stable run-instance identity supplied only by durable resume. + New runs generate one; resumed legs preserve the manifest value + so attended capabilities and effect idempotency remain bound to + the same logical run. execution_origin: Actual browser origin loaded before replay. It is evidence metadata only; hosted validation requires an exact match to its signed target origin. @@ -477,7 +482,7 @@ def run( # ``__run_id__`` param so an idempotency key can be bound PER-RUN (via # ``ValueExpr(param="__run_id__")``) instead of reusing a frozen demo # literal across unrelated runs. - self._run_id = ( + self._run_id = run_id or ( self.governed_authorization.authorization_id if self.governed_authorization is not None else uuid.uuid4().hex @@ -602,6 +607,7 @@ def run( durable_run = DurableRun( run_dir, + run_id=self._run_id, workflow_name=workflow.name, bundle_dir=bundle_dir, params=params, @@ -657,7 +663,18 @@ def run( if durable_run is not None: # Tier-3: verified step -> checkpoint; halt -> pending # escalation (resumable from the last checkpoint). - durable_run.record(step_index, step, result, params) + durable_run.record( + step_index, + step, + result, + params, + workflow=workflow, + transition_observation=( + self._attended_transition_observation() + if not result.ok + else None + ), + ) self._account_result(report, result) if not result.ok: break @@ -930,7 +947,7 @@ def _interpret_program( # Durably PAUSE (never just die): capture WHERE we stopped so an # approved resume can RESTORE the interpreter from here. if self._program_durable is not None: - self._record_program_pause(halt, report) + self._record_program_pause(halt, report, workflow=workflow) report.results.append( StepResult( step_id="", @@ -1484,7 +1501,9 @@ def _record_program_checkpoint( ) durable.record_program_checkpoint(checkpoint) - def _record_program_pause(self, halt: "_ProgramHalt", report: RunReport) -> None: + def _record_program_pause( + self, halt: "_ProgramHalt", report: RunReport, *, workflow: Workflow + ) -> None: """Persist a durable PROGRAM pause (the interpreter HALTED for a human). Uses the last executed state's failing result (an action failure) or, for @@ -1509,6 +1528,8 @@ def _record_program_pause(self, halt: "_ProgramHalt", report: RunReport) -> None intent=self._current_intent or failing.intent, result=failing, params=self._current_params, + workflow=workflow, + transition_observation=self._attended_transition_observation(), ) def revalidate_program_checkpoint( @@ -1556,6 +1577,254 @@ def revalidate_program_checkpoint( "the checkpoint" ) + def revalidate_attended_completion( + self, + workflow: Workflow, + *, + step_index: int, + params: dict[str, str], + bundle_dir: Path, + run_dir: Path, + run_id: str, + transition_baseline: Any, + transition_digest: Callable[[str, str], str], + ) -> StepResult: + """Verify a human-completed linear step without actuating it. + + This is intentionally not ``_run_step``: the person already completed + the challenge/task in the live application, so invoking the action + again could duplicate a consequential write. The method observes only: + + * the paused step's screen postconditions; + * its independently declared effects against the *current* system of + record, when the contract is meaningful without a pre-action delta; + * the next step's unique target and armed identity, proving the expected + continuation is actually available. + + Delivery receipts are never consulted. Transition-relative + postconditions are compared against the PHI-safe keyed baseline bound + into the signed pause capability. Delta/collateral-loss effects still + require a protected pre-delivery record baseline and are refused when + one is unavailable. + """ + from openadapt_flow.runtime.effects import EffectState + from openadapt_flow.runtime.effects._common import judge_records + + if not (0 <= step_index < len(workflow.steps)): + raise StateDiverged("the attended pause references no workflow step") + step = workflow.steps[step_index] + result = StepResult( + step_id=step.id, + intent=step.intent, + ok=False, + actuation="human_attended", + ) + self._run_id = run_id + (Path(run_dir) / "steps").mkdir(parents=True, exist_ok=True) + frame = self.vision.wait_settled(self.backend) + result.before_png = self._save_step_png( + run_dir, step.id, "attended-before", frame + ) + + relative_kinds = {"url_changed", "title_changed", "new_tab_opened"} + relative = [ + pc + for pc in step.expect + if (pc.kind.value if hasattr(pc.kind, "value") else pc.kind) + in relative_kinds + ] + live_structural = self._structural_state() if relative else {} + for condition in relative: + kind = ( + condition.kind.value + if hasattr(condition.kind, "value") + else str(condition.kind) + ) + if kind == "url_changed": + current = live_structural.get("url") + baseline_digest = getattr(transition_baseline, "url_digest", None) + changed = ( + isinstance(current, str) + and baseline_digest is not None + and transition_digest("url", current) != baseline_digest + ) + elif kind == "title_changed": + current = live_structural.get("page_title") + baseline_digest = getattr(transition_baseline, "title_digest", None) + changed = ( + isinstance(current, str) + and baseline_digest is not None + and transition_digest("page_title", current) != baseline_digest + ) + else: + current = live_structural.get("page_count") + baseline_count = getattr(transition_baseline, "page_count", None) + changed = ( + isinstance(current, int) + and isinstance(baseline_count, int) + and current > baseline_count + ) + if not changed: + result.postconditions_ok = False + result.error = ( + "the human-completed step's signed structural transition " + "was unavailable or unchanged; Continue is refused" + ) + result.after_png = result.before_png + return result + + nonrelative = [ + pc + for pc in step.expect + if (pc.kind.value if hasattr(pc.kind, "value") else pc.kind) + not in relative_kinds + ] + if nonrelative: + verification_step = step.model_copy(update={"expect": nonrelative}) + post_ok, frame, failed = self._check_postconditions( + verification_step, frame, bundle_dir, {}, result + ) + result.postconditions_ok = post_ok + if not post_ok: + # The failed descriptions can contain demonstrated text (and + # therefore PHI). Keep them out of the browser-facing decision + # message; protected screenshots remain in the local run dir. + result.error = ( + "the human-completed step's live postconditions are not " + f"satisfied ({len(failed)} condition(s) failed)" + ) + result.after_png = self._save_step_png( + run_dir, step.id, "attended-after", frame + ) + return result + + if step.effects: + if self.effect_verifier is None: + result.effect_verified = False + result.error = ( + "the human-completed consequential step has no independent " + "effect verifier; outcome is uncertain and Continue is refused" + ) + result.after_png = self._save_step_png( + run_dir, step.id, "attended-after", frame + ) + return result + current = self.effect_verifier.capture_pre_state() + if not current.reachable: + result.effect_verified = False + result.error = ( + "the system of record is unreachable; outcome is uncertain " + "and Continue is refused" + ) + result.after_png = self._save_step_png( + run_dir, step.id, "attended-after", frame + ) + return result + effects = self._resolve_effects(step.effects, params) + for effect in effects: + if effect.needs_operator_confirmation: + result.effect_verified = False + result.error = ( + "the effect contract is still a placeholder; a human " + "statement cannot replace independent verification" + ) + break + if effect.count_new_only or effect.forbid_collateral_loss: + result.effect_verified = False + result.error = ( + "the effect requires a pre-delivery delta or collateral-" + "loss baseline that is unavailable after human delivery; " + "outcome is uncertain and Continue is refused" + ) + break + baseline = EffectState( + substrate=current.substrate, + reachable=True, + records=[], + detail={"attended_current_state_readback": True}, + ) + verdict = judge_records( + effect, + baseline, + current.records, + substrate=current.substrate, + ) + result.effect_contract_hashes.append(effect.contract_hash()) + result.effect_results.append( + f"[attended-current-readback] {effect.kind.value}: " + f"{verdict.verdict.value} — {verdict.reason}" + ) + if not verdict.confirmed: + result.effect_verified = False + result.error = ( + "the independent current-state effect readback did not " + f"confirm the outcome ({verdict.verdict.value})" + ) + break + else: + result.effect_verified = True + if result.error is not None: + result.after_png = self._save_step_png( + run_dir, step.id, "attended-after", frame + ) + return result + + if not step.expect and not step.effects: + result.error = ( + "the human-completed step declares neither a postcondition nor " + "an independent effect; delivery cannot be confused with outcome " + "verification, so Continue is refused" + ) + result.after_png = self._save_step_png( + run_dir, step.id, "attended-after", frame + ) + return result + + # Prove the expected continuation's target is unique and, where the + # workflow armed identity, still identifies the intended entity. + next_index = step_index + 1 + if next_index < len(workflow.steps): + next_step = workflow.steps[next_index] + if next_step.anchor is not None: + resolution, _region, error = self._resolve_step( + next_step, frame, bundle_dir, workflow + ) + if error is not None or resolution is None: + result.error = ( + "the expected next transition's target did not resolve " + "uniquely; live state diverged and Continue is refused" + ) + result.after_png = self._save_step_png( + run_dir, step.id, "attended-after", frame + ) + return result + if next_step.identity_armed: + identity = self._verify_identity( + next_step, + resolution, + frame, + params, + workflow, + bundle_dir, + ) + result.identity = identity + if identity.status != "verified": + result.error = ( + "the expected next transition's target identity " + f"was {identity.status}; Continue is refused" + ) + result.after_png = self._save_step_png( + run_dir, step.id, "attended-after", frame + ) + return result + + result.ok = True + result.postconditions_ok = True if step.expect else None + result.after_png = self._save_step_png( + run_dir, step.id, "attended-after", frame + ) + return result + def _resolve_graph(self, workflow: Workflow, graph_id: str) -> ProgramGraph: """Resolve a durable ``graph_id`` back to a :class:`ProgramGraph` (``TOP_GRAPH_ID`` -> ``workflow.program``, else a named subflow).""" @@ -3475,6 +3744,25 @@ def _structural_state(self) -> dict[str, Any]: state[key] = value return state + def _attended_transition_observation(self) -> Any: + """Capture ephemeral browser structure for a signed pause baseline. + + Raw URL/title values live only in this return object long enough for + :class:`AttendedActionStore` to HMAC them. They are never written into + a checkpoint, report, pending escalation, or capability. + """ + from openadapt_flow.runtime.durable.attended import TransitionObservation + + state = self._structural_state() + url = state.get("url") + title = state.get("page_title") + page_count = state.get("page_count") + return TransitionObservation( + url=url if isinstance(url, str) else None, + page_title=title if isinstance(title, str) else None, + page_count=page_count if isinstance(page_count, int) else None, + ) + def _structural_changed( self, key: str, start_state: dict[str, Any] ) -> Optional[bool]: diff --git a/tests/test_attended_actions.py b/tests/test_attended_actions.py new file mode 100644 index 00000000..b574c8fe --- /dev/null +++ b/tests/test_attended_actions.py @@ -0,0 +1,1131 @@ +"""Adversarial contracts for the target-state attended action path.""" + +from __future__ import annotations + +import json +import os +import shutil +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import pytest +from fastapi.testclient import TestClient +from pydantic import ValidationError + +from openadapt_flow.console.app import create_app +from openadapt_flow.ir import ( + ActionKind, + Guard, + HaltObservation, + Postcondition, + PostconditionKind, + Predicate, + PredicateKind, + RunReport, + Step, + StepResult, + Workflow, +) +from openadapt_flow.runtime.durable.approval import ApprovalRecord +from openadapt_flow.runtime.durable.attended import ( + AttendedActionRefused, + AttendedActionRequest, + AttendedActionStore, + BoundAttendedExecutor, + TransitionObservation, + execute_attended_action, + issue_attended_capability, +) +from openadapt_flow.runtime.durable.checkpoint import ( + CheckpointStore, + PendingEscalation, + RunManifest, +) +from openadapt_flow.runtime.effects import Effect, EffectKind, EffectState +from openadapt_flow.runtime.replayer import Replayer +from tests.test_replayer import FakeBackend, FakeVision, Match + + +def _step(step_id: str, key: str, *, expect: str | None = None) -> Step: + return Step( + id=step_id, + intent=f"press {key}", + action=ActionKind.KEY, + key=key, + expect=( + [ + Postcondition( + kind=PostconditionKind.TEXT_PRESENT, + text=expect, + timeout_s=0.01, + ) + ] + if expect + else [] + ), + ) + + +def _paused( + tmp_path: Path, + *, + workflow: Workflow | None = None, + result: StepResult | None = None, + transition_observation: TransitionObservation | None = None, +): + workflow = workflow or Workflow( + name="attended", + steps=[_step("human", "A", expect="DONE"), _step("next", "B")], + ) + bundle = tmp_path / "bundle" + run = tmp_path / "run" + workflow.save(bundle) + store = CheckpointStore(run) + store.write_manifest( + RunManifest( + run_id="run-instance-a", + workflow_name=workflow.name, + bundle_dir=str(bundle), + params={}, + ) + ) + pending = PendingEscalation( + workflow_name=workflow.name, + step_index=0, + step_id=workflow.steps[0].id, + intent=workflow.steps[0].intent, + category="human_required", + reason="Please verify you are human", + proposed_options=["complete in live app", "resume"], + resume_from_index=0, + ) + store.write_pending(pending) + RunReport( + workflow_name=workflow.name, + started_at="2026-07-18T12:00:00+00:00", + success=False, + results=[ + result + or StepResult( + step_id=workflow.steps[0].id, + intent=workflow.steps[0].intent, + ok=False, + error="Please verify you are human", + ) + ], + halt=HaltObservation( + state_id=workflow.steps[0].id, + intent=workflow.steps[0].intent, + reason="Please verify you are human", + ), + ).save(run) + capability = issue_attended_capability( + run, + store=store, + pending=pending, + workflow=workflow, + result=result + or StepResult( + step_id=workflow.steps[0].id, + intent=workflow.steps[0].intent, + ok=False, + error="Please verify you are human", + ), + transition_observation=transition_observation, + ) + return workflow, bundle, run, store, capability + + +def _request(capability, action="continue", key="request-key-0001"): + return AttendedActionRequest( + capability_digest=capability.digest, + idempotency_key=key, + action=action, + disposition=( + "completed_by_operator" if action == "continue" else "not_applicable" + ), + ) + + +class _ResultExecutor: + def __init__(self): + self.calls = 0 + + def continue_run(self, run_dir, capability, approval): + from openadapt_flow.runtime.durable.attended import AttendedExecutionResult + + self.calls += 1 + return AttendedExecutionResult( + status="completed", + message="verified", + report_success=True, + next_transition=capability.expected_next_transition, + ) + + def skip_run(self, run_dir, capability, approval): + return self.continue_run(run_dir, capability, approval) + + +def test_capability_binds_run_bundle_pause_and_transition(tmp_path): + _workflow, _bundle, _run, store, capability = _paused(tmp_path) + assert capability.run_id == "run-instance-a" + assert capability.step_id == "human" + assert capability.expected_next_transition == "next" + assert capability.bundle_version.startswith("sha256:") + assert capability.expected_transition_digest.startswith("sha256:") + assert AttendedActionStore(store.run_dir).read() == capability + + pending = store.read_pending() + assert pending is not None + store.write_pending(pending.model_copy(update={"step_id": "other"})) + with pytest.raises(AttendedActionRefused, match="pause changed"): + execute_attended_action( + store.run_dir, + _request(capability), + operator="staff", + executor=_ResultExecutor(), + ) + + +def test_capability_derives_only_semantically_supported_actions(tmp_path): + _workflow, _bundle, _run, _store, verified = _paused(tmp_path / "verified") + assert verified.allowed_actions == ("continue", "teach", "escalate") + + unverified_workflow = Workflow( + name="unverified", + steps=[_step("human", "A")], + ) + _workflow, _bundle, _run, _store, unverified = _paused( + tmp_path / "unverified", workflow=unverified_workflow + ) + assert unverified.allowed_actions == ("teach", "escalate") + + optional_workflow = Workflow( + name="optional", + steps=[ + Step( + id="human", + intent="optional dismissal", + action=ActionKind.KEY, + key="A", + guard=Guard( + predicate=Predicate( + kind=PredicateKind.TEXT_PRESENT, text="OPTIONAL" + ), + on_unmet="skip", + ), + ) + ], + ) + _workflow, _bundle, _run, _store, optional = _paused( + tmp_path / "optional", workflow=optional_workflow + ) + assert optional.allowed_actions == ("skip", "teach", "escalate") + + absolute_effect_workflow = Workflow( + name="absolute-effect", + steps=[ + Step( + id="human", + intent="save", + action=ActionKind.KEY, + key="A", + effects=[ + Effect( + kind=EffectKind.RECORD_WRITTEN, + match={"id": "row-1"}, + forbid_collateral_loss=False, + ) + ], + ) + ], + ) + _workflow, _bundle, _run, _store, absolute = _paused( + tmp_path / "absolute", workflow=absolute_effect_workflow + ) + assert absolute.allowed_actions == ("continue", "teach", "escalate") + + delta_effect_workflow = absolute_effect_workflow.model_copy(deep=True) + delta_effect_workflow.name = "delta-effect" + delta_effect_workflow.steps[0].effects[0].count_new_only = True + _workflow, _bundle, _run, _store, delta = _paused( + tmp_path / "delta", workflow=delta_effect_workflow + ) + assert delta.allowed_actions == ("teach", "escalate") + + +def test_transition_baseline_is_keyed_signed_and_contains_no_raw_phi(tmp_path): + raw_url = "https://payer.example/eligibility?patient=Jane-Roe&member=ABC123" + raw_title = "Jane Roe — Eligibility ABC123" + workflow = Workflow( + name="relative", + steps=[ + Step( + id="human", + intent="complete login", + action=ActionKind.KEY, + key="A", + expect=[Postcondition(kind=PostconditionKind.URL_CHANGED)], + ) + ], + ) + _workflow, _bundle, run, _store, capability = _paused( + tmp_path, + workflow=workflow, + transition_observation=TransitionObservation( + url=raw_url, + page_title=raw_title, + page_count=1, + ), + ) + serialized = (run / "attended_capability.json").read_text() + assert raw_url not in serialized + assert raw_title not in serialized + assert "Jane Roe" not in serialized + assert capability.transition_baseline.url_digest.startswith("hmac-sha256:") + assert capability.transition_baseline.title_digest.startswith("hmac-sha256:") + assert capability.transition_baseline.page_count == 1 + assert capability.allowed_actions == ("continue", "teach", "escalate") + store = AttendedActionStore(run) + assert store.transition_value_digest("url", raw_url) == ( + capability.transition_baseline.url_digest + ) + assert store.read() == capability + + +@pytest.mark.parametrize( + ("kind", "baseline", "attribute", "changed", "unchanged"), + [ + ( + PostconditionKind.URL_CHANGED, + TransitionObservation(url="https://payer.example/login"), + "url", + "https://payer.example/home", + "https://payer.example/login", + ), + ( + PostconditionKind.TITLE_CHANGED, + TransitionObservation(page_title="Sign in"), + "page_title", + "Eligibility", + "Sign in", + ), + ( + PostconditionKind.NEW_TAB_OPENED, + TransitionObservation(page_count=1), + "page_count", + 2, + 1, + ), + ], +) +def test_signed_relative_transition_confirms_common_human_redirects( + tmp_path, kind, baseline, attribute, changed, unchanged +): + workflow = Workflow( + name=f"relative-{kind.value}", + steps=[ + Step( + id="human", + intent="complete human challenge", + action=ActionKind.KEY, + key="A", + expect=[Postcondition(kind=kind)], + ) + ], + ) + _workflow, _bundle, run, _store, capability = _paused( + tmp_path / "changed", + workflow=workflow, + transition_observation=baseline, + ) + backend = FakeBackend() + setattr(backend, attribute, changed) + accepted = execute_attended_action( + run, + _request(capability, key=f"relative-{kind.value}-changed"), + operator="staff", + executor=BoundAttendedExecutor( + lambda _manifest: Replayer( + backend, vision=FakeVision(), poll_interval_s=0.0 + ) + ), + ) + assert accepted.status == "completed" + assert not backend.actions + + _workflow, _bundle, run, store, capability = _paused( + tmp_path / "unchanged", + workflow=workflow, + transition_observation=baseline, + ) + backend = FakeBackend() + setattr(backend, attribute, unchanged) + refused = execute_attended_action( + run, + _request(capability, key=f"relative-{kind.value}-unchanged"), + operator="staff", + executor=BoundAttendedExecutor( + lambda _manifest: Replayer( + backend, vision=FakeVision(), poll_interval_s=0.0 + ) + ), + ) + assert refused.status == "refused" + assert "unchanged" in refused.message + assert store.read_pending() is not None + assert not backend.actions + + +def test_relative_continue_is_not_advertised_without_signed_baseline(tmp_path): + workflow = Workflow( + name="relative-no-baseline", + steps=[ + Step( + id="human", + intent="complete login", + action=ActionKind.KEY, + key="A", + expect=[Postcondition(kind=PostconditionKind.URL_CHANGED)], + ) + ], + ) + _workflow, _bundle, run, _store, capability = _paused(tmp_path, workflow=workflow) + assert capability.allowed_actions == ("teach", "escalate") + with pytest.raises(AttendedActionRefused, match="does not allow"): + execute_attended_action( + run, + _request(capability, key="relative-missing-baseline"), + operator="staff", + executor=_ResultExecutor(), + ) + + +def test_durable_halt_automatically_captures_protected_transition_baseline(tmp_path): + raw_url = "https://payer.example/member/Jane-Roe-ABC123" + workflow = Workflow( + name="auto-baseline", + steps=[ + Step( + id="human", + intent="complete challenge", + action=ActionKind.KEY, + key="A", + expect=[ + Postcondition( + kind=PostconditionKind.URL_CHANGED, + timeout_s=0.01, + ) + ], + ) + ], + ) + bundle = tmp_path / "bundle" + run = tmp_path / "run" + workflow.save(bundle) + backend = FakeBackend() + backend.url = raw_url + report = Replayer( + backend, + vision=FakeVision(), + durable=True, + poll_interval_s=0.0, + ).run(workflow, bundle_dir=bundle, run_dir=run) + assert report.success is False + capability = AttendedActionStore(run).read() + assert capability.transition_baseline.url_digest is not None + assert "continue" in capability.allowed_actions + assert raw_url not in (run / "attended_capability.json").read_text() + + +def test_program_pause_never_advertises_generic_continue_or_skip(tmp_path): + workflow, _bundle, run, store, _first = _paused(tmp_path) + pending = store.read_pending() + assert pending is not None + program_pending = pending.model_copy( + update={ + "program": True, + "state_id": "challenge-state", + "created_at": "2026-07-18T13:30:00+00:00", + } + ) + store.write_pending(program_pending) + capability = issue_attended_capability( + run, + store=store, + pending=program_pending, + workflow=workflow, + result=StepResult( + step_id="challenge-state", + intent="complete challenge", + ok=False, + error="MFA required", + ), + transition_observation=TransitionObservation(url="https://payer.example/mfa"), + ) + assert capability.allowed_actions == ("teach", "escalate") + + +def test_bound_executor_serializes_its_shared_live_session(tmp_path): + _workflow, _bundle, run, _store, capability = _paused(tmp_path) + executor = BoundAttendedExecutor(lambda _manifest: pytest.fail("factory called")) + approval = ApprovalRecord( + approver="staff", + resolution="completed by operator", + bundle_version=capability.bundle_version, + workflow_name=capability.workflow_name, + run_dir=str(run), + ) + assert executor._live_session_lock.acquire(blocking=False) + try: + result = executor.continue_run(run, capability, approval) + finally: + executor._live_session_lock.release() + assert result.status == "refused" + assert "serving another attended action" in result.message + + +def test_repeated_halt_on_same_step_gets_a_new_exact_pause_capability(tmp_path): + workflow, _bundle, run, store, first = _paused(tmp_path) + pending = store.read_pending() + assert pending is not None + repeated = pending.model_copy(update={"created_at": "2026-07-18T13:00:00+00:00"}) + store.write_pending(repeated) + second = issue_attended_capability( + run, + store=store, + pending=repeated, + workflow=workflow, + result=StepResult( + step_id="human", + intent="press A", + ok=False, + error="Please verify you are human", + ), + ) + assert second.pause_id != first.pause_id + assert second.pause_digest != first.pause_digest + history = json.loads((run / "attended_capability_history.json").read_text()) + assert [item["pause_id"] for item in history] == [first.pause_id] + with pytest.raises(AttendedActionRefused, match="stale"): + execute_attended_action( + run, + _request(first), + operator="staff", + executor=_ResultExecutor(), + ) + + +def test_tampered_capability_and_stale_page_refuse(tmp_path): + _workflow, _bundle, run, _store, capability = _paused(tmp_path) + path = run / "attended_capability.json" + raw = json.loads(path.read_text()) + raw["step_id"] = "attacker" + path.write_text(json.dumps(raw)) + with pytest.raises(AttendedActionRefused, match="signature"): + AttendedActionStore(run).read() + + # Rebuild and present a stale UI digest. + other = tmp_path / "other" + _workflow, _bundle, run, _store, capability = _paused(other) + stale = _request(capability).model_copy( + update={"capability_digest": "sha256:" + "0" * 64} + ) + with pytest.raises(AttendedActionRefused, match="stale"): + execute_attended_action( + run, stale, operator="staff", executor=_ResultExecutor() + ) + + +def test_capability_cannot_be_replayed_into_another_run(tmp_path): + _workflow, _bundle, run, _store, capability = _paused(tmp_path / "one") + copied = tmp_path / "two" / "run" + shutil.copytree(run, copied) + copied_store = CheckpointStore(copied) + manifest = copied_store.read_manifest() + assert manifest is not None + copied_store.write_manifest(manifest.model_copy(update={"run_id": "other-run"})) + with pytest.raises(AttendedActionRefused, match="transition binding"): + execute_attended_action( + copied, + _request(capability), + operator="staff", + executor=_ResultExecutor(), + ) + + +def test_bundle_revision_change_refuses_before_executor(tmp_path): + workflow, bundle, run, _store, capability = _paused(tmp_path) + workflow.steps.append(_step("changed", "C")) + workflow.save(bundle) + executor = _ResultExecutor() + with pytest.raises(Exception, match="bundle"): + execute_attended_action( + run, + _request(capability), + operator="staff", + executor=executor, + ) + assert executor.calls == 0 + + +def test_expired_capability_refuses_before_executor(tmp_path): + _workflow, _bundle, run, _store, capability = _paused(tmp_path) + executor = _ResultExecutor() + after_expiry = datetime.fromisoformat(capability.expires_at) + timedelta(seconds=1) + with pytest.raises(Exception, match="expired"): + execute_attended_action( + run, + _request(capability), + operator="staff", + executor=executor, + now=after_expiry, + ) + assert executor.calls == 0 + + +def test_same_request_is_idempotent_and_conflicting_reuse_refuses(tmp_path): + _workflow, _bundle, run, _store, capability = _paused(tmp_path) + executor = _ResultExecutor() + request = _request(capability) + first = execute_attended_action(run, request, operator="staff", executor=executor) + second = execute_attended_action(run, request, operator="staff", executor=executor) + assert first == second + assert executor.calls == 1 + conflict = request.model_copy( + update={"action": "skip", "disposition": "not_applicable"} + ) + with pytest.raises(AttendedActionRefused, match="different request"): + execute_attended_action(run, conflict, operator="staff", executor=executor) + + +def test_crash_after_delivery_started_becomes_uncertain_and_never_retries(tmp_path): + _workflow, _bundle, run, _store, capability = _paused(tmp_path) + + class Explodes(_ResultExecutor): + def continue_run(self, run_dir, capability, approval): + self.calls += 1 + raise RuntimeError("worker died after delivery boundary") + + executor = Explodes() + request = _request(capability) + with pytest.raises(RuntimeError): + execute_attended_action(run, request, operator="staff", executor=executor) + statuses = [ + item["status"] + for item in json.loads((run / "attended_decisions.json").read_text())[ + "decisions" + ] + ] + assert statuses == ["prepared", "delivery_started", "delivery_uncertain"] + with pytest.raises(AttendedActionRefused, match="automatic retry"): + execute_attended_action(run, request, operator="staff", executor=executor) + with pytest.raises(AttendedActionRefused, match="another request"): + execute_attended_action( + run, + request.model_copy(update={"idempotency_key": "request-key-0002"}), + operator="staff", + executor=executor, + ) + assert executor.calls == 1 + + +def test_challenge_payload_has_no_answer_code_or_raw_path_surface(): + with pytest.raises(ValidationError): + AttendedActionRequest.model_validate( + { + "capability_digest": "sha256:" + "0" * 64, + "idempotency_key": "request-key-0001", + "action": "continue", + "captcha_answer": "solve-me", + } + ) + with pytest.raises(ValidationError): + AttendedActionRequest.model_validate( + { + "capability_digest": "sha256:" + "0" * 64, + "idempotency_key": "request-key-0001", + "action": "teach", + "fix_path": "../../secret.json", + } + ) + + +def test_all_actions_require_operator_identity_and_matching_disposition(tmp_path): + _workflow, _bundle, run, _store, capability = _paused(tmp_path) + with pytest.raises(Exception, match="authenticated operator"): + execute_attended_action( + run, + _request(capability), + operator="", + executor=_ResultExecutor(), + ) + request = _request(capability).model_copy(update={"disposition": "cannot_complete"}) + with pytest.raises(AttendedActionRefused, match="disposition"): + execute_attended_action( + run, + request, + operator="staff", + executor=_ResultExecutor(), + ) + + +def test_missing_or_insecure_capability_secret_never_recreates_authority(tmp_path): + _workflow, _bundle, run, _store, _capability = _paused(tmp_path / "missing") + secret = run / ".attended_capability.key" + secret.unlink() + with pytest.raises(AttendedActionRefused, match="key is missing"): + AttendedActionStore(run).read() + assert not secret.exists() + + if os.name != "nt": + _workflow, _bundle, run, _store, _capability = _paused(tmp_path / "permissions") + secret = run / ".attended_capability.key" + secret.chmod(0o644) + with pytest.raises(AttendedActionRefused, match="permissions"): + AttendedActionStore(run).read() + + +def test_bound_continue_verifies_then_resumes_after_human_step(tmp_path): + _workflow, _bundle, run, store, capability = _paused(tmp_path) + backends: list[FakeBackend] = [] + + def factory(_manifest): + backend = FakeBackend() + backends.append(backend) + vision = FakeVision() + vision.text_results = { + "DONE": Match(point=(10, 10), region=(0, 0, 20, 20), confidence=1.0) + } + return Replayer(backend, vision=vision, poll_interval_s=0.0) + + decision = execute_attended_action( + run, + _request(capability), + operator="front-desk", + executor=BoundAttendedExecutor(factory), + ) + assert decision.status == "completed" + assert decision.report_success is True + assert len(backends) == 1 # verify and continue the exact live session + assert all(("press", "A") not in backend.actions for backend in backends) + assert ("press", "B") in backends[0].actions + checkpoints = store.checkpoints() + assert [checkpoint.step_id for checkpoint in checkpoints] == ["human", "next"] + assert checkpoints[0].actuation == "human_attended" + assert store.read_pending() is None + manifest = store.read_manifest() + assert manifest is not None and manifest.run_id == "run-instance-a" + + +def test_continue_refuses_live_postcondition_failure_without_actuation(tmp_path): + _workflow, _bundle, run, store, capability = _paused(tmp_path) + backends: list[FakeBackend] = [] + + def factory(_manifest): + backend = FakeBackend() + backends.append(backend) + return Replayer(backend, vision=FakeVision(), poll_interval_s=0.0) + + decision = execute_attended_action( + run, + _request(capability, key="request-key-refuse1"), + operator="staff", + executor=BoundAttendedExecutor(factory), + ) + assert decision.status == "refused" + assert decision.report_success is False + assert all(not backend.actions for backend in backends) + assert store.read_pending() is not None + + +def test_continue_that_halts_later_rotates_to_the_new_exact_pause(tmp_path): + workflow = Workflow( + name="attended-chain", + steps=[ + _step("human", "A", expect="DONE"), + _step("next", "B", expect="FINISHED"), + ], + ) + _workflow, _bundle, run, store, first = _paused(tmp_path, workflow=workflow) + backend = FakeBackend() + vision = FakeVision() + vision.text_results = { + "DONE": Match(point=(10, 10), region=(0, 0, 20, 20), confidence=1.0) + } + decision = execute_attended_action( + run, + _request(first, key="request-key-chain01"), + operator="staff", + executor=BoundAttendedExecutor( + lambda _manifest: Replayer( + backend, + vision=vision, + poll_interval_s=0.0, + ) + ), + ) + assert decision.status == "halted" + assert ("press", "A") not in backend.actions + assert ("press", "B") in backend.actions + pending = store.read_pending() + assert pending is not None and pending.step_id == "next" + second = AttendedActionStore(run).read() + assert second.pause_id != first.pause_id + assert second.step_id == "next" + history = json.loads((run / "attended_capability_history.json").read_text()) + assert [item["pause_id"] for item in history] == [first.pause_id] + + +def test_continue_refuses_effect_that_needs_missing_delivery_baseline(tmp_path): + effectful = Workflow( + name="effectful", + steps=[ + Step( + id="human", + intent="human save", + action=ActionKind.KEY, + key="A", + effects=[ + Effect( + kind=EffectKind.RECORD_WRITTEN, + match={"id": "row-1"}, + count_new_only=True, + ) + ], + ) + ], + ) + _workflow, _bundle, run, store, capability = _paused(tmp_path, workflow=effectful) + + class CurrentRecords: + substrate = "fake" + + def capture_pre_state(self, context=None): + return EffectState( + substrate="fake", + reachable=True, + records=[{"id": "row-1"}], + ) + + def verify(self, expected, before, context=None): + raise AssertionError("attended readback must not reuse delivery verify") + + def factory(_manifest): + return Replayer( + FakeBackend(), + vision=FakeVision(), + effect_verifier=CurrentRecords(), + poll_interval_s=0.0, + ) + + assert "continue" not in capability.allowed_actions + with pytest.raises(AttendedActionRefused, match="does not allow"): + execute_attended_action( + run, + _request(capability, key="request-key-effect1"), + operator="staff", + executor=BoundAttendedExecutor(factory), + ) + assert store.read_pending() is not None + + +def test_continue_confirms_absolute_effect_from_current_record_readback(tmp_path): + effectful = Workflow( + name="absolute-effect", + steps=[ + Step( + id="human", + intent="human save", + action=ActionKind.KEY, + key="A", + effects=[ + Effect( + kind=EffectKind.RECORD_WRITTEN, + match={"id": "row-1"}, + forbid_collateral_loss=False, + ) + ], + ) + ], + ) + _workflow, _bundle, run, store, capability = _paused(tmp_path, workflow=effectful) + assert "continue" in capability.allowed_actions + + class CurrentRecords: + substrate = "fake" + + def capture_pre_state(self, context=None): + return EffectState( + substrate="fake", + reachable=True, + records=[{"id": "row-1"}], + ) + + def verify(self, expected, before, context=None): + raise AssertionError("attended readback must not reuse delivery verify") + + backend = FakeBackend() + decision = execute_attended_action( + run, + _request(capability, key="request-key-absolute-effect"), + operator="staff", + executor=BoundAttendedExecutor( + lambda _manifest: Replayer( + backend, + vision=FakeVision(), + effect_verifier=CurrentRecords(), + poll_interval_s=0.0, + ) + ), + ) + assert decision.status == "completed" + assert store.checkpoints()[0].effect_verified is True + assert not backend.actions + + +def test_skip_requires_declared_nonconsequential_skip_semantics(tmp_path): + generic, _bundle, run, store, capability = _paused(tmp_path / "generic") + executor = BoundAttendedExecutor( + lambda _manifest: Replayer( + FakeBackend(), vision=FakeVision(), poll_interval_s=0.0 + ) + ) + assert "skip" not in capability.allowed_actions + with pytest.raises(AttendedActionRefused, match="does not allow"): + execute_attended_action( + run, + _request(capability, action="skip", key="request-key-skip1"), + operator="staff", + executor=executor, + ) + assert store.read_pending() is not None + assert generic.steps[0].guard is None + + optional = Workflow( + name="optional", + steps=[ + Step( + id="optional", + intent="optional dismissal", + action=ActionKind.KEY, + key="A", + guard=Guard( + predicate=Predicate( + kind=PredicateKind.TEXT_PRESENT, text="OPTIONAL" + ), + on_unmet="skip", + ), + ), + _step("next", "B"), + ], + ) + _workflow, _bundle, run, store, capability = _paused( + tmp_path / "optional", workflow=optional + ) + decision = execute_attended_action( + run, + _request(capability, action="skip", key="request-key-skip2"), + operator="staff", + executor=executor, + ) + assert decision.status == "completed" + assert store.checkpoints()[0].skipped is True + + +def test_teach_and_escalate_are_audited_without_actuation(tmp_path): + _workflow, _bundle, run, store, capability = _paused(tmp_path) + teach = execute_attended_action( + run, + AttendedActionRequest( + capability_digest=capability.digest, + idempotency_key="request-key-teach", + action="teach", + disposition="teach_requested", + ), + operator="staff", + ) + assert teach.status == "needs_demonstration" + assert "identity-evidence" in teach.message + escalated = execute_attended_action( + run, + AttendedActionRequest( + capability_digest=capability.digest, + idempotency_key="request-key-escalate", + action="escalate", + disposition="needs_assistance", + ), + operator="staff", + ) + assert escalated.status == "escalated" + assert store.read_pending() is not None + + +def test_encrypted_pause_uses_environment_key_and_protected_capability_secret( + tmp_path, monkeypatch +): + key = "correct horse battery staple" + workflow = Workflow( + name="sealed", + steps=[_step("human", "A", expect="DONE")], + ) + bundle = tmp_path / "bundle" + run = tmp_path / "run" + workflow.save(bundle, encrypt=True, key=key) + store = CheckpointStore(run, key=key) + store.write_manifest( + RunManifest( + run_id="sealed-run", + workflow_name=workflow.name, + bundle_dir=str(bundle), + ) + ) + pending = PendingEscalation( + workflow_name=workflow.name, + step_index=0, + step_id="human", + category="human_required", + reason="MFA required", + ) + store.write_pending(pending) + capability = issue_attended_capability( + run, + store=store, + pending=pending, + workflow=workflow, + result=StepResult( + step_id="human", intent="press A", ok=False, error="MFA required" + ), + ) + monkeypatch.setenv("OPENADAPT_BUNDLE_KEY", key) + decision = execute_attended_action( + run, + AttendedActionRequest( + capability_digest=capability.digest, + idempotency_key="request-key-sealed", + action="escalate", + disposition="needs_assistance", + ), + operator="staff", + ) + assert decision.status == "escalated" + assert (run / "pending_escalation.json.enc").is_file() + assert not (run / "pending_escalation.json").exists() + assert (run / ".attended_capability.key").stat().st_mode & 0o077 == 0 + + +def test_lease_refuses_concurrent_or_crashed_delivery(tmp_path): + _workflow, _bundle, run, _store, capability = _paused(tmp_path) + request = _request(capability) + store = AttendedActionStore(run) + with store.lease(request): + with pytest.raises(AttendedActionRefused, match="already in progress"): + with store.lease(request): + pass + + expired = { + "request_digest": "sha256:" + "0" * 64, + "idempotency_key": "old-request-key", + "acquired_at": "2020-01-01T00:00:00+00:00", + "expires_at": "2020-01-01T00:01:00+00:00", + } + store.lease_path.write_text(json.dumps(expired)) + with pytest.raises(AttendedActionRefused, match="delivery is uncertain"): + with store.lease( + request, + now=datetime(2026, 1, 1, tzinfo=timezone.utc), + ): + pass + + +def test_attended_http_action_requires_auth_csrf_and_exact_capability( + tmp_path, monkeypatch +): + _workflow, bundle, run, _store, capability = _paused(tmp_path) + monkeypatch.setattr( + "openadapt_flow.console.app._local_operator_identity", lambda: "staff" + ) + executor = _ResultExecutor() + app = create_app( + bundle.parent, + run.parent, + allow_actions=True, + attend=True, + attended_executor=executor, + ) + unauthenticated = TestClient(app, base_url="http://127.0.0.1") + assert unauthenticated.get("/api/attention").status_code == 401 + client = TestClient( + app, + base_url="http://127.0.0.1", + headers={ + "Authorization": f"Bearer {app.state.console_access_token}", + "Origin": "http://127.0.0.1", + "X-OpenAdapt-CSRF": app.state.console_csrf_token, + }, + ) + item = client.get("/api/attention").json()[0] + health = client.get("/api/health").json() + assert health["attended_decisions_ready"] is True + assert health["attended_actions_ready"] is True + assert item["capability"]["digest"] == capability.digest + assert "expected_next_transition" not in item["capability"] + payload = _request(capability, key="request-key-http1").model_dump() + response = client.post( + f"/api/attention/{item['id']}/actions/continue", + json=payload, + ) + assert response.status_code == 200 + assert response.json()["status"] == "completed" + assert set(response.json()) == { + "action", + "status", + "message", + "report_success", + } + assert executor.calls == 1 + + wrong_path = client.post( + f"/api/attention/{item['id']}/actions/skip", + json={**payload, "idempotency_key": "request-key-http2"}, + ) + assert wrong_path.status_code == 400 + + +def test_attended_http_can_teach_or_escalate_without_live_executor( + tmp_path, monkeypatch +): + _workflow, bundle, run, _store, capability = _paused(tmp_path) + monkeypatch.setattr( + "openadapt_flow.console.app._local_operator_identity", lambda: "staff" + ) + app = create_app( + bundle.parent, + run.parent, + allow_actions=True, + attend=True, + ) + client = TestClient( + app, + base_url="http://127.0.0.1", + headers={ + "Authorization": f"Bearer {app.state.console_access_token}", + "Origin": "http://127.0.0.1", + "X-OpenAdapt-CSRF": app.state.console_csrf_token, + }, + ) + health = client.get("/api/health").json() + assert health["attended_decisions_ready"] is True + assert health["attended_actions_ready"] is False + item = client.get("/api/attention").json()[0] + response = client.post( + f"/api/attention/{item['id']}/actions/teach", + json={ + "capability_digest": capability.digest, + "idempotency_key": "request-key-http-teach", + "action": "teach", + "disposition": "teach_requested", + }, + ) + assert response.status_code == 200 + assert response.json()["status"] == "needs_demonstration" diff --git a/tests/test_console.py b/tests/test_console.py index cfa706b1..14f9e824 100644 --- a/tests/test_console.py +++ b/tests/test_console.py @@ -25,7 +25,9 @@ import socket import threading import time +from contextlib import nullcontext from pathlib import Path +from types import SimpleNamespace import pytest @@ -292,6 +294,8 @@ def test_health_reports_read_only(console_env): body = _client(console_env).get("/api/health").json() assert body["status"] == "ok" assert body["read_only"] is True + assert body["attended_decisions_ready"] is False + assert body["attended_actions_ready"] is False assert "bundles_root" not in body assert "runs_root" not in body @@ -1387,17 +1391,10 @@ def test_attention_preserves_typed_effect_category_over_incidental_marker(consol assert run["human_required"] is False -def test_attend_mode_exposes_no_browser_mutations_even_when_requested( - console_env, monkeypatch +def test_attend_mode_preserves_normal_console_and_requires_pause_capability( + console_env, ): - from openadapt_flow.console import actions as actions_mod - _make_human_required_pause(console_env) - - def should_not_execute(_args): - raise AssertionError("attended mode must never execute a CLI action") - - monkeypatch.setattr(actions_mod, "_run_cli", should_not_execute) app = create_app( console_env["bundles"], console_env["runs"], @@ -1413,7 +1410,7 @@ def should_not_execute(_args): "X-OpenAdapt-CSRF": app.state.console_csrf_token, }, ) - assert client.get("/api/health").json()["read_only"] is True + assert client.get("/api/health").json()["read_only"] is False human = next( item for item in client.get("/api/attention").json() if item["human_required"] @@ -1423,31 +1420,32 @@ def should_not_execute(_args): ) workflow_id = _workflow_id(client, n_steps=3) - assert client.get(f"/api/runs/{paused_id}/actions").json() == [] - assert client.get(f"/api/workflows/{workflow_id}/actions").json() == [] - for path, payload in ( - (f"/api/runs/{paused_id}/actions/approve", {}), - (f"/api/workflows/{workflow_id}/actions/certify", {}), - ("/api/skills/not-a-skill/actions/promote", {}), - ): - response = client.post(path, json=payload) - assert response.status_code == 404 - assert response.json() == { - "detail": "actions are unavailable in attended read-only mode" - } - assert "command" not in response.text + # Attended mode is additive: the normal console catalogs remain available. + assert client.get(f"/api/runs/{paused_id}/actions").json() + assert client.get(f"/api/workflows/{workflow_id}/actions").json() route_paths = {getattr(route, "path", "") for route in app.routes} - assert "/api/attention/{run_id:path}/actions/approve-human-step" not in route_paths - removed = client.post( - f"/api/attention/{human['id']}/actions/approve-human-step", json={} + assert "/api/attention/{run_id:path}/actions/{action_id}" in route_paths + # This manually assembled legacy pause has no engine-issued capability; + # enabling actions cannot manufacture authority for it. + refused = client.post( + f"/api/attention/{human['id']}/actions/continue", + json={ + "capability_digest": "sha256:" + ("0" * 64), + "idempotency_key": "request-key-legacy-pause", + "action": "continue", + "disposition": "completed_by_operator", + }, ) - assert removed.status_code in {404, 405} + assert refused.status_code == 409 + assert "capability" in refused.text.lower() script = client.get("/static/console.js").text - assert "data-approve-human" not in script - assert "approve-human-step" not in script - assert "I completed the human step" not in script + assert "I fixed it — Continue" in script + assert "Teach the fix" in script + assert "!HEALTH.attend" not in script + assert "captcha_answer" not in script + assert "verification_code" not in script def test_symlinked_pause_never_enters_attention_queue_or_leaks(console_env, tmp_path): @@ -1477,7 +1475,7 @@ def test_symlinked_pause_never_enters_attention_queue_or_leaks(console_env, tmp_ assert run["paused"] is False -def test_attend_is_attention_first_and_hard_read_only(console_env, monkeypatch): +def test_attend_is_attention_first_and_actions_are_explicit(console_env, monkeypatch): from openadapt_flow.__main__ import build_parser args = build_parser().parse_args(["console", "--attend", "--allow-actions"]) @@ -1489,10 +1487,16 @@ def test_attend_is_attention_first_and_hard_read_only(console_env, monkeypatch): def fake_serve(_bundles, _runs, _skills, **kwargs): served.update(kwargs) + executor = object() + monkeypatch.setattr( + "openadapt_flow.__main__._attended_executor_from_args", + lambda _args: nullcontext(executor), + ) monkeypatch.setattr("openadapt_flow.console.server.serve", fake_serve) assert args.func(args) == 0 assert served["attend"] is True - assert served["allow_actions"] is False + assert served["allow_actions"] is True + assert served["attended_executor"] is executor app = create_app( console_env["bundles"], @@ -1507,10 +1511,138 @@ def fake_serve(_bundles, _runs, _skills, **kwargs): ) health = client.get("/api/health").json() assert health["attend"] is True - assert health["read_only"] is True + assert health["read_only"] is False + assert health["attended_decisions_ready"] is True + assert health["attended_actions_ready"] is False script = client.get("/static/console.js").text html = client.get("/").text assert 'HEALTH.attend ? "#/attention"' in script assert "Needs Attention" in html assert "