fix(engine): enforce access checks before protected root execution

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,57 @@ 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.
+
+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).
+
 ---
 
 ## Why Not YAML Roles?

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

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: ...

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]: ...

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)
@@ -1043,9 +1046,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:

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,8 @@
 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
 from agent_engine.engine.langgraph.nodes.agent_node import AgentNode
@@ -27,12 +30,15 @@
 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
 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 +147,37 @@ 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:
+            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]
+            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)
         builder.add_edge(path, END)
 
     def _build_orchestrator_node(