feat(guardrails): expose authenticated runtime context in GuardrailRequest (#3665)

* docs: guardrail runtime attribution spec

* docs: guardrail request attribution implementation plan

* feat(guardrails): add runtime user context and attribution fields to GuardrailRequest

Extend GuardrailRequest with optional runtime attribution fields so that
pluggable GuardrailProviders can access authenticated user context and
tool-call-level attribution:

- Gateway injects user_role, oauth_provider, oauth_id into runtime context
  alongside the existing user_id (server-authenticated only, client spoofing
  prevented)
- GuardrailRequest gains: user_id, user_role, oauth_provider, oauth_id,
  run_id, tool_call_id (all optional, backward compatible)
- GuardrailMiddleware reads these from ToolCallRequest.runtime.context
- thread_id now actually populated from context (was always None before)
- Tests: 15 new/expanded tests covering Gateway injection, runtime context
  reading, partial/missing fields, and client spoofing prevention
- Docs: new Runtime Attribution section in GUARDRAILS.md with provider
  example and YAML policy illustration

* fix(guardrails): propagate attribution to subagents

* fix(guardrails): complete subagent attribution propagation

---------

Co-authored-by: Miracle778 <miracle778@no-reply.com>

backend/app/gateway/services.py

@@ -184,6 +184,9 @@ def inject_authenticated_user_context(config: dict[str, Any], request: Request)
     runtime_context = config.setdefault("context", {})
     if isinstance(runtime_context, dict):
         runtime_context["user_id"] = str(user_id)
+        runtime_context["user_role"] = getattr(user, "system_role", None)
+        runtime_context["oauth_provider"] = getattr(user, "oauth_provider", None)
+        runtime_context["oauth_id"] = getattr(user, "oauth_id", None)
 
 
 def resolve_agent_factory(assistant_id: str | None):

backend/docs/GUARDRAILS.md

@@ -232,6 +232,8 @@ class MyGuardrailProvider:
         return GuardrailDecision(allow=True, reasons=[GuardrailReason(code="oap.allowed")])
 
     async def aevaluate(self, request):
+        # This skeleton reuses the sync path. If policy evaluation performs
+        # async I/O, call and await the async evaluator here instead.
         return self.evaluate(request)
 ```
 
@@ -251,6 +253,185 @@ Make sure `my_guardrail.py` is on the Python path (e.g. in the backend directory
 3. Start DeerFlow and ask: "Use bash to delete test.txt"
 4. Your provider blocks it
 
+#### Optional: Runtime Attribution
+
+Runtime attribution fields are optional. Providers that need richer policy context or audit records can read them, while simple tool allow/deny providers can ignore them:
+
+| Field | Example use |
+|---|---|
+| `user_id` | Attach the authenticated DeerFlow user to a provider-side policy or audit record |
+| `user_role` | Apply simple role-based policy, such as allowing an admin-only tool. Sourced from the authenticated user's `system_role` (renamed for the guardrail-facing surface, not a separate field) |
+| `oauth_provider` | Link a decision to an external identity provider, when present |
+| `oauth_id` | Link a decision to the external provider's subject/user id, when present |
+| `thread_id` | Link a decision back to the conversation thread |
+| `run_id` | Link a decision back to one execution run |
+| `tool_call_id` | Identify the exact tool call that was allowed or denied |
+
+These fields are populated by the Gateway from server-side auth state (`inject_authenticated_user_context` writes `user_id`/`user_role`/`oauth_provider`/`oauth_id` from `request.state.user`; the run worker always sets `thread_id`/`run_id`), and propagated into subagent runs so delegated tool calls are evaluated with the same identity as the lead agent. Client-supplied values cannot override them — the server-side assignment wins.
+
+**Known limitation — IM / internal-auth runs.** `inject_authenticated_user_context` early-returns for the internal system role that IM channel workers (Slack, Discord, Telegram, Feishu, DingTalk) and other internal-auth callers are stamped with, so on those runs `user_role`/`oauth_provider`/`oauth_id` stay `None`. `user_id` still resolves (via the channel-bound owner), but role-based policy cannot be applied to channel-delivered work in this release. Web-authenticated runs are unaffected. Threaded alongside the owner `system_role` in a follow-up if channel-originated role policy is needed.
+
+For example, if your deployment has user-scoped policy requirements, you can opt into a context-aware provider that passes the runtime fields into an external policy file. This keeps business policy out of Python code and `config.yaml`; the provider only normalizes context, evaluates a configured policy, and maps the result back to `GuardrailDecision`.
+
+```python
+import asyncio
+import json
+from pathlib import Path
+
+from deerflow.guardrails.provider import GuardrailDecision, GuardrailReason
+
+
+class ContextAwareGuardrailProvider:
+    """Illustrative provider skeleton; policy loading/evaluation is provider-defined."""
+
+    name = "context-aware-example"
+
+    def __init__(self, *, policy_path, audit_path="./logs/guardrail-audit.jsonl", **kwargs):
+        self.policy_path = Path(policy_path)
+        self.audit_path = Path(audit_path)
+        # Load policy rules here. In a real deployment this could call an
+        # internal policy service, OPA/Cedar, AGT, or another rule engine.
+        self.policy = self._load_policy(self.policy_path)
+
+    def evaluate(self, request):
+        decision = self._decide(request)
+        self._write_audit(request, decision)
+        return decision
+
+    async def aevaluate(self, request):
+        # ``_decide`` is in-memory policy work; the audit write is blocking
+        # file I/O, so offload it off the event loop with ``asyncio.to_thread``
+        # (DeerFlow enforces a blocking-IO gate in CI). If your policy
+        # evaluation itself does blocking I/O — external policy service, file
+        # read per call — move that behind ``asyncio.to_thread`` too, or
+        # implement a native async evaluator and await it here.
+        decision = self._decide(request)
+        await asyncio.to_thread(self._write_audit, request, decision)
+        return decision
+
+    def _decide(self, request):
+        # 1. Normalize DeerFlow request data into policy context.
+        context = {
+            "tool_name": request.tool_name,
+            "tool_input": request.tool_input,
+            "user_id": request.user_id,
+            "user_role": request.user_role,
+            "oauth_provider": request.oauth_provider,
+            "oauth_id": request.oauth_id,
+            "thread_id": request.thread_id,
+            "run_id": request.run_id,
+            "tool_call_id": request.tool_call_id,
+            "agent_id": request.agent_id,
+            "timestamp": request.timestamp,
+            # Derived fields make simple rule engines handle multi-field checks.
+            # Example policy: allow bash only for admin users.
+            "role_tool_key": f"{request.user_role or ''}:{request.tool_name}",
+            "command": request.tool_input.get("command", ""),
+            "message": json.dumps(request.tool_input, ensure_ascii=False, default=str),
+        }
+
+        # 2. Evaluate the provider-defined policy schema.
+        result = self._evaluate_policy(self.policy, context)
+
+        # 3. Convert the policy result back to DeerFlow's decision object.
+        return GuardrailDecision(
+            allow=result["allow"],
+            reasons=[
+                GuardrailReason(
+                    code=result["code"],
+                    message=result["message"],
+                )
+            ],
+            policy_id=result.get("policy_id"),
+            metadata={
+                "user_id": request.user_id,
+                "user_role": request.user_role,
+                "oauth_provider": request.oauth_provider,
+                "oauth_id": request.oauth_id,
+                "thread_id": request.thread_id,
+                "run_id": request.run_id,
+                "tool_call_id": request.tool_call_id,
+            },
+        )
+
+    def _write_audit(self, request, decision):
+        event = {
+            "decision": "allow" if decision.allow else "deny",
+            "reason": decision.reasons[0].message if decision.reasons else "",
+            "policy_id": decision.policy_id,
+            "tool_name": request.tool_name,
+            "user_id": request.user_id,
+            "user_role": request.user_role,
+            "oauth_provider": request.oauth_provider,
+            "oauth_id": request.oauth_id,
+            "thread_id": request.thread_id,
+            "run_id": request.run_id,
+            "tool_call_id": request.tool_call_id,
+            "agent_id": request.agent_id,
+            "timestamp": request.timestamp,
+        }
+        self.audit_path.parent.mkdir(parents=True, exist_ok=True)
+        with self.audit_path.open("a", encoding="utf-8") as f:
+            f.write(json.dumps(event, ensure_ascii=False) + "\n")
+
+    def _load_policy(self, path):
+        # Load your provider-defined policy file.
+        raise NotImplementedError
+
+    def _evaluate_policy(self, policy, context):
+        # Evaluate ordered rules and return:
+        # {"allow": bool, "code": str, "message": str, "policy_id": str | None}
+        raise NotImplementedError
+```
+
+**config.yaml:**
+```yaml
+guardrails:
+  enabled: true
+  provider:
+    use: my_guardrail:ContextAwareGuardrailProvider
+    config:
+      policy_path: ./policies/guardrail-policy.yml
+      audit_path: ./logs/guardrail-audit.jsonl
+```
+
+Many policy engines use a similar shape: normalize request context, evaluate ordered rules, and return an allow/deny decision. The exact schema is provider-defined; the YAML below is illustrative:
+
+```yaml
+# policies/guardrail-policy.yml
+version: "1.0"
+rules:
+  - name: allow-admin-bash
+    condition:
+      field: role_tool_key
+      operator: eq
+      value: admin:bash
+    action: allow
+    priority: 300
+    message: Admin users may execute bash
+
+  - name: deny-bash-for-other-roles
+    condition:
+      field: tool_name
+      operator: eq
+      value: bash
+    action: deny
+    priority: 200
+    message: bash is restricted to admin users
+
+  - name: deny-dangerous-command
+    condition:
+      field: message
+      operator: matches
+      value: "\\brm\\s+-rf\\b"
+    action: deny
+    priority: 100
+    message: Dangerous shell command detected
+
+defaults:
+  action: allow
+```
+
 ## Implementing a Provider
 
 ### Required Interface
@@ -277,8 +458,14 @@ Make sure `my_guardrail.py` is on the Python path (e.g. in the backend directory
 │  thread_id: str | None    │    │  metadata: dict           │
 │  is_subagent: bool        │    │                           │
 │  timestamp: str           │    │  GuardrailReason:         │
-│                           │    │    code: str              │
-└──────────────────────────┘    │    message: str           │
+│  user_id: str | None      │    │    code: str              │
+│  user_role: str | None    │    │    message: str           │
+│  oauth_provider: str | None│   │                           │
+│  oauth_id: str | None     │    │                           │
+│  run_id: str | None       │    │                           │
+│  tool_call_id: str | None │    │                           │
+│                           │    │                           │
+└──────────────────────────┘    │                           │
                                 └──────────────────────────┘
 ```
 

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

@@ -32,11 +32,23 @@ class GuardrailMiddleware(AgentMiddleware[AgentState]):
         self.passport = passport
 
     def _build_request(self, request: ToolCallRequest) -> GuardrailRequest:
+        runtime = getattr(request, "runtime", None)
+        context = getattr(runtime, "context", None) if runtime is not None else None
+        context = context if isinstance(context, dict) else {}
+
         return GuardrailRequest(
             tool_name=str(request.tool_call.get("name", "")),
             tool_input=request.tool_call.get("args", {}),
             agent_id=self.passport,
+            thread_id=context.get("thread_id"),
+            is_subagent=bool(context.get("is_subagent")),
             timestamp=datetime.now(UTC).isoformat(),
+            user_id=context.get("user_id"),
+            user_role=context.get("user_role"),
+            oauth_provider=context.get("oauth_provider"),
+            oauth_id=context.get("oauth_id"),
+            run_id=context.get("run_id"),
+            tool_call_id=request.tool_call.get("id"),
         )
 
     def _build_denied_message(self, request: ToolCallRequest, decision: GuardrailDecision) -> ToolMessage:

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

@@ -16,6 +16,12 @@ class GuardrailRequest:
     thread_id: str | None = None
     is_subagent: bool = False
     timestamp: str = ""
+    user_id: str | None = None
+    user_role: str | None = None
+    oauth_provider: str | None = None
+    oauth_id: str | None = None
+    run_id: str | None = None
+    tool_call_id: str | None = None
 
 
 @dataclass

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

@@ -289,6 +289,10 @@ class SubagentExecutor:
         thread_id: str | None = None,
         trace_id: str | None = None,
         user_id: str | None = None,
+        user_role: str | None = None,
+        oauth_provider: str | None = None,
+        oauth_id: str | None = None,
+        run_id: str | None = None,
     ):
         """Initialize the executor.
 
@@ -305,6 +309,12 @@ class SubagentExecutor:
             trace_id: Trace ID from parent for distributed tracing.
             user_id: User ID captured from the parent tool's runtime context.
                 When None, the tracing layer falls back to DEFAULT_USER_ID.
+            user_role: Authenticated user's role, propagated so GuardrailMiddleware
+                on the subagent can apply role-aware policy to delegated calls.
+            oauth_provider: External identity provider, when authenticated via SSO.
+            oauth_id: Subject id at the external identity provider.
+            run_id: Parent run id, so delegated guardrail decisions attribute to
+                the same run as the lead agent.
         """
         self.config = config
         self.app_config = app_config
