From 4831142585aa872ea5c506aed3c3cce750ca32e5 Mon Sep 17 00:00:00 2001 From: Asaf Varon Date: Thu, 22 Oct 2026 19:53:41 +0300 Subject: [PATCH 1/3] fix(engine): enforce access checks before protected root execution Apply the existing access policy before protected root agents or orchestrators resolve prompts, call models, or execute tools. Fail closed when the access plugin is missing, emit refusals through the stream sink, and recheck access during tool-approval replay. Add HTTP and engine regressions plus access-policy documentation. Fixes #151 --- docs/SIDECAR_CONTEXT_AUTH.md | 36 ++- src/agent_engine/engine/langgraph/engine.py | 4 +- .../engine/langgraph/graph/graph_builder.py | 29 ++- tasks/README.md | 1 + tasks/issue-0151-protected-root-access.md | 171 ++++++++++++++ tests/api/test_protected_access_e2e.py | 133 +++++++++++ tests/engine/test_protected_root.py | 211 ++++++++++++++++++ 7 files changed, 575 insertions(+), 10 deletions(-) create mode 100644 tasks/issue-0151-protected-root-access.md create mode 100644 tests/api/test_protected_access_e2e.py create mode 100644 tests/engine/test_protected_root.py diff --git a/docs/SIDECAR_CONTEXT_AUTH.md b/docs/SIDECAR_CONTEXT_AUTH.md index 5f96ea63..ca1379a0 100644 --- a/docs/SIDECAR_CONTEXT_AUTH.md +++ b/docs/SIDECAR_CONTEXT_AUTH.md @@ -175,8 +175,10 @@ missing variables in strict prompt rendering produce clear errors. ## Access Plugin -Authorization is opt-in per node. A node marked `protected: true` is hidden from -the router unless the fixed access plugin allows it. +Authorization is opt-in per node. A child marked `protected: true` is hidden +from its parent unless the fixed access plugin allows it. A protected root +agent or orchestrator passes through the same policy before its executor is +entered, since it has no parent to filter it. ```yaml agents: @@ -202,10 +204,38 @@ Rules: - Protected nodes are checked before routing. - If `can_access` returns true, the node remains a routing candidate. - If `can_access` returns false or raises, the node is hidden. +- On a new run of a protected root, denial returns `Access denied.` with + an empty visited route and no tool usage. Prompt resolvers, models, and tools + are not invoked. `/invoke` keeps its normal response shape and HTTP 200; + `/stream` emits the fixed refusal as an `answer_delta`, then a `final` event + containing the same refusal and empty route. Delta-only clients such as the + CLI therefore display the refusal too. + No model-generated answer or route event is emitted for a denied root. - `protected: true` without `plugins/access.py` is a configuration error at - startup. + startup, including when a Python host builds the engine directly. - `plugins/access.py` with no protected nodes is harmless and may be ignored. +The access resolver receives the per-request, non-secret `run_context` +projection for roots and children alike: available run, conversation, user and +organization IDs, metadata, and `auth` facts such as roles and scopes. The +inbound access token is excluded. A Python host supplies trusted `RunContext` +values; `agent_manager` derives them from its verified principal. The thin +engine API accepts no caller-asserted roles or user identity, including forged +`X-User-ID` or `X-Roles` headers. + +No universal role requirement is imposed by the engine. If the policy requires +identity or metadata that is missing or malformed, it must deny or raise; both +outcomes fail closed. An explicitly permissive policy can allow an empty +context. A root's decision is evaluated per execution, never cached between +requests. This enforces node access without implementing API authentication or +replacing authorization inside client tools and data services. + +The root policy is also rechecked when a pending tool approval replays the root +executor. If that policy now denies access, the reviewed invocation is never +reached. The existing replay-binding guard raises `ApprovalInvocationMismatch`, +rejects the pending approval and fails the run; the human decision is not +reported as an executed action. This differs from a new run's normal refusal. + --- ## Why Not YAML Roles? diff --git a/src/agent_engine/engine/langgraph/engine.py b/src/agent_engine/engine/langgraph/engine.py index 015492a0..5b5e625d 100644 --- a/src/agent_engine/engine/langgraph/engine.py +++ b/src/agent_engine/engine/langgraph/engine.py @@ -997,9 +997,7 @@ async def _stream_graph( def _setup_filters(self, spec: SystemSpec) -> list[RouteFilter]: filters: list[RouteFilter] = [] if has_protected_nodes(spec.graph): - access_plugin = self._base_dir / "plugins" / "access.py" - if access_plugin.is_file(): - filters.append(AccessFilter(self._base_dir)) + filters.append(AccessFilter(self._base_dir)) return filters def _build_graph(self, spec: SystemSpec) -> RunGraph: diff --git a/src/agent_engine/engine/langgraph/graph/graph_builder.py b/src/agent_engine/engine/langgraph/graph/graph_builder.py index 78bef498..d9019a11 100644 --- a/src/agent_engine/engine/langgraph/graph/graph_builder.py +++ b/src/agent_engine/engine/langgraph/graph/graph_builder.py @@ -3,6 +3,7 @@ from __future__ import annotations import inspect +import logging from collections.abc import Callable, Mapping, Sequence from pathlib import Path @@ -17,6 +18,7 @@ from agent_engine.core.spec import AgentSpec, BaseModelConfig, GraphNode, OrchestratorSpec from agent_engine.core.spec import ModelConfig as NodeModelConfig from agent_engine.engine.langgraph.checkpointing import CheckpointerHandle +from agent_engine.engine.langgraph.execution.node_executor import NodeExecutor from agent_engine.engine.langgraph.filters import RouteFilter from agent_engine.engine.langgraph.graph.traversal import node_id from agent_engine.engine.langgraph.nodes.agent_node import AgentNode @@ -28,11 +30,13 @@ from agent_engine.loaders.tool_loader import ToolLoader from agent_engine.runtime.hooks import HookManager from agent_engine.runtime.state import GraphState +from agent_engine.runtime.streaming import current_streams from agent_engine.tool_usage.context_provider import ToolUsageContextProvider from agent_engine.tool_usage.tracker import ToolUsageTracker ModelFactory = Callable[..., BaseChatModel] _MODEL_FACTORY_OPTIONAL_KWARGS = ("region", "max_tokens", "top_p") +logger = logging.getLogger(__name__) # noinspection PyTypeChecker RunGraphBuilder = StateGraph[GraphState, None, GraphState, GraphState] @@ -141,13 +145,30 @@ def _wire_node( parent_path: str | None, ) -> None: path = node_id(node, parent_path) + executor: NodeExecutor if isinstance(node.node, OrchestratorSpec): - orchestrator_node = self._build_orchestrator_node(node, parent_path) - builder.add_node(path, orchestrator_node.execute) + executor = self._build_orchestrator_node(node, parent_path) else: assert isinstance(node.node, AgentSpec) - agent_node = self._build_agent_node(node.node, path) - builder.add_node(path, agent_node.execute) + executor = self._build_agent_node(node.node, path) + filters = tuple(self._filters) + + async def execute_root(state: GraphState) -> GraphState: + # The root has no parent to filter it before delegation. Apply the + # same policy before entering either kind of protected executor. + candidates = [node.node] + for route_filter in filters: + candidates = route_filter.filter(state.get("run_context", {}), candidates) + if not candidates: + logger.warning("access denied for protected root node '%s'", path) + answer = "Access denied." + sink = current_streams.get().answer + if sink is not None: + sink(answer) + return {"visited": [], "answer": answer} + return await executor.execute(state) + + builder.add_node(path, execute_root if node.node.protected else executor.execute) builder.add_edge(path, END) def _build_orchestrator_node( diff --git a/tasks/README.md b/tasks/README.md index aeb30db0..3bb88dcc 100644 --- a/tasks/README.md +++ b/tasks/README.md @@ -22,5 +22,6 @@ finishing, Expected final report.** | 0010 | [Docker deployment](0010-docker-deployment.md) | 0009 | | 0011 | [Observability & tracing](0011-observability-and-tracing.md) | 0004 | | 0012 | [Tests & quality gates](0012-tests-and-quality-gates.md) | all | +| Issue 151 | [Protected root access](issue-0151-protected-root-access.md) | 0006 security follow-up | Before starting any task, read `AGENTS.md` and the skill(s) the task lists. diff --git a/tasks/issue-0151-protected-root-access.md b/tasks/issue-0151-protected-root-access.md new file mode 100644 index 00000000..fe7c8054 --- /dev/null +++ b/tasks/issue-0151-protected-root-access.md @@ -0,0 +1,171 @@ +# Issue 151 — Enforce access before protected root execution + +## Goal / authorization + +Fix [issue 151](https://github.com/extra-org/extra/issues/151), as requested by +the user. Apply the existing access resolver policy to a protected root agent +or orchestrator before any prompt resolution, model invocation, or tool work. + +## Context / read first + +AGENTS.md, .ai/README.md; project-architecture, senior-python-engineering, +testing, runtime-engine, sidecar-auth-context, architecture-review, +documentation and git-workflow; task 0006 and docs/SIDECAR_CONTEXT_AUTH.md. +Use systematic debugging, test-driven development and verification before +completion. The issue's HTTP reproduction distinguishes root bypass from +already-working protected-child filtering. + +## Scope / files allowed to change + +- src/agent_engine/engine/langgraph/graph/graph_builder.py: gate the root using + the existing request-scoped route filters before its executor is entered. +- src/agent_engine/engine/langgraph/engine.py: require the configured access + plugin when building a graph with protected nodes, including Python callers. +- tests/api/test_protected_access_e2e.py: real application regressions using a + fake model, covering root/child allow/deny and forged identity headers. +- tests/engine/test_protected_root.py: run/stream, agent/orchestrator roots, + missing/invalid policy context, resolver failures, missing access plugin, + streaming-console visibility and pending tool-approval replay. +- docs/SIDECAR_CONTEXT_AUTH.md: document root denial and identity semantics. +- This task and tasks/README.md. + +## Requirements / acceptance criteria + +- [x] Deny-all policy prevents protected root agent and orchestrator execution. +- [x] Permitted callers can execute protected roots using the same resolver + and non-secret context projection as protected children. +- [x] Missing/invalid required context and resolver errors fail closed. +- [x] Denied roots perform no prompt, model or tool work and expose no route. +- [x] Denial is a fixed refusal using existing result/SSE shapes; child + filtering and unprotected roots retain their existing behavior. +- [x] Missing access plugin is a clear startup error for protected graphs. +- [x] HTTP and engine regressions fail before the fix and pass afterwards. +- [x] Documentation reflects the implemented flow; make check passes. + +## Out of scope + +API authentication redesign, new identity claims or role rules, WhatsApp +changes, new HTTP/SSE contracts, configuration rejection of protected roots, +new persistence or dependencies, unrelated documentation cleanup, and merging. + +## Commands / expected final report + +Run the new regressions before implementation, the relevant engine/API tests, +make check, and git diff --check. Report using AGENTS.md section 9, including +test counts, the denial response, and the limits of the existing identity +boundary. Preserve the user's original checkout and work. + +## Implementation / validation result + +The compiled root entry applies the existing route filters before calling its +agent or orchestrator executor. Denial returns `Access denied.` and an empty +route using the existing result/SSE shapes. The wrapper captures startup +collaborators and receives request state as an argument; no request decision is +cached. Access plugin loading now fails at startup for a protected graph even +when a Python host bypasses the API's configuration validator. + +The initial regression run had 17 expected failures reproducing root execution +and missing-plugin acceptance, with nine existing allow/child/public behaviors +passing. Four additional prompt-plugin and per-request decision cases failed +when the root gate was temporarily bypassed, then passed with it restored. +The initial fix added 30 cases in the two test files. Its make check gate passed: +format/lint, mypy (293 files), 1,085 tests / 9 dependency warnings, and generated +stub checks. The first gate attempt caught import ordering and an unescaped +test regex; both were corrected before the passing run. All external models +are faked; storage uses temporary local paths. The final make check rerun also +passed after the wrapper's captured filters were made immutable; make validate +passed for both configured examples. git diff --check and new-file whitespace +checks passed. + +The fix preserves YAML, plugin signatures, HTTP schemas and SSE event types; +it corrects enforcement of the existing protected-node contract, so no new ADR +or migration is required. General API authentication and older project-status +documentation remain separate concerns. No commit, PR, merge or deployment +was performed during the initial implementation. + +## Post-implementation code review + +The user requested this review after the initial fix. Review date: 2026-10-22. + +### 1. Summary + +Reviewed root/child policy semantics, request isolation, startup failure, +stream consumers, lifecycle, and checkpointed approval replay. One display +regression was confirmed and corrected; no open blocking finding remains in +this scope. + +### 2. Blocking issues + +**[P2, fixed] Streaming clients could display a blank refusal.** Returning +only a final event skipped the answer sink. `agentctl run --stream` and the +local/remote streaming chat consumers render answer deltas, so the denial was +invisible. A regression using the real local chat consumer reproduced only +`Agent > ` without the refusal; both HTTP streaming cases also lacked the +expected delta. All three cases failed before correction. The denied-root +branch now emits its fixed, non-model-generated answer through the existing +request-scoped answer sink, followed by the usual final event. + +### 3. Non-blocking issues + +No additional actionable defect was confirmed in the scoped changes. Broader +API authentication and legacy project-status prose remain outside this task. + +### 4. Architecture concerns + +The guard runs before either protected root executor. It reuses the existing +access filter and keeps request decisions in invocation state; only immutable +startup filter references are captured by the compiled wrapper. The correction +uses an existing stream sink and event type. No new YAML/plugin/HTTP schema, +engine request state, persistence, dependency or ADR is introduced. + +### 5. Security concerns + +Denied new runs expose no visited path and execute no prompt plugin, model or +tool. Missing/invalid policy context and resolver failures remain fail closed. +Permitted and denied requests on one engine do not cache grants between calls. +Eight additional approval-replay cases cover root agents/orchestrators and +normal/streamed resumes: unchanged policy permits one real local tool effect; +policy revocation prevents that effect and the existing invocation-binding +guard rejects the decision, fails the run and leaves no pending approval or +recoverable success. This tests resolver policy rechecks, not a redesign of +identity propagation or the host's authentication boundary. + +### 6. Testing gaps + +The review added nine cases, bringing the two new files to 39 cases. The +streaming console regression uses the real consumer and engine; replay tests +observe a real temporary-file tool effect. Models are deterministic fakes. +Disabling the root gate made the replay-revocation regressions fail, confirming +they exercise enforcement. No confirmed coverage gap remains for this scope. + +The final make check passed: format/lint, mypy (293 files), **1,094 tests / +9 dependency warnings**, and generated-stub checks. make validate passed for +both configured examples; git diff --check passed. No warnings were suppressed. +One intervening full gate caught a stale source snapshot restored by the +mutation probe: the two HTTP streaming tests and console regression failed. +The stream correction was restored; the passing final run also verified its +source hash stayed unchanged throughout the gate. + +### 7. Suggested improvements + +No further change is required before reviewing this scoped fix for a PR. +Keep general authentication and outdated overview status documentation in +separate work. + +### 8. Final recommendation + +**Approve** the scoped fix with the streaming correction and added replay +coverage. This is a local self-review recommendation, not a GitHub approval. +No commit, push, PR, merge or deployment was performed during that review. + +## PR preparation + +On 2026-10-22, the user requested a pull request for the reviewed fix. This +authorizes committing and pushing the scoped change and opening a PR against +`main`. The PR includes the implementation, 39 regression cases, access-policy +documentation and this task record. Merging and deployment remain separate. + +Fresh pre-commit validation passed: make check (1,094 tests, 9 dependency +warnings, format/lint, mypy and generated-stub checks), make validate (both +configured examples), and git diff --check. The reviewed source hash was +unchanged throughout the gate. diff --git a/tests/api/test_protected_access_e2e.py b/tests/api/test_protected_access_e2e.py new file mode 100644 index 00000000..4f618b21 --- /dev/null +++ b/tests/api/test_protected_access_e2e.py @@ -0,0 +1,133 @@ +"""Issue 151: exercise protected root and child access through the real HTTP app.""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +import pytest +from fastapi.testclient import TestClient +from tests.fixtures.utils import FakeChatModel, fake_model_factory + +from agent_engine.api.app import create_app +from agent_engine.engine.langgraph.engine import LangGraphEngine + +_FORGED_HEADERS = { + "X-User-ID": "admin", + "X-Roles": "admin", + "Authorization": "Bearer forged-admin-token", +} +_MODEL = " model: {provider: openai, name: test-model}\n" + + +def _client( + base_dir: Path, + monkeypatch: pytest.MonkeyPatch, + *, + root_kind: str = "agent", + protected_root: bool = True, + allow: bool = False, +) -> TestClient: + plugins = base_dir / "plugins" + plugins.mkdir() + (plugins / "access.py").write_text( + "class AccessResolver:\n" + " def can_access(self, ctx, node_id):\n" + f" return {allow!r}\n", + encoding="utf-8", + ) + if root_kind == "agent": + declarations = ( + "agents:\n admin:\n description: Administrative agent.\n" + f" protected: {str(protected_root).lower()}\n{_MODEL}" + ) + graph = "graph:\n admin:\n" + else: + (base_dir / "route.md").write_text("Delegate to admin.", encoding="utf-8") + declarations = ( + "orchestrators:\n root:\n description: Routes requests.\n" + f" protected: {str(protected_root).lower()}\n" + f" prompts: {{orchestrator: route.md}}\n{_MODEL}" + f"agents:\n admin:\n description: Administrative agent.\n{_MODEL}" + " protected: true\n" + ) + graph = "graph:\n root:\n admin:\n" + config = base_dir / "agents.yaml" + config.write_text(f"system: {{name: Access Probe}}\n{declarations}{graph}", encoding="utf-8") + # Keep startup, validation, graph compilation and request execution real; + # replace only model construction at the engine's existing injection seam. + monkeypatch.setattr( + "agent_engine.api.app.LangGraphEngine", + lambda *args, **kwargs: LangGraphEngine(*args, model_factory=fake_model_factory, **kwargs), + ) + return TestClient(create_app(str(config))) + + +@pytest.fixture +def model_calls(monkeypatch: pytest.MonkeyPatch) -> list[list[Any]]: + calls: list[list[Any]] = [] + respond = FakeChatModel._respond + + def record(model: FakeChatModel, messages: list[Any]): + calls.append(messages) + return respond(model, messages) + + monkeypatch.setattr(FakeChatModel, "_respond", record) + return calls + + +@pytest.mark.parametrize("root_kind", ["agent", "orchestrator"]) +@pytest.mark.parametrize("endpoint", ["/invoke", "/stream"]) +def test_denied_protected_root_never_executes_over_http( + tmp_path, monkeypatch, model_calls, root_kind, endpoint +): + with _client(tmp_path, monkeypatch, root_kind=root_kind) as client: + response = client.post(endpoint, json={"message": "admin"}, headers=_FORGED_HEADERS) + + assert response.status_code == 200 + if endpoint == "/invoke": + body = response.json() + assert body["visited"] == [] + assert body["answer"] == "Access denied." + assert body["used_tools"] == [] + else: + events = [ + json.loads(line[6:]) for line in response.text.splitlines() if line.startswith("data: ") + ] + assert [event["type"] for event in events] == ["answer_delta", "final"] + assert events[0]["content"] == "Access denied." + assert events[1]["route"] == [] + assert events[1]["content"] == "Access denied." + assert model_calls == [] + + +@pytest.mark.parametrize("root_kind", ["agent", "orchestrator"]) +def test_protected_root_executes_when_policy_allows(tmp_path, monkeypatch, root_kind): + with _client(tmp_path, monkeypatch, root_kind=root_kind, allow=True) as client: + response = client.post("/invoke", json={"message": "admin"}) + + assert response.status_code == 200 + assert response.json()["visited"] == ( + ["admin"] if root_kind == "agent" else ["root", "root/admin"] + ) + + +@pytest.mark.parametrize("allow", [False, True]) +def test_protected_child_still_obeys_policy_with_forged_headers(tmp_path, monkeypatch, allow): + with _client( + tmp_path, monkeypatch, root_kind="orchestrator", protected_root=False, allow=allow + ) as client: + response = client.post("/invoke", json={"message": "admin"}, headers=_FORGED_HEADERS) + + assert response.status_code == 200 + assert response.json()["visited"] == (["root", "root/admin"] if allow else ["root"]) + + +def test_unprotected_root_still_executes_with_deny_all_policy(tmp_path, monkeypatch): + with _client(tmp_path, monkeypatch, protected_root=False) as client: + response = client.post("/invoke", json={"message": "hello"}) + + assert response.status_code == 200 + assert response.json()["visited"] == ["admin"] + assert response.json()["answer"] == "ok" diff --git a/tests/engine/test_protected_root.py b/tests/engine/test_protected_root.py new file mode 100644 index 00000000..7f519c75 --- /dev/null +++ b/tests/engine/test_protected_root.py @@ -0,0 +1,211 @@ +"""Protected roots use the existing access policy before entering their executor.""" + +from __future__ import annotations + +from dataclasses import replace +from pathlib import Path + +import pytest + +from agent_engine.approvals.errors import ApprovalInvocationMismatch +from agent_engine.approvals.models import RunStatus +from agent_engine.core.spec import ResolverSpec +from agent_engine.engine.langgraph.engine import LangGraphEngine +from agent_engine.runtime.hooks import AuthContext, RunContext +from tests.approvals.test_engine_hitl import _multi_tool_spec +from tests.approvals.test_resume_binding import write_tool +from tests.engine.test_engine_flow import ( + agent, + orchestrator, + system, + write_access, + write_context_access, + write_raising_access, +) +from tests.fixtures.utils import fake_model_factory + + +def _protected_root(kind): + root = agent("admin", protected=True) + if kind == "orchestrator": + root = orchestrator("admin", [agent("public")]) + root = replace(root, node=replace(root.node, protected=True)) + return system(root) + + +@pytest.mark.parametrize("kind", ["agent", "orchestrator"]) +@pytest.mark.parametrize("streaming", [False, True]) +@pytest.mark.parametrize("context_kind", ["missing", "invalid", "allowed"]) +async def test_root_policy_uses_request_context(tmp_path, kind, streaming, context_kind): + write_context_access(tmp_path) + context = None + if context_kind != "missing": + context = RunContext( + metadata={"allowed_nodes": ("admin",) if context_kind == "allowed" else None}, + auth_context=AuthContext( + inbound_access_token="test-only-secret", metadata={"groups": ("docs",)} + ), + ) + async with LangGraphEngine(tmp_path, model_factory=fake_model_factory) as engine: + await engine.build(_protected_root(kind)) + if streaming: + events = [event async for event in engine.stream("hello", context=context)] + result = events[-1] + visited, answer = list(result.route or ()), result.content + else: + response = await engine.run("hello", context=context) + visited, answer = response.visited, response.answer + + if context_kind == "allowed": + assert visited == (["admin"] if kind == "agent" else ["admin", "admin/public"]) + assert answer == "ok" + else: + assert visited == [] + assert answer == "Access denied." + + +@pytest.mark.parametrize("kind", ["agent", "orchestrator"]) +@pytest.mark.parametrize("exc", ["RuntimeError('private-policy-error')", "NotImplementedError"]) +async def test_root_denied_when_access_resolver_raises(tmp_path: Path, kind, exc): + write_raising_access(tmp_path, exc=exc) + async with LangGraphEngine(tmp_path, model_factory=fake_model_factory) as engine: + await engine.build(_protected_root(kind)) + result = await engine.run("hello") + + assert result.visited == [] + assert result.answer == "Access denied." + assert result.used_tools == () + + +async def test_protected_root_requires_access_plugin_for_python_callers(tmp_path: Path): + async with LangGraphEngine(tmp_path, model_factory=fake_model_factory) as engine: + with pytest.raises(FileNotFoundError, match=r"access\.py"): + await engine.build(_protected_root("agent")) + + +@pytest.mark.parametrize("kind", ["agent", "orchestrator"]) +async def test_denied_root_does_not_resolve_prompt_plugins(tmp_path: Path, kind): + write_access(tmp_path, allow=False) + accessed = tmp_path / "prompt-accessed" + resolvers = tmp_path / "plugins" / "resolvers" + resolvers.mkdir() + (resolvers / "admin.py").write_text( + "from pathlib import Path\n" + "class Resolver:\n" + " def private_data(self, ctx):\n" + f" Path({str(accessed)!r}).write_text('accessed')\n" + " return 'private prompt data'\n", + encoding="utf-8", + ) + spec = _protected_root(kind) + spec = replace( + spec, + graph=replace( + spec.graph, + node=replace(spec.graph.node, resolvers=(ResolverSpec("private_data", "agent"),)), + ), + ) + async with LangGraphEngine(tmp_path, model_factory=fake_model_factory) as engine: + await engine.build(spec) + result = await engine.run("hello") + + assert not accessed.exists() + assert result.visited == [] + assert result.answer == "Access denied." + + +@pytest.mark.parametrize("kind", ["agent", "orchestrator"]) +async def test_root_access_decision_is_evaluated_for_each_request(tmp_path: Path, kind): + write_context_access(tmp_path) + permitted = RunContext( + metadata={"allowed_nodes": ("admin",)}, + auth_context=AuthContext(metadata={"groups": ("docs",)}), + ) + async with LangGraphEngine(tmp_path, model_factory=fake_model_factory) as engine: + await engine.build(_protected_root(kind)) + first = await engine.run("hello", context=permitted) + denied = await engine.run("hello") + last = await engine.run("hello", context=permitted) + + expected = ["admin"] if kind == "agent" else ["admin", "admin/public"] + assert first.visited == last.visited == expected + assert denied.visited == [] + assert denied.answer == "Access denied." + + +async def test_denial_reaches_the_streaming_chat_console(tmp_path: Path, capsys): + from agentctl.chat import _answer_local_stream + + write_access(tmp_path, allow=False) + async with LangGraphEngine(tmp_path, model_factory=fake_model_factory) as engine: + await engine.build(_protected_root("agent")) + await _answer_local_stream( + engine, "hello", context=None, read_line=lambda _: "", echo=print + ) + + assert "Access denied." in capsys.readouterr().out + + +@pytest.mark.parametrize("kind", ["agent", "orchestrator"]) +@pytest.mark.parametrize("streaming", [False, True]) +@pytest.mark.parametrize("permitted", [False, True]) +async def test_root_policy_is_rechecked_before_approval_replay( + tmp_path: Path, kind, streaming, permitted +): + allowed = tmp_path / "allowed" + allowed.touch() + write_access(tmp_path, allow=True) + (tmp_path / "plugins" / "access.py").write_text( + "from pathlib import Path\n" + "class AccessResolver:\n" + " def can_access(self, ctx, node_id):\n" + f" return Path({str(allowed)!r}).exists()\n", + encoding="utf-8", + ) + output = tmp_path / "executed.log" + write_tool(tmp_path, "send_email", output) + spec = _multi_tool_spec("send_email") + if kind == "orchestrator": + spec = system(orchestrator("root", [spec.graph])) + spec = replace(spec, graph=replace(spec.graph, node=replace(spec.graph.node, protected=True))) + async with LangGraphEngine(tmp_path, model_factory=fake_model_factory) as engine: + await engine.build(spec) + pending = await engine.run( + "hello", context=RunContext(run_id="protected-run", user_id="user", conversation_id="s") + ) + assert pending.pending_approval is not None + assert not output.exists() + approval_id = pending.pending_approval.approval_id + if not permitted: + allowed.unlink() + + async def resume(): + kwargs = {"caller_user_id": "user", "caller_session_id": "s"} + if streaming: + return [ + event + async for event in engine.resume_stream( + "protected-run", approval_id, "allow once", **kwargs + ) + ] + return await engine.resume("protected-run", approval_id, "allow once", **kwargs) + + if permitted: + await resume() + assert len(output.read_text().splitlines()) == 1 + assert await engine.get_run_status("protected-run") == RunStatus.COMPLETED + else: + # Root denial prevents reaching the reviewed tool. The existing + # replay-binding guard rejects that decision rather than falsely + # reporting that the approved invocation was executed. + with pytest.raises(ApprovalInvocationMismatch): + await resume() + assert not output.exists() + assert await engine.get_run_status("protected-run") == RunStatus.FAILED + assert await engine.get_pending_approval("protected-run") is None + assert ( + await engine.get_processed_result( + "protected-run", approval_id, caller_user_id="user", caller_session_id="s" + ) + is None + ) From 7fe53b52084488b25bf0542c0934b27445e9dc56 Mon Sep 17 00:00:00 2001 From: Asaf Varon Date: Fri, 23 Oct 2026 18:35:26 +0300 Subject: [PATCH 2/3] fix(approvals): refresh authorization context before replay Carry the current verified principal through both approval ports and use its request-local permission facts for protected root and child checks. Do not inherit checkpointed roles, scopes, organization or metadata. Validate caller/run/session consistency before claiming the approval and keep credentials out of graph projections. Add role revocation, missing context, permission facts, ownership and concurrent resume regressions. Document the additive Python port keyword. Fixes #151 --- docs/SIDECAR_CONTEXT_AUTH.md | 29 ++- ...sation-lifecycle-and-immutable-branches.md | 17 ++ src/agent_engine/engine/approval_engine.py | 2 + .../engine/approval_streaming_engine.py | 2 + src/agent_engine/engine/langgraph/engine.py | 81 +++---- .../engine/langgraph/graph/graph_builder.py | 11 +- src/agent_engine/runtime/hooks/context.py | 27 +++ .../application/conversation_service.py | 26 ++- tasks/issue-0151-protected-root-access.md | 94 ++++++++ tests/agent_manager/test_api.py | 7 + tests/api/test_protected_access_e2e.py | 83 +++++++ tests/api/test_run_context.py | 4 +- tests/engine/test_protected_root.py | 221 +++++++++++++++++- 13 files changed, 543 insertions(+), 61 deletions(-) diff --git a/docs/SIDECAR_CONTEXT_AUTH.md b/docs/SIDECAR_CONTEXT_AUTH.md index ca1379a0..8fce12fd 100644 --- a/docs/SIDECAR_CONTEXT_AUTH.md +++ b/docs/SIDECAR_CONTEXT_AUTH.md @@ -230,11 +230,30 @@ context. A root's decision is evaluated per execution, never cached between requests. This enforces node access without implementing API authentication or replacing authorization inside client tools and data services. -The root policy is also rechecked when a pending tool approval replays the root -executor. If that policy now denies access, the reviewed invocation is never -reached. The existing replay-binding guard raises `ApprovalInvocationMismatch`, -rejects the pending approval and fails the run; the human decision is not -reported as an executed action. This differs from a new run's normal refusal. +Approval replay uses the **current approval request's** trusted context. Python +hosts pass `context=RunContext(...)` to `resume` / `resume_stream`; +`agent_manager` builds it from the verified principal on each decision request, +including that principal's current roles, organization and credential. Supplied +run/user/session identities must match the approval's bound caller and run. +The root wrapper replaces its execution-local context projection before access +checks and child delegation. Saved roles, scopes, organization and metadata are +never merged into it, and credentials are never added to the graph projection. +Omitting fresh permission facts does not recover the old grant. A permissive +policy can still explicitly allow a context with no role or scope facts. + +The root policy is rechecked before its executor is entered. If that policy +denies access, the reviewed invocation is never reached. The existing +replay-binding guard raises `ApprovalInvocationMismatch`, rejects the pending +approval and fails the run; the decision is not reported as an executed action. +Removing a protected child can instead trigger the existing checkpoint replay +mismatch guard, which also prevents execution and fails the run. Neither case +re-generates the reviewed model/tool request. A denied resume retains the +existing error handling rather than a new run's normal refusal response. + +The optional Python approval ports require implementations to accept the new +`context` keyword; a host must not silently discard it for a legacy adapter. +The HTTP decision body, SSE event types, access-plugin signature and YAML +remain unchanged. See [ADR 0002](adr/0002-conversation-lifecycle-and-immutable-branches.md). --- diff --git a/docs/adr/0002-conversation-lifecycle-and-immutable-branches.md b/docs/adr/0002-conversation-lifecycle-and-immutable-branches.md index d9e2d29b..bf86f67b 100644 --- a/docs/adr/0002-conversation-lifecycle-and-immutable-branches.md +++ b/docs/adr/0002-conversation-lifecycle-and-immutable-branches.md @@ -78,6 +78,13 @@ One `conversation_id` owns an immutable message tree. turns and resumed turns share the engine's owned graph-task cancellation path; resume retains the original run and transitions `PENDING_APPROVAL → RESUMING → RUNNING` before emitting work. +- Approval replay uses current authorization facts supplied by the trusted + host on the decision request. The manager derives them from its freshly + verified Principal, never from the suspended run. Root entry replaces the + execution-local context projection for root and child access checks while + preserving the reviewed checkpoint/model/tool binding. Missing permission + facts cannot inherit checkpointed roles, scopes, organization or metadata. + Credentials stay in request-local context and are excluded from the projection. No separate `branch_id` is introduced. A branch is identified by its leaf/head message, which is already a stable domain identity. @@ -91,6 +98,16 @@ message, which is already a stable domain identity. not represented as an approval decision or as `DENY`. - Approval decisions add `/decision/stream`; its `resume_started` event carries the original `run_id`, and disconnecting it cancels the active graph task. +- The Python `ApprovalEngine.resume` and `ApprovalStreamingEngine.resume_stream` + ports accept optional `context: RunContext | None`. Embedding hosts supply a + current verified snapshot on every decision; the manager forwards current + roles, organization and credential. Supplied caller/run/session identities + must agree with the existing approval binding before it is claimed. External + implementations of these optional capabilities must accept the keyword; + silently dropping context to support an older signature is unsafe. Existing + direct calls remain valid, but without fresh facts a role/scope/metadata + policy cannot reuse the original grant. No HTTP/YAML/access-plugin schema + changes or new persistence revision are required for this correction. - Conversation sessions gain `head_message_id`; manager persistence includes durable run records. Because no database environment currently exists, no new migration revision was created; the existing fresh-database baseline revision diff --git a/src/agent_engine/engine/approval_engine.py b/src/agent_engine/engine/approval_engine.py index 9a3737e3..f0e99350 100644 --- a/src/agent_engine/engine/approval_engine.py +++ b/src/agent_engine/engine/approval_engine.py @@ -6,6 +6,7 @@ from agent_engine.approvals.decision import ApprovalDecision from agent_engine.engine.types import PendingApproval, RunResult +from agent_engine.runtime.hooks.models import RunContext @runtime_checkable @@ -30,4 +31,5 @@ async def resume( caller_user_id: str | None = None, caller_session_id: str | None = None, access_token: str | None = None, + context: RunContext | None = None, ) -> RunResult: ... diff --git a/src/agent_engine/engine/approval_streaming_engine.py b/src/agent_engine/engine/approval_streaming_engine.py index 40805bda..8b077b83 100644 --- a/src/agent_engine/engine/approval_streaming_engine.py +++ b/src/agent_engine/engine/approval_streaming_engine.py @@ -6,6 +6,7 @@ from typing import Protocol, runtime_checkable from agent_engine.approvals.decision import ApprovalDecision +from agent_engine.runtime.hooks.models import RunContext from agent_engine.runtime.streaming import RunStreamEvent @@ -20,4 +21,5 @@ def resume_stream( caller_user_id: str | None = None, caller_session_id: str | None = None, access_token: str | None = None, + context: RunContext | None = None, ) -> AsyncIterator[RunStreamEvent]: ... diff --git a/src/agent_engine/engine/langgraph/engine.py b/src/agent_engine/engine/langgraph/engine.py index 9d6b5397..945e0a47 100644 --- a/src/agent_engine/engine/langgraph/engine.py +++ b/src/agent_engine/engine/langgraph/engine.py @@ -4,6 +4,7 @@ import logging from collections.abc import AsyncGenerator, AsyncIterator, Awaitable, Callable, Iterator, Sequence from contextlib import contextmanager, suppress +from dataclasses import replace from pathlib import Path from typing import Any, cast @@ -16,7 +17,11 @@ from agent_engine.approvals.approval_manager import ApprovalManager from agent_engine.approvals.coordinator import ApprovalCoordinator from agent_engine.approvals.decision import ApprovalDecision, parse_decision -from agent_engine.approvals.errors import ApprovalInvocationMismatch, RunNotFound +from agent_engine.approvals.errors import ( + ApprovalInvocationMismatch, + RunNotFound, + UnauthorizedApprover, +) from agent_engine.approvals.in_memory_approval_repository import InMemoryApprovalRepository from agent_engine.approvals.in_memory_session_approval_repository import ( InMemorySessionApprovalRepository, @@ -74,6 +79,7 @@ RunContext, current_run_context, ) +from agent_engine.runtime.hooks.context import public_run_context from agent_engine.runtime.streaming import ( RunStreamEvent, StreamSinks, @@ -112,36 +118,10 @@ def _initial_state( ], } if expose_run_context: - state["run_context"] = _state_run_context(ctx) + state["run_context"] = public_run_context(ctx) return state -def _state_run_context(ctx: RunContext) -> dict[str, Any]: - """Return the generic, non-secret context exposed to graph runtime filters.""" - data: dict[str, Any] = {} - for key in ("run_id", "conversation_id", "user_id", "organization_id"): - value = getattr(ctx, key) - if value is not None: - data[key] = value - if ctx.metadata: - data["metadata"] = dict(ctx.metadata) - if ctx.auth_context is not None: - auth: dict[str, Any] = {} - for key in ("user_id", "organization_id"): - value = getattr(ctx.auth_context, key) - if value is not None: - auth[key] = value - if ctx.auth_context.scopes: - auth["scopes"] = tuple(ctx.auth_context.scopes) - if ctx.auth_context.roles: - auth["roles"] = tuple(ctx.auth_context.roles) - if ctx.auth_context.metadata: - auth["metadata"] = dict(ctx.auth_context.metadata) - if auth: - data["auth"] = auth - return data - - def _trace_metadata(ctx: RunContext) -> dict[str, Any]: """Map per-run identity onto the metadata keys Langfuse reads to group traces. @@ -694,6 +674,7 @@ async def resume( caller_user_id: str | None = None, caller_session_id: str | None = None, access_token: str | None = None, + context: RunContext | None = None, ) -> RunResult: """Apply a human decision to a pending tool call and resume the same run. @@ -703,6 +684,10 @@ async def resume( either executes or denies the original tool call. ``ALLOW_FOR_SESSION`` additionally records a session permission so the tool is not re-prompted for the rest of the conversation. + + ``context`` must come from the current trusted host request. Permission + facts are replaced on replay, never recovered from the checkpoint. + With no context, role/scope/metadata-based policies see no saved grants. """ app, lifecycle = self._require_built("resuming") kind = parse_decision(decision) @@ -713,6 +698,7 @@ async def resume( caller_user_id=caller_user_id, caller_session_id=caller_session_id, access_token=access_token, + context=context, ) approved = kind != ApprovalDecision.DENY log( @@ -775,8 +761,9 @@ async def resume_stream( caller_user_id: str | None = None, caller_session_id: str | None = None, access_token: str | None = None, + context: RunContext | None = None, ) -> AsyncIterator[RunStreamEvent]: - """Resume one approval through the same owned stream used by new runs.""" + """Resume with current trusted context through the owned run stream.""" app, lifecycle = self._require_built("streaming an approval resume") kind = parse_decision(decision) ctx, approval = await self._activate_approval_resume( @@ -786,6 +773,7 @@ async def resume_stream( caller_user_id=caller_user_id, caller_session_id=caller_session_id, access_token=access_token, + context=context, ) approved = kind != ApprovalDecision.DENY @@ -834,6 +822,7 @@ async def _activate_approval_resume( caller_user_id: str | None, caller_session_id: str | None, access_token: str | None = None, + context: RunContext | None = None, ) -> tuple[RunContext, ApprovalRecord]: """Claim and activate one resume without an interruptible ownership gap. @@ -844,24 +833,38 @@ async def _activate_approval_resume( """ async def activate() -> tuple[RunContext, ApprovalRecord]: + request = context or RunContext() + auth = request.auth_context or AuthContext() + users = { + value + for value in (caller_user_id, request.user_id, auth.user_id) + if value is not None + } + sessions = { + value for value in (caller_session_id, request.conversation_id) if value is not None + } + if request.run_id not in (None, run_id) or len(users) > 1 or len(sessions) > 1: + raise UnauthorizedApprover(approval_id) approval = await self._approval_manager.claim( run_id=run_id, approval_id=approval_id, - caller_user_id=caller_user_id, - caller_auth_ref=caller_session_id, + caller_user_id=next(iter(users), None), + caller_auth_ref=next(iter(sessions), None), ) - ctx = RunContext( + ctx = request.replace( run_id=run_id, conversation_id=approval.auth_ref, user_id=approval.authorized_user_id, - organization_id=approval.organization_id, - metadata={"approval_id": approval.approval_id}, - # The approver's credential as of now, not one captured when the - # run started and possibly expired while it waited for a human. - auth_context=AuthContext( + organization_id=request.organization_id or auth.organization_id, + metadata={**request.metadata, "approval_id": approval.approval_id}, + # Authorization facts come only from this request. Saved facts + # describe the original run and cannot prove current access. + auth_context=replace( + auth, user_id=approval.authorized_user_id, - organization_id=approval.organization_id, - inbound_access_token=access_token, + inbound_access_token=access_token + if access_token is not None + else auth.inbound_access_token, ), ) await lifecycle.activate_resume(ctx) diff --git a/src/agent_engine/engine/langgraph/graph/graph_builder.py b/src/agent_engine/engine/langgraph/graph/graph_builder.py index d9019a11..63d2d123 100644 --- a/src/agent_engine/engine/langgraph/graph/graph_builder.py +++ b/src/agent_engine/engine/langgraph/graph/graph_builder.py @@ -18,6 +18,7 @@ from agent_engine.core.spec import AgentSpec, BaseModelConfig, GraphNode, OrchestratorSpec from agent_engine.core.spec import ModelConfig as NodeModelConfig from agent_engine.engine.langgraph.checkpointing import CheckpointerHandle +from agent_engine.engine.langgraph.execution.approval_resume import current_approval_resume from agent_engine.engine.langgraph.execution.node_executor import NodeExecutor from agent_engine.engine.langgraph.filters import RouteFilter from agent_engine.engine.langgraph.graph.traversal import node_id @@ -29,6 +30,7 @@ from agent_engine.loaders.resolver_loader import ResolverLoader from agent_engine.loaders.tool_loader import ToolLoader from agent_engine.runtime.hooks import HookManager +from agent_engine.runtime.hooks.context import current_run_context, public_run_context from agent_engine.runtime.state import GraphState from agent_engine.runtime.streaming import current_streams from agent_engine.tool_usage.context_provider import ToolUsageContextProvider @@ -154,6 +156,13 @@ def _wire_node( filters = tuple(self._filters) async def execute_root(state: GraphState) -> GraphState: + if current_approval_resume.get() is not None: + ctx = current_run_context.get() + # Checkpoints retain the reviewed work, not current authority. + # Replace facts locally for the root and all delegated children. + state = {**state, "run_context": public_run_context(ctx) if ctx else {}} + if not node.node.protected: + return await executor.execute(state) # The root has no parent to filter it before delegation. Apply the # same policy before entering either kind of protected executor. candidates = [node.node] @@ -168,7 +177,7 @@ async def execute_root(state: GraphState) -> GraphState: return {"visited": [], "answer": answer} return await executor.execute(state) - builder.add_node(path, execute_root if node.node.protected else executor.execute) + builder.add_node(path, execute_root) builder.add_edge(path, END) def _build_orchestrator_node( diff --git a/src/agent_engine/runtime/hooks/context.py b/src/agent_engine/runtime/hooks/context.py index 0bdf9727..58212ad0 100644 --- a/src/agent_engine/runtime/hooks/context.py +++ b/src/agent_engine/runtime/hooks/context.py @@ -10,7 +10,34 @@ from __future__ import annotations from contextvars import ContextVar +from typing import Any from agent_engine.runtime.hooks.models import RunContext current_run_context: ContextVar[RunContext | None] = ContextVar("current_run_context", default=None) + + +def public_run_context(ctx: RunContext) -> dict[str, Any]: + """Project request facts for access filters without exposing credentials.""" + data: dict[str, Any] = {} + for key in ("run_id", "conversation_id", "user_id", "organization_id"): + value = getattr(ctx, key) + if value is not None: + data[key] = value + if ctx.metadata: + data["metadata"] = dict(ctx.metadata) + if ctx.auth_context is not None: + auth: dict[str, Any] = {} + for key in ("user_id", "organization_id"): + value = getattr(ctx.auth_context, key) + if value is not None: + auth[key] = value + if ctx.auth_context.scopes: + auth["scopes"] = tuple(ctx.auth_context.scopes) + if ctx.auth_context.roles: + auth["roles"] = tuple(ctx.auth_context.roles) + if ctx.auth_context.metadata: + auth["metadata"] = dict(ctx.auth_context.metadata) + if auth: + data["auth"] = auth + return data diff --git a/src/agent_manager/application/conversation_service.py b/src/agent_manager/application/conversation_service.py index f49ae259..ce37f88b 100644 --- a/src/agent_manager/application/conversation_service.py +++ b/src/agent_manager/application/conversation_service.py @@ -67,18 +67,20 @@ def _assistant_message_id(session_id: str, run_id: str) -> str: return uuid.uuid5(uuid.NAMESPACE_URL, f"agent-manager:{session_id}:{run_id}:assistant").hex -def _run_context(turn: PreparedConversationTurn) -> RunContext: +def _run_context( + run_id: str, session_id: str, user_id: str | None, principal: Principal +) -> RunContext: """The per-run context one turn executes under, carrying the caller's own host credential so tools act as this user rather than share a privileged key.""" return RunContext( - run_id=turn.run_id, - conversation_id=turn.session_id, - user_id=turn.user_id, + run_id=run_id, + conversation_id=session_id, + user_id=user_id, auth_context=AuthContext( - user_id=turn.user_id, - organization_id=turn.principal.organization_id, - roles=turn.principal.roles, - inbound_access_token=turn.principal.access_token, + user_id=user_id, + organization_id=principal.organization_id, + roles=principal.roles, + inbound_access_token=principal.access_token, ), ) @@ -224,7 +226,9 @@ async def send( result = await self._engine.run( turn.message, history=turn.history, - context=_run_context(turn), + context=_run_context( + turn.run_id, turn.session_id, turn.user_id, turn.principal + ), ) except asyncio.CancelledError: await self.cancel_turn(turn) @@ -458,6 +462,7 @@ async def decide_approval( caller_user_id=session.user_id, caller_session_id=session.session_id, access_token=principal.access_token, + context=_run_context(run_id, session.session_id, session.user_id, principal), ) except ApprovalAlreadyProcessed: recovered = await engine.get_processed_result( @@ -527,6 +532,7 @@ async def events() -> AsyncIterator[RunStreamEvent]: caller_user_id=session.user_id, caller_session_id=session.session_id, access_token=principal.access_token, + context=_run_context(run_id, session.session_id, session.user_id, principal), ) async with self._retaining_run_usage(conversation_id, run_id): try: @@ -607,7 +613,7 @@ async def stream_turn(self, turn: PreparedConversationTurn) -> AsyncIterator[Run engine_stream = self._engine.stream( turn.message, history=turn.history, - context=_run_context(turn), + context=_run_context(turn.run_id, turn.session_id, turn.user_id, turn.principal), ) async with self._retaining_run_usage(turn.session_id, turn.run_id): try: diff --git a/tasks/issue-0151-protected-root-access.md b/tasks/issue-0151-protected-root-access.md index fe7c8054..198efbd5 100644 --- a/tasks/issue-0151-protected-root-access.md +++ b/tasks/issue-0151-protected-root-access.md @@ -169,3 +169,97 @@ Fresh pre-commit validation passed: make check (1,094 tests, 9 dependency warnings, format/lint, mypy and generated-stub checks), make validate (both configured examples), and git diff --check. The reviewed source hash was unchanged throughout the gate. + +## PR 154 review follow-up — Fresh approval authorization + +The user's 2026-10-23 request to fix PR 154 authorizes addressing +https://github.com/extra-org/extra/pull/154#discussion_r4360627342 and updating +the existing branch. The root guard currently reads checkpointed roles; the +manager forwards only the new credential on approval. A policy that reads +roles or scopes can therefore accept a revoked grant. + +Scope additions required for this correction: + +- engine/approval_engine.py and engine/approval_streaming_engine.py: an optional + typed RunContext on the existing Python approval ports, documented in ADR 0002. +- engine/langgraph/engine.py: use fresh context while preserving approval-owner + binding; missing fresh permission facts must never fall back to checkpoints. +- engine/langgraph/graph/graph_builder.py and runtime/hooks/context.py: project + the active request's non-secret context into root/child execution on replay. +- agent_manager/application/conversation_service.py: supply the current verified + Principal for both approval paths using the existing RunContext model. +- tests/api/test_protected_access_e2e.py and tests/engine/test_protected_root.py: + real HTTP and engine revocation/allow/omitted-context regressions. +- tests/agent_manager/test_api.py: update approval doubles to the additive port. +- tests/api/test_run_context.py: follow the shared non-secret projection helper + when it moves out of the engine module; keep the credential assertion intact. +- docs/SIDECAR_CONTEXT_AUTH.md, ADR 0002 and this task: describe semantics, + compatibility, scope and verification. No HTTP/YAML/plugin schema change. + +Acceptance criteria: + +- [x] Both approval endpoints use the current verified principal's roles. +- [x] Protected agents/orchestrators and children reject revoked permissions + before the pending tool effect, including streaming replay. +- [x] Fresh scope/metadata facts replace old authorization facts; missing facts + do not inherit a grant from the original run. +- [x] Still-authorized approvals execute the reviewed tool exactly once. +- [x] Owner binding, credential forwarding, checkpoint replay and cancellation + remain intact; request context is never stored on the shared engine. +- [x] The reviewed symptom fails before the fix; make check and examples pass. + +Merged current main (6b9ada3) after resolving the task-index conflict by +retaining both sets of entries. The combined baseline passed make check: +1,498 tests / 27 dependency warnings, format/lint, mypy and generated stubs. + +### Follow-up result and validation + +The manager now sends a current RunContext through both typed approval ports. +The engine validates caller/run/session consistency before claiming an +approval and binds the active request context to the original approved owner. +Only current role/scope/organization/metadata facts enter access checks. The +shared non-secret projection helper is reused on initial execution and replay; +root entry copies state locally so child filters receive the same fresh facts. +No checkpoint edit, model regeneration, global identity state or persistence +migration is required. ADR 0002 records the additive Python keyword and adapter +upgrade requirement; HTTP/YAML/access-plugin shapes are unchanged. + +The valid initial regression run reproduced six HTTP revoked-role cases that +executed the tool, plus two omitted-context resumes that reused old roles; +the six still-authorized HTTP cases already passed. An earlier HTTP test setup +used the engine's free-text decision alias instead of the HTTP enum; that was +corrected before observing the security failures. The final suite adds 40 cases +in total, including scope, metadata, organization, ownership and concurrent +approval isolation; the two protected-access files now contain 79 cases. + +Final make check passed: **1,538 tests / 27 dependency warnings**, 363 files +formatted, lint, mypy (359 files) and generated-stub checks. Both make validate +examples and the WhatsApp pilot configuration passed; git diff --check passed. +An intermediate gate caught stale helper imports and test typing after adding +the keyword; these were corrected before the passing gate. No warnings were +suppressed, and all models/providers are fakes with temporary local storage. + +### Follow-up code review + +1. **Summary:** The checkpointed-authority bypass is corrected for root and + child access checks on normal and streamed approval replay. +2. **Blocking issues:** The reviewed revoked-role tool effect is fixed; no + additional blocking finding was confirmed in the scoped changes. +3. **Non-blocking issues:** Removing a protected child can reach the existing + checkpoint mismatch error path (sanitized HTTP 500 or SSE error), while root + denial uses the existing approval mismatch response. Both stop the effect + and fail the run; changing those public error policies is separate work. +4. **Architecture concerns:** The manager supplies verified identity through + typed ports; the engine remains generic, startup-scoped and free of request + state. Replayed work and current authority remain separate. +5. **Security concerns:** Missing fresh facts never inherit old grants. Caller + identity conflicts fail before claiming the approval. Token exclusion, + original invocation binding and independent concurrent contexts are covered. +6. **Testing gaps:** Real manager HTTP verification and engine replay use fake + models and observable local tool effects. The full gate covers existing + cancellation, duplicate decisions, credential forwarding and nested replay. + No live identity/provider deployment was exercised. +7. **Suggested improvements:** External approval adapters must accept the + new context keyword; hosts must supply a trusted current snapshot. +8. **Final recommendation:** Approve the scoped correction after independent + review. This is a local self-review, not a GitHub approval or merge. diff --git a/tests/agent_manager/test_api.py b/tests/agent_manager/test_api.py index 3f3c0195..2fd0e59a 100644 --- a/tests/agent_manager/test_api.py +++ b/tests/agent_manager/test_api.py @@ -130,6 +130,7 @@ async def resume( caller_user_id: str | None = None, caller_session_id: str | None = None, access_token: str | None = None, + context: RunContext | None = None, ) -> RunResult: self.resume_calls.append((run_id, approval_id, decision, caller_user_id, caller_session_id)) self.resume_tokens.append(access_token) @@ -153,6 +154,7 @@ async def resume_stream( caller_user_id: str | None = None, caller_session_id: str | None = None, access_token: str | None = None, + context: RunContext | None = None, ) -> AsyncIterator[RunStreamEvent]: self.resume_calls.append((run_id, approval_id, decision, caller_user_id, caller_session_id)) self.resume_tokens.append(access_token) @@ -855,6 +857,7 @@ async def resume_stream( caller_user_id: str | None = None, caller_session_id: str | None = None, access_token: str | None = None, + context: RunContext | None = None, ) -> AsyncIterator[RunStreamEvent]: del approval_id, decision, caller_user_id, caller_session_id, access_token await runs.transition_if_allowed(run_id, RunStatus.RESUMING) @@ -1175,6 +1178,7 @@ async def resume( caller_user_id: str | None = None, caller_session_id: str | None = None, access_token: str | None = None, + context: RunContext | None = None, ) -> RunResult: del run_id, approval_id, decision, caller_user_id, caller_session_id, access_token raise RuntimeError("private approval failure") @@ -1211,6 +1215,7 @@ async def resume( caller_user_id: str | None = None, caller_session_id: str | None = None, access_token: str | None = None, + context: RunContext | None = None, ) -> RunResult: del run_id, decision, caller_user_id, caller_session_id, access_token raise ApprovalNotFound(f"private-{approval_id}") @@ -1246,6 +1251,7 @@ async def resume( caller_user_id: str | None = None, caller_session_id: str | None = None, access_token: str | None = None, + context: RunContext | None = None, ) -> RunResult: if run_id in self.completed_results: raise ApprovalAlreadyProcessed(approval_id, "approved") @@ -1256,6 +1262,7 @@ async def resume( caller_user_id=caller_user_id, caller_session_id=caller_session_id, access_token=access_token, + context=context, ) class FailFirstAssistantWrite(MemoryRepository): diff --git a/tests/api/test_protected_access_e2e.py b/tests/api/test_protected_access_e2e.py index 4f618b21..215a3032 100644 --- a/tests/api/test_protected_access_e2e.py +++ b/tests/api/test_protected_access_e2e.py @@ -3,15 +3,25 @@ from __future__ import annotations import json +from dataclasses import replace from pathlib import Path from typing import Any import pytest from fastapi.testclient import TestClient +from httpx import ASGITransport, AsyncClient +from tests.agent_manager.conftest import bearer, build_test_app +from tests.approvals.test_engine_hitl import _multi_tool_spec +from tests.approvals.test_resume_binding import write_tool +from tests.engine.test_engine_flow import orchestrator, system from tests.fixtures.utils import FakeChatModel, fake_model_factory from agent_engine.api.app import create_app +from agent_engine.approvals.decision import ApprovalDecision +from agent_engine.approvals.models import RunStatus from agent_engine.engine.langgraph.engine import LangGraphEngine +from agent_manager.application import ConversationService +from agent_manager.infrastructure.persistence.memory_repository import MemoryRepository _FORGED_HEADERS = { "X-User-ID": "admin", @@ -131,3 +141,76 @@ def test_unprotected_root_still_executes_with_deny_all_policy(tmp_path, monkeypa assert response.status_code == 200 assert response.json()["visited"] == ["admin"] assert response.json()["answer"] == "ok" + + +@pytest.mark.parametrize("location", ["agent", "orchestrator", "child"]) +@pytest.mark.parametrize("streaming", [False, True]) +@pytest.mark.parametrize("permitted", [False, True]) +async def test_manager_approval_uses_current_verified_roles( + tmp_path, location, streaming, permitted +): + output = tmp_path / "executed.log" + write_tool(tmp_path, "send_email", output) + (tmp_path / "plugins" / "access.py").write_text( + "class AccessResolver:\n" + " def can_access(self, ctx, node_id):\n" + " return 'admin' in ctx.get('auth', {}).get('roles', ())\n", + encoding="utf-8", + ) + spec = _multi_tool_spec("send_email") + if location in {"agent", "child"}: + spec = replace( + spec, graph=replace(spec.graph, node=replace(spec.graph.node, protected=True)) + ) + if location != "agent": + spec = system(orchestrator("root", [spec.graph])) + if location == "orchestrator": + spec = replace( + spec, graph=replace(spec.graph, node=replace(spec.graph.node, protected=True)) + ) + async with LangGraphEngine(tmp_path, model_factory=fake_model_factory) as engine: + await engine.build(spec) + service = ConversationService(engine, MemoryRepository()) + app = build_test_app(service) + initial = bearer("owner", roles=["admin"], org_id="org") + current = bearer("owner", roles=["admin"] if permitted else [], org_id="org") + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: + response = await client.post( + "/conversations", json={"session_id": "session"}, headers=initial + ) + assert response.status_code == 200 + response = await client.post( + "/conversations/session/messages", json={"message": "hello"}, headers=initial + ) + assert response.status_code == 200 + pending = response.json()["pending_approval"] + assert pending is not None and not output.exists() + suffix = "/stream" if streaming else "" + response = await client.post( + f"/conversations/session/runs/{pending['run_id']}/approvals/" + f"{pending['approval_id']}/decision{suffix}", + json={"decision": ApprovalDecision.ALLOW_ONCE.value}, + headers={**current, "X-Roles": "admin", "X-User-ID": "owner"}, + ) + + if permitted: + assert response.status_code == 200 + assert len(output.read_text().splitlines()) == 1 + assert await engine.get_run_status(pending["run_id"]) == RunStatus.COMPLETED + else: + assert not output.exists() + if streaming: + assert response.status_code == 200 + events = [ + json.loads(line[6:]) + for line in response.text.splitlines() + if line.startswith("data: ") and line != "data: [DONE]" + ] + assert any(event["type"] == "error" for event in events) + assert not any(event["type"] == "final" for event in events) + else: + # Removing a child changes the replayed model tool set, which + # the existing checkpoint guard rejects before tool execution. + assert response.status_code == (500 if location == "child" else 409) + assert await engine.get_run_status(pending["run_id"]) == RunStatus.FAILED + assert await engine.get_pending_approval(pending["run_id"]) is None diff --git a/tests/api/test_run_context.py b/tests/api/test_run_context.py index 0759f872..824587f5 100644 --- a/tests/api/test_run_context.py +++ b/tests/api/test_run_context.py @@ -4,7 +4,7 @@ from __future__ import annotations from agent_engine.api.app import _run_context -from agent_engine.engine.langgraph.engine import _state_run_context +from agent_engine.runtime.hooks.context import public_run_context TOKEN = "eyJhbGciOiJIUzI1NiJ9.host-token.signature" @@ -35,4 +35,4 @@ def test_the_credential_never_reaches_the_state_the_model_sees() -> None: """Graph state reaches prompts and callers; a credential there is a leak.""" ctx = _run_context(None, run_id="r1", authorization=f"Bearer {TOKEN}") - assert TOKEN not in repr(_state_run_context(ctx)) + assert TOKEN not in repr(public_run_context(ctx)) diff --git a/tests/engine/test_protected_root.py b/tests/engine/test_protected_root.py index 7f519c75..8114c50b 100644 --- a/tests/engine/test_protected_root.py +++ b/tests/engine/test_protected_root.py @@ -2,12 +2,14 @@ from __future__ import annotations +import asyncio +import json from dataclasses import replace from pathlib import Path import pytest -from agent_engine.approvals.errors import ApprovalInvocationMismatch +from agent_engine.approvals.errors import ApprovalInvocationMismatch, UnauthorizedApprover from agent_engine.approvals.models import RunStatus from agent_engine.core.spec import ResolverSpec from agent_engine.engine.langgraph.engine import LangGraphEngine @@ -180,15 +182,24 @@ async def test_root_policy_is_rechecked_before_approval_replay( allowed.unlink() async def resume(): - kwargs = {"caller_user_id": "user", "caller_session_id": "s"} if streaming: return [ event async for event in engine.resume_stream( - "protected-run", approval_id, "allow once", **kwargs + "protected-run", + approval_id, + "allow once", + caller_user_id="user", + caller_session_id="s", ) ] - return await engine.resume("protected-run", approval_id, "allow once", **kwargs) + return await engine.resume( + "protected-run", + approval_id, + "allow once", + caller_user_id="user", + caller_session_id="s", + ) if permitted: await resume() @@ -209,3 +220,205 @@ async def resume(): ) is None ) + + +@pytest.mark.parametrize("streaming", [False, True]) +async def test_approval_without_fresh_context_cannot_reuse_checkpoint_roles(tmp_path, streaming): + output = tmp_path / "executed.log" + write_tool(tmp_path, "send_email", output) + (tmp_path / "plugins" / "access.py").write_text( + "class AccessResolver:\n" + " def can_access(self, ctx, node_id):\n" + " return 'admin' in ctx.get('auth', {}).get('roles', ())\n", + encoding="utf-8", + ) + spec = _multi_tool_spec("send_email") + spec = replace(spec, graph=replace(spec.graph, node=replace(spec.graph.node, protected=True))) + async with LangGraphEngine(tmp_path, model_factory=fake_model_factory) as engine: + await engine.build(spec) + pending = await engine.run( + "hello", context=RunContext(run_id="r", auth_context=AuthContext(roles=("admin",))) + ) + assert pending.pending_approval is not None + with pytest.raises(ApprovalInvocationMismatch): + if streaming: + _ = [ + event + async for event in engine.resume_stream( + "r", pending.pending_approval.approval_id, "allow once" + ) + ] + else: + await engine.resume("r", pending.pending_approval.approval_id, "allow once") + assert not output.exists() + + +@pytest.mark.parametrize("fact", ["scopes", "metadata", "auth_metadata", "organization"]) +@pytest.mark.parametrize("streaming", [False, True]) +@pytest.mark.parametrize("permitted", [False, True]) +async def test_approval_authorization_facts_are_fresh(tmp_path, fact, streaming, permitted): + output = tmp_path / "executed.log" + policy_calls = tmp_path / "policy.jsonl" + write_tool(tmp_path, "send_email", output) + conditions = { + "scopes": "'write' in ctx.get('auth', {}).get('scopes', ())", + "metadata": "ctx.get('metadata', {}).get('permission') == 'write'", + "auth_metadata": "ctx.get('auth', {}).get('metadata', {}).get('permission') == 'write'", + "organization": "ctx.get('auth', {}).get('organization_id') == 'org'", + } + (tmp_path / "plugins" / "access.py").write_text( + "import json\n" + "class AccessResolver:\n" + " def can_access(self, ctx, node_id):\n" + f" with open({str(policy_calls)!r}, 'a') as output:\n" + " output.write(json.dumps(ctx) + '\\n')\n" + f" return {conditions[fact]}\n", + encoding="utf-8", + ) + spec = _multi_tool_spec("send_email") + spec = replace(spec, graph=replace(spec.graph, node=replace(spec.graph.node, protected=True))) + initial = RunContext( + run_id="r", + user_id="user", + conversation_id="s", + metadata={"permission": "write"}, + auth_context=AuthContext( + scopes=("write",), + organization_id="org", + metadata={"permission": "write"}, + inbound_access_token="old-secret", + ), + ) + # The resume host supplies a complete current snapshot; omitted facts must + # not be filled from the original run, even when the checkpoint has them. + current = initial.replace( + metadata={"permission": "write"} if permitted else {}, + auth_context=AuthContext( + scopes=("write",) if permitted else (), + organization_id="org" if permitted else "other", + metadata={"permission": "write"} if permitted else {}, + inbound_access_token="new-secret", + ), + ) + async with LangGraphEngine(tmp_path, model_factory=fake_model_factory) as engine: + await engine.build(spec) + pending = await engine.run("hello", context=initial) + assert pending.pending_approval is not None + approval_id = pending.pending_approval.approval_id + + async def resume(): + if streaming: + return [ + event + async for event in engine.resume_stream( + "r", approval_id, "allow once", context=current + ) + ] + return await engine.resume("r", approval_id, "allow once", context=current) + + if permitted: + await resume() + assert len(output.read_text().splitlines()) == 1 + else: + with pytest.raises(ApprovalInvocationMismatch): + await resume() + assert not output.exists() + calls = policy_calls.read_text() + assert "old-secret" not in calls and "new-secret" not in calls + latest = json.loads(calls.splitlines()[-1]) + assert latest["auth"].get("scopes", []) == (["write"] if permitted else []) + assert latest["auth"]["organization_id"] == ("org" if permitted else "other") + assert latest["metadata"].get("permission") == ("write" if permitted else None) + + +@pytest.mark.parametrize("identity", ["run", "session", "user", "auth_user"]) +@pytest.mark.parametrize("streaming", [False, True]) +async def test_resume_context_cannot_replace_approval_ownership(tmp_path, identity, streaming): + output = tmp_path / "executed.log" + write_tool(tmp_path, "send_email", output) + context = RunContext(run_id="r", user_id="owner", conversation_id="s") + contexts = { + "run": context.replace(run_id="another-run"), + "session": context.replace(conversation_id="another-session"), + "user": context.replace(user_id="another-user"), + "auth_user": context.replace(auth_context=AuthContext(user_id="another-user")), + } + async with LangGraphEngine(tmp_path, model_factory=fake_model_factory) as engine: + await engine.build(_multi_tool_spec("send_email")) + pending = await engine.run("hello", context=context) + assert pending.pending_approval is not None + with pytest.raises(UnauthorizedApprover): + if streaming: + _ = [ + event + async for event in engine.resume_stream( + "r", + pending.pending_approval.approval_id, + "allow once", + caller_user_id="owner", + caller_session_id="s", + context=contexts[identity], + ) + ] + else: + await engine.resume( + "r", + pending.pending_approval.approval_id, + "allow once", + caller_user_id="owner", + caller_session_id="s", + context=contexts[identity], + ) + assert not output.exists() + assert await engine.get_run_status("r") == RunStatus.PENDING_APPROVAL + assert await engine.get_pending_approval("r") is not None + + +@pytest.mark.parametrize("streaming", [False, True]) +async def test_concurrent_approval_requests_do_not_share_authorization(tmp_path, streaming): + output = tmp_path / "executed.log" + write_tool(tmp_path, "send_email", output) + (tmp_path / "plugins" / "access.py").write_text( + "class AccessResolver:\n" + " def can_access(self, ctx, node_id):\n" + " return 'admin' in ctx.get('auth', {}).get('roles', ())\n", + encoding="utf-8", + ) + spec = _multi_tool_spec("send_email") + spec = replace(spec, graph=replace(spec.graph, node=replace(spec.graph.node, protected=True))) + async with LangGraphEngine(tmp_path, model_factory=fake_model_factory) as engine: + await engine.build(spec) + approvals: dict[str, str] = {} + for run_id in ("allowed", "revoked"): + pending = await engine.run( + "hello", + context=RunContext(run_id=run_id, auth_context=AuthContext(roles=("admin",))), + ) + assert pending.pending_approval is not None + approvals[run_id] = pending.pending_approval.approval_id + + async def resume_one(run_id): + current = RunContext( + auth_context=AuthContext(roles=("admin",) if run_id == "allowed" else ()) + ) + + async def invoke(): + if streaming: + return [ + event + async for event in engine.resume_stream( + run_id, approvals[run_id], "allow once", context=current + ) + ] + return await engine.resume(run_id, approvals[run_id], "allow once", context=current) + + if run_id == "allowed": + await invoke() + else: + with pytest.raises(ApprovalInvocationMismatch): + await invoke() + + await asyncio.gather(resume_one("allowed"), resume_one("revoked")) + assert len(output.read_text().splitlines()) == 1 + assert await engine.get_run_status("allowed") == RunStatus.COMPLETED + assert await engine.get_run_status("revoked") == RunStatus.FAILED From eb5a460ac0669a0b3a6408c69e3330052e5c88e9 Mon Sep 17 00:00:00 2001 From: Asaf Varon Date: Sat, 24 Oct 2026 12:00:25 +0300 Subject: [PATCH 3/3] fix(cli): forward current context when approving local tools Pass the local chat RunContext through the approval port so protected roots retain current role, scope and metadata facts in both console modes. Add real-engine regressions for allowed and revoked permissions without exposing credentials or weakening fail-closed replay checks. --- src/agentctl/chat.py | 1 + tasks/issue-0151-protected-root-access.md | 51 +++++++++++++ tests/cli/test_chat_command.py | 88 +++++++++++++++++++++-- 3 files changed, 136 insertions(+), 4 deletions(-) diff --git a/src/agentctl/chat.py b/src/agentctl/chat.py index d38a8602..01798ca7 100644 --- a/src/agentctl/chat.py +++ b/src/agentctl/chat.py @@ -246,6 +246,7 @@ async def _resolve_local_approvals( decision, caller_user_id=context.user_id if context is not None else None, caller_session_id=context.conversation_id if context is not None else None, + context=context, ) return result diff --git a/tasks/issue-0151-protected-root-access.md b/tasks/issue-0151-protected-root-access.md index 198efbd5..0977834a 100644 --- a/tasks/issue-0151-protected-root-access.md +++ b/tasks/issue-0151-protected-root-access.md @@ -263,3 +263,54 @@ suppressed, and all models/providers are fakes with temporary local storage. new context keyword; hosts must supply a trusted current snapshot. 8. **Final recommendation:** Approve the scoped correction after independent review. This is a local self-review, not a GitHub approval or merge. + +## PR 154 local chat follow-up (2026-10-24) + +The user requested fixing the current PR review, including +https://github.com/extra-org/extra/pull/154#discussion_r4361338882. +Local chat receives a host-supplied RunContext but drops it at the approval +port. This makes an authorized protected root fail on local HITL replay. + +Scope: src/agentctl/chat.py forwards its existing context on every local +approval; tests/cli/test_chat_command.py adds real-engine regressions for both +local stream modes and updates its approval double to the existing port; +this task records scope, verification and self-review. No new contract or +identity source, checkpoint fallback, permission relaxation or migration. + +Acceptance criteria: + +- [x] Authorized local chat approvals retain current role/scope/metadata facts + and execute the pending tool once, with stream enabled or disabled. +- [x] Protected root agents and orchestrators are covered. +- [x] Revoking a metadata permission at the approval prompt still prevents the + tool effect, using current context rather than checkpointed facts. +- [x] Credentials remain absent from policy projections and console output. +- [x] Existing role-revocation HTTP/engine regressions and make check pass. + +After validation, commit and push this scoped correction to the existing PR +branch and verify the remote head in the final report. + +Validation: all eight new cases failed before the production change. Four +authorized cases failed to resume; four revoked cases showed that the actual +current metadata was omitted. All eight pass after forwarding context. The +focused local-chat/API/engine set passed, and make check passed **1,546 tests / +27 dependency warnings**, 363 formatting checks, lint, mypy (359 files) and +generated-stub checks. make validate and the WhatsApp pilot example passed; +git diff --check passed. Tests use fake models and observable temporary-file +tool effects. No live provider or development database was used. + +Local self-review: + +1. **Summary:** The local approval adapter now carries its host-supplied context + through the existing typed port in both console modes. +2. **Blocking issues:** The reported local HITL regression is fixed. +3. **Non-blocking issues:** No new actionable finding in this scoped change. +4. **Architecture concerns:** No new contract, shared request state or runtime + policy is introduced; the engine still enforces access. +5. **Security concerns:** Changed permission facts are observed before the tool + effect; credentials stay outside policy projections and console output. +6. **Testing gaps:** Both root kinds and console modes cover allow/revoke with + a real engine and fake models; existing revocation regressions also pass. +7. **Suggested improvements:** None required for this correction. +8. **Final recommendation:** Approve the scoped fix after independent review; + this is not a GitHub approval, merge or deployment. diff --git a/tests/cli/test_chat_command.py b/tests/cli/test_chat_command.py index 593bd7cf..e582ac72 100644 --- a/tests/cli/test_chat_command.py +++ b/tests/cli/test_chat_command.py @@ -1,13 +1,15 @@ """Tests for the interactive ``agentctl chat`` console (local + remote modes). -No real LLM, MCP, or network is touched: the local engine is a fake stand-in and -the remote server is an httpx ``MockTransport``. +No real LLM, MCP, or network is touched: local tests use fake engines or the +real engine with fake models, and remote tests use httpx ``MockTransport``. """ from __future__ import annotations import json from collections.abc import AsyncIterator, Callable, Sequence +from dataclasses import replace +from pathlib import Path import httpx import pytest @@ -16,12 +18,17 @@ from agent_engine.approvals.decision import ApprovalDecision from agent_engine.approvals.models import RunStatus from agent_engine.engine.engine import Engine +from agent_engine.engine.langgraph.engine import LangGraphEngine from agent_engine.engine.types import ChatMessage, PendingApproval, RunResult -from agent_engine.runtime.hooks.models import RunContext +from agent_engine.runtime.hooks.models import AuthContext, RunContext from agent_engine.runtime.streaming import RunStreamEvent from agentctl import chat as chat_mod from agentctl.chat import run_chat_loop, run_remote_chat from agentctl.main import cli +from tests.approvals.test_engine_hitl import _multi_tool_spec +from tests.approvals.test_resume_binding import write_tool +from tests.engine.test_engine_flow import orchestrator, system +from tests.fixtures.utils import fake_model_factory # --------------------------------------------------------------------------- # fakes / helpers @@ -157,8 +164,9 @@ async def resume( caller_user_id: str | None = None, caller_session_id: str | None = None, access_token: str | None = None, + context: RunContext | None = None, ) -> RunResult: - del caller_session_id, access_token + del caller_session_id, access_token, context assert isinstance(decision, ApprovalDecision) self.resume_calls.append((run_id, approval_id, decision, caller_user_id)) return next(self.results) @@ -540,6 +548,78 @@ async def test_local_stream_prompts_and_resumes_pending_approval() -> None: assert "Agent > stream resumed" in echo.lines +@pytest.mark.parametrize("stream", [False, True]) +@pytest.mark.parametrize("root_kind", ["agent", "orchestrator"]) +@pytest.mark.parametrize("revoke", [False, True]) +async def test_local_approval_retains_current_authorization( + tmp_path: Path, capsys: pytest.CaptureFixture[str], stream: bool, root_kind: str, revoke: bool +) -> None: + output = tmp_path / "executed.log" + policy = tmp_path / "policy.jsonl" + write_tool(tmp_path, "send_email", output) + (tmp_path / "plugins" / "access.py").write_text( + "import json\n" + "class AccessResolver:\n" + " def can_access(self, ctx, node_id):\n" + f" with open({str(policy)!r}, 'a') as output:\n" + " output.write(json.dumps(ctx) + '\\n')\n" + " auth = ctx.get('auth', {})\n" + " return ('admin' in auth.get('roles', ())\n" + " and 'write' in auth.get('scopes', ())\n" + " and ctx.get('metadata', {}).get('permission') is True)\n", + encoding="utf-8", + ) + spec = _multi_tool_spec("send_email") + if root_kind == "orchestrator": + spec = system(orchestrator("root", [spec.graph])) + spec = replace(spec, graph=replace(spec.graph, node=replace(spec.graph.node, protected=True))) + context = RunContext( + run_id="local-run", + user_id="owner", + conversation_id="local-session", + metadata={"permission": True}, + auth_context=AuthContext( + user_id="owner", + roles=("admin",), + scopes=("write",), + inbound_access_token="local-test-credential", + ), + ) + inputs = scripted_reader(["hello", "o"]) + + def read_line(prompt: str) -> str: + if prompt.startswith("Allow?") and revoke: + # Metadata is the explicitly mutable host/hook enrichment field. + context.metadata["permission"] = False + return inputs(prompt) + + echo = CollectingEcho() + async with LangGraphEngine(tmp_path, model_factory=fake_model_factory) as engine: + await engine.build(spec) + await run_chat_loop( + engine, stream=stream, context=context, read_line=read_line, echo=echo.write + ) + if revoke: + assert not output.exists() + assert await engine.get_run_status("local-run") == RunStatus.FAILED + assert await engine.get_pending_approval("local-run") is None + else: + assert not any(line.startswith("✗") for line in echo.lines), echo.lines + assert len(output.read_text().splitlines()) == 1 + assert await engine.get_run_status("local-run") == RunStatus.COMPLETED + assert any(line.startswith("Agent > ") for line in echo.lines) + decisions = policy.read_text() + latest = json.loads(decisions.splitlines()[-1]) + assert latest["metadata"]["permission"] is (not revoke) + assert latest["auth"]["roles"] == ["admin"] + assert latest["auth"]["scopes"] == ["write"] + assert latest["user_id"] == "owner" + assert latest["conversation_id"] == "local-session" + assert "local-test-credential" not in decisions + captured = capsys.readouterr() + assert "local-test-credential" not in "\n".join(echo.lines) + captured.out + captured.err + + # --------------------------------------------------------------------------- # remote mode: endpoint, error handling, stream flag # ---------------------------------------------------------------------------