diff --git a/README.md b/README.md index bb66faf08..1fd84ffc1 100644 --- a/README.md +++ b/README.md @@ -536,6 +536,19 @@ Once a channel is connected, you can interact with DeerFlow directly from the ch > Messages without a command prefix are treated as regular chat — DeerFlow creates a thread and responds conversationally. +#### Request Trace Correlation + +Gateway request trace correlation is disabled by default so existing HTTP responses and log formats stay unchanged. To enable it, set: + +```yaml +logging: + enhance: + enabled: true + format: text +``` + +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. + #### LangSmith Tracing DeerFlow has built-in [LangSmith](https://smith.langchain.com) integration for observability. When enabled, all LLM calls, agent runs, and tool executions are traced and visible in the LangSmith dashboard. @@ -570,6 +583,7 @@ If you are using a self-hosted Langfuse instance, set `LANGFUSE_BASE_URL` to you - `user_id` = effective user from `get_effective_user_id()` (falls back to `default` in no-auth mode) - `trace_name` = assistant id (defaults to `lead-agent`) - `tags` = `[env:, model:]` (omitted when not set) +- `metadata.deerflow_trace_id` = DeerFlow request correlation id, matching `X-Trace-Id` when request trace correlation is enabled These are injected into `RunnableConfig.metadata` at the graph invocation root for both the gateway path (`runtime/runs/worker.py::run_agent`) and the embedded path (`client.py::DeerFlowClient.stream`), so any LangChain-compatible callback can read them. Set `DEER_FLOW_ENV` (or `ENVIRONMENT`) to tag traces by deployment environment. diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 2074e28c7..16041ce28 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -248,7 +248,7 @@ Setup: Copy `config.example.yaml` to `config.yaml` in the **project root** direc **Config Hot-Reload Boundary**: Gateway dependencies route through `get_app_config()` on every request, so per-run fields like `models[*].max_tokens`, `summarization.*`, `title.*`, `memory.*`, `subagents.*`, `tools[*]`, and the agent system prompt pick up `config.yaml` edits on the next message. `AppConfig` is intentionally **not** cached on `app.state` — `lifespan()` keeps a local `startup_config` variable for one-shot bootstrap work and passes it to `langgraph_runtime(app, startup_config)`. -Infrastructure fields are **restart-required**. The authoritative list lives in `packages/harness/deerflow/config/reload_boundary.py::STARTUP_ONLY_FIELDS` and is mirrored by the standardised `"startup-only:"` prefix on the corresponding `Field(description=...)` in `AppConfig`, so IDE hover on those fields surfaces the reason inline (no need to context-switch into this table). Currently registered: `database`, `checkpointer`, `run_events`, `stream_bridge`, `sandbox`, `log_level`, `channels`, `channel_connections`. Adding a new restart-required field requires updating the registry; drift is pinned by `tests/test_reload_boundary.py`. +Infrastructure fields are **restart-required**. The authoritative list lives in `packages/harness/deerflow/config/reload_boundary.py::STARTUP_ONLY_FIELDS` and is mirrored by the standardised `"startup-only:"` prefix on the corresponding `Field(description=...)` in `AppConfig`, so IDE hover on those fields surfaces the reason inline (no need to context-switch into this table). Currently registered: `database`, `checkpointer`, `run_events`, `stream_bridge`, `sandbox`, `log_level`, `logging`, `channels`, `channel_connections`. Adding a new restart-required field requires updating the registry; drift is pinned by `tests/test_reload_boundary.py`. **Persistence backend resolution**: the unified `database` section selects the Gateway's LangGraph checkpointer, LangGraph Store, and DeerFlow SQL repositories. @@ -576,6 +576,33 @@ A terminal-native UI over the embedded harness, exposed as the `deerflow` consol **Tests**: `tests/test_tui_*.py` — pure layers via plain pytest, the app/palette/overlays via Textual's pilot harness with a fake in-process session, and `test_tui_persistence.py` for the `threads_meta` round-trip. +### 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: + +- **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. 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 same ContextVar value is injected into enhanced log records as `trace_id` and into Langfuse metadata as `deerflow_trace_id`. + +`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. + ### Tracing System (`packages/harness/deerflow/tracing/`) LangSmith and Langfuse are both supported. The wiring lives in two layers: @@ -591,6 +618,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(...)` | 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`. @@ -598,6 +626,7 @@ Returns `{}` when Langfuse is not in the enabled providers — LangSmith-only de **`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` - 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/app/gateway/__init__.py b/backend/app/gateway/__init__.py index cab046712..775fd1dd7 100644 --- a/backend/app/gateway/__init__.py +++ b/backend/app/gateway/__init__.py @@ -1,4 +1,12 @@ -from .app import app, create_app from .config import GatewayConfig, get_gateway_config __all__ = ["app", "create_app", "GatewayConfig", "get_gateway_config"] + + +def __getattr__(name: str): + """Lazily expose the FastAPI app without initializing it on package import.""" + if name in {"app", "create_app"}: + from .app import app, create_app + + return app if name == "app" else create_app + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") diff --git a/backend/app/gateway/app.py b/backend/app/gateway/app.py index 26041cd75..1880383c4 100644 --- a/backend/app/gateway/app.py +++ b/backend/app/gateway/app.py @@ -30,8 +30,9 @@ from app.gateway.routers import ( threads, uploads, ) +from app.gateway.trace_middleware import TraceMiddleware, resolve_trace_enabled from deerflow.config import app_config as deerflow_app_config -from deerflow.config.app_config import apply_logging_level +from deerflow.logging_config import DEFAULT_LOG_DATE_FORMAT, DEFAULT_LOG_FORMAT, configure_logging from deerflow.uploads.manager import cleanup_stale_upload_staging_files AppConfig = deerflow_app_config.AppConfig @@ -40,8 +41,8 @@ get_app_config = deerflow_app_config.get_app_config # Default logging; lifespan overrides from config.yaml log_level. logging.basicConfig( level=logging.INFO, - format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", - datefmt="%Y-%m-%d %H:%M:%S", + format=DEFAULT_LOG_FORMAT, + datefmt=DEFAULT_LOG_DATE_FORMAT, ) logger = logging.getLogger(__name__) @@ -174,7 +175,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: # snapshot on `app.state` to keep that contract enforceable. try: startup_config = get_app_config() - apply_logging_level(startup_config.log_level) + configure_logging(startup_config) logger.info("Configuration loaded successfully") warn_if_auth_disabled_enabled() except Exception as e: @@ -371,6 +372,14 @@ This gateway provides runtime endpoints for agent runs plus custom endpoints for allow_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()) + # Include routers # Models API is mounted at /api/models app.include_router(models.router) @@ -435,5 +444,15 @@ 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/routers/suggestions.py b/backend/app/gateway/routers/suggestions.py index 72b4dca8d..2bbe8dae2 100644 --- a/backend/app/gateway/routers/suggestions.py +++ b/backend/app/gateway/routers/suggestions.py @@ -1,5 +1,6 @@ import json import logging +import os import re from fastapi import APIRouter, Depends, Request @@ -10,6 +11,8 @@ from app.gateway.authz import require_permission from app.gateway.deps import get_config from deerflow.config.app_config import AppConfig from deerflow.models import create_chat_model +from deerflow.runtime.user_context import get_effective_user_id +from deerflow.tracing import inject_langfuse_metadata logger = logging.getLogger(__name__) @@ -175,7 +178,16 @@ async def generate_suggestions( try: model = create_chat_model(name=body.model_name, thinking_enabled=False, app_config=config) - response = await model.ainvoke([SystemMessage(content=system_instruction), HumanMessage(content=user_content)], config={"run_name": "suggest_agent"}) + invoke_config: dict = {"run_name": "suggest_agent"} + inject_langfuse_metadata( + invoke_config, + thread_id=thread_id, + user_id=get_effective_user_id(), + assistant_id="suggest_agent", + model_name=body.model_name, + environment=os.environ.get("DEER_FLOW_ENV") or os.environ.get("ENVIRONMENT"), + ) + response = await model.ainvoke([SystemMessage(content=system_instruction), HumanMessage(content=user_content)], config=invoke_config) raw = _extract_response_text(response.content) suggestions = _parse_json_string_list(raw) or [] cleaned = [s.replace("\n", " ").strip() for s in suggestions if s.strip()] diff --git a/backend/app/gateway/trace_middleware.py b/backend/app/gateway/trace_middleware.py new file mode 100644 index 000000000..014cebbe7 --- /dev/null +++ b/backend/app/gateway/trace_middleware.py @@ -0,0 +1,63 @@ +"""Gateway request trace middleware.""" + +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, request_trace_context + +logger = logging.getLogger(__name__) + + +class TraceMiddleware: + """Bind a request-level trace id and write it to HTTP response headers. + + 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``. + """ + + def __init__(self, app: ASGIApp, *, enabled: bool): + 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: + await self.app(scope, receive, send) + return + + headers = Headers(scope=scope) + incoming_trace_id = headers.get(TRACE_ID_HEADER) + + with request_trace_context(incoming_trace_id) as trace_id: + + 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) + + +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) diff --git a/backend/packages/harness/deerflow/agents/memory/queue.py b/backend/packages/harness/deerflow/agents/memory/queue.py index 129a28c66..d7a04e518 100644 --- a/backend/packages/harness/deerflow/agents/memory/queue.py +++ b/backend/packages/harness/deerflow/agents/memory/queue.py @@ -3,11 +3,13 @@ import logging import threading import time +from contextlib import nullcontext from dataclasses import dataclass, field from datetime import UTC, datetime from typing import Any from deerflow.config.memory_config import get_memory_config +from deerflow.trace_context import request_trace_context logger = logging.getLogger(__name__) @@ -21,6 +23,7 @@ class ConversationContext: timestamp: datetime = field(default_factory=lambda: datetime.now(UTC)) agent_name: str | None = None user_id: str | None = None + deerflow_trace_id: str | None = None correction_detected: bool = False reinforcement_detected: bool = False @@ -55,6 +58,7 @@ class MemoryUpdateQueue: messages: list[Any], agent_name: str | None = None, user_id: str | None = None, + deerflow_trace_id: str | None = None, correction_detected: bool = False, reinforcement_detected: bool = False, ) -> None: @@ -67,6 +71,8 @@ class MemoryUpdateQueue: user_id: The user ID captured at enqueue time. Stored in ConversationContext so it survives the threading.Timer boundary (ContextVar does not propagate across raw threads). + deerflow_trace_id: Request trace id captured at enqueue time so the + later Timer thread can attach it to memory LLM tracing metadata. correction_detected: Whether recent turns include an explicit correction signal. reinforcement_detected: Whether recent turns include a positive reinforcement signal. """ @@ -80,6 +86,7 @@ class MemoryUpdateQueue: messages=messages, agent_name=agent_name, user_id=user_id, + deerflow_trace_id=deerflow_trace_id, correction_detected=correction_detected, reinforcement_detected=reinforcement_detected, ) @@ -93,6 +100,7 @@ class MemoryUpdateQueue: messages: list[Any], agent_name: str | None = None, user_id: str | None = None, + deerflow_trace_id: str | None = None, correction_detected: bool = False, reinforcement_detected: bool = False, ) -> None: @@ -107,6 +115,7 @@ class MemoryUpdateQueue: messages=messages, agent_name=agent_name, user_id=user_id, + deerflow_trace_id=deerflow_trace_id, correction_detected=correction_detected, reinforcement_detected=reinforcement_detected, ) @@ -121,6 +130,7 @@ class MemoryUpdateQueue: messages: list[Any], agent_name: str | None, user_id: str | None, + deerflow_trace_id: str | None, correction_detected: bool, reinforcement_detected: bool, ) -> None: @@ -136,6 +146,7 @@ class MemoryUpdateQueue: messages=messages, agent_name=agent_name, user_id=user_id, + deerflow_trace_id=deerflow_trace_id, correction_detected=merged_correction_detected, reinforcement_detected=merged_reinforcement_detected, ) @@ -188,26 +199,35 @@ class MemoryUpdateQueue: updater = MemoryUpdater() for context in contexts_to_process: - try: - logger.info("Updating memory for thread %s", context.thread_id) - success = updater.update_memory( - messages=context.messages, - thread_id=context.thread_id, - agent_name=context.agent_name, - correction_detected=context.correction_detected, - reinforcement_detected=context.reinforcement_detected, - user_id=context.user_id, - ) - if success: - logger.info("Memory updated successfully for thread %s", context.thread_id) - else: - logger.warning("Memory update skipped/failed for thread %s", context.thread_id) - except Exception as e: - logger.error("Error updating memory for thread %s: %s", context.thread_id, e) + # Rebind the request-trace ContextVar from the value captured at + # enqueue time so ``TraceContextFilter`` attaches the correct + # trace id to every log record emitted below (this Timer thread + # does not inherit the enqueue-thread's ContextVar). Each + # iteration is scoped independently so id A does not leak into + # id B's logs. + trace_ctx = request_trace_context(context.deerflow_trace_id) if context.deerflow_trace_id else nullcontext() + with trace_ctx: + try: + logger.info("Updating memory for thread %s", context.thread_id) + success = updater.update_memory( + messages=context.messages, + thread_id=context.thread_id, + agent_name=context.agent_name, + correction_detected=context.correction_detected, + reinforcement_detected=context.reinforcement_detected, + user_id=context.user_id, + deerflow_trace_id=context.deerflow_trace_id, + ) + if success: + logger.info("Memory updated successfully for thread %s", context.thread_id) + else: + logger.warning("Memory update skipped/failed for thread %s", context.thread_id) + except Exception as e: + logger.error("Error updating memory for thread %s: %s", context.thread_id, e) - # Small delay between updates to avoid rate limiting - if len(contexts_to_process) > 1: - time.sleep(0.5) + # Small delay between updates to avoid rate limiting + if len(contexts_to_process) > 1: + time.sleep(0.5) finally: with self._lock: diff --git a/backend/packages/harness/deerflow/agents/memory/updater.py b/backend/packages/harness/deerflow/agents/memory/updater.py index f51a3dd78..1391c7190 100644 --- a/backend/packages/harness/deerflow/agents/memory/updater.py +++ b/backend/packages/harness/deerflow/agents/memory/updater.py @@ -7,8 +7,10 @@ import copy import json import logging import math +import os import re import uuid +from contextlib import nullcontext from typing import Any from deerflow.agents.memory.prompt import ( @@ -22,6 +24,8 @@ from deerflow.agents.memory.storage import ( ) from deerflow.config.memory_config import get_memory_config from deerflow.models import create_chat_model +from deerflow.trace_context import request_trace_context +from deerflow.tracing import inject_langfuse_metadata logger = logging.getLogger(__name__) @@ -385,9 +389,12 @@ class MemoryUpdater: def _get_model(self): """Get the model for memory updates.""" + return create_chat_model(name=self._resolve_model_name(), thinking_enabled=False) + + def _resolve_model_name(self) -> str | None: + """Return the configured model name for memory updates.""" config = get_memory_config() - model_name = self._model_name or config.model_name - return create_chat_model(name=model_name, thinking_enabled=False) + return self._model_name or config.model_name def _build_correction_hint( self, @@ -467,6 +474,7 @@ class MemoryUpdater: correction_detected: bool = False, reinforcement_detected: bool = False, user_id: str | None = None, + deerflow_trace_id: str | None = None, ) -> bool: """Update memory asynchronously by delegating to the sync path. @@ -484,6 +492,7 @@ class MemoryUpdater: correction_detected=correction_detected, reinforcement_detected=reinforcement_detected, user_id=user_id, + deerflow_trace_id=deerflow_trace_id, ) def _do_update_memory_sync( @@ -494,6 +503,7 @@ class MemoryUpdater: correction_detected: bool = False, reinforcement_detected: bool = False, user_id: str | None = None, + deerflow_trace_id: str | None = None, ) -> bool: """Pure-sync memory update using ``model.invoke()``. @@ -503,33 +513,53 @@ class MemoryUpdater: lead agent) is never touched — no cross-loop connection reuse is possible. """ - try: - prepared = self._prepare_update_prompt( - messages=messages, - agent_name=agent_name, - correction_detected=correction_detected, - reinforcement_detected=reinforcement_detected, - user_id=user_id, - ) - if prepared is None: - return False + # Callers may run us in a ``threading.Timer`` thread or an + # ``_SYNC_MEMORY_UPDATER_EXECUTOR`` worker — neither propagates the + # request-trace ContextVar. Rebind it here from the explicitly plumbed + # ``deerflow_trace_id`` so ``TraceContextFilter`` attaches the correct + # trace id to every log record emitted below (including model-invoke + # tracing-callback logs). ``nullcontext`` when unknown avoids + # fabricating a bogus id via ``request_trace_context(None)``. + trace_ctx = request_trace_context(deerflow_trace_id) if deerflow_trace_id else nullcontext() + with trace_ctx: + try: + prepared = self._prepare_update_prompt( + messages=messages, + agent_name=agent_name, + correction_detected=correction_detected, + reinforcement_detected=reinforcement_detected, + user_id=user_id, + ) + if prepared is None: + return False - current_memory, prompt = prepared - model = self._get_model() - response = model.invoke(prompt, config={"run_name": "memory_agent"}) - return self._finalize_update( - current_memory=current_memory, - response_content=response.content, - thread_id=thread_id, - agent_name=agent_name, - user_id=user_id, - ) - except json.JSONDecodeError as e: - logger.warning("Failed to parse LLM response for memory update: %s", e) - return False - except Exception as e: - logger.exception("Memory update failed: %s", e) - return False + current_memory, prompt = prepared + model_name = self._resolve_model_name() + model = self._get_model() + invoke_config: dict[str, Any] = {"run_name": "memory_agent"} + inject_langfuse_metadata( + invoke_config, + thread_id=thread_id, + user_id=user_id, + assistant_id="memory_agent", + model_name=model_name, + environment=os.environ.get("DEER_FLOW_ENV") or os.environ.get("ENVIRONMENT"), + deerflow_trace_id=deerflow_trace_id, + ) + response = model.invoke(prompt, config=invoke_config) + return self._finalize_update( + current_memory=current_memory, + response_content=response.content, + thread_id=thread_id, + agent_name=agent_name, + user_id=user_id, + ) + except json.JSONDecodeError as e: + logger.warning("Failed to parse LLM response for memory update: %s", e) + return False + except Exception as e: + logger.exception("Memory update failed: %s", e) + return False def update_memory( self, @@ -539,6 +569,7 @@ class MemoryUpdater: correction_detected: bool = False, reinforcement_detected: bool = False, user_id: str | None = None, + deerflow_trace_id: str | None = None, ) -> bool: """Synchronously update memory using the sync LLM path. @@ -577,6 +608,7 @@ class MemoryUpdater: correction_detected=correction_detected, reinforcement_detected=reinforcement_detected, user_id=user_id, + deerflow_trace_id=deerflow_trace_id, ) return future.result() except Exception: @@ -590,6 +622,7 @@ class MemoryUpdater: correction_detected=correction_detected, reinforcement_detected=reinforcement_detected, user_id=user_id, + deerflow_trace_id=deerflow_trace_id, ) def _apply_updates( @@ -691,6 +724,7 @@ def update_memory_from_conversation( correction_detected: bool = False, reinforcement_detected: bool = False, user_id: str | None = None, + deerflow_trace_id: str | None = None, ) -> bool: """Convenience function to update memory from a conversation. @@ -706,4 +740,4 @@ def update_memory_from_conversation( True if successful, False otherwise. """ updater = MemoryUpdater() - return updater.update_memory(messages, thread_id, agent_name, correction_detected, reinforcement_detected, user_id=user_id) + return updater.update_memory(messages, thread_id, agent_name, correction_detected, reinforcement_detected, user_id=user_id, deerflow_trace_id=deerflow_trace_id) diff --git a/backend/packages/harness/deerflow/agents/middlewares/memory_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/memory_middleware.py index ae5f9cfbb..21a8e96df 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/memory_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/memory_middleware.py @@ -12,6 +12,7 @@ from deerflow.agents.memory.message_processing import detect_correction, detect_ from deerflow.agents.memory.queue import get_memory_queue from deerflow.config.memory_config import get_memory_config from deerflow.runtime.user_context import get_effective_user_id +from deerflow.trace_context import DEERFLOW_TRACE_METADATA_KEY, get_current_trace_id, normalize_trace_id if TYPE_CHECKING: from deerflow.config.memory_config import MemoryConfig @@ -97,12 +98,24 @@ 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 = get_effective_user_id() + runtime_context = runtime.context if isinstance(runtime.context, dict) else {} + deerflow_trace_id = normalize_trace_id(runtime_context.get(DEERFLOW_TRACE_METADATA_KEY)) + if deerflow_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 {} + deerflow_trace_id = normalize_trace_id(config_metadata.get(DEERFLOW_TRACE_METADATA_KEY)) + if deerflow_trace_id is None: + deerflow_trace_id = get_current_trace_id() queue = get_memory_queue() queue.add( thread_id=thread_id, messages=filtered_messages, agent_name=self._agent_name, user_id=user_id, + deerflow_trace_id=deerflow_trace_id, correction_detected=correction_detected, reinforcement_detected=reinforcement_detected, ) diff --git a/backend/packages/harness/deerflow/client.py b/backend/packages/harness/deerflow/client.py index 5ecf1b15a..82ff61db7 100644 --- a/backend/packages/harness/deerflow/client.py +++ b/backend/packages/harness/deerflow/client.py @@ -37,13 +37,14 @@ from deerflow.agents.lead_agent.agent import build_middlewares from deerflow.agents.lead_agent.prompt import apply_prompt_template from deerflow.agents.thread_state import ThreadState from deerflow.config.agents_config import AGENT_NAME_PATTERN -from deerflow.config.app_config import get_app_config, reload_app_config +from deerflow.config.app_config import get_app_config, is_trace_correlation_enabled, reload_app_config from deerflow.config.extensions_config import ExtensionsConfig, SkillStateConfig, get_extensions_config, reload_extensions_config from deerflow.config.paths import get_paths from deerflow.models import create_chat_model from deerflow.runtime.user_context import get_effective_user_id from deerflow.skills.storage import get_or_new_skill_storage from deerflow.tools.builtins.tool_search import assemble_deferred_tools +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.tracing import build_tracing_callbacks, inject_langfuse_metadata from deerflow.uploads.manager import ( claim_unique_filename, @@ -518,6 +519,61 @@ class DeerFlowClient: *, thread_id: str | None = None, **kwargs, + ) -> 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. + """ + 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. + trace_id = get_current_trace_id() or generate_trace_id() + + # Bind the trace id only around each ``next()`` step, never across a + # ``yield``. ``stream()`` is a sync generator, which shares the + # caller's context — a ``with ensure_trace_context(): yield from ...`` + # would (1) leak the id into the caller's context between yields and + # (2) risk ``ValueError: Token was created in a different Context`` + # when GC finalizes an abandoned generator in a different context. + # 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) + _EXHAUSTED = object() + try: + while True: + token = set_current_trace_id(trace_id) + try: + try: + event = next(inner) + except StopIteration: + event = _EXHAUSTED + finally: + reset_current_trace_id(token) + if event is _EXHAUSTED: + break + yield event + finally: + inner.close() + + def _stream_without_trace_context( + self, + message: str, + *, + thread_id: str | None = None, + **kwargs, ) -> Generator[StreamEvent, None, None]: """Stream a conversation turn, yielding events incrementally. @@ -610,6 +666,7 @@ class DeerFlowClient: config["callbacks"] = [*existing_callbacks, *tracing_callbacks] configurable = config.get("configurable") or {} + deerflow_trace_id = get_current_trace_id() inject_langfuse_metadata( config, thread_id=thread_id, @@ -617,12 +674,15 @@ class DeerFlowClient: assistant_id=self._agent_name or "lead-agent", model_name=configurable.get("model_name") or self._model_name, environment=self._environment or os.environ.get("DEER_FLOW_ENV") or os.environ.get("ENVIRONMENT"), + deerflow_trace_id=deerflow_trace_id, ) self._ensure_agent(config) state: dict[str, Any] = {"messages": [HumanMessage(content=message)]} context = {"thread_id": thread_id} + if 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/app_config.py b/backend/packages/harness/deerflow/config/app_config.py index 8d12c5378..6ef27fdcf 100644 --- a/backend/packages/harness/deerflow/config/app_config.py +++ b/backend/packages/harness/deerflow/config/app_config.py @@ -4,7 +4,7 @@ import os from collections.abc import Mapping from contextvars import ContextVar from pathlib import Path -from typing import Any, Self +from typing import Any, Literal, Self import yaml from dotenv import load_dotenv @@ -57,6 +57,35 @@ class CircuitBreakerConfig(BaseModel): recovery_timeout_sec: int = Field(default=60, description="Time in seconds before attempting to recover the circuit") +class LoggingEnhanceConfig(BaseModel): + """Request trace logging enhancement settings.""" + + enabled: bool = Field(default=False, description="Enable request-level trace ids in Gateway response headers and log records.") + format: Literal["text", "json"] = Field(default="text", description="Enhanced log output format.") + + +class LoggingConfig(BaseModel): + """Logging configuration.""" + + 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] @@ -98,6 +127,13 @@ class AppConfig(BaseModel): field_doc="Logging level for deerflow and app modules (debug/info/warning/error); third-party libraries are not affected.", ), ) + logging: LoggingConfig = Field( + default_factory=LoggingConfig, + description=format_field_description( + "logging", + field_doc="Structured logging and request trace correlation settings.", + ), + ) token_usage: TokenUsageConfig = Field(default_factory=TokenUsageConfig, description="Token usage tracking configuration") token_budget: TokenBudgetConfig = Field(default_factory=TokenBudgetConfig, description="Token Budget tracking and limits configuration.") max_recursion_limit: int = Field( diff --git a/backend/packages/harness/deerflow/config/reload_boundary.py b/backend/packages/harness/deerflow/config/reload_boundary.py index 4c024fe52..f5d362495 100644 --- a/backend/packages/harness/deerflow/config/reload_boundary.py +++ b/backend/packages/harness/deerflow/config/reload_boundary.py @@ -51,6 +51,11 @@ STARTUP_ONLY_FIELDS: dict[str, str] = { "log_level": ( "apply_logging_level() runs only during app.py startup; it sets the deerflow/app logger levels and may lower root handler thresholds so configured messages can propagate. A freshly reloaded AppConfig does not retrigger it." ), + "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." + ), # Not part of the AppConfig Pydantic schema — channel credentials are # consumed directly by ``start_channel_service()`` once at lifespan # startup and the live channel clients are not rebuilt on diff --git a/backend/packages/harness/deerflow/logging_config.py b/backend/packages/harness/deerflow/logging_config.py new file mode 100644 index 000000000..164b62d21 --- /dev/null +++ b/backend/packages/harness/deerflow/logging_config.py @@ -0,0 +1,109 @@ +"""Logging setup helpers for DeerFlow.""" + +from __future__ import annotations + +import json +import logging +from datetime import UTC, datetime +from typing import Any + +from deerflow.config.app_config import apply_logging_level +from deerflow.trace_context import get_current_trace_id + +DEFAULT_LOG_DATE_FORMAT = "%Y-%m-%d %H:%M:%S" +DEFAULT_LOG_FORMAT = "%(asctime)s - %(name)s - %(levelname)s - %(message)s" +TRACE_TEXT_LOG_FORMAT = "%(asctime)s - %(name)s - %(levelname)s - [trace_id=%(trace_id)s] - %(message)s" +_TRACE_FILTER_NAME = "deerflow_trace_context_filter" + + +class TraceContextFilter(logging.Filter): + """Inject the current request trace id into every log record.""" + + name = _TRACE_FILTER_NAME + + def filter(self, record: logging.LogRecord) -> bool: + record.trace_id = get_current_trace_id() or "-" + return True + + +class JsonTraceFormatter(logging.Formatter): + """Small JSON formatter used when ``logging.enhance.format=json``.""" + + _deerflow_trace_formatter = True + + def format(self, record: logging.LogRecord) -> str: + if not hasattr(record, "trace_id"): + record.trace_id = get_current_trace_id() or "-" + payload: dict[str, Any] = { + "timestamp": datetime.fromtimestamp(record.created, UTC).isoformat(), + "logger": record.name, + "level": record.levelname, + "trace_id": record.trace_id, + "message": record.getMessage(), + } + if record.exc_info: + payload["exc_info"] = self.formatException(record.exc_info) + if record.stack_info: + payload["stack_info"] = self.formatStack(record.stack_info) + return json.dumps(payload, ensure_ascii=False) + + +class TraceTextFormatter(logging.Formatter): + """Marker subclass so trace formatting can be reverted cleanly in tests.""" + + _deerflow_trace_formatter = True + + +def _ensure_root_handler() -> None: + if logging.root.handlers: + return + logging.basicConfig(level=logging.INFO, format=DEFAULT_LOG_FORMAT, datefmt=DEFAULT_LOG_DATE_FORMAT) + + +def _has_trace_filter(handler: logging.Handler) -> bool: + return any(getattr(f, "name", None) == _TRACE_FILTER_NAME or isinstance(f, TraceContextFilter) for f in handler.filters) + + +def _install_trace_filter(handler: logging.Handler) -> None: + if not _has_trace_filter(handler): + handler.addFilter(TraceContextFilter()) + + +def _remove_trace_filter(handler: logging.Handler) -> None: + handler.filters = [f for f in handler.filters if not (getattr(f, "name", None) == _TRACE_FILTER_NAME or isinstance(f, TraceContextFilter))] + + +def _default_formatter() -> logging.Formatter: + return logging.Formatter(DEFAULT_LOG_FORMAT, datefmt=DEFAULT_LOG_DATE_FORMAT) + + +def _trace_formatter(format_name: str | None) -> logging.Formatter: + if (format_name or "text").strip().lower() == "json": + return JsonTraceFormatter() + return TraceTextFormatter(TRACE_TEXT_LOG_FORMAT, datefmt=DEFAULT_LOG_DATE_FORMAT) + + +def configure_logging(config: object) -> None: + """Configure DeerFlow logging from an AppConfig-like object. + + With logging enhancement disabled this preserves the previous + ``basicConfig + apply_logging_level`` behavior. With enhancement enabled, + root handlers gain a trace-context filter and a formatter that includes + only the additional ``trace_id`` field. + """ + _ensure_root_handler() + + logging_config = getattr(config, "logging", None) + enhance = getattr(logging_config, "enhance", None) + enhanced = bool(getattr(enhance, "enabled", False)) + + for handler in logging.root.handlers: + if enhanced: + _install_trace_filter(handler) + handler.setFormatter(_trace_formatter(getattr(enhance, "format", "text"))) + else: + _remove_trace_filter(handler) + if getattr(handler.formatter, "_deerflow_trace_formatter", False): + handler.setFormatter(_default_formatter()) + + apply_logging_level(getattr(config, "log_level", None)) diff --git a/backend/packages/harness/deerflow/runtime/runs/worker.py b/backend/packages/harness/deerflow/runtime/runs/worker.py index d51e16501..c8709f34d 100644 --- a/backend/packages/harness/deerflow/runtime/runs/worker.py +++ b/backend/packages/harness/deerflow/runtime/runs/worker.py @@ -30,6 +30,7 @@ from deerflow.config.app_config import AppConfig from deerflow.runtime.serialization import serialize from deerflow.runtime.stream_bridge import StreamBridge from deerflow.runtime.user_context import get_effective_user_id +from deerflow.trace_context import DEERFLOW_TRACE_METADATA_KEY, get_current_trace_id, normalize_trace_id from deerflow.tracing import inject_langfuse_metadata from .manager import RunManager, RunRecord @@ -91,6 +92,8 @@ 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"]) + if DEERFLOW_TRACE_METADATA_KEY in runtime_context: + existing_context.setdefault(DEERFLOW_TRACE_METADATA_KEY, runtime_context[DEERFLOW_TRACE_METADATA_KEY]) if "app_config" in runtime_context: existing_context["app_config"] = runtime_context["app_config"] return @@ -281,6 +284,10 @@ async def run_agent( # manually here because we drive the graph through ``agent.astream(config=...)`` # without passing the official ``context=`` parameter. runtime_ctx = _build_runtime_context(thread_id, run_id, config.get("context"), ctx.app_config) + incoming_metadata = config.get("metadata") if isinstance(config.get("metadata"), dict) else {} + deerflow_trace_id = normalize_trace_id(incoming_metadata.get(DEERFLOW_TRACE_METADATA_KEY)) or get_current_trace_id() + if deerflow_trace_id: + runtime_ctx[DEERFLOW_TRACE_METADATA_KEY] = deerflow_trace_id # 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 @@ -307,6 +314,7 @@ async def run_agent( assistant_id=record.assistant_id, model_name=record.model_name, environment=os.environ.get("DEER_FLOW_ENV") or os.environ.get("ENVIRONMENT"), + deerflow_trace_id=deerflow_trace_id, ) # Resolve after runtime context installation so context/configurable reflect diff --git a/backend/packages/harness/deerflow/subagents/executor.py b/backend/packages/harness/deerflow/subagents/executor.py index f47523860..1a85f610e 100644 --- a/backend/packages/harness/deerflow/subagents/executor.py +++ b/backend/packages/harness/deerflow/subagents/executor.py @@ -29,6 +29,7 @@ from deerflow.skills.types import Skill from deerflow.subagents.config import SubagentConfig, resolve_subagent_model_name 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.tracing import build_tracing_callbacks, inject_langfuse_metadata if TYPE_CHECKING: @@ -294,6 +295,7 @@ class SubagentExecutor: oauth_provider: str | None = None, oauth_id: str | None = None, run_id: str | None = None, + deerflow_trace_id: str | None = None, ): """Initialize the executor. @@ -316,6 +318,8 @@ class SubagentExecutor: oauth_id: Subject id at the external identity provider. run_id: Parent run id, so delegated guardrail decisions attribute to the same run as the lead agent. + deerflow_trace_id: DeerFlow request-level correlation id propagated + from the parent run for Langfuse metadata correlation. """ self.config = config self.app_config = app_config @@ -338,6 +342,7 @@ class SubagentExecutor: self.oauth_provider = oauth_provider self.oauth_id = oauth_id self.run_id = run_id + self.deerflow_trace_id = deerflow_trace_id self._base_tools = _filter_tools( tools, @@ -581,6 +586,7 @@ class SubagentExecutor: assistant_id=assistant_id, model_name=self.model_name, environment=os.environ.get("DEER_FLOW_ENV") or os.environ.get("ENVIRONMENT"), + deerflow_trace_id=self.deerflow_trace_id, ) context: dict[str, Any] = {} @@ -598,6 +604,8 @@ class SubagentExecutor: context["oauth_provider"] = self.oauth_provider context["oauth_id"] = self.oauth_id context["run_id"] = self.run_id + if 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 df2d970c4..4940ff08b 100644 --- a/backend/packages/harness/deerflow/tools/builtins/task_tool.py +++ b/backend/packages/harness/deerflow/tools/builtins/task_tool.py @@ -22,6 +22,7 @@ from deerflow.subagents.executor import ( request_cancel_background_task, ) from deerflow.tools.types import Runtime +from deerflow.trace_context import DEERFLOW_TRACE_METADATA_KEY, get_current_trace_id, normalize_trace_id if TYPE_CHECKING: from deerflow.config.app_config import AppConfig @@ -255,6 +256,7 @@ async def task_tool( parent_model = None trace_id = None user_id = None + deerflow_trace_id = None metadata: dict = {} if runtime is not None: @@ -287,6 +289,7 @@ async def task_tool( oauth_provider = parent_context.get("oauth_provider") oauth_id = parent_context.get("oauth_id") run_id = parent_context.get("run_id") + deerflow_trace_id = normalize_trace_id(parent_context.get(DEERFLOW_TRACE_METADATA_KEY)) or normalize_trace_id(metadata.get(DEERFLOW_TRACE_METADATA_KEY)) or get_current_trace_id() parent_available_skills = metadata.get("available_skills") if parent_available_skills is not None: @@ -330,6 +333,7 @@ async def task_tool( "oauth_provider": oauth_provider, "oauth_id": oauth_id, "run_id": run_id, + "deerflow_trace_id": deerflow_trace_id, } if resolved_app_config is not None: executor_kwargs["app_config"] = resolved_app_config diff --git a/backend/packages/harness/deerflow/trace_context.py b/backend/packages/harness/deerflow/trace_context.py new file mode 100644 index 000000000..1a7661798 --- /dev/null +++ b/backend/packages/harness/deerflow/trace_context.py @@ -0,0 +1,87 @@ +"""Request trace context helpers. + +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. +""" + +from __future__ import annotations + +import uuid +from collections.abc import Iterator +from contextlib import contextmanager +from contextvars import ContextVar, Token +from typing import Final + +TRACE_ID_HEADER: Final[str] = "X-Trace-Id" +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) + + +def generate_trace_id() -> str: + """Return a fresh header-safe trace id.""" + return uuid.uuid4().hex + + +def normalize_trace_id(value: object) -> str | None: + """Return a safe trace id string, or ``None`` when *value* is unusable. + + Only printable ASCII (0x20-0x7E) is accepted. Codepoints above 0x7E are + rejected because the trace id round-trips through HTTP response headers, + which Starlette encodes as latin-1: codepoints > 0xFF raise + ``UnicodeEncodeError`` inside ``MutableHeaders.__setitem__`` (forcing a + 500 before the response body is even dispatched), and C1 controls + (0x80-0x9F) technically encode but are stripped or rejected by hardened + intermediaries (nginx / envoy / cloudfront), silently breaking the + response. C0 controls (< 0x20) and DEL (0x7F) are rejected for the same + header-safety reason plus log-injection defense. + """ + if not isinstance(value, str): + return None + trace_id = value.strip() + if not trace_id or len(trace_id) > _MAX_TRACE_ID_LENGTH: + return None + if any(ord(ch) < 32 or ord(ch) > 126 for ch in trace_id): + return 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 _current_trace_id.get() + + +@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.""" + normalized = normalize_trace_id(trace_id) or generate_trace_id() + token = _current_trace_id.set(normalized) + try: + yield normalized + finally: + _current_trace_id.reset(token) + + +@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) + try: + yield normalized + finally: + _current_trace_id.reset(token) diff --git a/backend/packages/harness/deerflow/tracing/metadata.py b/backend/packages/harness/deerflow/tracing/metadata.py index 3dabf169a..2fe23f002 100644 --- a/backend/packages/harness/deerflow/tracing/metadata.py +++ b/backend/packages/harness/deerflow/tracing/metadata.py @@ -19,6 +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 # Lazy-imported below to avoid a circular import: ``deerflow.runtime`` eagerly # imports the run worker, which in turn needs ``deerflow.tracing``. @@ -32,6 +33,7 @@ def build_langfuse_trace_metadata( assistant_id: str | None = None, model_name: str | None = None, environment: str | None = None, + deerflow_trace_id: str | None = None, ) -> dict[str, Any]: """Return Langfuse trace-attribute metadata for ``RunnableConfig.metadata``. @@ -47,6 +49,8 @@ def build_langfuse_trace_metadata( model_name: Model name; emitted as ``model:`` in ``langfuse_tags``. 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. """ if "langfuse" not in get_enabled_tracing_providers(): return {} @@ -58,6 +62,9 @@ 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 tags: list[str] = [] if environment: @@ -78,6 +85,7 @@ def inject_langfuse_metadata( assistant_id: str | None = None, model_name: str | None = None, environment: str | None = None, + deerflow_trace_id: str | None = None, ) -> None: """Merge Langfuse trace-attribute metadata into ``config["metadata"]``. @@ -95,6 +103,7 @@ def inject_langfuse_metadata( assistant_id=assistant_id, model_name=model_name, environment=environment, + deerflow_trace_id=deerflow_trace_id, ) if not langfuse_metadata: return diff --git a/backend/tests/test_client_langfuse_metadata.py b/backend/tests/test_client_langfuse_metadata.py index 3116fd331..7f3a6ba10 100644 --- a/backend/tests/test_client_langfuse_metadata.py +++ b/backend/tests/test_client_langfuse_metadata.py @@ -15,6 +15,7 @@ from typing import Any import pytest from deerflow.client import DeerFlowClient +from deerflow.trace_context import DEERFLOW_TRACE_METADATA_KEY, request_trace_context class _FakeAgent: @@ -57,10 +58,18 @@ def _stub_agent_creation(monkeypatch, fake_agent: _FakeAgent) -> dict[str, Any]: return captured -def _make_client(_monkeypatch) -> DeerFlowClient: +def _make_client(_monkeypatch, *, enhance_enabled: bool = True) -> DeerFlowClient: """Build a client without going through ``__init__`` so we never load - config.yaml or perform any other side-effectful startup work.""" - fake_app_config = SimpleNamespace(models=[SimpleNamespace(name="stub-model")]) + 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)), + ) client = DeerFlowClient.__new__(DeerFlowClient) client._app_config = fake_app_config client._extensions_config = None @@ -102,6 +111,7 @@ def test_stream_injects_langfuse_metadata_when_enabled(monkeypatch): metadata = config.get("metadata") or {} assert metadata.get("langfuse_session_id") == "thread-client-1" assert metadata.get("langfuse_trace_name") == "lead-agent" + assert metadata.get(DEERFLOW_TRACE_METADATA_KEY) # Default no-auth context falls back to ``"default"`` user. assert metadata.get("langfuse_user_id") in {"default", "test-user-autouse"} callbacks = config.get("callbacks") or [] @@ -144,16 +154,154 @@ def test_stream_preserves_caller_metadata_overrides(monkeypatch): def patched_get_runnable_config(self, thread_id, **overrides): cfg = original_get_config(self, thread_id, **overrides) cfg["metadata"] = { + DEERFLOW_TRACE_METADATA_KEY: "explicit-client-trace", "langfuse_session_id": "explicit-session-override", "langfuse_user_id": "explicit-user", } return cfg monkeypatch.setattr(DeerFlowClient, "_get_runnable_config", patched_get_runnable_config) - list(client.stream("hi", thread_id="thread-client-3")) + with request_trace_context("client-trace-3"): + list(client.stream("hi", thread_id="thread-client-3")) metadata = captured["config"].get("metadata") or {} assert metadata["langfuse_session_id"] == "explicit-session-override" assert metadata["langfuse_user_id"] == "explicit-user" + assert metadata[DEERFLOW_TRACE_METADATA_KEY] == "explicit-client-trace" # ``trace_name`` was not supplied by caller so the worker still fills it. 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. + """ + monkeypatch.setenv("LANGFUSE_TRACING", "true") + monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-lf-test") + monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-lf-test") + from deerflow.config.tracing_config import reset_tracing_config + + reset_tracing_config() + monkeypatch.setattr("deerflow.client.build_tracing_callbacks", lambda: []) + + fake_agent = _FakeAgent() + captured = _stub_agent_creation(monkeypatch, fake_agent) + client = _make_client(monkeypatch, enhance_enabled=False) + + list(client.stream("hi", thread_id="thread-client-disabled")) + + 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 + + +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.""" + monkeypatch.setenv("LANGFUSE_TRACING", "true") + monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-lf-test") + monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-lf-test") + from deerflow.config.tracing_config import reset_tracing_config + + reset_tracing_config() + monkeypatch.setattr("deerflow.client.build_tracing_callbacks", lambda: []) + + fake_agent = _FakeAgent() + captured = _stub_agent_creation(monkeypatch, fake_agent) + client = _make_client(monkeypatch, enhance_enabled=False) + + with request_trace_context("caller-opt-in"): + list(client.stream("hi", thread_id="thread-client-opt-in")) + + metadata = captured["config"].get("metadata") or {} + assert metadata.get(DEERFLOW_TRACE_METADATA_KEY) == "caller-opt-in" + + +def test_stream_does_not_leak_trace_id_to_caller_context_between_yields(monkeypatch): + """Enable branch must bind the trace id only around each ``next()`` step + and reset it before yielding. ``stream()`` is a sync generator, which + shares the caller's context, so a ``with ensure_trace_context(): yield + from ...`` would leak the stream's id into the caller's context between + iterations — any caller code that read ``get_current_trace_id()`` (a + log filter, a follow-up ``inject_langfuse_metadata`` for unrelated + work) would pick up this stream's id instead of the caller's own trace + state. Per-step set/reset keeps the caller's context clean at every + yield boundary. + """ + monkeypatch.setattr("deerflow.client.build_tracing_callbacks", lambda: []) + + class _TwoEventAgent: + def __init__(self) -> None: + self.checkpointer = None + self.store = None + + def stream(self, state, *, config, context, stream_mode): + yield ("values", {"messages": [], "artifacts": []}) + yield ("values", {"messages": [], "artifacts": []}) + + _stub_agent_creation(monkeypatch, _TwoEventAgent()) + client = _make_client(monkeypatch, enhance_enabled=True) + + from deerflow.trace_context import get_current_trace_id + + # Caller's context starts with no trace id bound. + assert get_current_trace_id() is None + + observations: list[str | None] = [] + for _event in client.stream("hi", thread_id="thread-no-leak"): + observations.append(get_current_trace_id()) + + # Between every yield the caller sees their own (unbound) trace state, + # never the stream's minted id. + assert observations, "expected at least one event" + assert all(obs is None for obs in observations), observations + # After iteration completes, still no leak. + assert get_current_trace_id() is None + + +def test_stream_abandoned_generator_close_does_not_raise_cross_context(monkeypatch): + """Closing a partially-iterated stream from a different ``Context`` must + not raise ``ValueError: was created in a different Context``. + Sync generators share the caller's context on set/reset; a ``with`` + block spanning ``yield from`` would create a Token in the caller's + Context on the first ``next()`` and only release it via ``__exit__`` on + ``close()`` — GC-driven finalization on a different asyncio Task (or, + as simulated here, inside a ``copy_context()`` fork) would then blow up + with a cross-context reset. Per-step set/reset never leaves a Token + outstanding across yield boundaries. + """ + monkeypatch.setattr("deerflow.client.build_tracing_callbacks", lambda: []) + + class _InfiniteAgent: + def __init__(self) -> None: + self.checkpointer = None + self.store = None + + def stream(self, state, *, config, context, stream_mode): + while True: + yield ("values", {"messages": [], "artifacts": []}) + + _stub_agent_creation(monkeypatch, _InfiniteAgent()) + client = _make_client(monkeypatch, enhance_enabled=True) + + gen = client.stream("hi", thread_id="thread-cross-ctx") + # Pull one event in the current Context — a buggy implementation would + # bind a Token here that only this Context could reset. + next(gen) + + import contextvars + + isolated_ctx = contextvars.copy_context() + # Invoke ``gen.close()`` inside a distinct Context; the outer Context's + # Tokens (if any) cannot be reset from here. Reaching this line without + # a ``ValueError`` is the assertion. + isolated_ctx.run(gen.close) diff --git a/backend/tests/test_logging_config.py b/backend/tests/test_logging_config.py new file mode 100644 index 000000000..48ee1df0c --- /dev/null +++ b/backend/tests/test_logging_config.py @@ -0,0 +1,40 @@ +import io +import logging +from types import SimpleNamespace + +from deerflow.logging_config import TraceContextFilter, configure_logging +from deerflow.trace_context import request_trace_context + + +def test_trace_context_filter_injects_current_trace_id() -> None: + record = logging.LogRecord("deerflow.test", logging.INFO, __file__, 1, "hello", (), None) + + with request_trace_context("trace-log-1"): + assert TraceContextFilter().filter(record) is True + + assert record.trace_id == "trace-log-1" + + +def test_configure_logging_enhanced_text_includes_trace_id() -> None: + root = logging.getLogger() + old_handlers = root.handlers[:] + old_level = root.level + stream = io.StringIO() + handler = logging.StreamHandler(stream) + + try: + root.handlers = [handler] + root.setLevel(logging.INFO) + config = SimpleNamespace( + log_level="info", + logging=SimpleNamespace(enhance=SimpleNamespace(enabled=True, format="text")), + ) + configure_logging(config) + + with request_trace_context("trace-log-2"): + logging.getLogger("deerflow.test").info("hello") + + assert "[trace_id=trace-log-2]" in stream.getvalue() + finally: + root.handlers = old_handlers + root.setLevel(old_level) diff --git a/backend/tests/test_memory_queue.py b/backend/tests/test_memory_queue.py index 3d62f0497..421d28fc7 100644 --- a/backend/tests/test_memory_queue.py +++ b/backend/tests/test_memory_queue.py @@ -4,6 +4,7 @@ from unittest.mock import MagicMock, call, patch from deerflow.agents.memory.queue import ConversationContext, MemoryUpdateQueue from deerflow.config.memory_config import MemoryConfig +from deerflow.trace_context import get_current_trace_id, request_trace_context def _memory_config(**overrides: object) -> MemoryConfig: @@ -51,6 +52,7 @@ def test_process_queue_forwards_correction_flag_to_updater() -> None: correction_detected=True, reinforcement_detected=False, user_id=None, + deerflow_trace_id=None, ) @@ -92,6 +94,7 @@ def test_process_queue_forwards_reinforcement_flag_to_updater() -> None: correction_detected=False, reinforcement_detected=True, user_id=None, + deerflow_trace_id=None, ) @@ -235,6 +238,7 @@ def test_process_queue_updates_different_agents_in_same_thread_separately() -> N correction_detected=False, reinforcement_detected=False, user_id=None, + deerflow_trace_id=None, ), call( messages=["agent-b"], @@ -243,6 +247,158 @@ def test_process_queue_updates_different_agents_in_same_thread_separately() -> N correction_detected=False, reinforcement_detected=False, user_id=None, + deerflow_trace_id=None, ), ] ) + + +def test_process_queue_forwards_deerflow_trace_id_to_updater() -> None: + queue = MemoryUpdateQueue() + queue._queue = [ + ConversationContext( + thread_id="thread-1", + messages=["conversation"], + agent_name="lead_agent", + deerflow_trace_id="trace-memory-1", + ) + ] + mock_updater = MagicMock() + mock_updater.update_memory.return_value = True + + with patch("deerflow.agents.memory.updater.MemoryUpdater", return_value=mock_updater): + queue._process_queue() + + mock_updater.update_memory.assert_called_once_with( + messages=["conversation"], + thread_id="thread-1", + agent_name="lead_agent", + correction_detected=False, + reinforcement_detected=False, + user_id=None, + deerflow_trace_id="trace-memory-1", + ) + + +class TestProcessQueueBindsTraceContextVar: + """Regression: ``_process_queue`` runs in a Timer thread where the request + trace ContextVar is unbound. The per-context iteration must bind + ``ConversationContext.deerflow_trace_id`` into the ContextVar so + ``TraceContextFilter`` (which only reads the ContextVar) attaches the correct + ``trace_id`` to log records emitted from ``queue.py`` itself (``"Updating + memory for thread ..."``, ``"Memory updated successfully..."``, exception + logs) — not just from the deep memory-updater stack. + """ + + @staticmethod + def _run_process_queue_in_fresh_thread(queue: MemoryUpdateQueue, mock_updater: MagicMock) -> None: + def _target() -> None: + with patch("deerflow.agents.memory.updater.MemoryUpdater", return_value=mock_updater): + queue._process_queue() + + thread = threading.Thread(target=_target) + thread.start() + thread.join() + + def test_process_queue_binds_deerflow_trace_id_during_iteration(self) -> None: + queue = MemoryUpdateQueue() + queue._queue = [ + ConversationContext( + thread_id="thread-1", + messages=["conversation"], + agent_name="lead_agent", + deerflow_trace_id="trace-queue-abc", + ) + ] + captured: list[str | None] = [] + mock_updater = MagicMock() + + def _capture(**_kwargs) -> bool: + captured.append(get_current_trace_id()) + return True + + mock_updater.update_memory.side_effect = _capture + + self._run_process_queue_in_fresh_thread(queue, mock_updater) + + assert captured == ["trace-queue-abc"] + + def test_process_queue_binds_distinct_ids_per_context(self) -> None: + """Each queued context must be scoped independently — a per-iteration bind, + not a batch-level one — so id A's logs don't bleed into id B's iteration.""" + queue = MemoryUpdateQueue() + queue._queue = [ + ConversationContext( + thread_id="thread-1", + messages=["conv-a"], + agent_name="agent-a", + deerflow_trace_id="trace-a", + ), + ConversationContext( + thread_id="thread-2", + messages=["conv-b"], + agent_name="agent-b", + deerflow_trace_id="trace-b", + ), + ] + captured: list[str | None] = [] + mock_updater = MagicMock() + + def _capture(**_kwargs) -> bool: + captured.append(get_current_trace_id()) + return True + + mock_updater.update_memory.side_effect = _capture + + with ( + patch("deerflow.agents.memory.updater.MemoryUpdater", return_value=mock_updater), + patch("deerflow.agents.memory.queue.time.sleep"), + ): + queue._process_queue() + + assert captured == ["trace-a", "trace-b"] + + def test_process_queue_leaves_contextvar_unbound_when_no_trace_id(self) -> None: + """A queued context without ``deerflow_trace_id`` must not fabricate one; + the ContextVar stays unbound and log records fall through to '-'.""" + queue = MemoryUpdateQueue() + queue._queue = [ + ConversationContext( + thread_id="thread-1", + messages=["conversation"], + agent_name="lead_agent", + deerflow_trace_id=None, + ) + ] + captured: list[str | None] = [] + mock_updater = MagicMock() + + def _capture(**_kwargs) -> bool: + captured.append(get_current_trace_id()) + return True + + mock_updater.update_memory.side_effect = _capture + + self._run_process_queue_in_fresh_thread(queue, mock_updater) + + assert captured == [None] + + def test_process_queue_restores_outer_contextvar_after_return(self) -> None: + queue = MemoryUpdateQueue() + queue._queue = [ + ConversationContext( + thread_id="thread-1", + messages=["conversation"], + agent_name="lead_agent", + deerflow_trace_id="trace-inner", + ) + ] + mock_updater = MagicMock() + mock_updater.update_memory.return_value = True + + with ( + request_trace_context("trace-outer"), + patch("deerflow.agents.memory.updater.MemoryUpdater", return_value=mock_updater), + ): + queue._process_queue() + assert get_current_trace_id() == "trace-outer" diff --git a/backend/tests/test_memory_updater.py b/backend/tests/test_memory_updater.py index 9752de551..047a3a4b6 100644 --- a/backend/tests/test_memory_updater.py +++ b/backend/tests/test_memory_updater.py @@ -1,4 +1,5 @@ import asyncio +import threading from unittest.mock import AsyncMock, MagicMock, patch from deerflow.agents.memory.prompt import format_conversation_for_update @@ -12,6 +13,7 @@ from deerflow.agents.memory.updater import ( update_memory_fact, ) from deerflow.config.memory_config import MemoryConfig +from deerflow.trace_context import get_current_trace_id, request_trace_context def _make_memory(facts: list[dict[str, object]] | None = None) -> dict[str, object]: @@ -1200,3 +1202,151 @@ class TestUserIdForwarding: mock_load.assert_called_once_with(None, user_id="user-99") save_call = mock_storage.save.call_args assert save_call.kwargs.get("user_id") == "user-99" or (len(save_call.args) > 2 and save_call.args[2] == "user-99") + + def test_sync_update_injects_deerflow_trace_metadata_when_langfuse_enabled(self, monkeypatch): + monkeypatch.setenv("LANGFUSE_TRACING", "true") + monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-lf-test") + monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-lf-test") + from deerflow.config.tracing_config import reset_tracing_config + + reset_tracing_config() + updater = MemoryUpdater(model_name="memory-model") + valid_json = '{"user": {}, "history": {}, "newFacts": [], "factsToRemove": []}' + model = self._make_mock_model(valid_json) + mock_storage = MagicMock() + mock_storage.save = MagicMock(return_value=True) + + try: + with ( + patch.object(updater, "_get_model", return_value=model), + patch("deerflow.agents.memory.updater.get_memory_config", return_value=_memory_config(enabled=True)), + patch("deerflow.agents.memory.updater.get_memory_data", return_value=_make_memory()), + patch("deerflow.agents.memory.updater.get_memory_storage", return_value=mock_storage), + ): + msg = MagicMock() + msg.type = "human" + msg.content = "Hello" + ai_msg = MagicMock() + ai_msg.type = "ai" + ai_msg.content = "Hi" + ai_msg.tool_calls = [] + result = updater.update_memory([msg, ai_msg], thread_id="thread-memory", user_id="user-42", deerflow_trace_id="memory-trace-1") + finally: + reset_tracing_config() + + assert result is True + invoke_config = model.invoke.call_args.kwargs["config"] + metadata = invoke_config["metadata"] + assert metadata["deerflow_trace_id"] == "memory-trace-1" + assert metadata["langfuse_session_id"] == "thread-memory" + assert metadata["langfuse_user_id"] == "user-42" + assert metadata["langfuse_trace_name"] == "memory_agent" + + +class TestSyncUpdateBindsTraceContextVar: + """Regression: _do_update_memory_sync must bind ``deerflow_trace_id`` into the + request-trace ContextVar for the duration of the update. + + The memory pipeline plumbs ``deerflow_trace_id`` through ``ConversationContext`` + precisely because ContextVar does not propagate to ``threading.Timer`` threads + or ``ThreadPoolExecutor.submit(...)`` workers. Langfuse metadata is already + correct because it takes an explicit function argument, but the enhanced-log + ``TraceContextFilter`` only reads the ContextVar — so without this bind, every + log record emitted from the Timer/Executor path (model-error logs, tracing + callback logs) shows ``trace_id=-`` despite the correct id being available. + """ + + @staticmethod + def _make_updater_with_capturing_model(captured: list[str | None]) -> tuple[MemoryUpdater, MagicMock]: + updater = MemoryUpdater() + + def _capture_and_respond(*_args, **_kwargs): + captured.append(get_current_trace_id()) + response = MagicMock() + response.content = '{"user": {}, "history": {}, "newFacts": [], "factsToRemove": []}' + return response + + model = MagicMock() + model.invoke = MagicMock(side_effect=_capture_and_respond) + return updater, model + + @staticmethod + def _run_sync_update_in_fresh_thread(updater: MemoryUpdater, model: MagicMock, *, deerflow_trace_id: str | None) -> bool: + """Run ``_do_update_memory_sync`` in a bare ``threading.Thread`` to guarantee + no ContextVar inheritance from the pytest main thread (mirrors the Timer / + Executor worker execution model).""" + results: list[bool] = [] + + def _target() -> None: + with ( + patch.object(updater, "_get_model", return_value=model), + patch("deerflow.agents.memory.updater.get_memory_config", return_value=_memory_config(enabled=True)), + patch("deerflow.agents.memory.updater.get_memory_data", return_value=_make_memory()), + patch("deerflow.agents.memory.updater.get_memory_storage", return_value=MagicMock(save=MagicMock(return_value=True))), + ): + msg = MagicMock() + msg.type = "human" + msg.content = "Hello" + ai_msg = MagicMock() + ai_msg.type = "ai" + ai_msg.content = "Hi" + results.append( + updater._do_update_memory_sync( + messages=[msg, ai_msg], + deerflow_trace_id=deerflow_trace_id, + ) + ) + + thread = threading.Thread(target=_target) + thread.start() + thread.join() + return results[0] + + def test_binds_deerflow_trace_id_into_contextvar(self) -> None: + captured: list[str | None] = [] + updater, model = self._make_updater_with_capturing_model(captured) + + result = self._run_sync_update_in_fresh_thread(updater, model, deerflow_trace_id="trace-mem-xyz") + + assert result is True + assert captured == ["trace-mem-xyz"] + + def test_none_trace_id_does_not_fabricate_id(self) -> None: + """When no trace_id is provided the ContextVar must stay unbound — + fabricating a fresh id would produce log records with a bogus 'correlated' + id that has no relationship to any real request.""" + captured: list[str | None] = [] + updater, model = self._make_updater_with_capturing_model(captured) + + result = self._run_sync_update_in_fresh_thread(updater, model, deerflow_trace_id=None) + + assert result is True + assert captured == [None] + + def test_restores_outer_contextvar_after_return(self) -> None: + """The binding must be scoped to the function; a pre-existing outer trace + id in the caller's context must be intact after the call returns.""" + captured: list[str | None] = [] + updater, model = self._make_updater_with_capturing_model(captured) + + with ( + request_trace_context("outer-trace"), + patch.object(updater, "_get_model", return_value=model), + patch("deerflow.agents.memory.updater.get_memory_config", return_value=_memory_config(enabled=True)), + patch("deerflow.agents.memory.updater.get_memory_data", return_value=_make_memory()), + patch("deerflow.agents.memory.updater.get_memory_storage", return_value=MagicMock(save=MagicMock(return_value=True))), + ): + msg = MagicMock() + msg.type = "human" + msg.content = "Hello" + ai_msg = MagicMock() + ai_msg.type = "ai" + ai_msg.content = "Hi" + + updater._do_update_memory_sync( + messages=[msg, ai_msg], + deerflow_trace_id="inner-trace", + ) + + assert captured == ["inner-trace"] + assert get_current_trace_id() == "outer-trace" diff --git a/backend/tests/test_reload_boundary.py b/backend/tests/test_reload_boundary.py index daa33b0cc..639ec39f5 100644 --- a/backend/tests/test_reload_boundary.py +++ b/backend/tests/test_reload_boundary.py @@ -85,6 +85,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.", "database": "memory, sqlite, or postgres", "sandbox": "Sandbox provider", "run_events": "memory for dev", diff --git a/backend/tests/test_subagent_executor.py b/backend/tests/test_subagent_executor.py index 2c01d0abe..31e04d3a3 100644 --- a/backend/tests/test_subagent_executor.py +++ b/backend/tests/test_subagent_executor.py @@ -2109,7 +2109,7 @@ class TestSubagentTracingWiring: yield reset_tracing_config() - def _make_executor(self, classes, *, user_id=None, name="general-purpose", parent_model="test-model"): + def _make_executor(self, classes, *, user_id=None, name="general-purpose", parent_model="test-model", deerflow_trace_id=None): SubagentExecutor = classes["SubagentExecutor"] SubagentConfig = classes["SubagentConfig"] config = SubagentConfig( @@ -2126,6 +2126,7 @@ class TestSubagentTracingWiring: thread_id="thread-trace-1", trace_id="trace-1", user_id=user_id, + deerflow_trace_id=deerflow_trace_id, ) @pytest.mark.anyio @@ -2183,7 +2184,7 @@ class TestSubagentTracingWiring: sentinel = _Sentinel() monkeypatch.setattr(executor_module, "build_tracing_callbacks", lambda: [sentinel]) - executor = self._make_executor(classes, user_id="alice", name="general_purpose") + executor = self._make_executor(classes, user_id="alice", name="general_purpose", deerflow_trace_id="gateway-trace-sub") fake_agent = _FakeStreamAgent() monkeypatch.setattr(executor, "_build_initial_state", self._noop_build_initial_state) monkeypatch.setattr(executor, "_create_agent", lambda *a, **kw: fake_agent) @@ -2196,6 +2197,8 @@ class TestSubagentTracingWiring: # Underscores are normalized to hyphens so the trace name matches the # lead-agent naming shape. assert metadata.get("langfuse_trace_name") == "subagent:general-purpose" + assert metadata.get("deerflow_trace_id") == "gateway-trace-sub" + assert fake_agent.captured_context.get("deerflow_trace_id") == "gateway-trace-sub" tags = metadata.get("langfuse_tags") or [] assert any(t.startswith("model:") for t in tags), "model tag must be emitted for cost attribution" diff --git a/backend/tests/test_suggestions_router.py b/backend/tests/test_suggestions_router.py index 6a47d6f8c..dcb6bbb8f 100644 --- a/backend/tests/test_suggestions_router.py +++ b/backend/tests/test_suggestions_router.py @@ -2,7 +2,21 @@ import asyncio from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock +import pytest + from app.gateway.routers import suggestions +from deerflow.trace_context import request_trace_context + + +@pytest.fixture(autouse=True) +def _clear_langfuse_env(monkeypatch): + from deerflow.config.tracing_config import reset_tracing_config + + for name in ("LANGFUSE_TRACING", "LANGFUSE_PUBLIC_KEY", "LANGFUSE_SECRET_KEY", "LANGFUSE_BASE_URL", "DEER_FLOW_ENV", "ENVIRONMENT"): + monkeypatch.delenv(name, raising=False) + reset_tracing_config() + yield + reset_tracing_config() def test_strip_markdown_code_fence_removes_wrapping(): @@ -110,6 +124,38 @@ def test_generate_suggestions_parses_and_limits(monkeypatch): assert fake_model.ainvoke.await_args.kwargs["config"] == {"run_name": "suggest_agent"} +def test_generate_suggestions_injects_deerflow_trace_metadata_when_langfuse_enabled(monkeypatch): + monkeypatch.setenv("LANGFUSE_TRACING", "true") + monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-lf-test") + monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-lf-test") + from deerflow.config.tracing_config import reset_tracing_config + + reset_tracing_config() + req = suggestions.SuggestionsRequest( + messages=[ + suggestions.SuggestionMessage(role="user", content="Hi"), + suggestions.SuggestionMessage(role="assistant", content="Hello"), + ], + n=1, + model_name="suggest-model", + ) + fake_model = MagicMock() + fake_model.ainvoke = AsyncMock(return_value=MagicMock(content='["Q1"]')) + monkeypatch.setattr(suggestions, "create_chat_model", lambda **kwargs: fake_model) + + try: + with request_trace_context("suggest-trace-1"): + result = asyncio.run(suggestions.generate_suggestions.__wrapped__("thread-suggest", req, request=None, config=SimpleNamespace(suggestions=SimpleNamespace(enabled=True)))) + finally: + reset_tracing_config() + + assert result.suggestions == ["Q1"] + metadata = fake_model.ainvoke.await_args.kwargs["config"]["metadata"] + assert metadata["deerflow_trace_id"] == "suggest-trace-1" + assert metadata["langfuse_session_id"] == "thread-suggest" + assert metadata["langfuse_trace_name"] == "suggest_agent" + + def test_generate_suggestions_parses_list_block_content(monkeypatch): req = suggestions.SuggestionsRequest( messages=[ diff --git a/backend/tests/test_task_tool_core_logic.py b/backend/tests/test_task_tool_core_logic.py index dc0f844d3..4fc60e804 100644 --- a/backend/tests/test_task_tool_core_logic.py +++ b/backend/tests/test_task_tool_core_logic.py @@ -183,6 +183,7 @@ def test_task_tool_threads_runtime_app_config_to_subagent_dependencies(monkeypat def test_task_tool_emits_running_and_completed_events(monkeypatch): config = _make_subagent_config() runtime = _make_runtime() + runtime.context["deerflow_trace_id"] = "task-trace-1" events = [] captured = {} get_available_tools = MagicMock(return_value=["tool-a", "tool-b"]) @@ -231,6 +232,7 @@ def test_task_tool_emits_running_and_completed_events(monkeypatch): assert captured["task_id"] == "tc-123" assert captured["executor_kwargs"]["thread_id"] == "thread-1" assert captured["executor_kwargs"]["parent_model"] == "ark-model" + assert captured["executor_kwargs"]["deerflow_trace_id"] == "task-trace-1" assert captured["executor_kwargs"]["config"].max_turns == config.max_turns # Skills are no longer appended to system_prompt; they are loaded per-session # by SubagentExecutor and injected as conversation items (Codex pattern). diff --git a/backend/tests/test_trace_context.py b/backend/tests/test_trace_context.py new file mode 100644 index 000000000..9244e841f --- /dev/null +++ b/backend/tests/test_trace_context.py @@ -0,0 +1,86 @@ +"""Unit tests for ``deerflow.trace_context`` validation helpers. + +The middleware-level end-to-end coverage lives in ``test_trace_middleware.py``; +this file pins the character-set invariants of ``normalize_trace_id`` directly +so that a future relaxation of the check trips a targeted failure. +""" + +from __future__ import annotations + +import pytest + +from deerflow.trace_context import _MAX_TRACE_ID_LENGTH, normalize_trace_id + + +class TestNormalizeTraceIdAcceptsPrintableAscii: + def test_accepts_uuid_hex(self) -> None: + assert normalize_trace_id("0123456789abcdef0123456789abcdef") == "0123456789abcdef0123456789abcdef" + + def test_accepts_alphanumerics_and_punctuation(self) -> None: + assert normalize_trace_id("abc-123_XYZ.foo:bar/baz") == "abc-123_XYZ.foo:bar/baz" + + def test_strips_surrounding_whitespace(self) -> None: + assert normalize_trace_id(" trace-1 ") == "trace-1" + + def test_accepts_boundary_low(self) -> None: + assert normalize_trace_id("\x20abc") == "abc" + + def test_accepts_boundary_high(self) -> None: + assert normalize_trace_id("abc\x7e") == "abc\x7e" + + def test_accepts_maximum_length(self) -> None: + value = "a" * _MAX_TRACE_ID_LENGTH + assert normalize_trace_id(value) == value + + +class TestNormalizeTraceIdRejectsUnsafeInput: + def test_rejects_non_string(self) -> None: + assert normalize_trace_id(None) is None + assert normalize_trace_id(12345) is None + assert normalize_trace_id(b"abc") is None + + def test_rejects_empty_and_whitespace_only(self) -> None: + assert normalize_trace_id("") is None + assert normalize_trace_id(" \t ") is None + + def test_rejects_over_max_length(self) -> None: + assert normalize_trace_id("a" * (_MAX_TRACE_ID_LENGTH + 1)) is None + + @pytest.mark.parametrize( + "value", + [ + "trace\x00id", # NUL + "trace\x1fid", # last C0 control + "trace\tid", # embedded tab + "trace\nid", # LF — the classic log-injection / CRLF pivot + "trace\rid", # CR + ], + ) + def test_rejects_c0_controls(self, value: str) -> None: + assert normalize_trace_id(value) is None + + def test_rejects_del(self) -> None: + assert normalize_trace_id("trace\x7fid") is None + + @pytest.mark.parametrize("value", ["trace\x80id", "trace\x9fid"]) + def test_rejects_c1_controls_in_latin1_range(self, value: str) -> None: + """C1 controls latin-1-encode successfully but are stripped or + rejected by hardened intermediaries (nginx / envoy / cloudfront), + silently breaking the response. Reject at validation time.""" + assert normalize_trace_id(value) is None + + def test_rejects_latin1_supplement_characters(self) -> None: + assert normalize_trace_id("caf\xe9") is None # é = 0xE9 + + def test_rejects_cjk_characters(self) -> None: + """Codepoints > 0xFF raise UnicodeEncodeError inside + ``MutableHeaders.__setitem__`` before ``send`` is dispatched, forcing + a 500 on any endpoint. This is the exact case from the review.""" + assert normalize_trace_id("请求-1") is None + assert normalize_trace_id("トレース") is None + + def test_rejects_emoji(self) -> None: + assert normalize_trace_id("trace-\U0001f680") is None # 🚀 + + def test_rejects_surrogate_pair_pieces(self) -> None: + assert normalize_trace_id("trace-\ud83d") is None diff --git a/backend/tests/test_trace_middleware.py b/backend/tests/test_trace_middleware.py new file mode 100644 index 000000000..cca3a9a84 --- /dev/null +++ b/backend/tests/test_trace_middleware.py @@ -0,0 +1,180 @@ +from types import SimpleNamespace + +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 + + +def _make_app(*, enabled: bool) -> FastAPI: + app = FastAPI() + app.add_middleware(TraceMiddleware, enabled=enabled) + + @app.get("/plain") + async def plain() -> dict[str, str | None]: + return {"trace_id": get_current_trace_id()} + + @app.get("/stream") + async def stream() -> StreamingResponse: + async def body(): + yield f"trace={get_current_trace_id()}".encode() + + return StreamingResponse(body(), media_type="text/plain") + + @app.get("/pre-set") + async def pre_set() -> Response: + return Response("ok", headers={TRACE_ID_HEADER: "downstream"}) + + return app + + +def test_trace_header_absent_when_disabled() -> None: + client = TestClient(_make_app(enabled=False)) + + response = client.get("/plain") + + assert TRACE_ID_HEADER not in response.headers + assert response.json() == {"trace_id": None} + + +def test_trace_header_inherits_inbound_value_and_binds_context() -> None: + client = TestClient(_make_app(enabled=True)) + + response = client.get("/plain", headers={TRACE_ID_HEADER: "trace-from-upstream"}) + + assert response.headers[TRACE_ID_HEADER] == "trace-from-upstream" + assert response.json() == {"trace_id": "trace-from-upstream"} + + +def test_trace_header_generated_when_missing() -> None: + client = TestClient(_make_app(enabled=True)) + + response = client.get("/plain") + + trace_id = response.headers[TRACE_ID_HEADER] + assert trace_id + assert response.json() == {"trace_id": trace_id} + + +def test_trace_header_added_to_streaming_response_without_consuming_body() -> None: + client = TestClient(_make_app(enabled=True)) + + response = client.get("/stream", headers={TRACE_ID_HEADER: "stream-trace"}) + + assert response.headers[TRACE_ID_HEADER] == "stream-trace" + assert response.text == "trace=stream-trace" + + +def test_trace_header_overwrites_duplicate_downstream_value() -> None: + client = TestClient(_make_app(enabled=True)) + + response = client.get("/pre-set", headers={TRACE_ID_HEADER: "canonical-trace"}) + + assert response.headers[TRACE_ID_HEADER] == "canonical-trace" + assert response.headers.get_list(TRACE_ID_HEADER) == ["canonical-trace"] + + +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 + values either forced a 500 via ``UnicodeEncodeError`` inside + ``MutableHeaders.__setitem__`` (codepoints > 0xFF, e.g. UTF-8 CJK bytes + latin-1-decoded to high codepoints) or silently broke the response at + hardened intermediaries (nginx / envoy / cloudfront) for the 0x80-0xFF + range. The middleware must fall back to a freshly generated ASCII id. + + ``httpx`` refuses to ascii-encode non-ASCII string header values on the + client side, so we pass the header as raw bytes to mirror what an + 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)) + + # Raw UTF-8 bytes of "café-1"; Starlette latin-1-decodes them into + # a string containing 0xC3, 0xA9 — both > 0x7E. + crafted_bytes = b"caf\xc3\xa9-1" + crafted_decoded = crafted_bytes.decode("latin-1") + response = client.get("/plain", headers={TRACE_ID_HEADER: crafted_bytes}) + + assert response.status_code == 200 + returned = response.headers[TRACE_ID_HEADER] + assert returned != crafted_decoded + assert all(0x20 <= ord(ch) <= 0x7E for ch in returned), returned + assert response.json() == {"trace_id": returned} + + +def test_trace_header_rejects_crafted_c1_control_and_generates_fresh_id() -> None: + """C1 controls (0x80-0x9F) latin-1-encode successfully but are stripped + 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)) + + crafted_bytes = b"trace\x9fid" + crafted_decoded = crafted_bytes.decode("latin-1") + response = client.get("/plain", headers={TRACE_ID_HEADER: crafted_bytes}) + + assert response.status_code == 200 + returned = response.headers[TRACE_ID_HEADER] + assert returned != crafted_decoded + 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"]) + + @app.get("/plain") + async def plain() -> dict[str, str | None]: + return {"trace_id": get_current_trace_id()} + + 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 + + +def test_resolve_trace_enabled_walks_nested_config() -> None: + config = SimpleNamespace(logging=SimpleNamespace(enhance=SimpleNamespace(enabled=True))) + assert resolve_trace_enabled(config) is True + + config_off = SimpleNamespace(logging=SimpleNamespace(enhance=SimpleNamespace(enabled=False))) + assert resolve_trace_enabled(config_off) is False + + +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_gateway_app_construction_trace_flag_defaults_false_when_config_missing(monkeypatch) -> None: + import app.gateway.app as gateway_app + + def missing_config(): + raise FileNotFoundError("no config") + + monkeypatch.setattr(gateway_app, "get_app_config", missing_config) + + assert gateway_app._resolve_trace_enabled_for_app_construction() is False + + +def test_gateway_app_construction_trace_flag_uses_config_snapshot(monkeypatch) -> None: + import app.gateway.app as gateway_app + + 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 diff --git a/backend/tests/test_tracing_metadata.py b/backend/tests/test_tracing_metadata.py index 6c758e40d..981014d5b 100644 --- a/backend/tests/test_tracing_metadata.py +++ b/backend/tests/test_tracing_metadata.py @@ -4,6 +4,7 @@ from __future__ import annotations import pytest +from deerflow.trace_context import request_trace_context from deerflow.tracing import metadata as tracing_metadata @@ -135,3 +136,28 @@ def test_thread_id_none_still_produces_metadata(monkeypatch): assert result["langfuse_session_id"] is None assert result["langfuse_user_id"] == "u-1" + + +def test_deerflow_trace_id_comes_from_current_trace_context(monkeypatch): + _enable_langfuse(monkeypatch) + + with request_trace_context("gateway-trace-1"): + result = tracing_metadata.build_langfuse_trace_metadata( + thread_id="thread-abc", + user_id="user-42", + ) + + assert result["deerflow_trace_id"] == "gateway-trace-1" + + +def test_deerflow_trace_id_explicit_argument_wins(monkeypatch): + _enable_langfuse(monkeypatch) + + with request_trace_context("ambient-trace"): + result = tracing_metadata.build_langfuse_trace_metadata( + thread_id="thread-abc", + user_id="user-42", + deerflow_trace_id="explicit-trace", + ) + + assert result["deerflow_trace_id"] == "explicit-trace" diff --git a/backend/tests/test_worker_langfuse_metadata.py b/backend/tests/test_worker_langfuse_metadata.py index 63a2edf95..558e09585 100644 --- a/backend/tests/test_worker_langfuse_metadata.py +++ b/backend/tests/test_worker_langfuse_metadata.py @@ -14,6 +14,7 @@ import pytest from deerflow.runtime.runs.manager import RunRecord 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, request_trace_context class _FakeAgent: @@ -105,15 +106,16 @@ async def test_run_agent_injects_langfuse_metadata(monkeypatch): record.abort_event = asyncio.Event() ctx = RunContext(checkpointer=None) - await run_agent( - _FakeBridge(), - _FakeRunManager(), - record, - ctx=ctx, - agent_factory=agent_factory, - graph_input={"messages": []}, - config={"configurable": {"thread_id": "thread-xyz"}}, - ) + with request_trace_context("gateway-trace-1"): + await run_agent( + _FakeBridge(), + _FakeRunManager(), + record, + ctx=ctx, + agent_factory=agent_factory, + graph_input={"messages": []}, + config={"configurable": {"thread_id": "thread-xyz"}}, + ) assert fake_agent.captured_config is not None, "astream was not invoked" metadata = fake_agent.captured_config.get("metadata") or {} @@ -123,6 +125,8 @@ async def test_run_agent_injects_langfuse_metadata(monkeypatch): user_id = metadata.get("langfuse_user_id") assert user_id == "test-user-autouse", f"expected test-user-autouse, got {user_id}" assert metadata.get("langfuse_trace_name") == "lead-agent" + assert metadata.get(DEERFLOW_TRACE_METADATA_KEY) == "gateway-trace-1" + assert fake_agent.captured_config.get("context", {}).get(DEERFLOW_TRACE_METADATA_KEY) == "gateway-trace-1" tags = metadata.get("langfuse_tags") or [] assert "model:gpt-4o" in tags @@ -210,6 +214,7 @@ async def test_run_agent_preserves_caller_metadata_overrides(monkeypatch): config={ "configurable": {"thread_id": "thread-default"}, "metadata": { + DEERFLOW_TRACE_METADATA_KEY: "explicit-deerflow-trace", "langfuse_session_id": "custom-session-id", "langfuse_user_id": "explicit-user", }, @@ -220,6 +225,8 @@ 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" # Worker still fills in keys that the caller didn't set. assert metadata["langfuse_trace_name"] == "lead-agent" diff --git a/config.example.yaml b/config.example.yaml index a4bf3127c..ffedd6d7c 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -15,7 +15,7 @@ # ============================================================================ # Bump this number when the config schema changes. # Run `make config-upgrade` to merge new fields into your local config.yaml. -config_version: 16 +config_version: 17 # ============================================================================ # Logging @@ -23,6 +23,13 @@ config_version: 16 # 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. +logging: + enhance: + enabled: false + format: text + # ============================================================================ # Token Usage # ============================================================================