From 3bccd1474f06ac0c4deada48365741f217cc0025 Mon Sep 17 00:00:00 2001 From: Beautyl0ve <74452755+Beautyl0ve@users.noreply.github.com> Date: Sun, 6 Sep 2026 22:33:12 +0800 Subject: [PATCH] fix(client): scope embedded agent reuse by effective user (#5206) Signed-off-by: Beautyl0ve <74452755+Beautyl0ve@users.noreply.github.com> --- backend/packages/harness/deerflow/AGENTS.md | 1 + .../deerflow/agents/middlewares/AGENTS.md | 2 +- backend/packages/harness/deerflow/client.py | 21 ++++++---- backend/tests/test_client.py | 40 ++++++++++++++++++- 4 files changed, 54 insertions(+), 10 deletions(-) diff --git a/backend/packages/harness/deerflow/AGENTS.md b/backend/packages/harness/deerflow/AGENTS.md index 06286bb0c..13cb70a2e 100644 --- a/backend/packages/harness/deerflow/AGENTS.md +++ b/backend/packages/harness/deerflow/AGENTS.md @@ -61,6 +61,7 @@ drift. - `"end"` — stream finished (carries cumulative `usage` counted once per message id) - **Custom-event invariant** — production DeerFlow emitters must use `emit_custom_event` / `aemit_custom_event`, not call `StreamWriter` alone. Every built-in payload must carry a non-empty string `type`; typeless payloads remain writer-only and are intentionally absent from `astream_events`. The writer runs first and remains authoritative for Gateway, Web UI, and embedded-client compatibility; callback dispatch is best-effort and must not break that path. Async graph hooks must await the async helper rather than invoking synchronous dispatch on a running event loop. - Agent created lazily via `create_agent()` + `build_middlewares()`, same as `make_lead_agent` +- Cache graphs by effective storage `user_id` in every auth mode because prompts and middleware bind user SOUL, skills, and storage. `stream()` must materialize it before worker or isolated-loop boundaries. - Supports `checkpointer` parameter for state persistence across turns - `reset_agent()` forces agent recreation (e.g. after memory or skill changes) - See [docs/STREAMING.md](../../../docs/STREAMING.md) for the full design: why Gateway and DeerFlowClient are parallel paths, LangGraph's `stream_mode` semantics, the per-id dedup invariants, and regression testing strategy diff --git a/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md b/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md index d6c13500e..6935ac1df 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md +++ b/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md @@ -69,7 +69,7 @@ it to that middleware's declaration in the same change. 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. **ToolReceiptMiddleware + ToolErrorHandlingMiddleware** - `ToolReceiptMiddleware` is *(optional, if `verification.receipts_enabled`, default on)*. It is the **outermost `wrap_tool_call` layer** — registered ahead of entries 9-12 — because Guardrail/SandboxAudit/ReadBeforeWrite/ToolProgress can short-circuit a call with their own ToolMessage (and SandboxAudit rebuilds medium-risk results); an inner receipt layer would silently gap the ledger on those results (ordering constraints in `deerflow.extensions.ordering`). Normal results still carry the `deerflow_tool_meta` status ToolErrorHandlingMiddleware stamps on the inner return path; short-circuit messages self-stamp meta or fall back to `message.status`. It stamps deterministic provenance (tool name, status, args/output hashes, byte count, timestamp) onto direct `ToolMessage` results and every matching `ToolMessage` carried in `Command.update.messages`, including delegated `task`, `present_file`, `view_image`, and `tool_search` results; before model calls it derives a hidden receipt ledger (display ids r1..rN) from message state, and when the 2,000-character budget is exceeded the newest receipts are retained in chronological order with their original ids plus an older-receipts omission marker. Rendering returns both the text and its retained receipt subset; every response that received a ledger carries only that exact server-owned subset, never omitted receipts. Snapshot validation accepts a strictly consecutive positive original-id range (for example `r24`–`r30`) rather than requiring `r1`, so subagent terminal citation verification resolves ids against evidence present in the citing turn even when later summarization drops and renumbers tool messages. Model-generated citation IDs are digit-bounded before integer conversion; oversized IDs are ignored as malformed input rather than raising through task write-back. Citation parsing deduplicates exact `(id, anchor)` pairs, not IDs alone, so repeated identical references stay compact while every distinct anchor claim is verified. Gateway strips delegated receipts/verdicts from external messages. `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`. The built-in RBAC provider validates `authorization.default_role` during provider resolution so an unknown fallback role fails agent construction instead of degrading into an empty tool set. Task delegation carries `is_internal` plus copied attributes through `SubagentExecutor`, while `GuardrailMiddleware` maps the same runtime fields into `GuardrailRequest`. Phase 1B applies Layer 1 before deferred-tool assembly on the lead, native-subagent, and embedded-client paths, then passes the same provider instance into Layer 2. Framework-provided `describe_skill` and memory tools are included in Layer 1 but restored to their legacy post-`tool_search` ordering afterward. `DeerFlowClient.stream()` treats its in-process caller as trusted and accepts the same identity fields as keyword overrides; it includes the complete Principal in its agent cache key and deep-copies nested attributes so caller mutation cannot make a stale tool set look current. +Authorization identity is independent of enforcement. Gateway strips client identity overrides: only the server auth source sets `is_internal`, and only authenticated IM `body.context` supplies `channel_user_id` (never `body.config`). `build_principal_from_context` applies role defaults, strict provenance, and copied attributes; RBAC rejects unknown defaults. Delegation and `GuardrailMiddleware` share this identity. Layer 1 precedes deferred assembly across agent paths and its provider is reused for Layer 2; framework skill/memory ordering stays stable. Trusted `DeerFlowClient.stream()` accepts identity overrides. Its graph key always includes effective storage `user_id` and, when enforced, the full Principal; nested attributes are copied so mutation cannot hide stale cache state. Gateway route authorization uses `authz.py::resolve_route_permissions()` as the single provider integration point for both `AuthMiddleware` and decorator-only authentication. When enabled, it evaluates the six registered `threads:*` / `runs:*` permissions as `resource="route"` requests whose targets are the full `resource:action` strings. Decisions use the async provider API and are cached for the request in `AuthContext`; decorators do not call the provider again. Provider resolution or decision errors follow `authorization.fail_closed`, scoped per permission for decision errors. When authorization is disabled, the legacy complete permission set is returned without resolving a provider. Existing `owner_check` enforcement and `require_admin_user()` management gates remain independent and unchanged. Tests: `tests/test_authorization_route_permissions.py`, `tests/test_auth.py`, and `tests/test_auth_middleware.py`. diff --git a/backend/packages/harness/deerflow/client.py b/backend/packages/harness/deerflow/client.py index 6450c46ad..745ec6892 100644 --- a/backend/packages/harness/deerflow/client.py +++ b/backend/packages/harness/deerflow/client.py @@ -280,6 +280,13 @@ class DeerFlowClient: if context is not None: cfg.update(context) + # Prompt and middleware assembly bind user-scoped SOUL, skills, and + # storage even when authorization enforcement is disabled. Keep that + # storage identity in the graph cache key independently of the + # authorization principal so one trusted embedded client can safely + # serve more than one caller. + effective_user_id = cfg.get("user_id") or get_effective_user_id() + authorization_identity = None if self._app_config.authorization.enabled: principal = build_principal_from_context( @@ -306,6 +313,7 @@ class DeerFlowClient: frozenset(self._available_skills) if self._available_skills is not None else None, self._checkpoint_channel_mode, self._checkpoint_snapshot_frequency, + effective_user_id, authorization_identity, ) @@ -379,8 +387,6 @@ class DeerFlowClient: ) mcp_routing_hints_section = get_mcp_routing_hints_prompt_section(authorized_tools, deferred_names=deferred_setup.deferred_names) - effective_user_id = cfg.get("user_id") or get_effective_user_id() - kwargs: dict[str, Any] = { # attach_tracing=False because ``stream()`` injects tracing # callbacks at the graph invocation root so a single embedded run @@ -899,12 +905,11 @@ class DeerFlowClient: configurable = config.get("configurable") or {} deerflow_trace_id = ensure_trace_id() effective_user_id = context.get("user_id") or get_effective_user_id() - if self._app_config.authorization.enabled: - # Match the existing user-scoped storage/tracing identity when an - # embedded caller relies on CurrentUser instead of an explicit - # user_id override. Layer 1, Layer 2, and the agent cache must see - # the same actor. - context["user_id"] = effective_user_id + # Materialize the storage owner in runtime context in every auth mode. + # ContextVars normally propagate, but this explicit channel also + # survives worker/isolated-loop boundaries and matches the identity + # used by prompt assembly and the agent cache. + context["user_id"] = effective_user_id inject_langfuse_metadata( config, thread_id=thread_id, diff --git a/backend/tests/test_client.py b/backend/tests/test_client.py index d7f7ea66c..51a8baca0 100644 --- a/backend/tests/test_client.py +++ b/backend/tests/test_client.py @@ -1172,6 +1172,28 @@ class TestEnsureAgent: assert mock_create_agent.call_count == 2 + def test_disabled_authorization_cache_key_still_isolates_effective_users(self, client, mock_app_config): + """User-bound prompts/middleware must never be reused across embedded callers.""" + mock_app_config.authorization = AuthorizationConfig(enabled=False) + client._app_config = mock_app_config + + with ( + patch("deerflow.client.create_chat_model"), + patch("deerflow.client.create_agent", side_effect=[MagicMock(), MagicMock()]) as mock_create_agent, + patch("deerflow.client.build_middlewares", return_value=[]) as mock_build_middlewares, + patch("deerflow.client.apply_prompt_template", return_value="prompt") as mock_apply_prompt, + patch("deerflow.client.get_enabled_skills_for_config", return_value=[]), + patch.object(client, "_get_tools", return_value=[]), + patch("deerflow.runtime.checkpointer.get_checkpointer", return_value=None), + ): + config = client._get_runnable_config("t1") + client._ensure_agent(config, context={"user_id": "alice"}) + client._ensure_agent(config, context={"user_id": "bob"}) + + assert mock_create_agent.call_count == 2 + assert [call.kwargs["user_id"] for call in mock_build_middlewares.call_args_list] == ["alice", "bob"] + assert [call.kwargs["user_id"] for call in mock_apply_prompt.call_args_list] == ["alice", "bob"] + def test_authorization_cache_key_snapshots_nested_attributes(self, client, mock_app_config): mock_app_config.authorization = AuthorizationConfig( enabled=True, @@ -1353,7 +1375,7 @@ class TestEnsureAgent: """_ensure_agent does not recreate if config key unchanged.""" mock_agent = MagicMock() client._agent = mock_agent - client._agent_config_key = (None, True, False, False, None, None, None, None, "full", 10, None) + client._agent_config_key = (None, True, False, False, None, None, None, None, "full", 10, "test-user-autouse", None) config = client._get_runnable_config("t1") client._ensure_agent(config) @@ -2832,6 +2854,22 @@ class TestScenarioAgentRecreation: assert captured["user_id"] == "test-user-autouse" assert agent.stream.call_args.kwargs["context"]["user_id"] == "test-user-autouse" + def test_stream_uses_effective_user_context_when_authorization_is_disabled(self, client, mock_app_config): + mock_app_config.authorization = AuthorizationConfig(enabled=False) + client._app_config = mock_app_config + agent = _make_agent_mock([{"messages": [AIMessage(content="ok", id="ai-1")]}]) + captured: dict = {} + + def fake_ensure(config, *, context): + captured.update(context) + client._agent = agent + + with patch.object(client, "_ensure_agent", side_effect=fake_ensure): + list(client.stream("hi", thread_id="t1")) + + assert captured["user_id"] == "test-user-autouse" + assert agent.stream.call_args.kwargs["context"]["user_id"] == "test-user-autouse" + class TestScenarioThreadIsolation: """Scenario: Operations on different threads don't interfere."""