diff --git a/docs/DECISION_DELIVERY.md b/docs/DECISION_DELIVERY.md index bc2014b8..61177d20 100644 --- a/docs/DECISION_DELIVERY.md +++ b/docs/DECISION_DELIVERY.md @@ -89,13 +89,48 @@ NAT and on an ordinary practice broadband line. # context_tier defaults to remote_closed_context ``` -2. Staff open `app.openadapt.ai` on their phone and sign in. It is a web page; +2. Run the attended console with the lane on. Desktop does this for the + operator; the equivalent command by hand is: + + ``` + OPENADAPT_RUNNER_TOKEN=oar_... openadapt-flow console \ + --attend --allow-actions --remote-decisions --config deployment.yaml + ``` + +3. Staff open `app.openadapt.ai` on their phone and sign in. It is a web page; there is no app to install. That is the whole list. Nobody terminates TLS, because the only TLS involved is the runner's outbound connection to a public host with an ordinary public certificate. +### What actually runs, once step 2 is on + +`console.decision_supervisor.DecisionSupervisor` owns the whole `runs` root +rather than one run, which is what makes the lane automatic instead of a library +somebody has to call: + +- Every cycle it publishes **every** currently open attended pause, so a halt + becomes answerable on a phone without anyone doing anything. +- When an answer comes back it re-scans and requires the relayed `task_id` *and* + `capability_digest` to match a pause that is open **right now**. A decision + that matches none is acknowledged `stale` and never executed. This matters the + moment two runs are halted at once, which is the ordinary case: revalidation + would catch a *wrong* answer, but not a *correct* answer applied to the wrong + run. +- A relay whose deadline has passed, or whose deadline cannot be parsed, is + acknowledged `expired`. An unreadable deadline is not a licence to act. +- It runs as a daemon thread inside the attended console, because that process + already owns the deployment-bound action service, and because + `execute_attended_action` takes a single-flight lease over the pause — so an + answer from the phone and one from the local browser cannot both execute. + +Every refusal at start-up is a hard exit, never a disabled feature. An operator +who passed `--remote-decisions` and silently got a loopback-only console would +believe a phone can answer a halt while nothing is listening for one. Missing +runner token, remote issuance not enabled, a read-only console, or a plaintext +control-plane origin each stop the console rather than degrade it. + ### Runner-local portal — full fidelity, on the practice's own terms `openadapt-desktop`'s decision portal serves `local_full`, including protected @@ -243,6 +278,22 @@ says the frame is triage context rather than the source of any answer. Building the crypto before a pilot has told us the closed context is insufficient would be building the expensive half of the answer first. +## What is not done yet + +Stated plainly, because a half-wired lane that reads as finished is worse than +an honest gap. + +- **Desktop does not pass `--remote-decisions` yet.** `engine/portal/service.py` + spawns `openadapt-flow console --attend --allow-actions`; adding the flag is a + small change, but the installer bundles a pinned frozen Flow, so the pin has + to move to a release containing the flag first. Until then the lane is + available from the CLI and not from the Desktop toggle. +- **The hosted side must be deployed.** The control plane has to accept the two + new projection fields before the lane carries context; without that it still + publishes and answers, at `remote_identifiers`. +- **`max_remote_decision_tier` is not yet discriminating** (see above). It + becomes load-bearing when a tier exists that some profile must refuse. + ## What is deliberately not claimed `DecisionRelay` never says **delivered**. A successful POST proves the control diff --git a/docs/EXCEPTION_INBOX.md b/docs/EXCEPTION_INBOX.md index ac857a3f..726527c3 100644 --- a/docs/EXCEPTION_INBOX.md +++ b/docs/EXCEPTION_INBOX.md @@ -78,6 +78,16 @@ capability, allowed operation, expected transition, expiry, and idempotency scope. A signed task is presentation integrity, not execution authority, and a relayed answer is neither. +`console.decision_supervisor.DecisionSupervisor` drives that transport for a +whole `runs` root, so an open pause is published without a caller, and an +answered decision is matched back to the pause it was minted from — by task id +and capability digest, re-read at decision time — before it is executed. A +decision that matches no currently open pause is acknowledged `stale`; one whose +deadline has passed or cannot be parsed is acknowledged `expired`. Neither is +executed. `console --attend --allow-actions --remote-decisions` starts it beside +the loopback server, and every misconfiguration stops the console rather than +disabling the lane silently. + After a returned Continue, OpenAdapt observes a newly settled live frame, rechecks the human-completed postcondition and configured effect, proves the next state/target and armed identity, and commits the exact pause transition. diff --git a/openadapt_flow/__main__.py b/openadapt_flow/__main__.py index cb4c926e..1b230a32 100644 --- a/openadapt_flow/__main__.py +++ b/openadapt_flow/__main__.py @@ -4738,6 +4738,28 @@ def _repair_store_flag(rp: argparse.ArgumentParser) -> None: "fresh attended verification and deterministic continuation" ), ) + p.add_argument( + "--remote-decisions", + action="store_true", + help=( + "Publish open attended pauses to the hosted decision surface over " + "an OUTBOUND-ONLY connection, so a paired phone can answer them " + "from anywhere without any inbound port, certificate, or reverse " + "proxy. Requires --attend --allow-actions, a --config whose " + "human_decisions.remote is enabled with an exact tenant and runner, " + "and OPENADAPT_RUNNER_TOKEN. Only the PHI-free signed task and the " + "closed halt context cross the wire; screenshots never leave" + ), + ) + p.add_argument( + "--remote-decision-host", + default=None, + metavar="URL", + help=( + "Hosted control-plane origin for --remote-decisions " + "(default: the configured hosted host, https://app.openadapt.ai)" + ), + ) _add_backend_flags(p) _add_deployment_flags(p) p.set_defaults(func=_cmd_console) @@ -4867,6 +4889,69 @@ def _attended_service_from_args(args: argparse.Namespace) -> Iterator[Any]: raise SystemExit(str(exc)) from exc +def _decision_supervisor_from_args( + args: argparse.Namespace, attended_service: Any +) -> Any: + """Build the outbound decision lane, or fail loudly. Never returns silently. + + Every refusal here is a ``SystemExit`` rather than a disabled feature. An + operator who asked for ``--remote-decisions`` and got a loopback-only + console instead would believe a phone can answer a halt when nothing is + listening for one, which is the exact "looks like it works" failure this + lane must not have. + """ + if not getattr(args, "remote_decisions", False): + return None + if not (args.attend and args.allow_actions): + raise SystemExit( + "--remote-decisions requires --attend --allow-actions: a remote " + "answer is executed through the same governed attended path as a " + "local one, and that path is not available in a read-only console" + ) + if attended_service is None: # pragma: no cover - guarded by the check above + raise SystemExit("--remote-decisions requires an attended action service") + + from openadapt_flow.console.decision_relay import ( + DecisionRelay, + HttpxRelayTransport, + RelayRefused, + resolve_runner_token, + ) + from openadapt_flow.console.decision_supervisor import ( + DecisionSupervisor, + DecisionSupervisorThread, + ) + from openadapt_flow.hosted import HostedError, resolve_host + + try: + cfg, _effects_cfg, _actuation_cfg = _deployment_sections(args) + except (FileNotFoundError, ValueError) as exc: + raise SystemExit(str(exc)) from exc + remote = cfg.human_decisions.remote + if not remote.enabled: + raise SystemExit( + "--remote-decisions requires human_decisions.remote.enabled in " + "--config, with the exact tenant_id and runner_id the control " + "plane issued for this machine" + ) + try: + token = resolve_runner_token() + origin = resolve_host(getattr(args, "remote_decision_host", None)) + relay = DecisionRelay( + HttpxRelayTransport(origin, token), token=token, deployment=cfg + ) + except (RelayRefused, HostedError) as exc: + raise SystemExit(str(exc)) from exc + supervisor = DecisionSupervisor( + args.runs, relay=relay, deployment=cfg, executor=attended_service + ) + print( + f" remote decisions: outbound-only to {origin} " + f"(tier {remote.context_tier}; no inbound port, no certificate)" + ) + return DecisionSupervisorThread(supervisor) + + 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 @@ -4898,6 +4983,7 @@ def _cmd_console(args: argparse.Namespace) -> int: allow_actions=allow_actions, attend=args.attend, attended_service=attended_service, + decision_supervisor=_decision_supervisor_from_args(args, attended_service), port=args.port, ) return 0 diff --git a/openadapt_flow/console/decision_relay.py b/openadapt_flow/console/decision_relay.py index 369ff7b3..4a72ed2b 100644 --- a/openadapt_flow/console/decision_relay.py +++ b/openadapt_flow/console/decision_relay.py @@ -85,6 +85,7 @@ AttendedActionExecutor, AttendedActionRefused, AttendedDecision, + AttendedRelayBinding, ) #: Wire schema of a relayed decision, as minted by the hosted control plane. @@ -222,6 +223,18 @@ def relay_signature(self) -> str: def action(self) -> str: return str(self.relay["action"]) + def durable_binding(self) -> AttendedRelayBinding: + """Exact PHI-free fields required to recover a lost acknowledgement.""" + return AttendedRelayBinding( + decision_id=self.decision_id, + relay_digest=self.relay_digest, + relay_signature=self.relay_signature, + idempotency_key=str(self.relay["idempotency_key"]), + capability_digest=str(self.relay["capability_digest"]), + event_sequence=int(self.relay["event_sequence"]), + action=self.action, # type: ignore[arg-type] + ) + class RelayTransport(Protocol): """The one network capability this module needs. @@ -569,6 +582,7 @@ def execute( deployment=self._deployment, principal=principal, executor=executor, + relay_binding=decision.durable_binding(), key=key, ) diff --git a/openadapt_flow/console/decision_supervisor.py b/openadapt_flow/console/decision_supervisor.py new file mode 100644 index 00000000..7c7ab0ea --- /dev/null +++ b/openadapt_flow/console/decision_supervisor.py @@ -0,0 +1,576 @@ +"""Run the outbound decision lane: publish open pauses, execute the answers. + +:mod:`openadapt_flow.console.decision_relay` is a transport. It publishes *one* +pause the caller already resolved, and executes *one* decision against a +``run_dir`` and :class:`~openadapt_flow.console.attention.AttentionItem` the +caller already knows. Nothing in the engine called it, so the hosted lane was +reachable in principle and dead in practice: a dental practice with no reverse +proxy still had no way for a halt to arrive on a phone. + +This module is the missing half. It owns a ``runs`` root rather than one run: + +* it publishes **every** currently open attended pause under that root, so a + halt becomes answerable at ``app.openadapt.ai`` without anyone doing anything; +* it resolves an answered decision **back to the exact pause it was minted + from**, by capability digest, before executing it; and +* it runs as a supervised background thread beside the attended console, which + is the process that already holds the deployment-bound + :class:`~openadapt_flow.runtime.durable.attended_service.AttendedActionService` + a continuation needs. + +Why resolution is by capability digest, not by position +------------------------------------------------------- + +:meth:`DecisionRelay.serve_once` takes the run and item as arguments, which is +correct only when exactly one pause is open. A practice with two runs halted at +once would otherwise execute an answer against whichever pause the caller +happened to be holding. :meth:`DecisionSupervisor.resolve` instead re-scans the +runs root at decision time and requires **both** the relayed ``task_id`` and the +relayed ``capability_digest`` to equal the ones the engine's own signed +capability file produces right now. A decision that matches no open pause is +acknowledged ``stale`` and is never executed. + +That check is not the safety boundary — :func:`execute_remote_attended_action` +re-validates the capability, takes the single-flight lease, re-reads the live +application and re-proves every ``will_recheck`` contract regardless. It is what +keeps a *correct* answer from being applied to the *wrong* run, which +revalidation alone would not catch when both pauses are genuinely open. + +What this module refuses to claim +--------------------------------- + +The vocabulary stays the relay's: ``published``, ``already_published``, +``unknown``. A cycle reports what it observed and nothing more. In particular a +pause whose publish returned ``unknown`` is reported as ``unknown`` and is +**not** described as reachable; the local console remains the authoritative +surface for it. + +Re-publishing after ``unknown`` is deliberate, and it is not a blind retry. +``POST /api/human-decisions/tasks`` is an idempotent upsert keyed on the signed +task: an identical projection returns ``created: false``, and a *divergent* +projection for the same task is rejected by the control plane rather than +overwriting. So the next cycle re-POSTs the same bytes, which either resolves +the uncertainty or leaves it unchanged. The rule the relay states — never retry +an operation whose effect may already have happened *and would happen twice* — +is preserved, because this operation cannot happen twice. +""" + +from __future__ import annotations + +import threading +from dataclasses import dataclass, field +from datetime import datetime, timezone +from pathlib import Path +from typing import Callable, Optional + +from openadapt_flow.console import data +from openadapt_flow.console.attention import ( + AttentionItem, + list_attention, + resolve_attention, +) +from openadapt_flow.console.decision_relay import ( + DecisionRelay, + PublishOutcome, + PublishState, + RelayedDecision, + RelayRefused, +) +from openadapt_flow.deployment import DeploymentConfig +from openadapt_flow.runtime.durable.attended import ( + AttendedActionExecutor, + AttendedActionRefused, + AttendedActionStore, + AttendedDecision, + AttendedRelayAcknowledgement, +) + +#: Seconds a poll waits for an answer before the loop takes another turn. Also +#: the granularity at which :meth:`DecisionSupervisorThread.stop` is observed. +DEFAULT_POLL_WAIT_S = 25.0 + +#: Backoff bounds for a control plane that is refusing or unreachable. A +#: practice's broadband line goes down; the supervisor must not spin on it. +BACKOFF_BASE_S = 2.0 +BACKOFF_CAP_S = 120.0 + +#: Pause after a governed refusal. Not a backoff -- a refusal is an answer, and +#: the transport is healthy -- but a re-delivered decision returns from the poll +#: instantly, so without a floor the loop would spin on one it always refuses. +REFUSAL_PAUSE_S = 1.0 + + +@dataclass(frozen=True) +class OpenPause: + """One durably paused run, resolved from the runs root at scan time.""" + + run_dir: Path + item: AttentionItem + task_id: str + capability_digest: str + + +@dataclass(frozen=True) +class PublishReport: + """What one publish pass observed, per pause. No claim about a person.""" + + #: Pauses the control plane accepted for the first time. + published: tuple[str, ...] = () + #: Pauses the control plane said it already held, observed THIS pass. + already_published: tuple[str, ...] = () + #: Pauses the control plane accepted earlier in this process and that were + #: therefore not re-sent. Kept separate from ``already_published`` on + #: purpose: the supervisor did not ask the control plane about these this + #: pass, so reporting them as an observation would claim something it did + #: not observe. + previously_confirmed: tuple[str, ...] = () + #: Pauses whose POST left the process without a terminal response. These + #: may or may not be visible on a phone and must not be described as + #: reachable. + unknown: tuple[str, ...] = () + #: Pauses that could not be projected at all (no capability, closed pause, + #: remote issuance refused). Not an error; the local console still serves + #: them. + not_projectable: tuple[str, ...] = () + #: Pauses the control plane refused. Recorded rather than raised, so one + #: bad projection cannot make every other halt unreachable. + refused: tuple[str, ...] = () + + @property + def certain_count(self) -> int: + """Pauses known to be answerable, by observation or by prior accept.""" + return ( + len(self.published) + + len(self.already_published) + + len(self.previously_confirmed) + ) + + +@dataclass(frozen=True) +class CycleReport: + """The outcome of one supervisor cycle.""" + + publishes: PublishReport + #: ``None`` when no decision was waiting this cycle. + decision_id: Optional[str] = None + #: What the supervisor told the control plane it did: one of ``accepted``, + #: ``refused``, ``stale``, ``expired``, or ``None`` when there was nothing + #: to acknowledge. + acknowledged: Optional[str] = None + #: The engine decision, when one was executed. + outcome: Optional[AttendedDecision] = None + #: True when this cycle only repeated an acknowledgement for an exact + #: signed decision whose engine outcome was already retained in the run's + #: atomic decision journal. + #: The action is never executed a second time on this path. + reacknowledged: bool = False + + +def _parse_rfc3339(value: object) -> Optional[datetime]: + """Parse a relay timestamp, or ``None`` if it is not one. + + A relay whose ``expires_at`` cannot be parsed is treated as *expired*, not + as open-ended: an unreadable deadline is not a licence to act. + """ + if not isinstance(value, str): + return None + text = value.strip() + if text.endswith(("Z", "z")): + text = f"{text[:-1]}+00:00" + try: + parsed = datetime.fromisoformat(text) + except ValueError: + return None + if parsed.tzinfo is None: + return parsed.replace(tzinfo=timezone.utc) + return parsed + + +class DecisionSupervisor: + """Own the outbound decision lane for every open pause under one root.""" + + def __init__( + self, + runs_root: Path | str, + *, + relay: DecisionRelay, + deployment: DeploymentConfig, + executor: Optional[AttendedActionExecutor] = None, + now: Optional[Callable[[], datetime]] = None, + ) -> None: + self._runs_root = Path(runs_root) + self._relay = relay + self._deployment = deployment + self._executor = executor + self._now = now or (lambda: datetime.now(timezone.utc)) + #: ``(task_id, capability_digest)`` pairs the control plane has already + #: accepted in this process. Not durable on purpose: a restarted + #: supervisor republishes, which is idempotent, rather than trusting a + #: file to say a remote surface still holds something. + self._confirmed: set[tuple[str, str]] = set() + + # -- scanning --------------------------------------------------------- + + def open_pauses(self) -> list[OpenPause]: + """Every durably paused run under the root, with its exact capability. + + The capability file is the engine's own signed record of the pause, so + reading it here costs no signing and cannot mint authority. A run whose + capability is missing, unreadable, or unsigned is simply not an open + pause for this purpose. + """ + resolved: list[OpenPause] = [] + for scanned in list_attention(self._runs_root): + if not scanned.durably_paused: + continue + # `AttentionItem.id` is an opaque scan id, not a directory name. + # `resolve_attention` is the only supported way back to a path, and + # it re-scans, so a run that closed between the two calls resolves + # to None rather than to a stale directory. + located = resolve_attention(self._runs_root, scanned.id) + if located is None: + continue + run_dir, item = located + if not item.durably_paused: + continue + try: + capability = AttendedActionStore(run_dir).read() + except AttendedActionRefused: + continue + resolved.append( + OpenPause( + run_dir=run_dir, + item=item, + task_id=f"task_{capability.pause_id}", + capability_digest=capability.digest, + ) + ) + return resolved + + def resolve(self, decision: RelayedDecision) -> Optional[OpenPause]: + """The open pause this decision was minted from, or ``None``. + + Both the relayed ``task_id`` and the relayed ``capability_digest`` must + match. The digest alone would be sufficient today; requiring both means + that if a future change ever decouples the two, it shows up as a + decision that resolves to nothing rather than as a silent guess. + + That failure direction is deliberate. A supervisor that cannot resolve + a decision acknowledges it ``stale`` and executes nothing, which is an + outage. A supervisor that resolves it to the wrong open pause executes + a real answer against a real record. The first is recoverable. + """ + relay = decision.relay + task_id = relay.get("task_id") + capability_digest = relay.get("capability_digest") + for pause in self.open_pauses(): + if pause.task_id == task_id and pause.capability_digest == ( + capability_digest + ): + return pause + return None + + def retained_acknowledgement( + self, + decision: RelayedDecision, + ) -> Optional[tuple[Path, AttendedRelayAcknowledgement, AttendedDecision]]: + """Find one exact journaled outcome, including after process restart. + + Completed Continue, Skip, and Reject actions can remove a run from the + open queue. Scan the same bounded, symlink-refusing run root used by the + console, then require the complete signed relay binding and the retained + engine decision digest to verify before any acknowledgement is sent. + """ + if self._runs_root.is_symlink(): + raise RelayRefused( + "the configured runs root is a symlink; retained relay recovery " + "is refused" + ) + binding = decision.durable_binding() + retained: list[tuple[Path, AttendedRelayAcknowledgement, AttendedDecision]] = [] + for run_dir in data._scan(self._runs_root, data._is_run_dir): + store = AttendedActionStore(run_dir) + if not store.decisions_path.is_file(): + continue + try: + matched = store.relay_acknowledgement(binding) + except AttendedActionRefused as exc: + raise RelayRefused(str(exc)) from exc + if matched is not None: + record, outcome = matched + retained.append((run_dir, record, outcome)) + if len(retained) > 1: + raise RelayRefused( + "the relay decision is bound to more than one retained engine outcome" + ) + return retained[0] if retained else None + + # -- one cycle -------------------------------------------------------- + + def publish_open_pauses(self, *, timeout_s: float = 15.0) -> PublishReport: + """Make every open pause answerable from the hosted surface. + + One pause the control plane refuses must not silence the others. A + refusal is recorded per pause and the loop continues, because the + alternative -- letting it propagate -- would leave every OTHER halt in + the practice unreachable on a phone because of one bad projection. + """ + published: list[str] = [] + already: list[str] = [] + unknown: list[str] = [] + not_projectable: list[str] = [] + refused: list[str] = [] + memoized: list[str] = [] + confirmed: set[tuple[str, str]] = set() + for pause in self.open_pauses(): + key = (pause.task_id, pause.capability_digest) + if key in self._confirmed: + # Already accepted at this exact capability, and the signed task + # is a deterministic function of it, so re-POSTing would send + # identical bytes for no new information. A pause can stay open + # for hours; publishing it every cycle is noise, not safety. + memoized.append(pause.task_id) + confirmed.add(key) + continue + try: + outcome: PublishOutcome = self._relay.publish( + pause.run_dir, pause.item, timeout_s=timeout_s + ) + except AttendedActionRefused: + # The pause exists but cannot be projected remotely -- a closed + # pause, or a deployment whose remote issuance this run does not + # satisfy. The local console still serves it. + not_projectable.append(pause.task_id) + continue + except RelayRefused: + refused.append(pause.task_id) + continue + if outcome.state is PublishState.PUBLISHED: + published.append(pause.task_id) + confirmed.add(key) + elif outcome.state is PublishState.ALREADY_PUBLISHED: + already.append(pause.task_id) + confirmed.add(key) + else: + # Uncertain. Deliberately NOT confirmed, so the next cycle + # re-POSTs the identical idempotent projection and either + # resolves the uncertainty or leaves it unchanged. + unknown.append(pause.task_id) + # Rebuilt rather than updated, so a pause that closed stops being + # remembered and a later pause reusing its identity is republished. + self._confirmed = confirmed + return PublishReport( + published=tuple(published), + already_published=tuple(already), + previously_confirmed=tuple(memoized), + unknown=tuple(unknown), + not_projectable=tuple(not_projectable), + refused=tuple(refused), + ) + + def serve_once( + self, + *, + wait_s: float = DEFAULT_POLL_WAIT_S, + publish: bool = True, + ) -> CycleReport: + """Publish open pauses, then take at most one answered decision. + + A decision that resolves to no currently open pause is acknowledged + ``stale``; one whose relay deadline has passed is acknowledged + ``expired``. Neither is executed. A governed refusal is acknowledged + ``refused`` and re-raised, so the hosted surface can tell the operator + their answer was not accepted rather than leaving them looking at a + decision that appears to have been taken. + """ + publishes = self.publish_open_pauses() if publish else PublishReport() + decision = self._relay.poll(wait_s=wait_s) + if decision is None: + return CycleReport(publishes=publishes) + + retained_ack = self.retained_acknowledgement(decision) + if retained_ack is not None: + run_dir, record, outcome = retained_ack + confirmed = self._relay.acknowledge(decision, record.engine_ack_result) + if confirmed: + AttendedActionStore(run_dir).confirm_relay_acknowledgement( + decision.durable_binding() + ) + return CycleReport( + publishes=publishes, + decision_id=decision.decision_id, + acknowledged=record.engine_ack_result, + outcome=outcome, + reacknowledged=True, + ) + + expires_at = _parse_rfc3339(decision.relay.get("expires_at")) + if expires_at is None or expires_at <= self._now(): + self._relay.acknowledge(decision, "expired") + return CycleReport( + publishes=publishes, + decision_id=decision.decision_id, + acknowledged="expired", + ) + + pause = self.resolve(decision) + if pause is None: + self._relay.acknowledge(decision, "stale") + return CycleReport( + publishes=publishes, + decision_id=decision.decision_id, + acknowledged="stale", + ) + + try: + outcome = self._relay.execute( + pause.run_dir, pause.item, decision, executor=self._executor + ) + except AttendedActionRefused: + self._relay.acknowledge(decision, "refused") + raise + result = "accepted" if outcome.status != "refused" else "refused" + # ``DecisionRelay.execute`` atomically appended the completed local + # result and its exact signed relay binding to the run's existing + # decision journal before this network acknowledgement. A restart can + # therefore re-acknowledge without executing the action again. + confirmed = self._relay.acknowledge(decision, result) + if confirmed: + AttendedActionStore(pause.run_dir).confirm_relay_acknowledgement( + decision.durable_binding() + ) + return CycleReport( + publishes=publishes, + decision_id=decision.decision_id, + acknowledged=result, + outcome=outcome, + ) + + +@dataclass +class SupervisorStats: + """PHI-free counters a caller may log or surface. No identifiers.""" + + cycles: int = 0 + decisions_executed: int = 0 + decisions_refused: int = 0 + decisions_stale: int = 0 + decisions_expired: int = 0 + publish_unknown: int = 0 + consecutive_failures: int = 0 + last_error: Optional[str] = field(default=None) + + +class DecisionSupervisorThread: + """Run a :class:`DecisionSupervisor` beside the attended console. + + The console process already owns the deployment-bound action service a + continuation needs, and + :func:`~openadapt_flow.runtime.durable.attended.execute_attended_action` + takes a single-flight lease over the pause, so a decision arriving from the + phone and one taken in the local browser cannot both execute. That lease is + what makes running this in a thread beside the server correct rather than + merely convenient. + """ + + def __init__( + self, + supervisor: DecisionSupervisor, + *, + wait_s: float = DEFAULT_POLL_WAIT_S, + on_cycle: Optional[Callable[[CycleReport], None]] = None, + sleep: Optional[Callable[[float], None]] = None, + ) -> None: + self._supervisor = supervisor + self._wait_s = wait_s + self._on_cycle = on_cycle + self._stop = threading.Event() + self._thread: Optional[threading.Thread] = None + self._sleep = sleep or self._stop.wait # interruptible by stop() + self.stats = SupervisorStats() + + # -- lifecycle -------------------------------------------------------- + + def start(self) -> None: + if self._thread is not None: + raise RuntimeError("the decision supervisor is already running") + thread = threading.Thread( + target=self.run, name="openadapt-decision-relay", daemon=True + ) + self._thread = thread + thread.start() + + def stop(self, *, timeout_s: float = 5.0) -> None: + """Ask the loop to finish its current cycle and stop. + + A cycle can be inside a long poll, so the thread may outlive this call + by up to the poll wait. It is a daemon thread holding no lease of its + own, and a decision already in flight completes or refuses under the + engine's normal contracts, so an unjoined thread cannot leave a pause + half-decided. + """ + self._stop.set() + thread = self._thread + if thread is not None: + thread.join(timeout=timeout_s) + + @property + def running(self) -> bool: + thread = self._thread + return thread is not None and thread.is_alive() + + # -- the loop --------------------------------------------------------- + + def run(self) -> None: + """Cycle until stopped. Never raises; every outcome is counted.""" + while not self._stop.is_set(): + try: + report = self._supervisor.serve_once(wait_s=self._wait_s) + except AttendedActionRefused as exc: + # A governed refusal was already acknowledged as ``refused``. + # It is an answer, not a transport failure, so it does not raise + # the backoff level -- a refusal must not slow the lane down for + # everyone else. + # + # It does take one short pause, because the acknowledgement can + # itself be uncertain. In that case the decision stays leased + # server-side and is re-delivered immediately, and a poll with a + # decision waiting returns at once: without this pause the loop + # would spin at full speed on a decision it will refuse every + # time. + self.stats.decisions_refused += 1 + self.stats.last_error = type(exc).__name__ + self._sleep(REFUSAL_PAUSE_S) + continue + except RelayRefused as exc: + self.stats.consecutive_failures += 1 + self.stats.last_error = str(exc) + self._sleep(self._backoff_s()) + continue + except Exception as exc: # noqa: BLE001 - the loop must not die + self.stats.consecutive_failures += 1 + self.stats.last_error = type(exc).__name__ + self._sleep(self._backoff_s()) + continue + self._record(report) + if self._on_cycle is not None: + self._on_cycle(report) + + def _record(self, report: CycleReport) -> None: + self.stats.cycles += 1 + self.stats.consecutive_failures = 0 + self.stats.publish_unknown += len(report.publishes.unknown) + if report.acknowledged == "accepted" and not report.reacknowledged: + self.stats.decisions_executed += 1 + elif report.acknowledged == "refused": + self.stats.decisions_refused += 1 + elif report.acknowledged == "stale": + self.stats.decisions_stale += 1 + elif report.acknowledged == "expired": + self.stats.decisions_expired += 1 + + def _backoff_s(self) -> float: + return min( + BACKOFF_CAP_S, + BACKOFF_BASE_S * (2 ** min(self.stats.consecutive_failures - 1, 16)), + ) diff --git a/openadapt_flow/console/human_decisions.py b/openadapt_flow/console/human_decisions.py index 0d9501de..681b7a63 100644 --- a/openadapt_flow/console/human_decisions.py +++ b/openadapt_flow/console/human_decisions.py @@ -57,6 +57,7 @@ AttendedActionRequest, AttendedActionStore, AttendedDecision, + AttendedRelayBinding, execute_attended_action, ) @@ -731,6 +732,7 @@ def execute_remote_attended_action( deployment: DeploymentConfig, principal: RemoteDecisionPrincipal, executor: Optional[AttendedActionExecutor] = None, + relay_binding: Optional[AttendedRelayBinding] = None, key: Optional[str] = None, ) -> AttendedDecision: """Provider-neutral AAL2 admission followed by normal governed execution. @@ -752,6 +754,7 @@ def execute_remote_attended_action( engine_request, operator=principal.subject, executor=executor, + relay_binding=relay_binding, key=key, ) diff --git a/openadapt_flow/console/server.py b/openadapt_flow/console/server.py index c530af86..cd365266 100644 --- a/openadapt_flow/console/server.py +++ b/openadapt_flow/console/server.py @@ -11,6 +11,7 @@ from pathlib import Path from typing import Optional +from openadapt_flow.console.decision_supervisor import DecisionSupervisorThread from openadapt_flow.runtime.durable.attended_service import AttendedActionService #: The only address the console ever binds. Not configurable. @@ -27,9 +28,20 @@ def serve( allow_actions: bool = False, attend: bool = False, attended_service: Optional[AttendedActionService] = None, + decision_supervisor: Optional[DecisionSupervisorThread] = None, port: int = DEFAULT_PORT, ) -> None: - """Build the app and serve it on ``http://127.0.0.1:`` (blocking).""" + """Build the app and serve it on ``http://127.0.0.1:`` (blocking). + + Args: + decision_supervisor: When given, the outbound decision lane runs beside + the server for as long as it serves. The console process is the + right host for it because it already owns the deployment-bound + action service a continuation needs, and because + ``execute_attended_action`` takes a single-flight lease over the + pause -- so an answer from a phone and one from this browser cannot + both execute. + """ import uvicorn from openadapt_flow.console.app import create_app @@ -51,4 +63,10 @@ def serve( "Open this private console URL in your browser:\n" f" http://{LOOPBACK_HOST}:{port}/#token={access_token}" ) - uvicorn.run(app, host=LOOPBACK_HOST, port=port, log_level="info") + if decision_supervisor is not None: + decision_supervisor.start() + try: + uvicorn.run(app, host=LOOPBACK_HOST, port=port, log_level="info") + finally: + if decision_supervisor is not None: + decision_supervisor.stop() diff --git a/openadapt_flow/runtime/durable/attended.py b/openadapt_flow/runtime/durable/attended.py index b4408cfb..6b893871 100644 --- a/openadapt_flow/runtime/durable/attended.py +++ b/openadapt_flow/runtime/durable/attended.py @@ -24,11 +24,13 @@ import json import os import secrets +import stat import threading +import time from contextlib import contextmanager from datetime import datetime, timedelta, timezone from pathlib import Path -from typing import Any, Callable, Iterator, Literal, Optional, Protocol +from typing import Any, Callable, Iterator, Literal, Optional, Protocol, cast from pydantic import BaseModel, ConfigDict, Field @@ -58,10 +60,14 @@ CAPABILITY_HISTORY_FILENAME = "attended_capability_history.json" CAPABILITY_KEY_FILENAME = ".attended_capability.key" DECISIONS_FILENAME = "attended_decisions.json" +DECISIONS_LOCK_FILENAME = ".attended_decisions.lock" LEASE_FILENAME = ".attended_action.lease" PROGRAM_RECEIPTS_DIRNAME = ".attended_program_receipts" +RELAY_ACK_RECORD_DOMAIN = b"openadapt:relay-ack-record-v1\0" +RELAY_ACK_WORKFLOW_DOMAIN = b"openadapt:relay-ack-workflow-v1\0" DEFAULT_CAPABILITY_TTL_S = 24 * 3600.0 DEFAULT_LEASE_TTL_S = 15 * 60.0 +DEFAULT_DECISION_LOG_LOCK_TIMEOUT_S = 5.0 def _now() -> datetime: @@ -260,9 +266,82 @@ class AttendedDecision(BaseModel): transition_receipt_digest: Optional[str] = None +class AttendedRelayBinding(BaseModel): + """Exact signed hosted decision bound to one retained engine outcome.""" + + model_config = ConfigDict(extra="forbid") + + schema_version: Literal[1] = 1 + decision_id: str = Field( + min_length=8, + max_length=128, + pattern=r"^[A-Za-z0-9][A-Za-z0-9._:-]+$", + ) + relay_digest: str = Field(pattern=r"^sha256:[0-9a-f]{64}$") + relay_signature: str = Field(pattern=r"^hmac-sha256:[0-9a-f]{64}$") + idempotency_key: str = Field( + min_length=16, + max_length=200, + pattern=r"^[A-Za-z0-9._:-]+$", + ) + capability_digest: str = Field(pattern=r"^sha256:[0-9a-f]{64}$") + event_sequence: int = Field(ge=1) + action: Literal["continue", "skip", "reject", "teach", "escalate"] + + +RelayOutcomeStatus = Literal[ + "completed", + "refused", + "halted", + "needs_demonstration", + "escalated", + "rejected", +] + + +class AttendedRelayAcknowledgement(AttendedRelayBinding): + """Durable hosted acknowledgement linked to one journaled decision.""" + + engine_ack_result: Literal["accepted", "refused"] + run_id: str + workflow_digest: str = Field(pattern=r"^hmac-sha256:[0-9a-f]{64}$") + bundle_version: str = Field(pattern=r"^sha256:[0-9a-f]{64}$") + pause_id: str = Field(pattern=r"^[0-9a-f]{32}$") + retained_decision_id: str = Field(pattern=r"^[0-9a-f]{32}$") + retained_decision_digest: str = Field(pattern=r"^sha256:[0-9a-f]{64}$") + retained_request_digest: str = Field(pattern=r"^sha256:[0-9a-f]{64}$") + retained_status: RelayOutcomeStatus + confirmed: bool = False + created_at: str = Field(default_factory=lambda: _iso(_now())) + confirmed_at: Optional[str] = None + record_mac: str = Field(pattern=r"^hmac-sha256:[0-9a-f]{64}$") + + def unsigned(self) -> dict[str, Any]: + return self.model_dump(exclude={"record_mac"}, mode="json") + + def binding(self) -> AttendedRelayBinding: + return AttendedRelayBinding.model_validate( + self.model_dump( + include={ + "schema_version", + "decision_id", + "relay_digest", + "relay_signature", + "idempotency_key", + "capability_digest", + "event_sequence", + "action", + } + ) + ) + + class AttendedDecisionLog(BaseModel): schema_version: int = 1 decisions: list[AttendedDecision] = Field(default_factory=list) + relay_acknowledgements: list[AttendedRelayAcknowledgement] = Field( + default_factory=list + ) def _delivery_state( @@ -491,6 +570,7 @@ def __init__(self, run_dir: Path | str) -> None: 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.decisions_lock_path = self.run_dir / DECISIONS_LOCK_FILENAME self.lease_path = self.run_dir / LEASE_FILENAME @staticmethod @@ -552,6 +632,90 @@ def _key(self, *, create: bool) -> bytes: ) return key + @contextmanager + def _decision_log_lock( + self, + *, + timeout_s: float = DEFAULT_DECISION_LOG_LOCK_TIMEOUT_S, + ) -> Iterator[None]: + """Serialize cross-process decision-journal read-modify-write cycles. + + There is no stale automatic takeover. If a process dies while it owns + this lock, an operator must reconcile and remove the private lock file. + Guessing that a journal writer died would permit two writers to replace + one another's retained outcomes. + """ + self.run_dir.mkdir(parents=True, exist_ok=True) + deadline = time.monotonic() + max(0.0, timeout_s) + while True: + try: + fd = os.open( + self.decisions_lock_path, + os.O_CREAT | os.O_EXCL | os.O_WRONLY, + 0o600, + ) + break + except FileExistsError: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise AttendedActionBusy( + "the attended decision journal is being updated or " + "requires operator reconciliation" + ) from None + time.sleep(min(0.01, remaining)) + owner_nonce = secrets.token_bytes(32) + owner_stat = os.fstat(fd) + try: + if os.write(fd, owner_nonce) != len(owner_nonce): + raise OSError("the decision journal lock nonce write was incomplete") + os.fsync(fd) + self._fsync_parent(self.decisions_lock_path) + yield + finally: + cleanup_error: Optional[AttendedActionBusy] = None + try: + current_lstat = os.lstat(self.decisions_lock_path) + same_entry = ( + stat.S_ISREG(current_lstat.st_mode) + and current_lstat.st_dev == owner_stat.st_dev + and current_lstat.st_ino == owner_stat.st_ino + ) + if not same_entry: + raise OSError("the lock path no longer names the owner file") + current_fd = os.open( + self.decisions_lock_path, + os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0), + ) + try: + current_fstat = os.fstat(current_fd) + current_nonce = os.read(current_fd, len(owner_nonce) + 1) + finally: + os.close(current_fd) + if ( + not stat.S_ISREG(current_fstat.st_mode) + or current_fstat.st_dev != owner_stat.st_dev + or current_fstat.st_ino != owner_stat.st_ino + or not hmac.compare_digest(current_nonce, owner_nonce) + ): + raise OSError("the lock owner nonce or file identity changed") + final_lstat = os.lstat(self.decisions_lock_path) + if ( + final_lstat.st_dev != owner_stat.st_dev + or final_lstat.st_ino != owner_stat.st_ino + ): + raise OSError("the lock path changed before release") + self.decisions_lock_path.unlink() + self._fsync_parent(self.decisions_lock_path) + except OSError: + cleanup_error = AttendedActionBusy( + "the attended decision journal lock changed while owned; " + "the replacement lock was retained for reconciliation" + ) + finally: + os.close(fd) + if cleanup_error is not None: + raise cleanup_error + def _sign(self, capability: AttendedPauseCapability, *, create_key: bool) -> str: return ( "hmac-sha256:" @@ -562,6 +726,142 @@ def _sign(self, capability: AttendedPauseCapability, *, create_key: bool) -> str ).hexdigest() ) + def _workflow_digest(self, workflow_name: str) -> str: + return ( + "hmac-sha256:" + + hmac.new( + self._key(create=False), + RELAY_ACK_WORKFLOW_DOMAIN + workflow_name.encode("utf-8"), + hashlib.sha256, + ).hexdigest() + ) + + def _relay_ack_mac(self, record: AttendedRelayAcknowledgement) -> str: + return ( + "hmac-sha256:" + + hmac.new( + self._key(create=False), + RELAY_ACK_RECORD_DOMAIN + _canonical(record.unsigned()), + hashlib.sha256, + ).hexdigest() + ) + + def _verify_relay_ack_mac(self, record: AttendedRelayAcknowledgement) -> None: + expected = self._relay_ack_mac(record) + if not hmac.compare_digest(record.record_mac, expected): + raise AttendedActionRefused( + "the retained relay acknowledgement HMAC does not verify" + ) + + def _signed_capabilities(self) -> list[AttendedPauseCapability]: + """Read and authenticate the current and historical pause authorities.""" + if ( + self.capability_path.is_symlink() + or self.capability_history_path.is_symlink() + ): + raise AttendedActionRefused( + "the attended capability path must not be a symlink" + ) + capabilities: list[AttendedPauseCapability] = [] + if self.capability_path.is_file(): + try: + capabilities.append( + AttendedPauseCapability.model_validate_json( + self.capability_path.read_text() + ) + ) + except ValueError as exc: + raise AttendedActionRefused( + "the current attended capability is invalid" + ) from exc + if self.capability_history_path.is_file(): + try: + raw_history = json.loads(self.capability_history_path.read_text()) + if not isinstance(raw_history, list): + raise ValueError("history is not a list") + capabilities.extend( + AttendedPauseCapability.model_validate(raw) for raw in raw_history + ) + except (OSError, ValueError) as exc: + raise AttendedActionRefused( + "the attended capability history is invalid" + ) from exc + if not capabilities: + raise AttendedActionRefused( + "no signed attended capability can authenticate this relay outcome" + ) + for capability in capabilities: + expected = self._sign(capability, create_key=False) + if not hmac.compare_digest(capability.signature, expected): + raise AttendedActionRefused( + "an attended capability signature does not verify" + ) + return capabilities + + def _capability_for_relay_binding( + self, + binding: AttendedRelayBinding, + ) -> AttendedPauseCapability: + matches = [ + capability + for capability in self._signed_capabilities() + if capability.digest == binding.capability_digest + and capability.event_sequence == binding.event_sequence + ] + if len(matches) != 1: + raise AttendedActionRefused( + "the exact signed pause capability for this relay is missing " + "or ambiguous" + ) + return matches[0] + + def _verify_relay_ack_context( + self, + record: AttendedRelayAcknowledgement, + *, + key: Optional[str] = None, + ) -> AttendedPauseCapability: + """Verify the record against its signed pause and live run manifest.""" + capability = self._capability_for_relay_binding(record.binding()) + if ( + capability.run_id != record.run_id + or capability.bundle_version != record.bundle_version + or capability.pause_id != record.pause_id + or capability.event_sequence != record.event_sequence + or capability.digest != record.capability_digest + or self._workflow_digest(capability.workflow_name) != record.workflow_digest + ): + raise AttendedActionRefused( + "the retained relay acknowledgement does not match its signed " + "pause capability" + ) + + from openadapt_flow import crypto as _crypto + + manifest = CheckpointStore( + self.run_dir, key=_crypto.resolve_key(key) + ).read_manifest() + if manifest is None: + raise AttendedActionRefused( + "the live run manifest is missing for relay acknowledgement recovery" + ) + try: + live_bundle_version = bundle_version(manifest.bundle_dir) + except (OSError, ValueError) as exc: + raise AttendedActionRefused( + "the live bundle version cannot be verified for relay recovery" + ) from exc + if ( + manifest.run_id != record.run_id + or live_bundle_version != record.bundle_version + or self._workflow_digest(manifest.workflow_name) != record.workflow_digest + ): + raise AttendedActionRefused( + "the live run manifest does not match the retained relay " + "acknowledgement" + ) + return capability + def seal_human_decision_task(self, unsigned: dict[str, Any]) -> dict[str, Any]: """Sign one PHI-free task projection with a separate HMAC domain. @@ -897,12 +1197,229 @@ def unresolved_delivery(self, pause_id: str) -> Optional[AttendedDecision]: 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") + def _relay_acknowledgement( + self, + binding: AttendedRelayBinding, + decision: AttendedDecision, + *, + key: Optional[str] = None, + ) -> AttendedRelayAcknowledgement: + if decision.status in {"prepared", "delivery_started", "delivery_uncertain"}: + raise AttendedActionRefused( + "a non-terminal engine decision cannot authorize a relay " + "acknowledgement" + ) + capability = self._capability_for_relay_binding(binding) + record = AttendedRelayAcknowledgement( + **binding.model_dump(), + engine_ack_result=( + "refused" if decision.status == "refused" else "accepted" + ), + run_id=capability.run_id, + workflow_digest=self._workflow_digest(capability.workflow_name), + bundle_version=capability.bundle_version, + pause_id=capability.pause_id, + retained_decision_id=decision.decision_id, + retained_decision_digest=_digest(decision), + retained_request_digest=decision.request_digest, + retained_status=cast(RelayOutcomeStatus, decision.status), + record_mac="hmac-sha256:" + ("0" * 64), ) + record = record.model_copy(update={"record_mac": self._relay_ack_mac(record)}) + self._verify_relay_ack_context(record, key=key) + return record + + def _add_relay_acknowledgement( + self, + log: AttendedDecisionLog, + binding: AttendedRelayBinding, + decision: AttendedDecision, + *, + key: Optional[str] = None, + ) -> None: + acknowledgement = self._relay_acknowledgement(binding, decision, key=key) + existing = [ + record + for record in log.relay_acknowledgements + if record.decision_id == binding.decision_id + ] + if len(existing) > 1: + raise AttendedActionRefused( + "the relay decision has multiple retained acknowledgement records" + ) + if existing: + record = existing[0] + self._verify_relay_ack_mac(record) + self._verify_relay_ack_context(record, key=key) + if ( + record.binding() != binding + or record.engine_ack_result != acknowledgement.engine_ack_result + or record.retained_decision_id != acknowledgement.retained_decision_id + or record.retained_decision_digest + != acknowledgement.retained_decision_digest + or record.retained_request_digest + != acknowledgement.retained_request_digest + or record.retained_status != acknowledgement.retained_status + ): + raise AttendedActionRefused( + "the relay decision id is already bound to a different " + "signed decision or retained engine outcome" + ) + return + log.relay_acknowledgements.append(acknowledgement) + + def append( + self, + decision: AttendedDecision, + *, + relay_binding: Optional[AttendedRelayBinding] = None, + key: Optional[str] = None, + ) -> None: + with self._decision_log_lock(): + log = self._read_log() + log.decisions.append(decision) + if relay_binding is not None: + self._add_relay_acknowledgement(log, relay_binding, decision, key=key) + self._atomic_write( + self.decisions_path, log.model_dump_json(indent=2).encode("utf-8") + ) + + def retain_relay_acknowledgement( + self, + binding: AttendedRelayBinding, + decision: AttendedDecision, + *, + key: Optional[str] = None, + ) -> None: + """Bind a prior exact engine result for an idempotent remote replay.""" + with self._decision_log_lock(): + log = self._read_log() + retained = [ + candidate + for candidate in log.decisions + if candidate.decision_id == decision.decision_id + ] + if len(retained) != 1 or retained[0] != decision: + raise AttendedActionRefused( + "the retained engine outcome for this relay cannot be proved" + ) + before = len(log.relay_acknowledgements) + self._add_relay_acknowledgement(log, binding, decision, key=key) + if len(log.relay_acknowledgements) != before: + self._atomic_write( + self.decisions_path, + log.model_dump_json(indent=2).encode("utf-8"), + ) + + def _validated_relay_acknowledgement( + self, + log: AttendedDecisionLog, + binding: AttendedRelayBinding, + *, + key: Optional[str] = None, + ) -> Optional[tuple[AttendedRelayAcknowledgement, AttendedDecision]]: + records = [ + record + for record in log.relay_acknowledgements + if record.decision_id == binding.decision_id + ] + if not records: + orphaned = [ + decision + for decision in log.decisions + if decision.idempotency_key == binding.idempotency_key + and decision.capability_digest == binding.capability_digest + and decision.action == binding.action + and decision.status not in {"prepared", "delivery_started"} + ] + if orphaned: + raise AttendedActionRefused( + "a retained engine outcome has no authenticated relay " + "acknowledgement record" + ) + return None + if len(records) != 1: + raise AttendedActionRefused( + "the relay decision has multiple retained acknowledgement records" + ) + record = records[0] + self._verify_relay_ack_mac(record) + self._verify_relay_ack_context(record, key=key) + if record.binding() != binding: + raise AttendedActionRefused( + "a re-delivered decision changed its exact signed or " + "idempotency binding" + ) + retained = [ + decision + for decision in log.decisions + if decision.decision_id == record.retained_decision_id + ] + if len(retained) != 1: + raise AttendedActionRefused( + "the engine outcome retained for this relay is missing or ambiguous" + ) + outcome = retained[0] + if ( + _digest(outcome) != record.retained_decision_digest + or outcome.request_digest != record.retained_request_digest + or outcome.idempotency_key != record.idempotency_key + or outcome.capability_digest != record.capability_digest + or outcome.action != record.action + or outcome.status != record.retained_status + or ("refused" if outcome.status == "refused" else "accepted") + != record.engine_ack_result + ): + raise AttendedActionRefused( + "the engine outcome retained for this relay no longer verifies" + ) + return record, outcome + + def relay_acknowledgement( + self, + binding: AttendedRelayBinding, + *, + key: Optional[str] = None, + ) -> Optional[tuple[AttendedRelayAcknowledgement, AttendedDecision]]: + """Return an exact retained result, or refuse a changed relay binding.""" + return self._validated_relay_acknowledgement(self._read_log(), binding, key=key) + + def confirm_relay_acknowledgement( + self, + binding: AttendedRelayBinding, + *, + key: Optional[str] = None, + ) -> None: + """Mark the exact relay record only after Cloud accepted the ACK.""" + with self._decision_log_lock(): + log = self._read_log() + matched = self._validated_relay_acknowledgement(log, binding, key=key) + if matched is None: + raise AttendedActionRefused( + "the relay acknowledgement has no retained engine outcome" + ) + record, _outcome = matched + if record.confirmed: + return + index = next( + index + for index, candidate in enumerate(log.relay_acknowledgements) + if candidate == record + ) + candidate = record.model_copy( + update={ + "confirmed": True, + "confirmed_at": _iso(_now()), + "record_mac": "hmac-sha256:" + ("0" * 64), + } + ) + candidate = candidate.model_copy( + update={"record_mac": self._relay_ack_mac(candidate)} + ) + log.relay_acknowledgements[index] = candidate + self._atomic_write( + self.decisions_path, log.model_dump_json(indent=2).encode("utf-8") + ) @contextmanager def lease( @@ -1172,6 +1689,7 @@ def execute_attended_action( *, operator: str, executor: Optional[AttendedActionExecutor] = None, + relay_binding: Optional[AttendedRelayBinding] = None, key: Optional[str] = None, now: Optional[datetime] = None, ) -> AttendedDecision: @@ -1202,6 +1720,8 @@ def execute_attended_action( "automatic retry is refused until an audited reconciliation" ) if prior.status != "prepared": + if relay_binding is not None: + actions.retain_relay_acknowledgement(relay_binding, prior, key=key) return prior checkpoints = CheckpointStore(run_dir, key=key) @@ -1231,6 +1751,8 @@ def execute_attended_action( "automatic retry is refused until reconciliation" ) if prior.status != "prepared": + if relay_binding is not None: + actions.retain_relay_acknowledgement(relay_binding, prior, key=key) return prior request_digest = _digest(request) unresolved = actions.unresolved_delivery(capability.pause_id) @@ -1282,7 +1804,7 @@ def execute_attended_action( ), next_transition=capability.expected_next_transition, ) - actions.append(decision) + actions.append(decision, relay_binding=relay_binding, key=key) return decision if request.action == "teach": @@ -1303,7 +1825,7 @@ def execute_attended_action( ), next_transition=capability.expected_next_transition, ) - actions.append(decision) + actions.append(decision, relay_binding=relay_binding, key=key) return decision if request.action == "escalate": @@ -1322,7 +1844,7 @@ def execute_attended_action( ), next_transition=capability.expected_next_transition, ) - actions.append(decision) + actions.append(decision, relay_binding=relay_binding, key=key) return decision if executor is None: @@ -1399,7 +1921,7 @@ def execute_attended_action( next_transition=result.next_transition, transition_receipt_digest=result.transition_receipt_digest, ) - actions.append(decision) + actions.append(decision, relay_binding=relay_binding, key=key) return decision diff --git a/tests/test_decision_supervisor.py b/tests/test_decision_supervisor.py new file mode 100644 index 00000000..d476c98f --- /dev/null +++ b/tests/test_decision_supervisor.py @@ -0,0 +1,1286 @@ +"""The loop that makes the hosted lane real, and what it refuses to do. + +``decision_relay`` proved the wire. This file proves the two things a wire +cannot: that an open pause becomes answerable **without anyone calling +anything**, and that an answer is applied to the pause it was minted from +rather than to whichever run happened to be in hand. + +The second is the one worth the most scrutiny. ``DecisionRelay.serve_once`` +takes its run and item as arguments, which is safe only when exactly one pause +is open. A practice with two halted runs is the ordinary case, not the exotic +one, so ``test_a_decision_is_executed_against_the_pause_it_was_minted_from`` +opens two and answers the second. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import threading +import time +from pathlib import Path +from typing import Any, Optional + +import pytest + +from openadapt_flow.console.attention import attention_item +from openadapt_flow.console.decision_relay import ( + DecisionRelay, + RelayedDecision, + RelayRefused, + RelayUncertain, +) +from openadapt_flow.console.decision_supervisor import ( + DecisionSupervisor, + DecisionSupervisorThread, +) +from openadapt_flow.ir import ( + ActionKind, + Postcondition, + PostconditionKind, + Step, + Workflow, +) +from openadapt_flow.runtime.durable.attended import ( + AttendedActionBusy, + AttendedActionRefused, + AttendedActionStore, + AttendedRelayAcknowledgement, +) +from openadapt_flow.runtime.replayer import Replayer +from tests.test_attended_actions import _ResultExecutor +from tests.test_decision_relay import ( + TOKEN, + FakeTransport, + _bound_relay_payload, + _deployment, + _sign, +) +from tests.test_replayer import FakeBackend, FakeVision + +PROTECTED_VALUE = "Wilhelmina Featherstonehaugh" + +LONG_PAST = "2000-01-01T00:00:00Z" + + +# --------------------------------------------------------------- fixtures + + +def _halted_run(runs_root: Path, bundles_root: Path, name: str): + """One durably paused run under ``runs_root``, plus its attention item.""" + workflow = Workflow( + name=f"supervisor-{name}", + params={"patient": PROTECTED_VALUE}, + steps=[ + Step( + id="human", + intent=f"confirm coverage for {PROTECTED_VALUE}", + action=ActionKind.KEY, + key="A", + expect=[ + Postcondition( + kind=PostconditionKind.TEXT_PRESENT, + text="COVERAGE ACTIVE", + timeout_s=0.01, + ) + ], + ), + Step(id="next", intent="record it", action=ActionKind.KEY, key="B"), + ], + ) + bundle = bundles_root / name + run = runs_root / name + workflow.save(bundle) + report = Replayer( + FakeBackend(), vision=FakeVision(), durable=True, poll_interval_s=0.0 + ).run( + workflow, + bundle_dir=bundle, + run_dir=run, + params={"patient": PROTECTED_VALUE}, + ) + assert report.success is False + item = attention_item(runs_root, run) + assert item is not None + assert item.durably_paused is True + return run, item + + +def _supervisor( + runs_root: Path, + *, + deployment: Any = None, + executor: Any = None, + **responses: Any, +): + deployment = deployment or _deployment() + transport = FakeTransport(responses) + relay = DecisionRelay(transport, token=TOKEN, deployment=deployment) + supervisor = DecisionSupervisor( + runs_root, relay=relay, deployment=deployment, executor=executor + ) + return supervisor, transport + + +def _relayed_for(run: Path, item: Any, deployment: Any, **overrides: Any): + """Mint the relay the control plane would mint for one real open pause. + + Built from the same helper the transport tests use, so a change to the + projection cannot make this file agree with itself while disagreeing with + the wire. + """ + unsigned = _bound_relay_payload(run, item, deployment) + unsigned.update(overrides) + return _sign(unsigned) + + +# ------------------------------------------------------- publishing a queue + + +def test_an_open_pause_is_published_without_anyone_calling_anything(tmp_path): + """The whole point: a halt becomes answerable from a phone by itself.""" + runs = tmp_path / "runs" + _halted_run(runs, tmp_path / "bundles", "one") + supervisor, transport = _supervisor( + runs, tasks=(200, {"accepted": True, "created": True, "task_id": "task_x"}) + ) + + report = supervisor.publish_open_pauses() + + assert report.certain_count == 1 + assert len(report.published) == 1 + assert report.unknown == () + assert [path for path, _ in transport.calls] == ["/api/human-decisions/tasks"] + + +def test_every_open_pause_is_published_not_only_the_first(tmp_path): + runs = tmp_path / "runs" + bundles = tmp_path / "bundles" + _halted_run(runs, bundles, "one") + _halted_run(runs, bundles, "two") + supervisor, transport = _supervisor( + runs, tasks=(200, {"accepted": True, "created": True, "task_id": "task_x"}) + ) + + report = supervisor.publish_open_pauses() + + assert len(report.published) == 2 + assert len(transport.calls) == 2 + + +def test_nothing_protected_reaches_the_wire_from_the_whole_queue(tmp_path): + runs = tmp_path / "runs" + bundles = tmp_path / "bundles" + _halted_run(runs, bundles, "one") + _halted_run(runs, bundles, "two") + supervisor, transport = _supervisor( + runs, tasks=(200, {"accepted": True, "created": True, "task_id": "task_x"}) + ) + supervisor.publish_open_pauses() + + body = json.dumps([payload for _, payload in transport.calls]) + assert PROTECTED_VALUE not in body + assert "supervisor-one" not in body + assert "COVERAGE ACTIVE" not in body + + +def test_an_uncertain_publish_is_reported_as_unknown_never_as_reachable(tmp_path): + runs = tmp_path / "runs" + _halted_run(runs, tmp_path / "bundles", "one") + supervisor, _ = _supervisor(runs, tasks=RelayUncertain("connection reset")) + + report = supervisor.publish_open_pauses() + + assert len(report.unknown) == 1 + assert report.published == () + assert report.already_published == () + assert report.certain_count == 0 + + +def test_a_run_that_cannot_be_projected_is_recorded_not_raised(tmp_path): + """A pause with no remote issuance is still served by the local console.""" + runs = tmp_path / "runs" + run, _item = _halted_run(runs, tmp_path / "bundles", "one") + # Break the capability so projection refuses, exactly as a closed pause + # would. + AttendedActionStore(run).capability_path.write_text("{}") + supervisor, transport = _supervisor( + runs, tasks=(200, {"accepted": True, "created": True}) + ) + + report = supervisor.publish_open_pauses() + + assert report.published == () + assert report.already_published == () + assert transport.calls == [] + + +def test_a_run_that_is_not_durably_paused_is_never_published(tmp_path): + """A run that already resumed has no pause for a decision to bind to.""" + runs = tmp_path / "runs" + run, _item = _halted_run(runs, tmp_path / "bundles", "one") + supervisor, _ = _supervisor(runs) + + assert [pause.run_dir for pause in supervisor.open_pauses()] == [run] + + # Resuming clears the durable pause; the signed capability file survives it, + # which is exactly why `durably_paused` and not the capability is the gate. + (run / "pending_escalation.json").unlink() + assert AttendedActionStore(run).capability_path.exists() + assert supervisor.open_pauses() == [] + + +class _ScriptedTransport(FakeTransport): + """A transport whose /tasks response changes per call.""" + + def __init__(self, task_responses: list, **responses) -> None: + super().__init__(responses) + self.task_responses = list(task_responses) + + def post(self, path, payload, *, timeout_s): + self.calls.append((path, payload)) + if path.endswith("/tasks"): + response = self.task_responses.pop(0) + if isinstance(response, Exception): + raise response + return response + return super().post(path, payload, timeout_s=timeout_s) + + +class _SequencedTransport(FakeTransport): + """Script repeated polls and acknowledgements across supervisor cycles.""" + + def __init__(self, *, polls: list[Any], acknowledgements: list[Any]) -> None: + super().__init__() + self.polls = list(polls) + self.acknowledgements = list(acknowledgements) + + def post(self, path, payload, *, timeout_s): + self.calls.append((path, payload)) + if path.endswith("/tasks"): + return 200, {"accepted": True, "created": True, "task_id": "task_x"} + if path.endswith("/poll"): + response = self.polls.pop(0) + elif path.endswith("/ack"): + response = self.acknowledgements.pop(0) + else: # pragma: no cover - the relay has only these paths + response = (204, {}) + if isinstance(response, Exception): + raise response + return response + + +def _journaled_lost_ack(tmp_path: Path, name: str): + """Execute Continue once and retain its exact ACK record without confirming.""" + runs = tmp_path / "runs" + run, item = _halted_run(runs, tmp_path / "bundles", name) + deployment = _deployment() + executor = _ResultExecutor() + relay_body = _relayed_for(run, item, deployment) + transport = _SequencedTransport( + polls=[(200, {"decision": relay_body})], + acknowledgements=[RelayUncertain("the first ack response was lost")], + ) + relay = DecisionRelay(transport, token=TOKEN, deployment=deployment) + report = DecisionSupervisor( + runs, relay=relay, deployment=deployment, executor=executor + ).serve_once(wait_s=0.0) + assert report.outcome is not None + return runs, run, deployment, executor, relay_body, report.outcome + + +def _assert_retained_recovery_refuses( + runs: Path, + deployment: Any, + executor: Any, + relay_body: dict[str, Any], +) -> None: + transport = _SequencedTransport( + polls=[(200, {"decision": relay_body})], acknowledgements=[] + ) + relay = DecisionRelay(transport, token=TOKEN, deployment=deployment) + supervisor = DecisionSupervisor( + runs, relay=relay, deployment=deployment, executor=executor + ) + with pytest.raises(RelayRefused): + supervisor.serve_once(wait_s=0.0) + assert executor.calls == 1 + assert not any(path.endswith("/ack") for path, _ in transport.calls) + + +def _read_decision_log(run: Path) -> dict[str, Any]: + return json.loads(AttendedActionStore(run).decisions_path.read_text()) + + +def _write_decision_log(run: Path, payload: dict[str, Any]) -> None: + AttendedActionStore(run).decisions_path.write_text( + json.dumps(payload, indent=2, sort_keys=True) + ) + + +def _plain_digest(payload: dict[str, Any]) -> str: + encoded = json.dumps( + payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False + ).encode("utf-8") + return "sha256:" + hashlib.sha256(encoded).hexdigest() + + +def _resign_relay_ack_record( + store: AttendedActionStore, + payload: dict[str, Any], + **updates: Any, +) -> dict[str, Any]: + """Build a MAC-valid hostile record to test the independent bindings.""" + record = AttendedRelayAcknowledgement.model_validate(payload).model_copy( + update={ + **updates, + "record_mac": "hmac-sha256:" + ("0" * 64), + } + ) + return record.model_copy( + update={"record_mac": store._relay_ack_mac(record)} + ).model_dump(mode="json") + + +def test_one_refused_pause_does_not_silence_the_others(tmp_path): + """The failure that would leave a whole practice's halts unreachable.""" + runs = tmp_path / "runs" + bundles = tmp_path / "bundles" + _halted_run(runs, bundles, "one") + _halted_run(runs, bundles, "two") + deployment = _deployment() + transport = _ScriptedTransport( + [ + (400, {"error": "the projection is invalid"}), + (200, {"accepted": True, "created": True, "task_id": "task_x"}), + ] + ) + relay = DecisionRelay(transport, token=TOKEN, deployment=deployment) + supervisor = DecisionSupervisor(runs, relay=relay, deployment=deployment) + + report = supervisor.publish_open_pauses() + + assert len(report.refused) == 1 + assert len(report.published) == 1 + # Both were attempted; the refusal did not stop the loop. + assert len([p for p, _ in transport.calls if p.endswith("/tasks")]) == 2 + + +def test_an_accepted_pause_is_not_republished_every_cycle(tmp_path): + """An open pause can last hours; re-POSTing identical bytes is noise.""" + runs = tmp_path / "runs" + _halted_run(runs, tmp_path / "bundles", "one") + supervisor, transport = _supervisor( + runs, tasks=(200, {"accepted": True, "created": True, "task_id": "task_x"}) + ) + + first = supervisor.publish_open_pauses() + second = supervisor.publish_open_pauses() + + assert len(first.published) == 1 + # Reported as previously confirmed, NOT as an observation: the supervisor + # did not ask the control plane about it this pass. + assert second.already_published == () + assert len(second.previously_confirmed) == 1 + assert second.certain_count == 1 + assert len([p for p, _ in transport.calls if p.endswith("/tasks")]) == 1 + + +def test_an_uncertain_pause_is_republished_because_the_post_is_idempotent(tmp_path): + """Uncertainty is never memoized as success.""" + runs = tmp_path / "runs" + _halted_run(runs, tmp_path / "bundles", "one") + deployment = _deployment() + transport = _ScriptedTransport( + [ + RelayUncertain("connection reset"), + (200, {"accepted": True, "created": True, "task_id": "task_x"}), + ] + ) + relay = DecisionRelay(transport, token=TOKEN, deployment=deployment) + supervisor = DecisionSupervisor(runs, relay=relay, deployment=deployment) + + assert len(supervisor.publish_open_pauses().unknown) == 1 + assert len(supervisor.publish_open_pauses().published) == 1 + + +# ---------------------------------------------------- resolving an answer + + +def test_a_decision_is_executed_against_the_pause_it_was_minted_from(tmp_path): + """Two runs are halted; the answer belongs to exactly one of them.""" + runs = tmp_path / "runs" + bundles = tmp_path / "bundles" + _halted_run(runs, bundles, "one") + run_two, item_two = _halted_run(runs, bundles, "two") + deployment = _deployment() + + relay_body = _relayed_for(run_two, item_two, deployment) + supervisor, transport = _supervisor( + runs, + deployment=deployment, + executor=_ResultExecutor(), + tasks=(200, {"accepted": True, "created": True, "task_id": "task_x"}), + poll=(200, {"decision": relay_body}), + ack=(200, {"accepted": True}), + ) + + resolved = supervisor.resolve( + RelayedDecision(decision_id=str(relay_body["decision_id"]), relay=relay_body) + ) + assert resolved is not None + assert resolved.run_dir == run_two + + +def test_one_cycle_publishes_executes_and_acknowledges_without_a_caller(tmp_path): + """The whole lane, end to end, driven by nothing but the loop itself. + + This is the test that distinguishes a wired lane from a library that looks + wired: no caller resolves the run, no caller supplies the item, and the + engine -- not the relay -- performs the continuation. + """ + runs = tmp_path / "runs" + bundles = tmp_path / "bundles" + _halted_run(runs, bundles, "one") + run_two, item_two = _halted_run(runs, bundles, "two") + deployment = _deployment() + executor = _ResultExecutor() + relay_body = _relayed_for(run_two, item_two, deployment) + supervisor, transport = _supervisor( + runs, + deployment=deployment, + executor=executor, + tasks=(200, {"accepted": True, "created": True, "task_id": "task_x"}), + poll=(200, {"decision": relay_body}), + ack=(200, {"accepted": True}), + ) + + report = supervisor.serve_once(wait_s=0.0) + + # Both pauses were made answerable. + assert len(report.publishes.published) == 2 + # Exactly one was answered, and the engine ran it. + assert report.acknowledged == "accepted" + assert report.outcome is not None + assert report.outcome.status == "completed" + assert executor.calls == 1 + ack_path, ack_body = transport.calls[-1] + assert ack_path.endswith("/ack") + assert ack_body["result"] == "accepted" + + +@pytest.mark.parametrize( + ("action", "decision_action", "expected_status", "executor_calls"), + [ + ("continue", "verify_and_resume", "completed", 1), + ("reject", "reject", "rejected", 0), + ], +) +def test_a_lost_ack_survives_restart_without_a_second_action( + tmp_path, + action, + decision_action, + expected_status, + executor_calls, +): + """A completed action does not become stale when its first ack is lost. + + Continue proves the live-resume path. Reject proves a terminal action that + removes the pause from the open queue. Both re-deliver the exact signed + relay and its original idempotency key. + """ + runs = tmp_path / "runs" + run, item = _halted_run(runs, tmp_path / "bundles", action) + deployment = _deployment() + executor = _ResultExecutor() + relay_body = _relayed_for( + run, + item, + deployment, + action=action, + decision_action=decision_action, + ) + first_transport = _SequencedTransport( + polls=[(200, {"decision": relay_body})], + acknowledgements=[RelayUncertain("the engine result reached an uncertain ack")], + ) + first_relay = DecisionRelay(first_transport, token=TOKEN, deployment=deployment) + first_supervisor = DecisionSupervisor( + runs, relay=first_relay, deployment=deployment, executor=executor + ) + + first = first_supervisor.serve_once(wait_s=0.0) + + # A new relay and supervisor prove that no process-local cache is needed. + second_transport = _SequencedTransport( + polls=[(200, {"decision": relay_body})], + acknowledgements=[(200, {"accepted": True})], + ) + second_relay = DecisionRelay(second_transport, token=TOKEN, deployment=deployment) + second_supervisor = DecisionSupervisor( + runs, relay=second_relay, deployment=deployment, executor=executor + ) + second = second_supervisor.serve_once(wait_s=0.0) + + assert first.acknowledged == "accepted" + assert first.reacknowledged is False + assert first.outcome is not None + assert first.outcome.status == expected_status + assert second.acknowledged == "accepted" + assert second.reacknowledged is True + assert second.outcome == first.outcome + assert executor.calls == executor_calls + acknowledgements = [ + body + for transport in (first_transport, second_transport) + for path, body in transport.calls + if path.endswith("/ack") + ] + assert [body["result"] for body in acknowledgements] == [ + "accepted", + "accepted", + ] + retained = AttendedActionStore(run).relay_acknowledgement( + RelayedDecision( + decision_id=str(relay_body["decision_id"]), relay=relay_body + ).durable_binding() + ) + assert retained is not None + assert retained[0].confirmed is True + + +def test_lost_ack_recovery_refuses_a_changed_signed_or_idempotency_binding(tmp_path): + """A repeated decision id cannot select an earlier local outcome.""" + runs = tmp_path / "runs" + run, item = _halted_run(runs, tmp_path / "bundles", "one") + deployment = _deployment() + executor = _ResultExecutor() + relay_body = _relayed_for(run, item, deployment) + changed = { + key: value + for key, value in relay_body.items() + if key not in {"relay_digest", "relay_signature"} + } + changed["idempotency_key"] = "relay-idempotency-key-CHANGED-0002" + changed_relay = _sign(changed) + first_transport = _SequencedTransport( + polls=[(200, {"decision": relay_body})], + acknowledgements=[ + RelayUncertain("the first ack response was lost"), + ], + ) + first_relay = DecisionRelay(first_transport, token=TOKEN, deployment=deployment) + first_supervisor = DecisionSupervisor( + runs, relay=first_relay, deployment=deployment, executor=executor + ) + + first_supervisor.serve_once(wait_s=0.0) + + second_transport = _SequencedTransport( + polls=[(200, {"decision": changed_relay})], + acknowledgements=[], + ) + second_relay = DecisionRelay(second_transport, token=TOKEN, deployment=deployment) + second_supervisor = DecisionSupervisor( + runs, relay=second_relay, deployment=deployment, executor=executor + ) + + with pytest.raises(RelayRefused, match="exact signed or idempotency binding"): + second_supervisor.serve_once(wait_s=0.0) + + assert executor.calls == 1 + assert ( + len( + [ + path + for transport in (first_transport, second_transport) + for path, _ in transport.calls + if path.endswith("/ack") + ] + ) + == 1 + ) + + +def test_recovery_refuses_an_outcome_and_plain_digest_changed_together(tmp_path): + """A plain SHA beside mutable JSON cannot replace the per-run HMAC.""" + runs, run, deployment, executor, relay_body, _outcome = _journaled_lost_ack( + tmp_path, "plain-digest-tamper" + ) + log = _read_decision_log(run) + record = log["relay_acknowledgements"][0] + outcome = next( + item + for item in log["decisions"] + if item["decision_id"] == record["retained_decision_id"] + ) + outcome["status"] = "refused" + record["retained_status"] = "refused" + record["engine_ack_result"] = "refused" + record["retained_decision_digest"] = _plain_digest(outcome) + _write_decision_log(run, log) + + _assert_retained_recovery_refuses(runs, deployment, executor, relay_body) + + +def test_recovery_refuses_a_fabricated_accepted_outcome(tmp_path): + runs, run, deployment, executor, relay_body, _outcome = _journaled_lost_ack( + tmp_path, "fabricated-outcome" + ) + log = _read_decision_log(run) + record = log["relay_acknowledgements"][0] + original = next( + item + for item in log["decisions"] + if item["decision_id"] == record["retained_decision_id"] + ) + fabricated = { + **original, + "decision_id": "f" * 32, + "status": "completed", + "message": "fabricated local success", + } + log["decisions"].append(fabricated) + record["retained_decision_id"] = fabricated["decision_id"] + record["retained_decision_digest"] = _plain_digest(fabricated) + record["retained_status"] = "completed" + record["engine_ack_result"] = "accepted" + _write_decision_log(run, log) + + _assert_retained_recovery_refuses(runs, deployment, executor, relay_body) + + +@pytest.mark.parametrize( + ("field", "changed"), + [ + ("run_id", "changed-run-id"), + ("bundle_version", "sha256:" + ("0" * 64)), + ("pause_id", "0" * 32), + ("workflow_digest", "hmac-sha256:" + ("0" * 64)), + ("capability_digest", "sha256:" + ("0" * 64)), + ("event_sequence", None), + ], +) +def test_recovery_refuses_a_mac_valid_changed_run_or_pause_binding( + tmp_path, + field, + changed, +): + """The record MAC is necessary but not sufficient execution evidence.""" + runs, run, deployment, executor, relay_body, _outcome = _journaled_lost_ack( + tmp_path, f"changed-{field}" + ) + log = _read_decision_log(run) + record = log["relay_acknowledgements"][0] + if field == "event_sequence": + changed = int(record[field]) + 1 + log["relay_acknowledgements"][0] = _resign_relay_ack_record( + AttendedActionStore(run), record, **{field: changed} + ) + _write_decision_log(run, log) + + _assert_retained_recovery_refuses(runs, deployment, executor, relay_body) + + +def test_recovery_refuses_a_replaced_per_run_key(tmp_path): + runs, run, deployment, executor, relay_body, _outcome = _journaled_lost_ack( + tmp_path, "replaced-key" + ) + store = AttendedActionStore(run) + store.key_path.write_bytes(b"x" * 32) + if os.name != "nt": + os.chmod(store.key_path, 0o600) + + _assert_retained_recovery_refuses(runs, deployment, executor, relay_body) + + +@pytest.mark.parametrize("damage", ["truncate", "delete", "duplicate"]) +def test_recovery_refuses_a_damaged_or_missing_ack_record(tmp_path, damage): + runs, run, deployment, executor, relay_body, _outcome = _journaled_lost_ack( + tmp_path, f"record-{damage}" + ) + store = AttendedActionStore(run) + if damage == "truncate": + store.decisions_path.write_text("{") + else: + log = _read_decision_log(run) + if damage == "delete": + log["relay_acknowledgements"] = [] + else: + log["relay_acknowledgements"].append(dict(log["relay_acknowledgements"][0])) + _write_decision_log(run, log) + + _assert_retained_recovery_refuses(runs, deployment, executor, relay_body) + + +def test_recovery_refuses_when_the_signed_capability_is_missing(tmp_path): + runs, run, deployment, executor, relay_body, _outcome = _journaled_lost_ack( + tmp_path, "missing-capability" + ) + store = AttendedActionStore(run) + store.capability_path.unlink() + assert not store.capability_history_path.exists() + + _assert_retained_recovery_refuses(runs, deployment, executor, relay_body) + + +def test_recovery_refuses_a_journal_moved_to_another_run(tmp_path): + runs, run, deployment, executor, relay_body, _outcome = _journaled_lost_ack( + tmp_path, "moved-source" + ) + other, _item = _halted_run(runs, tmp_path / "bundles", "moved-destination") + source = AttendedActionStore(run).decisions_path + destination = AttendedActionStore(other).decisions_path + source.replace(destination) + + _assert_retained_recovery_refuses(runs, deployment, executor, relay_body) + + +def test_relay_ack_record_contains_only_closed_or_opaque_context(tmp_path): + _runs, run, _deployment_cfg, _executor, _relay, _outcome = _journaled_lost_ack( + tmp_path, "secret-free-record" + ) + record = _read_decision_log(run)["relay_acknowledgements"][0] + encoded = json.dumps(record, sort_keys=True) + + assert PROTECTED_VALUE not in encoded + assert "supervisor-secret-free-record" not in encoded + assert str(run) not in encoded + assert set(record).isdisjoint( + { + "actor_id", + "operator", + "workflow_name", + "task", + "task_content", + "screenshot", + "ocr_text", + "path", + "message", + } + ) + + +def test_confirm_and_append_are_serialized_without_a_lost_write( + tmp_path, + monkeypatch, +): + _runs, run, _deployment_cfg, _executor, relay_body, outcome = _journaled_lost_ack( + tmp_path, "serialized-journal" + ) + store = AttendedActionStore(run) + binding = RelayedDecision( + decision_id=str(relay_body["decision_id"]), relay=relay_body + ).durable_binding() + extra = outcome.model_copy(update={"decision_id": "e" * 32}) + append_entered = threading.Event() + release_append = threading.Event() + failures: list[BaseException] = [] + original_atomic_write = AttendedActionStore._atomic_write + + def slow_append(path, payload, *, mode=0o600): + if threading.current_thread().name == "append-worker": + append_entered.set() + assert release_append.wait(timeout=2.0) + return original_atomic_write(path, payload, mode=mode) + + monkeypatch.setattr(AttendedActionStore, "_atomic_write", staticmethod(slow_append)) + + def append_worker(): + try: + store.append(extra) + except BaseException as exc: # pragma: no cover - asserted below + failures.append(exc) + + def confirm_worker(): + try: + store.confirm_relay_acknowledgement(binding) + except BaseException as exc: # pragma: no cover - asserted below + failures.append(exc) + + append_thread = threading.Thread(target=append_worker, name="append-worker") + confirm_thread = threading.Thread(target=confirm_worker, name="confirm-worker") + append_thread.start() + assert append_entered.wait(timeout=2.0) + confirm_thread.start() + time.sleep(0.05) + assert confirm_thread.is_alive() + release_append.set() + append_thread.join(timeout=2.0) + confirm_thread.join(timeout=2.0) + + assert failures == [] + assert not append_thread.is_alive() + assert not confirm_thread.is_alive() + log = _read_decision_log(run) + assert any(item["decision_id"] == extra.decision_id for item in log["decisions"]) + assert log["relay_acknowledgements"][0]["confirmed"] is True + + +@pytest.mark.skipif( + os.name == "nt", + reason="Windows prevents replacement while the owner handle stays open", +) +def test_a_lock_owner_never_deletes_a_replacement_lock(tmp_path): + store = AttendedActionStore(tmp_path / "run") + replacement_nonce = b"replacement-lock-owner" + + with pytest.raises(AttendedActionBusy, match="lock changed while owned"): + with store._decision_log_lock(): + store.decisions_lock_path.unlink() + replacement_fd = os.open( + store.decisions_lock_path, + os.O_CREAT | os.O_EXCL | os.O_WRONLY, + 0o600, + ) + try: + os.write(replacement_fd, replacement_nonce) + os.fsync(replacement_fd) + finally: + os.close(replacement_fd) + + assert store.decisions_lock_path.read_bytes() == replacement_nonce + with pytest.raises(AttendedActionBusy): + with store._decision_log_lock(timeout_s=0.0): + raise AssertionError("a third writer acquired the replacement lock") + + +def test_a_reject_from_a_phone_ends_the_right_run(tmp_path): + """The action that terminates a run must reach the pause it names. + + `reject` is the one answer whose whole value is the disagreement signal it + records, so delivering it to the wrong open pause -- or not at all -- is the + failure that matters most here. + """ + runs = tmp_path / "runs" + bundles = tmp_path / "bundles" + _halted_run(runs, bundles, "one") + run_two, item_two = _halted_run(runs, bundles, "two") + deployment = _deployment() + executor = _ResultExecutor() + relay_body = _relayed_for( + run_two, item_two, deployment, action="reject", decision_action="reject" + ) + supervisor, transport = _supervisor( + runs, + deployment=deployment, + executor=executor, + tasks=(200, {"accepted": True, "created": True, "task_id": "task_x"}), + poll=(200, {"decision": relay_body}), + ack=(200, {"accepted": True}), + ) + + report = supervisor.serve_once(wait_s=0.0) + + assert report.acknowledged == "accepted" + assert report.outcome is not None + assert report.outcome.action == "reject" + assert report.outcome.pause_id == relay_body["task_id"].removeprefix("task_") + + +def test_a_governed_refusal_is_acknowledged_and_re_raised(tmp_path): + """A refused answer must reach the operator, not vanish into the loop.""" + runs = tmp_path / "runs" + run, item = _halted_run(runs, tmp_path / "bundles", "one") + deployment = _deployment() + # A binding the engine will not admit: the relay resolves to the right + # pause, and revalidation still refuses it. + relay_body = _relayed_for( + run, item, deployment, binding_digest="sha256:" + "9" * 64 + ) + supervisor, transport = _supervisor( + runs, + deployment=deployment, + executor=_ResultExecutor(), + tasks=(200, {"accepted": True, "created": True}), + poll=(200, {"decision": relay_body}), + ack=(200, {"accepted": True}), + ) + + with pytest.raises(AttendedActionRefused): + supervisor.serve_once(wait_s=0.0) + + assert transport.calls[-1][1]["result"] == "refused" + + +def test_a_decision_matching_no_open_pause_is_acknowledged_stale_not_executed( + tmp_path, +): + runs = tmp_path / "runs" + run, item = _halted_run(runs, tmp_path / "bundles", "one") + deployment = _deployment() + relay_body = _relayed_for( + run, item, deployment, capability_digest="sha256:" + "f" * 64 + ) + supervisor, transport = _supervisor( + runs, + deployment=deployment, + executor=_ResultExecutor(), + tasks=(200, {"accepted": True, "created": True}), + poll=(200, {"decision": relay_body}), + ack=(200, {"accepted": True}), + ) + + report = supervisor.serve_once(wait_s=0.0) + + assert report.acknowledged == "stale" + assert report.outcome is None + ack = [payload for path, payload in transport.calls if path.endswith("/ack")] + assert ack and ack[0]["result"] == "stale" + + +def test_an_expired_decision_is_acknowledged_expired_not_executed(tmp_path): + runs = tmp_path / "runs" + run, item = _halted_run(runs, tmp_path / "bundles", "one") + deployment = _deployment() + relay_body = _relayed_for(run, item, deployment, expires_at=LONG_PAST) + supervisor, transport = _supervisor( + runs, + deployment=deployment, + executor=_ResultExecutor(), + tasks=(200, {"accepted": True, "created": True}), + poll=(200, {"decision": relay_body}), + ack=(200, {"accepted": True}), + ) + + report = supervisor.serve_once(wait_s=0.0) + + assert report.acknowledged == "expired" + assert report.outcome is None + + +def test_an_unreadable_deadline_is_treated_as_expired_not_as_open_ended(tmp_path): + runs = tmp_path / "runs" + run, item = _halted_run(runs, tmp_path / "bundles", "one") + deployment = _deployment() + relay_body = _relayed_for(run, item, deployment, expires_at="whenever") + supervisor, _ = _supervisor( + runs, + deployment=deployment, + executor=_ResultExecutor(), + tasks=(200, {"accepted": True, "created": True}), + poll=(200, {"decision": relay_body}), + ack=(200, {"accepted": True}), + ) + + assert supervisor.serve_once(wait_s=0.0).acknowledged == "expired" + + +def test_no_waiting_decision_is_a_quiet_cycle_not_an_error(tmp_path): + runs = tmp_path / "runs" + _halted_run(runs, tmp_path / "bundles", "one") + supervisor, _ = _supervisor( + runs, + tasks=(200, {"accepted": True, "created": True}), + poll=(204, {}), + ) + + report = supervisor.serve_once(wait_s=0.0) + + assert report.decision_id is None + assert report.acknowledged is None + assert len(report.publishes.published) == 1 + + +def test_an_unverifiable_decision_is_refused_before_any_resolution(tmp_path): + """A relay this runner's credential cannot have produced is never run.""" + runs = tmp_path / "runs" + run, item = _halted_run(runs, tmp_path / "bundles", "one") + deployment = _deployment() + forged = _relayed_for(run, item, deployment) + forged = {**forged, "relay_signature": "hmac-sha256:" + "0" * 64} + supervisor, _ = _supervisor( + runs, + deployment=deployment, + executor=_ResultExecutor(), + tasks=(200, {"accepted": True, "created": True}), + poll=(200, {"decision": forged}), + ) + + with pytest.raises(RelayRefused): + supervisor.serve_once(wait_s=0.0) + + +# ------------------------------------------------------------- the thread + + +class _StubSupervisor: + """A supervisor whose cycles are scripted, so the loop is testable.""" + + def __init__(self, script: list[Any]) -> None: + self.script = list(script) + self.calls = 0 + + def serve_once(self, *, wait_s: float = 0.0, publish: bool = True) -> Any: + self.calls += 1 + if not self.script: + raise RelayUncertain("no more script") + step = self.script.pop(0) + if isinstance(step, Exception): + raise step + return step + + +def _cycle(acknowledged: Optional[str] = None, unknown: tuple[str, ...] = ()): + from openadapt_flow.console.decision_supervisor import CycleReport, PublishReport + + return CycleReport( + publishes=PublishReport(unknown=unknown), acknowledged=acknowledged + ) + + +def test_the_loop_counts_outcomes_and_stops_when_asked(): + stub = _StubSupervisor([_cycle("accepted"), _cycle("stale"), _cycle("expired")]) + thread = DecisionSupervisorThread(stub, wait_s=0.0, sleep=lambda _s: None) + + stop_after = {"n": 0} + + def on_cycle(_report: Any) -> None: + stop_after["n"] += 1 + if stop_after["n"] >= 3: + thread.stop(timeout_s=0.0) + + thread._on_cycle = on_cycle + thread.run() + + assert thread.stats.cycles == 3 + assert thread.stats.decisions_executed == 1 + assert thread.stats.decisions_stale == 1 + assert thread.stats.decisions_expired == 1 + + +def test_a_relay_refusal_backs_the_loop_off_instead_of_spinning(): + slept: list[float] = [] + stub = _StubSupervisor( + [RelayRefused("control plane refused"), RelayRefused("again"), _cycle()] + ) + thread = DecisionSupervisorThread(stub, wait_s=0.0, sleep=slept.append) + + def on_cycle(_report: Any) -> None: + thread.stop(timeout_s=0.0) + + thread._on_cycle = on_cycle + thread.run() + + assert slept == [2.0, 4.0] + # A successful cycle clears the backoff, so an outage does not permanently + # slow the lane. + assert thread.stats.consecutive_failures == 0 + + +def test_a_governed_refusal_is_an_answer_not_an_outage(): + """A refused decision must not back the transport off; nothing is wrong. + + It does take one short pause. An acknowledgement can itself be uncertain, + which leaves the decision leased and re-delivered at once, and a poll with a + decision waiting returns immediately -- so the loop would otherwise spin at + full speed on a decision it refuses every time. + """ + slept: list[float] = [] + stub = _StubSupervisor( + [ + AttendedActionRefused("revalidation failed"), + AttendedActionRefused("again"), + _cycle(), + ] + ) + thread = DecisionSupervisorThread(stub, wait_s=0.0, sleep=slept.append) + + def on_cycle(_report: Any) -> None: + thread.stop(timeout_s=0.0) + + thread._on_cycle = on_cycle + thread.run() + + # A flat floor, not a doubling backoff: the transport is healthy. + assert slept == [1.0, 1.0] + assert thread.stats.decisions_refused == 2 + assert thread.stats.consecutive_failures == 0 + + +def test_the_loop_never_dies_on_an_unexpected_error(): + slept: list[float] = [] + stub = _StubSupervisor([ValueError("something unforeseen"), _cycle()]) + thread = DecisionSupervisorThread(stub, wait_s=0.0, sleep=slept.append) + + def on_cycle(_report: Any) -> None: + thread.stop(timeout_s=0.0) + + thread._on_cycle = on_cycle + thread.run() + + assert thread.stats.cycles == 1 + assert slept == [2.0] + + +def test_starting_twice_is_refused(): + stub = _StubSupervisor([]) + thread = DecisionSupervisorThread(stub, wait_s=0.0, sleep=lambda _s: None) + thread._stop.set() # so the thread exits immediately + thread.start() + with pytest.raises(RuntimeError): + thread.start() + thread.stop() + + +def test_the_thread_is_a_daemon_so_it_cannot_hold_the_console_open(): + stub = _StubSupervisor([]) + thread = DecisionSupervisorThread(stub, wait_s=0.0, sleep=lambda _s: None) + thread._stop.set() + thread.start() + assert isinstance(thread._thread, threading.Thread) + assert thread._thread.daemon is True + thread.stop() + assert thread.running is False + + +# ------------------------------------------------------------------- the CLI +# +# `--remote-decisions` must fail loudly, never quietly. An operator who asked +# for it and silently got a loopback-only console would believe a phone can +# answer a halt when nothing is listening for one. + + +def _console_args(*extra: str, config: Optional[Path] = None): + from openadapt_flow.__main__ import build_parser + + argv = ["console", *extra] + if config is not None: + argv += ["--config", str(config)] + return build_parser().parse_args(argv) + + +def _remote_config(tmp_path: Path, *, enabled: bool = True) -> Path: + import yaml + + path = tmp_path / "deployment.yaml" + path.write_text( + yaml.safe_dump( + { + "human_decisions": { + "remote": { + "enabled": enabled, + "tenant_id": "tenant_exact_01", + "runner_id": "runner_exact_01", + } + } + } + ) + ) + return path + + +def test_remote_decisions_is_off_unless_asked_for(): + from openadapt_flow.__main__ import _decision_supervisor_from_args + + args = _console_args("--attend") + assert _decision_supervisor_from_args(args, None) is None + + +def test_remote_decisions_refuses_a_read_only_console(tmp_path): + from openadapt_flow.__main__ import _decision_supervisor_from_args + + args = _console_args( + "--attend", "--remote-decisions", config=_remote_config(tmp_path) + ) + with pytest.raises(SystemExit, match="--attend --allow-actions"): + _decision_supervisor_from_args(args, None) + + +def test_remote_decisions_refuses_a_deployment_that_did_not_enable_it(tmp_path): + from openadapt_flow.__main__ import _decision_supervisor_from_args + + args = _console_args( + "--attend", + "--allow-actions", + "--remote-decisions", + config=_remote_config(tmp_path, enabled=False), + ) + with pytest.raises(SystemExit, match="human_decisions.remote.enabled"): + _decision_supervisor_from_args(args, object()) + + +def test_remote_decisions_refuses_without_a_runner_credential(tmp_path, monkeypatch): + from openadapt_flow.__main__ import _decision_supervisor_from_args + + monkeypatch.delenv("OPENADAPT_RUNNER_TOKEN", raising=False) + args = _console_args( + "--attend", + "--allow-actions", + "--remote-decisions", + config=_remote_config(tmp_path), + ) + with pytest.raises(SystemExit, match="OPENADAPT_RUNNER_TOKEN"): + _decision_supervisor_from_args(args, object()) + + +def test_remote_decisions_refuses_a_plaintext_control_plane(tmp_path, monkeypatch): + """A runner credential is never sent over plaintext, even to localhost.""" + from openadapt_flow.__main__ import _decision_supervisor_from_args + + monkeypatch.setenv("OPENADAPT_RUNNER_TOKEN", TOKEN) + args = _console_args( + "--attend", + "--allow-actions", + "--remote-decisions", + "--remote-decision-host", + "http://localhost:3000", + config=_remote_config(tmp_path), + ) + with pytest.raises(SystemExit, match="https"): + _decision_supervisor_from_args(args, object()) + + +def test_remote_decisions_builds_a_supervisor_when_fully_configured( + tmp_path, monkeypatch +): + from openadapt_flow.__main__ import _decision_supervisor_from_args + + monkeypatch.setenv("OPENADAPT_RUNNER_TOKEN", TOKEN) + args = _console_args( + "--attend", + "--allow-actions", + "--remote-decisions", + "--remote-decision-host", + "https://app.openadapt.test", + config=_remote_config(tmp_path), + ) + supervisor = _decision_supervisor_from_args(args, object()) + assert isinstance(supervisor, DecisionSupervisorThread) + assert supervisor.running is False + + +def test_the_console_starts_and_stops_the_lane_with_the_server(monkeypatch): + """The lane's lifetime is the server's; a stopped console leaves none.""" + import openadapt_flow.console.server as server_mod + + started: list[str] = [] + monkeypatch.setattr( + server_mod, "create_app", lambda *a, **k: object(), raising=False + ) + + class _FakeUvicorn: + @staticmethod + def run(*_args: Any, **_kwargs: Any) -> None: + started.append("served") + + class _Lane: + def __init__(self) -> None: + self.events: list[str] = [] + + def start(self) -> None: + self.events.append("start") + + def stop(self) -> None: + self.events.append("stop") + + lane = _Lane() + monkeypatch.setitem(__import__("sys").modules, "uvicorn", _FakeUvicorn) + monkeypatch.setattr( + "openadapt_flow.console.app.create_app", lambda *a, **k: object() + ) + server_mod.serve("bundles", "runs", None, decision_supervisor=lane, port=0) + + assert started == ["served"] + assert lane.events == ["start", "stop"]