feat(mcp): read the approval mode; answer a waiting save as a delegated agent (carry-over of openadapt-agent#80)

packages/agent/README.md

@@ -276,7 +276,8 @@ Flow's [execution profiles](https://github.com/OpenAdaptAI/openadapt-flow/blob/m
 | `--config YAML` | Flow's deployment config ([example](https://github.com/OpenAdaptAI/openadapt-flow/blob/main/docs/deployment.example.yaml)), which sets the profile and whether model grounding is allowed. |
 | `--policy NAME-OR-PATH` | The policy Flow certifies against, such as `clinical-write` or `permissive`. |
 | `--timeout SECONDS` | How long one run may work (default 600). Time a person spends approving a save doesn't count. A run stopped at the limit is `not_sure_if_saved`. |
-| `--approval-mode MODE` | Whether a person approves each save. Without it, Flow decides: under `standard` and `regulated` a run stops before each save until a person approves it (`each_item`). `not_gated` turns that off for every run this server starts; each result records it under `technical.approval_mode`. |
+| `--approval-mode MODE` | Whether each save waits for approval. Without it, Flow decides: under `standard` and `regulated` a run stops before each save until it's approved (`each_item`). `not_gated` turns that off for every run this server starts; each result records it under `technical.approval_mode`. |
+| `--delegate-approvals AGENT_ID` | Lets your agent answer a waiting save as `AGENT_ID` (an opaque id such as `agent:intake-bot`), with `list_approval_requests` and `answer_approval_request`. Flow accepts the answer only where the run's approver policy allows an agent for that step, and records the agent as the approver. By default only a person may answer. Needs an openadapt-flow with delegated approvers (not released yet). |
 | `--headed` | Shows the browser. A person needs it to continue or skip a stopped run in a web app. |
 | `--extra-run-arg ARG`, `--flow-cli CMD` | Change how Flow is called. Either one can relax Flow's checks, for example `--extra-run-arg=--allow-unencrypted`. |
 | `--allow-protected-export`, `--allow-synthetic-recorded-defaults`, `--allow-model-grounding` | Off by default. See [Security](#security). |
@@ -409,7 +410,7 @@ openadapt-agent seal verify https://example.com/receipt.json --json
 - The check that stops a second entry covers one runs folder. Pass the same absolute `--runs-dir` every time.
 - If the server exits during a run, the run reads as `not_sure_if_saved`, and Flow may still finish it.
 - This server issues no signed receipts itself. `proof` is `sealed` only when a run is connected to OpenAdapt Cloud Execute, Cloud signed its receipt, and that signature verifies against keys you pinned (see [Seals](#seals)).
-- With openadapt-flow 2, every save waits for a person's approval unless the operator names another mode (`serve --approval-mode problems_only` or `not_gated`). Your agent gets `needs_review` with the reason `waiting_for_approval`, and a person approves on the computer that runs OpenAdapt. Nothing is saved without that answer.
+- With openadapt-flow 2, each save waits for approval unless the operator names another mode (`serve --approval-mode problems_only` or `not_gated`). Your agent gets `needs_review` with the reason `waiting_for_approval`. A person answers on the computer that runs OpenAdapt, or your agent does when the server runs with `--delegate-approvals` and the run's approver policy allows an agent for that step. Nothing is saved without that answer. `get_approval_mode` tells your agent which mode applies; no tool changes it.
 - Each call carries one request. There's no batch call.
 - Accounts, separation between customers, fleet policy, and hosted runs belong to OpenAdapt Cloud.
 

packages/agent/tests/test_delegated_approval.py

@@ -0,0 +1,157 @@
+"""Delegated approval tools. Made-up values only.
+
+The agent can always read the approval mode. It can answer a waiting save
+only when the operator started the server with --delegate-approvals, and
+then only through Flow's item-approval command as decided_by automation
+from the mcp route, naming the agent. Nothing changes the mode, and no
+tool admits a check.
+"""
+
+from __future__ import annotations
+
+import dataclasses
+
+import pytest
+
+from openadapt_agent.bridge import AgentBridge, BridgeError
+from openadapt_agent.runner import FlowRunner
+
+REQUEST_ID = "a" * 32
+DIGEST = "sha256:" + "b" * 64
+
+
+class _Runner(FlowRunner):
+    def __init__(self, config, reply):
+        super().__init__(config)
+        self.reply = reply
+        self.calls: list[tuple[str, str, list[str]]] = []
+
+    def item_approval(self, verb, run_dir, args):
+        self.calls.append((verb, str(run_dir), list(args)))
+        return dict(self.reply)
+
+
+def _bridge(bundles_root, runner_config, *, delegate=None, reply=None):
+    config = dataclasses.replace(runner_config, delegate_approver_id=delegate)
+    runner = _Runner(config, reply or {"ok": True})
+    bridge = AgentBridge(bundles_root, config, allow_run=True, runner=runner)
+    return bridge, runner, config
+
+
+def _names(bridge):
+    return {spec.name for spec in bridge.list_tool_specs()}
+
+
+def test_mode_is_readable_and_answer_tools_are_off_by_default(bundles_root, runner_config):
+    bridge, runner, _ = _bridge(bundles_root, runner_config)
+    names = _names(bridge)
+    assert "get_approval_mode" in names
+    assert not names & {"list_approval_requests", "answer_approval_request"}
+    view = bridge.dispatch("get_approval_mode", {})
+    assert view == {
+        "approval_mode": "flow_default",
+        "delegated_approvals": False,
+        "delegate_id": None,
+    }
+    with pytest.raises(BridgeError):
+        bridge.dispatch(
+            "answer_approval_request",
+            {"run_id": "r1", "request_id": REQUEST_ID, "request_sha256": DIGEST, "decision": "approve"},
+        )
+    assert runner.calls == []
+
+
+def test_no_tool_changes_the_mode_or_admits(bundles_root, runner_config):
+    bridge, _, _ = _bridge(bundles_root, runner_config, delegate="agent:intake-bot")
+    for name in _names(bridge):
+        assert not name.startswith("set_approval")
+        assert "admit" not in name
+
+
+def test_an_answer_goes_to_flow_as_this_agent(bundles_root, runner_config):
+    bridge, runner, config = _bridge(
+        bundles_root,
+        runner_config,
+        delegate="agent:intake-bot",
+        reply={"ok": True, "request_id": REQUEST_ID, "decision": "approve", "answer_sha256": "x"},
+    )
+    (config.runs_dir / "r1").mkdir(parents=True)
+    result = bridge.dispatch(
+        "answer_approval_request",
+        {"run_id": "r1", "request_id": REQUEST_ID, "request_sha256": DIGEST, "decision": "approve"},
+    )
+    assert result["ok"] is True
+    verb, _run_dir, args = runner.calls[-1]
+    assert verb == "approve"
+    pairs = dict(zip(args[::2], args[1::2]))
+    assert pairs["--decided-by"] == "automation"
+    assert pairs["--route"] == "mcp"
+    assert pairs["--delegate-kind"] == "agent"
+    assert pairs["--delegate-id"] == "agent:intake-bot"
+
+
+def test_flow_refusals_come_back_as_is(bundles_root, runner_config):
+    bridge, _, config = _bridge(
+        bundles_root,
+        runner_config,
+        delegate="agent:intake-bot",
+        reply={"ok": False, "code": "approver_kind_not_allowed", "message": "m"},
+    )
+    (config.runs_dir / "r1").mkdir(parents=True)
+    result = bridge.dispatch(
+        "answer_approval_request",
+        {
+            "run_id": "r1",
+            "request_id": REQUEST_ID,
+            "request_sha256": DIGEST,
+            "decision": "decline",
+            "decline_reason": "wrong_values",
+        },
+    )
+    assert result == {"ok": False, "code": "approver_kind_not_allowed", "message": "m"}
+
+
+@pytest.mark.parametrize(
+    "arguments",
+    [
+        {"run_id": "r1", "request_id": "nope", "request_sha256": DIGEST, "decision": "approve"},
+        {"run_id": "r1", "request_id": REQUEST_ID, "request_sha256": "x", "decision": "approve"},
+        {"run_id": "r1", "request_id": REQUEST_ID, "request_sha256": DIGEST, "decision": "decline"},
+        {"run_id": "../x", "request_id": REQUEST_ID, "request_sha256": DIGEST, "decision": "approve"},
+    ],
+)
+def test_bad_answers_never_reach_flow(bundles_root, runner_config, arguments):
+    bridge, runner, config = _bridge(bundles_root, runner_config, delegate="agent:intake-bot")
+    (config.runs_dir / "r1").mkdir(parents=True)
+    with pytest.raises(BridgeError):
+        bridge.dispatch("answer_approval_request", arguments)
+    assert runner.calls == []
+
+
+def test_request_list_drops_everything_but_the_allow_list(bundles_root, runner_config):
+    bridge, _, config = _bridge(
+        bundles_root,
+        runner_config,
+        delegate="agent:intake-bot",
+        reply={
+            "schema": "openadapt.item-approval-list/v1",
+            "run_dir": "/private/path",
+            "requests": [
+                {
+                    "request_id": REQUEST_ID,
+                    "request_sha256": DIGEST,
+                    "status": "waiting",
+                    "step_id": "save",
+                    "step_class": "consequential",
+                    "step_intent": "Type Zyxw Qopn's DOB",
+                    "screenshot": "/private/path/shot.png",
+                }
+            ],
+        },
+    )
+    (config.runs_dir / "r1").mkdir(parents=True)
+    result = bridge.dispatch("list_approval_requests", {"run_id": "r1"})
+    assert result["ok"] is True
+    (row,) = result["requests"]
+    assert set(row) == {"request_id", "request_sha256", "status", "step_id", "step_class"}
+    assert "private" not in str(result)

packages/flow/openadapt_flow/agent/bridge.py

@@ -37,6 +37,7 @@
 from pathlib import Path
 from typing import Any, Optional
 
+from openadapt_flow.agent import delegated
 from openadapt_flow.agent.approval import report_approval
 from openadapt_flow.agent.attended import (
     ATTENDED_TOOLS,
@@ -351,8 +352,94 @@ def _contract_specs(self) -> list[ToolSpec]:
             )
         )
         specs.extend(self._repair_specs())
+        specs.extend(self._approval_specs())
         return specs
 
+    def _delegate_id(self) -> Optional[str]:
+        if not self.allow_run or self.sandbox is not None:
+            return None
+        return self.runner_config.delegate_approver_id
+
+    def _approval_specs(self) -> list[ToolSpec]:
+        """``get_approval_mode`` always; the answer tools only when the
+        operator started the server with ``--delegate-approvals``."""
+        if not self.allow_run or self.sandbox is not None:
+            return []
+        specs = [
+            ToolSpec(
+                name="get_approval_mode",
+                description=delegated.GET_MODE_DESCRIPTION,
+                input_schema={
+                    "type": "object",
+                    "properties": {},
+                    "additionalProperties": False,
+                },
+                annotations=_READ_ONLY_ANNOTATIONS,
+            )
+        ]
+        if self._delegate_id() is None:
+            return specs
+        list_schema = delegated.list_schema()
+        answer_schema = delegated.answer_schema()
+        for schema in (list_schema, answer_schema):
+            schema["properties"]["run_id"]["pattern"] = _RUN_ID_PATTERN
+        specs += [
+            ToolSpec(
+                name="list_approval_requests",
+                description=delegated.LIST_DESCRIPTION,
+                input_schema=list_schema,
+                annotations=_READ_ONLY_ANNOTATIONS,
+            ),
+            ToolSpec(
+                name="answer_approval_request",
+                description=delegated.ANSWER_DESCRIPTION,
+                input_schema=answer_schema,
+                annotations={
+                    "readOnlyHint": False,
+                    "destructiveHint": True,
+                    "idempotentHint": True,
+                    "openWorldHint": True,
+                },
+            ),
+        ]
+        return specs
+
+    def _safe_run_dir(self, run_id: str) -> Path:
+        if not run_id or not is_safe_run_id(run_id):
+            raise BridgeError("run_id must be a single path component")
+        runs_candidate = Path(self.runner_config.runs_dir)
+        run_candidate = runs_candidate / run_id
+        if runs_candidate.is_symlink() or run_candidate.is_symlink():
+            raise BridgeError("the configured run evidence boundary is unavailable")
+        run_dir = run_candidate.resolve()
+        if runs_candidate.resolve() not in run_dir.parents or not run_dir.is_dir():
+            raise BridgeError("no such run in the server's runs directory")
+        return run_dir
+
+    def _approval_dispatch(self, name: str, arguments: dict) -> dict:
+        if name == "get_approval_mode":
+            return delegated.mode_view(
+                self.runner_config.approval_mode, self._delegate_id()
+            )
+        delegate_id = self._delegate_id()
+        if delegate_id is None:
+            raise BridgeError(
+                "this agent can't answer approvals: the operator didn't start the "
+                "server with --delegate-approvals"
+            )
+        run_dir = self._safe_run_dir(str(arguments.get("run_id", "")))
+        if name == "list_approval_requests":
+            return delegated.project_requests(
+                self.runner.item_approval("show", run_dir, [])
+            )
+        try:
+            verb, args = delegated.answer_args(arguments, delegate_id)
+        except delegated.DelegationRefused as exc:
+            raise BridgeError(str(exc)) from exc
+        result = self.runner.item_approval(verb, run_dir, args)
+        allowed = ("ok", "code", "message", "request_id", "decision", "answer_sha256")
+        return {k: result[k] for k in allowed if k in result}
+
     def _repair_tools_enabled(self) -> bool:
         from openadapt_flow.agent.repair import teach_supported
 
@@ -592,6 +679,14 @@ def dispatch(self, name: str, arguments: Optional[dict]) -> dict:
                 )
             except ServiceError as exc:
                 raise BridgeError(str(exc)) from exc
+        if name in (
+            "get_approval_mode",
+            "list_approval_requests",
+            "answer_approval_request",
+        ):
+            if not any(spec.name == name for spec in self._approval_specs()):
+                raise BridgeError("unknown tool name")
+            return self._approval_dispatch(name, arguments)
         if name in ("get_repair_screen", "teach_repair"):
             if not self._repair_tools_enabled():
                 raise BridgeError("repair tools are off on this server")

packages/flow/openadapt_flow/agent/cli.py

@@ -236,12 +236,26 @@ def build_parser() -> argparse.ArgumentParser:
         help=(
             "Per-save approval passed to every `openadapt-flow run`. Unset "
             "(the default): Flow decides, and under the Standard and Regulated "
-            "profiles a person approves each save (each_item); a run waiting "
+            "profiles each save waits for approval (each_item); a run waiting "
             "for that approval reports needs_review. not_gated turns the pause "
             "off for every run this server starts, and each run's result and "
             "Flow's report record that. Callers can't change it."
         ),
     )
+    p.add_argument(
+        "--delegate-approvals",
+        default=None,
+        metavar="AGENT_ID",
+        help=(
+            "Let the calling agent answer saves that wait for an approval, as "
+            "the opaque id AGENT_ID (for example agent:intake-bot). Adds the "
+            "list_approval_requests and answer_approval_request tools. Flow "
+            "accepts the agent's answer only where the run's approver policy "
+            "(openadapt-flow run --approver-policy) allows an agent for that "
+            "step; by default only a person may answer. Flow records the agent "
+            "as the approver."
+        ),
+    )
     p.add_argument(
         "--allow-url-override",
         action="store_true",
@@ -538,6 +552,25 @@ def _cmd_serve(args: argparse.Namespace) -> int:
                 file=sys.stderr,
             )
             return 2
+    if args.delegate_approvals is not None:
+        import re as _re
+
+        from openadapt_flow.agent.delegated import DELEGATE_ID_PATTERN
+
+        if mode not in {"production", "attended"}:
+            print(
+                "serve: --delegate-approvals needs --mode production or --mode "
+                "attended with --bundles",
+                file=sys.stderr,
+            )
+            return 2
+        if not _re.match(DELEGATE_ID_PATTERN, args.delegate_approvals):
+            print(
+                "serve: --delegate-approvals needs an opaque id such as "
+                "agent:intake-bot, never a name or email",
+                file=sys.stderr,
+            )
+            return 2
     if args.allow_synthetic_recorded_defaults and not (args.allow_run and args.bundles):
         print(
             "serve: --allow-synthetic-recorded-defaults requires --allow-run",
@@ -664,6 +697,7 @@ def _cmd_serve(args: argparse.Namespace) -> int:
             allow_url_override=args.allow_url_override,
             extra_run_args=tuple(extra_run_args),
             approval_mode=args.approval_mode,
+            delegate_approver_id=args.delegate_approvals,
             hold_on_halt_s=(
                 max(0.0, min(args.hold_on_halt, 1800.0)) if args.repair_access else 0.0
             ),
@@ -702,7 +736,7 @@ def _cmd_serve(args: argparse.Namespace) -> int:
                 + (
                     f"approval mode {args.approval_mode} (chosen with --approval-mode); "
                     if args.approval_mode
-                    else "approval mode set by Flow (a person approves each save "
+                    else "approval mode set by Flow (each save waits for approval "
                     "under the Standard and Regulated profiles); "
                 )
                 + f"runs folder {runs_dir}",