feat(authz): propagate trusted authorization principal context (#4203)

backend/AGENTS.md

@@ -237,6 +237,10 @@ Lead-agent middlewares are assembled in strict order across three functions: the
 12. **ToolProgressMiddleware** - *(optional, if `tool_progress.enabled`)* State-machine-based stagnation guard (RFC #3177). Outer wrapper around ToolErrorHandlingMiddleware so its `wrap_tool_call` receives results already stamped with `deerflow_tool_meta`. Tracks per-(thread, tool) consecutive "no-new-info" calls across three error categories: (a) `recoverable_by_model=True` (no_results, not_found, permission, Jaccard-duplicate success): ACTIVE → WARNED (terminal — hint re-injected on each subsequent problem); (b) `recoverable_by_model=False, action≠stop` (rate_limited, transient): ACTIVE → WARNED → BLOCKED after `warn_escalation_count` more problems; (c) `recoverable_by_model=False, action=stop` (auth, config, internal): immediately BLOCKED on first occurrence. **Division of labor with LoopDetectionMiddleware:** ToolProgressMiddleware is a result-quality guard — fires after tool execution and blocks specific tools that stop producing new information; LoopDetectionMiddleware is a call-pattern guard — fires after the model responds and hard-stops the whole turn when the model repeatedly issues identical tool_calls. Both can inject HumanMessage hints in the same model call without conflict; neither reads the other's internal state.
 13. **ToolErrorHandlingMiddleware** - Receives `AppConfig`, converts tool exceptions into error `ToolMessage`s so the run can continue instead of aborting, stamps every result with `deerflow_tool_meta` (status / error_type / recoverable_by_model / recommended_next_action / source) via `tool_result_meta.normalize_tool_result`, stamps structured metadata for task exception wrappers, and stamps skill-read metadata for downstream durable-context capture. Task tool result text is generated from the same status/result/error inputs as the structured metadata so callers do not hand-write a second protocol string.
 
+Authorization identity plumbing is independent of whether authorization enforcement is enabled. Gateway removes client-supplied `is_internal` / `authz_attributes` / `channel_user_id`, derives `is_internal` only from the server-owned `request.state.auth_source`, and accepts `channel_user_id` only from an internally authenticated IM caller's top-level `body.context`; free-form `body.config` can never supply it. `build_principal_from_context` is the shared Principal builder for assembly-time authorization and `GuardrailAuthorizationAdapter`; it applies `default_role`, strict-boolean internal provenance, and copy-on-read `authz_attributes`. Task delegation carries `is_internal` plus copied attributes through `SubagentExecutor`, while `GuardrailMiddleware` maps the same runtime fields into `GuardrailRequest`. Phase 1A provides this trusted identity chain only; automatic Layer 1 filtering and Layer 2 authorization middleware wiring remain separate enforcement work.
+
+Before changing a later authorization phase, read the [authorization RFC](../docs/plans/2026-07-10-pluggable-authorization-rfc.md) and its [implementation notes](../docs/plans/2026-07-10-pluggable-authorization-implementation-notes.md). The notes are the cumulative handoff record for merged PR behavior, reviewer feedback, trust-boundary decisions, deferred scope, and required regression coverage.
+
 **Lead-only middlewares** (`build_middlewares`, appended after the base):
 
 14. **DynamicContextMiddleware** - Injects the current date (and optionally memory) as a `<system-reminder>` into the first HumanMessage, keeping the base system prompt fully static for prefix-cache reuse
@@ -510,7 +514,7 @@ Bridges external messaging platforms (Feishu, Slack, Telegram, Discord, DingTalk
 1. External platform -> Channel impl -> `MessageBus.publish_inbound()`
    - For GitHub, the webhook router verifies the delivery then calls `fanout_event(bus, ...)`; matching agent bindings publish one `InboundMessage` each instead of a long-polling channel worker.
 2. `ChannelManager._dispatch_loop()` consumes from queue
-3. For user-owned channel connections, incoming messages carry `connection_id`, `owner_user_id`, and `workspace_id`; `owner_user_id` becomes the DeerFlow run `user_id`, while the raw platform user id remains `channel_user_id`. The Gateway forwards `channel_user_id` from `body.context` into the runtime context only (never `configurable`, which is checkpointed), and `bash_tool` exposes it to sandbox commands as the fixed env var `DEERFLOW_CHANNEL_USER_ID` — via a shell-quoted command-string prefix, NOT the `execute_command(env=...)` channel, which is reserved for request-scoped secrets and would switch `AioSandbox` onto the `bash.exec` path (image >= 1.9.3, fresh session per call). Per-call injection keeps group-chat identity correct (one thread/sandbox, many senders) **without depending on the AIO shell's session semantics**: every IM-channel command carries an explicit `export VAR=<id>; ` (valid id) or `unset VAR; ` (empty / non-str / over the 256-char cap, since `body.context` is client-writable). The AIO no-env path reuses a persistent shell session (the reason for the class lock, #1433), so a bare command could otherwise resolve a stale id an earlier sender exported; the `unset` closes the window the length/type guard would open (a dropped id would inherit the previous sender's value). Non-IM runs (no `channel_user_id` in context) are left untouched. Not injected on the Windows local sandbox (its PowerShell/cmd.exe fallback has no `export`/`unset`). Propagates across `task` delegation: `task_tool` captures the dispatching turn's id and the subagent executor forwards it into the subagent's runtime context, same as the guardrail attribution fields. The var is informational, never authorization-grade: any bash command can overwrite it (and web clients can set `body.context.channel_user_id`), so skills must not treat it as authenticated identity. Tests: `tests/test_channel_user_id_env.py`
+3. For user-owned channel connections, incoming messages carry `connection_id`, `owner_user_id`, and `workspace_id`; `owner_user_id` becomes the DeerFlow run `user_id`, while the raw platform user id remains `channel_user_id`. The Gateway accepts `channel_user_id` only from an internally authenticated channel caller's top-level `body.context`, clears it from both free-form `body.config` sections, and writes it into runtime context only (never `configurable`, which is checkpointed). `bash_tool` exposes it to sandbox commands as the fixed env var `DEERFLOW_CHANNEL_USER_ID` — via a shell-quoted command-string prefix, NOT the `execute_command(env=...)` channel, which is reserved for request-scoped secrets and would switch `AioSandbox` onto the `bash.exec` path (image >= 1.9.3, fresh session per call). Per-call injection keeps group-chat identity correct (one thread/sandbox, many senders) **without depending on the AIO shell's session semantics**: every IM-channel command carries an explicit `export VAR=<id>; ` (valid id) or `unset VAR; ` (empty / non-str / over the 256-char cap). The AIO no-env path reuses a persistent shell session (the reason for the class lock, #1433), so a bare command could otherwise resolve a stale id an earlier sender exported; the `unset` closes the window the length/type guard would open (a dropped id would inherit the previous sender's value). Non-IM runs (no `channel_user_id` in context) are left untouched. Not injected on the Windows local sandbox (its PowerShell/cmd.exe fallback has no `export`/`unset`). Propagates across `task` delegation: `task_tool` captures the dispatching turn's id and the subagent executor forwards it into the subagent's runtime context, same as the guardrail attribution fields. The runtime-context value is authorization-grade at the Gateway/guardrail boundary, but the exported shell variable remains informational because any bash command can overwrite its own environment; skills must not treat the shell variable itself as authenticated identity. Tests: `tests/test_gateway_services.py`, `tests/test_channel_user_id_env.py`
 4. For chat: look up/create thread through Gateway's LangGraph-compatible API
 5. Feishu/Telegram chat: `runs.stream()` → accumulate AI text → publish multiple outbound updates (`is_final=False`) → publish final outbound (`is_final=True`)
 6. Slack/Discord chat: `runs.wait()` → extract final response → publish outbound

backend/app/gateway/services.py

@@ -210,6 +210,15 @@ _CONTEXT_CONFIGURABLE_KEYS: frozenset[str] = frozenset(
 # arbitrary HTTP/IM clients must not be able to force autonomous execution.
 _CONTEXT_INTERNAL_CALLER_KEYS: frozenset[str] = frozenset({"non_interactive"})
 
+# Server-owned authorization identity fields. These must never be accepted from
+# client-supplied ``body.config.context`` or ``body.config.configurable``. They
+# are either produced by Gateway auth state or admitted from a separately
+# authenticated internal request channel.
+#   ``is_internal``       — derived from ``request.state.auth_source``
+#   ``authz_attributes`` — Phase 1A has no Gateway-side producer; always cleared.
+#   ``channel_user_id``  — accepted only from trusted internal ``body.context``.
+_SERVER_OWNED_AUTHZ_CONTEXT_KEYS: frozenset[str] = frozenset({"is_internal", "authz_attributes", "channel_user_id"})
+
 # Keys forwarded from ``body.context`` into ``config['context']`` ONLY (the
 # runtime context that becomes ``ToolRuntime.context`` / ``runtime.context``),
 # never into ``config['configurable']``. These are read by tools and
@@ -255,8 +264,9 @@ def merge_run_context_overrides(config: dict[str, Any], context: Mapping[str, An
     ``setdefault`` so a server-authenticated id stamped by
     :func:`inject_authenticated_user_context` always wins over the client-supplied one.
 
-    :data:`_CONTEXT_INTERNAL_CALLER_KEYS`; those keys are dropped from client
-    requests.
+    :data:`_CONTEXT_INTERNAL_CALLER_KEYS` are also forwarded when ``internal``
+    is True; for non-internal callers those keys are dropped from client requests
+    by :func:`strip_internal_context_keys`.
 
     A second set of keys (``_CONTEXT_RUNTIME_ONLY_KEYS`` — e.g. ``github_token``,
     ``disable_clarification``) is forwarded into ``config['context']`` only, never
@@ -282,11 +292,6 @@ def merge_run_context_overrides(config: dict[str, Any], context: Mapping[str, An
             runtime_context.setdefault(key, context[key])
     if "user_id" in context and isinstance(runtime_context, dict):
         runtime_context.setdefault("user_id", context["user_id"])
-    # The raw platform user id from IM channels (Feishu open_id, Slack Uxxx, ...)
-    # follows the same runtime-context-only rule as user_id: tools may read it,
-    # but it never enters ``configurable`` (checkpointed with the thread).
-    if "channel_user_id" in context and isinstance(runtime_context, dict):
-        runtime_context.setdefault("channel_user_id", context["channel_user_id"])
 
 
 async def resolve_trusted_internal_owner_for_attribution(request: Request, owner_user_id: str | None) -> Any | None:
@@ -309,14 +314,39 @@ def inject_authenticated_user_context(
     request: Request,
     *,
     internal_owner_user: Any | None = None,
+    request_context: Mapping[str, Any] | None = None,
 ) -> None:
     """Stamp the authenticated user into the run context for background tools.
 
     Tool execution may happen after the request handler has returned, so tools
     that persist user-scoped files should not rely only on ambient ContextVars.
     The value comes from server-side auth state, never from client context.
+
+    ``request_context.channel_user_id`` is the sole exception: it is honored
+    only after ``request.state.auth_source`` proves the caller is internal.
+    Values copied through the free-form RunnableConfig are always cleared.
     """
 
+    # --- Server-owned authorization identity fields ---
+    # Clear any client-forged values from both config sections, then write the
+    # authoritative is_internal. This runs before ALL early returns so that
+    # even user_id-is-None paths get a defined is_internal value.
+    runtime_context = config.setdefault("context", {})
+    if not isinstance(runtime_context, dict):
+        raise TypeError("run context must be a mapping")
+    for key in _SERVER_OWNED_AUTHZ_CONTEXT_KEYS:
+        runtime_context.pop(key, None)
+    configurable = config.get("configurable")
+    if isinstance(configurable, dict):
+        for key in _SERVER_OWNED_AUTHZ_CONTEXT_KEYS:
+            configurable.pop(key, None)
+    auth_source = getattr(getattr(request, "state", None), "auth_source", None)
+    runtime_context["is_internal"] = auth_source == AUTH_SOURCE_INTERNAL
+    if auth_source == AUTH_SOURCE_INTERNAL and request_context is not None:
+        channel_user_id = request_context.get("channel_user_id")
+        if channel_user_id is not None:
+            runtime_context["channel_user_id"] = channel_user_id
+
     user = getattr(request.state, "user", None)
     user_id = getattr(user, "id", None)
     if user_id is None:
@@ -704,7 +734,12 @@ async def start_run(
             # ``build_run_config``; scrub internal-only keys smuggled there.
             strip_internal_context_keys(config)
         internal_owner_user = await resolve_trusted_internal_owner_for_attribution(request, owner_user_id)
-        inject_authenticated_user_context(config, request, internal_owner_user=internal_owner_user)
+        inject_authenticated_user_context(
+            config,
+            request,
+            internal_owner_user=internal_owner_user,
+            request_context=getattr(body, "context", None),
+        )
 
         stream_modes = normalize_stream_modes(body.stream_mode)
 

backend/packages/harness/deerflow/authz/__init__.py

@@ -1,6 +1,7 @@
 """Pluggable fine-grained authorization (resource-level RBAC and beyond)."""
 
 from deerflow.authz.adapter import GuardrailAuthorizationAdapter
+from deerflow.authz.principal import build_principal_from_context, normalize_authz_attributes
 from deerflow.authz.provider import AuthorizationProvider, AuthzDecision, AuthzReason, AuthzRequest, Principal
 
 __all__ = [
@@ -10,4 +11,6 @@ __all__ = [
     "AuthorizationProvider",
     "GuardrailAuthorizationAdapter",
     "Principal",
+    "build_principal_from_context",
+    "normalize_authz_attributes",
 ]

backend/packages/harness/deerflow/authz/adapter.py

@@ -8,11 +8,17 @@ The adapter maps :class:`~deerflow.guardrails.provider.GuardrailRequest`
 fields to :class:`~deerflow.authz.provider.AuthzRequest` fields, calls the
 authorization provider, and converts the :class:`~deerflow.authz.provider.AuthzDecision`
 back to a :class:`~deerflow.guardrails.provider.GuardrailDecision`.
+
+Principal construction delegates to
+:func:`~deerflow.authz.principal.build_principal_from_context` so Layer 1
+(tool assembly) and Layer 2 (this adapter) share a single identity builder
+with consistent ``default_role`` and ``attributes`` semantics.
 """
 
 from __future__ import annotations
 
-from deerflow.authz.provider import AuthorizationProvider, AuthzDecision, AuthzRequest, Principal
+from deerflow.authz.principal import build_principal_from_context
+from deerflow.authz.provider import AuthorizationProvider, AuthzDecision, AuthzRequest
 from deerflow.guardrails.provider import GuardrailDecision, GuardrailReason, GuardrailRequest
 
 
@@ -23,13 +29,13 @@ class GuardrailAuthorizationAdapter:
     which is correct for the tool-execution path. A different resource/action
     pair can be injected if the adapter is reused outside the tool path.
 
-    .. note::
-
-        ``Principal.is_internal`` is not populated by this adapter — the
-        correct signal (``auth_source == AUTH_SOURCE_INTERNAL``) lives on
-        ``request.state``, not on :class:`GuardrailRequest`. The field
-        retains its dataclass default (``False``) until Phase 1 threads it
-        into run context, so both layers derive it from one source.
+    Args:
+        provider: The authorization provider to delegate decisions to.
+        default_role: Role used when ``user_role`` is absent or empty in the
+            runtime context. Must be passed by Phase 1B wiring from
+            ``AuthorizationConfig.default_role``.
+        resource_type: Resource type for all ``AuthzRequest`` instances.
+        action: Action for all ``AuthzRequest`` instances.
     """
 
     name = "authorization"
@@ -38,22 +44,31 @@ class GuardrailAuthorizationAdapter:
         self,
         provider: AuthorizationProvider,
         *,
+        default_role: str = "user",
         resource_type: str = "tool",
         action: str = "call",
     ) -> None:
         self._provider = provider
+        self._default_role = default_role
         self._resource_type = resource_type
         self._action = action
 
     def _to_authz(self, gr: GuardrailRequest) -> AuthzRequest:
         """Map a guardrail request to an authorization request."""
+        principal = build_principal_from_context(
+            {
+                "user_id": gr.user_id,
+                "user_role": gr.user_role,
+                "oauth_provider": gr.oauth_provider,
+                "oauth_id": gr.oauth_id,
+                "channel_user_id": gr.channel_user_id,
+                "is_internal": gr.is_internal,
+                "authz_attributes": gr.authz_attributes,
+            },
+            default_role=self._default_role,
+        )
         return AuthzRequest(
-            principal=Principal(
-                user_id=gr.user_id,
-                role=gr.user_role,
-                oauth_provider=gr.oauth_provider,
-                oauth_id=gr.oauth_id,
-            ),
+            principal=principal,
             resource=self._resource_type,
             action=self._action,
             target=gr.tool_name,

backend/packages/harness/deerflow/authz/principal.py

@@ -0,0 +1,62 @@
+"""Principal builder — the single sanctioned way to construct a Principal.
+
+Both Layer 1 (tool assembly) and Layer 2 (GuardrailAuthorizationAdapter) must
+use this builder so identity semantics stay consistent. It is a pure function:
+no global config reads, no caching, no input mutation.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+from deerflow.authz.provider import Principal
+
+
+def normalize_authz_attributes(raw: Any) -> dict[str, Any]:
+    """Validate and copy ``authz_attributes`` into a fresh dict.
+
+    This is the single normalization point shared by the Principal builder and
+    all propagation sites (middleware, executor, task_tool). Keeping it in one
+    place ensures every in-process consumption boundary raises ``TypeError``
+    for non-Mapping values rather than silently coercing.
+
+    Raises:
+        TypeError: If *raw* is not ``None`` and not a ``Mapping``.
+    """
+    if raw is None:
+        return {}
+    if isinstance(raw, Mapping):
+        return dict(raw)
+    raise TypeError(f"authz_attributes must be a Mapping, got {type(raw).__name__}")
+
+
+def build_principal_from_context(
+    context: Mapping[str, Any],
+    *,
+    default_role: str,
+) -> Principal:
+    """Build a :class:`Principal` from a runtime context mapping.
+
+    Args:
+        context: The runtime context (``config["context"]`` or a dict assembled
+            from a :class:`~deerflow.guardrails.provider.GuardrailRequest`).
+        default_role: Role used when ``user_role`` is ``None`` or empty string.
+            Unknown but non-empty roles are **not** replaced — only missing ones.
+
+    Raises:
+        TypeError: If ``authz_attributes`` is present but not a ``Mapping``.
+    """
+    resolved_role = context.get("user_role")
+    if resolved_role is None or resolved_role == "":
+        resolved_role = default_role
+
+    return Principal(
+        user_id=context.get("user_id"),
+        role=resolved_role,
+        oauth_provider=context.get("oauth_provider"),
+        oauth_id=context.get("oauth_id"),
+        channel_user_id=context.get("channel_user_id"),
+        is_internal=context.get("is_internal") is True,
+        attributes=normalize_authz_attributes(context.get("authz_attributes")),
+    )

backend/packages/harness/deerflow/authz/provider.py

@@ -28,11 +28,13 @@ from typing import Any, Protocol, runtime_checkable
 
 @dataclass
 class Principal:
-    """The actor. Built once per run from ``request.state.user`` + context.
+    """The actor resolved from trusted runtime identity context.
 
     Identity fields mirror what ``inject_authenticated_user_context``
     (``app/gateway/services.py``) already stamps into the run context, so the
-    provider sees a single resolved actor rather than re-deriving it per call.
+    provider sees one consistent identity shape. Layer 1 and the execution-time
+    guardrail adapter both use ``build_principal_from_context``; the adapter
+    rebuilds the value per request so it never caches stale runtime identity.
     """
 
     user_id: str | None = None

backend/packages/harness/deerflow/guardrails/middleware.py

@@ -12,6 +12,7 @@ from langgraph.errors import GraphBubbleUp
 from langgraph.prebuilt.tool_node import ToolCallRequest
 from langgraph.types import Command
 
+from deerflow.authz.principal import normalize_authz_attributes
 from deerflow.guardrails.provider import GuardrailDecision, GuardrailProvider, GuardrailReason, GuardrailRequest
 
 logger = logging.getLogger(__name__)
@@ -53,6 +54,9 @@ class GuardrailMiddleware(AgentMiddleware[AgentState]):
             oauth_id=context.get("oauth_id"),
             run_id=context.get("run_id"),
             tool_call_id=request.tool_call.get("id"),
+            channel_user_id=context.get("channel_user_id"),
+            is_internal=context.get("is_internal") is True,
+            authz_attributes=normalize_authz_attributes(context.get("authz_attributes")),
         )
 
     def _build_denied_message(self, request: ToolCallRequest, decision: GuardrailDecision) -> ToolMessage:

backend/packages/harness/deerflow/guardrails/provider.py

@@ -22,6 +22,12 @@ class GuardrailRequest:
     oauth_id: str | None = None
     run_id: str | None = None
     tool_call_id: str | None = None
+    # Authorization identity fields (populated by GuardrailMiddleware from
+    # runtime context). Default values ensure backward compatibility for
+    # providers that don't read them.
+    channel_user_id: str | None = None
+    is_internal: bool = False
+    authz_attributes: dict[str, Any] = field(default_factory=dict)
 
 
 @dataclass

backend/packages/harness/deerflow/sandbox/tools.py

@@ -1610,9 +1610,10 @@ CHANNEL_USER_ID_ENV = "DEERFLOW_CHANNEL_USER_ID"
 
 _CHANNEL_USER_ID_CONTEXT_KEY = "channel_user_id"
 
-# body.context is client-writable on web requests, so bound the value: real
-# platform ids are tens of chars; anything past this is hostile or corrupt and
-# must not bloat every command string sent to the sandbox.
+# Gateway accepts this identity only from internally authenticated channel
+# requests, but embedded runtimes can still construct context directly. Bound
+# the value defensively: real platform ids are tens of chars; anything past
+# this is corrupt and must not bloat every sandbox command string.
 _CHANNEL_USER_ID_MAX_LEN = 256
 
 

backend/packages/harness/deerflow/subagents/executor.py

@@ -7,7 +7,7 @@ import logging
 import os
 import threading
 import uuid
-from collections.abc import Callable, Coroutine
+from collections.abc import Callable, Coroutine, Mapping
 from concurrent.futures import Future, ThreadPoolExecutor
 from concurrent.futures import TimeoutError as FuturesTimeoutError
 from contextvars import Context, copy_context
@@ -23,6 +23,7 @@ from langchain_core.runnables import RunnableConfig
 from langgraph.errors import GraphRecursionError
 
 from deerflow.agents.thread_state import SandboxState, ThreadDataState, ThreadState
+from deerflow.authz.principal import normalize_authz_attributes
 from deerflow.config import get_app_config
 from deerflow.config.app_config import AppConfig
 from deerflow.models import create_chat_model
@@ -409,6 +410,8 @@ class SubagentExecutor:
         oauth_id: str | None = None,
         run_id: str | None = None,
         channel_user_id: str | None = None,
+        is_internal: bool = False,
+        authz_attributes: Mapping[str, Any] | None = None,
         deerflow_trace_id: str | None = None,
     ):
         """Initialize the executor.
@@ -460,6 +463,11 @@ class SubagentExecutor:
         # chats share one thread across senders, so delegated bash commands
         # must export the dispatching turn's id, not none at all.
         self.channel_user_id = channel_user_id
+        # Authorization identity propagated from the parent runtime context.
+        # is_internal is written unconditionally (including False) so the
+        # subagent's GuardrailMiddleware sees the same provenance as the lead.
+        self.is_internal = is_internal
+        self.authz_attributes = normalize_authz_attributes(authz_attributes)
         self.deerflow_trace_id = deerflow_trace_id
 
         self._base_tools = _filter_tools(
@@ -785,6 +793,10 @@ class SubagentExecutor:
             context["run_id"] = self.run_id
             if self.channel_user_id:
                 context["channel_user_id"] = self.channel_user_id
+            # Authorization identity: is_internal written unconditionally
+            # (including False); attributes copied again on write-back.
+            context["is_internal"] = self.is_internal
+            context["authz_attributes"] = dict(self.authz_attributes)
             if self.deerflow_trace_id:
                 context[DEERFLOW_TRACE_METADATA_KEY] = self.deerflow_trace_id
             context["is_subagent"] = True

backend/packages/harness/deerflow/tools/builtins/task_tool.py

@@ -12,6 +12,7 @@ from langchain_core.messages import ToolMessage
 from langgraph.config import get_stream_writer
 from langgraph.types import Command
 
+from deerflow.authz.principal import normalize_authz_attributes
 from deerflow.config import get_app_config
 from deerflow.runtime.user_context import resolve_runtime_user_id
 from deerflow.sandbox.security import LOCAL_BASH_SUBAGENT_DISABLED_MESSAGE, is_host_bash_allowed
@@ -341,6 +342,11 @@ async def task_tool(
     # IM-channel sender identity: group chats share one thread across senders,
     # so delegated bash commands need the dispatching turn's channel_user_id.
     channel_user_id = parent_context.get("channel_user_id")
+    # Propagate authorization identity: is_internal (strict bool) and
+    # authz_attributes (validated Mapping, copied). These follow the same
+    # server-side provenance as user_role/oauth — see inject_authenticated_user_context.
+    is_internal = parent_context.get("is_internal") is True
+    authz_attributes = normalize_authz_attributes(parent_context.get("authz_attributes"))
     deerflow_trace_id = normalize_trace_id(parent_context.get(DEERFLOW_TRACE_METADATA_KEY)) or normalize_trace_id(metadata.get(DEERFLOW_TRACE_METADATA_KEY)) or get_current_trace_id()
 
     parent_available_skills = metadata.get("available_skills")
@@ -386,6 +392,8 @@ async def task_tool(
         "oauth_id": oauth_id,
         "run_id": run_id,
         "channel_user_id": channel_user_id,
+        "is_internal": is_internal,
+        "authz_attributes": authz_attributes,
         "deerflow_trace_id": deerflow_trace_id,
     }
     if resolved_app_config is not None:

backend/tests/test_authorization_principal.py

@@ -0,0 +1,146 @@
+"""Tests for build_principal_from_context — the single Principal builder.
+
+This builder is the only sanctioned way to construct a Principal from runtime
+context. Both Layer 1 (tool assembly) and Layer 2 (GuardrailAuthorizationAdapter)
+must use it so identity semantics stay consistent.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from deerflow.authz.principal import build_principal_from_context
+
+
+class TestPrincipalBuilderFields:
+    """Verify all 7 Principal fields are explicitly constructed."""
+
+    def test_empty_context(self):
+        p = build_principal_from_context({}, default_role="user")
+        assert p.user_id is None
+        assert p.role == "user"
+        assert p.oauth_provider is None
+        assert p.oauth_id is None
+        assert p.channel_user_id is None
+        assert p.is_internal is False
+        assert p.attributes == {}
+
+    def test_full_field_mapping(self):
+        context = {
+            "user_id": "u1",
+            "user_role": "admin",
+            "oauth_provider": "github",
+            "oauth_id": "gh-123",
+            "channel_user_id": "ou_sender_1",
+            "is_internal": True,
+            "authz_attributes": {"department": "eng"},
+        }
+        p = build_principal_from_context(context, default_role="user")
+        assert p.user_id == "u1"
+        assert p.role == "admin"
+        assert p.oauth_provider == "github"
+        assert p.oauth_id == "gh-123"
+        assert p.channel_user_id == "ou_sender_1"
+        assert p.is_internal is True
+        assert p.attributes == {"department": "eng"}
+
+    def test_partial_context(self):
+        context = {"user_id": "u1", "user_role": "user"}
+        p = build_principal_from_context(context, default_role="admin")
+        assert p.user_id == "u1"
+        assert p.role == "user"
+        assert p.oauth_provider is None
+        assert p.oauth_id is None
+
+
+class TestRoleResolution:
+    """Role fallback rules."""
+
+    def test_none_role_uses_default(self):
+        p = build_principal_from_context({"user_role": None}, default_role="guest")
+        assert p.role == "guest"
+
+    def test_empty_string_role_uses_default(self):
+        p = build_principal_from_context({"user_role": ""}, default_role="guest")
+        assert p.role == "guest"
+
+    def test_missing_role_uses_default(self):
+        p = build_principal_from_context({}, default_role="guest")
+        assert p.role == "guest"
+
+    def test_unknown_role_preserved(self):
+        """A non-empty but unknown role must NOT fall back to default_role."""
+        p = build_principal_from_context({"user_role": "editor"}, default_role="user")
+        assert p.role == "editor"
+
+
+class TestIsInternalStrictBool:
+    """is_internal must only be True when the value is strictly True."""
+
+    @pytest.mark.parametrize(
+        "value,expected",
+        [
+            (True, True),
+            (False, False),
+            (1, False),
+            ("true", False),
+            ("1", False),
+            (None, False),
+            ([], False),
+            ({}, False),
+        ],
+    )
+    def test_is_internal_strict_bool(self, value, expected):
+        p = build_principal_from_context({"is_internal": value}, default_role="user")
+        assert p.is_internal is expected
+
+
+class TestAttributes:
+    """Attributes copy and validation semantics."""
+
+    def test_missing_attributes(self):
+        p = build_principal_from_context({}, default_role="user")
+        assert p.attributes == {}
+
+    def test_none_attributes(self):
+        p = build_principal_from_context({"authz_attributes": None}, default_role="user")
+        assert p.attributes == {}
+
+    def test_mapping_attributes_copied(self):
+        attrs = {"team": "platform"}
+        p = build_principal_from_context({"authz_attributes": attrs}, default_role="user")
+        assert p.attributes == {"team": "platform"}
+        # Mutating input after build must not affect Principal
+        attrs["team"] = "changed"
+        assert p.attributes["team"] == "platform"
+
+    def test_empty_mapping_attributes(self):
+        p = build_principal_from_context({"authz_attributes": {}}, default_role="user")
+        assert p.attributes == {}
+
+    @pytest.mark.parametrize(
+        "invalid",
+        [
+            [("key", "value")],  # list of tuples (not a Mapping at runtime)
+            "not a mapping",
+            42,
+            [1, 2, 3],
+        ],
+    )
+    def test_non_mapping_attributes_raises_type_error(self, invalid):
+        with pytest.raises(TypeError, match="authz_attributes must be a Mapping"):
+            build_principal_from_context({"authz_attributes": invalid}, default_role="user")
+
+    def test_type_error_includes_actual_type(self):
+        with pytest.raises(TypeError, match="list"):
+            build_principal_from_context({"authz_attributes": [1, 2]}, default_role="user")
+
+
+class TestPureFunction:
+    """Builder must not modify its input."""
+
+    def test_input_not_modified(self):
+        context = {"user_id": "u1", "user_role": "admin", "authz_attributes": {"k": "v"}}
+        original = dict(context)
+        build_principal_from_context(context, default_role="user")
+        assert context == original

backend/tests/test_authorization_provider.py

@@ -209,20 +209,30 @@ def _make_guardrail_request(
     tool_input: dict | None = None,
     user_id: str | None = "u1",
     user_role: str | None = "user",
+    oauth_provider: str | None = None,
+    oauth_id: str | None = None,
+    channel_user_id: str | None = None,
     thread_id: str | None = "t1",
     is_subagent: bool = False,
     agent_id: str | None = None,
     timestamp: str = "",
+    is_internal: bool = False,
+    authz_attributes: dict | None = None,
 ) -> GuardrailRequest:
     return GuardrailRequest(
         tool_name=tool_name,
         tool_input=tool_input or {},
         user_id=user_id,
         user_role=user_role,
+        oauth_provider=oauth_provider,
+        oauth_id=oauth_id,
+        channel_user_id=channel_user_id,
         thread_id=thread_id,
         is_subagent=is_subagent,
         agent_id=agent_id,
         timestamp=timestamp,
+        is_internal=is_internal,
+        authz_attributes=authz_attributes if authz_attributes is not None else {},
     )
 
 
@@ -254,8 +264,8 @@ class TestGuardrailAuthorizationAdapter:
         assert decision.allow is False
         assert decision.policy_id == "test.deny.v1"
 
-    def test_evaluate_maps_principal_identity(self):
-        """Verify user_role and user_id flow into the AuthzRequest principal."""
+    def test_evaluate_maps_complete_principal_identity(self):
+        """Every GuardrailRequest identity field flows into Principal."""
         captured: list[AuthzRequest] = []
 
         class _CapturingProvider:
@@ -272,25 +282,55 @@ class TestGuardrailAuthorizationAdapter:
                 return list(candidates)
 
         adapter = GuardrailAuthorizationAdapter(_CapturingProvider())
-        gr_req = _make_guardrail_request(user_id="user-42", user_role="admin", tool_name="write_file")
+        gr_req = _make_guardrail_request(
+            user_id="user-42",
+            user_role="admin",
+            oauth_provider="github",
+            oauth_id="gh-42",
+            channel_user_id="channel-42",
+            is_internal=True,
+            authz_attributes={"department": "engineering"},
+            tool_name="write_file",
+        )
         adapter.evaluate(gr_req)
 
         assert len(captured) == 1
         authz_req = captured[0]
         assert authz_req.principal.user_id == "user-42"
         assert authz_req.principal.role == "admin"
+        assert authz_req.principal.oauth_provider == "github"
+        assert authz_req.principal.oauth_id == "gh-42"
+        assert authz_req.principal.channel_user_id == "channel-42"
+        assert authz_req.principal.is_internal is True
+        assert authz_req.principal.attributes == {"department": "engineering"}
         assert authz_req.resource == "tool"
         assert authz_req.action == "call"
         assert authz_req.target == "write_file"
 
-    def test_evaluate_does_not_populate_is_internal_in_phase0(self):
-        """is_internal is not populated by the adapter in Phase 0.
+    def test_evaluate_maps_is_internal_true(self):
+        """is_internal=True on GuardrailRequest maps to Principal.is_internal=True."""
+        captured: list[AuthzRequest] = []
 
-        The correct signal (auth_source == AUTH_SOURCE_INTERNAL) lives on
-        request.state, not on GuardrailRequest. The adapter does not set
-        is_internal, so Principal retains its dataclass default (False).
-        Phase 1 will thread the signal into run context.
-        """
+        class _CapturingProvider:
+            name = "capturing"
+
+            def authorize(self, request: AuthzRequest) -> AuthzDecision:
+                captured.append(request)
+                return AuthzDecision(allow=True)
+
+            async def aauthorize(self, request: AuthzRequest) -> AuthzDecision:
+                return self.authorize(request)
+
+            def filter_resources(self, principal: Principal, resource_type: str, candidates: list[str]) -> list[str]:
+                return list(candidates)
+
+        adapter = GuardrailAuthorizationAdapter(_CapturingProvider())
+        gr_req = _make_guardrail_request(user_role="user", is_internal=True)
+        adapter.evaluate(gr_req)
+        assert captured[0].principal.is_internal is True
+
+    def test_evaluate_maps_is_internal_false(self):
+        """is_internal=False (default) maps to Principal.is_internal=False."""
         captured: list[AuthzRequest] = []
 
         class _CapturingProvider:
@@ -308,10 +348,100 @@ class TestGuardrailAuthorizationAdapter:
 
         adapter = GuardrailAuthorizationAdapter(_CapturingProvider())
         adapter.evaluate(_make_guardrail_request(user_role="user"))
-
-        # is_internal retains its dataclass default — adapter does not set it
         assert captured[0].principal.is_internal is False
 
+    def test_evaluate_applies_default_role_when_missing(self):
+        """Adapter uses default_role when user_role is absent."""
+        captured: list[AuthzRequest] = []
+
+        class _CapturingProvider:
+            name = "capturing"
+
+            def authorize(self, request: AuthzRequest) -> AuthzDecision:
+                captured.append(request)
+                return AuthzDecision(allow=True)
+
+            async def aauthorize(self, request: AuthzRequest) -> AuthzDecision:
+                return self.authorize(request)
+
+            def filter_resources(self, principal: Principal, resource_type: str, candidates: list[str]) -> list[str]:
+                return list(candidates)
+
+        adapter = GuardrailAuthorizationAdapter(_CapturingProvider(), default_role="guest")
+        gr_req = _make_guardrail_request(user_role=None)
+        adapter.evaluate(gr_req)
+        assert captured[0].principal.role == "guest"
+
+    def test_evaluate_preserves_unknown_role(self):
+        """Unknown non-empty role is NOT replaced by default_role."""
+        captured: list[AuthzRequest] = []
+
+        class _CapturingProvider:
+            name = "capturing"
+
+            def authorize(self, request: AuthzRequest) -> AuthzDecision:
+                captured.append(request)
+                return AuthzDecision(allow=True)
+
+            async def aauthorize(self, request: AuthzRequest) -> AuthzDecision:
+                return self.authorize(request)
+
+            def filter_resources(self, principal: Principal, resource_type: str, candidates: list[str]) -> list[str]:
+                return list(candidates)
+
+        adapter = GuardrailAuthorizationAdapter(_CapturingProvider(), default_role="user")
+        gr_req = _make_guardrail_request(user_role="editor")
+        adapter.evaluate(gr_req)
+        assert captured[0].principal.role == "editor"
+
+    def test_evaluate_sync_async_same_principal(self):
+        """Sync and async paths produce identical Principal fields."""
+        captured_sync: list[AuthzRequest] = []
+        captured_async: list[AuthzRequest] = []
+
+        class _SyncCapture:
+            name = "sync-capture"
+
+            def authorize(self, request: AuthzRequest) -> AuthzDecision:
+                captured_sync.append(request)
+                return AuthzDecision(allow=True)
+
+            async def aauthorize(self, request: AuthzRequest) -> AuthzDecision:
+                return self.authorize(request)
+
+            def filter_resources(self, principal: Principal, resource_type: str, candidates: list[str]) -> list[str]:
+                return list(candidates)
+
+        class _AsyncCapture:
+            name = "async-capture"
+
+            def authorize(self, request: AuthzRequest) -> AuthzDecision:
+                captured_async.append(request)
+                return AuthzDecision(allow=True)
+
+            async def aauthorize(self, request: AuthzRequest) -> AuthzDecision:
+                captured_async.append(request)
+                return AuthzDecision(allow=True)
+
+            def filter_resources(self, principal: Principal, resource_type: str, candidates: list[str]) -> list[str]:
+                return list(candidates)
+
+        gr_req = _make_guardrail_request(
+            user_id="user-42",
+            user_role="admin",
+            oauth_provider="github",
+            oauth_id="gh-42",
+            channel_user_id="channel-42",
+            is_internal=True,
+            authz_attributes={"department": "engineering"},
+        )
+        GuardrailAuthorizationAdapter(_SyncCapture()).evaluate(gr_req)
+        asyncio.run(GuardrailAuthorizationAdapter(_AsyncCapture()).aevaluate(gr_req))
+
+        p1 = captured_sync[0].principal
+        p2 = captured_async[0].principal
+        assert p1 == p2
+
     def test_evaluate_maps_context_fields(self):
         """Verify thread_id, tool_input, and is_subagent flow into AuthzRequest.context."""
         captured: list[AuthzRequest] = []

backend/tests/test_channel_user_id_env.py

@@ -1,9 +1,8 @@
 """Tests for exposing the IM-channel platform user id to sandbox commands (#3914).
 
 Two halves:
-- Gateway: ``merge_run_context_overrides`` forwards ``channel_user_id`` from
-  ``body.context`` into ``config['context']`` (runtime context) only — never
-  into ``configurable`` (which is checkpointed).
+- Gateway: only an internally authenticated caller's top-level ``body.context``
+  may supply ``channel_user_id``; free-form RunnableConfig values are cleared.
 - Sandbox: ``bash_tool`` exposes the id as the fixed env var
   ``DEERFLOW_CHANNEL_USER_ID`` via an ``export`` prefix on the command string.
   It must NOT ride the ``env=`` parameter: on ``AioSandbox`` a non-empty env
@@ -52,33 +51,51 @@ def _run_bash(monkeypatch, runtime, command: str = "echo hi") -> _CapturingSandb
     return sandbox
 
 
-class TestMergeRunContextOverridesChannelUserId:
-    def test_channel_user_id_propagates_to_runtime_context_only(self):
-        from app.gateway.services import build_run_config, merge_run_context_overrides
+class TestGatewayChannelUserIdTrustBoundary:
+    @staticmethod
+    def _request(auth_source: str):
+        return SimpleNamespace(
+            state=SimpleNamespace(
+                auth_source=auth_source,
+                user=SimpleNamespace(id="u1", system_role="user"),
+            )
+        )
+
+    def test_internal_channel_user_id_propagates_to_runtime_context_only(self):
+        from app.gateway.services import build_run_config, inject_authenticated_user_context
 
         config = build_run_config("thread-1", None, None)
-        merge_run_context_overrides(config, {"channel_user_id": "ou_feishu_123"})
+        inject_authenticated_user_context(
+            config,
+            self._request("internal"),
+            request_context={"channel_user_id": "ou_feishu_123"},
+        )
 
         assert config["context"]["channel_user_id"] == "ou_feishu_123"
         # Never into configurable: that mapping is checkpointed with the thread.
         assert "channel_user_id" not in config["configurable"]
 
-    def test_existing_runtime_context_value_wins(self):
-        """setdefault semantics: a server-side value stamped earlier must not be
-        overridden by the client-supplied body.context."""
-        from app.gateway.services import build_run_config, merge_run_context_overrides
+    def test_free_form_config_value_cannot_override_internal_sender(self):
+        from app.gateway.services import build_run_config, inject_authenticated_user_context
 
-        config = build_run_config("thread-1", None, None)
-        config.setdefault("context", {})["channel_user_id"] = "server-stamped"
-        merge_run_context_overrides(config, {"channel_user_id": "client-supplied"})
+        config = build_run_config(
+            "thread-1",
+            {"context": {"channel_user_id": "forged-config-sender"}},
+            None,
+        )
+        inject_authenticated_user_context(
+            config,
+            self._request("internal"),
+            request_context={"channel_user_id": "trusted-im-sender"},
+        )
 
-        assert config["context"]["channel_user_id"] == "server-stamped"
+        assert config["context"]["channel_user_id"] == "trusted-im-sender"
 
     def test_absent_channel_user_id_adds_nothing(self):
-        from app.gateway.services import build_run_config, merge_run_context_overrides
+        from app.gateway.services import build_run_config, inject_authenticated_user_context
 
         config = build_run_config("thread-1", None, None)
-        merge_run_context_overrides(config, {"model_name": "gpt"})
+        inject_authenticated_user_context(config, self._request("internal"), request_context={"model_name": "gpt"})
 
         assert "channel_user_id" not in config.get("context", {})
 
@@ -191,6 +208,7 @@ class TestBashToolChannelIdentityPrefix:
         monkeypatch.setattr("deerflow.sandbox.tools.ensure_sandbox_initialized", lambda runtime: sandbox)
         monkeypatch.setattr("deerflow.sandbox.tools.ensure_thread_directories_exist", lambda runtime: None)
         monkeypatch.setattr("deerflow.sandbox.tools.is_host_bash_allowed", lambda: True)
+        monkeypatch.setattr("deerflow.sandbox.tools._is_windows", lambda: False)
 
         bash_tool.func(runtime=runtime, description="test", command="echo hi")
 

backend/tests/test_gateway_services.py

@@ -6,6 +6,7 @@ import json
 
 import pytest
 
+from app.gateway.auth_disabled import AUTH_SOURCE_INTERNAL
 from deerflow.config.app_config import AppConfig, reset_app_config, set_app_config
 
 
@@ -1062,6 +1063,7 @@ def test_start_run_uses_internal_owner_header_for_persistence(_stub_app_config):
     from langgraph.checkpoint.memory import InMemorySaver
     from langgraph.store.memory import InMemoryStore
 
+    from app.gateway.auth_disabled import AUTH_SOURCE_INTERNAL
     from app.gateway.internal_auth import INTERNAL_OWNER_USER_ID_HEADER_NAME, INTERNAL_SYSTEM_ROLE
     from app.gateway.services import start_run
     from deerflow.persistence.thread_meta.memory import MemoryThreadMetaStore
@@ -1085,7 +1087,10 @@ def test_start_run_uses_internal_owner_header_for_persistence(_stub_app_config):
         )
         request = SimpleNamespace(
             headers={INTERNAL_OWNER_USER_ID_HEADER_NAME: "owner-1"},
-            state=SimpleNamespace(user=SimpleNamespace(id="default", system_role=INTERNAL_SYSTEM_ROLE)),
+            state=SimpleNamespace(
+                auth_source=AUTH_SOURCE_INTERNAL,
+                user=SimpleNamespace(id="default", system_role=INTERNAL_SYSTEM_ROLE),
+            ),
             app=SimpleNamespace(state=state),
         )
         body = SimpleNamespace(
@@ -1139,6 +1144,7 @@ def test_start_run_stamps_internal_owner_guardrail_attribution(_stub_app_config)
     from langgraph.checkpoint.memory import InMemorySaver
     from langgraph.store.memory import InMemoryStore
 
+    from app.gateway.auth_disabled import AUTH_SOURCE_INTERNAL
     from app.gateway.internal_auth import INTERNAL_OWNER_USER_ID_HEADER_NAME, INTERNAL_SYSTEM_ROLE
     from app.gateway.services import start_run
     from deerflow.persistence.thread_meta.memory import MemoryThreadMetaStore
@@ -1170,7 +1176,10 @@ def test_start_run_stamps_internal_owner_guardrail_attribution(_stub_app_config)
         )
         request = SimpleNamespace(
             headers={INTERNAL_OWNER_USER_ID_HEADER_NAME: "owner-1"},
-            state=SimpleNamespace(user=SimpleNamespace(id="default", system_role=INTERNAL_SYSTEM_ROLE)),
+            state=SimpleNamespace(
+                auth_source=AUTH_SOURCE_INTERNAL,
+                user=SimpleNamespace(id="default", system_role=INTERNAL_SYSTEM_ROLE),
+            ),
             app=SimpleNamespace(state=state),
         )
         body = SimpleNamespace(
@@ -1182,9 +1191,10 @@ def test_start_run_stamps_internal_owner_guardrail_attribution(_stub_app_config)
                     "user_role": "admin",
                     "oauth_provider": "spoofed-provider",
                     "oauth_id": "spoofed-subject",
+                    "channel_user_id": "forged-config-sender",
                 }
             },
-            context={"user_id": "spoofed-client"},
+            context={"user_id": "spoofed-client", "channel_user_id": "trusted-im-sender"},
             on_disconnect="cancel",
             multitask_strategy="reject",
             stream_mode=None,
@@ -1213,6 +1223,92 @@ def test_start_run_stamps_internal_owner_guardrail_attribution(_stub_app_config)
     assert context["user_role"] == "user"
     assert context["oauth_provider"] == "keycloak"
     assert context["oauth_id"] == "subject-123"
+    assert context["channel_user_id"] == "trusted-im-sender"
+    assert context["is_internal"] is True
+
+
+def test_start_run_session_caller_anti_forgery(_stub_app_config):
+    """A session (non-internal) caller cannot forge is_internal, authz_attributes,
+    or channel_user_id via body.config. Exercises the real start_run path, not
+    a replay, so ordering or gating drift would be caught."""
+    import asyncio
+    from types import SimpleNamespace
+    from unittest.mock import patch
+
+    from langgraph.checkpoint.memory import InMemorySaver
+    from langgraph.store.memory import InMemoryStore
+
+    from app.gateway.services import start_run
+    from deerflow.persistence.thread_meta.memory import MemoryThreadMetaStore
+    from deerflow.runtime import RunManager
+    from deerflow.runtime.runs.store.memory import MemoryRunStore
+
+    async def _scenario():
+        thread_store = MemoryThreadMetaStore(InMemoryStore())
+        await thread_store.create("thread-session-authz", user_id="u1", metadata={})
+        run_manager = RunManager(store=MemoryRunStore())
+        state = SimpleNamespace(
+            stream_bridge=SimpleNamespace(),
+            run_manager=run_manager,
+            checkpointer=InMemorySaver(),
+            store=InMemoryStore(),
+            run_event_store=SimpleNamespace(),
+            run_events_config=None,
+            thread_store=thread_store,
+        )
+        request = SimpleNamespace(
+            headers={},
+            state=SimpleNamespace(
+                auth_source="session",
+                user=SimpleNamespace(id="u1", system_role="user"),
+            ),
+            app=SimpleNamespace(state=state),
+        )
+        body = SimpleNamespace(
+            assistant_id="lead_agent",
+            input={"messages": [{"role": "human", "content": "hi"}]},
+            metadata={},
+            config={
+                "context": {
+                    "is_internal": True,
+                    "authz_attributes": {"forged": True},
+                    "channel_user_id": "forged-sender",
+                },
+                "configurable": {
+                    "is_internal": True,
+                    "authz_attributes": {"forged": True},
+                },
+            },
+            context=None,
+            on_disconnect="cancel",
+            multitask_strategy="reject",
+            stream_mode=None,
+            stream_subgraphs=False,
+            interrupt_before=None,
+            interrupt_after=None,
+        )
+        captured_context: dict[str, object] = {}
+
+        async def fake_run_agent(*args, **kwargs):
+            captured_context.update(kwargs["config"]["context"])
+
+        with (
+            patch("app.gateway.services.resolve_agent_factory", return_value=object()),
+            patch("app.gateway.services.run_agent", side_effect=fake_run_agent),
+        ):
+            record = await start_run(body, "thread-session-authz", request)
+            await record.task
+
+        return captured_context
+
+    context = asyncio.run(_scenario())
+
+    # is_internal must be False (server-derived from auth_source="session")
+    assert context["is_internal"] is False
+    # authz_attributes must be stripped (no Gateway-side producer)
+    assert "authz_attributes" not in context
+    # channel_user_id must not survive from body.config for a session caller
+    assert context.get("channel_user_id") is None
 
 
 def test_launch_scheduled_thread_run_marks_context_non_interactive(_stub_app_config):
@@ -1380,3 +1476,160 @@ def test_strip_internal_context_keys_scrubs_config_smuggled_non_interactive():
     via_configurable = build_run_config("thread-1", {"configurable": {"non_interactive": True}}, None)
     strip_internal_context_keys(via_configurable)
     assert "non_interactive" not in via_configurable["configurable"]
+
+
+# --- Authorization identity anti-forgery tests ---
+
+
+def _make_request_with_auth_source(auth_source: str | None, *, user_id="u1", system_role="user"):
+    """Build a minimal fake request with the given auth_source."""
+    from types import SimpleNamespace
+
+    return SimpleNamespace(
+        state=SimpleNamespace(
+            auth_source=auth_source,
+            user=SimpleNamespace(id=user_id, system_role=system_role) if user_id else None,
+        ),
+    )
+
+
+def _assemble_authz_run_config(request_config: dict, request, *, body_context: dict | None = None):
+    """Replay the real start_run config sequence for authz identity tests."""
+    from app.gateway.services import (
+        build_run_config,
+        inject_authenticated_user_context,
+        merge_run_context_overrides,
+        strip_internal_context_keys,
+    )
+
+    is_internal = request.state.auth_source == AUTH_SOURCE_INTERNAL
+    config = build_run_config("thread-authz", request_config, None)
+    merge_run_context_overrides(config, body_context, internal=is_internal)
+    if not is_internal:
+        strip_internal_context_keys(config)
+    inject_authenticated_user_context(config, request, request_context=body_context)
+    return config
+
+
+class TestInjectAuthenticatedUserContextAuthz:
+    """Verify is_internal and authz_attributes anti-forgery in inject_authenticated_user_context."""
+
+    def test_clears_forged_is_internal_from_context_section(self):
+        """Client forges is_internal=True via body.config['context'] → must be cleared."""
+        request = _make_request_with_auth_source("session")
+        config = _assemble_authz_run_config({"context": {"is_internal": True}}, request)
+        # The forged value must be replaced by the server-side value (False for session)
+        assert config["context"]["is_internal"] is False
+
+    def test_clears_forged_is_internal_from_configurable_section(self):
+        """Client forges is_internal=True via body.config['configurable'] → must be cleared."""
+        request = _make_request_with_auth_source("session")
+        config = _assemble_authz_run_config({"configurable": {"is_internal": True}}, request)
+        assert "is_internal" not in config["configurable"]
+        assert config["context"]["is_internal"] is False
+
+    def test_clears_forged_authz_attributes_from_context_section(self):
+        """Client forges authz_attributes via body.config['context'] → must be cleared."""
+        request = _make_request_with_auth_source("session")
+        config = _assemble_authz_run_config(
+            {"context": {"authz_attributes": [("forged", True)]}},
+            request,
+        )
+        assert "authz_attributes" not in config["context"]
+
+    def test_clears_forged_authz_attributes_from_configurable_section(self):
+        """Client forges authz_attributes via body.config['configurable'] → must be cleared."""
+        request = _make_request_with_auth_source("session")
+        config = _assemble_authz_run_config({"configurable": {"authz_attributes": {"forged": True}}}, request)
+        assert "authz_attributes" not in config["configurable"]
+
+    def test_internal_auth_source_writes_is_internal_true(self):
+        """Internal caller gets is_internal=True."""
+        from app.gateway.services import inject_authenticated_user_context
+
+        config = {"context": {}, "configurable": {}}
+        inject_authenticated_user_context(config, _make_request_with_auth_source(AUTH_SOURCE_INTERNAL))
+        assert config["context"]["is_internal"] is True
+
+    def test_session_auth_source_writes_is_internal_false(self):
+        """Session caller gets is_internal=False."""
+        from app.gateway.services import inject_authenticated_user_context
+
+        config = {"context": {}, "configurable": {}}
+        inject_authenticated_user_context(config, _make_request_with_auth_source("session"))
+        assert config["context"]["is_internal"] is False
+
+    def test_user_none_still_writes_is_internal(self):
+        """Even when user_id is None (early return path), is_internal is written."""
+        from app.gateway.services import inject_authenticated_user_context
+
+        config = {"context": {}, "configurable": {}}
+        # user=None triggers the first early return, but is_internal must still be set
+        inject_authenticated_user_context(config, _make_request_with_auth_source("session", user_id=None))
+        assert config["context"]["is_internal"] is False
+
+    def test_user_none_internal_source_writes_true(self):
+        """When auth_source is internal but user is None, is_internal is still True."""
+        from app.gateway.services import inject_authenticated_user_context
+
+        config = {"context": {}, "configurable": {}}
+        inject_authenticated_user_context(config, _make_request_with_auth_source(AUTH_SOURCE_INTERNAL, user_id=None))
+        assert config["context"]["is_internal"] is True
+
+    def test_internal_caller_attributes_also_cleared(self):
+        """Even internal callers can't forge authz_attributes."""
+        request = _make_request_with_auth_source(AUTH_SOURCE_INTERNAL)
+        config = _assemble_authz_run_config({"context": {"authz_attributes": {"forged": True}}}, request)
+        assert "authz_attributes" not in config["context"]
+        assert config["context"]["is_internal"] is True
+
+    def test_session_body_context_cannot_inject_channel_user_id(self):
+        request = _make_request_with_auth_source("session")
+        config = _assemble_authz_run_config(
+            {},
+            request,
+            body_context={"channel_user_id": "forged-sender"},
+        )
+        assert "channel_user_id" not in config["context"]
+
+    def test_session_config_sections_cannot_inject_channel_user_id(self):
+        request = _make_request_with_auth_source("session")
+        config = _assemble_authz_run_config(
+            {
+                "context": {"channel_user_id": "forged-context-sender"},
+                "configurable": {"channel_user_id": "forged-configurable-sender"},
+            },
+            request,
+        )
+        assert "channel_user_id" not in config["context"]
+        assert "channel_user_id" not in config["configurable"]
+
+    def test_internal_body_context_preserves_channel_user_id(self):
+        request = _make_request_with_auth_source(AUTH_SOURCE_INTERNAL)
+        config = _assemble_authz_run_config(
+            {},
+            request,
+            body_context={"channel_user_id": "trusted-im-sender"},
+        )
+        assert config["context"]["channel_user_id"] == "trusted-im-sender"
+
+    def test_internal_config_sections_cannot_override_channel_user_id(self):
+        request = _make_request_with_auth_source(AUTH_SOURCE_INTERNAL)
+        config = _assemble_authz_run_config(
+            {
+                "context": {"channel_user_id": "forged-context-sender"},
+                "configurable": {"channel_user_id": "forged-configurable-sender"},
+            },
+            request,
+            body_context={"channel_user_id": "trusted-im-sender"},
+        )
+        assert config["context"]["channel_user_id"] == "trusted-im-sender"
+        assert "channel_user_id" not in config["configurable"]
+
+    def test_non_dict_context_raises_type_error(self):
+        """Non-dict runtime context must raise TypeError, not silently skip."""
+        from app.gateway.services import inject_authenticated_user_context
+
+        config = {"context": "not a dict"}
+        with pytest.raises(TypeError, match="run context must be a mapping"):
+            inject_authenticated_user_context(config, _make_request_with_auth_source("session"))

backend/tests/test_guardrail_middleware.py

@@ -544,6 +544,9 @@ class TestGuardrailRequestAttribution:
         assert guardrail_request.oauth_id is None
         assert guardrail_request.run_id is None
         assert guardrail_request.tool_call_id is None
+        assert guardrail_request.channel_user_id is None
+        assert guardrail_request.is_internal is False
+        assert guardrail_request.authz_attributes == {}
 
     def test_only_user_id_present(self):
         runtime = self._make_runtime_mock(context={"user_id": "user_abc"})
@@ -559,12 +562,16 @@ class TestGuardrailRequestAttribution:
         assert guardrail_request.tool_call_id is None
 
     def test_authenticated_user_context_present(self):
+        attributes = {"department": "engineering"}
         runtime = self._make_runtime_mock(
             context={
                 "user_id": "user_abc",
                 "user_role": "admin",
                 "oauth_provider": "github",
                 "oauth_id": "gh_123",
+                "channel_user_id": "channel_123",
+                "is_internal": True,
+                "authz_attributes": attributes,
             }
         )
         req = self._make_request(runtime=runtime, tool_call={"name": "bash", "args": {}})
@@ -575,6 +582,19 @@ class TestGuardrailRequestAttribution:
         assert guardrail_request.user_role == "admin"
         assert guardrail_request.oauth_provider == "github"
         assert guardrail_request.oauth_id == "gh_123"
+        assert guardrail_request.channel_user_id == "channel_123"
+        assert guardrail_request.is_internal is True
+        assert guardrail_request.authz_attributes == {"department": "engineering"}
+
+        attributes["department"] = "changed"
+        assert guardrail_request.authz_attributes == {"department": "engineering"}
+
+    def test_non_mapping_authz_attributes_raise_type_error(self):
+        runtime = self._make_runtime_mock(context={"authz_attributes": ["not", "a", "mapping"]})
+        req = self._make_request(runtime=runtime, tool_call={"name": "bash", "args": {}})
+
+        with pytest.raises(TypeError, match="authz_attributes must be a Mapping"):
+            self._capture_guardrail_request(req)
 
     def test_only_run_id_present(self):
         runtime = self._make_runtime_mock(context={"run_id": "run_xyz"})

backend/tests/test_subagent_executor.py

@@ -3133,3 +3133,121 @@ class TestSubagentGuardrailAttribution:
         from langchain_core.messages import HumanMessage
 
         return ({"messages": [HumanMessage(content=task)]}, [], None)
+
+    @pytest.mark.anyio
+    async def test_aexecute_writes_is_internal_true(
+        self,
+        classes,
+        monkeypatch,
+    ):
+        """is_internal=True must propagate to subagent context."""
+        SubagentExecutor = classes["SubagentExecutor"]
+        SubagentConfig = classes["SubagentConfig"]
+        executor = SubagentExecutor(
+            config=SubagentConfig(
+                name="general-purpose",
+                description="is_internal test",
+                system_prompt="test",
+                max_turns=5,
+                timeout_seconds=30,
+            ),
+            tools=[],
+            thread_id="t1",
+            is_internal=True,
+        )
+        fake_agent = _FakeStreamAgent()
+        monkeypatch.setattr(executor, "_build_initial_state", self._noop_build_initial_state)
+        monkeypatch.setattr(executor, "_create_agent", lambda *a, **kw: fake_agent)
+
+        await executor._aexecute("do something")
+
+        context = fake_agent.captured_context
+        assert context is not None
+        assert context.get("is_internal") is True
+
+    @pytest.mark.anyio
+    async def test_aexecute_writes_is_internal_false(
+        self,
+        classes,
+        monkeypatch,
+    ):
+        """is_internal=False must be written explicitly, not omitted."""
+        SubagentExecutor = classes["SubagentExecutor"]
+        SubagentConfig = classes["SubagentConfig"]
+        executor = SubagentExecutor(
+            config=SubagentConfig(
+                name="general-purpose",
+                description="is_internal false test",
+                system_prompt="test",
+                max_turns=5,
+                timeout_seconds=30,
+            ),
+            tools=[],
+            thread_id="t1",
+            is_internal=False,
+        )
+        fake_agent = _FakeStreamAgent()
+        monkeypatch.setattr(executor, "_build_initial_state", self._noop_build_initial_state)
+        monkeypatch.setattr(executor, "_create_agent", lambda *a, **kw: fake_agent)
+
+        await executor._aexecute("do something")
+
+        context = fake_agent.captured_context
+        assert context is not None
+        assert context.get("is_internal") is False
+
+    @pytest.mark.anyio
+    async def test_aexecute_copies_attributes_on_writeback(
+        self,
+        classes,
+        monkeypatch,
+    ):
+        """authz_attributes must be copied on write-back; mutating context copy
+        doesn't affect executor's internal copy."""
+        SubagentExecutor = classes["SubagentExecutor"]
+        SubagentConfig = classes["SubagentConfig"]
+        source_attributes = {"dept": "eng"}
+        executor = SubagentExecutor(
+            config=SubagentConfig(
+                name="general-purpose",
+                description="attributes copy test",
+                system_prompt="test",
+                max_turns=5,
+                timeout_seconds=30,
+            ),
+            tools=[],
+            thread_id="t1",
+            authz_attributes=source_attributes,
+        )
+        source_attributes["dept"] = "changed-before-run"
+        assert executor.authz_attributes == {"dept": "eng"}
+        fake_agent = _FakeStreamAgent()
+        monkeypatch.setattr(executor, "_build_initial_state", self._noop_build_initial_state)
+        monkeypatch.setattr(executor, "_create_agent", lambda *a, **kw: fake_agent)
+
+        await executor._aexecute("do something")
+
+        context = fake_agent.captured_context
+        assert context is not None
+        assert context.get("authz_attributes") == {"dept": "eng"}
+        # Mutate the context copy
+        context["authz_attributes"]["dept"] = "changed"
+        # Executor's internal copy should be unaffected
+        assert executor.authz_attributes["dept"] == "eng"
+
+    def test_executor_rejects_non_mapping_attributes(self, classes):
+        """Constructor must raise TypeError for non-Mapping authz_attributes."""
+        SubagentExecutor = classes["SubagentExecutor"]
+        SubagentConfig = classes["SubagentConfig"]
+        with pytest.raises(TypeError, match="authz_attributes must be a Mapping"):
+            SubagentExecutor(
+                config=SubagentConfig(
+                    name="general-purpose",
+                    description="test",
+                    system_prompt="test",
+                    max_turns=5,
+                    timeout_seconds=30,
+                ),
+                tools=[],
+                authz_attributes=["not", "a", "mapping"],
+            )

backend/tests/test_task_tool_core_logic.py

@@ -300,6 +300,118 @@ def test_task_tool_forwards_channel_user_id_to_executor(monkeypatch):
     assert captured["executor_kwargs"]["channel_user_id"] == "ou_group_sender_1"
 
 
+def test_task_tool_forwards_is_internal_true_to_executor(monkeypatch):
+    """is_internal=True must propagate to SubagentExecutor."""
+    runtime = _make_runtime()
+    runtime.context["is_internal"] = True
+    captured = {}
+
+    class DummyExecutor:
+        def __init__(self, **kwargs):
+            captured["executor_kwargs"] = kwargs
+
+        def execute_async(self, prompt, task_id=None):
+            return task_id or "generated-task-id"
+
+    monkeypatch.setattr(task_tool_module, "SubagentStatus", FakeSubagentStatus)
+    monkeypatch.setattr(task_tool_module, "SubagentExecutor", DummyExecutor)
+    monkeypatch.setattr(task_tool_module, "get_subagent_config", lambda _: _make_subagent_config())
+    monkeypatch.setattr(
+        task_tool_module,
+        "get_background_task_result",
+        lambda _: _make_result(FakeSubagentStatus.COMPLETED, result="done"),
+    )
+    monkeypatch.setattr(task_tool_module, "get_stream_writer", lambda: lambda _event: None)
+    monkeypatch.setattr(task_tool_module.asyncio, "sleep", _no_sleep)
+    monkeypatch.setattr("deerflow.tools.get_available_tools", lambda **kwargs: [])
+
+    _run_task_tool(runtime=runtime, description="test", prompt="p", subagent_type="general-purpose", tool_call_id="tc-1")
+    assert captured["executor_kwargs"]["is_internal"] is True
+
+
+def test_task_tool_forwards_is_internal_false_to_executor(monkeypatch):
+    """is_internal=False must also propagate explicitly (not skipped)."""
+    runtime = _make_runtime()
+    runtime.context["is_internal"] = False
+    captured = {}
+
+    class DummyExecutor:
+        def __init__(self, **kwargs):
+            captured["executor_kwargs"] = kwargs
+
+        def execute_async(self, prompt, task_id=None):
+            return task_id or "generated-task-id"
+
+    monkeypatch.setattr(task_tool_module, "SubagentStatus", FakeSubagentStatus)
+    monkeypatch.setattr(task_tool_module, "SubagentExecutor", DummyExecutor)
+    monkeypatch.setattr(task_tool_module, "get_subagent_config", lambda _: _make_subagent_config())
+    monkeypatch.setattr(
+        task_tool_module,
+        "get_background_task_result",
+        lambda _: _make_result(FakeSubagentStatus.COMPLETED, result="done"),
+    )
+    monkeypatch.setattr(task_tool_module, "get_stream_writer", lambda: lambda _event: None)
+    monkeypatch.setattr(task_tool_module.asyncio, "sleep", _no_sleep)
+    monkeypatch.setattr("deerflow.tools.get_available_tools", lambda **kwargs: [])
+
+    _run_task_tool(runtime=runtime, description="test", prompt="p", subagent_type="general-purpose", tool_call_id="tc-1")
+    assert captured["executor_kwargs"]["is_internal"] is False
+
+
+def test_task_tool_copies_attributes_to_executor(monkeypatch):
+    """Mapping authz_attributes must be copied; mutating parent doesn't affect executor."""
+    runtime = _make_runtime()
+    runtime.context["authz_attributes"] = {"dept": "eng"}
+    captured = {}
+
+    class DummyExecutor:
+        def __init__(self, **kwargs):
+            captured["executor_kwargs"] = kwargs
+
+        def execute_async(self, prompt, task_id=None):
+            return task_id or "generated-task-id"
+
+    monkeypatch.setattr(task_tool_module, "SubagentStatus", FakeSubagentStatus)
+    monkeypatch.setattr(task_tool_module, "SubagentExecutor", DummyExecutor)
+    monkeypatch.setattr(task_tool_module, "get_subagent_config", lambda _: _make_subagent_config())
+    monkeypatch.setattr(
+        task_tool_module,
+        "get_background_task_result",
+        lambda _: _make_result(FakeSubagentStatus.COMPLETED, result="done"),
+    )
+    monkeypatch.setattr(task_tool_module, "get_stream_writer", lambda: lambda _event: None)
+    monkeypatch.setattr(task_tool_module.asyncio, "sleep", _no_sleep)
+    monkeypatch.setattr("deerflow.tools.get_available_tools", lambda **kwargs: [])
+
+    _run_task_tool(runtime=runtime, description="test", prompt="p", subagent_type="general-purpose", tool_call_id="tc-1")
+    executor_attrs = captured["executor_kwargs"]["authz_attributes"]
+    assert executor_attrs == {"dept": "eng"}
+    # Mutate the executor's copy; original context should not change
+    executor_attrs["dept"] = "changed"
+    assert runtime.context["authz_attributes"]["dept"] == "eng"
+
+
+def test_task_tool_rejects_non_mapping_attributes(monkeypatch):
+    """Non-Mapping authz_attributes must raise TypeError, not silently become {}."""
+
+    class DummyExecutor:
+        def __init__(self, **kwargs):
+            pass
+
+        def execute_async(self, prompt, task_id=None):
+            return task_id or "generated-task-id"
+
+    monkeypatch.setattr(task_tool_module, "SubagentStatus", FakeSubagentStatus)
+    monkeypatch.setattr(task_tool_module, "SubagentExecutor", DummyExecutor)
+    monkeypatch.setattr(task_tool_module, "get_subagent_config", lambda _: _make_subagent_config())
+    monkeypatch.setattr("deerflow.tools.get_available_tools", lambda **kwargs: [])
+
+    runtime = _make_runtime()
+    runtime.context["authz_attributes"] = ["not", "a", "mapping"]
+    with pytest.raises(TypeError, match="authz_attributes must be a Mapping"):
+        _run_task_tool(runtime=runtime, description="test", prompt="p", subagent_type="general-purpose", tool_call_id="tc-1")
+
+
 def test_task_tool_rejects_bash_subagent_when_host_bash_disabled(monkeypatch):
     monkeypatch.setattr(task_tool_module, "get_subagent_config", lambda _: _make_subagent_config())
     monkeypatch.setattr(task_tool_module, "is_host_bash_allowed", lambda: False)

docs/plans/2026-07-10-pluggable-authorization-implementation-notes.md

@@ -0,0 +1,223 @@
+# 可插拔授权系统实施记录
+
+<!-- Language: Chinese. This document is the cumulative handoff record for the
+     pluggable authorization system (RFC #4063). An English summary is provided
+     below for navigation; the detailed content is in Chinese. -->
+
+> **English summary:** This is the cumulative implementation log for the
+> pluggable authorization RFC ([#4063](https://github.com/bytedance/deer-flow/issues/4063)).
+> It records merged contracts, reviewer-confirmed decisions, and required
+> regression coverage for each phase. Sections:
+> - **每个 RFC PR 的必读要求** — Pre-PR checklist for every authorization change
+> - **信息优先级** — Information precedence when sources conflict
+> - **Phase 0：已合并基线** — Phase 0 merged baseline (PR #4127)
+> - **PR #4127 多轮修改的原因** — Root causes of multi-round review iterations
+> - **所有后续阶段必须保持的不变量** — Invariants all phases must preserve
+> - **Phase 1 实施前确认** — Phase 1 pre-implementation confirmations
+> - **每次更新 PR 前的固定清单** — Fixed checklist before each PR update
+> - **决策日志** — Append-only decision log
+> - **当前连续性风险** — Current continuity risks
+
+本文档是可插拔授权 RFC（[#4063](https://github.com/bytedance/deer-flow/issues/4063)）
+的持续实施记忆。它用于补充设计 RFC，记录已经实际合并的内容、review 中确认的契约，
+以及每个后续 PR 必须验证的事项。
+
+## 每个 RFC PR 的必读要求
+
+修改任何后续阶段前，必须阅读：
+
+1. [设计 RFC](2026-07-10-pluggable-authorization-rfc.md)。
+2. 本实施记录。
+3. 前一阶段已经合并的代码和测试。如果它们与旧 RFC 示例不一致，以已合并契约为准。
+
+每个 PR 描述中必须复制并确认以下内容：
+
+```markdown
+## Authorization RFC 连续性确认
+
+- [ ] 已阅读 `docs/plans/2026-07-10-pluggable-authorization-rfc.md`。
+- [ ] 已阅读 `docs/plans/2026-07-10-pluggable-authorization-implementation-notes.md`。
+- [ ] 已核对所有前置阶段的决策和延期事项。
+- [ ] 已用本 PR 的新决策和后续事项更新实施记录。
+```
+
+## 信息优先级
+
+不同来源发生冲突时，按以下顺序判断：
+
+1. 已合并代码和回归测试。
+2. 已接受的 review 决策和最终合并的 PR 描述。
+3. 本实施记录。
+4. 设计 RFC 中较早的示例或阶段划分。
+
+不能静默修改或重新解释冲突。必须写入决策日志；涉及架构或安全行为时，还要在
+issue #4063 中确认。
+
+## Phase 0：已合并基线
+
+PR [#4127](https://github.com/bytedance/deer-flow/pull/4127) 于 2026-07-15
+以提交 `1300c6d3` 合并，确立了以下契约：
+
+- `AuthorizationProvider` 是可在运行时检查的 Protocol，包含同步授权、异步授权和
+  `filter_resources`。
+- `filter_resources` 是必需方法。Protocol 中方法体为 `...` 不代表存在默认实现。
+  没有静态映射的 provider 必须自行实现逐项授权。
+- `GuardrailAuthorizationAdapter` 有意让 provider 异常向上传播。
+  `GuardrailMiddleware` 统一负责异常处理、审计以及 fail-open/fail-closed 执行。
+- 不能通过 `user_role == "internal"` 推导 `Principal.is_internal`。权威信号是内部
+  认证状态 `auth_source`，必须从 Gateway 上下文传递。
+- Adapter 上下文保留 `thread_id`、`run_id`、`tool_call_id`、`tool_input`、
+  `is_subagent`、`agent_id` 和 `timestamp`。
+- `AuthorizationConfig` 已加入 AppConfig，默认 `enabled: false`，并参与 singleton
+  加载；Phase 0 尚无运行时代码读取它。
+- Adapter 在结构上符合 `GuardrailProvider`；同步和异步异常传播均由测试固定。
+
+以下原 RFC Phase 0 项目没有在 PR #4127 中落地，仍属于后续工作：Principal 构建器、
+内置 RBAC provider、Layer 1 过滤、Layer 2 自动装配和内部 Principal 填充。
+
+## PR #4127 多轮修改的原因
+
+实现方向获得认可，但首版没有完整验证后续阶段将继承的关键契约：
+
+- 直接沿用了 RFC 关于 `filter_resources` 默认行为的假设，没有先验证 Python
+  Protocol 的真实语义。willem-bd 和 zhfeng **独立**指出了同一个问题——多个 reviewer
+  从不同角度指向同一处，说明该缺陷在 review 中非常显眼。后续阶段中如果再次出现
+  多人独立指出同一处，应当视为最高优先级，不再需要多方确认。
+- 身份字段按照表面数据结构映射，没有追踪到运行时权威来源。
+- 配置测试验证了 Pydantic 对象构建，但最初绕过了真实 singleton 加载生命周期。
+- schema 变化最初遗漏 `config_version`；同时主线发生版本竞争，需要 rebase 后重新
+  选择版本号。
+- Helm 中的配置版本镜像和仓库内 RFC 文档较晚才在 review 中被发现。
+- 代码、测试、注释和 PR 描述没有在同一次 push 中同步，导致旧描述让已修问题再次
+  被提出。
+- 对 `GuardrailMiddleware` 已有的 fail-closed 机制理解不足，在 adapter 中写了
+  "Phase 1 加 try/except" 的 TODO，暗示 adapter 应当自行处理异常。实际上 fail-closed
+  是 middleware 的职责，adapter 刻意不 catch 异常。描述与架构意图不一致导致了额外的
+  review 轮次。
+
+这些问题主要是实施前检查和可追踪性不足，并非两层授权设计被否定。
+
+## 所有后续阶段必须保持的不变量
+
+- `authorization.enabled: false` 必须保持现有行为不变。
+- Layer 1 和 Layer 2 必须使用同一个 provider 和同一个 Principal。
+- Layer 1 必须在 `assemble_deferred_tools` 之前过滤；被移除的工具不能进入
+  `DeferredToolCatalog`，也不能被 `tool_search` 再次提升。
+- Layer 1 必须覆盖 lead agent、native subagent 和 `DeerFlowClient` 三条装配路径。
+- Layer 2 复用 `GuardrailMiddleware`；不能在 adapter 中重复实现异常处理、审计、
+  deny 消息或 fail-closed 逻辑。
+- ownership 检查和现有 `require_admin_user()` 管理端点保护必须保留。细粒度授权只能
+  增加策略，不能削弱现有保护。
+- deny 优先于 allow。身份缺失、未知角色、provider 故障和 provider 返回值格式错误
+  都必须具有明确且经过测试的行为。
+- 同步和异步路径必须具有一致的授权决策和失败语义。
+- internal、Web、关闭认证、已绑定频道、未绑定频道、scheduler 和 subagent 的身份
+  都必须从真实来源追踪。
+- 新增 resource 或 action 时，必须检查所有消费者、allowlist、配置示例、文档和测试。
+- 新组件嵌入现有中间件前，必须先完整理解宿主中间件已有的机制（异常处理、审计、
+  fail-closed 等）。新组件不重复实现宿主已有的逻辑；如果看似缺少某功能，先确认是
+  否由宿主在上游或下游统一处理。
+
+## Phase 1 实施前确认
+
+Phase 1 是工具授权。编码前必须先确定：
+
+- Principal 在哪里构建，以及如何进入三条工具装配路径。
+- `auth_source`、owner role、`default_role` 和 subagent 继承如何组合。
+- provider 如何实例化，以及配置热更新后如何刷新。
+- authorization 与显式配置的 guardrail 如何共存，二者都不能静默替换或绕过对方。
+- provider 构造异常和授权决策异常是否使用同一 fail-closed 策略，以及各自由哪层处理。
+- 内置 RBAC provider 对 allow、deny 和通配符的精确定义。
+
+Phase 1 最低验证要求：
+
+- 每角色 allow、deny、通配符、deny 优先、未知角色和默认角色测试。
+- lead agent、subagent 和 embedded client 的工具可见性测试。
+- 证明被拒绝工具不会进入 deferred catalog。
+- Layer 2 的 allow、deny、provider 异常、审计、同步和异步测试。
+- prompt injection 回归测试，证明装配阶段被过滤的工具无法执行。
+- internal、未绑定频道、关闭认证和 subagent 的 Principal 测试。
+- 通过真实生命周期执行 AppConfig 加载与热更新测试。
+- 证明关闭 authorization 时现有工具集合完全不变。
+
+## 每次更新 PR 前的固定清单
+
+- [ ] 选择配置版本号前，已 fetch 并 rebase 最新 `upstream/main`。不能使用本地缓存的
+      旧版本号——主线可能在此期间已被其他 PR bump 过。先 fetch、读最新值、+1，再在
+      `config.example.yaml` + `deploy/helm/deer-flow/values.yaml` + `deploy/helm/deer-flow/README.md`
+      三处同步。
+- [ ] 已搜索 issue #4063 和当前阶段是否存在并行工作。
+- [ ] 每个新字段都已追踪到权威生产者，而不只是确认类型。
+- [ ] 测试经过公开运行时生命周期，而不只是直接构造配置模型。
+- [ ] 已按需覆盖 lead、subagent、embedded、同步和异步路径。
+- [ ] 已执行负向变异检查：删除新增 wiring 后，至少一个回归测试必须失败。
+- [ ] 已搜索配置的所有镜像，包括 Helm values 和相关文档。
+- [ ] 代码注释、测试、RFC 记录和 PR 描述已在同一次 push 中更新。
+- [ ] 已明确列出延期阶段，且没有把延期功能带入当前范围。
+- [ ] 已在下方记录新决策和未解决问题。
+- [ ] 如果多个 reviewer 独立指出同一处问题，视为高置信信号，立即修复，不再等待
+      进一步确认。
+
+## 决策日志
+
+只追加新记录。需要推翻旧决策时，必须新增一条“替代决策”，不能直接重写历史。
+
+### 2026-07-15 — Phase 0 / PR #4127
+
+- **决策：** `filter_resources` 为必需方法，不提供 Protocol fallback。
+- **决策：** provider 异常穿过 adapter，由 `GuardrailMiddleware` 处理。
+- **决策：** internal 身份来自认证上下文，不使用角色名称约定推导。
+- **决策：** Phase 0 默认保持运行时行为不变。
+- **延期：** Principal 构建、RBAC provider、两层执行接入和 internal Principal 传递
+  移至 Phase 1。
+
+### 2026-07-15 — Phase 1A-1 / 可信 Principal 链路
+
+- **背景：** Phase 0 建立了 `AuthorizationProvider` Protocol 和 adapter，但
+  `Principal.is_internal` 无可信来源，adapter 手工构造 Principal（与未来 Layer 1 的
+  builder 不一致），客户端可伪造身份字段。
+- **决策：** `build_principal_from_context()` 是唯一 Principal builder，Layer 1 和
+  Layer 2（adapter）必须共用。
+- **决策：** `is_internal` 来自 `request.state.auth_source == AUTH_SOURCE_INTERNAL`，
+  在 `inject_authenticated_user_context` 最顶部（所有 early return 之前）用直接赋值
+  写入 runtime context，不用 `setdefault`。
+- **决策：** `is_internal`、`authz_attributes` 和 `channel_user_id` 列为
+  `_SERVER_OWNED_AUTHZ_CONTEXT_KEYS`，
+  从 `config["context"]` 和 `config["configurable"]` 清除客户端值。
+- **决策：** `channel_user_id` 只接受内部认证 IM 调用方的顶层 `body.context` 值；普通
+  session 调用和 `body.config` 两个 section 均不能提供该授权身份字段。
+- **决策：** Phase 1A-1 没有 Gateway 侧 `authz_attributes` 权威生产者；Gateway 请求中
+  的 `authz_attributes` 一律删除（默认 `{}`）。
+- **决策：** adapter 的 `evaluate`/`aevaluate` 通过 `build_principal_from_context()`
+  构造 Principal，接收 `default_role` 参数。
+- **决策：** `authz_attributes` 在所有进程内消费边界统一使用 `isinstance(x, Mapping)`
+  + `dict()` 复制；非 Mapping 抛 `TypeError`。
+- **决策：** subagent 的 `is_internal` 无条件写回 context（包括 `False`）。
+- **证据：** 323 个目标与边界测试通过，覆盖 Gateway 防伪、channel sender 信任边界、subagent 继承、
+  `GuardrailMiddleware` runtime 字段映射、adapter builder 复用和 harness/app 边界。
+- **兼容性：** `authorization.enabled: false` 时工具集合和执行决策不变；runtime context
+  新增 `is_internal` 字段是有意的可观察变化。
+- **延期：** RBAC provider、provider factory、Layer 1 过滤、Layer 2 自动接线移至
+  Phase 1A-2 / Phase 1B。
+
+### 新记录模板
+
+```markdown
+### YYYY-MM-DD — Phase N / PR #NNNN
+
+- **背景：** 本次变化或 review 发现了什么？
+- **决策：** 新的正式契约是什么？
+- **证据：** 相关代码路径、测试、issue 评论或 benchmark。
+- **否决方案：** 考虑过哪些方案，为什么不采用？
+- **兼容性：** 如何保持现有部署和关闭功能时的行为？
+- **延期：** 哪些工作留给下一阶段？
+```
+
+## 当前连续性风险
+
+- 原始 RFC 仍包含“`filter_resources` 存在默认实现”的草案示例；本文件记录的已合并
+  Phase 0 契约优先于该示例。
+- RFC 原始 Phase 0 范围大于 PR #4127 的实际落地范围。后续实现必须依据已合并基线
+  和明确延期清单，不能假设这些功能已经存在。
+- 主线配置版本可能并发变化。不能提前占用版本号；必须先 rebase，再在所有镜像文件中
+  使用下一个有效版本。

docs/plans/2026-07-10-pluggable-authorization-rfc.md

@@ -1,6 +1,11 @@
 <!-- Authored by @zhfeng, discussed in https://github.com/bytedance/deer-flow/issues/4063.
      Added to the PR per @WillemJiang's request for design tracking. -->
 
+> **实施连续性要求：** 每个阶段的 PR 都必须阅读并更新
+> [可插拔授权系统实施记录](2026-07-10-pluggable-authorization-implementation-notes.md)。
+> 该文件记录已合并契约、review 决策、延期工作和实施前检查项。旧 RFC 示例与已合并
+> 代码及测试不一致时，以已合并契约为准。
+
 # RFC: Pluggable Fine-Grained Authorization
 
 **Status:** Draft for feedback (responds to [#3462](https://github.com/bytedance/deer-flow/issues/3462)).

docs/plans/2026-07-15-authz-phase1a-implementation-plan.md

@@ -0,0 +1,398 @@
+# Phase 1A 实施计划：可信身份链路与内置 RBAC 核心
+
+> RFC：`docs/plans/2026-07-10-pluggable-authorization-rfc.md`
+> 前情记录：`docs/plans/2026-07-10-pluggable-authorization-implementation-notes.md`
+> Phase 0 基线：PR #4127 / `1300c6d3`
+
+## 1. 目标与拆分
+
+Phase 1A 只建立身份和策略核心，不把授权接入工具装配或执行路径。为降低 review
+复杂度，拆成两个可独立合并的 PR：
+
+1. **Phase 1A-1：可信 Principal 链路**
+   - 从 Gateway 的服务端认证状态产生 `is_internal`。
+   - 使用唯一的 Principal builder 解析身份。
+   - 将完整身份传递到 subagent 和 Guardrail adapter。
+2. **Phase 1A-2：内置 RBAC 与 provider factory**
+   - 实现严格、确定性的内置 RBAC provider。
+   - 实现统一的 provider 实例化和 Protocol 校验。
+
+Phase 1A 合并后的准确兼容性承诺是：
+
+- `authorization.enabled: false` 时，工具集合和工具执行决策不变。
+- runtime context 会新增服务端生成的身份字段，这是有意的可观察变化。
+- Phase 1A 不执行 Layer 1 过滤，也不自动安装 Layer 2 middleware。
+
+## 2. 固定架构与职责
+
+```text
+Gateway 认证状态
+    │
+    ├─ 清除客户端伪造的服务端身份字段
+    ├─ 注入 user_id / user_role / oauth_* / is_internal
+    ▼
+runtime context
+    │
+    ├─ build_principal_from_context(default_role)
+    │      └─ Phase 1B Layer 1 使用
+    │
+    ├─ task tool → SubagentExecutor → subagent runtime context
+    │
+    └─ GuardrailMiddleware → GuardrailRequest
+             → GuardrailAuthorizationAdapter
+             → build_principal_from_context(default_role)
+             → AuthorizationProvider
+```
+
+职责必须保持单一：
+
+- **Gateway**：确认字段来源可信，覆盖或清除客户端值。
+- **Principal builder**：补齐缺失角色、规范化类型、复制 attributes。
+- **RBAC provider**：只解释已经解析好的 Principal 和角色策略。
+- **provider factory**：加载、构造、校验 provider，不执行授权策略。
+- **GuardrailMiddleware**：处理 provider 异常以及 `fail_closed`。
+- **adapter**：只做数据映射和决策类型转换，不捕获异常。
+
+## 3. 全局语义锁
+
+### 3.1 Principal
+
+- `role` 仅在 `user_role` 为 `None` 或空字符串时使用 `default_role`。
+- 未知但非空的角色不能回退到 `default_role`。
+- `is_internal` 只有原值严格等于 `True` 时才为 `True`。
+- `authz_attributes` 必须是 `Mapping`；其他非空类型抛 `TypeError`。
+- `attributes` 总是复制为新字典，不共享输入引用。
+- Layer 1 和 Layer 2 必须通过同一个 builder 获得语义一致的 Principal。
+
+### 3.2 服务端身份字段
+
+`is_internal`、`authz_attributes` 和 `channel_user_id` 是服务端拥有字段：
+
+- Gateway 必须从 `config.context` 和 `config.configurable` 清除客户端值。
+- `is_internal` 只能由 `request.state.auth_source == AUTH_SOURCE_INTERNAL` 产生。
+- `channel_user_id` 只能来自内部认证 IM 调用方的顶层 `body.context`；普通 session
+  请求和 `body.config` 两个 section 中的同名值必须删除。
+- 写入使用直接赋值，不能使用 `setdefault`。
+- 写入必须发生在 `user_id is None` 等所有 early return 之前。
+- Phase 1A 尚无 Gateway 侧 `authz_attributes` 权威生产者，因此 Gateway 路径固定为空；
+  不能接受普通 HTTP 客户端提供的 attributes。
+- 嵌入式 Python 调用属于进程内可信调用，可直接使用 builder 构造 attributes。
+
+### 3.3 RBAC
+
+每个角色的资源策略使用以下语义：
+
+| 配置 | 行为 |
+|---|---|
+| `allow: "*"` 或 `allow: true` | 默认允许全部候选 |
+| `allow: [...]` | 只允许列表成员 |
+| `allow: []` 或 `allow: false` | 全部拒绝 |
+| `allow` 缺失 | 默认允许，仍应用 deny |
+| `deny: [...]` | 从允许集合移除；deny 永远优先 |
+| 资源配置缺失 | 此 provider 不限制该资源 |
+| Principal 角色缺失 | 视为 builder/调用方错误，抛明确异常 |
+| Principal 角色未知 | 抛明确异常，由执行层按 `fail_closed` 处理 |
+| 配置类型、成员类型或字段非法 | provider 构造时抛 `ValueError` |
+
+资源名必须显式映射，禁止通过简单加 `s` 猜测：
+
+```python
+RESOURCE_POLICY_KEYS = {
+    "tool": "tools",
+    "model": "models",
+    "skill": "skills",
+    "sandbox": "sandbox",
+    "mcp_server": "mcp_servers",
+    "route": "routes",
+}
+```
+
+未知 resource 使用原名查找；未配置时按“资源配置缺失”处理。Phase 1A 的行为测试
+以 `tool -> tools` 为主，其他资源只固定映射契约，不接线。
+
+### 3.4 fail-closed
+
+- `fail_closed` 不传给 provider，也不由 provider 或 adapter解释。
+- provider 遇到未知身份或内部错误时抛异常。
+- Phase 1B 的 Layer 1 helper 和现有 `GuardrailMiddleware` 分别在边界处应用
+  `fail_closed`。
+- provider 主动返回 `allow=True` 不属于异常，middleware 不会替它改判；因此未知角色
+  绝不能返回 allow。
+
+## 4. PR Phase 1A-1：可信 Principal 链路
+
+建议标题：
+
+```text
+feat(authz): propagate trusted authorization principal context
+```
+
+### Step 1：先写失败测试
+
+新增 `backend/tests/test_authorization_principal.py`：
+
+- 空 context 和部分 context。
+- 缺失、`None`、空字符串角色使用 `default_role`。
+- 未知非空角色原样保留。
+- `is_internal` 仅接受严格布尔 `True`。
+- attributes 缺失、复制、输入修改不反向影响 Principal。
+- attributes 非 Mapping 时抛 `TypeError`。
+
+扩展 `backend/tests/test_gateway_services.py`：
+
+- 普通请求通过 `body.config.context` 伪造 `is_internal=True`，最终为 `False`。
+- 普通请求通过 `body.config.configurable` 伪造该值，最终被清除。
+- `user=None` 时仍写入 `is_internal=False`。
+- 内部认证请求写入 `is_internal=True`。
+- 普通请求注入的 `authz_attributes` 从两个 config section 中被清除。
+- 原有 user/owner/oauth 注入测试继续通过。
+
+扩展现有 subagent/guardrail 测试：
+
+- `is_internal=True/False` 均能原样传递，不能只在 truthy 时写回。
+- `channel_user_id` 和 attributes 一同传递。
+- subagent 对 attributes 使用副本。
+- adapter 同步、异步路径映射出相同 Principal。
+- adapter 通过 builder 应用 `default_role`，不维护第二套回退逻辑。
+
+### Step 2：实现 Principal builder
+
+新增：
+
+```text
+backend/packages/harness/deerflow/authz/principal.py
+```
+
+接口：
+
+```python
+def build_principal_from_context(
+    context: Mapping[str, Any],
+    *,
+    default_role: str,
+) -> Principal:
+    ...
+```
+
+实现保持纯函数，不读取全局 AppConfig，不缓存结果，不修改输入。
+
+### Step 3：保护并注入 Gateway 身份
+
+修改：
+
+```text
+backend/app/gateway/services.py
+```
+
+新增独立常量，例如：
+
+```python
+_SERVER_OWNED_AUTHZ_CONTEXT_KEYS = frozenset(
+    {"is_internal", "authz_attributes", "channel_user_id"}
+)
+```
+
+在 `inject_authenticated_user_context()` 开头：
+
+1. 从 `context` 和 `configurable` 删除上述客户端值。
+2. 确保 runtime context 是字典；若输入类型非法，使用现有配置错误约定明确失败，
+   不能静默保留伪造值。
+3. 直接写入服务端计算的 `is_internal`。
+4. 再执行现有 user/internal owner 分支和 early return。
+
+不要把 `is_internal` 加进允许 internal caller 自定义的普通 override 白名单；它始终由
+认证中间件产生。
+
+### Step 4：完整传递 subagent 身份
+
+修改：
+
+```text
+backend/packages/harness/deerflow/tools/builtins/task_tool.py
+backend/packages/harness/deerflow/subagents/executor.py
+```
+
+从 parent runtime context 捕获：
+
+```text
+user_id / user_role / oauth_provider / oauth_id / channel_user_id
+is_internal / authz_attributes
+```
+
+`SubagentExecutor` 构造时复制 attributes；写回 subagent context 时再次复制。
+`is_internal` 必须无条件写回布尔值，包括 `False`。
+
+### Step 5：扩展 GuardrailRequest 并复用 builder
+
+修改：
+
+```text
+backend/packages/harness/deerflow/guardrails/provider.py
+backend/packages/harness/deerflow/guardrails/middleware.py
+backend/packages/harness/deerflow/authz/adapter.py
+backend/tests/test_authorization_provider.py
+```
+
+`GuardrailRequest` 增加向后兼容的默认字段：
+
+```python
+channel_user_id: str | None = None
+is_internal: bool = False
+authz_attributes: dict[str, Any] = field(default_factory=dict)
+```
+
+adapter 构造函数增加 `default_role`，并在 `_to_authz()` 中调用
+`build_principal_from_context()`；删除 Phase 0 中“不映射 is_internal”的说明。
+
+### Step 6：导出与文档
+
+修改 `deerflow/authz/__init__.py` 导出 builder。向 implementation notes 的决策日志
+追加 Phase 1A-1 记录，不改写 Phase 0 历史。
+
+### Phase 1A-1 验收
+
+- 删除 Gateway 对 `is_internal` 的直接赋值后，防伪测试必须失败。
+- 普通请求无法从 `body.context` 或 `body.config` 注入 `channel_user_id`，内部认证 IM
+  请求只保留顶层 `body.context` 的 sender id。
+- 删除任意一段 subagent 传递后，继承测试必须失败。
+- Gateway、subagent、adapter 得到的身份字段一致。
+- 没有 Layer 1/Layer 2 自动接线。
+- 现有 guardrail、Gateway、subagent 测试全部通过。
+
+## 5. PR Phase 1A-2：内置 RBAC 与 provider factory
+
+建议标题：
+
+```text
+feat(authz): add built-in RBAC provider and provider factory
+```
+
+前置：Phase 1A-1 已合并或当前分支已 rebase 到其提交。
+
+### Step 1：先写失败测试
+
+新增 `backend/tests/test_rbac_authorization_provider.py`：
+
+- wildcard、布尔 allow、列表 allow、空列表、allow 缺失。
+- deny 优先于所有 allow 形式。
+- 资源配置缺失时不限制。
+- `tool -> tools` 等显式资源映射。
+- 未知角色和缺失角色抛异常，绝不返回 allow。
+- 非法 roles、role policy、resource policy、allow/deny 类型和非字符串成员。
+- `authorize()` 与 `aauthorize()` 决策一致。
+- `filter_resources()` 与逐项 `authorize()` 结果一致。
+- 过滤保持 candidates 顺序和重复项，不增加输入中不存在的资源。
+- 构造后修改原配置不会改变 provider 行为。
+
+新增 `backend/tests/test_authorization_runtime.py`：
+
+- disabled 时直接返回 `None`，且不尝试 import 无效 class path。
+- enabled 但 provider 缺失时抛明确错误。
+- 路径不存在、目标不是 class、构造失败时错误包含 class path 并保留异常链。
+- 实例不符合 `AuthorizationProvider` Protocol 时明确失败。
+- provider factory 不注入 `fail_closed` 或 `default_role`。
+- 内置 RBAC 能通过相同标准路径解析，不写特殊分支。
+
+### Step 2：实现 RBAC provider
+
+新增：
+
+```text
+backend/packages/harness/deerflow/authz/rbac.py
+```
+
+要求：
+
+- 构造时完成全部配置校验和规范化。
+- allow/deny 预编译为不可变集合或明确的“全部/全部拒绝”标记。
+- 请求路径只做 O(1) membership 和 O(n) candidates 遍历。
+- 返回稳定 reason code；拒绝消息包含 role、resource、target，但不包含敏感配置。
+- `filter_resources()` 保持输入顺序，不修改输入列表。
+- 不读取全局 config，不处理 `fail_closed`，不二次应用 `default_role`。
+
+### Step 3：实现 provider factory
+
+新增：
+
+```text
+backend/packages/harness/deerflow/authz/runtime.py
+```
+
+接口：
+
+```python
+def resolve_authorization_provider(
+    config: AuthorizationConfig,
+) -> AuthorizationProvider | None:
+    ...
+```
+
+固定顺序：
+
+1. `enabled=False`：立即返回 `None`。
+2. `enabled=True` 且 provider 缺失：抛 `ValueError`。
+3. 使用 `resolve_variable(path, expected_type=type)` 解析 class。
+4. 只用 `provider.config` 中显式提供的 kwargs 构造实例。
+5. 使用 `isinstance(instance, AuthorizationProvider)` 做结构校验。
+6. 包装错误时包含 class path、保留 `raise ... from err`，不打印 kwargs。
+
+factory 不缓存 provider。Phase 1B 在每次 agent build 时解析一次，并把同一个实例传给
+Layer 1 和 Layer 2。
+
+### Step 4：导出与文档
+
+修改 `deerflow/authz/__init__.py` 导出 RBAC provider 和 factory。向 implementation
+notes 追加 Phase 1A-2 决策记录。
+
+Phase 1A-2 不修改 `config.example.yaml`：在执行层尚未接线时展示“启用 RBAC”的用户配置
+会造成已经生效的错觉。完整 RBAC 示例随 Phase 1B enforcement 一起加入；届时同时搜索
+Helm values 和相关文档镜像。Phase 1A 不改变配置 schema，因此不 bump `config_version`。
+
+### Phase 1A-2 验收
+
+- 未知角色无法静默放行。
+- deny 在 `authorize` 和 `filter_resources` 中都优先。
+- 内置和自定义 provider 走相同 factory 路径。
+- disabled 路径不 import、不构造 provider。
+- provider 配置在构造后不可被外部可变引用改变。
+- 没有 Layer 1/Layer 2 自动接线。
+
+## 6. 测试与检查命令
+
+每个 PR 按 TDD 顺序执行：先提交/观察失败测试，再实现到通过。
+
+```powershell
+cd backend
+uv run pytest tests/test_authorization_principal.py -q
+uv run pytest tests/test_authorization_provider.py tests/test_gateway_services.py -q
+uv run pytest tests/test_rbac_authorization_provider.py tests/test_authorization_runtime.py -q
+uv run pytest tests/test_harness_boundary.py -q
+uv run ruff check packages/harness/deerflow/authz app/gateway/services.py tests
+uv run ruff format --check packages/harness/deerflow/authz app/gateway/services.py tests
+```
+
+提交前再运行 `make test`；如果全量测试受环境依赖阻塞，PR 描述必须列出已运行命令、
+通过结果和具体阻塞，不得只写“tests passed”。
+
+## 7. Phase 1A 明确不做
+
+- Lead agent、native subagent、embedded client 的 Layer 1 工具过滤。
+- `DeferredToolCatalog` 和 `tool_search` 的授权集成。
+- Layer 2 `GuardrailMiddleware` 自动装配以及与显式 guardrail 的组合顺序。
+- `DeerFlowClient` 现有 skill filter 缺口修复。
+- route、model、skill、sandbox、MCP server 的实际授权接线。
+- provider 缓存、跨 build singleton 或热更新生命周期优化。
+- 前端权限展示。
+
+这些工作进入 Phase 1B 或后续独立 PR；Phase 1A 不提前加入未被消费的执行逻辑。
+
+## 8. Phase 1B 前置验收清单
+
+- [ ] Principal 的每个字段都有明确权威来源。
+- [ ] Gateway 的服务端字段不可通过两个 config section 伪造。
+- [ ] lead/subagent/adapter 使用同一 builder 语义。
+- [ ] 未知角色、非法策略和 provider 构造失败均有明确异常。
+- [ ] RBAC 的同步、异步、批量过滤结果一致。
+- [ ] Phase 1A-1 与 1A-2 的决策已追加到 implementation notes。
+- [ ] 分支已 rebase 最新 upstream/main。
+- [ ] 未提前修改配置版本号。

docs/plans/2026-07-15-authz-phase1a1-trusted-principal-implementation-plan.md

@@ -0,0 +1,514 @@
+# Phase 1A-1 实施计划：可信 Principal 链路
+
+建议 PR 标题：
+
+```text
+feat(authz): propagate trusted authorization principal context
+```
+
+前置基线：PR #4127 / `1300c6d3`
+总计划：`docs/plans/2026-07-15-authz-phase1a-implementation-plan.md`
+实施记录：`docs/plans/2026-07-10-pluggable-authorization-implementation-notes.md`
+
+## 1. 目标
+
+本 PR 完成一条可信、可测试的授权身份链路：
+
+```text
+Gateway 认证状态
+  → 清除客户端伪造字段
+  → 服务端注入 is_internal
+  → lead runtime context
+  → task_tool 捕获
+  → SubagentExecutor 复制
+  → subagent runtime context
+  → GuardrailRequest
+  → GuardrailAuthorizationAdapter
+  → build_principal_from_context()
+  → Principal
+```
+
+本 PR 不接入 Layer 1 工具过滤，也不自动安装 Layer 2 middleware。合并后的兼容性承诺：
+
+- `authorization.enabled: false` 时，工具集合和工具执行决策不变。
+- runtime context 新增服务端身份字段是有意的可观察变化。
+- 现有 adapter 异常传播语义保持不变。
+
+## 2. 固定安全契约
+
+### 2.1 Principal
+
+`build_principal_from_context()` 必须显式构造全部 7 个字段：
+
+```text
+user_id / role / oauth_provider / oauth_id / channel_user_id /
+is_internal / attributes
+```
+
+规则：
+
+- `user_role` 为 `None` 或空字符串时使用 `default_role`。
+- 未知但非空的角色原样保留；不能回退到默认角色。
+- `is_internal` 只有原值严格等于 `True` 时才为 `True`。
+- `authz_attributes` 为 `None` 或缺失时解析为 `{}`。
+- 非空 `authz_attributes` 必须实现 `Mapping`，否则抛 `TypeError`。
+- attributes 每次都复制为新字典，不共享可变引用。
+- builder 是纯函数：不读全局 config、不缓存、不修改输入。
+
+### 2.2 Gateway 可信边界
+
+`is_internal`、`authz_attributes` 和 `channel_user_id` 是服务端拥有字段：
+
+- 必须从 `config["context"]` 和 `config["configurable"]` 清除客户端值。
+- `is_internal` 只能由 `request.state.auth_source == AUTH_SOURCE_INTERNAL` 产生。
+- `channel_user_id` 只能来自内部认证 IM 调用方的顶层 `body.context`；普通 session
+  请求和 `body.config` 两个 section 中的值必须删除。
+- 必须使用直接赋值，不能使用 `setdefault`。
+- 必须在 `user_id is None` 等所有 early return 之前写入。
+- Phase 1A-1 没有 Gateway 侧 attributes 权威生产者，因此 Gateway 请求中的
+  `authz_attributes` 一律删除。
+
+### 2.3 传递语义
+
+- Subagent 必须继承 parent context 的 `is_internal` 和合法 attributes。
+- `is_internal=False` 也必须显式写回，不能按 truthy 条件省略。
+- attributes 在 task 捕获、executor 构造和 context 写回时均使用副本。
+- 非 Mapping attributes 不能在某些层抛错、另一些层静默变成 `{}`；所有进程内消费
+  边界统一抛 `TypeError`。
+- Guardrail adapter 必须复用 Principal builder，不能维护第二套默认角色或 attributes
+  解析逻辑。
+
+## 3. TDD 实施顺序
+
+### Step 1：新增 Principal builder 失败测试
+
+新增：
+
+```text
+backend/tests/test_authorization_principal.py
+```
+
+覆盖：
+
+- 空 context、部分 context 和全部 7 个字段映射。
+- 缺失、`None`、空字符串 role 使用 `default_role`。
+- 未知非空 role 原样保留。
+- `is_internal` 对 `True`、`False`、`1`、`"true"`、`None` 的严格布尔行为。
+- attributes 缺失或 `None` 得到 `{}`。
+- Mapping attributes 被复制；修改输入不影响 Principal。
+- 非 Mapping attributes 抛 `TypeError`，错误包含实际类型。
+- oauth 和 channel identity 正确映射。
+
+先运行该测试并确认因 builder 不存在而失败。
+
+### Step 2：实现 Principal builder
+
+新增：
+
+```text
+backend/packages/harness/deerflow/authz/principal.py
+```
+
+接口和核心实现：
+
+```python
+from collections.abc import Mapping
+from typing import Any
+
+from deerflow.authz.provider import Principal
+
+
+def build_principal_from_context(
+    context: Mapping[str, Any],
+    *,
+    default_role: str,
+) -> Principal:
+    resolved_role = context.get("user_role")
+    if resolved_role is None or resolved_role == "":
+        resolved_role = default_role
+
+    raw_attributes = context.get("authz_attributes")
+    if raw_attributes is None:
+        attributes: dict[str, Any] = {}
+    elif isinstance(raw_attributes, Mapping):
+        attributes = dict(raw_attributes)
+    else:
+        raise TypeError(
+            "authz_attributes must be a Mapping, "
+            f"got {type(raw_attributes).__name__}"
+        )
+
+    return Principal(
+        user_id=context.get("user_id"),
+        role=resolved_role,
+        oauth_provider=context.get("oauth_provider"),
+        oauth_id=context.get("oauth_id"),
+        channel_user_id=context.get("channel_user_id"),
+        is_internal=context.get("is_internal") is True,
+        attributes=attributes,
+    )
+```
+
+修改 `backend/packages/harness/deerflow/authz/__init__.py` 导出
+`build_principal_from_context`。
+
+### Step 3：新增 Gateway 防伪失败测试
+
+修改：
+
+```text
+backend/tests/test_gateway_services.py
+```
+
+必须复用真实配置装配顺序：
+
+```text
+build_run_config
+  → merge_run_context_overrides
+  → strip_internal_context_keys（非 internal 请求）
+  → inject_authenticated_user_context
+```
+
+分别测试以下入口，不能合并为一个配置：
+
+1. `body.config["context"]` 伪造 `is_internal=True`。
+2. `body.config["configurable"]` 伪造 `is_internal=True`。
+3. 两个入口分别伪造 `authz_attributes`。
+
+断言：
+
+- 普通/session 请求最终 `context.is_internal is False`。
+- `configurable.is_internal` 被删除。
+- 两个 section 中的 `authz_attributes` 都被删除。
+- internal 请求最终 `context.is_internal is True`。
+- internal 请求携带的客户端伪造 attributes 同样被删除。
+- `user=None`、auth source 缺失或 session 时仍写入 `False`。
+- `user=None`、`auth_source=internal` 时仍写入 `True`。
+- 非 Mapping runtime context 显式抛 `TypeError`，不静默跳过。
+- 原有 user、owner、role、oauth 测试继续通过。
+
+这些测试必须证明新增覆盖逻辑是唯一通过原因，不能使用本来就被 `body.context`
+白名单过滤的空路径。
+
+### Step 4：实现 Gateway 防伪与注入
+
+修改：
+
+```text
+backend/app/gateway/services.py
+```
+
+新增：
+
+```python
+_SERVER_OWNED_AUTHZ_CONTEXT_KEYS = frozenset(
+    {"is_internal", "authz_attributes", "channel_user_id"}
+)
+```
+
+在 `inject_authenticated_user_context()` 最顶部、读取 `user_id` 之前执行：
+
+```python
+runtime_context = config.setdefault("context", {})
+if not isinstance(runtime_context, dict):
+    raise TypeError("run context must be a mapping")
+
+for key in _SERVER_OWNED_AUTHZ_CONTEXT_KEYS:
+    runtime_context.pop(key, None)
+
+configurable = config.get("configurable")
+if isinstance(configurable, dict):
+    for key in _SERVER_OWNED_AUTHZ_CONTEXT_KEYS:
+        configurable.pop(key, None)
+
+auth_source = getattr(getattr(request, "state", None), "auth_source", None)
+runtime_context["is_internal"] = auth_source == AUTH_SOURCE_INTERNAL
+```
+
+之后保留现有普通用户、internal owner 和 early return 逻辑。`AUTH_SOURCE_INTERNAL`
+已经由当前模块导入，不重复增加角色名称推导。
+
+### Step 5：新增 task_tool 身份捕获失败测试
+
+修改：
+
+```text
+backend/tests/test_task_tool_core_logic.py
+```
+
+复用现有 `test_task_tool_forwards_channel_user_id_to_executor` 的 DummyExecutor 模式，覆盖：
+
+- parent `is_internal=True` 传给 executor。
+- parent `is_internal=False` 仍显式传给 executor。
+- Mapping attributes 被复制后传给 executor。
+- 捕获后修改 parent attributes 不影响 executor kwargs。
+- 非 Mapping attributes 抛 `TypeError`，不能静默变成 `{}`。
+
+### Step 6：实现 task_tool 身份捕获
+
+修改：
+
+```text
+backend/packages/harness/deerflow/tools/builtins/task_tool.py
+```
+
+在现有 parent identity capture 块增加：
+
+```python
+is_internal = parent_context.get("is_internal") is True
+raw_attributes = parent_context.get("authz_attributes")
+if raw_attributes is None:
+    authz_attributes: dict[str, Any] = {}
+elif isinstance(raw_attributes, Mapping):
+    authz_attributes = dict(raw_attributes)
+else:
+    raise TypeError(
+        "authz_attributes must be a Mapping, "
+        f"got {type(raw_attributes).__name__}"
+    )
+```
+
+将二者无条件加入 `executor_kwargs`。
+
+### Step 7：新增 executor 写回失败测试
+
+修改：
+
+```text
+backend/tests/test_subagent_executor.py
+```
+
+复用现有 channel identity context 测试模式，覆盖：
+
+- 构造参数 `is_internal=True/False` 都写回 subagent context。
+- attributes 在构造时复制。
+- attributes 在 context write-back 时再次复制。
+- 修改调用方原字典或写回后的字典均不影响 executor 内部副本。
+- executor 直接收到非 Mapping attributes 时抛 `TypeError`。
+
+### Step 8：实现 executor 身份写回
+
+修改：
+
+```text
+backend/packages/harness/deerflow/subagents/executor.py
+```
+
+构造参数增加：
+
+```python
+is_internal: bool = False,
+authz_attributes: Mapping[str, Any] | None = None,
+```
+
+构造时严格校验并复制：
+
+```python
+self.is_internal = is_internal
+if authz_attributes is None:
+    self.authz_attributes = {}
+elif isinstance(authz_attributes, Mapping):
+    self.authz_attributes = dict(authz_attributes)
+else:
+    raise TypeError(
+        "authz_attributes must be a Mapping, "
+        f"got {type(authz_attributes).__name__}"
+    )
+```
+
+context write-back 无条件执行：
+
+```python
+context["is_internal"] = self.is_internal
+context["authz_attributes"] = dict(self.authz_attributes)
+```
+
+### Step 9：新增 Guardrail/adapter 失败测试
+
+修改：
+
+```text
+backend/tests/test_authorization_provider.py
+backend/tests/test_guardrail_middleware.py
+```
+
+覆盖：
+
+- `GuardrailRequest` 新字段默认值向后兼容。
+- middleware 映射 `channel_user_id`、严格布尔 `is_internal`、合法 attributes。
+- middleware 遇到非 Mapping attributes 时抛 `TypeError`。
+- adapter 同步和异步路径产生相同 Principal。
+- adapter 正确映射全部 7 个 Principal 字段。
+- role 缺失时 adapter 通过 builder 应用 `default_role`。
+- 未知非空 role 不回退。
+- 将旧的 Phase 0 `is_internal` 不映射测试改为正确映射测试。
+- provider 异常仍穿过 adapter，现有异常传播测试继续通过。
+
+middleware context 映射断言放在现有 `test_guardrail_middleware.py`，adapter 转换断言
+放在 `test_authorization_provider.py`；不得新建重复测试模块。
+
+### Step 10：扩展 GuardrailRequest 和 adapter
+
+修改：
+
+```text
+backend/packages/harness/deerflow/guardrails/provider.py
+backend/packages/harness/deerflow/guardrails/middleware.py
+backend/packages/harness/deerflow/authz/adapter.py
+```
+
+`GuardrailRequest` 增加带默认值的字段：
+
+```python
+channel_user_id: str | None = None
+is_internal: bool = False
+authz_attributes: dict[str, Any] = field(default_factory=dict)
+```
+
+middleware 在构造 request 前用普通分支完成 attributes 校验，不使用 walrus one-liner：
+
+```python
+raw_attributes = context.get("authz_attributes")
+if raw_attributes is None:
+    authz_attributes: dict[str, Any] = {}
+elif isinstance(raw_attributes, Mapping):
+    authz_attributes = dict(raw_attributes)
+else:
+    raise TypeError(...)
+```
+
+adapter：
+
+- 构造函数增加 `default_role: str = "user"` 并保存。
+- `_to_authz()` 从 GuardrailRequest 组装 context 字典。
+- 调用 `build_principal_from_context()`，不得手工构造 Principal。
+- 删除 Phase 0 关于 `is_internal` 尚未映射的 note。
+- 不捕获 provider 异常，不实现 `fail_closed`。
+
+同时更新 `authz/provider.py` 中 Principal 的生命周期说明：adapter 每次请求实时构建
+Principal，不得继续描述为“每个 run 只构建一次”。
+
+Phase 1B 自动装配 adapter 时必须显式传入
+`AuthorizationConfig.default_role`；本 PR 不进行该装配。
+
+### Step 11：更新实施记录
+
+修改：
+
+```text
+docs/plans/2026-07-10-pluggable-authorization-implementation-notes.md
+```
+
+只追加 Phase 1A-1 决策日志，记录：
+
+- internal 来自服务端 auth source。
+- Gateway 清除服务端拥有字段。
+- adapter 复用唯一 Principal builder。
+- attributes 使用严格 Mapping + copy 语义。
+- subagent 无条件继承 internal 布尔值。
+- RBAC、provider factory 和执行接线继续延期。
+
+不修改 Phase 0 历史记录，不 bump `config_version`。
+
+## 4. 精确文件清单
+
+新增 5 个文件：
+
+```text
+backend/packages/harness/deerflow/authz/principal.py
+backend/tests/test_authorization_principal.py
+docs/plans/2026-07-10-pluggable-authorization-implementation-notes.md
+docs/plans/2026-07-15-authz-phase1a-implementation-plan.md
+docs/plans/2026-07-15-authz-phase1a1-trusted-principal-implementation-plan.md
+```
+
+修改 17 个文件：
+
+```text
+backend/AGENTS.md
+backend/packages/harness/deerflow/authz/__init__.py
+backend/packages/harness/deerflow/authz/adapter.py
+backend/packages/harness/deerflow/authz/provider.py
+backend/packages/harness/deerflow/guardrails/provider.py
+backend/packages/harness/deerflow/guardrails/middleware.py
+backend/app/gateway/services.py
+backend/packages/harness/deerflow/tools/builtins/task_tool.py
+backend/packages/harness/deerflow/subagents/executor.py
+backend/packages/harness/deerflow/sandbox/tools.py
+backend/tests/test_authorization_provider.py
+backend/tests/test_channel_user_id_env.py
+backend/tests/test_guardrail_middleware.py
+backend/tests/test_task_tool_core_logic.py
+backend/tests/test_subagent_executor.py
+backend/tests/test_gateway_services.py
+docs/plans/2026-07-10-pluggable-authorization-rfc.md
+```
+
+合计 22 个文件。文件数较多是因为它固定一条跨 Gateway、lead、subagent 和 guardrail
+的端到端身份链；继续拆分会产生可合并但身份链不完整的中间状态。
+
+## 5. 明确不修改
+
+```text
+backend/packages/harness/deerflow/config/authorization_config.py
+config.example.yaml
+backend/packages/harness/deerflow/agents/lead_agent/agent.py
+backend/packages/harness/deerflow/client.py
+```
+
+本 PR 不实现：
+
+- RBAC provider 和 provider factory（Phase 1A-2）。
+- Layer 1 工具过滤和 deferred catalog 防提升（Phase 1B）。
+- Layer 2 自动接线及与显式 guardrail 的组合顺序（Phase 1B）。
+- DeerFlowClient skill filter gap（独立 bugfix PR）。
+- route、model、skill、sandbox、MCP server 授权。
+- RBAC 配置示例和配置版本变更。
+
+## 6. 测试命令
+
+按 TDD 步骤先确认新增测试失败，再逐段实现：
+
+```powershell
+cd backend
+
+uv run pytest tests/test_authorization_principal.py -q
+uv run pytest tests/test_gateway_services.py -q
+uv run pytest tests/test_task_tool_core_logic.py -q
+uv run pytest tests/test_subagent_executor.py -q
+uv run pytest tests/test_authorization_provider.py -q
+uv run pytest tests/test_guardrail_middleware.py -q
+uv run pytest tests/test_channel_user_id_env.py -q
+
+uv run pytest tests/test_authorization_principal.py tests/test_gateway_services.py tests/test_channel_user_id_env.py tests/test_task_tool_core_logic.py tests/test_subagent_executor.py tests/test_authorization_provider.py tests/test_guardrail_middleware.py -q
+uv run pytest tests/test_harness_boundary.py -q
+
+uv run ruff check packages/harness/deerflow/authz packages/harness/deerflow/guardrails packages/harness/deerflow/tools/builtins/task_tool.py packages/harness/deerflow/subagents/executor.py packages/harness/deerflow/sandbox/tools.py app/gateway/services.py tests/test_authorization_principal.py tests/test_authorization_provider.py tests/test_channel_user_id_env.py tests/test_task_tool_core_logic.py tests/test_subagent_executor.py tests/test_gateway_services.py
+uv run ruff format --check packages/harness/deerflow/authz packages/harness/deerflow/guardrails packages/harness/deerflow/tools/builtins/task_tool.py packages/harness/deerflow/subagents/executor.py packages/harness/deerflow/sandbox/tools.py app/gateway/services.py tests/test_authorization_principal.py tests/test_authorization_provider.py tests/test_channel_user_id_env.py tests/test_task_tool_core_logic.py tests/test_subagent_executor.py tests/test_gateway_services.py
+```
+
+提交前运行：
+
+```powershell
+make test
+make lint
+make format
+```
+
+若全量测试受外部环境阻塞，PR 描述必须列出具体命令、通过结果和阻塞原因。
+
+## 7. 验收门槛
+
+- [ ] 删除 `is_internal` 服务端直接赋值后，Gateway 防伪测试失败。
+- [ ] 删除 `configurable` 清理后，对应伪造测试失败。
+- [ ] 普通请求不能通过 `body.context` 或 `body.config` 注入 `channel_user_id`；internal
+      请求只接受顶层 `body.context` 的值。
+- [ ] 删除任意一段 task/executor identity wiring 后，subagent 测试失败。
+- [ ] builder、task、executor、middleware 对 attributes 非 Mapping 的行为一致。
+- [ ] adapter 同步和异步生成语义一致的完整 Principal。
+- [ ] `user=None` 的 internal 和 non-internal 分支都被固定。
+- [ ] 原有 Gateway、guardrail 和 subagent 测试继续通过。
+- [ ] `authorization.enabled: false` 时工具集合与执行决策不变。
+- [ ] 没有 Layer 1、Layer 2、RBAC 或 client 额外改动混入。
+- [ ] implementation notes 已追加决策记录。
+- [ ] `git diff --check`、ruff、目标测试和全量可运行测试通过。