From 2bf32061fa4d067996702f2b9c541a556a46e201 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Mon, 27 Jul 2026 22:19:42 -0400 Subject: [PATCH 1/8] feat(decisions): run the outbound decision lane, so a halt reaches a phone `DecisionRelay` is a transport: it publishes ONE pause the caller already resolved and executes ONE decision against a run_dir and 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 is the missing half. `console.decision_supervisor.DecisionSupervisor` owns a `runs` root rather than one run. Every cycle it publishes EVERY currently open attended pause, so a halt becomes answerable at app.openadapt.ai without anyone calling anything. Resolution is by capability digest, not by position. `DecisionRelay.serve_once` takes its run and item as arguments, which is correct only when exactly one pause is open; a practice with two halted runs is the ordinary case, not the exotic one. `DecisionSupervisor.resolve` re-scans the runs root at decision time and requires BOTH the relayed `task_id` and the relayed `capability_digest` to equal what the engine's own signed capability file produces right now. That is not the safety boundary -- `execute_remote_attended_action` still revalidates the capability, takes the single-flight lease, re-reads the live application and re-proves every `will_recheck` contract. 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. A decision matching no open pause is acknowledged `stale`. One whose deadline has passed, or whose deadline cannot be parsed, is acknowledged `expired`. An unreadable deadline is not a licence to act. Neither is executed. `console --attend --allow-actions --remote-decisions` starts the lane as a daemon thread beside the loopback server. That process already owns the deployment-bound action service a continuation needs, and `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 start-up refusal is a hard exit, never a disabled feature: a missing runner token, remote issuance not enabled, a read-only console, or a plaintext control-plane origin each stop the console. 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. The vocabulary stays the relay's. A publish reports `published`, `already_published` or `unknown`, and an `unknown` pause is never described as reachable. Re-publishing after `unknown` is deliberate and is not a blind retry: `POST /api/human-decisions/tasks` is an idempotent upsert keyed on the signed task, and a divergent projection for the same task is rejected rather than overwriting, so the operation cannot happen twice. `docs/DECISION_DELIVERY.md` gains the exact command a customer runs and a "what is not done yet" section: Desktop does not pass the flag yet (the installer pins a frozen Flow, so the pin has to move first), and the control plane must be deployed before the lane carries context rather than identifiers alone. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01NyCHrzA1psrKMFfroYbzaM --- docs/DECISION_DELIVERY.md | 53 +- docs/EXCEPTION_INBOX.md | 10 + openadapt_flow/__main__.py | 86 +++ openadapt_flow/console/decision_supervisor.py | 441 +++++++++++ openadapt_flow/console/server.py | 22 +- tests/test_decision_supervisor.py | 685 ++++++++++++++++++ 6 files changed, 1294 insertions(+), 3 deletions(-) create mode 100644 openadapt_flow/console/decision_supervisor.py create mode 100644 tests/test_decision_supervisor.py 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_supervisor.py b/openadapt_flow/console/decision_supervisor.py new file mode 100644 index 00000000..16c72bdf --- /dev/null +++ b/openadapt_flow/console/decision_supervisor.py @@ -0,0 +1,441 @@ +"""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.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, +) + +#: 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 + + +@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 already held at this exact revision. + already_published: 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, ...] = () + + @property + def certain_count(self) -> int: + return len(self.published) + len(self.already_published) + + +@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 + + +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)) + + # -- 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, and requiring both + means a future change that decouples the two surfaces as a mismatch + rather than silently picking one. + """ + 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 + + # -- one cycle -------------------------------------------------------- + + def publish_open_pauses(self, *, timeout_s: float = 15.0) -> PublishReport: + """Make every open pause answerable from the hosted surface.""" + published: list[str] = [] + already: list[str] = [] + unknown: list[str] = [] + not_projectable: list[str] = [] + for pause in self.open_pauses(): + 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 + if outcome.state is PublishState.PUBLISHED: + published.append(pause.task_id) + elif outcome.state is PublishState.ALREADY_PUBLISHED: + already.append(pause.task_id) + else: + unknown.append(pause.task_id) + return PublishReport( + published=tuple(published), + already_published=tuple(already), + unknown=tuple(unknown), + not_projectable=tuple(not_projectable), + ) + + 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) + + 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" + self._relay.acknowledge(decision, result) + 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 back + # the loop off. + self.stats.decisions_refused += 1 + self.stats.last_error = type(exc).__name__ + 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": + 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/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/tests/test_decision_supervisor.py b/tests/test_decision_supervisor.py new file mode 100644 index 00000000..c7c909c7 --- /dev/null +++ b/tests/test_decision_supervisor.py @@ -0,0 +1,685 @@ +"""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 threading +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 ( + AttendedActionRefused, + AttendedActionStore, +) +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() + + import json + + 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() == [] + + +# ---------------------------------------------------- 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" + + +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.""" + slept: list[float] = [] + stub = _StubSupervisor([AttendedActionRefused("revalidation failed"), _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 == [] + assert thread.stats.decisions_refused == 1 + + +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"] From 9d3fc5fa715477defb58e88a7a88a81e38630c6b Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Mon, 27 Jul 2026 22:42:33 -0400 Subject: [PATCH 2/8] fix(decisions): one refused pause must not silence the rest of the queue `publish_open_pauses` let a `RelayRefused` propagate. A single pause the control plane rejects would therefore abort the whole pass, back the loop off, and leave every OTHER halt in the practice unreachable on a phone -- the failure mode is inverted from the one the refusal is protecting against. It is now recorded per pause, in a new `PublishReport.refused`, and the loop continues. Also stop re-POSTing a pause the control plane has already accepted. A pause can stay open for hours, and the signed task is a deterministic function of the capability, so publishing it every 25-second cycle sends identical bytes for no new information. The memo is keyed on `(task_id, capability_digest)` and REBUILT each pass rather than updated, so a pause that closed stops being remembered and a later pause that reuses its identity is republished. It is in-process only, deliberately: a restarted supervisor republishes, which is idempotent, rather than trusting a file's claim that a remote surface still holds something. An `unknown` publish is deliberately not memoized. It is re-POSTed next cycle, which either resolves the uncertainty or leaves it unchanged, and cannot happen twice because the endpoint is an idempotent upsert keyed on the signed task. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01NyCHrzA1psrKMFfroYbzaM --- openadapt_flow/console/decision_supervisor.py | 39 +++++++++- tests/test_decision_supervisor.py | 75 +++++++++++++++++++ 2 files changed, 113 insertions(+), 1 deletion(-) diff --git a/openadapt_flow/console/decision_supervisor.py b/openadapt_flow/console/decision_supervisor.py index 16c72bdf..e954a4a5 100644 --- a/openadapt_flow/console/decision_supervisor.py +++ b/openadapt_flow/console/decision_supervisor.py @@ -119,6 +119,9 @@ class PublishReport: #: 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: @@ -177,6 +180,11 @@ def __init__( 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 --------------------------------------------------------- @@ -237,12 +245,29 @@ def resolve(self, decision: RelayedDecision) -> Optional[OpenPause]: # -- one cycle -------------------------------------------------------- def publish_open_pauses(self, *, timeout_s: float = 15.0) -> PublishReport: - """Make every open pause answerable from the hosted surface.""" + """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] = [] + 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. + already.append(pause.task_id) + confirmed.add(key) + continue try: outcome: PublishOutcome = self._relay.publish( pause.run_dir, pause.item, timeout_s=timeout_s @@ -253,17 +278,29 @@ def publish_open_pauses(self, *, timeout_s: float = 15.0) -> PublishReport: # 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), unknown=tuple(unknown), not_projectable=tuple(not_projectable), + refused=tuple(refused), ) def serve_once( diff --git a/tests/test_decision_supervisor.py b/tests/test_decision_supervisor.py index c7c909c7..d4f3c57c 100644 --- a/tests/test_decision_supervisor.py +++ b/tests/test_decision_supervisor.py @@ -227,6 +227,81 @@ def test_a_run_that_is_not_durably_paused_is_never_published(tmp_path): 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) + + +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 + assert len(second.already_published) == 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 From 0c41c1076163925e990bf704944c97a0c89fe183 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Mon, 27 Jul 2026 22:44:28 -0400 Subject: [PATCH 3/8] fix(decisions): do not report a memoized publish as an observation `already_published` means "the control plane said it already held this". A pause skipped by the in-process memo was not asked about this pass, so counting it there claims an observation the supervisor did not make -- the same discipline the relay applies when it refuses to say "delivered". It now reports under `previously_confirmed`. `certain_count` still counts it, because the pause IS answerable; only the provenance of that knowledge is now distinguishable by a caller rendering operator copy. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01NyCHrzA1psrKMFfroYbzaM --- openadapt_flow/console/decision_supervisor.py | 19 ++++++++++++++++--- tests/test_decision_supervisor.py | 6 +++++- 2 files changed, 21 insertions(+), 4 deletions(-) diff --git a/openadapt_flow/console/decision_supervisor.py b/openadapt_flow/console/decision_supervisor.py index e954a4a5..e2c2dc9e 100644 --- a/openadapt_flow/console/decision_supervisor.py +++ b/openadapt_flow/console/decision_supervisor.py @@ -109,8 +109,14 @@ class PublishReport: #: Pauses the control plane accepted for the first time. published: tuple[str, ...] = () - #: Pauses the control plane already held at this exact revision. + #: 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. @@ -125,7 +131,12 @@ class PublishReport: @property def certain_count(self) -> int: - return len(self.published) + len(self.already_published) + """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) @@ -257,6 +268,7 @@ def publish_open_pauses(self, *, timeout_s: float = 15.0) -> PublishReport: 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) @@ -265,7 +277,7 @@ def publish_open_pauses(self, *, timeout_s: float = 15.0) -> PublishReport: # 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. - already.append(pause.task_id) + memoized.append(pause.task_id) confirmed.add(key) continue try: @@ -298,6 +310,7 @@ def publish_open_pauses(self, *, timeout_s: float = 15.0) -> PublishReport: return PublishReport( published=tuple(published), already_published=tuple(already), + previously_confirmed=tuple(memoized), unknown=tuple(unknown), not_projectable=tuple(not_projectable), refused=tuple(refused), diff --git a/tests/test_decision_supervisor.py b/tests/test_decision_supervisor.py index d4f3c57c..0c9b8d36 100644 --- a/tests/test_decision_supervisor.py +++ b/tests/test_decision_supervisor.py @@ -280,7 +280,11 @@ def test_an_accepted_pause_is_not_republished_every_cycle(tmp_path): second = supervisor.publish_open_pauses() assert len(first.published) == 1 - assert len(second.already_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 From 0fc31396140cef56a81bcafe0341d7d26d476c19 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Mon, 27 Jul 2026 22:46:30 -0400 Subject: [PATCH 4/8] docs(decisions): say why an unresolvable decision is the safe failure Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01NyCHrzA1psrKMFfroYbzaM --- openadapt_flow/console/decision_supervisor.py | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/openadapt_flow/console/decision_supervisor.py b/openadapt_flow/console/decision_supervisor.py index e2c2dc9e..e40cba80 100644 --- a/openadapt_flow/console/decision_supervisor.py +++ b/openadapt_flow/console/decision_supervisor.py @@ -239,9 +239,14 @@ 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, and requiring both - means a future change that decouples the two surfaces as a mismatch - rather than silently picking one. + 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") From 5f7a569bdf6c29aabce591241e39551ab2af08fd Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Mon, 27 Jul 2026 22:58:49 -0400 Subject: [PATCH 5/8] fix(decisions): stop the loop spinning on a decision it always refuses A governed refusal is acknowledged and re-raised, and the loop treats it as an answer rather than an outage -- correctly, because the transport is healthy and a refusal must not slow the lane down for every other pause. But if the acknowledgement is itself uncertain, the decision stays leased server-side and is re-delivered at once, and a poll with a decision waiting returns immediately. The loop would then spin at full speed on a decision it refuses every time. It now takes a flat one-second floor after a refusal. Not a backoff: the consecutive-failure counter is untouched, so a refusal never raises the delay for anything else. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01NyCHrzA1psrKMFfroYbzaM --- openadapt_flow/console/decision_supervisor.py | 18 +++++++++++++-- tests/test_decision_supervisor.py | 22 +++++++++++++++---- 2 files changed, 34 insertions(+), 6 deletions(-) diff --git a/openadapt_flow/console/decision_supervisor.py b/openadapt_flow/console/decision_supervisor.py index e40cba80..4bfa5749 100644 --- a/openadapt_flow/console/decision_supervisor.py +++ b/openadapt_flow/console/decision_supervisor.py @@ -92,6 +92,11 @@ 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: @@ -457,10 +462,19 @@ def run(self) -> None: 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 back - # the loop off. + # 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 diff --git a/tests/test_decision_supervisor.py b/tests/test_decision_supervisor.py index 0c9b8d36..48f10b09 100644 --- a/tests/test_decision_supervisor.py +++ b/tests/test_decision_supervisor.py @@ -562,9 +562,21 @@ def on_cycle(_report: Any) -> None: def test_a_governed_refusal_is_an_answer_not_an_outage(): - """A refused decision must not back the transport off; nothing is wrong.""" + """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"), _cycle()]) + stub = _StubSupervisor( + [ + AttendedActionRefused("revalidation failed"), + AttendedActionRefused("again"), + _cycle(), + ] + ) thread = DecisionSupervisorThread(stub, wait_s=0.0, sleep=slept.append) def on_cycle(_report: Any) -> None: @@ -573,8 +585,10 @@ def on_cycle(_report: Any) -> None: thread._on_cycle = on_cycle thread.run() - assert slept == [] - assert thread.stats.decisions_refused == 1 + # 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(): From 153532298f838c9534850f0cf4658f296494393b Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Mon, 27 Jul 2026 23:10:50 -0400 Subject: [PATCH 6/8] test(decisions): pin that a phone's Reject ends the run it names `reject` terminates a run and `escalate` parks it, so a Reject delivered to the wrong open pause is the worst outcome this lane can produce. The test opens two halted runs, rejects the second, and asserts the executed decision carries that pause id. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01NyCHrzA1psrKMFfroYbzaM --- tests/test_decision_supervisor.py | 33 +++++++++++++++++++++++++++++++ 1 file changed, 33 insertions(+) diff --git a/tests/test_decision_supervisor.py b/tests/test_decision_supervisor.py index 48f10b09..966fdfcf 100644 --- a/tests/test_decision_supervisor.py +++ b/tests/test_decision_supervisor.py @@ -371,6 +371,39 @@ def test_one_cycle_publishes_executes_and_acknowledges_without_a_caller(tmp_path assert ack_body["result"] == "accepted" +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" From 07d43f712b11f0ad887b58bcde055e79c65596e8 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Tue, 28 Jul 2026 01:25:12 -0400 Subject: [PATCH 7/8] fix(decisions): preserve completed outcome across lost ack --- openadapt_flow/console/decision_supervisor.py | 94 ++++++++++++- tests/test_decision_supervisor.py | 123 ++++++++++++++++++ 2 files changed, 215 insertions(+), 2 deletions(-) diff --git a/openadapt_flow/console/decision_supervisor.py b/openadapt_flow/console/decision_supervisor.py index 4bfa5749..fafcbb4c 100644 --- a/openadapt_flow/console/decision_supervisor.py +++ b/openadapt_flow/console/decision_supervisor.py @@ -157,6 +157,58 @@ class CycleReport: 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 memory. + #: The action is never executed a second time on this path. + reacknowledged: bool = False + + +@dataclass(frozen=True) +class _PendingAcknowledgement: + """One completed engine decision whose hosted acknowledgement is uncertain. + + ``relay_digest`` already binds every signed relay field. The explicit + signature and idempotency fields make the recovery invariant visible here: + a re-delivery must be the exact signed decision, not merely the same opaque + decision id. + """ + + relay_digest: str + relay_signature: str + idempotency_key: str + capability_digest: str + action: str + result: str + outcome: AttendedDecision + + @classmethod + def completed( + cls, + decision: RelayedDecision, + *, + result: str, + outcome: AttendedDecision, + ) -> "_PendingAcknowledgement": + relay = decision.relay + return cls( + relay_digest=decision.relay_digest, + relay_signature=decision.relay_signature, + idempotency_key=str(relay["idempotency_key"]), + capability_digest=str(relay["capability_digest"]), + action=str(relay["action"]), + result=result, + outcome=outcome, + ) + + def matches(self, decision: RelayedDecision) -> bool: + relay = decision.relay + return ( + self.relay_digest == decision.relay_digest + and self.relay_signature == decision.relay_signature + and self.idempotency_key == str(relay["idempotency_key"]) + and self.capability_digest == str(relay["capability_digest"]) + and self.action == str(relay["action"]) + ) def _parse_rfc3339(value: object) -> Optional[datetime]: @@ -201,6 +253,11 @@ def __init__( #: 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() + #: Completed local outcomes whose acknowledgement may not have reached + #: the control plane. This is deliberately process-local: it closes + #: the ordinary lost-response/re-delivery loop without minting a second + #: durable authority beside the engine's existing decision journal. + self._pending_acknowledgements: dict[str, _PendingAcknowledgement] = {} # -- scanning --------------------------------------------------------- @@ -346,6 +403,24 @@ def serve_once( if decision is None: return CycleReport(publishes=publishes) + pending_ack = self._pending_acknowledgements.get(decision.decision_id) + if pending_ack is not None: + if not pending_ack.matches(decision): + raise RelayRefused( + "a re-delivered decision changed its exact signed or " + "idempotency binding" + ) + confirmed = self._relay.acknowledge(decision, pending_ack.result) + if confirmed: + self._pending_acknowledgements.pop(decision.decision_id, None) + return CycleReport( + publishes=publishes, + decision_id=decision.decision_id, + acknowledged=pending_ack.result, + outcome=pending_ack.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") @@ -372,7 +447,22 @@ def serve_once( self._relay.acknowledge(decision, "refused") raise result = "accepted" if outcome.status != "refused" else "refused" - self._relay.acknowledge(decision, result) + # Record the completed local result BEFORE the network acknowledgement. + # If that response is lost, the control plane re-delivers the decision. + # Continue, Skip, and Reject can close the pause, so resolving only from + # the current open queue would then misreport a real completed action as + # stale. The next cycle instead re-acknowledges this exact signed relay + # without executing it again. + self._pending_acknowledgements[decision.decision_id] = ( + _PendingAcknowledgement.completed( + decision, + result=result, + outcome=outcome, + ) + ) + confirmed = self._relay.acknowledge(decision, result) + if confirmed: + self._pending_acknowledgements.pop(decision.decision_id, None) return CycleReport( publishes=publishes, decision_id=decision.decision_id, @@ -494,7 +584,7 @@ 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": + if report.acknowledged == "accepted" and not report.reacknowledged: self.stats.decisions_executed += 1 elif report.acknowledged == "refused": self.stats.decisions_refused += 1 diff --git a/tests/test_decision_supervisor.py b/tests/test_decision_supervisor.py index 966fdfcf..0703d33f 100644 --- a/tests/test_decision_supervisor.py +++ b/tests/test_decision_supervisor.py @@ -244,6 +244,29 @@ def post(self, path, payload, *, timeout_s): 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 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" @@ -371,6 +394,106 @@ def test_one_cycle_publishes_executes_and_acknowledges_without_a_caller(tmp_path 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_reacknowledges_the_exact_decision_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, + ) + transport = _SequencedTransport( + polls=[ + (200, {"decision": relay_body}), + (200, {"decision": relay_body}), + ], + acknowledgements=[ + RelayUncertain("the engine result reached an uncertain ack"), + (200, {"accepted": True}), + ], + ) + relay = DecisionRelay(transport, token=TOKEN, deployment=deployment) + supervisor = DecisionSupervisor( + runs, relay=relay, deployment=deployment, executor=executor + ) + + first = supervisor.serve_once(wait_s=0.0) + 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 path, body in transport.calls if path.endswith("/ack")] + assert [body["result"] for body in acknowledgements] == [ + "accepted", + "accepted", + ] + + +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) + transport = _SequencedTransport( + polls=[ + (200, {"decision": relay_body}), + (200, {"decision": changed_relay}), + ], + acknowledgements=[ + RelayUncertain("the first ack response was lost"), + ], + ) + relay = DecisionRelay(transport, token=TOKEN, deployment=deployment) + supervisor = DecisionSupervisor( + runs, relay=relay, deployment=deployment, executor=executor + ) + + supervisor.serve_once(wait_s=0.0) + with pytest.raises(RelayRefused, match="exact signed or idempotency binding"): + supervisor.serve_once(wait_s=0.0) + + assert executor.calls == 1 + assert len([path for path, _ in transport.calls if path.endswith("/ack")]) == 1 + + 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. From f0e1220c0e3aa33edc1414fcdb57d589bdf009d9 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Tue, 28 Jul 2026 01:50:28 -0400 Subject: [PATCH 8/8] fix(decisions): recover lost acknowledgements after restart --- openadapt_flow/console/decision_relay.py | 14 + openadapt_flow/console/decision_supervisor.py | 134 ++--- openadapt_flow/console/human_decisions.py | 3 + openadapt_flow/runtime/durable/attended.py | 542 +++++++++++++++++- tests/test_decision_supervisor.py | 410 ++++++++++++- 5 files changed, 985 insertions(+), 118 deletions(-) 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 index fafcbb4c..7c7ab0ea 100644 --- a/openadapt_flow/console/decision_supervisor.py +++ b/openadapt_flow/console/decision_supervisor.py @@ -63,6 +63,7 @@ from pathlib import Path from typing import Callable, Optional +from openadapt_flow.console import data from openadapt_flow.console.attention import ( AttentionItem, list_attention, @@ -81,6 +82,7 @@ AttendedActionRefused, AttendedActionStore, AttendedDecision, + AttendedRelayAcknowledgement, ) #: Seconds a poll waits for an answer before the loop takes another turn. Also @@ -158,59 +160,12 @@ class CycleReport: #: 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 memory. + #: 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 -@dataclass(frozen=True) -class _PendingAcknowledgement: - """One completed engine decision whose hosted acknowledgement is uncertain. - - ``relay_digest`` already binds every signed relay field. The explicit - signature and idempotency fields make the recovery invariant visible here: - a re-delivery must be the exact signed decision, not merely the same opaque - decision id. - """ - - relay_digest: str - relay_signature: str - idempotency_key: str - capability_digest: str - action: str - result: str - outcome: AttendedDecision - - @classmethod - def completed( - cls, - decision: RelayedDecision, - *, - result: str, - outcome: AttendedDecision, - ) -> "_PendingAcknowledgement": - relay = decision.relay - return cls( - relay_digest=decision.relay_digest, - relay_signature=decision.relay_signature, - idempotency_key=str(relay["idempotency_key"]), - capability_digest=str(relay["capability_digest"]), - action=str(relay["action"]), - result=result, - outcome=outcome, - ) - - def matches(self, decision: RelayedDecision) -> bool: - relay = decision.relay - return ( - self.relay_digest == decision.relay_digest - and self.relay_signature == decision.relay_signature - and self.idempotency_key == str(relay["idempotency_key"]) - and self.capability_digest == str(relay["capability_digest"]) - and self.action == str(relay["action"]) - ) - - def _parse_rfc3339(value: object) -> Optional[datetime]: """Parse a relay timestamp, or ``None`` if it is not one. @@ -253,11 +208,6 @@ def __init__( #: 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() - #: Completed local outcomes whose acknowledgement may not have reached - #: the control plane. This is deliberately process-local: it closes - #: the ordinary lost-response/re-delivery loop without minting a second - #: durable authority beside the engine's existing decision journal. - self._pending_acknowledgements: dict[str, _PendingAcknowledgement] = {} # -- scanning --------------------------------------------------------- @@ -320,6 +270,41 @@ def resolve(self, decision: RelayedDecision) -> Optional[OpenPause]: 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: @@ -403,21 +388,19 @@ def serve_once( if decision is None: return CycleReport(publishes=publishes) - pending_ack = self._pending_acknowledgements.get(decision.decision_id) - if pending_ack is not None: - if not pending_ack.matches(decision): - raise RelayRefused( - "a re-delivered decision changed its exact signed or " - "idempotency binding" - ) - confirmed = self._relay.acknowledge(decision, pending_ack.result) + 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: - self._pending_acknowledgements.pop(decision.decision_id, None) + AttendedActionStore(run_dir).confirm_relay_acknowledgement( + decision.durable_binding() + ) return CycleReport( publishes=publishes, decision_id=decision.decision_id, - acknowledged=pending_ack.result, - outcome=pending_ack.outcome, + acknowledged=record.engine_ack_result, + outcome=outcome, reacknowledged=True, ) @@ -447,22 +430,15 @@ def serve_once( self._relay.acknowledge(decision, "refused") raise result = "accepted" if outcome.status != "refused" else "refused" - # Record the completed local result BEFORE the network acknowledgement. - # If that response is lost, the control plane re-delivers the decision. - # Continue, Skip, and Reject can close the pause, so resolving only from - # the current open queue would then misreport a real completed action as - # stale. The next cycle instead re-acknowledges this exact signed relay - # without executing it again. - self._pending_acknowledgements[decision.decision_id] = ( - _PendingAcknowledgement.completed( - decision, - result=result, - outcome=outcome, - ) - ) + # ``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: - self._pending_acknowledgements.pop(decision.decision_id, None) + AttendedActionStore(pause.run_dir).confirm_relay_acknowledgement( + decision.durable_binding() + ) return CycleReport( publishes=publishes, decision_id=decision.decision_id, 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/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 index 0703d33f..d476c98f 100644 --- a/tests/test_decision_supervisor.py +++ b/tests/test_decision_supervisor.py @@ -14,7 +14,11 @@ from __future__ import annotations +import hashlib +import json +import os import threading +import time from pathlib import Path from typing import Any, Optional @@ -39,8 +43,10 @@ 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 @@ -173,8 +179,6 @@ def test_nothing_protected_reaches_the_wire_from_the_whole_queue(tmp_path): ) supervisor.publish_open_pauses() - import json - body = json.dumps([payload for _, payload in transport.calls]) assert PROTECTED_VALUE not in body assert "supervisor-one" not in body @@ -267,6 +271,78 @@ def post(self, path, payload, *, timeout_s): 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" @@ -401,7 +477,7 @@ def test_one_cycle_publishes_executes_and_acknowledges_without_a_caller(tmp_path ("reject", "reject", "rejected", 0), ], ) -def test_a_lost_ack_reacknowledges_the_exact_decision_without_a_second_action( +def test_a_lost_ack_survives_restart_without_a_second_action( tmp_path, action, decision_action, @@ -425,23 +501,27 @@ def test_a_lost_ack_reacknowledges_the_exact_decision_without_a_second_action( action=action, decision_action=decision_action, ) - transport = _SequencedTransport( - polls=[ - (200, {"decision": relay_body}), - (200, {"decision": relay_body}), - ], - acknowledgements=[ - RelayUncertain("the engine result reached an uncertain ack"), - (200, {"accepted": True}), - ], + first_transport = _SequencedTransport( + polls=[(200, {"decision": relay_body})], + acknowledgements=[RelayUncertain("the engine result reached an uncertain ack")], ) - relay = DecisionRelay(transport, token=TOKEN, deployment=deployment) - supervisor = DecisionSupervisor( - runs, relay=relay, deployment=deployment, executor=executor + 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 = supervisor.serve_once(wait_s=0.0) + 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 @@ -451,11 +531,23 @@ def test_a_lost_ack_reacknowledges_the_exact_decision_without_a_second_action( assert second.reacknowledged is True assert second.outcome == first.outcome assert executor.calls == executor_calls - acknowledgements = [body for path, body in transport.calls if path.endswith("/ack")] + 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): @@ -472,26 +564,286 @@ def test_lost_ack_recovery_refuses_a_changed_signed_or_idempotency_binding(tmp_p } changed["idempotency_key"] = "relay-idempotency-key-CHANGED-0002" changed_relay = _sign(changed) - transport = _SequencedTransport( - polls=[ - (200, {"decision": relay_body}), - (200, {"decision": changed_relay}), - ], + first_transport = _SequencedTransport( + polls=[(200, {"decision": relay_body})], acknowledgements=[ RelayUncertain("the first ack response was lost"), ], ) - relay = DecisionRelay(transport, token=TOKEN, deployment=deployment) - supervisor = DecisionSupervisor( - runs, relay=relay, deployment=deployment, executor=executor + 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 ) - supervisor.serve_once(wait_s=0.0) with pytest.raises(RelayRefused, match="exact signed or idempotency binding"): - supervisor.serve_once(wait_s=0.0) + second_supervisor.serve_once(wait_s=0.0) assert executor.calls == 1 - assert len([path for path, _ in transport.calls if path.endswith("/ack")]) == 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):