diff --git a/pyproject.toml b/pyproject.toml index bfe5e2d..76e3049 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -32,7 +32,12 @@ keywords = ["mcp", "agent-skills", "gui", "automation", "workflow", "openadapt", dependencies = [ # Governed workflow compiler/runtime this package bridges. Execution # shells out to its CLI; bundle metadata loads via its IR. - "openadapt-flow>=1.18.1,<2", + # FLOOR IS LOAD-BEARING. `reject`, its `rejected_by_operator` disposition, + # and the `decided_by` provenance keyword are named symbols that only exist + # from 1.26.0. A resolver satisfied by 1.25.x fails at CALL time, inside the + # attended decision path, with a TypeError rather than at resolve. Raise + # this with every new symbol taken from Flow's attended contract. + "openadapt-flow>=1.26.0,<2", # Official Model Context Protocol SDK (stdio server transport). "mcp>=1.28,<2", # Structured concurrency runtime the mcp SDK already uses; we call diff --git a/src/openadapt_agent/attended.py b/src/openadapt_agent/attended.py index 4cfe20a..7d7769f 100644 --- a/src/openadapt_agent/attended.py +++ b/src/openadapt_agent/attended.py @@ -46,6 +46,14 @@ "The escalation was recorded and the durable pause remains available " "for qualified assistance." ), + # Deliberately not worded like the escalation above. That one says the + # pause remains; this one says the run is over. A caller told the wrong + # one of those acts on it. + "rejected": ( + "The rejection was recorded and the run is terminal. Nothing was " + "actuated, no approval can resume it, and the durable pause is " + "retained only as the audit record of what was rejected." + ), } @@ -130,6 +138,22 @@ class AttendedTool: "pause for a qualified operator." ), ), + "reject_attention": AttendedTool( + action="reject", + confirmation="confirm_run_must_not_proceed", + disposition="rejected_by_operator", + description=( + "End this run because it must not proceed. Use it only after " + "reading the live application and concluding OpenAdapt was RIGHT " + "to stop. This TERMINATES the run: no approval resumes it, and " + "the durable pause is kept only as the audit record. It is not " + "escalate_attention, which parks the run for a colleague who can " + "still continue it, and it is not teach_attention, which changes " + "future runs. It asserts nothing about the saved workflow and " + "actuates nothing. Flow independently refuses a rejection whose " + "delivery may already have landed." + ), + ), } @@ -206,9 +230,22 @@ def live_actions_ready(self) -> bool: return self.allow_actions and self.service is not None def enabled_action_tools(self) -> tuple[str, ...]: + """Which action tools this bridge exposes, and why reject is not gated. + + ``reject_attention`` sits with teach and escalate rather than behind + ``live_actions_ready``. That gate exists because continue and skip need + Flow's deployment-bound live executor to re-read the application and + act on it. Rejecting actuates nothing and resumes nothing, so it has + nothing to gate on -- the same reason Flow's own ``_allowed_actions`` + offers it at a pause carrying no resolvable action step at all. + + Withholding it would also leave this bridge able to say "proceed" -- + which writes to the system of record -- while unable to say "stop". + The only thing removed by that configuration is the brake. + """ if not self.allow_actions: return () - tools = ["teach_attention", "escalate_attention"] + tools = ["reject_attention", "teach_attention", "escalate_attention"] if self.live_actions_ready: tools[0:0] = ["continue_attention", "skip_attention"] return tuple(tools) @@ -327,17 +364,24 @@ def act(self, tool_name: str, arguments: dict[str, Any]) -> dict[str, Any]: disposition=tool.disposition, ) try: + # This bridge submits on behalf of a MODEL, and it derives + # `operator` from the same local OS identity a person's own console + # uses -- so the identity alone cannot tell the two apart. Declared + # here, at the one place that knows, so an agreement rate computed + # over Flow's journal can filter to decisions people actually made. if self.service is not None: decision = self.service.execute( path, request, operator=self.operator, + decided_by="automation", ) else: decision = execute_attended_action( path, request, operator=self.operator, + decided_by="automation", ) except (ApprovalRequired, AttendedActionRefused, ResumeRefused) as exc: _LOG.info( diff --git a/tests/test_attended_bridge.py b/tests/test_attended_bridge.py index aee6031..c29da14 100644 --- a/tests/test_attended_bridge.py +++ b/tests/test_attended_bridge.py @@ -4,6 +4,7 @@ import json import os +from pathlib import Path import pytest @@ -144,14 +145,19 @@ class DirectFlowService: def __init__(self, executor): self.executor = executor + self.deciders = [] - def execute(self, run_dir, request, *, operator): + def execute(self, run_dir, request, *, operator, decided_by="unknown"): from openadapt_flow.runtime.durable import execute_attended_action + # Recorded rather than defaulted: a fake that silently accepted + # `unknown` would hide a bridge that stopped declaring itself. + self.deciders.append(decided_by) return execute_attended_action( run_dir, request, operator=operator, + decided_by=decided_by, executor=self.executor, ) @@ -173,6 +179,7 @@ def item_and_args(bridge, *, action="continue"): confirmation = { "continue": "human_completed", "skip": "confirmed_not_applicable", + "reject": "confirm_run_must_not_proceed", "teach": "request_demonstration", "escalate": "request_assistance", }[action] @@ -200,8 +207,11 @@ def test_needs_attention_is_phi_safe_and_always_readable(paused_attention): item = listing["items"][0] assert item["human_required"] is True + # Relayed verbatim from Flow's signed capability. `reject` appears because + # the engine offers it at this pause, not because the bridge added it. assert item["capability"]["allowed_actions"] == [ "continue", + "reject", "teach", "escalate", ] @@ -285,8 +295,9 @@ def test_stale_capability_extra_fields_and_false_confirmation_never_execute( def test_flow_refusal_details_do_not_cross_the_phi_safe_bridge(paused_attention): class ProtectedRefusalService: - def execute(self, _run_dir, _request, *, operator): + def execute(self, _run_dir, _request, *, operator, decided_by="unknown"): assert operator + assert decided_by == "automation" from openadapt_flow.runtime.durable import AttendedActionRefused raise AttendedActionRefused("protected patient name and local workflow details") @@ -511,3 +522,116 @@ def test_symlinked_runs_root_is_not_scanned(paused_attention, tmp_path): alias.symlink_to(paused_attention["runs"], target_is_directory=True) bridge = AttendedBridge(alias) assert bridge.list()["items"] == [] + + +# --------------------------------------------------------------------------- +# Reject, and saying which kind of decider answered +# --------------------------------------------------------------------------- + + +def _journal(runs_dir): + """Flow's append-only attended decision records for the single paused run.""" + from openadapt_flow.runtime.durable.attended import AttendedActionStore + + run = next(p for p in Path(runs_dir).iterdir() if p.is_dir()) + return AttendedActionStore(run)._read_log().decisions + + +def test_reject_is_offered_without_a_live_executor_unlike_continue_and_skip( + paused_attention, +): + """The brake must not be gated on the machinery only the accelerator needs. + + ``live_actions_ready`` gates continue and skip because they need Flow's + deployment-bound executor to re-read the application and act on it. + Rejecting actuates nothing and resumes nothing, so it has nothing to gate + on. Gating it anyway would leave a configuration that can say "proceed" -- + which writes to the system of record -- but cannot say "stop", and the only + thing that removes is the brake. + """ + bridge = make_bridge(paused_attention, allow_actions=True, service=None) + specs = {spec.name for spec in bridge.list_tool_specs()} + + assert "reject_attention" in specs + assert {"continue_attention", "skip_attention"} & specs == set(), ( + "the premise is that no live executor is configured here" + ) + + # And it is still off entirely when actions are not enabled at all. + assert "reject_attention" not in { + spec.name for spec in make_bridge(paused_attention).list_tool_specs() + } + + +def test_reject_ends_the_run_and_is_recorded_as_an_automated_decision( + paused_attention, +): + """Two properties in one path, because they are only meaningful together. + + The bridge can now end a run. That is precisely why the record must say a + MODEL ended it: `operator` is derived from the same local OS identity a + person's own console uses, so without this the two are indistinguishable + and any agreement rate silently mixes them. + """ + bridge = make_bridge(paused_attention, allow_actions=True, service=None) + item, arguments = item_and_args(bridge, action="reject") + + result = bridge.dispatch("reject_attention", dict(arguments)) + assert result["action"] == "reject" + assert result["status"] == "rejected" + assert result["success"] is False + # The rejection copy must not read like the escalation copy: that one says + # the pause remains, this one says the run is over. + assert "terminal" in result["message"] + assert "remains available" not in result["message"] + + decision = _journal(paused_attention["runs"])[-1] + assert decision.action == "reject" + assert decision.decided_by == "automation", ( + "a model's answer must never be recorded as a person's" + ) + + +def test_every_bridge_decision_declares_itself_automated_including_continue( + paused_attention, +): + """Not only reject. The bias this fixes was always about `continue`. + + `continue` resumes the run and can write to the system of record, and it + has been model-callable all along. If provenance were attached only to the + action added last, the population that actually matters would stay + unlabelled. + """ + executor = ResultExecutor() + service = DirectFlowService(executor) + bridge = make_bridge(paused_attention, allow_actions=True, service=service) + item, arguments = item_and_args(bridge, action="continue") + + bridge.dispatch("continue_attention", dict(arguments)) + + assert service.deciders == ["automation"], service.deciders + assert _journal(paused_attention["runs"])[-1].decided_by == "automation" + + +def test_the_reject_tool_schema_is_closed_and_needs_explicit_confirmation( + paused_attention, +): + """Ending a run is not something a model may do by omission.""" + bridge = make_bridge(paused_attention, allow_actions=True, service=None) + schema = next( + spec.input_schema for spec in bridge.list_tool_specs() if spec.name == "reject_attention" + ) + assert schema["additionalProperties"] is False + assert schema["properties"]["confirm_run_must_not_proceed"]["const"] is True + assert "confirm_run_must_not_proceed" in schema["required"] + # No free text may reach a governed decision from a model. + assert not any( + prop.get("type") == "string" and "pattern" not in prop and "const" not in prop + for name, prop in schema["properties"].items() + ), schema["properties"] + + item, arguments = item_and_args(bridge, action="reject") + arguments["confirm_run_must_not_proceed"] = False + # `dispatch` re-raises the bridge's refusal as the public BridgeError. + with pytest.raises(BridgeError, match="must be explicitly true"): + bridge.dispatch("reject_attention", arguments)