feat(attended): give the bridge a brake, and say when a model pulled it

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

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(

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)