@@ -322,6 +332,11 @@ class SubagentExecutor:
         # Generate trace_id if not provided (for top-level calls)
         self.trace_id = trace_id or str(uuid.uuid4())[:8]
         self.user_id = user_id
+        # Guardrail attribution propagated from the parent runtime context.
+        self.user_role = user_role
+        self.oauth_provider = oauth_provider
+        self.oauth_id = oauth_id
+        self.run_id = run_id
 
         self._base_tools = _filter_tools(
             tools,
@@ -564,6 +579,16 @@ class SubagentExecutor:
                 context["thread_id"] = self.thread_id
             if self.app_config is not None:
                 context["app_config"] = self.app_config
+            # Propagate guardrail attribution so delegated tool calls are
+            # evaluated with the parent run's identity (role-aware policy,
+            # audit). user_id reuses the resolved tracing id; on every
+            # authenticated/IM path this equals the parent context value.
+            context["user_id"] = self.user_id
+            context["user_role"] = self.user_role
+            context["oauth_provider"] = self.oauth_provider
+            context["oauth_id"] = self.oauth_id
+            context["run_id"] = self.run_id
+            context["is_subagent"] = True
 
             logger.info(f"[trace={self.trace_id}] Subagent {self.config.name} starting async execution with max_turns={self.config.max_turns}")
 

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

@@ -274,6 +274,20 @@ async def task_tool(
     # Get user_id for tracing (uses standard resolution order)
     user_id = resolve_runtime_user_id(runtime)
 
+    # Propagate the authenticated runtime context so delegated tool calls are
+    # evaluated by GuardrailMiddleware with the same identity/attribution as
+    # the lead agent. Sourced from the server-side context written by
+    # inject_authenticated_user_context (and run_id by the run worker); stays
+    # None when absent (e.g. internal-auth runs) so guardrail behavior is
+    # unchanged. Without this, role-aware policy silently mis-attributes any
+    # tool call delegated to a subagent (user_role=None).
+    parent_context = runtime.context if runtime is not None else None
+    parent_context = parent_context if isinstance(parent_context, dict) else {}
+    user_role = parent_context.get("user_role")
+    oauth_provider = parent_context.get("oauth_provider")
+    oauth_id = parent_context.get("oauth_id")
+    run_id = parent_context.get("run_id")
+
     parent_available_skills = metadata.get("available_skills")
     if parent_available_skills is not None:
         overrides["skills"] = _merge_skill_allowlists(list(parent_available_skills), config.skills)
@@ -312,6 +326,10 @@ async def task_tool(
         "thread_id": thread_id,
         "trace_id": trace_id,
         "user_id": user_id,
+        "user_role": user_role,
+        "oauth_provider": oauth_provider,
+        "oauth_id": oauth_id,
+        "run_id": run_id,
     }
     if resolved_app_config is not None:
         executor_kwargs["app_config"] = resolved_app_config

backend/tests/test_guardrail_middleware.py

@@ -300,6 +300,145 @@ class TestGuardrailMiddleware:
             asyncio.run(run())
 
 
+class TestGuardrailRequestAttribution:
+    """Tests for GuardrailRequest runtime attribution fields."""
+
+    def _make_runtime_mock(self, context: dict | None = None):
+        runtime = MagicMock()
+        runtime.context = context
+        return runtime
+
+    def _make_request(self, runtime=None, tool_call: dict | None = None):
+        req = MagicMock()
+        req.runtime = runtime
+        req.tool_call = tool_call or {"name": "bash", "args": {}}
+        req.tool = None
+        req.state = {}
+        return req
+
+    def _capture_guardrail_request(self, req):
+        captured = {}
+
+        class CaptureProvider:
+            name = "capture"
+
+            def evaluate(self, request):
+                captured["request"] = request
+                return GuardrailDecision(allow=True)
+
+            async def aevaluate(self, request):
+                return self.evaluate(request)
+
+        mw = GuardrailMiddleware(CaptureProvider())
+        mw.wrap_tool_call(req, MagicMock())
+        return captured["request"]
+
+    def test_no_attribution_fields_are_none(self):
+        req = self._make_request(runtime=None, tool_call={"name": "bash", "args": {}})
+
+        guardrail_request = self._capture_guardrail_request(req)
+
+        assert guardrail_request.user_id is None
+        assert guardrail_request.user_role is None
+        assert guardrail_request.oauth_provider is None
+        assert guardrail_request.oauth_id is None
+        assert guardrail_request.run_id is None
+        assert guardrail_request.tool_call_id is None
+
+    def test_only_user_id_present(self):
+        runtime = self._make_runtime_mock(context={"user_id": "user_abc"})
+        req = self._make_request(runtime=runtime, tool_call={"name": "bash", "args": {}})
+
+        guardrail_request = self._capture_guardrail_request(req)
+
+        assert guardrail_request.user_id == "user_abc"
+        assert guardrail_request.user_role is None
+        assert guardrail_request.oauth_provider is None
+        assert guardrail_request.oauth_id is None
+        assert guardrail_request.run_id is None
+        assert guardrail_request.tool_call_id is None
+
+    def test_authenticated_user_context_present(self):
+        runtime = self._make_runtime_mock(
+            context={
+                "user_id": "user_abc",
+                "user_role": "admin",
+                "oauth_provider": "github",
+                "oauth_id": "gh_123",
+            }
+        )
+        req = self._make_request(runtime=runtime, tool_call={"name": "bash", "args": {}})
+
+        guardrail_request = self._capture_guardrail_request(req)
+
+        assert guardrail_request.user_id == "user_abc"
+        assert guardrail_request.user_role == "admin"
+        assert guardrail_request.oauth_provider == "github"
+        assert guardrail_request.oauth_id == "gh_123"
+
+    def test_only_run_id_present(self):
+        runtime = self._make_runtime_mock(context={"run_id": "run_xyz"})
+        req = self._make_request(runtime=runtime, tool_call={"name": "bash", "args": {}})
+
+        guardrail_request = self._capture_guardrail_request(req)
+
+        assert guardrail_request.user_id is None
+        assert guardrail_request.run_id == "run_xyz"
+        assert guardrail_request.tool_call_id is None
+
+    def test_only_tool_call_id_present(self):
+        req = self._make_request(runtime=None, tool_call={"name": "web_search", "args": {"query": "test"}, "id": "call_42"})
+
+        guardrail_request = self._capture_guardrail_request(req)
+
+        assert guardrail_request.user_id is None
+        assert guardrail_request.run_id is None
+        assert guardrail_request.tool_call_id == "call_42"
+
+    def test_all_attribution_fields_present(self):
+        runtime = self._make_runtime_mock(
+            context={
+                "user_id": "user_abc",
+                "user_role": "user",
+                "oauth_provider": "google",
+                "oauth_id": "google_123",
+                "run_id": "run_xyz",
+                "is_subagent": True,
+            }
+        )
+        req = self._make_request(runtime=runtime, tool_call={"name": "bash", "args": {}, "id": "call_all"})
+
+        guardrail_request = self._capture_guardrail_request(req)
+
+        assert guardrail_request.user_id == "user_abc"
+        assert guardrail_request.user_role == "user"
+        assert guardrail_request.oauth_provider == "google"
+        assert guardrail_request.oauth_id == "google_123"
+        assert guardrail_request.run_id == "run_xyz"
+        assert guardrail_request.tool_call_id == "call_all"
+        assert guardrail_request.is_subagent is True
+
+    def test_partial_attribution_fields_present(self):
+        runtime = self._make_runtime_mock(context={"user_id": "user_partial"})
+        req = self._make_request(runtime=runtime, tool_call={"name": "bash", "args": {}, "id": "call_partial"})
+
+        guardrail_request = self._capture_guardrail_request(req)
+
+        assert guardrail_request.user_id == "user_partial"
+        assert guardrail_request.run_id is None
+        assert guardrail_request.tool_call_id == "call_partial"
+
+    def test_empty_context_with_tool_call(self):
+        runtime = self._make_runtime_mock(context={})
+        req = self._make_request(runtime=runtime, tool_call={"name": "bash", "args": {}, "id": "call_empty_context"})
+
+        guardrail_request = self._capture_guardrail_request(req)
+
+        assert guardrail_request.user_id is None
+        assert guardrail_request.run_id is None
+        assert guardrail_request.tool_call_id == "call_empty_context"
+
+
 # --- Config tests ---
 
 

backend/tests/test_setup_agent_e2e_user_isolation.py

@@ -46,13 +46,25 @@ from deerflow.runtime.runs.worker import _build_runtime_context, _install_runtim
 # ---------------------------------------------------------------------------
 
 
-def _make_request(user_id_str: str | None) -> SimpleNamespace:
+def _make_request(
+    user_id_str: str | None,
+    *,
+    system_role: str = "user",
+    oauth_provider: str | None = None,
+    oauth_id: str | None = None,
+) -> SimpleNamespace:
     """Build a fake FastAPI Request that carries an authenticated user."""
     if user_id_str is None:
         user = None
     else:
         # User.id is UUID in production; honour that
-        user = SimpleNamespace(id=UUID(user_id_str), email="alice@local")
+        user = SimpleNamespace(
+            id=UUID(user_id_str),
+            email="alice@local",
+            system_role=system_role,
+            oauth_provider=oauth_provider,
+            oauth_id=oauth_id,
+        )
     return SimpleNamespace(state=SimpleNamespace(user=user))
 
 
@@ -61,13 +73,24 @@ def _assemble_config(
     body_config: dict | None,
     body_context: dict | None,
     request_user_id: str | None,
+    request_user_role: str = "user",
+    request_oauth_provider: str | None = None,
+    request_oauth_id: str | None = None,
     thread_id: str = "thread-e2e",
     assistant_id: str = "lead_agent",
 ) -> dict:
     """Replay the **exact** start_run config-assembly sequence."""
     config = build_run_config(thread_id, body_config, None, assistant_id=assistant_id)
     merge_run_context_overrides(config, body_context)
-    inject_authenticated_user_context(config, _make_request(request_user_id))
+    inject_authenticated_user_context(
+        config,
+        _make_request(
+            request_user_id,
+            system_role=request_user_role,
+            oauth_provider=request_oauth_provider,
+            oauth_id=request_oauth_id,
+        ),
+    )
     return config
 
 
@@ -111,6 +134,22 @@ class TestConfigAssembly:
         runtime_ctx = _build_runtime_context("thread-e2e", "run-1", config.get("context"), None)
         assert runtime_ctx["user_id"] == "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"
 
+    def test_authenticated_user_context_includes_role_and_oauth_identity(self):
+        """Server-authenticated user attributes should reach runtime.context."""
+        config = _assemble_config(
+            body_config={"recursion_limit": 1000},
+            body_context=None,
+            request_user_id="aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee",
+            request_user_role="admin",
+            request_oauth_provider="github",
+            request_oauth_id="gh_123",
+        )
+        runtime_ctx = _build_runtime_context("thread-e2e", "run-1", config.get("context"), None)
+        assert runtime_ctx["user_id"] == "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"
+        assert runtime_ctx["user_role"] == "admin"
+        assert runtime_ctx["oauth_provider"] == "github"
+        assert runtime_ctx["oauth_id"] == "gh_123"
+
     def test_body_context_empty_dict_still_injects_user_id(self):
         """body.context={} (falsy) path: inject must still produce user_id."""
         config = _assemble_config(
@@ -131,15 +170,60 @@ class TestConfigAssembly:
         runtime_ctx = _build_runtime_context("thread-e2e", "run-1", config.get("context"), None)
         assert runtime_ctx["user_id"] == "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"
 
-    def test_client_supplied_user_id_is_overridden(self):
-        """Spoofed client user_id must be overwritten by inject (auth-trusted source)."""
+    def test_body_context_user_id_is_overridden(self):
+        """``body.context`` may carry a legacy/non-web user_id, but server auth wins.
+
+        This covers the whitelisted ``body.context`` merge path only. Full
+        identity spoofing coverage lives in the ``body.config.context`` test
+        below, because that path copies arbitrary context keys before inject
+        overwrites them.
+        """
         config = _assemble_config(
             body_config={"recursion_limit": 1000},
-            body_context={"agent_name": "myagent", "user_id": "spoofed"},
+            body_context={
+                "agent_name": "myagent",
+                "user_id": "spoofed",
+            },
             request_user_id="11111111-2222-3333-4444-555555555555",
+            request_user_role="user",
         )
         runtime_ctx = _build_runtime_context("thread-e2e", "run-1", config.get("context"), None)
         assert runtime_ctx["user_id"] == "11111111-2222-3333-4444-555555555555"
+        assert runtime_ctx["user_role"] == "user"
+        assert runtime_ctx["oauth_provider"] is None
+        assert runtime_ctx["oauth_id"] is None
+
+    def test_spoofed_context_in_body_config_is_overridden_by_inject(self):
+        """The real spoofing vector is ``body.config.context``: ``build_run_config``
+        copies it wholesale (no whitelist, unlike ``body.context``), so only
+        ``inject_authenticated_user_context``'s unconditional assignment can
+        defeat a client that spoofs ``user_id``/``user_role``/``oauth_*`` there.
+
+        The companion test above covers only ``body.context.user_id``. This
+        test spoofs via ``body.config.context`` so all spoofed values actually
+        reach ``config['context']`` and ``inject``'s overwrite is the only thing
+        standing between them and ``runtime_ctx``.
+        """
+        config = _assemble_config(
+            body_config={
+                "context": {
+                    "user_id": "spoofed-id",
+                    "user_role": "admin",
+                    "oauth_provider": "spoofed-provider",
+                    "oauth_id": "spoofed-subject",
+                },
+            },
+            body_context=None,
+            request_user_id="11111111-2222-3333-4444-555555555555",
+            request_user_role="user",
+            request_oauth_provider="keycloak",
+            request_oauth_id="real-subject",
+        )
+        runtime_ctx = _build_runtime_context("thread-e2e", "run-1", config.get("context"), None)
+        assert runtime_ctx["user_id"] == "11111111-2222-3333-4444-555555555555"
+        assert runtime_ctx["user_role"] == "user"
+        assert runtime_ctx["oauth_provider"] == "keycloak"
+        assert runtime_ctx["oauth_id"] == "real-subject"
 
     def test_unauthenticated_request_does_not_inject(self):
         """If request.state.user is missing (impossible under fail-closed auth, but

backend/tests/test_subagent_executor.py

@@ -2250,3 +2250,114 @@ class TestSubagentTracingWiring:
         from langchain_core.messages import HumanMessage
 
         return ({"messages": [HumanMessage(content=task)]}, [], None)
+
+
+class TestSubagentGuardrailAttribution:
+    """GuardrailMiddleware runs on subagents too, so the authenticated runtime
+    context captured at the lead-agent layer must reach the subagent's own
+    ``astream`` context — otherwise delegated tool calls are evaluated with
+    ``user_role=None`` and role-aware policy silently mis-attributes them.
+    """
+
+    @pytest.fixture
+    def executor_module(self, _setup_executor_classes):
+        executor = importlib.import_module("deerflow.subagents.executor")
+        return _patch_default_get_app_config(importlib.reload(executor))
+
+    def _make_executor(
+        self,
+        classes,
+        *,
+        user_id=None,
+        user_role=None,
+        oauth_provider=None,
+        oauth_id=None,
+        run_id=None,
+        name="general-purpose",
+        parent_model="test-model",
+    ):
+        SubagentExecutor = classes["SubagentExecutor"]
+        SubagentConfig = classes["SubagentConfig"]
+        config = SubagentConfig(
+            name=name,
+            description="Guardrail attribution test agent",
+            system_prompt="You are a guardrail attribution test agent.",
+            max_turns=5,
+            timeout_seconds=30,
+        )
+        return SubagentExecutor(
+            config=config,
+            tools=[],
+            parent_model=parent_model,
+            thread_id="thread-attrib-1",
+            trace_id="trace-attrib-1",
+            user_id=user_id,
+            user_role=user_role,
+            oauth_provider=oauth_provider,
+            oauth_id=oauth_id,
+            run_id=run_id,
+        )
+
+    @pytest.mark.anyio
+    async def test_aexecute_propagates_attribution_to_subagent_context(
+        self,
+        classes,
+        executor_module,
+        monkeypatch,
+    ):
+        """The authenticated runtime context captured at task_tool must reach
+        the subagent's ``astream`` context so GuardrailMiddleware sees the
+        same identity/attribution as the lead agent.
+        """
+        executor = self._make_executor(
+            classes,
+            user_id="alice",
+            user_role="admin",
+            oauth_provider="keycloak",
+            oauth_id="subj-123",
+            run_id="run-42",
+        )
+        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, "subagent context must be passed to astream"
+        assert context.get("user_id") == "alice"
+        assert context.get("user_role") == "admin"
+        assert context.get("oauth_provider") == "keycloak"
+        assert context.get("oauth_id") == "subj-123"
+        assert context.get("run_id") == "run-42"
+        assert context.get("is_subagent") is True
+
+    @pytest.mark.anyio
+    async def test_aexecute_context_defaults_to_none_when_attribution_absent(
+        self,
+        classes,
+        executor_module,
+        monkeypatch,
+    ):
+        """When no authenticated context is propagated (e.g. internal-auth
+        runs), the subagent context still carries the attribution keys as
+        None so GuardrailRequest fields stay None rather than KeyError-ing.
+        """
+        executor = self._make_executor(classes)
+        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("user_role") is None
+        assert context.get("oauth_provider") is None
+        assert context.get("oauth_id") is None
+        assert context.get("run_id") is None
+
+    async def _noop_build_initial_state(self, task):  # noqa: ARG002 - signature parity
+        from langchain_core.messages import HumanMessage
+
+        return ({"messages": [HumanMessage(content=task)]}, [], None)

docs/superpowers/plans/2026-06-19-guardrail-request-attribution.md

@@ -0,0 +1,186 @@
+# GuardrailRequest 运行时用户上下文与归因字段补充 — 实现计划
+
+**Goal:** 将服务端认证后的用户上下文和运行时归因信息传递给 `GuardrailProvider`：`user_id`、`user_role`、`oauth_provider`、`oauth_id`、`thread_id`、`run_id`、`tool_call_id`。
+
+**Architecture:** Gateway 只从 `request.state.user` 注入可信用户上下文；run worker 复用现有 `runtime.context` 传递机制；GuardrailMiddleware 只消费 `ToolCallRequest.runtime.context`，不依赖 FastAPI request、DB 或 auth repository。
+
+**Tech Stack:** Python 3.12+, dataclasses, LangGraph `ToolCallRequest.runtime.context`, pytest
+
+---
+
+## 文件结构
+
+| 操作 | 文件 | 职责 |
+|------|------|------|
+| Modify | `backend/app/gateway/services.py` | 注入 authenticated user context |
+| Modify | `backend/packages/harness/deerflow/guardrails/provider.py` | `GuardrailRequest` 扩字段 |
+| Modify | `backend/packages/harness/deerflow/guardrails/middleware.py` | 从 runtime context 填充字段 |
+| Modify | `backend/tests/test_setup_agent_e2e_user_isolation.py` | 覆盖 Gateway → runtime context |
+| Modify | `backend/tests/test_guardrail_middleware.py` | 覆盖 runtime context → GuardrailRequest |
+| Modify | `backend/docs/GUARDRAILS.md` | 更新 custom provider 示例 |
+
+---
+
+## Task 1: Gateway 注入 authenticated user context
+
+**Files:**
+- Modify: `backend/app/gateway/services.py`
+
+- [x] **Step 1: 扩展 `inject_authenticated_user_context()`**
+
+在已有 `runtime_context["user_id"] = str(user_id)` 后追加：
+
+```python
+runtime_context["user_role"] = getattr(user, "system_role", None)
+runtime_context["oauth_provider"] = getattr(user, "oauth_provider", None)
+runtime_context["oauth_id"] = getattr(user, "oauth_id", None)
+```
+
+**约束：**
+- 只读取 `request.state.user`，不信任 client `body.context`。
+- `INTERNAL_SYSTEM_ROLE` 仍保持跳过注入。
+- local user 的 OAuth 字段允许为 `None`。
+
+- [x] **Step 2: 扩展 config assembly 测试**
+
+在 `TestConfigAssembly` 中覆盖：
+
+- authenticated user 的 `user_role/oauth_provider/oauth_id` 进入 runtime context。
+- client-supplied `user_id/user_role/oauth_provider/oauth_id` 不覆盖 server-authenticated user。
+
+---
+
+## Task 2: GuardrailRequest 扩展字段
+
+**Files:**
+- Modify: `backend/packages/harness/deerflow/guardrails/provider.py`
+
+- [x] **Step 1: 在 `GuardrailRequest` 末尾新增 optional 字段**
+
+```python
+user_id: str | None = None
+user_role: str | None = None
+oauth_provider: str | None = None
+oauth_id: str | None = None
+run_id: str | None = None
+tool_call_id: str | None = None
+```
+
+**约束：**
+- 不修改 `GuardrailDecision`。
+- 不修改现有 provider protocol。
+- 所有字段 optional，保持向后兼容。
+
+---
+
+## Task 3: GuardrailMiddleware 从 runtime context 填充
+
+**Files:**
+- Modify: `backend/packages/harness/deerflow/guardrails/middleware.py`
+
+- [x] **Step 1: 安全读取 `ToolCallRequest.runtime.context`**
+
+```python
+runtime = getattr(request, "runtime", None)
+context = getattr(runtime, "context", None) if runtime is not None else None
+context = context if isinstance(context, dict) else {}
+```
+
+- [x] **Step 2: 构造 `GuardrailRequest` 时填充字段**
+
+```python
+GuardrailRequest(
+    tool_name=str(request.tool_call.get("name", "")),
+    tool_input=request.tool_call.get("args", {}),
+    agent_id=self.passport,
+    thread_id=context.get("thread_id"),
+    timestamp=datetime.now(UTC).isoformat(),
+    user_id=context.get("user_id"),
+    user_role=context.get("user_role"),
+    oauth_provider=context.get("oauth_provider"),
+    oauth_id=context.get("oauth_id"),
+    run_id=context.get("run_id"),
+    tool_call_id=request.tool_call.get("id"),
+)
+```
+
+**约束：**
+- runtime/context 缺失时字段为 `None`。
+- `thread_id` 是已有字段，本次补填。
+- 不在 GuardrailMiddleware 中访问 request/auth/DB。
+
+---
+
+## Task 4: 文档与示例
+
+**Files:**
+- Modify: `backend/docs/GUARDRAILS.md`
+
+- [x] **Step 1: 增加 Runtime Attribution 小节**
+
+说明以下字段及用途：
+
+- `user_id`：稳定审计归因。
+- `user_role`：简单 role-based policy。
+- `oauth_provider/oauth_id`：外部身份 provider/subject 映射，local user 可为空。
+- `thread_id/run_id/tool_call_id`：run/tool-call 定位。
+
+- [x] **Step 2: 更新 Context-Aware Provider 示例**
+
+示例 provider 只接收 `policy_path/audit_path`，把 `GuardrailRequest` 归一化成 provider-defined policy context，并展示：
+
+```python
+"role_tool_key": f"{request.user_role or ''}:{request.tool_name}"
+```
+
+示例 policy 使用：
+
+```yaml
+field: role_tool_key
+operator: eq
+value: admin:bash
+```
+
+**文案约束：**
+- 不声称 DeerFlow 内置该 YAML schema。
+- 不写成对标 AGT。
+- AGT/OPA/Cedar 只作为可选 policy engine 示例。
+
+---
+
+## Task 5: 测试与验证
+
+- [x] **Step 1: 运行目标测试**
+
+```bash
+cd backend
+PYTHONPATH=. uv run pytest \
+  tests/test_guardrail_middleware.py::TestGuardrailRequestAttribution \
+  tests/test_setup_agent_e2e_user_isolation.py::TestConfigAssembly -v
+```
+
+Expected:
+
+```text
+15 passed
+```
+
+- [x] **Step 2: 验证覆盖点**
+
+| 测试 | 覆盖 |
+|------|------|
+| `test_authenticated_user_context_includes_role_and_oauth_identity` | Gateway 注入完整用户上下文 |
+| `test_client_supplied_user_id_is_overridden` | 防止客户端伪造身份字段 |
+| `test_authenticated_user_context_present` | GuardrailRequest 接收用户上下文 |
+| `test_all_attribution_fields_present` | 用户上下文与 run/tool-call 归因同时存在 |
+| missing context tests | 字段缺失时为 `None` |
+
+---
+
+## 自检清单
+
+1. **Scope:** 改动保持在 authenticated runtime context 与 Guardrail provider 入参，不新增治理系统。
+2. **Security:** `user_role/oauth_provider/oauth_id` 只从服务端认证态注入，不信任客户端 context。
+3. **Compatibility:** 新字段全部 optional；现有 provider 不读取时行为不变。
+4. **Docs:** Guardrails 文档示例使用 role-based policy，不再建议手写 UUID 策略。
+5. **Tests:** 目标测试已通过：`15 passed`。

docs/superpowers/specs/2026-06-19-guardrail-request-attribution-design.md

@@ -0,0 +1,134 @@
+# GuardrailRequest 运行时用户上下文与归因字段补充
+
+## 概述
+
+为 `GuardrailRequest` 补充可选的运行时用户上下文和工具调用归因字段，使可插拔 `GuardrailProvider` 能访问 DeerFlow 已认证用户、外部身份映射、run/thread/tool-call 定位信息。
+
+本设计不新增治理系统、不定义统一 policy schema，也不改变默认 allow/deny 行为。它只把 DeerFlow 运行时已经掌握的上下文传给 provider。
+
+## 背景
+
+DeerFlow 已通过 `GuardrailMiddleware` + 可插拔 `GuardrailProvider` 实现工具调用前授权。当前 `GuardrailDecision` 已能表达 allow/deny、原因、policy_id 和 metadata；缺口在 `GuardrailRequest` 侧：provider 只能看到 `tool_name` 和 `tool_input`，无法可靠知道“谁发起了这次工具调用”以及“这次调用属于哪个 run/tool_call”。
+
+源码中 DeerFlow 已有用户身份模型：
+
+- `users.id`：DeerFlow 内部稳定用户 ID。
+- `users.system_role`：`admin` / `user`。
+- `users.oauth_provider`、`users.oauth_id`：未来 OAuth/SSO 外部身份映射字段，local user 下可为空。
+
+这些字段当前已存在于认证后的 `request.state.user`，但 run-time middleware/tool 阶段只能通过 `runtime.context` 访问上下文。因此应在 Gateway 构建 run config 时注入 server-authenticated user context，再由 GuardrailMiddleware 消费。
+
+## 改动范围
+
+- `app/gateway/services.py`：`inject_authenticated_user_context()` 注入更多 authenticated user context。
+- `guardrails/provider.py`：`GuardrailRequest` 新增 optional 字段。
+- `guardrails/middleware.py`：从 `ToolCallRequest.runtime.context` 读取字段。
+- `tests/test_setup_agent_e2e_user_isolation.py`：验证 Gateway 注入到 runtime context。
+- `tests/test_guardrail_middleware.py`：验证 runtime context 进入 `GuardrailRequest`。
+- `backend/docs/GUARDRAILS.md`：更新 custom provider 示例。
+
+## 设计
+
+### GuardrailRequest 字段
+
+```python
+@dataclass
+class GuardrailRequest:
+    tool_name: str
+    tool_input: dict[str, Any]
+    agent_id: str | None = None
+    thread_id: str | None = None
+    is_subagent: bool = False
+    timestamp: str = ""
+
+    user_id: str | None = None
+    user_role: str | None = None
+    oauth_provider: str | None = None
+    oauth_id: str | None = None
+    run_id: str | None = None
+    tool_call_id: str | None = None
+```
+
+所有新增字段均为 optional，缺失时保持 `None`。`GuardrailDecision` 不变。
+
+### 字段来源
+
+| 字段 | 来源 | 说明 |
+|------|------|------|
+| `user_id` | `request.state.user.id` → `runtime.context["user_id"]` | DeerFlow 内部稳定用户 ID |
+| `user_role` | `request.state.user.system_role` → `runtime.context["user_role"]` | 可用于简单 role-based policy |
+| `oauth_provider` | `request.state.user.oauth_provider` → `runtime.context["oauth_provider"]` | OAuth/SSO 外部 provider，local user 可为空 |
+| `oauth_id` | `request.state.user.oauth_id` → `runtime.context["oauth_id"]` | 外部 provider subject/user id，local user 可为空 |
+| `run_id` | `_build_runtime_context()` 写入 `runtime.context["run_id"]` | run 级审计归因 |
+| `thread_id` | `_build_runtime_context()` 写入 `runtime.context["thread_id"]` | 修正已有字段未填充问题 |
+| `tool_call_id` | `request.tool_call.get("id")` | 单次 tool call 定位 |
+
+Gateway 注入只信任服务端认证态 `request.state.user`。客户端 `body.context` 里的 `user_id/user_role/oauth_*` 不应覆盖 authenticated user。
+
+## 收益
+
+### 稳定审计归因
+
+`user_id/run_id/thread_id/tool_call_id` 让 provider 或外部审计系统能回答：
+
+- 哪个 DeerFlow 用户触发了 tool call？
+- 哪个 run 里发生了 deny？
+- 同一轮中多次同名工具调用时，具体是哪一次？
+
+### 可读的本地策略示例
+
+`user_id` 是 UUID，不适合直接写人工 policy。`user_role` 可以支持简单示例：
+
+```yaml
+field: role_tool_key
+operator: eq
+value: admin:bash
+```
+
+provider 可派生：
+
+```python
+role_tool_key = f"{request.user_role or ''}:{request.tool_name}"
+```
+
+### 外部身份映射
+
+`oauth_provider/oauth_id` 保留了未来接 OAuth/SSO/IAM 时的外部 subject 信息。当前 OAuth 路由仍是 placeholder，local user 下这些字段通常为 `None`，但字段 optional，不影响现有部署。
+
+## 兼容性
+
+- 所有新增字段 optional。
+- 现有 provider 不读取新字段时行为不变。
+- `GuardrailDecision` 不变。
+- 未认证或无 runtime context 时字段为 `None`。
+- local user 没有 OAuth 信息时 `oauth_provider/oauth_id` 为 `None`。
+- `agent_id` 语义不变，仍保持现有 passport/agent hint 含义。
+
+## 测试
+
+新增或扩展以下测试：
+
+| 测试 | 覆盖 |
+|------|------|
+| `TestConfigAssembly::test_authenticated_user_context_includes_role_and_oauth_identity` | Gateway 将 `user_id/user_role/oauth_provider/oauth_id` 注入 runtime context |
+| `TestConfigAssembly::test_client_supplied_user_id_is_overridden` | 客户端伪造 identity context 不覆盖服务端认证态 |
+| `TestGuardrailRequestAttribution::test_authenticated_user_context_present` | `runtime.context` 中用户上下文进入 `GuardrailRequest` |
+| `TestGuardrailRequestAttribution::test_all_attribution_fields_present` | 用户上下文 + run/tool_call 归因字段同时传递 |
+| 缺失 runtime/context 测试 | 字段保持 `None`，向后兼容 |
+
+验证命令：
+
+```bash
+cd backend
+PYTHONPATH=. uv run pytest \
+  tests/test_guardrail_middleware.py::TestGuardrailRequestAttribution \
+  tests/test_setup_agent_e2e_user_isolation.py::TestConfigAssembly -v
+```
+
+## 未涉及
+
+- 不新增 central governance subsystem。
+- 不新增 DeerFlow 内置 policy schema。
+- 不修改 MCP 配置机制。
+- 不修改 OAuth/SSO 实现状态。
+- 不让 GuardrailMiddleware 直接依赖 FastAPI request、DB 或 auth repository。