diff --git a/CHANGELOG.md b/CHANGELOG.md index 57224d696..c7735cf4a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,20 @@ This section accumulates work toward the **2.1.0** milestone ### ⚠ Breaking changes +- **gateway:** Request trace ids are now issued unconditionally, and every + Gateway HTTP response carries an `X-Trace-Id` header. Previously both were + gated behind `logging.enhance.enabled`, which now controls **log output + only** — whether records carry a `trace_id` field, and in which format. The + header cannot be turned off; installations running the default + `enabled: false` will start seeing it after upgrading. Scheduled tasks, MCP + task notification runs, IM channel messages, and the embedded + `DeerFlowClient` bind an id per unit of work, so the id also reaches the run + record, the checkpoint metadata, and Langfuse traces that previously had + none. A `deerflow_trace_id` supplied in a run request's `metadata` or + `config.context` is now ignored and overwritten so the response header, the + logs, and the persisted run cannot disagree — send the `X-Trace-Id` request + header to pin a correlation id across services. `logging` remains + restart-required. No config keys were added or removed. ([#5119]) - **skills:** Sandboxes now reserve `/mnt/skills` for managed enabled-only projections. `DEER_FLOW_HOST_SKILLS_PATH` and `SKILLS_HOST_PATH` are no longer used; Docker/AIO and hostPath deployments derive projection paths from @@ -400,6 +414,34 @@ This section accumulates work toward the **2.1.0** milestone ### Fixed +- **gateway:** Stop persisting a caller-supplied `deerflow_trace_id` on the run + record. `body.metadata` reaches both the live run config, which the run + worker restamps, and the run record echoed verbatim by the runs API; only the + first was covered, so a client could make the most durable surface of a run + disagree with the `X-Trace-Id` and the log lines from the same request. The + id is now stamped once at the trust boundary, `config.context` is closed off + the same way, and a thread's own metadata is no longer seeded with the + run-scoped id of whichever run created it. ([#5119]) +- **gateway:** Expose `X-Trace-Id` in `Access-Control-Expose-Headers`. It is not + CORS-safelisted, so split-origin browser clients — the ones that cannot read + the Gateway's logs either — could not read the correlation id they are meant + to quote in a bug report. ([#5119]) +- **gateway:** Keep `X-Trace-Id` on unhandled-exception 500s. Starlette's + `ServerErrorMiddleware` emits those through the raw send outside every user + middleware, so the 500 for a server bug — the response most in need of + correlation — was the only one shipped without the id. `TraceMiddleware` now + sends its own 500 carrying the header before re-raising; the server's + exception logging is untouched and mid-stream failures propagate unchanged. + This fallback is emitted outside `CORSMiddleware` and stays CORS-opaque, so + split-origin browser clients cannot read the id on this one response — same + as the `ServerErrorMiddleware` 500 it replaces. ([#5119]) +- **gateway:** Strip a forged `deerflow_trace_id` from the persisted request + echo. `body.config` is stored verbatim as `runs.kwargs_json` and served back + by the runs API, so a forged id in `config.metadata` or `config.context` + survived on that one surface while every other carried the real id. + `redact_config_secrets` now drops the key from both containers, and + `build_run_config` merges run metadata onto a copy so the server-stamped id + can no longer be written through into the caller's request body. ([#5119]) - **artifacts:** Keep explicit full-file loading scoped to the source thread, so a same-path artifact in another conversation keeps its 1 MiB preview. ([#4634]) - **sandbox:** `SandboxAuditMiddleware` no longer blocks ordinary command substitution that only captures output. The rule now judges *position* instead @@ -2085,3 +2127,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#4983]: https://github.com/bytedance/deer-flow/pull/4983 [#4987]: https://github.com/bytedance/deer-flow/pull/4987 [#4998]: https://github.com/bytedance/deer-flow/pull/4998 +[#5119]: https://github.com/bytedance/deer-flow/pull/5119 diff --git a/README.md b/README.md index cdd9843ae..943cbef3c 100644 --- a/README.md +++ b/README.md @@ -717,16 +717,38 @@ Once a channel is connected, you can interact with DeerFlow directly from the ch #### Request Trace Correlation -Gateway request trace correlation is disabled by default so existing HTTP responses and log formats stay unchanged. To enable it, set: +Every Gateway HTTP response carries an `X-Trace-Id` header. The id is inherited +from an inbound `X-Trace-Id` when the caller sends one and generated otherwise, so +a proxy or an upstream service can pin one id across services. It needs no +configuration and cannot be turned off. + +The same id stays attached to work that outlives the HTTP response: the detached +run task, any subagents it delegates to, and the background memory-update threads. +It is recorded as `deerflow_trace_id` on the run record (visible in the runs API), +in the thread's checkpoint metadata, and in Langfuse traces. Scheduled tasks, MCP +task notification runs, and IM channel messages start outside HTTP and mint their +own id per occurrence. + +Log records carry that id only when enhanced logging is on: ```yaml logging: enhance: - enabled: true - format: text + enabled: true # print trace_id into log records + format: text # or json ``` -When enabled, every Gateway HTTP response includes `X-Trace-Id`, logs include `trace_id`, and Langfuse traces created by that request include `metadata.deerflow_trace_id` with the same value. +This is off by default because turning it on changes the log format. `logging` is +restart-required, so edit `config.yaml` and restart the Gateway. The setting +affects log output only — the id, the response header, and the run metadata are +unaffected. + +`deerflow_trace_id` is a DeerFlow correlation id: it is not a run id, and it is not +a provider's native trace id. It is not a lookup key either — nothing resolves a +thread or a run from it; use it to correlate log lines. A `deerflow_trace_id` sent +in a run request's `metadata` or `config.context` is ignored and overwritten, so +the response header, the logs, and the persisted run can never disagree. To pin a +correlation id, send the `X-Trace-Id` header. Gateway run history also records one terminal `run.delivery` receipt per run, including zero-output and crash-recovered runs. The receipt is persisted before diff --git a/README_fr.md b/README_fr.md index 769cbc8fc..6e5090971 100644 --- a/README_fr.md +++ b/README_fr.md @@ -501,7 +501,7 @@ Si vous utilisez une instance Langfuse auto-hébergée, définissez `LANGFUSE_BA - `user_id` = utilisateur effectif issu de `get_effective_user_id()` (revient à `default` en mode sans authentification) - `trace_name` = assistant id (par défaut `lead-agent`) - `tags` = `[env:, model:]` (omis lorsqu'ils ne sont pas définis) -- `metadata.deerflow_trace_id` = id de corrélation de requête DeerFlow, identique à `X-Trace-Id` lorsque la corrélation de trace des requêtes est activée +- `metadata.deerflow_trace_id` = id de corrélation de requête DeerFlow, toujours identique à l'en-tête de réponse `X-Trace-Id` renvoyé par la même requête (`logging.enhance.enabled` contrôle uniquement si cet id est écrit dans les logs) Ces champs sont injectés dans `RunnableConfig.metadata` à la racine de l'invocation du graphe, à la fois pour le chemin gateway (`runtime/runs/worker.py::run_agent`) et le chemin embarqué (`client.py::DeerFlowClient.stream`), de sorte que tout callback compatible LangChain puisse les lire. Définissez `DEER_FLOW_ENV` (ou `ENVIRONMENT`) pour étiqueter les traces par environnement de déploiement. diff --git a/README_ja.md b/README_ja.md index 409a9eda2..241c13f42 100644 --- a/README_ja.md +++ b/README_ja.md @@ -488,7 +488,7 @@ LANGFUSE_BASE_URL=https://cloud.langfuse.com - `user_id` = `get_effective_user_id()`から取得した有効なユーザー（認証なしモードでは`default`にフォールバック） - `trace_name` = assistant id（デフォルトは`lead-agent`） - `tags` = `[env:, model:]`（未設定の場合は省略） -- `metadata.deerflow_trace_id` = DeerFlowのリクエスト関連付けid。リクエストトレース関連付けが有効な場合は`X-Trace-Id`と一致します +- `metadata.deerflow_trace_id` = DeerFlowのリクエスト関連付けid。常に同じリクエストが返す`X-Trace-Id`レスポンスヘッダーと一致します（`logging.enhance.enabled`はこのidをログに出力するかどうかのみを制御します） これらは、gatewayパス（`runtime/runs/worker.py::run_agent`）と埋め込みパス（`client.py::DeerFlowClient.stream`）の両方で、グラフ呼び出しのルートで`RunnableConfig.metadata`に注入されるため、LangChain互換の任意のcallbackから読み取れます。`DEER_FLOW_ENV`（または`ENVIRONMENT`）を設定すると、デプロイ環境ごとにトレースにタグを付けられます。 diff --git a/README_ru.md b/README_ru.md index 1eda8d192..6015ee09d 100644 --- a/README_ru.md +++ b/README_ru.md @@ -445,7 +445,7 @@ LANGFUSE_BASE_URL=https://cloud.langfuse.com - `user_id` = эффективный пользователь из `get_effective_user_id()` (возвращается к `default` в режиме без аутентификации) - `trace_name` = assistant id (по умолчанию `lead-agent`) - `tags` = `[env:, model:]` (опускается, если не заданы) -- `metadata.deerflow_trace_id` = идентификатор корреляции запросов DeerFlow, совпадающий с `X-Trace-Id`, когда корреляция трассировки запросов включена +- `metadata.deerflow_trace_id` = идентификатор корреляции запросов DeerFlow, всегда совпадающий с заголовком ответа `X-Trace-Id` того же запроса (`logging.enhance.enabled` управляет только тем, выводится ли этот идентификатор в логи) Эти поля внедряются в `RunnableConfig.metadata` в корне вызова графа как для gateway-пути (`runtime/runs/worker.py::run_agent`), так и для встроенного пути (`client.py::DeerFlowClient.stream`), поэтому любой LangChain-совместимый callback может их прочитать. Установите `DEER_FLOW_ENV` (или `ENVIRONMENT`) для тегирования трасс по среде развёртывания. diff --git a/README_zh.md b/README_zh.md index 2d44426c3..f8cdbb161 100644 --- a/README_zh.md +++ b/README_zh.md @@ -538,7 +538,7 @@ LANGFUSE_BASE_URL=https://cloud.langfuse.com - `user_id` = 来自 `get_effective_user_id()` 的有效用户（在无鉴权模式下回退为 `default`） - `trace_name` = assistant id（默认为 `lead-agent`） - `tags` = `[env:, model:]`（未设置时省略） -- `metadata.deerflow_trace_id` = DeerFlow 的请求关联 id，当启用请求链路关联（request trace correlation）时与 `X-Trace-Id` 一致 +- `metadata.deerflow_trace_id` = DeerFlow 的请求关联 id，始终与同一请求返回的 `X-Trace-Id` 响应头一致（`logging.enhance.enabled` 只控制该 id 是否打印到日志中） 这些字段会在图（graph）调用的根部注入到 `RunnableConfig.metadata`，同时覆盖 gateway 路径（`runtime/runs/worker.py::run_agent`）和内嵌路径（`client.py::DeerFlowClient.stream`），因此任何兼容 LangChain 的 callback 都能读取到它们。设置 `DEER_FLOW_ENV`（或 `ENVIRONMENT`）可按部署环境为 trace 打标签。 diff --git a/backend/app/channels/manager.py b/backend/app/channels/manager.py index eb75407fb..86f5abde6 100644 --- a/backend/app/channels/manager.py +++ b/backend/app/channels/manager.py @@ -48,6 +48,7 @@ from deerflow.runtime.user_context import get_effective_user_id from deerflow.skills.slash import parse_slash_skill_reference from deerflow.skills.storage import get_or_new_skill_storage from deerflow.skills.storage.skill_storage import SkillStorage +from deerflow.trace_context import ensure_trace_context from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY logger = logging.getLogger(__name__) @@ -1702,50 +1703,56 @@ class ChannelManager: except asyncio.CancelledError: raise - dedupe_recorded = False - try: - # Dedupe before logging "received" so a provider retrying an - # event N times does not log N accepts. Provider ack side - # effects may still happen before this manager-level dedupe. - if await self._is_duplicate_inbound(msg): - continue - dedupe_recorded = self._inbound_dedupe_key(msg) is not None - logger.info( - "[Manager] received inbound: channel=%s, chat_id=%s, type=%s, text_len=%d, files=%d", - msg.channel_name, - msg.chat_id, - msg.msg_type.value, - len(msg.text or ""), - len(msg.files), - ) - # Deliberately awaited inline: never create a task per message. - await self._handle_message(msg) - except asyncio.CancelledError: - # A cancellation after dedupe admission must make provider - # redelivery retryable rather than retaining a TTL-long key for - # work that never completed. - if dedupe_recorded: - try: - await self._release_inbound_dedupe_key(msg) - except Exception: - logger.exception("[Manager] failed to release inbound dedupe key during worker cancellation") - raise - except Exception: - logger.exception( - "[Manager] inbound worker %d failed handling channel=%s chat_id=%s", - worker_index, - msg.channel_name, - msg.chat_id, - ) - if dedupe_recorded: - try: - await self._release_inbound_dedupe_key(msg) - except Exception: - # A dedupe backend outage must not shrink the fixed - # worker pool by letting cleanup escape this loop. - logger.exception("[Manager] failed to release inbound dedupe key after worker error") - finally: - self.bus.inbound_task_done() + # Inbound IM messages are a non-HTTP entry point: channels hold + # long-lived provider connections, so no ASGI middleware ever runs + # for them. Scope one trace id per message here -- the worker task + # is long-lived and reused, so the scope must close with the + # message rather than leak into the next one. + with ensure_trace_context(): + dedupe_recorded = False + try: + # Dedupe before logging "received" so a provider retrying an + # event N times does not log N accepts. Provider ack side + # effects may still happen before this manager-level dedupe. + if await self._is_duplicate_inbound(msg): + continue + dedupe_recorded = self._inbound_dedupe_key(msg) is not None + logger.info( + "[Manager] received inbound: channel=%s, chat_id=%s, type=%s, text_len=%d, files=%d", + msg.channel_name, + msg.chat_id, + msg.msg_type.value, + len(msg.text or ""), + len(msg.files), + ) + # Deliberately awaited inline: never create a task per message. + await self._handle_message(msg) + except asyncio.CancelledError: + # A cancellation after dedupe admission must make provider + # redelivery retryable rather than retaining a TTL-long key for + # work that never completed. + if dedupe_recorded: + try: + await self._release_inbound_dedupe_key(msg) + except Exception: + logger.exception("[Manager] failed to release inbound dedupe key during worker cancellation") + raise + except Exception: + logger.exception( + "[Manager] inbound worker %d failed handling channel=%s chat_id=%s", + worker_index, + msg.channel_name, + msg.chat_id, + ) + if dedupe_recorded: + try: + await self._release_inbound_dedupe_key(msg) + except Exception: + # A dedupe backend outage must not shrink the fixed + # worker pool by letting cleanup escape this loop. + logger.exception("[Manager] failed to release inbound dedupe key after worker error") + finally: + self.bus.inbound_task_done() @staticmethod def _inbound_dedupe_key(msg: InboundMessage) -> tuple[str, str, str, str] | None: diff --git a/backend/app/gateway/AGENTS.md b/backend/app/gateway/AGENTS.md index 3fdde4804..e0f001ce7 100644 --- a/backend/app/gateway/AGENTS.md +++ b/backend/app/gateway/AGENTS.md @@ -57,7 +57,7 @@ owner-scoped assistant version selection remains enabled. | **Artifacts** (`/api/threads/{id}/artifacts`) | `GET /{path}` - stream regular text and binary artifacts with `FileResponse`, including byte-`Range` 206/416 behavior used by bounded text previews and media seeking; active content types (`text/html`, `application/xhtml+xml`, `image/svg+xml`) are always forced as download attachments to reduce XSS risk; `?download=true` still forces download for other file types. `PUT /{path}` atomically replaces an existing UTF-8 text file under `/mnt/user-data/outputs` when its expected SHA-256 still matches; active runs conflict, and non-mounted sandbox providers receive the same update explicitly. Atomic replacement applies the existing POSIX permission handling when descriptor-based APIs are available and otherwise keeps the platform-native temporary-file permissions (Windows). | | **Suggestions** (`/api/suggestions`) | `GET /config` - returns global suggestions config boolean; `POST /threads/{id}/suggestions` - generate follow-up questions; rich list/block model content is normalized and inline reasoning (`...`, including unclosed/truncated blocks from reasoning models like MiniMax-M3) is stripped before JSON parsing | | **Input Polish** (`/api/input-polish`) | `POST /` - rewrite a composer draft before it is sent. This is a short authenticated `runs:create` LLM request using `input_polish` config; it does not create a LangGraph run, persist a message, or modify thread state. Shares the non-graph one-shot LLM path (`deerflow.utils.oneshot_llm.run_oneshot_llm`) with the suggestions route so model build + Langfuse metadata + invoke stay in one place; validates the same stripped view of the draft it sends to the model, and preserves literal `` substrings in the rewrite (`strip_think_blocks(truncate_unclosed=False)`) | -| **Thread Runs** (`/api/threads/{id}/runs`) | `POST /` - create background run; `POST /stream` - create + SSE stream; `POST /wait` - create + block. Before the first journaled run, seed an empty event feed from an existing checkpoint so legacy checkpoint-only history keeps earlier thread-global ordering and stays visible; skip without a checkpoint or when the feed is populated. `POST /regenerate/prepare` - prepare clean input + checkpoint metadata for regenerating the latest completed or interrupted assistant answer, carrying the latest non-empty thread title in graph input so resuming an older checkpoint cannot roll back a later manual rename (#4457); `POST /edit-regenerate/prepare` - prepare a checkpoint replay from the latest editable human turn with a replacement user message and edit replay metadata; it carries the current thread title the same way, but only when the replay base already has one — an untitled base belongs to a thread the title middleware has not named yet, so pinning the current title there would keep a name generated from the prompt the edit just replaced; `GET /` - list runs; `GET /{rid}` - run details; `POST /{rid}/cancel` - cancel; `GET /{rid}/join` - join SSE; `GET /{rid}/stream` hides action/wait; GET action 405 pre-owner; POST needs `runs:cancel`; `GET /{rid}/messages` - paginated per-run messages `{data, has_more}`; `GET /{rid}/events` - full event stream; `GET /{rid}/workspace-changes` - workspace/output file change summary and optional diffs; `GET /../messages` - legacy thread message array; `GET /../messages/page` - backward thread-global `seq` history page with middleware/subagent-AI/successful-regenerate/edit-replay filtering and page-run-scoped feedback enrichment; subagent AI callbacks remain available through run events while parent `task` ToolMessages stay visible for card restoration; `GET /../token-usage` - aggregate tokens plus an optional `context_usage` percentage. Context usage approximately counts messages from the latest materialized thread state through `build_thread_checkpoint_state_accessor`, so full and delta checkpoint modes expose the same input. The percentage uses the latest run's model and its configured `context_window`. | +| **Thread Runs** (`/api/threads/{id}/runs`) | `POST /` - create background run; `POST /stream` - create + SSE stream; `POST /wait` - create + block. Before the first journaled run, seed an empty event feed from an existing checkpoint so legacy checkpoint-only history keeps earlier thread-global ordering and stays visible; skip without a checkpoint or when the feed is populated. `POST /regenerate/prepare` - prepare clean input + checkpoint metadata for regenerating the latest completed or interrupted assistant answer, carrying the latest non-empty thread title in graph input so resuming an older checkpoint cannot roll back a later manual rename (#4457); `POST /edit-regenerate/prepare` - prepare a checkpoint replay from the latest editable human turn with a replacement user message and edit replay metadata; it carries the current thread title the same way, but only when the replay base already has one — an untitled base belongs to a thread the title middleware has not named yet, so pinning the current title there would keep a name generated from the prompt the edit just replaced; `GET /` - list runs; `GET /{rid}` - run details; `POST /{rid}/cancel` - cancel; `GET /{rid}/join` - join SSE; `GET /{rid}/stream` hides action/wait; GET action 405 pre-owner; POST needs `runs:cancel`; `GET /{rid}/messages` - paginated per-run messages `{data, has_more}`; `GET /{rid}/events` - full event stream; `GET /{rid}/workspace-changes` - workspace/output file change summary and optional diffs; `GET /../messages` - legacy thread message array; `GET /../messages/page` - backward thread-global `seq` history page with middleware/subagent-AI/successful-regenerate/edit-replay filtering and page-run-scoped feedback enrichment; subagent AI callbacks remain available through run events while parent `task` ToolMessages stay visible for card restoration; `GET /../token-usage` - aggregate tokens plus an optional `context_usage` percentage. Context usage approximately counts messages from the latest materialized thread state through `build_thread_checkpoint_state_accessor`, so full and delta checkpoint modes expose the same input. The percentage uses the latest run's model and its `context_window`. | | **Feedback** (`/api/threads/{id}/runs/{rid}/feedback`) | `PUT /` - upsert feedback; `DELETE /` - delete user feedback; `POST /` - create feedback; `GET /` - list feedback; `GET /stats` - aggregate stats; `DELETE /{fid}` - delete specific | | **Runs** (`/api/runs`) | `POST /stream`, `/wait` - stateless runs requiring `runs:create`; optional body `thread_id` is owner-checked. Scheduled-task create/update/resume/trigger also require `threads:write` plus `runs:create`. `GET /{rid}/messages`, `/feedback` - run messages/feedback | | **GitHub Webhooks** (`/api/webhooks/github`) | `POST /` - receive GitHub App / repo webhook deliveries. Verifies `X-Hub-Signature-256` against `GITHUB_WEBHOOK_SECRET`; exempt from auth + CSRF because authenticity is enforced by HMAC. The route is fail-closed: mounted only when `GITHUB_WEBHOOK_SECRET` is set, or when explicit dev opt-in `DEER_FLOW_ALLOW_UNVERIFIED_GITHUB_WEBHOOKS=1` is set. Recognized events include `ping`, `issues`, `issue_comment`, `pull_request`, `pull_request_review`, and `pull_request_review_comment`; unknown events return 200 with `handled=false`. Fan-out runtime failures return 503, keeping the delivery recorded as failed for manual/API/scripted redelivery (GitHub does not automatically retry any failed delivery, 5xx included); permanent/non-retryable conditions such as `channels.github.enabled: false`, unknown events, malformed payloads, or unavailable channel service return 200 with a skipped/handled response. | diff --git a/backend/app/gateway/app.py b/backend/app/gateway/app.py index 7e6696b2e..abd283afc 100644 --- a/backend/app/gateway/app.py +++ b/backend/app/gateway/app.py @@ -41,7 +41,7 @@ from app.gateway.routers import ( threads, uploads, ) -from app.gateway.trace_middleware import TraceMiddleware, resolve_trace_enabled +from app.gateway.trace_middleware import TraceMiddleware from deerflow.config import app_config as deerflow_app_config from deerflow.logging_config import DEFAULT_LOG_DATE_FORMAT, DEFAULT_LOG_FORMAT, configure_logging from deerflow.tracing.monocle import setup_monocle_tracing_if_enabled @@ -737,13 +737,11 @@ This gateway provides runtime endpoints for agent runs plus custom endpoints for expose_headers=list(CORS_EXPOSED_HEADERS), ) - # Request trace correlation: when logging.enhance.enabled=true, bind one - # trace id per Gateway HTTP request and write it to response start headers. - # `logging` is registered as restart-required (see reload_boundary.py) so we - # snapshot the flag from the startup AppConfig instead of reading live; a - # runtime toggle would otherwise leave the log formatter (installed once by - # configure_logging() at lifespan startup) out of sync with the middleware. - app.add_middleware(TraceMiddleware, enabled=_resolve_trace_enabled_for_app_construction()) + # Request trace correlation: bind one trace id per Gateway HTTP request + # and write it to the response start headers. Ungated, so it works without + # a config.yaml and needs no restart; logging.enhance.enabled only decides + # whether that id is printed into log records. + app.add_middleware(TraceMiddleware) # Python extensions load once while the Gateway app is constructed. Agent # middleware builders consume the same immutable set through the process @@ -761,10 +759,9 @@ This gateway provides runtime endpoints for agent runs plus custom endpoints for # fail-open guard below: a config.yaml that exists but cannot be parsed or # validated is a configuration failure, not an extension failure. Reporting # it as the latter would silently drop a `required: true` extension instead - # of failing the boot. Only an absent config.yaml is tolerated, mirroring - # _resolve_trace_enabled_for_app_construction() — create_app() runs at - # import time, and lifespan still performs strict config loading before - # serving. + # of failing the boot. Only an absent config.yaml is tolerated — create_app() + # runs at import time, and lifespan still performs strict config loading + # before serving. try: configured_plugins = get_app_config().plugins except FileNotFoundError: @@ -896,15 +893,5 @@ This gateway provides runtime endpoints for agent runs plus custom endpoints for return app -def _resolve_trace_enabled_for_app_construction() -> bool: - """Resolve the trace middleware flag without making imports require config.yaml.""" - try: - return resolve_trace_enabled(get_app_config()) - except FileNotFoundError: - # Startup lifespan still performs strict config loading before serving. - logger.debug("config.yaml not found while constructing Gateway app; TraceMiddleware disabled for this app instance") - return False - - # Create app instance for uvicorn app = create_app() diff --git a/backend/app/gateway/csrf_middleware.py b/backend/app/gateway/csrf_middleware.py index aa823c2c2..01ef02ebe 100644 --- a/backend/app/gateway/csrf_middleware.py +++ b/backend/app/gateway/csrf_middleware.py @@ -18,6 +18,7 @@ from app.gateway.auth.config import get_auth_config from app.gateway.auth.session_cookie_state import SESSION_COOKIE_ISSUED_STATE_ATTR, SESSION_COOKIE_MAX_AGE_STATE_ATTR, SESSION_COOKIE_SECURE_STATE_ATTR, SKIP_AUTH_CSRF_COOKIE_STATE_ATTR from app.gateway.auth_disabled import is_auth_disabled from app.gateway.request_path import get_request_route_path +from deerflow.trace_context import TRACE_ID_HEADER CSRF_COOKIE_NAME = "csrf_token" CSRF_HEADER_NAME = "X-CSRF-Token" @@ -130,7 +131,11 @@ def get_configured_cors_origins() -> set[str]: # CORS-safelisted set is visible to JS by default, and the created run's id # travels in `Content-Location` — the LangGraph SDK resolves run metadata from # it, so withholding it leaves such a client unable to learn its own run id. -CORS_EXPOSED_HEADERS: tuple[str, ...] = ("Content-Location",) +# `X-Trace-Id` is listed for the same reason: TraceMiddleware puts it on every +# response as the correlation id to quote in a bug report, and unexposed it is +# readable on same-origin nginx deployments but invisible to exactly the +# split-origin clients that cannot see the Gateway's logs either. +CORS_EXPOSED_HEADERS: tuple[str, ...] = ("Content-Location", TRACE_ID_HEADER) def _first_header_value(value: str | None) -> str | None: diff --git a/backend/app/gateway/services.py b/backend/app/gateway/services.py index 21c6a6553..3aa656722 100644 --- a/backend/app/gateway/services.py +++ b/backend/app/gateway/services.py @@ -77,6 +77,7 @@ from deerflow.runtime.secret_context import ( from deerflow.runtime.stream_modes import normalize_stream_modes from deerflow.runtime.user_context import reset_current_user, set_current_user from deerflow.subagents.status_contract import SUBAGENT_ACCEPTANCE_VERDICT_KEY, SUBAGENT_RECEIPT_VERDICT_KEY, SUBAGENT_TOOL_RECEIPTS_KEY +from deerflow.trace_context import DEERFLOW_TRACE_METADATA_KEY, ensure_trace_context, ensure_trace_id from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY from deerflow.utils.thread_id import validate_thread_id @@ -197,7 +198,10 @@ async def _ensure_thread_metadata( await thread_store.create( record.thread_id, assistant_id=record.assistant_id, - metadata=record.metadata, + # Seeded from the run that created the thread, minus the run-scoped + # trace id: a thread spans many runs and as many trace ids, so + # pinning the first one here would be misleading rather than useful. + metadata={key: value for key, value in (record.metadata or {}).items() if key != DEERFLOW_TRACE_METADATA_KEY}, ) @@ -795,7 +799,15 @@ def build_run_config( external_values.pop(INTERNAL_CHECKPOINT_MODE_KEY, None) if metadata: - config.setdefault("metadata", {}).update(metadata) + # Merged onto a copy: config["metadata"] is the same dict object as the + # caller's body.config["metadata"] (the passthrough above copies + # references), and an in-place update would write server-stamped keys + # -- the trace id -- through into the request body that is persisted + # and echoed as the run's kwargs. + existing_metadata = config.get("metadata") + merged_metadata = dict(existing_metadata) if isinstance(existing_metadata, dict) else {} + merged_metadata.update(metadata) + config["metadata"] = merged_metadata return config @@ -1315,7 +1327,18 @@ async def start_run( graph_input = Command(resume=command["resume"]) else: graph_input = normalize_input(body.input, trusted_internal=is_internal_caller) - config = build_run_config(thread_id, body.config, body.metadata, assistant_id=body.assistant_id) + # deerflow_trace_id is server-issued, so the caller's value is replaced + # here at the trust boundary. body.metadata forks two ways -- through + # build_run_config into config["metadata"], which the run worker + # restamps, and through create_or_reject into the run record, which the + # runs API echoes verbatim. Only the first is covered downstream, so + # without this the run record is the one surface that persists a forged + # id, disagreeing with the response header, the logs, and the + # checkpoint. The caller's own metadata keys are preserved. + run_metadata = dict(body.metadata) if isinstance(body.metadata, dict) else {} + run_metadata[DEERFLOW_TRACE_METADATA_KEY] = ensure_trace_id() + + config = build_run_config(thread_id, body.config, run_metadata, assistant_id=body.assistant_id) await apply_checkpoint_to_run_config(config, body=body, thread_id=thread_id, request=request) # Merge DeerFlow-specific context overrides into both ``configurable`` and ``context``. @@ -1432,7 +1455,7 @@ async def start_run( thread_id, body.assistant_id, on_disconnect=disconnect, - metadata=body.metadata or {}, + metadata=run_metadata, # Persist a secret-redacted copy of the config: the run record is # written to runs.kwargs_json and echoed by the run API, so a # request-scoped secret (#3861) must not ride along. The live @@ -1527,12 +1550,19 @@ async def launch_scheduled_thread_run( ) scheduled_task_run_id = (metadata or {}).get("scheduled_task_run_id") idempotency_key = f"scheduled-task:{scheduled_task_run_id}" if isinstance(scheduled_task_run_id, str) else None - record = await start_run( - body, - thread_id, - request, - idempotency_key=idempotency_key, - ) + # Non-HTTP entry point: the lifespan scheduler calls this with a synthetic + # request, so TraceMiddleware never runs. The scope is opened per launch, + # never around the poller loop, or every scheduled run would collapse onto + # one id. Reached from inside an HTTP request -- a manual trigger, or the + # scheduler service's own per-occurrence scope -- ensure_trace_context + # keeps that trace instead of minting a competing one. + with ensure_trace_context(): + record = await start_run( + body, + thread_id, + request, + idempotency_key=idempotency_key, + ) return {"run_id": record.run_id, "thread_id": record.thread_id} @@ -1604,14 +1634,18 @@ async def launch_mcp_task_notification_run( feedback_keys=None, ) idempotency_key = f"mcp-task:{task_id}:{dispatch_version}:{dispatch_attempt}" + # Non-HTTP entry point, same as launch_scheduled_thread_run above: the MCP + # task service drives this from its own background loop, so one scope per + # notification keeps every delivery attempt separately correlatable. try: - record = await start_run( - body, - thread_id, - request, - idempotency_key=idempotency_key, - require_existing_thread=True, - ) + with ensure_trace_context(): + record = await start_run( + body, + thread_id, + request, + idempotency_key=idempotency_key, + require_existing_thread=True, + ) except HTTPException as exc: if exc.status_code == 409: raise ConflictError(str(exc.detail)) from exc diff --git a/backend/app/gateway/trace_middleware.py b/backend/app/gateway/trace_middleware.py index 057b65e82..1fa275647 100644 --- a/backend/app/gateway/trace_middleware.py +++ b/backend/app/gateway/trace_middleware.py @@ -2,73 +2,91 @@ from __future__ import annotations -import logging -from typing import Any - from starlette.datastructures import Headers, MutableHeaders from starlette.types import ASGIApp, Message, Receive, Scope, Send -from deerflow.config.app_config import is_trace_correlation_enabled -from deerflow.trace_context import ( - TRACE_ID_HEADER, - mark_trace_id_from_request_header, - normalize_trace_id, - request_trace_context, - reset_trace_id_from_request_header, -) - -logger = logging.getLogger(__name__) +from deerflow.trace_context import TRACE_ID_HEADER, request_trace_context class TraceMiddleware: - """Bind a request-level trace id and write it to HTTP response headers. + """Bind a trace id to every HTTP request and write it to the response. - The ``enabled`` flag is a **startup snapshot** rather than a per-request - live read: ``logging`` is registered as restart-required in - ``deerflow.config.reload_boundary.STARTUP_ONLY_FIELDS`` because - ``configure_logging()`` only installs the trace-context filter and - formatter during app.py lifespan startup. Reading ``logging.enhance.enabled`` - live here would let a runtime config edit surface the response - ``X-Trace-Id`` header and Langfuse ``deerflow_trace_id`` immediately while - the log formatter stays on its startup value, contradicting the - restart-required contract IDE hover surfaces on ``AppConfig.logging``. + Deliberately ungated. The id has to exist on every path so that everything + downstream -- the run worker's run metadata, delegated subagents, the + background memory threads -- reads one ContextVar instead of branching on + "there might be no trace id". ``logging.enhance.enabled`` only decides + whether log records print it (``logging_config.configure_logging``), so + this middleware reads no ``AppConfig`` and is not entangled with the + restart-required contract on that field. + + The header is written at ``http.response.start`` rather than on the + finished response, which covers SSE and other streaming responses without + consuming the body. ``CORS_EXPOSED_HEADERS`` lists it so split-origin + browser clients can read it back. + + Unhandled exceptions get their own 500 here rather than in Starlette's + ``ServerErrorMiddleware``: that middleware sits outside every user + middleware and emits through the raw send, so its 500 -- the one response + a user most needs to correlate with a log line -- would be the only one + without the header. + + That fallback 500 is CORS-opaque: this middleware sits outside + ``CORSMiddleware``, so the exception has already unwound past it and the + 500 carries no ``Access-Control-Allow-Origin`` -- a split-origin browser + client cannot read the id on this one response, unchanged from the + ``ServerErrorMiddleware`` 500 it replaces. Deliberately not fixed here: + replicating the origin allowlist outside ``CORSMiddleware`` would let the + two policies drift. """ - def __init__(self, app: ASGIApp, *, enabled: bool): + def __init__(self, app: ASGIApp): self.app = app - self.enabled = bool(enabled) async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: - if scope["type"] != "http" or not self.enabled: + if scope["type"] != "http": await self.app(scope, receive, send) return - headers = Headers(scope=scope) - incoming_trace_id = headers.get(TRACE_ID_HEADER) - header_provided = normalize_trace_id(incoming_trace_id) is not None + incoming_trace_id = Headers(scope=scope).get(TRACE_ID_HEADER) with request_trace_context(incoming_trace_id) as trace_id: - header_token = mark_trace_id_from_request_header(from_header=header_provided) + response_started = False + + async def send_with_trace(message: Message) -> None: + nonlocal response_started + if message["type"] == "http.response.start": + response_started = True + MutableHeaders(scope=message)[TRACE_ID_HEADER] = trace_id + await send(message) + try: - - async def send_with_trace(message: Message) -> None: - if message["type"] == "http.response.start": - response_headers = MutableHeaders(scope=message) - response_headers[TRACE_ID_HEADER] = trace_id - await send(message) - await self.app(scope, receive, send_with_trace) - finally: - reset_trace_id_from_request_header(header_token) - - -def resolve_trace_enabled(config: Any) -> bool: - """Read ``logging.enhance.enabled`` from an ``AppConfig``-like object. - - Thin backwards-compatible alias around - :func:`deerflow.config.app_config.is_trace_correlation_enabled`, kept so - existing gateway callers and tests do not have to switch imports. Both - the Gateway middleware and the embedded ``DeerFlowClient`` resolve the - gate through the same harness helper so their behaviour cannot drift. - """ - return is_trace_correlation_enabled(config) + except Exception: + # Before the response has started, ship a plain 500 carrying + # the header and re-raise: the outer ServerErrorMiddleware sees + # the response already started and only re-raises too, so the + # server's exception logging is untouched. Mid-stream failures + # propagate unchanged -- a second response start cannot be + # sent, and the already-written header stands. The id is + # printable ASCII by construction (``normalize_trace_id`` / + # ``generate_trace_id``), which makes the raw latin-1 header + # encoding safe. + if not response_started: + body = b"Internal Server Error" + await send( + { + "type": "http.response.start", + "status": 500, + # content-length keeps the framing byte-identical + # to the ServerErrorMiddleware response this + # replaces; without it the ASGI server picks + # (chunked on HTTP/1.1, close-delimited on 1.0). + "headers": [ + (b"content-type", b"text/plain; charset=utf-8"), + (b"content-length", str(len(body)).encode("latin-1")), + (TRACE_ID_HEADER.encode("latin-1"), trace_id.encode("latin-1")), + ], + } + ) + await send({"type": "http.response.body", "body": body}) + raise diff --git a/backend/app/scheduler/service.py b/backend/app/scheduler/service.py index 0c5ac2326..8b3b8f6c7 100644 --- a/backend/app/scheduler/service.py +++ b/backend/app/scheduler/service.py @@ -12,6 +12,7 @@ from fastapi import HTTPException from deerflow.persistence.scheduled_task_runs import ActiveScheduledRunConflict, ScheduledTaskAdmissionRejected from deerflow.runtime import ConflictError, RunRecord from deerflow.scheduler.schedules import next_run_at +from deerflow.trace_context import ensure_trace_context from deerflow.utils.thread_id import validate_thread_id logger = logging.getLogger(__name__) @@ -223,6 +224,24 @@ class ScheduledTaskService: queued: dict[str, Any], *, now: datetime, + ) -> dict[str, Any]: + """Turn one queued occurrence into a live run under its own trace scope. + + The poller is a non-HTTP entry point, so no ``TraceMiddleware`` has + bound anything: each occurrence opens its own scope rather than + sharing one id across a whole poll cycle. A manual trigger arrives + inside a Gateway request and keeps that request's trace instead, so + the launched run stays correlated with the call that asked for it. + """ + with ensure_trace_context(): + return await self._launch_queued_occurrence(task, queued, now=now) + + async def _launch_queued_occurrence( + self, + task: dict[str, Any], + queued: dict[str, Any], + *, + now: datetime, ) -> dict[str, Any]: task_run_id = queued["id"] execution_thread_id = queued["thread_id"] diff --git a/backend/packages/harness/deerflow/AGENTS.md b/backend/packages/harness/deerflow/AGENTS.md index 1c2e4b74c..53c3830c2 100644 --- a/backend/packages/harness/deerflow/AGENTS.md +++ b/backend/packages/harness/deerflow/AGENTS.md @@ -1,11 +1,31 @@ ### Request Trace Context (`packages/harness/deerflow/trace_context.py`) -Request trace correlation is controlled by `logging.enhance.enabled` at **both** entry points, gated through the shared helper `deerflow.config.app_config.is_trace_correlation_enabled` so the Gateway and embedded paths cannot drift: +DeerFlow's request-level correlation id — the `X-Trace-Id` header and the `deerflow_trace_id` key. Not Langfuse's trace id, not `run_id`, not the short subagent `trace_id` log label. -- **Gateway HTTP**: `app.gateway.trace_middleware.TraceMiddleware` binds one request-level trace id per HTTP request, inheriting inbound `X-Trace-Id` when present or generating a new id otherwise. A **valid** inbound header also marks the request so `runtime/runs/worker.py` prefers that id over `config.metadata.deerflow_trace_id`, keeping logs, response headers, Langfuse, and runtime context aligned when callers send both. The middleware writes the final value to every HTTP response at `http.response.start`, which covers SSE / streaming responses without consuming the body. -- **Embedded / TUI / CLI**: `DeerFlowClient.stream()` mints (or inherits) a request-level trace id per turn only when the flag is on. When it is off, no fresh id is minted — a caller that explicitly wraps `stream()` in `request_trace_context(...)` still opts in, because the downstream `get_current_trace_id()` read propagates that value into Langfuse metadata regardless of the flag. Because `stream()` is a sync generator (which shares the caller's context), the id binding is set/reset around each `next()` step rather than around `yield from`: this keeps LangGraph node execution and its log records inside the binding, while returning control to the caller with the ContextVar restored — avoids cross-request leak between yields and `ValueError: was created in a different Context` on GC-driven close of an abandoned generator (regression pinned by `tests/test_client_langfuse_metadata.py::test_stream_does_not_leak_trace_id_to_caller_context_between_yields` and `::test_stream_abandoned_generator_close_does_not_raise_cross_context`). +**The ContextVar is the only source.** Every path that reaches a run binds one first; downstream treats the id as a plain `str`, no `if trace_id:` guards. -The same ContextVar value is injected into enhanced log records as `trace_id` and into Langfuse metadata as `deerflow_trace_id`. +Entry points and binders: Gateway HTTP — `TraceMiddleware`; scheduled occurrence — `ScheduledTaskService._attempt_queued_run` → `launch_scheduled_thread_run`; MCP task notification — `launch_mcp_task_notification_run`; IM inbound — `ChannelManager._worker_loop`; embedded / TUI / CLI turn — `DeerFlowClient.stream()`. + +Only the first is HTTP; the rest run outside ASGI, so the binding cannot live in middleware alone. Each scopes **one unit of work**, never a poller loop — a leaked binding on a reused worker task would tag later occurrences with the first id. `ensure_trace_context` inherits, keeping layered scheduled bindings and a manual trigger inside a Gateway request on one trace. + +**Every other carrier is a derived output, never read back as an input.** `worker._bind_trace_id` stamps the runtime context and `config["metadata"]`; `services.start_run` stamps the run record; a caller-sent `deerflow_trace_id` (`body.metadata`, `body.config.context`) is replaced — honouring it would let the persisted run disagree with the header and the logs. `_SERVER_OWNED_RUNTIME_CONTEXT_KEYS` covers the embedded path, `redact_config_secrets` scrubs the kwargs echo (`runs.kwargs_json`), and `build_run_config` merges metadata onto a copy so the stamp cannot reach `body.config`. Callers pin an id with `X-Trace-Id`. + +Accepted divergence: a crash-recovered scheduled launch reuses its run via the idempotency key without restamping — the record keeps the first attempt's id, the retry's logs a fresh one; restamping would rewrite an existing record. Not a bug. Thread metadata omits the key entirely — a thread spans many runs. + +**Do not open-code fallback chains.** Two helpers own the resolution order: + +- `resolve_trace_id(*carriers)` — first usable carrier, else ambient. For ids travelling as data in `runtime.context`; ContextVars do not survive a bare thread hop. +- `ensure_trace_context(trace_id)` — reuse the surrounding scope, else start a self-contained one. For boundary crossings (`SubagentExecutor._aexecute`, the memory `trace_context_manager` hook) and non-HTTP entry points; no argument mints a scoped id. + +`request_trace_context` (HTTP) deliberately does **not** inherit: a crafted header must not fall back to the previous request's id. + +`get_current_trace_id()` stays nullable only for the logging filter (pre-entry-point records render as `trace_id=-`); everything else uses `ensure_trace_id()`/`resolve_trace_id()`. + +`DeerFlowClient.stream()` binds per `next()` step and around `inner.close()`, never across a `yield`: a sync generator shares the caller's context, so a scope held across yields would leak the id and break on cross-context GC finalization. + +`logging.enhance.enabled` gates **log output only** (`trace_id` field presence and format) — not the id, the header, or the run metadata — so `TraceMiddleware` reads no `AppConfig`; `logging` stays restart-required (`STARTUP_ONLY_FIELDS["logging"]`). `X-Trace-Id` is in `CORS_EXPOSED_HEADERS` (not safelisted). Unhandled-exception 500s keep the header — `TraceMiddleware` sends its own plain 500 (CORS-opaque, see its docstring) before re-raising; mid-stream failures propagate unchanged. + +Tests: the `tests/test_trace_*` and `tests/test_worker_trace_binding.py` suites, `test_gateway_services.py`, `test_run_metadata_secret_safety.py`, plus the Langfuse suites in `tracing/AGENTS.md`. ### Managed Lark CLI credentials (`integrations/lark_cli.py`) @@ -17,24 +37,6 @@ leave `config.json` with a dangling keychain reference. The transaction snapshot still supplies the previous OAuth data for logout and restores the complete old tree if any switch step fails. -`logging` is registered as a **restart-required** field -(`STARTUP_ONLY_FIELDS["logging"]`): `configure_logging()` installs the trace-context -filter and enhanced formatter on root handlers only during app.py lifespan startup, -and `TraceMiddleware` captures `logging.enhance.enabled` once when the FastAPI app -is constructed (via `resolve_trace_enabled(get_app_config())` in `create_app()`, -itself a thin alias for `is_trace_correlation_enabled`). This keeps the response -`X-Trace-Id` header, log `trace_id` fields, and Langfuse `deerflow_trace_id` -coherent — a runtime `config.yaml` edit to `logging.enhance.*` needs a Gateway -restart to take effect. The `deerflow_trace_id` chain inherits this guarantee -transitively because every injection point ultimately reads the same -`trace_context` ContextVar that the middleware alone populates. `DeerFlowClient` -reads its own `self._app_config` snapshot (captured at `__init__`) through the -same helper for the embedded gate. - -`deerflow_trace_id` is a DeerFlow correlation metadata key, not Langfuse's native -trace id and not a DeerFlow `run_id`. Keep the existing subagent `trace_id` field -separate: that short id is still only for subagent execution logs/status. - ### Browser Progress Screenshots (`community/browser_automation/`) Hidden per-action browser progress frames use JPEG at quality 80 to keep their diff --git a/backend/packages/harness/deerflow/agents/memory/manager.py b/backend/packages/harness/deerflow/agents/memory/manager.py index 7edee32d6..c1879830f 100644 --- a/backend/packages/harness/deerflow/agents/memory/manager.py +++ b/backend/packages/harness/deerflow/agents/memory/manager.py @@ -831,12 +831,12 @@ def _collect_host_hooks() -> dict[str, Any]: of its own) -- building an unused default on every startup would waste time. The others are direct values (cheap function refs). """ - from deerflow.trace_context import request_trace_context + from deerflow.trace_context import ensure_trace_context return { "callbacks": LangfuseMemoryCallbacks(), "should_keep_hidden_message": _host_default_should_keep_hidden_message, - "trace_context_manager": request_trace_context, + "trace_context_manager": ensure_trace_context, "host_llm_factory": _host_default_llm, "extraction_callback": _host_default_extraction_callback, } diff --git a/backend/packages/harness/deerflow/agents/middlewares/memory_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/memory_middleware.py index e0e946726..8ff283888 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/memory_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/memory_middleware.py @@ -12,7 +12,7 @@ from langgraph.runtime import Runtime from deerflow.agents.memory import get_memory_manager from deerflow.config.memory_config import get_memory_config from deerflow.runtime.user_context import resolve_runtime_user_id -from deerflow.trace_context import DEERFLOW_TRACE_METADATA_KEY, get_current_trace_id, normalize_trace_id +from deerflow.trace_context import DEERFLOW_TRACE_METADATA_KEY, resolve_trace_id if TYPE_CHECKING: from deerflow.config.memory_config import MemoryConfig @@ -50,7 +50,7 @@ class MemoryMiddleware(AgentMiddleware[MemoryMiddlewareState]): self._agent_name = agent_name self._memory_config = memory_config - def _resolve_add_args(self, state: MemoryMiddlewareState, runtime: Runtime) -> tuple[str, list, str, str | None] | None: + def _resolve_add_args(self, state: MemoryMiddlewareState, runtime: Runtime) -> tuple[str, list, str, str] | None: """Resolve one write request without invoking the manager.""" config = self._memory_config or get_memory_config() if not config.enabled: @@ -75,17 +75,13 @@ class MemoryMiddleware(AgentMiddleware[MemoryMiddlewareState]): # threading.Timer fires on a different thread where ContextVar values are not # propagated, so we must store user_id explicitly in ConversationContext. user_id = resolve_runtime_user_id(runtime) + # The memory update fires on a threading.Timer thread that inherits no + # ContextVars, so the id is captured here, while the request context is + # still alive, and carried as data. The runtime context is authoritative + # (worker._bind_trace_id always fills it); the ambient fallback covers + # embedded callers driving the agent outside a Gateway run. runtime_context = runtime.context if isinstance(runtime.context, dict) else {} - trace_id = normalize_trace_id(runtime_context.get(DEERFLOW_TRACE_METADATA_KEY)) - if trace_id is None: - try: - config_data = get_config() - except RuntimeError: - config_data = {} - config_metadata = config_data.get("metadata", {}) if isinstance(config_data.get("metadata"), dict) else {} - trace_id = normalize_trace_id(config_metadata.get(DEERFLOW_TRACE_METADATA_KEY)) - if trace_id is None: - trace_id = get_current_trace_id() + trace_id = resolve_trace_id(runtime_context.get(DEERFLOW_TRACE_METADATA_KEY)) return thread_id, messages, user_id, trace_id diff --git a/backend/packages/harness/deerflow/client.py b/backend/packages/harness/deerflow/client.py index a5cdd5ffe..829ac509e 100644 --- a/backend/packages/harness/deerflow/client.py +++ b/backend/packages/harness/deerflow/client.py @@ -38,7 +38,7 @@ from deerflow.agents.lead_agent.prompt import apply_prompt_template, get_enabled from deerflow.agents.thread_state import get_thread_state_schema, normalize_middleware_state_schemas from deerflow.authz.principal import build_principal_from_context from deerflow.config.agents_config import AGENT_NAME_PATTERN -from deerflow.config.app_config import get_app_config, is_trace_correlation_enabled, reload_app_config +from deerflow.config.app_config import get_app_config, reload_app_config from deerflow.config.extensions_config import ( ExtensionsConfig, SkillStateConfig, @@ -64,7 +64,7 @@ from deerflow.skills.describe import build_skill_search_setup from deerflow.skills.storage import get_or_new_user_skill_storage from deerflow.subagents.capacity import configure_subagent_execution_capacity from deerflow.tools.builtins.tool_search import assemble_deferred_tools, build_mcp_routing_middleware, get_mcp_routing_hints_prompt_section -from deerflow.trace_context import DEERFLOW_TRACE_METADATA_KEY, generate_trace_id, get_current_trace_id, reset_current_trace_id, set_current_trace_id +from deerflow.trace_context import DEERFLOW_TRACE_METADATA_KEY, bind_trace_id, ensure_trace_id, generate_trace_id, get_current_trace_id, reset_trace_id from deerflow.tracing import build_tracing_callbacks, inject_langfuse_metadata from deerflow.uploads.manager import ( claim_unique_filename, @@ -715,23 +715,12 @@ class DeerFlowClient: ) -> Generator[StreamEvent, None, None]: """Stream a conversation turn with a DeerFlow request trace context. - Mirrors the Gateway ``TraceMiddleware`` gate: when - ``logging.enhance.enabled`` is off the embedded client does **not** - create a fresh request-level trace id, so Langfuse traces from - embedded / TUI / CLI callers keep their pre-enhancement schema and - do not gain a ``metadata.deerflow_trace_id`` key by default. A - caller that explicitly binds its own trace via - :func:`deerflow.trace_context.request_trace_context` still opts in: - the inner ``get_current_trace_id()`` read propagates that value - into Langfuse metadata regardless of the flag. + The embedded entry point, and like every other one it binds a trace id + for the turn so logs, Langfuse metadata, and delegated work correlate. + A caller that opened its own scope with ``request_trace_context`` keeps + that id; otherwise the turn gets a fresh one. """ - if not is_trace_correlation_enabled(self._app_config): - yield from self._stream_without_trace_context(message, thread_id=thread_id, **kwargs) - return - - # Resolve the trace id once, without mutating the caller's context. - # Inherits an ambient id if the caller opted in via - # ``request_trace_context``; otherwise mints a fresh one. + # Resolve the id once, without mutating the caller's context. trace_id = get_current_trace_id() or generate_trace_id() # Bind the trace id only around each ``next()`` step, never across a @@ -743,25 +732,35 @@ class DeerFlowClient: # Per-step set/reset keeps LangGraph node execution and its log # records inside the binding while returning control to the caller # with the ContextVar restored. - inner = self._stream_without_trace_context(message, thread_id=thread_id, **kwargs) + inner = self._stream_turn(message, thread_id=thread_id, **kwargs) _EXHAUSTED = object() try: while True: - token = set_current_trace_id(trace_id) + token = bind_trace_id(trace_id) try: try: event = next(inner) except StopIteration: event = _EXHAUSTED finally: - reset_current_trace_id(token) + reset_trace_id(token) if event is _EXHAUSTED: break yield event finally: - inner.close() + # close() drives the inner generator's finally path (GeneratorExit + # on an abandoned stream), which still logs and fires callbacks -- + # bind the turn's id around it so that cleanup correlates with the + # turn it belongs to. Set and reset in this same frame, never + # across a yield, so the per-step cross-context safety holds even + # when GC closes the generator from another Context. + token = bind_trace_id(trace_id) + try: + inner.close() + finally: + reset_trace_id(token) - def _stream_without_trace_context( + def _stream_turn( self, message: str, *, @@ -885,7 +884,7 @@ class DeerFlowClient: context[key] = kwargs[key] configurable = config.get("configurable") or {} - deerflow_trace_id = get_current_trace_id() + 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 @@ -906,8 +905,7 @@ class DeerFlowClient: self._ensure_agent(config, context=context) state: dict[str, Any] = {"messages": [HumanMessage(content=message, additional_kwargs={"run_id": run_id})]} - if deerflow_trace_id: - context[DEERFLOW_TRACE_METADATA_KEY] = deerflow_trace_id + context[DEERFLOW_TRACE_METADATA_KEY] = deerflow_trace_id if self._agent_name: context["agent_name"] = self._agent_name diff --git a/backend/packages/harness/deerflow/config/AGENTS.md b/backend/packages/harness/deerflow/config/AGENTS.md index 991677779..feec437e5 100644 --- a/backend/packages/harness/deerflow/config/AGENTS.md +++ b/backend/packages/harness/deerflow/config/AGENTS.md @@ -48,7 +48,7 @@ Extensions are optional only in the fallback *search* mode (priority 3-4 above): **`config.yaml`** key sections: - `models[]` - LLM configs with `use` class path, `supports_thinking`, `supports_vision`, provider-specific fields -- `logging.enhance` - Optional request trace correlation (`enabled`, `format`) for Gateway `X-Trace-Id`, log `trace_id`, and Langfuse `deerflow_trace_id` +- `logging.enhance` - Log output only (`enabled`, `format`): whether log records carry a `trace_id` field, and in which format. Trace ids are issued unconditionally — the Gateway `X-Trace-Id` header and Langfuse `deerflow_trace_id` metadata are always present whatever this says (see the Request Trace Context section in `packages/harness/deerflow/AGENTS.md`); restart-required - vLLM reasoning models should use `deerflow.models.vllm_provider:VllmChatModel`; for Qwen-style parsers prefer `when_thinking_enabled.extra_body.chat_template_kwargs.enable_thinking`, and DeerFlow will also normalize the older `thinking` alias - `tools[]` - Tool configs with `use` variable path and `group` - `tool_groups[]` - Logical groupings for tools diff --git a/backend/packages/harness/deerflow/config/app_config.py b/backend/packages/harness/deerflow/config/app_config.py index 5af8109b6..b03aef5ad 100644 --- a/backend/packages/harness/deerflow/config/app_config.py +++ b/backend/packages/harness/deerflow/config/app_config.py @@ -130,9 +130,18 @@ class LlmCallConfig(BaseModel): class LoggingEnhanceConfig(BaseModel): - """Request trace logging enhancement settings.""" + """Request trace logging enhancement settings. - enabled: bool = Field(default=False, description="Enable request-level trace ids in Gateway response headers and log records.") + Trace ids are issued unconditionally (``TraceMiddleware`` for HTTP, + ``ensure_trace_context`` elsewhere) and always returned in the + ``X-Trace-Id`` response header. This block decides only whether log + records carry that id, and in which format. + """ + + enabled: bool = Field( + default=False, + description="Print the request trace id into log records. Trace ids are always issued and always returned in the X-Trace-Id response header; this controls log output only.", + ) format: Literal["text", "json"] = Field(default="text", description="Enhanced log output format.") @@ -142,22 +151,6 @@ class LoggingConfig(BaseModel): enhance: LoggingEnhanceConfig = Field(default_factory=LoggingEnhanceConfig, description="Request trace correlation logging settings.") -def is_trace_correlation_enabled(config: Any) -> bool: - """Return ``True`` when ``logging.enhance.enabled`` is set on *config*. - - Single source of truth for the request-trace-correlation gate, shared by - the Gateway ``TraceMiddleware`` and the embedded ``DeerFlowClient`` so - the two entry points cannot drift on when ``deerflow_trace_id`` is - emitted (Langfuse metadata) and when a request-level trace id is bound - at all. Accepts any object exposing ``logging.enhance.enabled`` via - ``getattr`` chains (``AppConfig``, ``SimpleNamespace`` fixtures, etc.); - missing intermediate attributes silently degrade to ``False``. - """ - logging_config = getattr(config, "logging", None) - enhance = getattr(logging_config, "enhance", None) - return bool(getattr(enhance, "enabled", False)) - - def _legacy_config_candidates() -> tuple[Path, ...]: """Return source-tree config.yaml locations for monorepo compatibility.""" backend_dir = Path(__file__).resolve().parents[4] @@ -203,7 +196,7 @@ class AppConfig(BaseModel): default_factory=LoggingConfig, description=format_field_description( "logging", - field_doc="Structured logging and request trace correlation settings.", + field_doc="Structured logging settings: whether request trace ids appear in log records, and in which format.", ), ) token_usage: TokenUsageConfig = Field(default_factory=TokenUsageConfig, description="Token usage tracking configuration") diff --git a/backend/packages/harness/deerflow/config/reload_boundary.py b/backend/packages/harness/deerflow/config/reload_boundary.py index 0d1e72e05..99949bb00 100644 --- a/backend/packages/harness/deerflow/config/reload_boundary.py +++ b/backend/packages/harness/deerflow/config/reload_boundary.py @@ -59,8 +59,8 @@ STARTUP_ONLY_FIELDS: dict[str, str] = { ), "logging": ( "configure_logging() runs only during app.py startup; it installs/removes the trace-context filter and the enhanced formatter on root handlers, " - "and TraceMiddleware captures logging.enhance.enabled once at startup so response X-Trace-Id headers, log trace_id fields, and Langfuse " - "deerflow_trace_id stay coherent. A freshly reloaded AppConfig does not retrigger any of this." + "and a freshly reloaded AppConfig does not retrigger it, so a runtime edit to logging.enhance.* needs a Gateway restart. Only log output is " + "affected: trace ids are issued unconditionally and always returned in the X-Trace-Id response header, whatever this setting says." ), # Not part of the AppConfig Pydantic schema — channel credentials are # consumed directly by ``start_channel_service()`` once at lifespan diff --git a/backend/packages/harness/deerflow/runtime/runs/worker.py b/backend/packages/harness/deerflow/runtime/runs/worker.py index fab07e482..7fba8d692 100644 --- a/backend/packages/harness/deerflow/runtime/runs/worker.py +++ b/backend/packages/harness/deerflow/runtime/runs/worker.py @@ -31,7 +31,7 @@ from contextvars import Context from dataclasses import dataclass, field from datetime import datetime from functools import lru_cache -from typing import Any, Literal, cast +from typing import Any, Final, Literal, cast from langgraph.checkpoint.base import empty_checkpoint from langgraph.types import Overwrite @@ -76,11 +76,7 @@ from deerflow.runtime.serialization import serialize from deerflow.runtime.stream_bridge import StreamBridge from deerflow.runtime.stream_modes import normalize_stream_modes, to_langgraph_stream_modes from deerflow.runtime.user_context import get_effective_user_id, resolve_runtime_user_id -from deerflow.trace_context import ( - DEERFLOW_TRACE_METADATA_KEY, - is_trace_id_from_request_header, - resolve_deerflow_trace_id, -) +from deerflow.trace_context import DEERFLOW_TRACE_METADATA_KEY, ensure_trace_id from deerflow.tracing import inject_langfuse_metadata from deerflow.utils.messages import message_to_text from deerflow.workspace_changes import capture_workspace_snapshot, get_changed_output_paths, record_workspace_changes @@ -517,6 +513,19 @@ class _LargeFileToolChunkBatcher: return chunks +# Runtime-context keys the worker owns outright. A same-named key in the +# caller's ``config['context']`` is dropped rather than merged: the Gateway +# strips ``__``-prefixed keys in build_run_config, but embedded harness callers +# have no such filter and ``deerflow_trace_id`` carries no prefix to be caught +# by it anyway. +_SERVER_OWNED_RUNTIME_CONTEXT_KEYS: Final[frozenset[str]] = frozenset( + { + CURRENT_RUN_PRE_EXISTING_MESSAGE_IDS_KEY, + DEERFLOW_TRACE_METADATA_KEY, + } +) + + def _build_runtime_context( thread_id: str, run_id: str, @@ -529,9 +538,9 @@ def _build_runtime_context( Always includes ``thread_id`` and ``run_id``. Additional keys from the caller's ``config['context']`` (e.g. ``agent_name`` for the bootstrap flow — issue #2677) - are merged in but never override ``thread_id``/``run_id``. The resolved - ``AppConfig`` is added by the worker so tools can consume it without ambient - global lookups. + are merged in but never override ``thread_id``/``run_id`` or the server-owned + keys in ``_SERVER_OWNED_RUNTIME_CONTEXT_KEYS``. The resolved ``AppConfig`` is + added by the worker so tools can consume it without ambient global lookups. langgraph 1.1+ surfaces this as ``runtime.context`` via the parent runtime stored under ``config['configurable']['__pregel_runtime']`` — see @@ -540,7 +549,7 @@ def _build_runtime_context( runtime_ctx: dict[str, Any] = {"thread_id": thread_id, "run_id": run_id} if isinstance(caller_context, dict): for key, value in caller_context.items(): - if key == CURRENT_RUN_PRE_EXISTING_MESSAGE_IDS_KEY: + if key in _SERVER_OWNED_RUNTIME_CONTEXT_KEYS: continue runtime_ctx.setdefault(key, value) if app_config is not None: @@ -592,8 +601,13 @@ def _install_runtime_context(config: dict, runtime_context: dict[str, Any]) -> N if isinstance(existing_context, dict): existing_context.setdefault("thread_id", runtime_context["thread_id"]) existing_context.setdefault("run_id", runtime_context["run_id"]) + # Assigned, not setdefault: this is a server-owned key, the same rule + # _bind_trace_id applies to the runtime context and the run metadata. A + # deerflow_trace_id the caller put in body.config.context is an echo of + # a past output, not an input, and leaving it would make this one dict + # disagree with the response header and the logs. if DEERFLOW_TRACE_METADATA_KEY in runtime_context: - existing_context.setdefault(DEERFLOW_TRACE_METADATA_KEY, runtime_context[DEERFLOW_TRACE_METADATA_KEY]) + existing_context[DEERFLOW_TRACE_METADATA_KEY] = runtime_context[DEERFLOW_TRACE_METADATA_KEY] if "app_config" in runtime_context: existing_context["app_config"] = runtime_context["app_config"] if CURRENT_RUN_PRE_EXISTING_MESSAGE_IDS_KEY in runtime_context: @@ -696,6 +710,32 @@ class _SubagentEventBuffer: logger.warning("Run %s: failed to persist %d subagent step event(s)", self._run_id, len(batch), exc_info=True) +def _bind_trace_id(config: dict[str, Any], runtime_ctx: dict[str, Any]) -> str: + """Record the current request trace id on the runtime context and metadata. + + The ContextVar is the only source. A ``deerflow_trace_id`` the caller sent + in ``config["metadata"]`` is overwritten rather than read: honouring it + would let the persisted run disagree with the ``X-Trace-Id`` and the log + lines the same request already produced, which is the correlation the id + exists to provide in the first place. + + The two destinations serve different purposes. ``runtime_ctx`` is the + carrier across boundaries the ContextVar does not cross -- subagent + delegation, the memory update running on a Timer/executor thread -- while + ``config["metadata"]`` is persisted with the checkpoint and is what makes a + finished run traceable after the fact. + """ + trace_id = ensure_trace_id() + runtime_ctx[DEERFLOW_TRACE_METADATA_KEY] = trace_id + incoming_metadata = config.get("metadata") + # Replaced rather than mutated through: this mapping can be shared with the + # caller's request body. + merged_metadata = dict(incoming_metadata) if isinstance(incoming_metadata, dict) else {} + merged_metadata[DEERFLOW_TRACE_METADATA_KEY] = trace_id + config["metadata"] = merged_metadata + return trace_id + + async def run_agent( bridge: StreamBridge, run_manager: RunManager, @@ -972,14 +1012,7 @@ async def run_agent( task_store, extensions, ) - incoming_metadata = config.get("metadata") if isinstance(config.get("metadata"), dict) else {} - deerflow_trace_id = resolve_deerflow_trace_id(incoming_metadata.get(DEERFLOW_TRACE_METADATA_KEY)) - if deerflow_trace_id: - runtime_ctx[DEERFLOW_TRACE_METADATA_KEY] = deerflow_trace_id - if is_trace_id_from_request_header(): - merged_metadata = dict(incoming_metadata) - merged_metadata[DEERFLOW_TRACE_METADATA_KEY] = deerflow_trace_id - config["metadata"] = merged_metadata + deerflow_trace_id = _bind_trace_id(config, runtime_ctx) # Expose the run-scoped journal under a sentinel key so middleware can # write audit events (e.g. SafetyFinishReasonMiddleware recording # suppressed tool calls). Double-underscore prefix marks it as a diff --git a/backend/packages/harness/deerflow/runtime/secret_context.py b/backend/packages/harness/deerflow/runtime/secret_context.py index c94fc1276..7106553de 100644 --- a/backend/packages/harness/deerflow/runtime/secret_context.py +++ b/backend/packages/harness/deerflow/runtime/secret_context.py @@ -16,6 +16,8 @@ from __future__ import annotations from typing import Any +from deerflow.trace_context import DEERFLOW_TRACE_METADATA_KEY + # Reserved sub-key of the run context that holds request-scoped secrets supplied # by the caller. Source of truth for what a skill *may* receive. SECRETS_CONTEXT_KEY = "secrets" @@ -155,6 +157,11 @@ def redact_config_secrets(config: Any) -> Any: protected config surface is persisted or returned, while the live config that drives the run (built separately) keeps them. Ordinary metadata is preserved. Non-dict configs pass through unchanged. + + ``deerflow_trace_id`` is dropped from both containers as well: the id is + server-issued and ignored as an input, so echoing a caller-supplied one + back would only manufacture disagreement with the ``X-Trace-Id`` header, + the logs, and the run record's own stamped metadata. """ if not isinstance(config, dict): return config @@ -162,10 +169,14 @@ def redact_config_secrets(config: Any) -> Any: redacted = dict(config) context = config.get("context") if isinstance(context, dict): - redacted["context"] = redact_secret_context_keys(context) + scrubbed_context = redact_secret_context_keys(context) + scrubbed_context.pop(DEERFLOW_TRACE_METADATA_KEY, None) + redacted["context"] = scrubbed_context metadata = config.get("metadata") if isinstance(metadata, dict): - redacted["metadata"] = redact_metadata_secrets(metadata) + scrubbed_metadata = redact_metadata_secrets(metadata) + scrubbed_metadata.pop(DEERFLOW_TRACE_METADATA_KEY, None) + redacted["metadata"] = scrubbed_metadata return redacted diff --git a/backend/packages/harness/deerflow/subagents/executor.py b/backend/packages/harness/deerflow/subagents/executor.py index 27a778000..2a0990875 100644 --- a/backend/packages/harness/deerflow/subagents/executor.py +++ b/backend/packages/harness/deerflow/subagents/executor.py @@ -45,7 +45,7 @@ from deerflow.subagents.report_contract import ( ) from deerflow.subagents.step_events import capture_new_step_messages from deerflow.subagents.token_collector import SubagentTokenCollector -from deerflow.trace_context import DEERFLOW_TRACE_METADATA_KEY +from deerflow.trace_context import DEERFLOW_TRACE_METADATA_KEY, ensure_trace_context, resolve_trace_id from deerflow.tracing import build_tracing_callbacks, inject_langfuse_metadata from deerflow.utils.messages import message_content_to_text @@ -797,7 +797,9 @@ class SubagentExecutor: run_id: Parent run id, so delegated guardrail decisions attribute to the same run as the lead agent. deerflow_trace_id: DeerFlow request-level correlation id propagated - from the parent run for Langfuse metadata correlation. + from the parent run for Langfuse metadata correlation. Falls + back to the ambient trace so the attribute is always a real + id, never ``None``. extensions: The parent run's immutable ``LoadedExtensions`` snapshot, captured at ``task_tool`` dispatch. When None (embedded client, standalone LangGraph Server), ``_aexecute`` falls back to the @@ -844,7 +846,10 @@ class SubagentExecutor: # 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 + # Resolved, not stored raw: the attribute is part of the non-nullable + # trace contract, and ``_aexecute`` rebinds it because a subagent runs + # on the isolated loop thread where the parent ContextVar may be gone. + self.deerflow_trace_id = resolve_trace_id(deerflow_trace_id) # Parent run's extension snapshot. Binding it here (rather than reading # the singleton at execution time) is what keeps one run on a single # extension generation: a concurrent ``set_loaded_extensions()`` between @@ -1242,7 +1247,14 @@ class SubagentExecutor: return state, final_tools, deferred_setup async def _aexecute(self, task: str, result_holder: SubagentResult | None = None) -> SubagentResult: - """Execute after acquiring the process-wide native-subagent slot.""" + """Execute after acquiring the process-wide native-subagent slot. + + Rebinds the parent's request trace id for the whole execution. Sync + callers reach here on the persistent isolated loop thread, which is + entered through a copied ``Context`` -- so the binding is usually + still intact and this is a no-op -- but the id also travels as data + precisely because that copy is not guaranteed on every path. + """ result = result_holder if result is None: result = SubagentResult( @@ -1250,21 +1262,22 @@ class SubagentExecutor: trace_id=self.trace_id, status=SubagentStatus.PENDING, ) - try: - capacity = self.execution_capacity or get_subagent_execution_capacity() - async with capacity.slot(): - with result._state_lock: - if not result.status.is_terminal: - result.status = SubagentStatus.RUNNING - result.started_at = datetime.now() - return await self._aexecute_admitted(task, result) - except SubagentCapacityError as exc: - result.try_set_terminal( - SubagentStatus.FAILED, - error=str(exc), - admission_failure=True, - ) - return result + with ensure_trace_context(self.deerflow_trace_id): + try: + capacity = self.execution_capacity or get_subagent_execution_capacity() + async with capacity.slot(): + with result._state_lock: + if not result.status.is_terminal: + result.status = SubagentStatus.RUNNING + result.started_at = datetime.now() + return await self._aexecute_admitted(task, result) + except SubagentCapacityError as exc: + result.try_set_terminal( + SubagentStatus.FAILED, + error=str(exc), + admission_failure=True, + ) + return result async def _aexecute_admitted(self, task: str, result_holder: SubagentResult | None = None) -> SubagentResult: """Execute a task asynchronously. @@ -1443,8 +1456,7 @@ class SubagentExecutor: # (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[DEERFLOW_TRACE_METADATA_KEY] = self.deerflow_trace_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}") diff --git a/backend/packages/harness/deerflow/tools/builtins/task_tool.py b/backend/packages/harness/deerflow/tools/builtins/task_tool.py index 3830be146..f2c928d60 100644 --- a/backend/packages/harness/deerflow/tools/builtins/task_tool.py +++ b/backend/packages/harness/deerflow/tools/builtins/task_tool.py @@ -40,7 +40,7 @@ from deerflow.subagents.status_contract import ( make_subagent_additional_kwargs, ) from deerflow.tools.types import Runtime -from deerflow.trace_context import DEERFLOW_TRACE_METADATA_KEY, get_current_trace_id, normalize_trace_id +from deerflow.trace_context import DEERFLOW_TRACE_METADATA_KEY, resolve_trace_id from deerflow.utils.custom_events import aemit_custom_event if TYPE_CHECKING: @@ -751,7 +751,11 @@ async def task_tool( # None outside that path (embedded client, standalone LangGraph Server), where # the executor keeps its process-singleton fallback. run_extensions = resolve_run_extensions(parent_context) - 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() + # Request-level correlation id, distinct from the short ``trace_id`` above + # that labels this one subagent execution in log prefixes. The parent + # runtime context is authoritative (worker._bind_trace_id always fills it); + # the ambient fallback covers tools invoked outside a Gateway run. + deerflow_trace_id = resolve_trace_id(parent_context.get(DEERFLOW_TRACE_METADATA_KEY)) parent_available_skills = metadata.get("available_skills") if parent_available_skills is not None: diff --git a/backend/packages/harness/deerflow/trace_context.py b/backend/packages/harness/deerflow/trace_context.py index 5f9726996..9fdcfa386 100644 --- a/backend/packages/harness/deerflow/trace_context.py +++ b/backend/packages/harness/deerflow/trace_context.py @@ -2,6 +2,36 @@ The value stored here is DeerFlow's request-level correlation id. It is separate from Langfuse's own trace id and from DeerFlow run ids. + +**This ContextVar is the only source of a trace id.** Every path that reaches a +run binds one first: the Gateway ``TraceMiddleware`` for HTTP, and +:func:`ensure_trace_context` for the entry points that never touch ASGI -- +scheduled occurrences, MCP task notification runs, IM channel messages, and the +embedded :class:`~deerflow.client.DeerFlowClient`. Downstream code can therefore +treat the trace id as a plain ``str`` and use :func:`ensure_trace_id` or +:func:`resolve_trace_id` instead of the ``if trace_id:`` guards a nullable id +used to require. + +Everything else that carries the id -- the ``X-Trace-Id`` response header, +``runtime.context[DEERFLOW_TRACE_METADATA_KEY]``, the run record's metadata, +log records -- is a **derived output, never read back as an input**. A caller +that sends ``metadata.deerflow_trace_id`` on a run request has it replaced +rather than honoured: reading it back would let the persisted run disagree with +the header the same request already returned, and a trace id you cannot trust +to match the logs is worse than no trace id at all. Callers that need to pin a +correlation id across services send ``X-Trace-Id``. + +``logging.enhance.enabled`` gates log output only -- whether records carry a +``trace_id`` field, and in which format. It does not gate the id's existence, +the response header, or the run metadata. + +Crossing execution boundaries +----------------------------- +The ContextVar is task-local and does not survive a bare thread hop, which is +why the id also travels as data. Code re-entering on the far side of such a +boundary rebinds with :func:`ensure_trace_context` and reads carriers through +:func:`resolve_trace_id`, keeping the fallback order in one place instead of +open-coding it per call site. """ from __future__ import annotations @@ -17,10 +47,6 @@ DEERFLOW_TRACE_METADATA_KEY: Final[str] = "deerflow_trace_id" _MAX_TRACE_ID_LENGTH: Final[int] = 512 _current_trace_id: Final[ContextVar[str | None]] = ContextVar("deerflow_current_trace_id", default=None) -_trace_id_from_request_header: Final[ContextVar[bool]] = ContextVar( - "deerflow_trace_id_from_request_header", - default=False, -) def generate_trace_id() -> str: @@ -51,55 +77,73 @@ def normalize_trace_id(value: object) -> str | None: return trace_id -def set_current_trace_id(trace_id: str) -> Token[str | None]: - """Bind *trace_id* to the current execution context.""" - normalized = normalize_trace_id(trace_id) - if normalized is None: - normalized = generate_trace_id() - return _current_trace_id.set(normalized) - - -def reset_current_trace_id(token: Token[str | None]) -> None: - """Restore the trace context captured by *token*.""" - _current_trace_id.reset(token) - - def get_current_trace_id() -> str | None: - """Return the current request trace id, if one is bound.""" + """Return the bound trace id, or ``None`` when nothing is bound. + + Prefer :func:`ensure_trace_id` or :func:`resolve_trace_id`, which honour + the non-nullable contract. This nullable accessor exists for callers that + must neither mutate context nor fabricate a value: the logging filter, + which runs on records emitted before any entry point (import time, + third-party threads) and renders those as ``trace_id=-``. + """ return _current_trace_id.get() -def mark_trace_id_from_request_header(*, from_header: bool) -> Token[bool]: - """Record whether the current trace id came from a valid inbound header.""" - return _trace_id_from_request_header.set(from_header) +def ensure_trace_id() -> str: + """Return the ambient trace id, minting and binding one when unset. - -def reset_trace_id_from_request_header(token: Token[bool]) -> None: - """Restore the inbound-header flag captured by *token*.""" - _trace_id_from_request_header.reset(token) - - -def is_trace_id_from_request_header() -> bool: - """Return ``True`` when a valid ``X-Trace-Id`` header bound the request.""" - return _trace_id_from_request_header.get() - - -def resolve_deerflow_trace_id(metadata_trace_id: object) -> str | None: - """Resolve the effective ``deerflow_trace_id`` for a run. - - When Gateway ``TraceMiddleware`` bound a valid inbound ``X-Trace-Id``, - that value wins over ``config.metadata.deerflow_trace_id`` so logs, - response headers, Langfuse, and runtime context stay aligned. Otherwise - caller metadata wins, then the ambient request trace context. + Binding rather than returning a throwaway id is what makes repeated calls + inside one context agree. """ - if is_trace_id_from_request_header(): - return get_current_trace_id() - return normalize_trace_id(metadata_trace_id) or get_current_trace_id() + trace_id = _current_trace_id.get() + if trace_id is None: + trace_id = generate_trace_id() + _current_trace_id.set(trace_id) + return trace_id + + +def resolve_trace_id(*carriers: object) -> str: + """Return the first usable value in *carriers*, else the ambient trace id. + + The single place that knows the carrier fallback order, so a consumer + reading the id back out of ``runtime.context`` states its carriers and + nothing else. Carriers are listed most authoritative first and validated + with :func:`normalize_trace_id`, so an absent key and a malformed value + fall through identically. + """ + for carrier in carriers: + normalized = normalize_trace_id(carrier) + if normalized is not None: + return normalized + return ensure_trace_id() + + +def bind_trace_id(trace_id: str | None) -> Token[str | None]: + """Bind *trace_id* in the current context; ``None`` clears the binding. + + The low-level pair for callers that cannot use the context managers: a + sync generator that must bind per step (``DeerFlowClient.stream``), and + test harnesses restoring an unbound baseline. Values are normalized, and + an unusable one clears rather than fabricating an id -- every caller here + has already resolved the value it means to bind. + """ + return _current_trace_id.set(normalize_trace_id(trace_id)) + + +def reset_trace_id(token: Token[str | None]) -> None: + """Restore the binding captured by *token*.""" + _current_trace_id.reset(token) @contextmanager def request_trace_context(trace_id: str | None = None) -> Iterator[str]: - """Bind a request trace id for the duration of a request or entry point.""" + """Open a trace scope for an HTTP request, always binding a fresh id. + + *trace_id* is the inbound ``X-Trace-Id``; an absent or unusable one is + replaced by a generated id. Deliberately does **not** inherit the ambient + context: a crafted header must not silently fall back to the id of + whatever request ran before it on the same task. + """ normalized = normalize_trace_id(trace_id) or generate_trace_id() token = _current_trace_id.set(normalized) try: @@ -110,10 +154,30 @@ def request_trace_context(trace_id: str | None = None) -> Iterator[str]: @contextmanager def ensure_trace_context(trace_id: str | None = None) -> Iterator[str]: - """Bind *trace_id*, inherit the current trace, or create a fresh one.""" - normalized = normalize_trace_id(trace_id) or get_current_trace_id() or generate_trace_id() - token = _current_trace_id.set(normalized) + """Open a trace scope that inherits the ambient one when there is one. + + Two callers, one rule -- *reuse the surrounding scope, otherwise start a + self-contained one*: + + - Non-HTTP entry points (a scheduled occurrence, an MCP task notification + run, an inbound IM message) called with no id: the scope mints one, and + unbinds it on exit so the next unit of work on the same long-lived + worker task does not inherit it. Reached from inside an HTTP request -- + a manual scheduled trigger, say -- it stays on the caller's trace + instead of minting a competing id. + - Crossing an execution boundary (a thread hop, a background task, a queue + hand-off) where the ContextVar may not have survived: pass the id that + travelled as data alongside the work. + """ + normalized = normalize_trace_id(trace_id) + inherited = _current_trace_id.get() + if inherited is not None and (normalized is None or inherited == normalized): + yield inherited + return + + resolved = normalized or generate_trace_id() + token = _current_trace_id.set(resolved) try: - yield normalized + yield resolved finally: _current_trace_id.reset(token) diff --git a/backend/packages/harness/deerflow/tracing/AGENTS.md b/backend/packages/harness/deerflow/tracing/AGENTS.md index 9ef8c04fc..d3028554d 100644 --- a/backend/packages/harness/deerflow/tracing/AGENTS.md +++ b/backend/packages/harness/deerflow/tracing/AGENTS.md @@ -13,7 +13,7 @@ LangSmith and Langfuse are both supported. The wiring lives in two layers: | `langfuse_user_id` | `get_effective_user_id()` (`default` in no-auth); for subagents, captured from `runtime.context` at `task_tool` time via `resolve_runtime_user_id()` | | `langfuse_trace_name` | `RunRecord.assistant_id` / client `agent_name` (defaults to `lead-agent`); for subagents, `subagent:` (lowercased, `_` → `-`) | | `langfuse_tags` | `env:` + `model:` | -| `deerflow_trace_id` | Current request/entry trace id from `deerflow.trace_context`; matches `X-Trace-Id` for enhanced Gateway HTTP requests. Gated by `logging.enhance.enabled` in both gateway and embedded paths via `is_trace_correlation_enabled` — off by default; embedded callers can still opt in per-turn by wrapping `stream()` in `request_trace_context(...)` | +| `deerflow_trace_id` | Current entry trace id from `deerflow.trace_context`; always written, and always equal to the `X-Trace-Id` the same Gateway request returned. Not gated by config — see `packages/harness/deerflow/AGENTS.md` | Returns `{}` when Langfuse is not in the enabled providers — LangSmith-only deployments are unaffected. Set `DEER_FLOW_ENV` (or `ENVIRONMENT`) to tag traces by deployment environment. Tests live in `tests/test_tracing_factory.py`, `tests/test_tracing_metadata.py`, `tests/test_worker_langfuse_metadata.py`, `tests/test_client_langfuse_metadata.py`, and `tests/test_subagent_executor.py::TestSubagentTracingWiring`. diff --git a/backend/packages/harness/deerflow/tracing/metadata.py b/backend/packages/harness/deerflow/tracing/metadata.py index 2fe23f002..13c30a1a5 100644 --- a/backend/packages/harness/deerflow/tracing/metadata.py +++ b/backend/packages/harness/deerflow/tracing/metadata.py @@ -19,7 +19,7 @@ from __future__ import annotations from typing import Any from deerflow.config import get_enabled_tracing_providers -from deerflow.trace_context import DEERFLOW_TRACE_METADATA_KEY, get_current_trace_id, normalize_trace_id +from deerflow.trace_context import DEERFLOW_TRACE_METADATA_KEY, resolve_trace_id # Lazy-imported below to avoid a circular import: ``deerflow.runtime`` eagerly # imports the run worker, which in turn needs ``deerflow.tracing``. @@ -50,7 +50,9 @@ def build_langfuse_trace_metadata( environment: Deployment env (e.g. ``"production"``); emitted as ``env:`` in ``langfuse_tags``. deerflow_trace_id: Optional DeerFlow request trace id; falls back to - the current request trace context when omitted. + the current request trace context when omitted. Always emitted -- + it is what ties a Langfuse trace back to the log lines and the + ``X-Trace-Id`` the same request returned. """ if "langfuse" not in get_enabled_tracing_providers(): return {} @@ -62,9 +64,7 @@ def build_langfuse_trace_metadata( "langfuse_user_id": user_id or DEFAULT_USER_ID, "langfuse_trace_name": assistant_id or _DEFAULT_TRACE_NAME, } - request_trace_id = normalize_trace_id(deerflow_trace_id) or get_current_trace_id() - if request_trace_id: - metadata[DEERFLOW_TRACE_METADATA_KEY] = request_trace_id + metadata[DEERFLOW_TRACE_METADATA_KEY] = resolve_trace_id(deerflow_trace_id) tags: list[str] = [] if environment: diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index f2eff3c19..012273b93 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -136,6 +136,25 @@ def _restore_title_config_singleton(): reset_title_config() +@pytest.fixture(autouse=True) +def _isolate_trace_context(): + """Give every test an unbound request trace context. + + Entry points bind a trace id unconditionally, and ``ensure_trace_id()`` + binds one for the remainder of whatever context it is called in. pytest + runs the whole session in a single context, so without this reset one + test's trace would leak into the next and quietly satisfy assertions + about ids the test under exercise never bound. + """ + from deerflow.trace_context import bind_trace_id, reset_trace_id + + token = bind_trace_id(None) + try: + yield + finally: + reset_trace_id(token) + + @pytest.fixture(autouse=True) def _auto_user_context(request): """Inject a default ``test-user-autouse`` into the contextvar. diff --git a/backend/tests/test_client_langfuse_metadata.py b/backend/tests/test_client_langfuse_metadata.py index 51daac199..d9dd12db9 100644 --- a/backend/tests/test_client_langfuse_metadata.py +++ b/backend/tests/test_client_langfuse_metadata.py @@ -60,17 +60,12 @@ def _stub_agent_creation(monkeypatch, fake_agent: _FakeAgent) -> dict[str, Any]: return captured -def _make_client(_monkeypatch, *, enhance_enabled: bool = True) -> DeerFlowClient: +def _make_client(_monkeypatch) -> DeerFlowClient: """Build a client without going through ``__init__`` so we never load config.yaml or perform any other side-effectful startup work. - - ``enhance_enabled`` seeds the ``logging.enhance.enabled`` flag that - :func:`DeerFlowClient.stream` consults to gate request-trace binding - (mirrors the Gateway ``TraceMiddleware`` startup snapshot). """ fake_app_config = SimpleNamespace( models=[SimpleNamespace(name="stub-model")], - logging=SimpleNamespace(enhance=SimpleNamespace(enabled=enhance_enabled)), authorization=AuthorizationConfig(enabled=False), ) client = DeerFlowClient.__new__(DeerFlowClient) @@ -176,12 +171,10 @@ def test_stream_preserves_caller_metadata_overrides(monkeypatch): assert metadata["langfuse_trace_name"] == "lead-agent" -def test_stream_omits_deerflow_trace_id_when_enhance_disabled(monkeypatch): - """With ``logging.enhance.enabled=false`` the embedded client must not - forge a fresh request trace id. Otherwise embedded / TUI callers on the - default config would silently gain a new indexed ``deerflow_trace_id`` - key on every Langfuse trace they emit — the exact schema change the - enhancement flag exists to opt into. +def test_stream_always_binds_a_trace_id(monkeypatch): + """Embedded turns are correlated like every other entry point. The id is + unconditional so downstream consumers -- delegated subagents, the memory + threads -- read one ContextVar instead of branching on its absence. """ monkeypatch.setenv("LANGFUSE_TRACING", "true") monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-lf-test") @@ -193,24 +186,22 @@ def test_stream_omits_deerflow_trace_id_when_enhance_disabled(monkeypatch): fake_agent = _FakeAgent() captured = _stub_agent_creation(monkeypatch, fake_agent) - client = _make_client(monkeypatch, enhance_enabled=False) + client = _make_client(monkeypatch) - list(client.stream("hi", thread_id="thread-client-disabled")) + list(client.stream("hi", thread_id="thread-client-bound")) metadata = captured["config"].get("metadata") or {} - # Session / user still bind — those are Langfuse-native trace attributes - # unrelated to the request-trace-correlation enhancement. - assert metadata.get("langfuse_session_id") == "thread-client-disabled" - assert metadata.get("langfuse_trace_name") == "lead-agent" - # The gated key stays out of metadata. - assert DEERFLOW_TRACE_METADATA_KEY not in metadata + assert metadata.get("langfuse_session_id") == "thread-client-bound" + assert metadata[DEERFLOW_TRACE_METADATA_KEY] + # The same id reaches the runtime context, which is what carries it across + # the boundaries the ContextVar does not survive. + assert captured["context"][DEERFLOW_TRACE_METADATA_KEY] == metadata[DEERFLOW_TRACE_METADATA_KEY] -def test_stream_respects_caller_bound_trace_when_enhance_disabled(monkeypatch): - """Even with the enhancement disabled, a caller that explicitly binds - :func:`request_trace_context` has opted into propagation. The embedded - client must not swallow that id — the flag only gates *implicit* - per-turn id creation, not caller-supplied context.""" +def test_stream_keeps_a_caller_bound_trace(monkeypatch): + """A caller that opened its own scope with :func:`request_trace_context` + is pinning a correlation id across several turns; the client must join + that trace rather than mint a competing one per turn.""" monkeypatch.setenv("LANGFUSE_TRACING", "true") monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-lf-test") monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-lf-test") @@ -221,13 +212,13 @@ def test_stream_respects_caller_bound_trace_when_enhance_disabled(monkeypatch): fake_agent = _FakeAgent() captured = _stub_agent_creation(monkeypatch, fake_agent) - client = _make_client(monkeypatch, enhance_enabled=False) + client = _make_client(monkeypatch) - with request_trace_context("caller-opt-in"): - list(client.stream("hi", thread_id="thread-client-opt-in")) + with request_trace_context("caller-pinned"): + list(client.stream("hi", thread_id="thread-client-pinned")) metadata = captured["config"].get("metadata") or {} - assert metadata.get(DEERFLOW_TRACE_METADATA_KEY) == "caller-opt-in" + assert metadata.get(DEERFLOW_TRACE_METADATA_KEY) == "caller-pinned" def test_stream_does_not_leak_trace_id_to_caller_context_between_yields(monkeypatch): @@ -253,7 +244,7 @@ def test_stream_does_not_leak_trace_id_to_caller_context_between_yields(monkeypa yield ("values", {"messages": [], "artifacts": []}) _stub_agent_creation(monkeypatch, _TwoEventAgent()) - client = _make_client(monkeypatch, enhance_enabled=True) + client = _make_client(monkeypatch) from deerflow.trace_context import get_current_trace_id @@ -295,7 +286,7 @@ def test_stream_abandoned_generator_close_does_not_raise_cross_context(monkeypat yield ("values", {"messages": [], "artifacts": []}) _stub_agent_creation(monkeypatch, _InfiniteAgent()) - client = _make_client(monkeypatch, enhance_enabled=True) + client = _make_client(monkeypatch) gen = client.stream("hi", thread_id="thread-cross-ctx") # Pull one event in the current Context — a buggy implementation would @@ -309,3 +300,43 @@ def test_stream_abandoned_generator_close_does_not_raise_cross_context(monkeypat # Tokens (if any) cannot be reset from here. Reaching this line without # a ``ValueError`` is the assertion. isolated_ctx.run(gen.close) + + +def test_stream_abandoned_generator_cleanup_stays_inside_trace_binding(monkeypatch): + """Abandoning a stream runs the inner generator's ``finally`` path via + ``GeneratorExit``, and that cleanup still logs and fires callbacks. The + per-step binding has already been reset by then, so without a binding + around ``inner.close()`` the cleanup would observe no trace id (or an + unrelated ambient one) and its records would not correlate with the turn + they belong to. + """ + monkeypatch.setattr("deerflow.client.build_tracing_callbacks", lambda: []) + + from deerflow.trace_context import get_current_trace_id + + observed: dict[str, str | None] = {} + + class _RecordingAgent: + def __init__(self) -> None: + self.checkpointer = None + self.store = None + + def stream(self, state, *, config, context, stream_mode): + observed["body"] = get_current_trace_id() + try: + while True: + yield ("values", {"messages": [], "artifacts": []}) + finally: + observed["cleanup"] = get_current_trace_id() + + _stub_agent_creation(monkeypatch, _RecordingAgent()) + client = _make_client(monkeypatch) + + gen = client.stream("hi", thread_id="thread-abandoned-cleanup") + next(gen) + gen.close() + + assert observed["body"] is not None + assert observed["cleanup"] == observed["body"] + # The close-time binding is local set/reset: nothing leaks to the caller. + assert get_current_trace_id() is None diff --git a/backend/tests/test_extension_app_loading.py b/backend/tests/test_extension_app_loading.py index a75dcb747..57d2efcb9 100644 --- a/backend/tests/test_extension_app_loading.py +++ b/backend/tests/test_extension_app_loading.py @@ -165,8 +165,8 @@ def test_create_app_fails_closed_when_a_required_extension_cannot_load(monkeypat def test_create_app_tolerates_a_missing_config_file_and_loads_no_extensions(monkeypatch): """``create_app()`` runs at import time, so an absent config.yaml must not break it. - Mirrors ``_resolve_trace_enabled_for_app_construction()``: lifespan still - performs strict config loading before the Gateway serves traffic. + Only an absent config.yaml is tolerated; lifespan still performs strict + config loading before the Gateway serves traffic. """ import app.gateway.app as app_module import deerflow.extensions as extensions_module diff --git a/backend/tests/test_gateway_services.py b/backend/tests/test_gateway_services.py index 911914de0..9b2f8d47c 100644 --- a/backend/tests/test_gateway_services.py +++ b/backend/tests/test_gateway_services.py @@ -13,6 +13,7 @@ import pytest from app.gateway.auth_disabled import AUTH_SOURCE_INTERNAL from deerflow.config.app_config import AppConfig, reset_app_config, set_app_config from deerflow.runtime.events.store.memory import MemoryRunEventStore +from deerflow.trace_context import DEERFLOW_TRACE_METADATA_KEY @pytest.fixture @@ -2007,9 +2008,14 @@ def test_start_run_preserves_ordinary_metadata(_stub_app_config): ) await record.task - assert record.metadata == metadata + # The run is additionally stamped with the server-issued trace id; + # the caller's own keys pass through untouched, and both metadata forks + # agree. Thread metadata is not run-scoped -- one thread spans many + # runs and many trace ids -- so it keeps only what the caller sent. + assert record.metadata[DEERFLOW_TRACE_METADATA_KEY] + assert record.metadata == {**metadata, DEERFLOW_TRACE_METADATA_KEY: record.metadata[DEERFLOW_TRACE_METADATA_KEY]} + assert captured["config"]["metadata"] == record.metadata assert (await thread_store.get(thread_id))["metadata"] == metadata - assert captured["config"]["metadata"] == metadata asyncio.run(_scenario()) @@ -3348,3 +3354,137 @@ def test_client_forged_user_id_never_selects_another_users_credential(): runtime = SimpleNamespace(server_info=None, context=config["context"]) assert resolve_runtime_user_id(runtime) == "attacker-own-id" + + +def _make_trace_start_run_request(run_manager): + from types import SimpleNamespace + + from langgraph.checkpoint.memory import InMemorySaver + from langgraph.store.memory import InMemoryStore + + from deerflow.persistence.thread_meta.memory import MemoryThreadMetaStore + + return SimpleNamespace( + headers={}, + state=SimpleNamespace(auth_source=None), + app=SimpleNamespace( + state=SimpleNamespace( + stream_bridge=SimpleNamespace(), + run_manager=run_manager, + checkpointer=InMemorySaver(), + store=InMemoryStore(), + run_event_store=MemoryRunEventStore(), + run_events_config=None, + thread_store=MemoryThreadMetaStore(InMemoryStore()), + ) + ), + ) + + +async def _start_run_capturing_config(body, thread_id): + """Run ``start_run`` far enough to see both metadata forks.""" + from unittest.mock import patch + + from app.gateway.services import start_run + from deerflow.runtime.runs.manager import RunManager + from deerflow.runtime.runs.store.memory import MemoryRunStore + + run_manager = RunManager(store=MemoryRunStore()) + request = _make_trace_start_run_request(run_manager) + captured: dict[str, object] = {} + + async def fake_run_agent(*args, **kwargs): + captured["config"] = kwargs["config"] + + 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_id, request) + await record.task + + return record, captured["config"] + + +@pytest.mark.anyio +async def test_start_run_replaces_a_caller_supplied_trace_id(_stub_app_config): + """``body.metadata`` forks two ways: through ``build_run_config`` into the + live run config, which the worker restamps, and through + ``create_or_reject`` into the run record that the runs API echoes back. + Only the first is covered downstream, so a forged ``deerflow_trace_id`` + used to survive on the most visible surface of the two. + """ + from deerflow.trace_context import request_trace_context + + body = _run_create_request(metadata={DEERFLOW_TRACE_METADATA_KEY: "forged-by-caller", "caller_key": "kept"}) + + with request_trace_context("gateway-issued"): + record, config = await _start_run_capturing_config(body, "thread-trace-forgery") + + assert record.metadata[DEERFLOW_TRACE_METADATA_KEY] == "gateway-issued" + assert config["metadata"][DEERFLOW_TRACE_METADATA_KEY] == "gateway-issued" + # Only the server-owned key is replaced; the caller's own metadata stays. + assert record.metadata["caller_key"] == "kept" + + +@pytest.mark.anyio +async def test_start_run_stamps_the_run_record_without_caller_metadata(_stub_app_config): + """The run record always carries the id, so "the run records its trace id" + holds for every run rather than only the ones that asked for it.""" + from deerflow.trace_context import request_trace_context + + body = _run_create_request() + + with request_trace_context("gateway-issued"): + record, config = await _start_run_capturing_config(body, "thread-trace-stamp") + + assert record.metadata[DEERFLOW_TRACE_METADATA_KEY] == "gateway-issued" + assert config["metadata"][DEERFLOW_TRACE_METADATA_KEY] == "gateway-issued" + + +def test_build_run_config_merges_metadata_onto_a_copy(_stub_app_config): + """The nested values of ``request_config`` are reference copies of + ``body.config``, so an in-place metadata merge would write server-stamped + keys through into the client's request body before it is persisted as the + kwargs echo -- contaminating the "what the client sent" record.""" + from app.gateway.services import build_run_config + + caller_metadata = {"caller_key": "kept"} + request_config = {"metadata": caller_metadata} + + config = build_run_config("thread-copy-merge", request_config, {DEERFLOW_TRACE_METADATA_KEY: "gateway-issued"}) + + assert config["metadata"] == {"caller_key": "kept", DEERFLOW_TRACE_METADATA_KEY: "gateway-issued"} + assert caller_metadata == {"caller_key": "kept"} + + +@pytest.mark.anyio +async def test_start_run_strips_forged_trace_id_from_the_kwargs_echo(_stub_app_config): + """``create_or_reject`` persists ``body.config`` as ``runs.kwargs_json``, + which the runs API serves back. A forged ``deerflow_trace_id`` in + ``config.metadata`` or ``config.context`` must neither survive there nor be + replaced by a server value written through into the caller's request body: + the id is ignored as an input on that surface, so any echo of it only + manufactures disagreement with the header, the logs, and the run record.""" + from deerflow.trace_context import request_trace_context + + forged_config = { + "metadata": {DEERFLOW_TRACE_METADATA_KEY: "forged-in-config", "caller_key": "kept"}, + "context": {DEERFLOW_TRACE_METADATA_KEY: "forged-in-context", "model_name": "default"}, + } + body = _run_create_request( + metadata={DEERFLOW_TRACE_METADATA_KEY: "forged-by-caller"}, + config=forged_config, + ) + + with request_trace_context("gateway-issued"): + record, config = await _start_run_capturing_config(body, "thread-trace-echo") + + echoed = record.kwargs["config"] + assert DEERFLOW_TRACE_METADATA_KEY not in echoed["metadata"] + assert DEERFLOW_TRACE_METADATA_KEY not in echoed["context"] + assert echoed["metadata"]["caller_key"] == "kept" + # The caller's own request body is not mutated by the merge either. + assert forged_config["metadata"][DEERFLOW_TRACE_METADATA_KEY] == "forged-in-config" + # The live run config still carries the authoritative id. + assert config["metadata"][DEERFLOW_TRACE_METADATA_KEY] == "gateway-issued" diff --git a/backend/tests/test_memory_middleware.py b/backend/tests/test_memory_middleware.py index b1b76e0a1..0ec0dd8fb 100644 --- a/backend/tests/test_memory_middleware.py +++ b/backend/tests/test_memory_middleware.py @@ -45,4 +45,7 @@ def test_after_agent_queues_memory_under_runtime_user(monkeypatch): ) assert call.kwargs["agent_name"] == "researcher" assert call.kwargs["user_id"] == "runtime-user" - assert call.kwargs["trace_id"] is None + # Entry points always bind a trace id, so the memory queue -- which fires + # on a Timer thread that inherits no ContextVars -- always gets a real one + # captured at enqueue time rather than a None it would have to guard. + assert call.kwargs["trace_id"] diff --git a/backend/tests/test_memory_updater.py b/backend/tests/test_memory_updater.py index 1882bcd9c..42a0c4464 100644 --- a/backend/tests/test_memory_updater.py +++ b/backend/tests/test_memory_updater.py @@ -1716,6 +1716,8 @@ class TestUserIdForwarding: assert result is True invoke_config = model.invoke.call_args.kwargs["config"] metadata = invoke_config["metadata"] + # The update runs on a Timer thread that inherits no ContextVars, so the + # id captured at enqueue time is what keeps this trace correlated. assert metadata["deerflow_trace_id"] == "memory-trace-1" assert metadata["langfuse_session_id"] == "thread-memory" assert metadata["langfuse_user_id"] == "user-42" diff --git a/backend/tests/test_reload_boundary.py b/backend/tests/test_reload_boundary.py index fca782d37..40741aecb 100644 --- a/backend/tests/test_reload_boundary.py +++ b/backend/tests/test_reload_boundary.py @@ -86,7 +86,7 @@ def test_appconfig_descriptions_retain_original_field_documentation(): hover documents what the field is *and* why a restart is needed.""" descriptions = { "log_level": "debug/info/warning/error", - "logging": "Structured logging and request trace correlation settings.", + "logging": "Structured logging settings: whether request trace ids appear in log records, and in which format.", "database": "memory, sqlite, or postgres", "sandbox": "Sandbox provider", "run_events": "memory for dev", diff --git a/backend/tests/test_run_metadata_secret_safety.py b/backend/tests/test_run_metadata_secret_safety.py index 8d8f5e10d..f2f223bea 100644 --- a/backend/tests/test_run_metadata_secret_safety.py +++ b/backend/tests/test_run_metadata_secret_safety.py @@ -10,6 +10,7 @@ from deerflow.runtime.secret_context import ( redact_metadata_secrets, validate_run_metadata_secrets, ) +from deerflow.trace_context import DEERFLOW_TRACE_METADATA_KEY @pytest.mark.parametrize("value", ["secret", "", None, {"nested": True}]) @@ -72,6 +73,26 @@ def test_redact_config_secrets_hides_legacy_config_metadata_without_mutating_sou assert redacted["context"] is not source["context"] +def test_redact_config_secrets_drops_trace_id_from_metadata_and_context(): + """``body.config`` is persisted as ``runs.kwargs_json`` and echoed verbatim + by the runs API. ``deerflow_trace_id`` is ignored as an input everywhere + else, so echoing a caller-supplied one back would only manufacture + disagreement with the response header, the logs, and the run record.""" + source = { + "metadata": {DEERFLOW_TRACE_METADATA_KEY: "forged-meta", "token_usage": 7}, + "context": {DEERFLOW_TRACE_METADATA_KEY: "forged-ctx", "model_name": "default"}, + } + + redacted = redact_config_secrets(source) + + assert redacted == { + "metadata": {"token_usage": 7}, + "context": {"model_name": "default"}, + } + assert source["metadata"][DEERFLOW_TRACE_METADATA_KEY] == "forged-meta" + assert source["context"][DEERFLOW_TRACE_METADATA_KEY] == "forged-ctx" + + def test_run_response_hides_historical_auth_token_without_mutating_record(): legacy_metadata = {"auth_token": "legacy-secret", "token_usage": 7} record = RunRecord( diff --git a/backend/tests/test_subagent_executor.py b/backend/tests/test_subagent_executor.py index 80f4a5de7..f787edc71 100644 --- a/backend/tests/test_subagent_executor.py +++ b/backend/tests/test_subagent_executor.py @@ -31,6 +31,7 @@ from packaging.version import Version from deerflow.skills.types import Skill from deerflow.subagents.capacity import SubagentCapacityRejected +from deerflow.trace_context import request_trace_context # Module names that need to be mocked to break circular imports _MOCKED_MODULE_NAMES = [ @@ -3397,6 +3398,50 @@ class TestSubagentTracingWiring: assert len(callbacks) >= 2, "existing callbacks must be preserved when tracing is injected" assert result.status.value == SubagentStatus.COMPLETED.value + def test_deerflow_trace_id_is_never_none(self, classes): + """The attribute is part of the non-nullable trace contract: consumers + write it into the child runtime context unconditionally, so an + undelegated id must resolve rather than propagate ``None``.""" + executor = self._make_executor(classes, deerflow_trace_id=None) + + assert executor.deerflow_trace_id + + def test_deerflow_trace_id_falls_back_to_the_ambient_trace(self, classes): + with request_trace_context("ambient-trace-1"): + executor = self._make_executor(classes, deerflow_trace_id=None) + + assert executor.deerflow_trace_id == "ambient-trace-1" + + @pytest.mark.anyio + async def test_aexecute_rebinds_the_parent_trace_on_the_isolated_loop( + self, + classes, + executor_module, + monkeypatch, + ): + """Sync callers reach execution on the persistent isolated loop thread, + where the parent ContextVar is not guaranteed to have survived. The id + also travels as data precisely so it can be rebound here.""" + from deerflow.trace_context import get_current_trace_id + + executor = self._make_executor(classes, deerflow_trace_id="parent-trace-1") + fake_agent = _FakeStreamAgent() + monkeypatch.setattr(executor, "_build_initial_state", self._noop_build_initial_state) + monkeypatch.setattr(executor, "_create_agent", lambda *a, **kw: fake_agent) + + seen: list[str | None] = [] + original = executor._aexecute_admitted + + async def capture(*args, **kwargs): + seen.append(get_current_trace_id()) + return await original(*args, **kwargs) + + monkeypatch.setattr(executor, "_aexecute_admitted", capture) + + await executor._aexecute("do something") + + assert seen == ["parent-trace-1"] + @pytest.mark.anyio async def test_aexecute_injects_langfuse_session_user_and_trace_name( self, diff --git a/backend/tests/test_trace_context.py b/backend/tests/test_trace_context.py index 29de503a2..d663bca2d 100644 --- a/backend/tests/test_trace_context.py +++ b/backend/tests/test_trace_context.py @@ -11,12 +11,14 @@ import pytest from deerflow.trace_context import ( _MAX_TRACE_ID_LENGTH, - is_trace_id_from_request_header, - mark_trace_id_from_request_header, + bind_trace_id, + ensure_trace_context, + ensure_trace_id, + get_current_trace_id, normalize_trace_id, request_trace_context, - reset_trace_id_from_request_header, - resolve_deerflow_trace_id, + reset_trace_id, + resolve_trace_id, ) @@ -94,28 +96,106 @@ class TestNormalizeTraceIdRejectsUnsafeInput: assert normalize_trace_id("trace-\ud83d") is None -class TestResolveDeerflowTraceId: - def test_header_marker_defaults_false_and_resets(self) -> None: - assert is_trace_id_from_request_header() is False - token = mark_trace_id_from_request_header(from_header=True) +class TestEnsureTraceId: + """The non-nullable accessor that lets consumers drop presence guards.""" + + def test_returns_the_bound_id(self) -> None: + with request_trace_context("bound-1"): + assert ensure_trace_id() == "bound-1" + + def test_mints_and_binds_when_unset(self) -> None: + assert get_current_trace_id() is None + trace_id = ensure_trace_id() + assert trace_id + # Binding is what makes repeated reads inside one context agree. + assert get_current_trace_id() == trace_id + assert ensure_trace_id() == trace_id + + +class TestResolveTraceId: + """Carrier fallback: the one place that knows the order.""" + + def test_first_usable_carrier_wins(self) -> None: + with request_trace_context("ambient"): + assert resolve_trace_id("runtime-ctx", "config-metadata") == "runtime-ctx" + + def test_falls_through_absent_and_malformed_carriers_alike(self) -> None: + with request_trace_context("ambient"): + assert resolve_trace_id(None, "trace\nid", "config-metadata") == "config-metadata" + + def test_falls_back_to_ambient_trace(self) -> None: + with request_trace_context("ambient"): + assert resolve_trace_id(None, None) == "ambient" + + def test_never_returns_none_without_any_binding(self) -> None: + assert get_current_trace_id() is None + assert resolve_trace_id(None) + + +class TestBindTraceId: + """Low-level pair for callers that cannot use the context managers.""" + + def test_binds_and_restores(self) -> None: + token = bind_trace_id("step-1") try: - assert is_trace_id_from_request_header() is True + assert get_current_trace_id() == "step-1" finally: - reset_trace_id_from_request_header(token) - assert is_trace_id_from_request_header() is False + reset_trace_id(token) + assert get_current_trace_id() is None - def test_metadata_wins_without_inbound_header(self) -> None: - with request_trace_context("ambient-trace"): - assert resolve_deerflow_trace_id("metadata-trace") == "metadata-trace" - - def test_inbound_header_overrides_metadata(self) -> None: - with request_trace_context("header-trace"): - token = mark_trace_id_from_request_header(from_header=True) + def test_none_clears_the_binding(self) -> None: + """How a test harness restores an unbound baseline — see the autouse + ``_isolate_trace_context`` fixture in conftest.""" + with request_trace_context("outer"): + token = bind_trace_id(None) try: - assert resolve_deerflow_trace_id("metadata-trace") == "header-trace" + assert get_current_trace_id() is None finally: - reset_trace_id_from_request_header(token) + reset_trace_id(token) + assert get_current_trace_id() == "outer" - def test_falls_back_to_ambient_context(self) -> None: - with request_trace_context("ambient-only"): - assert resolve_deerflow_trace_id(None) == "ambient-only" + +class TestRequestTraceContext: + def test_generates_when_no_inbound_id(self) -> None: + with request_trace_context() as trace_id: + assert trace_id + assert get_current_trace_id() == trace_id + assert get_current_trace_id() is None + + def test_never_inherits_an_ambient_id(self) -> None: + """A crafted header must not silently fall back to the id of whatever + request ran before it on the same task.""" + with request_trace_context("outer"): + with request_trace_context("trace\nid") as inner: + assert inner != "outer" + assert get_current_trace_id() == "outer" + + +class TestEnsureTraceContext: + def test_rebinds_a_propagated_id_across_a_boundary(self) -> None: + with ensure_trace_context("carried-over") as trace_id: + assert trace_id == "carried-over" + assert get_current_trace_id() == "carried-over" + assert get_current_trace_id() is None + + def test_inherits_rather_than_minting_when_no_id_is_carried(self) -> None: + """A non-HTTP launch reached from inside a request stays on the + caller's trace instead of minting a competing id.""" + with request_trace_context("gateway-request-1"): + with ensure_trace_context() as trace_id: + assert trace_id == "gateway-request-1" + + def test_mints_a_scoped_id_for_a_non_http_entry_point(self) -> None: + """A long-lived worker task must not leak one unit of work's id into + the next, so the minted id is unbound on exit.""" + with ensure_trace_context() as first: + assert first + assert get_current_trace_id() is None + with ensure_trace_context() as second: + assert second != first + + def test_keeps_the_ambient_binding_when_the_id_already_matches(self) -> None: + with request_trace_context("outer"): + with ensure_trace_context("outer") as trace_id: + assert trace_id == "outer" + assert get_current_trace_id() == "outer" diff --git a/backend/tests/test_trace_entry_points.py b/backend/tests/test_trace_entry_points.py new file mode 100644 index 000000000..659fe44e4 --- /dev/null +++ b/backend/tests/test_trace_entry_points.py @@ -0,0 +1,307 @@ +"""Trace binding at the entry points that no ASGI middleware can reach. + +``TraceMiddleware`` covers Gateway HTTP traffic (``test_trace_middleware.py``) +and ``DeerFlowClient.stream`` covers embedded callers +(``test_client_langfuse_metadata.py``). The remaining ways work enters DeerFlow +hold no HTTP request at all: the scheduled-task poller, MCP task notification +runs, and IM channels, which keep long-lived provider connections. Each must +bind a trace id of its own, scoped to one unit of work, or everything +downstream falls back to an unattributed id. +""" + +from __future__ import annotations + +import asyncio +from datetime import UTC, datetime, timedelta +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from app.channels.manager import ChannelManager +from app.channels.message_bus import InboundMessage, MessageBus +from app.channels.store import ChannelStore +from app.scheduler.service import ScheduledTaskService +from deerflow.config.app_config import AppConfig, reset_app_config, set_app_config +from deerflow.trace_context import get_current_trace_id, request_trace_context + +# -------------------------------------------------------------------------- +# Scheduled tasks +# -------------------------------------------------------------------------- + + +class _StubTaskRepo: + def __init__(self, rows): + self.rows = rows + self.claimed = False + + async def claim_due_tasks(self, **_kwargs): + if self.claimed: + return [] + self.claimed = True + return self.rows + + async def claim_dispatch_lease(self, task_id, **_kwargs): + return next((dict(row) for row in self.rows if row["id"] == task_id), None) + + async def release_queued_admission_lease(self, task_id): + return False + + async def release_dispatch_lease(self, task_id, **_kwargs): + return True + + async def get_internal(self, task_id): + row = next((item for item in self.rows if item["id"] == task_id), None) + return dict(row) if row is not None else None + + async def update_after_launch(self, *_args, **_kwargs): + return None + + +class _StubRunRepo: + async def list_queued_runs(self, *, limit): + return [] + + async def expire_queued_runs(self, **_kwargs): + return [] + + async def recover_expired_launch_claims(self, **_kwargs): + return 0 + + async def get_active_run(self, task_id): + return None + + async def claim_queued_run(self, run_record_id, **_kwargs): + return {"id": run_record_id, "status": "launching"} + + async def create(self, **kwargs): + return {"id": kwargs["run_record_id"]} + + async def reconcile_launched_run(self, run_record_id, **_kwargs): + return True + + async def update_status(self, run_record_id, **_kwargs): + return True + + +def _scheduled_task(task_id: str) -> dict: + return { + "id": task_id, + "user_id": "user-1", + "thread_id": f"thread-{task_id}", + "context_mode": "reuse_thread", + "assistant_id": "lead_agent", + "prompt": "Summarize thread", + "schedule_type": "once", + "schedule_spec": {"run_at": "2026-07-02T01:00:00+00:00"}, + "timezone": "UTC", + } + + +def _make_service(rows, launch_run) -> ScheduledTaskService: + return ScheduledTaskService( + task_repo=_StubTaskRepo(rows), + task_run_repo=_StubRunRepo(), + launch_run=launch_run, + poll_interval_seconds=5, + lease_seconds=120, + max_concurrent_runs=3, + ) + + +@pytest.mark.asyncio +async def test_scheduled_launch_runs_under_a_bound_trace_id(): + launched: list[str | None] = [] + + async def fake_launch(**kwargs): + launched.append(get_current_trace_id()) + return {"run_id": "run-1", "thread_id": kwargs["thread_id"]} + + service = _make_service([_scheduled_task("task-1")], fake_launch) + + await service.run_once(now=datetime.now(UTC) + timedelta(days=1)) + + assert launched == [launched[0]] + assert launched[0], "a scheduled occurrence must not launch without a trace id" + + +@pytest.mark.asyncio +async def test_each_scheduled_occurrence_gets_its_own_trace_id(): + """One id per poll cycle would merge unrelated tasks into a single trace.""" + launched: list[str | None] = [] + + async def fake_launch(**kwargs): + launched.append(get_current_trace_id()) + return {"run_id": "run-1", "thread_id": kwargs["thread_id"]} + + service = _make_service([_scheduled_task("task-1"), _scheduled_task("task-2")], fake_launch) + + await service.run_once(now=datetime.now(UTC) + timedelta(days=1)) + + assert len(launched) == 2 + assert all(launched) + assert launched[0] != launched[1] + + +@pytest.mark.asyncio +async def test_scheduled_trace_scope_closes_after_the_occurrence(): + """The poller task is long-lived, so a leaked binding would attribute every + later cycle to the first occurrence it ever ran.""" + + async def fake_launch(**kwargs): + return {"run_id": "run-1", "thread_id": kwargs["thread_id"]} + + service = _make_service([_scheduled_task("task-1")], fake_launch) + + await service.run_once(now=datetime.now(UTC) + timedelta(days=1)) + + assert get_current_trace_id() is None + + +@pytest.mark.asyncio +async def test_manual_trigger_keeps_the_requesting_trace(): + """A manual trigger arrives inside a Gateway request, so the launched run + stays correlated with the call that asked for it.""" + launched: list[str | None] = [] + + async def fake_launch(**kwargs): + launched.append(get_current_trace_id()) + return {"run_id": "run-1", "thread_id": kwargs["thread_id"]} + + task = _scheduled_task("task-1") + service = _make_service([task], fake_launch) + + with request_trace_context("gateway-request-1"): + await service.dispatch_task(task, now=datetime.now(UTC), trigger="manual") + + assert launched == ["gateway-request-1"] + + +# -------------------------------------------------------------------------- +# IM channels +# -------------------------------------------------------------------------- + + +def _inbound(index: int) -> InboundMessage: + return InboundMessage( + channel_name="slack", + chat_id="C1", + user_id="U1", + text=f"message-{index}", + metadata={}, + ) + + +@pytest.mark.asyncio +async def test_inbound_messages_are_handled_under_distinct_trace_scopes(tmp_path: Path): + """Channels hold long-lived provider connections, so no ASGI middleware + ever runs for them, and one worker task serves many messages in sequence.""" + bus = MessageBus(inbound_queue_maxsize=4) + manager = ChannelManager( + bus=bus, + store=ChannelStore(path=tmp_path / "store.json"), + max_concurrency=1, + ) + seen: list[str | None] = [] + + async def capture_handler(msg: InboundMessage) -> None: + seen.append(get_current_trace_id()) + + manager._handle_message = capture_handler # type: ignore[method-assign] + await manager.start() + try: + await bus.publish_inbound(_inbound(0)) + await bus.publish_inbound(_inbound(1)) + async with asyncio.timeout(2): + while len(seen) < 2: + await asyncio.sleep(0) + finally: + await manager.stop() + + assert all(seen), "an inbound message must not be handled without a trace id" + assert seen[0] != seen[1], "each message is its own unit of work" + assert get_current_trace_id() is None + + +# -------------------------------------------------------------------------- +# Gateway run launchers +# -------------------------------------------------------------------------- + + +@pytest.fixture +def _stub_app_config(): + """Keep the launchers independent from a developer-local config.yaml.""" + set_app_config(AppConfig.model_validate({"sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"}})) + yield + reset_app_config() + + +@pytest.fixture +def launcher_traces(monkeypatch): + """Capture the trace id bound around each ``start_run`` the launchers make.""" + seen: list[str | None] = [] + + async def fake_start_run(_body, thread_id, _request, **_kwargs): + seen.append(get_current_trace_id()) + return SimpleNamespace(run_id="run-1", thread_id=thread_id) + + monkeypatch.setattr("app.gateway.services.start_run", fake_start_run) + return seen + + +@pytest.mark.asyncio +async def test_scheduled_launcher_binds_a_trace_context(_stub_app_config, launcher_traces): + from app.gateway.services import launch_scheduled_thread_run + + await launch_scheduled_thread_run( + app=SimpleNamespace(), + thread_id="thread-sched", + assistant_id="lead_agent", + prompt="Summarize thread", + owner_user_id="user-1", + metadata={"scheduled_task_run_id": "run-row-1"}, + ) + + assert launcher_traces[0], "a scheduled launch must not reach start_run untraced" + assert get_current_trace_id() is None + + +@pytest.mark.asyncio +async def test_mcp_notification_launcher_binds_a_trace_context(_stub_app_config, launcher_traces): + """Driven from the MCP task service's own background loop, so one scope per + notification keeps every delivery attempt separately correlatable.""" + from app.gateway.services import launch_mcp_task_notification_run + + for attempt in (1, 2): + await launch_mcp_task_notification_run( + app=SimpleNamespace(), + thread_id="thread-mcp", + assistant_id="lead_agent", + owner_user_id="user-1", + task_id="task-1", + dispatch_version=1, + dispatch_attempt=attempt, + event={"status": "completed"}, + ) + + assert all(launcher_traces) + assert launcher_traces[0] != launcher_traces[1] + assert get_current_trace_id() is None + + +@pytest.mark.asyncio +async def test_launcher_keeps_the_requesting_trace(_stub_app_config, launcher_traces): + """Reached from inside a Gateway request -- a manual scheduled trigger -- + the launched run stays correlated with the call that asked for it.""" + from app.gateway.services import launch_scheduled_thread_run + + with request_trace_context("gateway-request-1"): + await launch_scheduled_thread_run( + app=SimpleNamespace(), + thread_id="thread-sched", + assistant_id="lead_agent", + prompt="Summarize thread", + owner_user_id="user-1", + ) + + assert launcher_traces == ["gateway-request-1"] diff --git a/backend/tests/test_trace_middleware.py b/backend/tests/test_trace_middleware.py index cd10eb91e..097b685b7 100644 --- a/backend/tests/test_trace_middleware.py +++ b/backend/tests/test_trace_middleware.py @@ -1,29 +1,23 @@ -from types import SimpleNamespace +import asyncio +import pytest from fastapi import FastAPI from fastapi.responses import Response, StreamingResponse from starlette.testclient import TestClient -from app.gateway.trace_middleware import TraceMiddleware, resolve_trace_enabled -from deerflow.trace_context import ( - TRACE_ID_HEADER, - get_current_trace_id, - is_trace_id_from_request_header, -) +from app.gateway.csrf_middleware import CORS_EXPOSED_HEADERS +from app.gateway.trace_middleware import TraceMiddleware +from deerflow.trace_context import TRACE_ID_HEADER, get_current_trace_id -def _make_app(*, enabled: bool) -> FastAPI: +def _make_app() -> FastAPI: app = FastAPI() - app.add_middleware(TraceMiddleware, enabled=enabled) + app.add_middleware(TraceMiddleware) @app.get("/plain") async def plain() -> dict[str, str | None]: return {"trace_id": get_current_trace_id()} - @app.get("/header-flag") - async def header_flag() -> dict[str, bool]: - return {"from_header": is_trace_id_from_request_header()} - @app.get("/stream") async def stream() -> StreamingResponse: async def body(): @@ -38,17 +32,25 @@ def _make_app(*, enabled: bool) -> FastAPI: return app -def test_trace_header_absent_when_disabled() -> None: - client = TestClient(_make_app(enabled=False)) +def test_every_response_carries_a_trace_id() -> None: + """Ungated by design: downstream reads one ContextVar instead of branching + on whether a trace id happens to exist.""" + client = TestClient(_make_app()) response = client.get("/plain") - assert TRACE_ID_HEADER not in response.headers - assert response.json() == {"trace_id": None} + assert response.headers[TRACE_ID_HEADER] + assert response.json()["trace_id"] == response.headers[TRACE_ID_HEADER] + + +def test_trace_id_header_is_exposed_to_split_origin_clients() -> None: + """Not CORS-safelisted, so a browser client on a separate origin cannot + read the id it is meant to quote in a bug report unless it is listed.""" + assert TRACE_ID_HEADER in CORS_EXPOSED_HEADERS def test_trace_header_inherits_inbound_value_and_binds_context() -> None: - client = TestClient(_make_app(enabled=True)) + client = TestClient(_make_app()) response = client.get("/plain", headers={TRACE_ID_HEADER: "trace-from-upstream"}) @@ -57,7 +59,7 @@ def test_trace_header_inherits_inbound_value_and_binds_context() -> None: def test_trace_header_generated_when_missing() -> None: - client = TestClient(_make_app(enabled=True)) + client = TestClient(_make_app()) response = client.get("/plain") @@ -67,7 +69,7 @@ def test_trace_header_generated_when_missing() -> None: def test_trace_header_added_to_streaming_response_without_consuming_body() -> None: - client = TestClient(_make_app(enabled=True)) + client = TestClient(_make_app()) response = client.get("/stream", headers={TRACE_ID_HEADER: "stream-trace"}) @@ -76,7 +78,7 @@ def test_trace_header_added_to_streaming_response_without_consuming_body() -> No def test_trace_header_overwrites_duplicate_downstream_value() -> None: - client = TestClient(_make_app(enabled=True)) + client = TestClient(_make_app()) response = client.get("/pre-set", headers={TRACE_ID_HEADER: "canonical-trace"}) @@ -84,16 +86,6 @@ def test_trace_header_overwrites_duplicate_downstream_value() -> None: assert response.headers.get_list(TRACE_ID_HEADER) == ["canonical-trace"] -def test_trace_header_marks_inbound_header_flag() -> None: - client = TestClient(_make_app(enabled=True)) - - with_header = client.get("/header-flag", headers={TRACE_ID_HEADER: "trace-from-upstream"}) - without_header = client.get("/header-flag") - - assert with_header.json() == {"from_header": True} - assert without_header.json() == {"from_header": False} - - def test_trace_header_rejects_crafted_non_ascii_and_generates_fresh_id() -> None: """A caller-crafted ``X-Trace-Id`` containing codepoints > 0x7E must not reach the response header. Prior to tightening ``normalize_trace_id`` such @@ -108,7 +100,7 @@ def test_trace_header_rejects_crafted_non_ascii_and_generates_fresh_id() -> None attacker's ``curl -H 'X-Trace-Id: 请求-1'`` would put on the wire (UTF-8 bytes that Starlette then latin-1-decodes into codepoints > 0x7E). """ - client = TestClient(_make_app(enabled=True)) + client = TestClient(_make_app()) # Raw UTF-8 bytes of "café-1"; Starlette latin-1-decodes them into # a string containing 0xC3, 0xA9 — both > 0x7E. @@ -128,7 +120,7 @@ def test_trace_header_rejects_crafted_c1_control_and_generates_fresh_id() -> Non or rejected by hardened intermediaries, so they must not survive validation either. Sent as raw bytes to bypass the ``httpx`` client-side ASCII check.""" - client = TestClient(_make_app(enabled=True)) + client = TestClient(_make_app()) crafted_bytes = b"trace\x9fid" crafted_decoded = crafted_bytes.decode("latin-1") @@ -140,59 +132,99 @@ def test_trace_header_rejects_crafted_c1_control_and_generates_fresh_id() -> Non assert all(0x20 <= ord(ch) <= 0x7E for ch in returned), returned -def test_enabled_is_a_startup_snapshot_not_a_live_read() -> None: - """`logging` is startup-only (see reload_boundary.STARTUP_ONLY_FIELDS), so - the middleware must capture the flag by value at construction time. A - later mutation of the source object must not flip request-time behavior, - otherwise the response `X-Trace-Id` would drift out of sync with the - log formatter installed once by `configure_logging()` at startup. - """ - source = {"enabled": True} - app = FastAPI() - app.add_middleware(TraceMiddleware, enabled=source["enabled"]) +def test_create_app_wires_trace_middleware_into_the_real_stack(monkeypatch) -> None: + """Every other case here pins the middleware's behavior on a hand-built + app; this one pins the wiring. ``create_app()`` must install + ``TraceMiddleware`` itself — dropping that ``add_middleware`` line (or + short-circuiting above it) would strip the header and the ambient id that + the run-record stamp and enhanced log records derive from, while every + hand-wired suite still passed.""" + import app.gateway.app as app_module + import deerflow.extensions as extensions_module + from deerflow.config.app_config import AppConfig + from deerflow.config.sandbox_config import SandboxConfig + from deerflow.extensions import reset_loaded_extensions + from deerflow.extensions.registry import ExtensionRegistry - @app.get("/plain") - async def plain() -> dict[str, str | None]: - return {"trace_id": get_current_trace_id()} + monkeypatch.setattr(app_module, "get_app_config", lambda: AppConfig(sandbox=SandboxConfig(use="test"))) + monkeypatch.setattr(extensions_module, "load_extensions", lambda plugins: (ExtensionRegistry().build(), [])) + try: + client = TestClient(app_module.create_app()) + response = client.get("/health", headers={TRACE_ID_HEADER: "wired-through-create-app"}) + finally: + reset_loaded_extensions() - client = TestClient(app) - - source["enabled"] = False # would matter if the middleware read live - response = client.get("/plain") - - assert TRACE_ID_HEADER in response.headers - assert response.json()["trace_id"] is not None + assert response.status_code == 200 + assert response.headers[TRACE_ID_HEADER] == "wired-through-create-app" -def test_resolve_trace_enabled_walks_nested_config() -> None: - config = SimpleNamespace(logging=SimpleNamespace(enhance=SimpleNamespace(enabled=True))) - assert resolve_trace_enabled(config) is True +def test_unhandled_exception_500_carries_trace_header() -> None: + """Starlette's ServerErrorMiddleware sits outside every user middleware and + emits unhandled-exception 500s through the raw send, so those responses + never pass the header-writing wrapper -- yet the 500 for a server bug is + exactly the response a user most needs to correlate with a log line. The + middleware must ship its own 500 carrying the id before re-raising.""" + app = _make_app() - config_off = SimpleNamespace(logging=SimpleNamespace(enhance=SimpleNamespace(enabled=False))) - assert resolve_trace_enabled(config_off) is False + @app.get("/boom") + async def boom() -> None: + raise RuntimeError("server bug") + + client = TestClient(app, raise_server_exceptions=False) + + response = client.get("/boom", headers={TRACE_ID_HEADER: "trace-from-upstream"}) + + assert response.status_code == 500 + assert response.headers[TRACE_ID_HEADER] == "trace-from-upstream" + # Byte-identical to the ServerErrorMiddleware response it replaces: an + # explicit content-length, not server-chosen framing (chunked on HTTP/1.1, + # close-delimited on HTTP/1.0). + assert response.text == "Internal Server Error" + assert response.headers["content-length"] == str(len(b"Internal Server Error")) -def test_resolve_trace_enabled_defaults_to_false_when_fields_missing() -> None: - assert resolve_trace_enabled(SimpleNamespace()) is False - assert resolve_trace_enabled(SimpleNamespace(logging=None)) is False - assert resolve_trace_enabled(SimpleNamespace(logging=SimpleNamespace(enhance=None))) is False +def test_unhandled_exception_500_carries_generated_trace_header() -> None: + app = _make_app() + + @app.get("/boom") + async def boom() -> None: + raise RuntimeError("server bug") + + client = TestClient(app, raise_server_exceptions=False) + + response = client.get("/boom") + + assert response.status_code == 500 + returned = response.headers[TRACE_ID_HEADER] + assert returned + assert all(0x20 <= ord(ch) <= 0x7E for ch in returned), returned -def test_gateway_app_construction_trace_flag_defaults_false_when_config_missing(monkeypatch) -> None: - import app.gateway.app as gateway_app +def test_midstream_exception_propagates_without_second_response_start() -> None: + """An exception after ``http.response.start`` keeps propagating unchanged: + a second response start cannot be sent, and the already-written header + stands on the one that was.""" + sent: list[dict] = [] - def missing_config(): - raise FileNotFoundError("no config") + async def failing_app(scope, receive, send) -> None: + await send({"type": "http.response.start", "status": 200, "headers": []}) + await send({"type": "http.response.body", "body": b"partial", "more_body": True}) + raise RuntimeError("mid-stream bug") - monkeypatch.setattr(gateway_app, "get_app_config", missing_config) + async def record(message) -> None: + sent.append(message) - assert gateway_app._resolve_trace_enabled_for_app_construction() is False + middleware = TraceMiddleware(failing_app) + scope = {"type": "http", "method": "GET", "path": "/", "headers": []} + async def scenario() -> None: + with pytest.raises(RuntimeError, match="mid-stream bug"): + await middleware(scope, None, record) -def test_gateway_app_construction_trace_flag_uses_config_snapshot(monkeypatch) -> None: - import app.gateway.app as gateway_app + asyncio.run(scenario()) - config = SimpleNamespace(logging=SimpleNamespace(enhance=SimpleNamespace(enabled=True))) - monkeypatch.setattr(gateway_app, "get_app_config", lambda: config) - - assert gateway_app._resolve_trace_enabled_for_app_construction() is True + starts = [message for message in sent if message["type"] == "http.response.start"] + assert len(starts) == 1 + assert starts[0]["status"] == 200 + header_names = {name.lower() for name, _ in starts[0]["headers"]} + assert TRACE_ID_HEADER.lower().encode("latin-1") in header_names diff --git a/backend/tests/test_worker_langfuse_metadata.py b/backend/tests/test_worker_langfuse_metadata.py index 0010c52c6..d7d2f5df0 100644 --- a/backend/tests/test_worker_langfuse_metadata.py +++ b/backend/tests/test_worker_langfuse_metadata.py @@ -16,9 +16,7 @@ from deerflow.runtime.runs.schemas import DisconnectMode, RunStatus from deerflow.runtime.runs.worker import RunContext, run_agent from deerflow.trace_context import ( DEERFLOW_TRACE_METADATA_KEY, - mark_trace_id_from_request_header, request_trace_context, - reset_trace_id_from_request_header, ) @@ -296,15 +294,20 @@ async def test_run_agent_preserves_caller_metadata_overrides(monkeypatch): # Caller-supplied keys win. assert metadata["langfuse_session_id"] == "custom-session-id" assert metadata["langfuse_user_id"] == "explicit-user" - assert metadata[DEERFLOW_TRACE_METADATA_KEY] == "explicit-deerflow-trace" - assert fake_agent.captured_config.get("context", {}).get(DEERFLOW_TRACE_METADATA_KEY) == "explicit-deerflow-trace" + # ...except deerflow_trace_id, which the server issues. Honouring the + # caller here would let the persisted run point at an id that matches + # neither the response header nor the log lines for the same request. + assert metadata[DEERFLOW_TRACE_METADATA_KEY] != "explicit-deerflow-trace" + assert metadata[DEERFLOW_TRACE_METADATA_KEY] == fake_agent.captured_config["context"][DEERFLOW_TRACE_METADATA_KEY] # Worker still fills in keys that the caller didn't set. assert metadata["langfuse_trace_name"] == "lead-agent" @pytest.mark.asyncio -async def test_run_agent_inbound_header_trace_overrides_metadata(monkeypatch): - """A valid inbound ``X-Trace-Id`` wins over ``config.metadata.deerflow_trace_id``.""" +async def test_run_agent_overwrites_caller_supplied_trace_id(monkeypatch): + """The bound request trace is the only source. A ``deerflow_trace_id`` in + the caller's metadata is replaced, not honoured, so the persisted run + cannot disagree with the header and the logs from the same request.""" monkeypatch.setenv("LANGFUSE_TRACING", "true") monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-lf-test") monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-lf-test") @@ -328,24 +331,20 @@ async def test_run_agent_inbound_header_trace_overrides_metadata(monkeypatch): ctx = RunContext(checkpointer=None) with request_trace_context("header-trace-1"): - header_token = mark_trace_id_from_request_header(from_header=True) - try: - await run_agent( - _FakeBridge(), - _FakeRunManager(), - record, - ctx=ctx, - agent_factory=agent_factory, - graph_input={"messages": []}, - config={ - "configurable": {"thread_id": "thread-header"}, - "metadata": { - DEERFLOW_TRACE_METADATA_KEY: "metadata-trace-ignored", - }, + await run_agent( + _FakeBridge(), + _FakeRunManager(), + record, + ctx=ctx, + agent_factory=agent_factory, + graph_input={"messages": []}, + config={ + "configurable": {"thread_id": "thread-header"}, + "metadata": { + DEERFLOW_TRACE_METADATA_KEY: "metadata-trace-ignored", }, - ) - finally: - reset_trace_id_from_request_header(header_token) + }, + ) metadata = fake_agent.captured_config.get("metadata") or {} assert metadata[DEERFLOW_TRACE_METADATA_KEY] == "header-trace-1" diff --git a/backend/tests/test_worker_trace_binding.py b/backend/tests/test_worker_trace_binding.py new file mode 100644 index 000000000..56fe33734 --- /dev/null +++ b/backend/tests/test_worker_trace_binding.py @@ -0,0 +1,189 @@ +"""``run_agent`` stamps the request trace id onto everything it hands the graph. + +The trace ContextVar is the only source. These tests pin the other half of +that contract: a ``deerflow_trace_id`` arriving on the run request is a +caller's echo of a past output, not an input, and must not survive into the +runtime context, the run metadata, or the checkpoint. Otherwise a client can +make the most durable surfaces of a run disagree with the ``X-Trace-Id`` and +the log lines the same request produced. +""" + +from __future__ import annotations + +import asyncio + +import pytest + +from deerflow.runtime.runs.manager import RunRecord, RunStartOutcome +from deerflow.runtime.runs.schemas import DisconnectMode, RunStatus +from deerflow.runtime.runs.worker import RunContext, _build_runtime_context, run_agent +from deerflow.trace_context import DEERFLOW_TRACE_METADATA_KEY, get_current_trace_id, request_trace_context + + +class _FakeAgent: + def __init__(self) -> None: + self.captured_config: dict | None = None + self.metadata: dict = {} + self.checkpointer = None + self.store = None + self.interrupt_before_nodes: list[str] = [] + self.interrupt_after_nodes: list[str] = [] + + async def astream(self, graph_input, *, config, stream_mode, **kwargs): + self.captured_config = config + return + yield # pragma: no cover (makes this an async generator) + + +class _FakeRunManager: + async def try_start(self, _run_id: str) -> RunStartOutcome: + return RunStartOutcome.started + + async def wait_for_prior_finalizing(self, *_args, **_kwargs) -> None: + return None + + async def has_later_run(self, *_args, **_kwargs) -> bool: + return False + + async def has_later_started_run(self, *_args, **_kwargs) -> bool: + return False + + async def set_status(self, *_args, **_kwargs) -> None: + return None + + async def set_status_if_not_cancelled(self, *_args, **_kwargs) -> None: + return None + + async def update_model_name(self, *_args, **_kwargs) -> None: + return None + + async def update_run_completion(self, *_args, **_kwargs) -> None: + return None + + async def cleanup(self, *_args, **_kwargs) -> None: + return None + + +class _FakeBridge: + async def publish(self, _run_id, event, payload) -> None: + return None + + async def publish_end(self, _run_id) -> None: + return None + + async def cleanup(self, _run_id, *, delay: int = 0) -> None: + return None + + +async def _run(config: dict) -> dict: + """Drive ``run_agent`` once and return the config the graph received.""" + fake_agent = _FakeAgent() + record = RunRecord( + run_id="run-trace-binding", + thread_id="thread-trace-binding", + assistant_id="lead-agent", + status=RunStatus.pending, + on_disconnect=DisconnectMode.cancel, + ) + record.abort_event = asyncio.Event() + + await run_agent( + _FakeBridge(), + _FakeRunManager(), + record, + ctx=RunContext(checkpointer=None), + agent_factory=lambda **_kwargs: fake_agent, + graph_input={"messages": []}, + config=config, + ) + assert fake_agent.captured_config is not None + return fake_agent.captured_config + + +@pytest.mark.asyncio +async def test_runtime_context_and_metadata_carry_the_bound_trace_id(): + """Both destinations get the same id: the runtime context carries it across + boundaries the ContextVar does not cross, the metadata persists with the + checkpoint.""" + with request_trace_context("gateway-issued"): + captured = await _run({"configurable": {"thread_id": "thread-trace-binding"}}) + + assert captured["context"][DEERFLOW_TRACE_METADATA_KEY] == "gateway-issued" + assert captured["metadata"][DEERFLOW_TRACE_METADATA_KEY] == "gateway-issued" + + +@pytest.mark.asyncio +async def test_caller_supplied_metadata_trace_id_is_overwritten(): + with request_trace_context("gateway-issued"): + captured = await _run( + { + "configurable": {"thread_id": "thread-trace-binding"}, + "metadata": {DEERFLOW_TRACE_METADATA_KEY: "forged", "caller_key": "kept"}, + } + ) + + assert captured["metadata"][DEERFLOW_TRACE_METADATA_KEY] == "gateway-issued" + # Only the server-owned key is replaced. + assert captured["metadata"]["caller_key"] == "kept" + + +@pytest.mark.asyncio +async def test_caller_supplied_context_trace_id_is_overwritten(): + """``config['context']`` is a second, separate way in. The Gateway filters + ``__``-prefixed keys out of it, but ``deerflow_trace_id`` carries no prefix + and embedded harness callers pass through no such filter at all.""" + with request_trace_context("gateway-issued"): + captured = await _run( + { + "configurable": {"thread_id": "thread-trace-binding"}, + "context": {DEERFLOW_TRACE_METADATA_KEY: "forged", "agent_name": "kept"}, + } + ) + + assert captured["context"][DEERFLOW_TRACE_METADATA_KEY] == "gateway-issued" + assert captured["context"]["agent_name"] == "kept" + + +@pytest.mark.asyncio +async def test_both_forks_agree_when_the_caller_forges_both(): + """The failure this rules out is disagreement, not any single wrong value.""" + with request_trace_context("gateway-issued"): + captured = await _run( + { + "configurable": {"thread_id": "thread-trace-binding"}, + "metadata": {DEERFLOW_TRACE_METADATA_KEY: "forged-metadata"}, + "context": {DEERFLOW_TRACE_METADATA_KEY: "forged-context"}, + } + ) + + assert captured["metadata"][DEERFLOW_TRACE_METADATA_KEY] == captured["context"][DEERFLOW_TRACE_METADATA_KEY] == "gateway-issued" + + +@pytest.mark.asyncio +async def test_run_without_an_ambient_trace_still_gets_one(): + """A run reached outside any entry point -- a standalone harness caller -- + is still correlatable rather than falling back to an absent id.""" + assert get_current_trace_id() is None + + captured = await _run({"configurable": {"thread_id": "thread-trace-binding"}}) + + assert captured["metadata"][DEERFLOW_TRACE_METADATA_KEY] + assert captured["context"][DEERFLOW_TRACE_METADATA_KEY] == captured["metadata"][DEERFLOW_TRACE_METADATA_KEY] + + +def test_build_runtime_context_drops_a_caller_supplied_trace_id(): + """Pinned on the builder itself, not through ``run_agent``. + + ``_bind_trace_id`` overwrites the key immediately afterwards, so at the one + current call site this guard is masked. It is the builder's own contract + that server-owned keys never come from the caller, and a second call site + added later must inherit that without having to remember the ordering. + """ + runtime_ctx = _build_runtime_context( + "thread-1", + "run-1", + {DEERFLOW_TRACE_METADATA_KEY: "forged", "agent_name": "kept"}, + ) + + assert DEERFLOW_TRACE_METADATA_KEY not in runtime_ctx + assert runtime_ctx["agent_name"] == "kept" diff --git a/config.example.yaml b/config.example.yaml index 97783fe2b..6a260c23f 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -23,8 +23,10 @@ config_version: 38 # Log level for deerflow modules (debug/info/warning/error) log_level: info -# Request trace correlation for Gateway logs, HTTP response headers, and -# Langfuse metadata. Disabled by default to preserve existing HTTP/log output. +# Trace ids are always issued and always returned in the `X-Trace-Id` response +# header; this block controls log output only — whether records carry a +# `trace_id` field, and in which format. Off by default because enabling it +# changes the log format. Restart required (see reload_boundary.py). logging: enhance: enabled: false