From 6f61401a91d671f07190c4fa5bbe10d061c01ed3 Mon Sep 17 00:00:00 2001 From: Yunlu Wen Date: Tue, 22 Sep 2026 04:35:33 +0000 Subject: [PATCH] feat(agent): export agent trajectory with a separated tracer (#42651) --- dify-agent/.example.env | 1 + dify-agent/docs/dify-agent/guide/index.md | 37 + .../execution-context-layer/index.md | 25 + .../src/dify_agent/runtime/agent_factory.py | 15 +- .../src/dify_agent/runtime/observability.py | 290 +++++++ .../src/dify_agent/runtime/run_scheduler.py | 5 + dify-agent/src/dify_agent/runtime/runner.py | 12 + dify-agent/src/dify_agent/server/app.py | 18 +- .../src/dify_agent/server/observability.py | 188 ++++- dify-agent/src/dify_agent/server/settings.py | 63 +- .../agent_stub/server/test_agent_stub_app.py | 14 + .../runtime/test_agent_observability.py | 723 ++++++++++++++++++ .../dify_agent/runtime/test_run_scheduler.py | 31 + .../local/dify_agent/runtime/test_runner.py | 61 ++ .../runtime_backend/test_profile.py | 7 +- .../tests/local/dify_agent/server/test_app.py | 100 +++ .../dify_agent/server/test_observability.py | 461 ++++++++++- .../server/test_observability_isolation.py | 285 +++++++ .../local/dify_agent/server/test_settings.py | 268 +++++++ .../envs/core-services/dify-agent.env.example | 13 + 20 files changed, 2570 insertions(+), 47 deletions(-) create mode 100644 dify-agent/src/dify_agent/runtime/observability.py create mode 100644 dify-agent/tests/local/dify_agent/runtime/test_agent_observability.py create mode 100644 dify-agent/tests/local/dify_agent/server/test_observability_isolation.py diff --git a/dify-agent/.example.env b/dify-agent/.example.env index c54af515e45..00afe2698aa 100644 --- a/dify-agent/.example.env +++ b/dify-agent/.example.env @@ -107,3 +107,4 @@ DIFY_AGENT_OUTBOUND_HTTP_KEEPALIVE_EXPIRY=30 DIFY_AGENT_LOCAL_SANDBOX_MATERIALIZED_HOME_ROOT=/home/dify DIFY_AGENT_LOCAL_SANDBOX_WORKSPACE_ROOT=/workspace DIFY_AGENT_LOCAL_SANDBOX_HOME_SNAPSHOT_ROOT=/home/dify/.snapshots +LOGFIRE_CONSOLE=false diff --git a/dify-agent/docs/dify-agent/guide/index.md b/dify-agent/docs/dify-agent/guide/index.md index 9fb5fe0eb96..6be5b103f4b 100644 --- a/dify-agent/docs/dify-agent/guide/index.md +++ b/dify-agent/docs/dify-agent/guide/index.md @@ -81,6 +81,43 @@ in [E2B runtime metering](runtime-metering.md). | `DIFY_AGENT_OUTBOUND_HTTP_MAX_CONNECTIONS` | `100` | Maximum total shared outbound HTTP connections. | | `DIFY_AGENT_OUTBOUND_HTTP_MAX_KEEPALIVE_CONNECTIONS` | `20` | Maximum idle shared outbound HTTP connections. | | `DIFY_AGENT_OUTBOUND_HTTP_KEEPALIVE_EXPIRY` | `30` | Idle keep-alive expiry in seconds. | +| `DIFY_AGENT_TRAJECTORY_ENABLED` | `false` | Opt-in switch for the separate Agent trajectory export pipeline. | +| `DIFY_AGENT_TRAJECTORY_OTLP_TRACES_ENDPOINT` | empty | Full HTTP(S) OTLP traces endpoint required when trajectory export is enabled. | +| `DIFY_AGENT_TRAJECTORY_OTLP_HEADERS` | `{}` | JSON object of request headers sent with Agent OTLP trace exports. | +| `DIFY_AGENT_TRAJECTORY_SERVICE_NAME` | `dify-agent-trajectory` | Resource service name attached to Agent trajectory spans. | +| `DIFY_AGENT_TRAJECTORY_INCLUDE_CONTENT` | `false` | Include message and tool content in Agent trajectory spans. | +| `DIFY_AGENT_TRAJECTORY_TRACE_CONTEXT_MODE` | `isolated` | `isolated` starts independent platform/Agent traces; `shared` preserves cross-instance parent context. Requires process restart to change. | +| `DIFY_AGENT_TRAJECTORY_MAX_QUEUE_SIZE` | `2048` | Maximum number of spans queued for Agent trace export; must be positive. | +| `DIFY_AGENT_TRAJECTORY_MAX_EXPORT_BATCH_SIZE` | `512` | Maximum spans per export batch; must be positive and no greater than the queue size. | +| `DIFY_AGENT_TRAJECTORY_SCHEDULE_DELAY_MS` | `5000` | Delay between scheduled Agent span exports, in milliseconds; must be positive. | +| `DIFY_AGENT_TRAJECTORY_EXPORT_TIMEOUT_MS` | `5000` | Batch processor export timeout setting, in milliseconds; must be positive. Does not change the OTLP HTTP exporter request timeout. | + +Platform observability uses the process-level Logfire instance and standard +`OTEL_*` / `LOGFIRE_*` configuration. Agent observability uses a separate local +Logfire instance and is disabled by default. Set `DIFY_AGENT_TRAJECTORY_ENABLED=true` +and a full `DIFY_AGENT_TRAJECTORY_OTLP_TRACES_ENDPOINT` to opt in; +`DIFY_AGENT_TRAJECTORY_OTLP_HEADERS` is a JSON object of request headers. Agent +traces do not inherit platform endpoints, credentials, sampling, or console +export. Message/tool content requires the separate +`DIFY_AGENT_TRAJECTORY_INCLUDE_CONTENT` opt-in. Platform FastAPI parameter/error +payloads, HTTPX bodies/headers and Redis statements are not captured by these +instrumentations. The Agent instance is created at service startup and shut down +after active runs finish. These are deployment-level settings, not yet +per-tenant authorization or routing; a shared backend does not provide storage +isolation. + +With `DIFY_AGENT_TRAJECTORY_TRACE_CONTEXT_MODE=isolated` (the default), Agent +run spans start independent traces instead of inheriting platform request +parents. Platform HTTPX/Redis instrumentation also detaches local Agent +parents while preserving platform-only and incoming distributed trace +relationships. In `shared` mode both pipelines retain cross-instance parent +context and the Agent sampler respects the parent sampling decision. Use +`shared` only when both pipelines export to the same queryable backend space; +otherwise missing-root spans can recur. Service names do not select the mode. +The two SDK instances, exporters, enable switch, and content opt-in remain +separate. The mode is fixed at process initialization and requires a restart +to change. No extra shared root span is added; incomplete traces may still +appear temporarily while a run or export is in progress. Example `.env`: diff --git a/dify-agent/docs/dify-agent/user-manual/execution-context-layer/index.md b/dify-agent/docs/dify-agent/user-manual/execution-context-layer/index.md index e73fd3c19bb..582d7fac155 100644 --- a/dify-agent/docs/dify-agent/user-manual/execution-context-layer/index.md +++ b/dify-agent/docs/dify-agent/user-manual/execution-context-layer/index.md @@ -18,6 +18,31 @@ on this layer to reach the plugin daemon. | `invoke_from` | `Literal[...]` | Dify caller category recorded for observability and correlation. | | `app_id` / `workflow_id` / `workflow_run_id` / `node_id` / `node_execution_id` / `conversation_id` / `agent_id` / `agent_config_version_id` / `trace_id` | `str \| None` | Optional Dify-owned execution identifiers forwarded with the run. | +## Trajectory span attributes + +When Agent trajectory export is enabled, this layer's fields are attached to every +trajectory span of the run. The names follow the Dify Enterprise Data Push +dictionary so a trajectory span joins the platform's own workflow and node signals +on the same keys. Fields you leave unset are omitted, so an Agent App turn carries +`dify.conversation.id` while a workflow Agent node carries the workflow keys. + +| Config field | Span attribute | +| --- | --- | +| `tenant_id` | `dify.tenant_id` | +| `app_id` | `dify.app_id` | +| `agent_id` | `dify.agent_id` | +| `user_id` | `gen_ai.user.id` | +| `invoke_from` | `dify.invoke_from` | +| `conversation_id` | `dify.conversation.id` | +| `workflow_id` | `dify.workflow.id` | +| `workflow_run_id` | `dify.workflow.run_id` | +| `node_id` | `dify.node.id` | +| `node_execution_id` | `dify.node.execution_id` | +| `trace_id` | `dify.trace_id` | + +`app_id` is the app the run was invoked through, so a workflow Agent node reports +the workflow's own app rather than the Agent's. + The execution-context layer type id is `dify.execution_context`. ## Basic usage diff --git a/dify-agent/src/dify_agent/runtime/agent_factory.py b/dify-agent/src/dify_agent/runtime/agent_factory.py index e82f246d866..0161d4159fb 100644 --- a/dify-agent/src/dify_agent/runtime/agent_factory.py +++ b/dify-agent/src/dify_agent/runtime/agent_factory.py @@ -11,7 +11,7 @@ optional JSON Schema output layer share the same ``Agent`` construction path. """ from collections.abc import Sequence -from typing import Any, cast +from typing import Any, Final, cast from pydantic_ai import Agent from pydantic_ai.messages import UserContent @@ -21,6 +21,9 @@ from pydantic_ai.output import OutputSpec from agenton.layers.types import PydanticAITool +DIFY_AGENT_RUN_NAME: Final[str] = "dify-agent" + + def create_agent( model: Model[Any], *, @@ -35,7 +38,13 @@ def create_agent( carries the Pydantic hooks needed for schema exposure and runtime validation, so agent construction does not need to register a separate validator. """ - return cast(Agent[None, object], Agent(model, output_type=output_type, tools=tools)) + # TODO: propagate roster name, so gen_ai.agent.name can pick that up + agent = cast( + Agent[None, object], + Agent(model, name=DIFY_AGENT_RUN_NAME, output_type=output_type, tools=tools), + ) + agent.instrument = False + return agent def normalize_user_input(user_prompts: Sequence[UserContent]) -> str | Sequence[UserContent]: @@ -45,4 +54,4 @@ def normalize_user_input(user_prompts: Sequence[UserContent]) -> str | Sequence[ return list(user_prompts) -__all__ = ["create_agent", "normalize_user_input"] +__all__ = ["DIFY_AGENT_RUN_NAME", "create_agent", "normalize_user_input"] diff --git a/dify-agent/src/dify_agent/runtime/observability.py b/dify-agent/src/dify_agent/runtime/observability.py new file mode 100644 index 00000000000..8933dea279c --- /dev/null +++ b/dify-agent/src/dify_agent/runtime/observability.py @@ -0,0 +1,290 @@ +from __future__ import annotations + +import asyncio +from collections.abc import Sequence +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any, Final, Literal + +from opentelemetry.context import Context +from opentelemetry.sdk.trace import ReadableSpan +from opentelemetry.trace import ( + INVALID_SPAN, + Link, + Span, + SpanKind, + Tracer, + TracerProvider, + get_current_span, + set_span_in_context, +) +from opentelemetry.util.types import Attributes + +from dify_agent.layers.execution_context.configs import DifyExecutionContextLayerConfig + +if TYPE_CHECKING: + from logfire import Logfire + from pydantic_ai import Agent + +DIFY_TRACE_ID_ATTRIBUTE: Final[str] = "dify.trace_id" +DIFY_TENANT_ID_ATTRIBUTE: Final[str] = "dify.tenant_id" +DIFY_APP_ID_ATTRIBUTE: Final[str] = "dify.app_id" +DIFY_AGENT_ID_ATTRIBUTE: Final[str] = "dify.agent_id" +DIFY_INVOKE_FROM_ATTRIBUTE: Final[str] = "dify.invoke_from" +DIFY_CONVERSATION_ID_ATTRIBUTE: Final[str] = "dify.conversation.id" +DIFY_WORKFLOW_ID_ATTRIBUTE: Final[str] = "dify.workflow.id" +DIFY_WORKFLOW_RUN_ID_ATTRIBUTE: Final[str] = "dify.workflow.run_id" +DIFY_NODE_ID_ATTRIBUTE: Final[str] = "dify.node.id" +DIFY_NODE_EXECUTION_ID_ATTRIBUTE: Final[str] = "dify.node.execution_id" +GEN_AI_USER_ID_ATTRIBUTE: Final[str] = "gen_ai.user.id" + + +def dify_run_attributes(execution_context: DifyExecutionContextLayerConfig) -> tuple[tuple[str, str], ...]: + return tuple( + (key, value) + for key, value in ( + (DIFY_TRACE_ID_ATTRIBUTE, execution_context.trace_id), + (DIFY_TENANT_ID_ATTRIBUTE, execution_context.tenant_id), + (DIFY_APP_ID_ATTRIBUTE, execution_context.app_id), + (DIFY_AGENT_ID_ATTRIBUTE, execution_context.agent_id), + (GEN_AI_USER_ID_ATTRIBUTE, execution_context.user_id), + (DIFY_INVOKE_FROM_ATTRIBUTE, execution_context.invoke_from), + (DIFY_CONVERSATION_ID_ATTRIBUTE, execution_context.conversation_id), + (DIFY_WORKFLOW_ID_ATTRIBUTE, execution_context.workflow_id), + (DIFY_WORKFLOW_RUN_ID_ATTRIBUTE, execution_context.workflow_run_id), + (DIFY_NODE_ID_ATTRIBUTE, execution_context.node_id), + (DIFY_NODE_EXECUTION_ID_ATTRIBUTE, execution_context.node_execution_id), + ) + if value + ) + + +@dataclass(frozen=True) +class IsolatedTracerProvider(TracerProvider): + client: Logfire + preserve_external_parent: bool = False + + def get_tracer( + self, + instrumenting_module_name: str, + instrumenting_library_version: str | None = None, + schema_url: str | None = None, + attributes: Attributes = None, + ) -> Tracer: + delegate = self.client.config.get_tracer_provider().get_tracer( + instrumenting_module_name, + instrumenting_library_version, + schema_url=schema_url, + attributes=attributes, + ) + return _IsolatedTracer(delegate, self.client, self.preserve_external_parent) + + +@dataclass(frozen=True) +class _IsolatedTracer(Tracer): + delegate: Tracer + client: Logfire + preserve_external_parent: bool + + def _parent_context(self, context: Context | None) -> Context | None: + parent = get_current_span(context) + if not parent.get_span_context().is_valid: + return context + if isinstance(parent, ReadableSpan): + instance_id = self.client.resource_attributes.get("service.instance.id") + if instance_id is not None and parent.resource.attributes.get("service.instance.id") == instance_id: + return context + elif self.preserve_external_parent: + return context + return set_span_in_context(INVALID_SPAN, context) + + def start_span( + self, + name: str, + context: Context | None = None, + kind: SpanKind = SpanKind.INTERNAL, + attributes: Attributes = None, + links: Sequence[Link] | None = None, + start_time: int | None = None, + record_exception: bool = True, + set_status_on_exception: bool = True, + ) -> Span: + return self.delegate.start_span( + name, + context=self._parent_context(context), + kind=kind, + attributes=attributes, + links=links, + start_time=start_time, + record_exception=record_exception, + set_status_on_exception=set_status_on_exception, + ) + + def start_as_current_span( + self, + name: str, + context: Context | None = None, + kind: SpanKind = SpanKind.INTERNAL, + attributes: Attributes = None, + links: Sequence[Link] | None = None, + start_time: int | None = None, + record_exception: bool = True, + set_status_on_exception: bool = True, + end_on_exit: bool = True, + ): + return self.delegate.start_as_current_span( + name, + context=self._parent_context(context), + kind=kind, + attributes=attributes, + links=links, + start_time=start_time, + record_exception=record_exception, + set_status_on_exception=set_status_on_exception, + end_on_exit=end_on_exit, + ) + + +@dataclass(frozen=True) +class RunScopedTracerProvider(TracerProvider): + """Stamp one run's fixed attributes on every span a delegate provider creates.""" + + delegate: TracerProvider + attributes: tuple[tuple[str, str], ...] + + def get_tracer( + self, + instrumenting_module_name: str, + instrumenting_library_version: str | None = None, + schema_url: str | None = None, + attributes: Attributes = None, + ) -> Tracer: + delegate = self.delegate.get_tracer( + instrumenting_module_name, + instrumenting_library_version, + schema_url=schema_url, + attributes=attributes, + ) + return _RunScopedTracer(delegate, self.attributes) + + +@dataclass(frozen=True) +class _RunScopedTracer(Tracer): + delegate: Tracer + attributes: tuple[tuple[str, str], ...] + + def _run_attributes(self, attributes: Attributes) -> Attributes: + return {**(attributes or {}), **dict(self.attributes)} + + def start_span( + self, + name: str, + context: Context | None = None, + kind: SpanKind = SpanKind.INTERNAL, + attributes: Attributes = None, + links: Sequence[Link] | None = None, + start_time: int | None = None, + record_exception: bool = True, + set_status_on_exception: bool = True, + ) -> Span: + return self.delegate.start_span( + name, + context=context, + kind=kind, + attributes=self._run_attributes(attributes), + links=links, + start_time=start_time, + record_exception=record_exception, + set_status_on_exception=set_status_on_exception, + ) + + def start_as_current_span( + self, + name: str, + context: Context | None = None, + kind: SpanKind = SpanKind.INTERNAL, + attributes: Attributes = None, + links: Sequence[Link] | None = None, + start_time: int | None = None, + record_exception: bool = True, + set_status_on_exception: bool = True, + end_on_exit: bool = True, + ): + return self.delegate.start_as_current_span( + name, + context=context, + kind=kind, + attributes=self._run_attributes(attributes), + links=links, + start_time=start_time, + record_exception=record_exception, + set_status_on_exception=set_status_on_exception, + end_on_exit=end_on_exit, + ) + + +@dataclass(frozen=True, slots=True) +class AgentObservability: + client: Logfire + include_content: bool = False + trace_context_mode: Literal["isolated", "shared"] = "isolated" + + def instrument( + self, + agent: Agent[Any, Any], + *, + execution_context: DifyExecutionContextLayerConfig | None = None, + ) -> None: + """Instrument one run's agent and attribute its spans to its Dify context. + + ``execution_context`` is the run's Dify identity and correlation context. + Its Data Push-aligned attributes are attached to every span the + instrumented agent produces; see ``dify_run_attributes``. Runs without an + execution context are still instrumented, just without Dify attribution. + """ + self.client.instrument_pydantic_ai( + agent, + include_content=self.include_content, + include_binary_content=False, + tracer_provider=self._tracer_provider(execution_context), + ) + + def _tracer_provider(self, execution_context: DifyExecutionContextLayerConfig | None) -> TracerProvider: + provider: TracerProvider = ( + IsolatedTracerProvider(self.client) + if self.trace_context_mode == "isolated" + else self.client.config.get_tracer_provider() + ) + if execution_context is None: + return provider + run_attributes = dify_run_attributes(execution_context) + if not run_attributes: + return provider + return RunScopedTracerProvider(provider, run_attributes) + + async def aclose(self) -> None: + """Flush and tear down this instance's own export pipeline. + + The owning scope closes the instance; shutdown is blocking, so it runs off + the event loop. This only affects the Agent pipeline and leaves the + platform Logfire instance running. + """ + await asyncio.to_thread(self.client.shutdown, timeout_millis=5000) + + +__all__ = [ + "DIFY_AGENT_ID_ATTRIBUTE", + "DIFY_APP_ID_ATTRIBUTE", + "DIFY_CONVERSATION_ID_ATTRIBUTE", + "DIFY_INVOKE_FROM_ATTRIBUTE", + "DIFY_NODE_EXECUTION_ID_ATTRIBUTE", + "DIFY_NODE_ID_ATTRIBUTE", + "DIFY_TENANT_ID_ATTRIBUTE", + "DIFY_TRACE_ID_ATTRIBUTE", + "DIFY_WORKFLOW_ID_ATTRIBUTE", + "DIFY_WORKFLOW_RUN_ID_ATTRIBUTE", + "GEN_AI_USER_ID_ATTRIBUTE", + "AgentObservability", + "IsolatedTracerProvider", + "RunScopedTracerProvider", + "dify_run_attributes", +] diff --git a/dify-agent/src/dify_agent/runtime/run_scheduler.py b/dify-agent/src/dify_agent/runtime/run_scheduler.py index 5a57999547f..db3592b3cd1 100644 --- a/dify-agent/src/dify_agent/runtime/run_scheduler.py +++ b/dify-agent/src/dify_agent/runtime/run_scheduler.py @@ -29,6 +29,7 @@ from dify_agent.runtime.event_coalescer import ( DEFAULT_TEXT_DELTA_MAX_CHARS, ) from dify_agent.runtime.event_sink import RunEventSink, RunFinalizationResult, emit_run_failed +from dify_agent.runtime.observability import AgentObservability from dify_agent.runtime.runner import DEFAULT_AGENT_RUN_TIMEOUT_SECONDS, AgentRunRunner from dify_agent.server.schemas import RunRecord @@ -119,6 +120,7 @@ class RunScheduler: layer_providers: tuple[LayerProviderInput, ...] plugin_daemon_http_client: httpx.AsyncClient dify_api_http_client: httpx.AsyncClient + agent_observability: AgentObservability | None _lifecycle_lock: asyncio.Lock def __init__( @@ -134,6 +136,7 @@ class RunScheduler: stream_text_delta_max_chars: int = DEFAULT_TEXT_DELTA_MAX_CHARS, layer_providers: tuple[LayerProviderInput, ...] | None = None, runner_factory: RunRunnerFactory | None = None, + agent_observability: AgentObservability | None = None, ) -> None: self.store = store self.shutdown_grace_seconds = shutdown_grace_seconds @@ -147,6 +150,7 @@ class RunScheduler: self.dify_api_http_client = dify_api_http_client self.layer_providers = layer_providers if layer_providers is not None else create_default_layer_providers() self.runner_factory = runner_factory + self.agent_observability = agent_observability self._lifecycle_lock = asyncio.Lock() async def create_run(self, request: CreateRunRequest) -> RunRecord: @@ -320,6 +324,7 @@ class RunScheduler: stream_text_delta_coalescing_enabled=self.stream_text_delta_coalescing_enabled, stream_text_delta_flush_interval_seconds=self.stream_text_delta_flush_interval_seconds, stream_text_delta_max_chars=self.stream_text_delta_max_chars, + agent_observability=self.agent_observability, ) def _discard_active_run(self, run_id: str) -> None: diff --git a/dify-agent/src/dify_agent/runtime/runner.py b/dify-agent/src/dify_agent/runtime/runner.py index 4590cd721bd..71e367d0b33 100644 --- a/dify-agent/src/dify_agent/runtime/runner.py +++ b/dify-agent/src/dify_agent/runtime/runner.py @@ -82,6 +82,7 @@ from dify_agent.runtime.event_sink import ( emit_run_started, emit_run_succeeded, ) +from dify_agent.runtime.observability import AgentObservability from dify_agent.runtime.history import ( get_history_layer, replace_run_history, @@ -198,6 +199,7 @@ class AgentRunRunner: stream_text_delta_coalescing_enabled: bool stream_text_delta_flush_interval_seconds: float stream_text_delta_max_chars: int + agent_observability: AgentObservability | None _terminal_session_snapshot: CompositorSessionSnapshot | None _terminal_usage: AgentRunUsage | None @@ -215,6 +217,7 @@ class AgentRunRunner: stream_text_delta_coalescing_enabled: bool = True, stream_text_delta_flush_interval_seconds: float = DEFAULT_TEXT_DELTA_FLUSH_INTERVAL_SECONDS, stream_text_delta_max_chars: int = DEFAULT_TEXT_DELTA_MAX_CHARS, + agent_observability: AgentObservability | None = None, ) -> None: if stream_text_delta_flush_interval_seconds <= 0: raise ValueError("stream_text_delta_flush_interval_seconds must be positive") @@ -231,6 +234,7 @@ class AgentRunRunner: self.stream_text_delta_coalescing_enabled = stream_text_delta_coalescing_enabled self.stream_text_delta_flush_interval_seconds = stream_text_delta_flush_interval_seconds self.stream_text_delta_max_chars = stream_text_delta_max_chars + self.agent_observability = agent_observability self._terminal_session_snapshot = None self._terminal_usage = None @@ -390,6 +394,14 @@ class AgentRunRunner: tools=tools, output_type=_resolve_agent_output_type(output_contract.output_type, ask_human_layer is not None), ) + if self.agent_observability is not None: + # The model layer's required execution-context dependency is the + # run's only carrier of Dify identity, and its node name is + # caller-chosen, so read it through the typed dependency. + self.agent_observability.instrument( + agent, + execution_context=llm_layer.deps.execution_context.config, + ) run_timeout = asyncio.timeout(self.run_timeout_seconds) try: with capture_run_messages() as captured_messages: diff --git a/dify-agent/src/dify_agent/server/app.py b/dify-agent/src/dify_agent/server/app.py index cd5f1d650b0..36219c9c6e5 100644 --- a/dify-agent/src/dify_agent/server/app.py +++ b/dify-agent/src/dify_agent/server/app.py @@ -11,9 +11,11 @@ adds no reporting to business operations. Redis is not used as a run job queue. stay state-only: they borrow the lifespan-owned clients through the runner and receive runtime-backend and Shell settings through provider construction rather than reading environment variables themselves. The standard server mounts the -HTTP Agent Stub router. Process-level Logfire instrumentation is configured at -app construction time and only exports remotely when Logfire's default -environment configuration provides a token. +HTTP Agent Stub router. Process-level platform Logfire instrumentation is +configured at app construction time. Logfire-platform export is token-gated; +standard OTLP export follows the SDK environment configuration, including +server dotenv files. Agent trajectory export is a separate opt-in Logfire +instance owned by the app lifespan and injected into each run's runner. """ from collections.abc import AsyncGenerator @@ -30,7 +32,7 @@ from dify_agent.layers.execution_context import DifyExecutionContextLayerConfig from dify_agent.runtime.compositor_factory import create_default_layer_providers from dify_agent.runtime.run_scheduler import RunScheduler from dify_agent.server.auth import create_bearer_token_dependency -from dify_agent.server.observability import configure_server_observability +from dify_agent.server.observability import configure_agent_observability, configure_server_observability from dify_agent.server.routes.e2b_usage import create_e2b_usage_router from dify_agent.server.routes.runs import create_runs_router from dify_agent.server.routes.execution_bindings import create_execution_bindings_router @@ -102,6 +104,8 @@ def create_app(settings: ServerSettings | None = None) -> FastAPI: @asynccontextmanager async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: + agent_observability = configure_agent_observability(resolved_settings) + _app.state.agent_observability = agent_observability redis = Redis.from_url(resolved_settings.redis_url) plugin_daemon_http_client = create_plugin_daemon_http_client(resolved_settings) dify_api_inner_http_client = create_dify_api_inner_http_client(resolved_settings) @@ -121,6 +125,7 @@ def create_app(settings: ServerSettings | None = None) -> FastAPI: stream_text_delta_flush_interval_seconds=(resolved_settings.stream_text_delta_flush_interval_ms / 1000), stream_text_delta_max_chars=resolved_settings.stream_text_delta_max_chars, layer_providers=layer_providers, + agent_observability=agent_observability, ) state["store"] = store state["scheduler"] = scheduler @@ -131,9 +136,12 @@ def create_app(settings: ServerSettings | None = None) -> FastAPI: await dify_api_inner_http_client.aclose() await plugin_daemon_http_client.aclose() await redis.aclose() + # Closed last so runs draining above still reach the exporter. + if agent_observability is not None: + await agent_observability.aclose() app = FastAPI(title="Dify Agent Run Server", version="0.1.0", lifespan=lifespan) - configure_server_observability(app) + app.state.platform_observability = configure_server_observability(app, settings=resolved_settings) def get_store() -> RedisRunStore: return cast(RedisRunStore, state["store"]) diff --git a/dify-agent/src/dify_agent/server/observability.py b/dify-agent/src/dify_agent/server/observability.py index 7893dfe4769..01fecfd0cee 100644 --- a/dify-agent/src/dify_agent/server/observability.py +++ b/dify-agent/src/dify_agent/server/observability.py @@ -1,47 +1,181 @@ """Process-level Logfire setup for the Dify Agent run server. The run server performs observability setup at the FastAPI app boundary rather -than inside agent runtime code. Global instrumentations cover shared HTTPX, -Redis, and Pydantic AI clients once per process; the FastAPI instrumentation is -applied per app instance because tests and embedded callers can build multiple -apps in one Python process. Remote export remains token-gated through Logfire's -``if-token-present`` mode and Logfire's default environment-variable handling, -so development without a token only writes Logfire's console output locally. +than inside agent runtime code. Global instrumentations cover shared HTTPX and +Redis clients once per process; the FastAPI instrumentation is applied per app +instance because tests and embedded callers can build multiple apps in one +Python process. ``OTEL_*`` and ``LOGFIRE_*`` values captured in the +``ServerSettings`` dotenv snapshot are forwarded before SDK setup without +overriding the process environment. Logfire-platform export remains token-gated through +``if-token-present``; standard OTLP export is configured independently. + +The optional Agent trajectory pipeline is a separate, deployment-opt-in Logfire +instance configured only from ``DIFY_AGENT_TRAJECTORY_*`` settings. It is +created while the inherited ``OTEL_*``/``LOGFIRE_*`` environment is temporarily +stripped so platform endpoints, credentials, and sampling never leak into the +Agent pipeline; the isolation is startup-only and does not coordinate external +code that mutates ``os.environ`` concurrently. """ from __future__ import annotations +import os +from collections.abc import Iterator +from contextlib import contextmanager +from threading import RLock +from typing import Any, Literal + import logfire -from fastapi import FastAPI +from fastapi import FastAPI, Request, WebSocket +from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter +from opentelemetry.sdk.trace.export import BatchSpanProcessor +from opentelemetry.sdk.trace.sampling import ALWAYS_ON, ParentBased + +from dify_agent.runtime.observability import AgentObservability, IsolatedTracerProvider +from dify_agent.server.settings import ServerSettings _global_instrumentation_ready = False +_global_trace_context_mode: Literal["isolated", "shared"] | None = None +_observability_configuration_lock = RLock() -def configure_server_observability(app: FastAPI) -> None: +@contextmanager +def _isolated_agent_environment() -> Iterator[None]: + """Serialize SDK configuration and strip inherited SDK env for Agent setup.""" + with _observability_configuration_lock: + inherited = {key: value for key, value in os.environ.items() if key.startswith(("OTEL_", "LOGFIRE_"))} + for key in inherited: + del os.environ[key] + try: + yield + finally: + for key in tuple(os.environ): + if key.startswith(("OTEL_", "LOGFIRE_")): + del os.environ[key] + os.environ.update(inherited) + + +def _platform_request_attributes(_request: Request | WebSocket, _attributes: dict[str, Any]) -> None: + return None + + +def configure_server_observability(app: FastAPI, *, settings: ServerSettings | None = None) -> logfire.Logfire: """Configure Logfire and instrument the server's framework/client boundaries. - Instrumentation calls intentionally use Logfire's defaults instead of - re-exposing capture options through Dify settings. The only Dify-owned - policy here is that remote export stays token-gated while local console - output still works without a token. + Platform instrumentation captures infrastructure metadata without parsed + request payloads or HTTP bodies. SDK environment variables + are forwarded from the settings dotenv snapshot with process-environment + priority. + Logfire-platform export stays token-gated; OTLP endpoints do not require a + Logfire token. FastAPI parsed argument/error payloads, HTTPX bodies and + headers, and Redis statements are not captured. The trace context mode is a + startup-only choice and cannot change after instrumentation. Returns the + platform Logfire instance. """ global _global_instrumentation_ready + global _global_trace_context_mode - logfire.configure( - send_to_logfire="if-token-present", - inspect_arguments=False, + resolved_settings = settings or ServerSettings() + mode = resolved_settings.trajectory_trace_context_mode + + with _observability_configuration_lock: + if _global_instrumentation_ready and _global_trace_context_mode != mode: + raise ValueError("Trace context mode cannot change after platform instrumentation; restart the process") + for key, value in resolved_settings.observability_dotenv.items(): + if key.startswith(("OTEL_", "LOGFIRE_")): + os.environ.setdefault(key, value.get_secret_value()) + + platform = logfire.configure( + send_to_logfire="if-token-present", + inspect_arguments=False, + ) + + tracer_provider: IsolatedTracerProvider | Any = ( + IsolatedTracerProvider(platform, preserve_external_parent=True) + if mode == "isolated" + else platform.config.get_tracer_provider() + ) + if not _global_instrumentation_ready: + platform.instrument_httpx( + capture_all=False, + capture_headers=False, + capture_request_body=False, + capture_response_body=False, + tracer_provider=tracer_provider, + ) + platform.instrument_redis(capture_statement=False, tracer_provider=tracer_provider) + _global_instrumentation_ready = True + _global_trace_context_mode = mode + + if getattr(app.state, "dify_agent_logfire_instrumented", False): + return platform + platform.instrument_fastapi( + app, + request_attributes_mapper=_platform_request_attributes, + capture_headers=False, + tracer_provider=tracer_provider, + ) + app.state.dify_agent_logfire_instrumented = True + return platform + + +def configure_agent_observability(settings: ServerSettings) -> AgentObservability | None: + """Build the opt-in Agent trajectory pipeline from deployment settings only. + + This is one process-wide pipeline and is meant to stay that way. Per-tenant + destinations belong in a routing ``SpanExporter`` that reads a span's + ``dify.tenant_id`` and fans out to a cached per-tenant exporter, not in a + tracer instance per tenant: a second instance costs another + ``logfire.configure()``, which means another + ``_isolated_agent_environment()`` window on a live process and another batch + processor thread. Routing keeps setup static and makes a tenant destination + plain data. Such a router must fail closed on an unknown tenant rather than + fall back to this deployment-wide endpoint. + """ + if not settings.trajectory_enabled: + return None + endpoint = settings.trajectory_otlp_traces_endpoint + if endpoint is None: + raise ValueError("trajectory_otlp_traces_endpoint is required when trajectory_enabled is true") + with _isolated_agent_environment(): + exporter = OTLPSpanExporter( + endpoint=str(endpoint), + headers={key: value.get_secret_value() for key, value in settings.trajectory_otlp_headers.items()}, + timeout=5, + ) + processor = BatchSpanProcessor( + exporter, + max_queue_size=settings.trajectory_max_queue_size, + max_export_batch_size=settings.trajectory_max_export_batch_size, + schedule_delay_millis=settings.trajectory_schedule_delay_ms, + export_timeout_millis=settings.trajectory_export_timeout_ms, + ) + try: + client = logfire.configure( + local=True, + send_to_logfire=False, + service_name=settings.trajectory_service_name, + console=False, + metrics=False, + inspect_arguments=False, + add_baggage_to_attributes=False, + sampling=logfire.SamplingOptions( + head=ALWAYS_ON if settings.trajectory_trace_context_mode == "isolated" else ParentBased(ALWAYS_ON) + ), + additional_span_processors=[processor], + ) + except BaseException: + processor.shutdown() + raise + return AgentObservability( + client=client, + include_content=settings.trajectory_include_content, + trace_context_mode=settings.trajectory_trace_context_mode, ) - if not _global_instrumentation_ready: - logfire.instrument_httpx() - logfire.instrument_redis() - logfire.instrument_pydantic_ai() - _global_instrumentation_ready = True - if getattr(app.state, "dify_agent_logfire_instrumented", False): - return - logfire.instrument_fastapi(app) - app.state.dify_agent_logfire_instrumented = True - - -__all__ = ["configure_server_observability"] +__all__ = [ + "AgentObservability", + "configure_agent_observability", + "configure_server_observability", +] diff --git a/dify-agent/src/dify_agent/server/settings.py b/dify-agent/src/dify_agent/server/settings.py index 2f0ab43183e..207e14b7675 100644 --- a/dify-agent/src/dify_agent/server/settings.py +++ b/dify-agent/src/dify_agent/server/settings.py @@ -14,8 +14,9 @@ import httpx from typing import ClassVar, Literal, cast -from pydantic import AliasChoices, AnyHttpUrl, Field, TypeAdapter, field_validator, model_validator +from pydantic import AliasChoices, AnyHttpUrl, Field, SecretStr, TypeAdapter, field_validator, model_validator from pydantic_settings import BaseSettings, SettingsConfigDict +from pydantic_settings.sources import DotEnvSettingsSource, InitSettingsSource, PydanticBaseSettingsSource from dify_agent.agent_stub.protocol.agent_stub import normalize_agent_stub_api_base_url from dify_agent.agent_stub.server.agent_stub_config import DifyApiAgentStubConfigRequestHandler @@ -45,7 +46,12 @@ DEFAULT_RUN_EVENT_STREAM_MAX_LENGTH = 5000 class ServerSettings(BaseSettings): - """Environment settings for scheduling, outbound HTTP, and runtime resources.""" + """Environment settings for scheduling, outbound HTTP, and runtime resources. + + ``observability_dotenv`` is a construction-time snapshot of the effective + dotenv source restricted to ``OTEL_*``/``LOGFIRE_*`` keys; it is excluded + from serialization and repr, and is consumed only by observability setup. + """ redis_url: str = "redis://localhost:6379/0" redis_prefix: str = "dify-agent" @@ -124,6 +130,16 @@ class ServerSettings(BaseSettings): binding_file_download_command_timeout_seconds: float = Field(default=210.0, gt=0) server_secret_key: str | None = None api_token: str | None = None + trajectory_enabled: bool = False + trajectory_otlp_traces_endpoint: AnyHttpUrl | None = None + trajectory_otlp_headers: dict[str, SecretStr] = Field(default_factory=dict, repr=False) + trajectory_service_name: str = Field(default="dify-agent-trajectory", min_length=1) + trajectory_include_content: bool = False + trajectory_trace_context_mode: Literal["isolated", "shared"] = "isolated" + trajectory_max_queue_size: int = Field(default=2048, gt=0) + trajectory_max_export_batch_size: int = Field(default=512, gt=0) + trajectory_schedule_delay_ms: int = Field(default=5000, gt=0) + trajectory_export_timeout_ms: int = Field(default=5000, gt=0) shell_redact_patterns: str = "" outbound_http_connect_timeout: float = Field(default=10.0, ge=0) outbound_http_read_timeout: float = Field(default=600.0, ge=0) @@ -132,6 +148,7 @@ class ServerSettings(BaseSettings): outbound_http_max_connections: int = Field(default=100, ge=1) outbound_http_max_keepalive_connections: int = Field(default=20, ge=0) outbound_http_keepalive_expiry: float = Field(default=30.0, ge=0) + observability_dotenv: dict[str, SecretStr] = Field(default_factory=dict, exclude=True, repr=False) model_config: ClassVar[SettingsConfigDict] = SettingsConfigDict( env_prefix="DIFY_AGENT_", @@ -140,6 +157,29 @@ class ServerSettings(BaseSettings): populate_by_name=True, ) + @classmethod + def settings_customise_sources( + cls, + settings_cls: type[BaseSettings], + init_settings: PydanticBaseSettingsSource, + env_settings: PydanticBaseSettingsSource, + dotenv_settings: PydanticBaseSettingsSource, + file_secret_settings: PydanticBaseSettingsSource, + ) -> tuple[PydanticBaseSettingsSource, ...]: + variables: dict[str, str] = {} + if isinstance(dotenv_settings, DotEnvSettingsSource): + for key, value in dotenv_settings.env_vars.items(): + name = key if dotenv_settings.case_sensitive else key.upper() + if value is not None and name.startswith(("OTEL_", "LOGFIRE_")): + variables[name] = value + return ( + InitSettingsSource(settings_cls, {"observability_dotenv": variables}), + init_settings, + env_settings, + dotenv_settings, + file_secret_settings, + ) + @field_validator("agent_stub_api_base_url") @classmethod def normalize_agent_stub_api_base_url_value(cls, value: str | None) -> str | None: @@ -195,6 +235,16 @@ class ServerSettings(BaseSettings): raise ValueError("DIFY_AGENT_INNER_API_URL must not include a query string or fragment") return parsed + @field_validator("trajectory_otlp_traces_endpoint") + @classmethod + def validate_trajectory_otlp_traces_endpoint(cls, value: AnyHttpUrl | None) -> AnyHttpUrl | None: + """Reject embedded credentials or fragments in the Agent OTLP endpoint.""" + if value is None: + return None + if value.username is not None or value.password is not None or value.fragment is not None: + raise ValueError("DIFY_AGENT_TRAJECTORY_OTLP_TRACES_ENDPOINT must not include credentials or a fragment") + return value + @field_validator("inner_api_key", "api_token") @classmethod def normalize_optional_api_token(cls, value: str | None) -> str | None: @@ -230,6 +280,15 @@ class ServerSettings(BaseSettings): ) return self + @model_validator(mode="after") + def validate_trajectory_requirements(self) -> "ServerSettings": + """Validate the Agent endpoint and batch processor limits.""" + if self.trajectory_enabled and self.trajectory_otlp_traces_endpoint is None: + raise ValueError("trajectory_otlp_traces_endpoint is required when trajectory_enabled is true") + if self.trajectory_max_export_batch_size > self.trajectory_max_queue_size: + raise ValueError("trajectory_max_export_batch_size must not exceed trajectory_max_queue_size") + return self + def build_runtime_backend_profile(self) -> RuntimeBackendProfile | None: """Build the deployment-selected resource backend without adding service state.""" if self.runtime_backend == "local" and not self.local_sandbox_endpoint: diff --git a/dify-agent/tests/local/dify_agent/agent_stub/server/test_agent_stub_app.py b/dify-agent/tests/local/dify_agent/agent_stub/server/test_agent_stub_app.py index 1da2d399b52..4513a6f981b 100644 --- a/dify-agent/tests/local/dify_agent/agent_stub/server/test_agent_stub_app.py +++ b/dify-agent/tests/local/dify_agent/agent_stub/server/test_agent_stub_app.py @@ -2,9 +2,11 @@ from __future__ import annotations import base64 import importlib +import os import time import httpx +import pytest from fastapi.testclient import TestClient from dify_agent.agent_stub.server.app import create_agent_stub_app @@ -12,6 +14,18 @@ from dify_agent.layers.execution_context import DifyExecutionContextLayerConfig from dify_agent.server.settings import ServerSettings +@pytest.fixture(autouse=True) +def _isolated_agent_stub_settings(monkeypatch: pytest.MonkeyPatch) -> None: + # ServerSettings reads the process environment and a dotenv file resolved + # against the current directory. A developer's local dify-agent/.env can + # configure the Agent Stub, which would silently invalidate the unconfigured + # cases here, so every case states the settings it exercises. + for name in tuple(os.environ): + if name.startswith("DIFY_AGENT_"): + monkeypatch.delenv(name) + monkeypatch.setitem(ServerSettings.model_config, "env_file", None) + + def _base64url_secret(value: bytes) -> str: return base64.urlsafe_b64encode(value).rstrip(b"=").decode("ascii") diff --git a/dify-agent/tests/local/dify_agent/runtime/test_agent_observability.py b/dify-agent/tests/local/dify_agent/runtime/test_agent_observability.py new file mode 100644 index 00000000000..3a20a40e766 --- /dev/null +++ b/dify-agent/tests/local/dify_agent/runtime/test_agent_observability.py @@ -0,0 +1,723 @@ +"""Real Logfire/OTel SDK tests for the opt-in Agent observability instance.""" + +from __future__ import annotations + +import asyncio +import os +from collections.abc import Iterator +from typing import Literal, cast + +import httpx +import logfire +import pytest +from opentelemetry import context as otel_context +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter +from opentelemetry.trace import NonRecordingSpan, SpanContext, TraceFlags, get_current_span, set_span_in_context +from pydantic_ai import Agent, Tool +from pydantic_ai.models.instrumented import InstrumentationSettings +from pydantic_ai.models.test import TestModel + +import dify_agent.server.observability as server_observability +from dify_agent.layers.execution_context import DifyExecutionContextLayerConfig +from dify_agent.runtime.agent_factory import DIFY_AGENT_RUN_NAME, create_agent +from dify_agent.runtime.observability import ( + DIFY_TENANT_ID_ATTRIBUTE, + GEN_AI_USER_ID_ATTRIBUTE, + AgentObservability, + IsolatedTracerProvider, + dify_run_attributes, +) +from dify_agent.server.settings import ServerSettings + +_AGENT_RUN_SPAN_NAME = f"invoke_agent {DIFY_AGENT_RUN_NAME}" + + +@pytest.fixture(autouse=True) +def isolate_environment(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None: + monkeypatch.chdir(tmp_path) + monkeypatch.setattr(os, "environ", dict(os.environ)) + for key in tuple(os.environ): + if key.startswith(("OTEL_", "LOGFIRE_")): + monkeypatch.delenv(key) + + +@pytest.fixture(autouse=True) +def restore_agent_instrumentation(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]: + monkeypatch.setattr(Agent, "_instrument_default", False) + yield + + +def _local_client(exporter: InMemorySpanExporter) -> logfire.Logfire: + return logfire.configure( + local=True, + send_to_logfire=False, + console=False, + metrics=False, + inspect_arguments=False, + additional_span_processors=[SimpleSpanProcessor(exporter)], + ) + + +def _platform_provider(exporter: InMemorySpanExporter) -> TracerProvider: + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + return provider + + +def _static_tool() -> str: + return "tool-sentinel-output" + + +def _serialized_attributes(exporter: InMemorySpanExporter) -> str: + return repr([dict(span.attributes or {}) for span in exporter.get_finished_spans()]) + + +def test_create_agent_instrument_disabled_survives_global_instrument_all() -> None: + platform_exporter = InMemorySpanExporter() + platform_provider = _platform_provider(platform_exporter) + try: + Agent.instrument_all(InstrumentationSettings(tracer_provider=platform_provider)) + + agent = create_agent(TestModel(custom_output_text="done"), tools=[]) + assert agent.instrument is False + + _ = agent.run_sync("test-only input") + platform_provider.force_flush() + + assert not platform_exporter.get_finished_spans() + finally: + platform_provider.shutdown() + + +def test_agent_observability_runs_agent_spans_to_business_exporter_only() -> None: + platform_exporter = InMemorySpanExporter() + platform_provider = _platform_provider(platform_exporter) + business_exporter = InMemorySpanExporter() + client = _local_client(business_exporter) + try: + Agent.instrument_all(InstrumentationSettings(tracer_provider=platform_provider)) + observability = AgentObservability(client=client) + agent = create_agent(TestModel(custom_output_text="done"), tools=[Tool(_static_tool)]) + observability.instrument(agent) + + _ = agent.run_sync("test-only input") + assert client.force_flush(timeout_millis=10000) + platform_provider.force_flush() + + business_spans = business_exporter.get_finished_spans() + assert business_spans + assert any( + span.instrumentation_scope is not None and "pydantic" in span.instrumentation_scope.name + for span in business_spans + ) + span_names = [span.name for span in business_spans] + assert any("_static_tool" in name or "execute_tool" in name for name in span_names), span_names + assert not platform_exporter.get_finished_spans() + finally: + client.shutdown(timeout_millis=5000) + platform_provider.shutdown() + + +def test_agent_observability_excludes_content_by_default() -> None: + exporter = InMemorySpanExporter() + client = _local_client(exporter) + try: + observability = AgentObservability(client=client, include_content=False) + agent = create_agent(TestModel(custom_output_text="done"), tools=[Tool(_static_tool)]) + observability.instrument(agent) + + _ = agent.run_sync("content-sentinel-input-1") + assert client.force_flush(timeout_millis=10000) + + spans = exporter.get_finished_spans() + assert spans + serialized = _serialized_attributes(exporter) + assert "content-sentinel-input-1" not in serialized + assert "tool-sentinel-output" not in serialized + finally: + client.shutdown(timeout_millis=5000) + + +def test_agent_observability_includes_content_when_opted_in() -> None: + exporter = InMemorySpanExporter() + client = _local_client(exporter) + try: + observability = AgentObservability(client=client, include_content=True) + agent = create_agent(TestModel(custom_output_text="done"), tools=[Tool(_static_tool)]) + observability.instrument(agent) + + _ = agent.run_sync("content-sentinel-input-2") + assert client.force_flush(timeout_millis=10000) + + spans = exporter.get_finished_spans() + assert spans + serialized = _serialized_attributes(exporter) + assert "content-sentinel-input-2" in serialized + finally: + client.shutdown(timeout_millis=5000) + + +def test_business_agent_root_detaches_from_platform_parent_span() -> None: + platform_exporter = InMemorySpanExporter() + platform_client = _local_client(platform_exporter) + business_exporter = InMemorySpanExporter() + business_client = _local_client(business_exporter) + try: + observability = AgentObservability(client=business_client) + agent = create_agent(TestModel(custom_output_text="done"), tools=[Tool(_static_tool)]) + observability.instrument(agent) + + with platform_client.span("platform-incoming-request"): + _ = agent.run_sync("test-only input") + assert platform_client.force_flush(timeout_millis=10000) + assert business_client.force_flush(timeout_millis=10000) + + platform_root = next( + span for span in platform_exporter.get_finished_spans() if span.name == "platform-incoming-request" + ) + agent_root = next(span for span in business_exporter.get_finished_spans() if span.name == _AGENT_RUN_SPAN_NAME) + assert agent_root.parent is None + assert agent_root.context is not None + assert agent_root.context.trace_id != platform_root.context.trace_id + finally: + business_client.shutdown(timeout_millis=5000) + platform_client.shutdown(timeout_millis=5000) + + +def test_agent_shutdown_does_not_stop_platform_pipeline() -> None: + platform_exporter = InMemorySpanExporter() + platform_client = _local_client(platform_exporter) + business_exporter = InMemorySpanExporter() + business_client = _local_client(business_exporter) + try: + Agent.instrument_all(InstrumentationSettings(tracer_provider=platform_client.config.get_tracer_provider())) + observability = AgentObservability(client=business_client) + agent = create_agent(TestModel(custom_output_text="done"), tools=[]) + observability.instrument(agent) + _ = agent.run_sync("test-only input") + assert business_client.force_flush(timeout_millis=10000) + assert business_exporter.get_finished_spans() + + business_client.shutdown(timeout_millis=5000) + business_client = None + + with platform_client.span("platform-marker-after-agent-shutdown"): + pass + assert platform_client.force_flush(timeout_millis=10000) + assert any( + span.name == "platform-marker-after-agent-shutdown" for span in platform_exporter.get_finished_spans() + ) + finally: + if business_client is not None: + business_client.shutdown(timeout_millis=5000) + platform_client.shutdown(timeout_millis=5000) + + +def _spans_by_name(exporter: InMemorySpanExporter, name: str) -> list: + return [span for span in exporter.get_finished_spans() if span.name == name] + + +def _assert_parents_resolve_within(exporter: InMemorySpanExporter) -> None: + spans = exporter.get_finished_spans() + local_ids = {span.context.span_id for span in spans} + for span in spans: + if span.parent is not None: + assert span.parent.span_id in local_ids, span.name + + +def test_business_descendants_stay_in_business_trace_and_caller_context_restored() -> None: + platform_exporter = InMemorySpanExporter() + platform_client = _local_client(platform_exporter) + business_exporter = InMemorySpanExporter() + business_client = _local_client(business_exporter) + try: + observability = AgentObservability(client=business_client) + agent = create_agent(TestModel(custom_output_text="done"), tools=[Tool(_static_tool)]) + observability.instrument(agent) + + with platform_client.span("platform-incoming-request") as platform_span: + _ = agent.run_sync("test-only input") + assert get_current_span().get_span_context() == platform_span.get_span_context() + assert platform_client.force_flush(timeout_millis=10000) + assert business_client.force_flush(timeout_millis=10000) + + platform_trace_ids = {span.context.trace_id for span in platform_exporter.get_finished_spans()} + business_spans = business_exporter.get_finished_spans() + roots = _spans_by_name(business_exporter, _AGENT_RUN_SPAN_NAME) + assert len(roots) == 1 + root = roots[0] + assert root.parent is None + descendants = [span for span in business_spans if span is not root] + assert descendants + for span in descendants: + assert span.parent is not None + assert span.context.trace_id == root.context.trace_id + _assert_parents_resolve_within(business_exporter) + assert root.context.trace_id not in platform_trace_ids + finally: + business_client.shutdown(timeout_millis=5000) + platform_client.shutdown(timeout_millis=5000) + + +def test_platform_httpx_client_span_detaches_from_business_parent() -> None: + platform_exporter = InMemorySpanExporter() + platform_client = _local_client(platform_exporter) + business_exporter = InMemorySpanExporter() + business_client = _local_client(business_exporter) + http = httpx.Client(transport=httpx.MockTransport(lambda _request: httpx.Response(200, json={"ok": True}))) + try: + platform_client.instrument_httpx( + http, + capture_all=False, + tracer_provider=IsolatedTracerProvider(platform_client, preserve_external_parent=True), + ) + + def net_tool() -> str: + return http.get("http://test-only.local/ping").text + + observability = AgentObservability(client=business_client) + agent = create_agent(TestModel(custom_output_text="done"), tools=[Tool(net_tool)]) + observability.instrument(agent) + + with platform_client.span("platform-incoming-request"): + _ = agent.run_sync("test-only input") + http.close() + assert platform_client.force_flush(timeout_millis=10000) + assert business_client.force_flush(timeout_millis=10000) + + platform_spans = platform_exporter.get_finished_spans() + http_spans = [span for span in platform_spans if span.name.startswith("GET")] + assert http_spans + business_spans = business_exporter.get_finished_spans() + business_trace_ids = {span.context.trace_id for span in business_spans} + business_span_ids = {span.context.span_id for span in business_spans} + for span in http_spans: + assert span.parent is None or span.parent.span_id not in business_span_ids + assert span.context.trace_id not in business_trace_ids + _assert_parents_resolve_within(platform_exporter) + _assert_parents_resolve_within(business_exporter) + assert _spans_by_name(business_exporter, _AGENT_RUN_SPAN_NAME) + finally: + http.close() + business_client.shutdown(timeout_millis=5000) + platform_client.shutdown(timeout_millis=5000) + + +def test_isolated_tracer_preserves_same_instance_and_external_parents() -> None: + exporter = InMemorySpanExporter() + client = _local_client(exporter) + try: + tracer = IsolatedTracerProvider(client, preserve_external_parent=True).get_tracer("test-scope") + with client.span("platform-outer"): + with tracer.start_as_current_span("platform-inner"): + pass + remote_ctx = set_span_in_context( + NonRecordingSpan( + SpanContext( + trace_id=0x1111, + span_id=0x2222, + is_remote=True, + trace_flags=TraceFlags(TraceFlags.SAMPLED), + ) + ) + ) + with tracer.start_as_current_span("from-remote", context=remote_ctx): + pass + assert client.force_flush(timeout_millis=10000) + + spans = {span.name: span for span in exporter.get_finished_spans()} + inner = spans["platform-inner"] + outer = spans["platform-outer"] + remote = spans["from-remote"] + assert inner.parent is not None + assert inner.parent.span_id == outer.context.span_id + assert inner.context.trace_id == outer.context.trace_id + assert remote.parent is not None + assert remote.parent.span_id == 0x2222 + assert remote.context.trace_id == 0x1111 + finally: + client.shutdown(timeout_millis=5000) + + +@pytest.mark.parametrize("method", ["start_span", "start_as_current_span"]) +@pytest.mark.parametrize("sampled", [True, False]) +def test_isolated_tracer_remote_and_same_instance_parent_policy(method: str, sampled: bool) -> None: + platform_exporter = InMemorySpanExporter() + platform_client = _local_client(platform_exporter) + business_exporter = InMemorySpanExporter() + business_client = _local_client(business_exporter) + try: + platform_tracer = IsolatedTracerProvider(platform_client, preserve_external_parent=True).get_tracer("platform") + business_tracer = IsolatedTracerProvider(business_client).get_tracer("business") + remote = NonRecordingSpan( + SpanContext( + trace_id=0xABCD, + span_id=0x1234, + is_remote=True, + trace_flags=TraceFlags(TraceFlags.SAMPLED if sampled else 0), + ) + ) + remote_ctx = otel_context.set_value("preserved-key", "preserved-value", set_span_in_context(remote)) + + def invoke(tracer, name, ctx) -> None: + if method == "start_span": + span = tracer.start_span(name, context=ctx) + span.end() + else: + with tracer.start_as_current_span(name, context=ctx): + pass + + invoke(platform_tracer, "platform-remote-child", remote_ctx) + invoke(business_tracer, "business-remote-child", remote_ctx) + with business_client.span("business-outer"): + invoke(business_tracer, "business-same-child", set_span_in_context(get_current_span())) + assert platform_client.force_flush(timeout_millis=10000) + assert business_client.force_flush(timeout_millis=10000) + + platform_remotes = _spans_by_name(platform_exporter, "platform-remote-child") + if sampled: + platform_remote = platform_remotes[0] + assert platform_remote.parent is not None + assert platform_remote.parent.span_id == 0x1234 + assert platform_remote.context.trace_id == 0xABCD + else: + assert platform_remotes == [] + business_remote = _spans_by_name(business_exporter, "business-remote-child")[0] + assert business_remote.parent is None + assert business_remote.context.trace_id != 0xABCD + business_same = _spans_by_name(business_exporter, "business-same-child")[0] + business_outer_span = _spans_by_name(business_exporter, "business-outer")[0] + assert business_same.parent is not None + assert business_same.parent.span_id == business_outer_span.context.span_id + assert business_same.context.trace_id == business_outer_span.context.trace_id + finally: + business_client.shutdown(timeout_millis=5000) + platform_client.shutdown(timeout_millis=5000) + + +def test_concurrent_agent_runs_have_independent_business_roots() -> None: + platform_exporter = InMemorySpanExporter() + platform_client = _local_client(platform_exporter) + business_exporter = InMemorySpanExporter() + business_client = _local_client(business_exporter) + try: + observability = AgentObservability(client=business_client) + agents = [] + for index in range(2): + agent = create_agent(TestModel(custom_output_text=f"done-{index}"), tools=[Tool(_static_tool)]) + observability.instrument(agent) + agents.append(agent) + + async def scenario() -> None: + with platform_client.span("platform-shared-parent") as platform_span: + await asyncio.gather(*(agent.run("test-only input") for agent in agents)) + assert get_current_span().get_span_context() == platform_span.get_span_context() + + asyncio.run(scenario()) + assert platform_client.force_flush(timeout_millis=10000) + assert business_client.force_flush(timeout_millis=10000) + + roots = _spans_by_name(business_exporter, _AGENT_RUN_SPAN_NAME) + assert len(roots) == 2 + assert all(root.parent is None for root in roots) + assert roots[0].context.trace_id != roots[1].context.trace_id + for span in business_exporter.get_finished_spans(): + if span in roots: + continue + assert span.parent is not None + assert span.context.trace_id in {root.context.trace_id for root in roots} + _assert_parents_resolve_within(business_exporter) + finally: + business_client.shutdown(timeout_millis=5000) + platform_client.shutdown(timeout_millis=5000) + + +def test_isolated_tracer_restores_context_on_exception() -> None: + exporter = InMemorySpanExporter() + client = _local_client(exporter) + try: + tracer = IsolatedTracerProvider(client).get_tracer("test-scope") + with client.span("outer") as outer: + with pytest.raises(RuntimeError, match="boom"): + with tracer.start_as_current_span("failing-span"): + raise RuntimeError("boom") + assert get_current_span().get_span_context() == outer.get_span_context() + assert client.force_flush(timeout_millis=10000) + assert _spans_by_name(exporter, "failing-span") + finally: + client.shutdown(timeout_millis=5000) + + +def test_isolated_tracer_restores_context_on_task_cancellation() -> None: + exporter = InMemorySpanExporter() + client = _local_client(exporter) + try: + tracer = IsolatedTracerProvider(client).get_tracer("test-scope") + + async def scenario() -> None: + with client.span("outer") as outer: + started = asyncio.Event() + + async def body() -> None: + with tracer.start_as_current_span("cancelled-span"): + started.set() + await asyncio.Event().wait() + + task = asyncio.create_task(body()) + await asyncio.wait_for(started.wait(), timeout=1) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert get_current_span().get_span_context() == outer.get_span_context() + + asyncio.run(scenario()) + finally: + client.shutdown(timeout_millis=5000) + + +def test_shared_trace_context_mode_links_business_run_to_platform_parent() -> None: + platform_exporter = InMemorySpanExporter() + platform_client = _local_client(platform_exporter) + business_exporter = InMemorySpanExporter() + business_client = _local_client(business_exporter) + http = httpx.Client(transport=httpx.MockTransport(lambda _request: httpx.Response(200, json={"ok": True}))) + try: + platform_client.instrument_httpx( + http, + capture_all=False, + tracer_provider=platform_client.config.get_tracer_provider(), + ) + + def net_tool() -> str: + return http.get("http://test-only.local/ping").text + + observability = AgentObservability(client=business_client, trace_context_mode="shared") + agent = create_agent(TestModel(custom_output_text="done"), tools=[Tool(net_tool)]) + observability.instrument(agent) + + with platform_client.span("shared-platform-root"): + platform_context = get_current_span().get_span_context() + _ = agent.run_sync("test-only input") + http.close() + assert platform_client.force_flush(timeout_millis=10000) + assert business_client.force_flush(timeout_millis=10000) + + platform_spans = platform_exporter.get_finished_spans() + business_spans = business_exporter.get_finished_spans() + all_spans = [*platform_spans, *business_spans] + assert len({span.context.trace_id for span in all_spans}) == 1 + assert all(span.context.trace_id == platform_context.trace_id for span in all_spans) + root = _spans_by_name(business_exporter, _AGENT_RUN_SPAN_NAME)[0] + assert root.parent is not None + assert root.parent.span_id == platform_context.span_id + http_spans = [span for span in platform_spans if span.name.startswith("GET")] + assert http_spans + business_ids = {span.context.span_id for span in business_spans} + for span in http_spans: + assert span.parent is not None + assert span.parent.span_id in business_ids + all_ids = {(span.context.trace_id, span.context.span_id) for span in all_spans} + for span in all_spans: + if span.parent is not None: + assert (span.context.trace_id, span.parent.span_id) in all_ids + assert not any( + span.instrumentation_scope is not None and "pydantic" in span.instrumentation_scope.name + for span in platform_spans + ) + assert not _spans_by_name(business_exporter, "shared-platform-root") + finally: + http.close() + business_client.shutdown(timeout_millis=5000) + platform_client.shutdown(timeout_millis=5000) + + +def test_shared_mode_agent_pipeline_respects_remote_parent_sampling(monkeypatch: pytest.MonkeyPatch) -> None: + business_exporter = InMemorySpanExporter() + monkeypatch.setattr(server_observability, "OTLPSpanExporter", lambda **_kwargs: business_exporter) + settings = ServerSettings( + _env_file=None, + trajectory_enabled=True, + trajectory_otlp_traces_endpoint="http://127.0.0.1:1/v1/traces", + trajectory_service_name="dify-agent-trajectory", + trajectory_trace_context_mode="shared", + ) + observability_instance = server_observability.configure_agent_observability(settings) + assert observability_instance is not None + try: + agent = create_agent(TestModel(custom_output_text="done"), tools=[Tool(_static_tool)]) + observability_instance.instrument(agent) + + unsampled = set_span_in_context( + NonRecordingSpan(SpanContext(trace_id=0x9999, span_id=0x1111, is_remote=True, trace_flags=TraceFlags(0))) + ) + token = otel_context.attach(unsampled) + try: + _ = agent.run_sync("unsampled-parent input") + finally: + otel_context.detach(token) + assert observability_instance.client.force_flush(timeout_millis=10000) + assert not business_exporter.get_finished_spans() + + sampled = set_span_in_context( + NonRecordingSpan( + SpanContext( + trace_id=0x8888, + span_id=0x2222, + is_remote=True, + trace_flags=TraceFlags(TraceFlags.SAMPLED), + ) + ) + ) + token = otel_context.attach(sampled) + try: + _ = agent.run_sync("sampled-parent input") + finally: + otel_context.detach(token) + assert observability_instance.client.force_flush(timeout_millis=10000) + spans = business_exporter.get_finished_spans() + assert spans + assert all(span.context.trace_id == 0x8888 for span in spans) + root = _spans_by_name(business_exporter, _AGENT_RUN_SPAN_NAME)[0] + assert root.parent is not None + assert root.parent.span_id == 0x2222 + finally: + observability_instance.client.shutdown(timeout_millis=5000) + + +def _workflow_execution_context() -> DifyExecutionContextLayerConfig: + return DifyExecutionContextLayerConfig( + tenant_id="tenant-1", + app_id="app-1", + agent_id="agent-1", + user_id="user-1", + user_from="account", + workflow_id="workflow-1", + workflow_run_id="workflow-run-1", + node_id="node-1", + node_execution_id="node-execution-1", + agent_mode="workflow_run", + invoke_from="service-api", + ) + + +def test_dify_run_attributes_use_data_push_names_and_drop_absent_fields() -> None: + workflow_attributes = dict(dify_run_attributes(_workflow_execution_context())) + + assert workflow_attributes == { + "dify.tenant_id": "tenant-1", + "dify.app_id": "app-1", + "dify.agent_id": "agent-1", + "gen_ai.user.id": "user-1", + "dify.invoke_from": "service-api", + "dify.workflow.id": "workflow-1", + "dify.workflow.run_id": "workflow-run-1", + "dify.node.id": "node-1", + "dify.node.execution_id": "node-execution-1", + } + + agent_app_attributes = dict( + dify_run_attributes( + DifyExecutionContextLayerConfig( + tenant_id="tenant-1", + app_id="app-1", + agent_id="agent-1", + conversation_id="conversation-1", + trace_id="trace-1", + user_from="end-user", + agent_mode="agent_app", + invoke_from="web-app", + ) + ) + ) + + # An Agent App turn has no workflow graph, so those keys stay absent instead of + # being exported empty; the conversation and business trace ids take their place. + assert agent_app_attributes == { + "dify.trace_id": "trace-1", + "dify.tenant_id": "tenant-1", + "dify.app_id": "app-1", + "dify.agent_id": "agent-1", + "dify.invoke_from": "web-app", + "dify.conversation.id": "conversation-1", + } + + +@pytest.mark.parametrize("trace_context_mode", ["isolated", "shared"]) +def test_agent_observability_stamps_dify_context_on_every_run_span(trace_context_mode: str) -> None: + exporter = InMemorySpanExporter() + client = _local_client(exporter) + expected = dict(dify_run_attributes(_workflow_execution_context())) + try: + observability = AgentObservability( + client=client, + trace_context_mode=cast(Literal["isolated", "shared"], trace_context_mode), + ) + agent = create_agent(TestModel(custom_output_text="done"), tools=[Tool(_static_tool)]) + observability.instrument(agent, execution_context=_workflow_execution_context()) + + _ = agent.run_sync("test-only input") + assert client.force_flush(timeout_millis=10000) + + spans = exporter.get_finished_spans() + assert spans + for span in spans: + attributes = dict(span.attributes or {}) + assert {key: attributes.get(key) for key in expected} == expected, span.name + run_span = _spans_by_name(exporter, _AGENT_RUN_SPAN_NAME)[0] + assert run_span.attributes is not None + assert run_span.attributes["gen_ai.operation.name"] == "invoke_agent" + finally: + client.shutdown(timeout_millis=5000) + + +def test_agent_observability_omits_dify_attributes_without_an_execution_context() -> None: + exporter = InMemorySpanExporter() + client = _local_client(exporter) + try: + observability = AgentObservability(client=client) + agent = create_agent(TestModel(custom_output_text="done"), tools=[Tool(_static_tool)]) + observability.instrument(agent) + + _ = agent.run_sync("test-only input") + assert client.force_flush(timeout_millis=10000) + + spans = exporter.get_finished_spans() + assert spans + for span in spans: + attributes = dict(span.attributes or {}) + assert not [key for key in attributes if key.startswith("dify.")], span.name + assert GEN_AI_USER_ID_ATTRIBUTE not in attributes + finally: + client.shutdown(timeout_millis=5000) + + +def test_dify_attributes_do_not_break_isolated_parent_policy() -> None: + platform_exporter = InMemorySpanExporter() + platform_client = _local_client(platform_exporter) + business_exporter = InMemorySpanExporter() + business_client = _local_client(business_exporter) + try: + observability = AgentObservability(client=business_client) + agent = create_agent(TestModel(custom_output_text="done"), tools=[Tool(_static_tool)]) + observability.instrument(agent, execution_context=_workflow_execution_context()) + + with platform_client.span("platform-incoming-request"): + _ = agent.run_sync("test-only input") + assert platform_client.force_flush(timeout_millis=10000) + assert business_client.force_flush(timeout_millis=10000) + + platform_root = next( + span for span in platform_exporter.get_finished_spans() if span.name == "platform-incoming-request" + ) + root = _spans_by_name(business_exporter, _AGENT_RUN_SPAN_NAME)[0] + assert root.parent is None + assert root.context is not None + assert root.context.trace_id != platform_root.context.trace_id + _assert_parents_resolve_within(business_exporter) + for span in platform_exporter.get_finished_spans(): + assert DIFY_TENANT_ID_ATTRIBUTE not in dict(span.attributes or {}) + finally: + business_client.shutdown(timeout_millis=5000) + platform_client.shutdown(timeout_millis=5000) diff --git a/dify-agent/tests/local/dify_agent/runtime/test_run_scheduler.py b/dify-agent/tests/local/dify_agent/runtime/test_run_scheduler.py index 7c00424f243..71af4d7a3f0 100644 --- a/dify-agent/tests/local/dify_agent/runtime/test_run_scheduler.py +++ b/dify-agent/tests/local/dify_agent/runtime/test_run_scheduler.py @@ -1,6 +1,7 @@ import asyncio from collections import defaultdict from collections.abc import Mapping +from typing import cast import httpx import pytest @@ -26,6 +27,7 @@ from dify_agent.protocol.schemas import ( utc_now, ) from dify_agent.runtime.cancellation import RunCancellationIntent +from dify_agent.runtime.observability import AgentObservability from dify_agent.runtime.event_sink import ( NonTerminalRunEvent, RunFinalizationResult, @@ -487,6 +489,35 @@ def test_default_runner_factory_passes_runtime_limits_to_runner() -> None: asyncio.run(scenario()) +def test_default_runner_factory_passes_agent_observability_instance() -> None: + async def scenario() -> None: + store = FakeStore() + record = await store.create_run() + sentinel = cast(AgentObservability, object()) + async with httpx.AsyncClient() as client: + scheduler = RunScheduler( + store=store, + plugin_daemon_http_client=client, + dify_api_http_client=client, + agent_observability=sentinel, + ) + + runner = scheduler._default_runner_factory(record, _request(), is_cancelled=lambda: False) + default_scheduler = RunScheduler( + store=store, + plugin_daemon_http_client=client, + dify_api_http_client=client, + ) + default_runner = default_scheduler._default_runner_factory(record, _request(), is_cancelled=lambda: False) + + assert isinstance(runner, AgentRunRunner) + assert runner.agent_observability is sentinel + assert isinstance(default_runner, AgentRunRunner) + assert default_runner.agent_observability is None + + asyncio.run(scenario()) + + def test_create_run_starts_background_task_and_returns_running() -> None: async def scenario() -> None: store = FakeStore() diff --git a/dify-agent/tests/local/dify_agent/runtime/test_runner.py b/dify-agent/tests/local/dify_agent/runtime/test_runner.py index aa72f1a15af..be963726254 100644 --- a/dify-agent/tests/local/dify_agent/runtime/test_runner.py +++ b/dify-agent/tests/local/dify_agent/runtime/test_runner.py @@ -69,6 +69,8 @@ from dify_agent.protocol.schemas import ( RunLayerSpec, RunSucceededEvent, ) + +from dify_agent.runtime.observability import AgentObservability from dify_agent.runtime.event_sink import InMemoryRunEventSink from dify_agent.runtime.compositor_factory import create_default_layer_providers from dify_agent.runtime.runner import ( @@ -264,6 +266,8 @@ def _request( ask_human_config: DifyAskHumanLayerConfig | None = None, llm_layer_name: str = DIFY_AGENT_MODEL_LAYER_ID, execution_context_layer_name: str = "execution_context", + app_id: str | None = None, + agent_id: str | None = None, on_exit: LayerExitSignals | None = None, output_config: Mapping[str, object] | DifyOutputLayerConfig | None = None, model_settings: ModelSettings | None = None, @@ -296,6 +300,8 @@ def _request( type=DIFY_EXECUTION_CONTEXT_LAYER_TYPE_ID, config=DifyExecutionContextLayerConfig( tenant_id="tenant-1", + app_id=app_id, + agent_id=agent_id, user_from="account", agent_mode="workflow_run", invoke_from="service-api", @@ -739,6 +745,61 @@ def test_runner_passes_context_compaction(monkeypatch: pytest.MonkeyPatch) -> No assert sink.statuses["run-compaction"] == "succeeded" +def test_runner_instruments_agent_with_injected_observability(monkeypatch: pytest.MonkeyPatch) -> None: + class FakeAgentObservability: + def __init__(self) -> None: + self.instrumented: list[tuple[object, DifyExecutionContextLayerConfig | None]] = [] + + def instrument( + self, + agent: object, + *, + execution_context: DifyExecutionContextLayerConfig | None = None, + ) -> None: + self.instrumented.append((agent, execution_context)) + + class FakeAgent: + async def run(self, *_args: object, **_kwargs: object) -> FakeAgentRunResult: + return FakeAgentRunResult("done") + + observability = FakeAgentObservability() + created_agent = FakeAgent() + + def fake_get_model(_self: DifyPluginLLMLayer, *, http_client: httpx.AsyncClient, agent_run_id: str): + assert http_client.is_closed is False + return TestModel(custom_output_text="unused") # pyright: ignore[reportReturnType] + + monkeypatch.setattr(DifyPluginLLMLayer, "get_model", fake_get_model) + monkeypatch.setattr("dify_agent.runtime.runner.create_agent", lambda *_args, **_kwargs: created_agent) + sink = InMemoryRunEventSink() + + async def scenario() -> None: + async with httpx.AsyncClient() as client: + await AgentRunRunner( + sink=sink, + request=_request(app_id="app-1", agent_id="agent-1"), + run_id="run-observed", + plugin_daemon_http_client=client, + dify_api_http_client=client, + agent_observability=cast(AgentObservability, observability), + ).run() + + asyncio.run(scenario()) + + assert sink.statuses["run-observed"] == "succeeded" + # The run's Dify context reaches observability through the model layer's + # execution-context dependency, whose node name the caller chooses. + assert len(observability.instrumented) == 1 + instrumented_agent, execution_context = observability.instrumented[0] + assert instrumented_agent is created_agent + assert execution_context is not None + assert (execution_context.tenant_id, execution_context.app_id, execution_context.agent_id) == ( + "tenant-1", + "app-1", + "agent-1", + ) + + def test_runner_rejects_compaction_budget_before_model_resolution_or_invocation( monkeypatch: pytest.MonkeyPatch, ) -> None: diff --git a/dify-agent/tests/local/dify_agent/runtime_backend/test_profile.py b/dify-agent/tests/local/dify_agent/runtime_backend/test_profile.py index 532a2a552b6..cfba7a09305 100644 --- a/dify-agent/tests/local/dify_agent/runtime_backend/test_profile.py +++ b/dify-agent/tests/local/dify_agent/runtime_backend/test_profile.py @@ -21,11 +21,14 @@ from dify_agent.runtime_backend.profile import ( @pytest.fixture(autouse=True) def _isolated_backend_env(monkeypatch: pytest.MonkeyPatch) -> None: - # RuntimeBackendSettings reads the process environment, so a developer's - # exported DIFY_AGENT_* / E2B_* values must not leak into these tests. + # RuntimeBackendSettings reads both the process environment and a dotenv file + # resolved against the current directory, so neither a developer's exported + # DIFY_AGENT_* / E2B_* values nor their local dify-agent/.env may leak in. + # Every case here states the settings it exercises. for name in list(os.environ): if name.startswith("DIFY_AGENT_") or name in ("E2B_API_KEY", "E2B_API_TOKEN"): monkeypatch.delenv(name) + monkeypatch.setitem(RuntimeBackendSettings.model_config, "env_file", None) def test_e2b_backend_uses_prepared_dify_template_and_one_hour_lease_by_default() -> None: diff --git a/dify-agent/tests/local/dify_agent/server/test_app.py b/dify-agent/tests/local/dify_agent/server/test_app.py index 23c9da4343b..8bab0a95bf2 100644 --- a/dify-agent/tests/local/dify_agent/server/test_app.py +++ b/dify-agent/tests/local/dify_agent/server/test_app.py @@ -1,6 +1,7 @@ from __future__ import annotations import base64 +import os import time from typing import ClassVar @@ -9,6 +10,7 @@ import pytest from fastapi.testclient import TestClient import dify_agent.server.app as app_module +import dify_agent.server.observability as server_observability from dify_agent.layers.execution_context import DifyExecutionContextLayerConfig from dify_agent.layers.execution_context.layer import DifyExecutionContextLayer from dify_agent.layers.knowledge.configs import DifyKnowledgeBaseLayerConfig @@ -24,6 +26,24 @@ from dify_agent.server.settings import ServerSettings from dify_agent.storage.redis_run_store import RedisRunStore +@pytest.fixture(autouse=True) +def _isolated_app_settings(monkeypatch: pytest.MonkeyPatch) -> None: + """Keep app construction independent of the developer's dotenv and SDK env. + + ``ServerSettings`` resolves ``.env``/``dify-agent/.env`` against the current + directory, and importing this module builds ``app_module.app``, which already + latched the process-global trace context mode from whatever it found there. + Pinning the settings source and resetting that latch lets each case build an + app from the settings it states, in any order and on any machine. + """ + for name in tuple(os.environ): + if name.startswith(("DIFY_AGENT_", "OTEL_", "LOGFIRE_")): + monkeypatch.delenv(name) + monkeypatch.setitem(ServerSettings.model_config, "env_file", None) + monkeypatch.setattr(server_observability, "_global_instrumentation_ready", False) + monkeypatch.setattr(server_observability, "_global_trace_context_mode", None) + + def _base64url_secret(value: bytes) -> str: return base64.urlsafe_b64encode(value).rstrip(b"=").decode("ascii") @@ -89,6 +109,7 @@ class FakeRunScheduler: stream_text_delta_flush_interval_seconds: float, stream_text_delta_max_chars: int, layer_providers: tuple[DifyAgentLayerProvider, ...], + agent_observability: object | None = None, ) -> None: self.store = store self.shutdown_grace_seconds = shutdown_grace_seconds @@ -99,6 +120,7 @@ class FakeRunScheduler: self.layer_providers = layer_providers self.plugin_daemon_http_client = plugin_daemon_http_client self.dify_api_http_client = dify_api_http_client + self.agent_observability = agent_observability self.shutdown_called = False self.created.append(self) @@ -449,6 +471,84 @@ def test_create_dify_api_inner_http_client_uses_generic_outbound_httpx_construct assert client.trust_env is False +def test_create_app_lifecycle_owns_agent_observability_instance(monkeypatch: pytest.MonkeyPatch) -> None: + _patch_app_lifecycle(monkeypatch) + events: list[str] = [] + + class FakeAgentObservability: + async def aclose(self) -> None: + events.append("agent-observability-close") + + sentinel = FakeAgentObservability() + + def fake_configure_agent_observability(_settings: ServerSettings) -> object: + events.append("agent-observability-configure") + return sentinel + + async def recording_shutdown(self: FakeRunScheduler) -> None: + self.shutdown_called = True + events.append("scheduler-shutdown") + + monkeypatch.setattr(app_module, "configure_agent_observability", fake_configure_agent_observability) + monkeypatch.setattr(FakeRunScheduler, "shutdown", recording_shutdown) + FakeRunScheduler.created.clear() + + app = create_app(ServerSettings(redis_url="redis://example.invalid/0")) + with TestClient(app): + assert app.state.agent_observability is sentinel + assert FakeRunScheduler.created[0].agent_observability is sentinel + assert events == ["agent-observability-configure"] + + # The instance is closed after the scheduler drains, so spans from runs + # finishing during shutdown still reach the exporter. + assert events == ["agent-observability-configure", "scheduler-shutdown", "agent-observability-close"] + + +def test_create_app_defaults_agent_observability_to_disabled(monkeypatch: pytest.MonkeyPatch) -> None: + _patch_app_lifecycle(monkeypatch) + FakeRunScheduler.created.clear() + settings = ServerSettings(redis_url="redis://example.invalid/0", trajectory_enabled=False) + created: list[ServerSettings] = [] + real_configure = app_module.configure_agent_observability + + def recording_configure(settings: ServerSettings): + created.append(settings) + return real_configure(settings) + + monkeypatch.setattr(app_module, "configure_agent_observability", recording_configure) + app = create_app(settings) + with TestClient(app): + assert app.state.agent_observability is None + assert FakeRunScheduler.created[0].agent_observability is None + + assert created == [settings] + assert app.state.platform_observability is not None + + +def test_create_app_passes_settings_to_server_observability(monkeypatch: pytest.MonkeyPatch) -> None: + _patch_app_lifecycle(monkeypatch) + captured: dict[str, object] = {} + sentinel = object() + + def fake_configure(app, *, settings=None): + captured["app"] = app + captured["settings"] = settings + return sentinel + + monkeypatch.setattr(app_module, "configure_server_observability", fake_configure) + settings = ServerSettings( + _env_file=None, + redis_url="redis://example.invalid/0", + trajectory_trace_context_mode="shared", + ) + + app = create_app(settings) + + assert captured["app"] is app + assert captured["settings"] is settings + assert app.state.platform_observability is sentinel + + def test_server_settings_use_generic_outbound_http_args_for_shared_clients() -> None: model_fields = ServerSettings.model_fields diff --git a/dify-agent/tests/local/dify_agent/server/test_observability.py b/dify-agent/tests/local/dify_agent/server/test_observability.py index 534298415ba..d730498e81e 100644 --- a/dify-agent/tests/local/dify_agent/server/test_observability.py +++ b/dify-agent/tests/local/dify_agent/server/test_observability.py @@ -1,15 +1,47 @@ from __future__ import annotations -from typing import ClassVar +import asyncio +import os +from typing import ClassVar, cast +import logfire +import pytest from fastapi import FastAPI +from opentelemetry.sdk.trace.sampling import ALWAYS_ON, ParentBased +from opentelemetry.trace import NoOpTracerProvider import dify_agent.server.observability as observability +from dify_agent.runtime.observability import AgentObservability from dify_agent.server.observability import configure_server_observability +from dify_agent.server.settings import ServerSettings + + +@pytest.fixture(autouse=True) +def isolate_observability_environment(monkeypatch, tmp_path): + # The temporary working directory hides a developer's dotenv, while stripping + # the SDK and DIFY_AGENT_* variables hides exported ones. Cases that exercise + # a dotenv or a specific variable write it themselves after this runs. + monkeypatch.chdir(tmp_path) + monkeypatch.setattr(os, "environ", dict(os.environ)) + for key in tuple(os.environ): + if key.startswith(("OTEL_", "LOGFIRE_", "DIFY_AGENT_")): + monkeypatch.delenv(key) + + +_FAKE_NATIVE_PROVIDER = NoOpTracerProvider() + + +class _FakeLogfireConfig: + @staticmethod + def get_tracer_provider() -> NoOpTracerProvider: + return _FAKE_NATIVE_PROVIDER class FakeLogfireModule: + SamplingOptions: ClassVar[type] = logfire.SamplingOptions + config: ClassVar[_FakeLogfireConfig] = _FakeLogfireConfig() configure_calls: ClassVar[list[dict[str, object]]] = [] + configure_envs: ClassVar[list[dict[str, str]]] = [] fastapi_calls: ClassVar[list[dict[str, object]]] = [] httpx_calls: ClassVar[list[dict[str, object]]] = [] redis_calls: ClassVar[list[dict[str, object]]] = [] @@ -18,14 +50,19 @@ class FakeLogfireModule: @classmethod def reset(cls) -> None: cls.configure_calls.clear() + cls.configure_envs.clear() cls.fastapi_calls.clear() cls.httpx_calls.clear() cls.redis_calls.clear() cls.pydantic_ai_calls.clear() @classmethod - def configure(cls, **kwargs: object) -> None: + def configure(cls, **kwargs: object) -> "type[FakeLogfireModule]": cls.configure_calls.append(kwargs) + cls.configure_envs.append( + {key: value for key, value in os.environ.items() if key.startswith(("OTEL_", "LOGFIRE_"))} + ) + return cls @classmethod def instrument_fastapi(cls, app: FastAPI, **kwargs: object) -> None: @@ -50,14 +87,16 @@ def test_configure_server_observability_keeps_remote_export_token_gated_by_logfi monkeypatch.setattr(observability, "_global_instrumentation_ready", False) app = FastAPI() - configure_server_observability(app) + platform = configure_server_observability(app) + assert platform is FakeLogfireModule assert FakeLogfireModule.configure_calls == [ { "send_to_logfire": "if-token-present", "inspect_arguments": False, } ] + assert FakeLogfireModule.configure_envs == [{}] def test_configure_server_observability_instruments_server_boundaries_once(monkeypatch) -> None: @@ -70,12 +109,418 @@ def test_configure_server_observability_instruments_server_boundaries_once(monke configure_server_observability(first_app) configure_server_observability(second_app) - assert FakeLogfireModule.httpx_calls == [{}] - assert FakeLogfireModule.redis_calls == [{}] - assert FakeLogfireModule.pydantic_ai_calls == [{}] + expected_provider = observability.IsolatedTracerProvider(FakeLogfireModule, preserve_external_parent=True) + assert FakeLogfireModule.httpx_calls == [ + { + "capture_all": False, + "capture_headers": False, + "capture_request_body": False, + "capture_response_body": False, + "tracer_provider": expected_provider, + } + ] + assert FakeLogfireModule.redis_calls == [{"capture_statement": False, "tracer_provider": expected_provider}] + assert FakeLogfireModule.pydantic_ai_calls == [] assert FakeLogfireModule.fastapi_calls == [ - {"app": first_app}, - {"app": second_app}, + { + "app": first_app, + "request_attributes_mapper": observability._platform_request_attributes, + "capture_headers": False, + "tracer_provider": expected_provider, + }, + { + "app": second_app, + "request_attributes_mapper": observability._platform_request_attributes, + "capture_headers": False, + "tracer_provider": expected_provider, + }, ] assert first_app.state.dify_agent_logfire_instrumented is True assert second_app.state.dify_agent_logfire_instrumented is True + + +def test_configure_server_observability_shared_mode_uses_native_tracer_provider(monkeypatch) -> None: + FakeLogfireModule.reset() + monkeypatch.setattr(observability, "logfire", FakeLogfireModule) + monkeypatch.setattr(observability, "_global_instrumentation_ready", False) + monkeypatch.setattr(observability, "_global_trace_context_mode", None) + app = FastAPI() + + configure_server_observability(app, settings=ServerSettings(trajectory_trace_context_mode="shared")) + + assert FakeLogfireModule.httpx_calls == [ + { + "capture_all": False, + "capture_headers": False, + "capture_request_body": False, + "capture_response_body": False, + "tracer_provider": _FAKE_NATIVE_PROVIDER, + } + ] + assert FakeLogfireModule.redis_calls == [{"capture_statement": False, "tracer_provider": _FAKE_NATIVE_PROVIDER}] + assert FakeLogfireModule.fastapi_calls == [ + { + "app": app, + "request_attributes_mapper": observability._platform_request_attributes, + "capture_headers": False, + "tracer_provider": _FAKE_NATIVE_PROVIDER, + } + ] + assert FakeLogfireModule.pydantic_ai_calls == [] + + +def test_configure_server_observability_reads_trace_context_mode_from_environment(monkeypatch) -> None: + FakeLogfireModule.reset() + monkeypatch.setattr(observability, "logfire", FakeLogfireModule) + monkeypatch.setattr(observability, "_global_instrumentation_ready", False) + monkeypatch.setattr(observability, "_global_trace_context_mode", None) + monkeypatch.setenv("DIFY_AGENT_TRAJECTORY_TRACE_CONTEXT_MODE", "shared") + + configure_server_observability(FastAPI()) + + assert FakeLogfireModule.httpx_calls[0]["tracer_provider"] is _FAKE_NATIVE_PROVIDER + + +def test_configure_server_observability_rejects_trace_context_mode_change(monkeypatch) -> None: + FakeLogfireModule.reset() + monkeypatch.setattr(observability, "logfire", FakeLogfireModule) + monkeypatch.setattr(observability, "_global_instrumentation_ready", False) + monkeypatch.setattr(observability, "_global_trace_context_mode", None) + + configure_server_observability(FastAPI(), settings=ServerSettings(trajectory_trace_context_mode="isolated")) + + with pytest.raises(ValueError, match="Trace context mode cannot change"): + configure_server_observability(FastAPI(), settings=ServerSettings(trajectory_trace_context_mode="shared")) + assert len(FakeLogfireModule.configure_calls) == 1 + + configure_server_observability(FastAPI(), settings=ServerSettings(trajectory_trace_context_mode="isolated")) + assert len(FakeLogfireModule.configure_calls) == 2 + + +@pytest.mark.parametrize("dotenv_path", [".env", "dify-agent/.env"]) +def test_configure_server_observability_forwards_otel_logfire_from_dotenv( + monkeypatch, tmp_path, dotenv_path: str +) -> None: + FakeLogfireModule.reset() + monkeypatch.setattr(observability, "logfire", FakeLogfireModule) + monkeypatch.setattr(observability, "_global_instrumentation_ready", False) + dotenv_file = tmp_path / dotenv_path + dotenv_file.parent.mkdir(parents=True, exist_ok=True) + dotenv_file.write_text( + 'OTEL_EXPORTER_OTLP_ENDPOINT="http://127.0.0.1:4318"\n' + 'OTEL_EXPORTER_OTLP_HEADERS="Authorization=Bearer test-only"\n' + "LOGFIRE_SERVICE_NAME=dify-agent-test\n" + 'OTEL_RESOURCE_ATTRIBUTES="service.namespace=dify,deployment.environment.name=local"\n' + "UNRELATED_SECRET=not-forwarded\n" + "DIFY_AGENT_API_TOKEN=not-forwarded\n" + "OTLP_API_KEY=not-forwarded\n" + "NOT_OTEL_VALUE=not-forwarded\n" + "OTELISH=not-forwarded\n" + "OTEL_UNSET\n" + "LOGFIRE_EMPTY=\n" + ) + + configure_server_observability(FastAPI()) + + assert FakeLogfireModule.configure_envs == [ + { + "OTEL_EXPORTER_OTLP_ENDPOINT": "http://127.0.0.1:4318", + "OTEL_EXPORTER_OTLP_HEADERS": "Authorization=Bearer test-only", + "LOGFIRE_SERVICE_NAME": "dify-agent-test", + "OTEL_RESOURCE_ATTRIBUTES": "service.namespace=dify,deployment.environment.name=local", + "LOGFIRE_EMPTY": "", + } + ] + for key in ( + "UNRELATED_SECRET", + "DIFY_AGENT_API_TOKEN", + "OTLP_API_KEY", + "NOT_OTEL_VALUE", + "OTELISH", + "OTEL_UNSET", + ): + assert key not in os.environ + + +def test_configure_server_observability_later_dotenv_file_wins(monkeypatch, tmp_path) -> None: + FakeLogfireModule.reset() + monkeypatch.setattr(observability, "logfire", FakeLogfireModule) + monkeypatch.setattr(observability, "_global_instrumentation_ready", False) + (tmp_path / ".env").write_text("OTEL_EXPORTER_OTLP_ENDPOINT=http://root:4318\nLOGFIRE_SERVICE_NAME=root\n") + nested_dir = tmp_path / "dify-agent" + nested_dir.mkdir() + (nested_dir / ".env").write_text("OTEL_EXPORTER_OTLP_ENDPOINT=http://nested:4318\n") + + configure_server_observability(FastAPI()) + + assert FakeLogfireModule.configure_envs == [ + { + "OTEL_EXPORTER_OTLP_ENDPOINT": "http://nested:4318", + "LOGFIRE_SERVICE_NAME": "root", + } + ] + + +@pytest.mark.parametrize("existing_endpoint", ["http://existing:4318", ""]) +def test_configure_server_observability_process_env_wins_and_is_idempotent( + monkeypatch, tmp_path, existing_endpoint: str +) -> None: + FakeLogfireModule.reset() + monkeypatch.setattr(observability, "logfire", FakeLogfireModule) + monkeypatch.setattr(observability, "_global_instrumentation_ready", False) + monkeypatch.setenv("OTEL_EXPORTER_OTLP_ENDPOINT", existing_endpoint) + monkeypatch.setenv("LOGFIRE_SERVICE_NAME", "existing-service") + (tmp_path / ".env").write_text( + "OTEL_EXPORTER_OTLP_ENDPOINT=http://dotenv:4318\n" + "LOGFIRE_SERVICE_NAME=dotenv-service\n" + "OTEL_EXPORTER_OTLP_PROTOCOL=http/protobuf\n" + ) + + configure_server_observability(FastAPI()) + configure_server_observability(FastAPI()) + + expected = { + "OTEL_EXPORTER_OTLP_ENDPOINT": existing_endpoint, + "LOGFIRE_SERVICE_NAME": "existing-service", + "OTEL_EXPORTER_OTLP_PROTOCOL": "http/protobuf", + } + assert FakeLogfireModule.configure_envs == [expected, expected] + assert os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] == existing_endpoint + + +@pytest.mark.parametrize("use_custom_file", [True, False]) +def test_configure_server_observability_reads_configured_env_file_paths( + monkeypatch, tmp_path, use_custom_file: bool +) -> None: + FakeLogfireModule.reset() + monkeypatch.setattr(observability, "logfire", FakeLogfireModule) + monkeypatch.setattr(observability, "_global_instrumentation_ready", False) + (tmp_path / ".env").write_text("LOGFIRE_SERVICE_NAME=default-dotenv\n") + if use_custom_file: + custom_file = tmp_path / "custom.env" + custom_file.write_text("LOGFIRE_SERVICE_NAME=custom-dotenv\n") + monkeypatch.setitem(ServerSettings.model_config, "env_file", custom_file) + else: + monkeypatch.setitem(ServerSettings.model_config, "env_file", None) + + configure_server_observability(FastAPI()) + + if use_custom_file: + assert FakeLogfireModule.configure_envs == [{"LOGFIRE_SERVICE_NAME": "custom-dotenv"}] + else: + assert FakeLogfireModule.configure_envs == [{}] + + +class FakeOTLPSpanExporter: + calls: ClassVar[list[dict[str, object]]] = [] + shutdown_called: bool + + def __init__(self, **kwargs: object) -> None: + self.kwargs = kwargs + self.shutdown_called = False + type(self).calls.append(kwargs) + + def shutdown(self) -> None: + self.shutdown_called = True + + +class FakeBatchSpanProcessor: + calls: ClassVar[list[dict[str, object]]] = [] + instances: ClassVar[list["FakeBatchSpanProcessor"]] = [] + exporter: FakeOTLPSpanExporter + shutdown_called: bool + + def __init__(self, exporter: FakeOTLPSpanExporter, **kwargs: object) -> None: + self.exporter = exporter + self.kwargs = kwargs + self.shutdown_called = False + type(self).calls.append(kwargs) + type(self).instances.append(self) + + def shutdown(self) -> None: + self.shutdown_called = True + + +class RaisingLogfireModule(FakeLogfireModule): + @classmethod + def configure(cls, **_kwargs: object) -> "type[RaisingLogfireModule]": + cls.configure_calls.append(_kwargs) + raise RuntimeError("configure failed") + + +def _patch_agent_observability_dependencies(monkeypatch, logfire_module=FakeLogfireModule) -> None: + logfire_module.reset() + FakeOTLPSpanExporter.calls.clear() + FakeBatchSpanProcessor.calls.clear() + monkeypatch.setattr(observability, "logfire", logfire_module) + monkeypatch.setattr(observability, "OTLPSpanExporter", FakeOTLPSpanExporter) + monkeypatch.setattr(observability, "BatchSpanProcessor", FakeBatchSpanProcessor) + FakeBatchSpanProcessor.instances.clear() + + +def _enabled_trajectory_settings(**overrides: object) -> ServerSettings: + values: dict[str, object] = { + "trajectory_enabled": True, + "trajectory_otlp_traces_endpoint": "http://127.0.0.1:4318/v1/traces", + } + values.update(overrides) + return ServerSettings.model_validate(values) + + +def test_configure_agent_observability_disabled_creates_no_sdk_or_exporter(monkeypatch) -> None: + _patch_agent_observability_dependencies(monkeypatch) + settings = _enabled_trajectory_settings(trajectory_enabled=False) + + assert observability.configure_agent_observability(settings) is None + assert FakeLogfireModule.configure_calls == [] + assert FakeOTLPSpanExporter.calls == [] + assert FakeBatchSpanProcessor.calls == [] + + +def test_configure_agent_observability_missing_endpoint_fails_before_sdk_setup(monkeypatch) -> None: + _patch_agent_observability_dependencies(monkeypatch) + settings = _enabled_trajectory_settings() + settings.trajectory_otlp_traces_endpoint = None + + with pytest.raises(ValueError, match="trajectory_otlp_traces_endpoint"): + observability.configure_agent_observability(settings) + assert FakeLogfireModule.configure_calls == [] + assert FakeOTLPSpanExporter.calls == [] + + +def test_configure_agent_observability_builds_independent_pipeline_without_inherited_env(monkeypatch) -> None: + _patch_agent_observability_dependencies(monkeypatch) + monkeypatch.setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://platform:4318") + monkeypatch.setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", "http://platform:4318/v1/traces") + monkeypatch.setenv("OTEL_EXPORTER_OTLP_HEADERS", "Authorization=platform-test-only") + monkeypatch.setenv("LOGFIRE_TOKEN", "platform-test-only") + settings = _enabled_trajectory_settings( + trajectory_otlp_headers={"Authorization": "agent-test-only"}, + trajectory_service_name="dify-agent-trajectory", + trajectory_include_content=True, + ) + + result = observability.configure_agent_observability(settings) + + assert isinstance(result, AgentObservability) + assert result.client is FakeLogfireModule + assert result.include_content is True + assert FakeOTLPSpanExporter.calls == [ + { + "endpoint": "http://127.0.0.1:4318/v1/traces", + "headers": {"Authorization": "agent-test-only"}, + "timeout": 5, + } + ] + assert FakeBatchSpanProcessor.calls == [ + { + "max_queue_size": 2048, + "max_export_batch_size": 512, + "schedule_delay_millis": 5000, + "export_timeout_millis": 5000, + } + ] + configure_kwargs = FakeLogfireModule.configure_calls[0] + assert configure_kwargs["local"] is True + assert configure_kwargs["send_to_logfire"] is False + assert configure_kwargs["service_name"] == "dify-agent-trajectory" + assert configure_kwargs["console"] is False + assert configure_kwargs["metrics"] is False + assert configure_kwargs["inspect_arguments"] is False + assert configure_kwargs["add_baggage_to_attributes"] is False + assert configure_kwargs["sampling"] is not None + assert configure_kwargs["sampling"].head is ALWAYS_ON + assert result.trace_context_mode == "isolated" + assert FakeLogfireModule.configure_envs == [{}] + assert os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] == "http://platform:4318" + assert os.environ["OTEL_EXPORTER_OTLP_HEADERS"] == "Authorization=platform-test-only" + assert os.environ["LOGFIRE_TOKEN"] == "platform-test-only" + + +def test_configure_agent_observability_shared_mode_uses_parent_based_sampling(monkeypatch) -> None: + _patch_agent_observability_dependencies(monkeypatch) + settings = _enabled_trajectory_settings( + trajectory_trace_context_mode="shared", + trajectory_include_content=True, + ) + + result = observability.configure_agent_observability(settings) + + assert isinstance(result, AgentObservability) + assert result.trace_context_mode == "shared" + assert result.include_content is True + configure_kwargs = FakeLogfireModule.configure_calls[0] + assert isinstance(configure_kwargs["sampling"].head, ParentBased) + + +@pytest.mark.parametrize("mode", ["isolated", "shared"]) +def test_configure_agent_observability_disabled_ignores_trace_context_mode(monkeypatch, mode: str) -> None: + _patch_agent_observability_dependencies(monkeypatch) + settings = _enabled_trajectory_settings(trajectory_enabled=False, trajectory_trace_context_mode=mode) + + assert observability.configure_agent_observability(settings) is None + assert FakeLogfireModule.configure_calls == [] + assert FakeOTLPSpanExporter.calls == [] + + +def test_configure_agent_observability_passes_batch_processor_limits_to_processor(monkeypatch) -> None: + _patch_agent_observability_dependencies(monkeypatch) + settings = _enabled_trajectory_settings( + trajectory_max_queue_size=1024, + trajectory_max_export_batch_size=128, + trajectory_schedule_delay_ms=250, + trajectory_export_timeout_ms=1500, + ) + + observability.configure_agent_observability(settings) + + assert FakeBatchSpanProcessor.calls == [ + { + "max_queue_size": 1024, + "max_export_batch_size": 128, + "schedule_delay_millis": 250, + "export_timeout_millis": 1500, + } + ] + assert FakeOTLPSpanExporter.calls[0]["timeout"] == 5 + + +def test_configure_agent_observability_empty_headers_do_not_inherit_platform_auth(monkeypatch) -> None: + _patch_agent_observability_dependencies(monkeypatch) + monkeypatch.setenv("OTEL_EXPORTER_OTLP_HEADERS", "Authorization=platform-test-only") + + observability.configure_agent_observability(_enabled_trajectory_settings()) + + assert FakeOTLPSpanExporter.calls[0]["headers"] == {} + assert os.environ["OTEL_EXPORTER_OTLP_HEADERS"] == "Authorization=platform-test-only" + + +def test_configure_agent_observability_restores_env_and_shuts_processor_on_configure_failure( + monkeypatch, +) -> None: + _patch_agent_observability_dependencies(monkeypatch, logfire_module=RaisingLogfireModule) + monkeypatch.setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://platform:4318") + + with pytest.raises(RuntimeError, match="configure failed"): + observability.configure_agent_observability(_enabled_trajectory_settings()) + + assert len(FakeBatchSpanProcessor.instances) == 1 + assert FakeBatchSpanProcessor.instances[0].shutdown_called is True + assert os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] == "http://platform:4318" + + +class FakeAgentLogfireClient: + def __init__(self) -> None: + self.shutdown_calls: list[dict[str, object]] = [] + + def shutdown(self, **kwargs: object) -> None: + self.shutdown_calls.append(kwargs) + + +def test_agent_observability_aclose_shuts_down_only_the_agent_client() -> None: + client = FakeAgentLogfireClient() + instance = AgentObservability(client=cast(logfire.Logfire, client)) + + asyncio.run(instance.aclose()) + + assert client.shutdown_calls == [{"timeout_millis": 5000}] diff --git a/dify-agent/tests/local/dify_agent/server/test_observability_isolation.py b/dify-agent/tests/local/dify_agent/server/test_observability_isolation.py new file mode 100644 index 00000000000..7d48d45fd2e --- /dev/null +++ b/dify-agent/tests/local/dify_agent/server/test_observability_isolation.py @@ -0,0 +1,285 @@ +"""Real two-endpoint OTLP isolation between platform and Agent observability.""" + +from __future__ import annotations + +import os +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +import logfire +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient +from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ( + ExportTraceServiceRequest, + ExportTraceServiceResponse, +) +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter +from pydantic import BaseModel +from pydantic_ai import Agent, Tool +from pydantic_ai.models.test import TestModel + +import dify_agent.server.observability as observability +from dify_agent.runtime.agent_factory import create_agent +from dify_agent.runtime.observability import IsolatedTracerProvider +from dify_agent.server.observability import configure_agent_observability +from dify_agent.server.settings import ServerSettings + + +class _OTLPReceiver: + """Collect decoded OTLP trace exports and request headers on a local port.""" + + def __init__(self) -> None: + received: list[tuple[dict[str, str], ExportTraceServiceRequest]] = [] + + class Handler(BaseHTTPRequestHandler): + def do_POST(self) -> None: + length = int(self.headers.get("Content-Length", "0")) + request = ExportTraceServiceRequest() + request.ParseFromString(self.rfile.read(length)) + received.append(({key.casefold(): value for key, value in self.headers.items()}, request)) + body = ExportTraceServiceResponse().SerializeToString() + self.send_response(200) + self.send_header("Content-Type", "application/x-protobuf") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, *_args: object) -> None: + return None + + self.received = received + self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) + self.thread.start() + + @property + def endpoint(self) -> str: + host, port = self.server.server_address[:2] + return f"http://{host}:{port}/v1/traces" + + def service_names(self) -> list[str]: + names: list[str] = [] + for _headers, request in self.received: + for resource_spans in request.resource_spans: + for attribute in resource_spans.resource.attributes: + if attribute.key == "service.name": + names.append(attribute.value.string_value) + return names + + def span_names(self) -> list[str]: + names: list[str] = [] + for _headers, request in self.received: + for resource_spans in request.resource_spans: + for scope_spans in resource_spans.scope_spans: + names.extend(span.name for span in scope_spans.spans) + return names + + def scope_names(self) -> list[str]: + names: list[str] = [] + for _headers, request in self.received: + for resource_spans in request.resource_spans: + for scope_spans in resource_spans.scope_spans: + names.append(scope_spans.scope.name) + return names + + def authorizations(self) -> list[str | None]: + return [headers.get("authorization") for headers, _request in self.received] + + def close(self) -> None: + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=5) + + +@pytest.fixture(autouse=True) +def isolate_environment(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None: + monkeypatch.chdir(tmp_path) + monkeypatch.setattr(os, "environ", dict(os.environ)) + for key in tuple(os.environ): + if key.startswith(("OTEL_", "LOGFIRE_")): + monkeypatch.delenv(key) + + +@pytest.fixture +def restore_agent_instrumentation(monkeypatch: pytest.MonkeyPatch): + monkeypatch.setattr(Agent, "_instrument_default", False) + + +def _static_tool() -> str: + return "tool-sentinel-output" + + +def _platform_client(endpoint: str, headers: dict[str, str]) -> logfire.Logfire: + from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter + from opentelemetry.sdk.trace.export import BatchSpanProcessor + + return logfire.configure( + local=True, + send_to_logfire=False, + console=False, + metrics=False, + inspect_arguments=False, + additional_span_processors=[ + BatchSpanProcessor(OTLPSpanExporter(endpoint=endpoint, headers=headers, timeout=5)) + ], + ) + + +def test_agent_pipeline_is_isolated_from_platform_endpoint_and_credentials( + restore_agent_instrumentation, +) -> None: + platform_receiver = _OTLPReceiver() + agent_receiver = _OTLPReceiver() + platform: logfire.Logfire | None = None + agent_observability = None + try: + platform_port = platform_receiver.server.server_address[1] + os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = f"http://127.0.0.1:{platform_port}" + os.environ["OTEL_EXPORTER_OTLP_TRACES_ENDPOINT"] = platform_receiver.endpoint + os.environ["OTEL_EXPORTER_OTLP_METRICS_ENDPOINT"] = f"http://127.0.0.1:{platform_port}/v1/metrics" + os.environ["OTEL_EXPORTER_OTLP_LOGS_ENDPOINT"] = f"http://127.0.0.1:{platform_port}/v1/logs" + os.environ["OTEL_EXPORTER_OTLP_HEADERS"] = "Authorization=platform-test-only" + os.environ["OTEL_RESOURCE_ATTRIBUTES"] = "service.name=platform-test" + os.environ["LOGFIRE_TOKEN"] = "platform-test-only" + os.environ["LOGFIRE_HTTPX_CAPTURE_ALL"] = "true" + environment_snapshot = dict(os.environ) + + platform = _platform_client(platform_receiver.endpoint, {"Authorization": "platform-test-only"}) + settings = ServerSettings( + trajectory_enabled=True, + trajectory_otlp_traces_endpoint=agent_receiver.endpoint, + trajectory_otlp_headers={"Authorization": "agent-test-only"}, + trajectory_service_name="dify-agent-trajectory", + ) + agent_observability = configure_agent_observability(settings) + assert agent_observability is not None + + agent = create_agent(TestModel(custom_output_text="done"), tools=[Tool(_static_tool)]) + assert agent.instrument is False + agent_observability.instrument(agent) + + with platform.span("platform-marker"): + pass + result = agent.run_sync("business-marker-input") + assert result.output == "done" + + assert platform.force_flush(timeout_millis=10000) + assert agent_observability.client.force_flush(timeout_millis=10000) + + platform_names = platform_receiver.span_names() + agent_names = agent_receiver.span_names() + assert "platform-marker" in platform_names + assert "platform-marker" not in agent_names + assert agent_names, "expected agent run spans on the business receiver" + assert not any("pydantic" in scope for scope in platform_receiver.scope_names()) + assert set(platform_receiver.service_names()) == {"platform-test"} + assert set(agent_receiver.service_names()) == {"dify-agent-trajectory"} + assert platform_receiver.authorizations() == ["platform-test-only"] * len(platform_receiver.authorizations()) + assert agent_receiver.authorizations() == ["agent-test-only"] * len(agent_receiver.authorizations()) + assert dict(os.environ) == environment_snapshot + finally: + if agent_observability is not None: + agent_observability.client.shutdown(timeout_millis=5000) + if platform is not None: + platform.shutdown(timeout_millis=5000) + platform_receiver.close() + agent_receiver.close() + + +def test_agent_pipeline_empty_headers_send_no_platform_credentials( + restore_agent_instrumentation, +) -> None: + agent_receiver = _OTLPReceiver() + agent_observability = None + try: + os.environ["OTEL_EXPORTER_OTLP_HEADERS"] = "Authorization=platform-test-only" + settings = ServerSettings( + trajectory_enabled=True, + trajectory_otlp_traces_endpoint=agent_receiver.endpoint, + trajectory_service_name="dify-agent-trajectory", + ) + agent_observability = configure_agent_observability(settings) + assert agent_observability is not None + + agent = create_agent(TestModel(custom_output_text="done"), tools=[]) + agent_observability.instrument(agent) + _ = agent.run_sync("business-marker-input") + assert agent_observability.client.force_flush(timeout_millis=10000) + + assert agent_receiver.received, "expected spans on the business receiver" + assert all(authorization is None for authorization in agent_receiver.authorizations()) + assert os.environ["OTEL_EXPORTER_OTLP_HEADERS"] == "Authorization=platform-test-only" + finally: + if agent_observability is not None: + agent_observability.client.shutdown(timeout_millis=5000) + agent_receiver.close() + + +def test_agent_observability_disabled_creates_no_instance_or_export( + restore_agent_instrumentation, +) -> None: + agent_receiver = _OTLPReceiver() + try: + settings = ServerSettings( + trajectory_enabled=False, + trajectory_otlp_traces_endpoint=agent_receiver.endpoint, + ) + + assert configure_agent_observability(settings) is None + agent = create_agent(TestModel(custom_output_text="done"), tools=[]) + assert agent.instrument is False + _ = agent.run_sync("business-marker-input") + + assert agent_receiver.received == [] + finally: + agent_receiver.close() + + +class _SmokePayload(BaseModel): + text: str + count: int + + +def test_platform_fastapi_instrumentation_omits_parsed_argument_payloads() -> None: + exporter = InMemorySpanExporter() + client = logfire.configure( + local=True, + send_to_logfire=False, + console=False, + metrics=False, + inspect_arguments=False, + additional_span_processors=[SimpleSpanProcessor(exporter)], + ) + app = FastAPI() + + @app.post("/echo") + def echo(payload: _SmokePayload) -> dict[str, str]: + return {"text": payload.text} + + try: + with client.instrument_fastapi( + app, + request_attributes_mapper=observability._platform_request_attributes, + capture_headers=False, + tracer_provider=IsolatedTracerProvider(client, preserve_external_parent=True), + ): + with TestClient(app) as http: + assert http.post("/echo", json={"text": "valid-sentinel-payload", "count": 1}).status_code == 200 + assert http.post("/echo", json={"text": "invalid-sentinel-payload", "count": "x"}).status_code == 422 + assert client.force_flush(timeout_millis=10000) + + spans = exporter.get_finished_spans() + assert spans + serialized = repr([dict(span.attributes or {}) for span in spans]) + serialized_events = repr([dict(event.attributes or {}) for span in spans for event in span.events]) + assert any("/echo" in name for name in [span.name for span in spans]) + assert "http.status_code" in serialized + for sentinel in ("valid-sentinel-payload", "invalid-sentinel-payload"): + assert sentinel not in serialized + assert sentinel not in serialized_events + assert "fastapi.arguments.values" not in serialized + assert "fastapi.arguments.errors" not in serialized + finally: + client.shutdown(timeout_millis=5000) diff --git a/dify-agent/tests/local/dify_agent/server/test_settings.py b/dify-agent/tests/local/dify_agent/server/test_settings.py index 561ac117caa..7dd46e16e06 100644 --- a/dify-agent/tests/local/dify_agent/server/test_settings.py +++ b/dify-agent/tests/local/dify_agent/server/test_settings.py @@ -1,6 +1,7 @@ from __future__ import annotations from pathlib import Path +import os import secrets import pytest @@ -24,6 +25,11 @@ from dify_agent.runtime_backend.openshell import ( ) +@pytest.fixture(autouse=True) +def isolate_settings_dotenv(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + monkeypatch.chdir(tmp_path) + + def _base64url_secret(value: bytes) -> str: import base64 @@ -526,6 +532,268 @@ def test_server_settings_rejects_non_array_shell_redact_patterns(monkeypatch: py _ = settings.get_shell_redact_patterns() +def _isolate_trajectory_env(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + monkeypatch.chdir(tmp_path) + for key in ( + "DIFY_AGENT_TRAJECTORY_ENABLED", + "DIFY_AGENT_TRAJECTORY_OTLP_TRACES_ENDPOINT", + "DIFY_AGENT_TRAJECTORY_OTLP_HEADERS", + "DIFY_AGENT_TRAJECTORY_SERVICE_NAME", + "DIFY_AGENT_TRAJECTORY_INCLUDE_CONTENT", + "DIFY_AGENT_TRAJECTORY_MAX_QUEUE_SIZE", + "DIFY_AGENT_TRAJECTORY_MAX_EXPORT_BATCH_SIZE", + "DIFY_AGENT_TRAJECTORY_SCHEDULE_DELAY_MS", + "DIFY_AGENT_TRAJECTORY_EXPORT_TIMEOUT_MS", + "DIFY_AGENT_TRAJECTORY_TRACE_CONTEXT_MODE", + ): + monkeypatch.delenv(key, raising=False) + + +def test_server_settings_trajectory_defaults_disabled( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + _isolate_trajectory_env(monkeypatch, tmp_path) + + settings = ServerSettings() + + assert settings.trajectory_enabled is False + assert settings.trajectory_otlp_traces_endpoint is None + assert settings.trajectory_otlp_headers == {} + assert settings.trajectory_service_name == "dify-agent-trajectory" + assert settings.trajectory_include_content is False + assert settings.trajectory_max_queue_size == 2048 + assert settings.trajectory_max_export_batch_size == 512 + assert settings.trajectory_schedule_delay_ms == 5000 + assert settings.trajectory_export_timeout_ms == 5000 + assert settings.trajectory_trace_context_mode == "isolated" + + +def test_server_settings_reads_trajectory_opt_in_from_env(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("DIFY_AGENT_TRAJECTORY_ENABLED", "true") + monkeypatch.setenv("DIFY_AGENT_TRAJECTORY_OTLP_TRACES_ENDPOINT", "http://127.0.0.1:4318/v1/traces") + monkeypatch.setenv("DIFY_AGENT_TRAJECTORY_OTLP_HEADERS", '{"Authorization": "Bearer test-only"}') + monkeypatch.setenv("DIFY_AGENT_TRAJECTORY_SERVICE_NAME", "custom-trajectory") + monkeypatch.setenv("DIFY_AGENT_TRAJECTORY_INCLUDE_CONTENT", "true") + + settings = ServerSettings() + + assert settings.trajectory_enabled is True + assert str(settings.trajectory_otlp_traces_endpoint) == "http://127.0.0.1:4318/v1/traces" + assert settings.trajectory_otlp_headers["Authorization"].get_secret_value() == "Bearer test-only" + assert "test-only" not in repr(settings.trajectory_otlp_headers) + assert settings.trajectory_service_name == "custom-trajectory" + assert settings.trajectory_include_content is True + + +def test_server_settings_trajectory_enabled_requires_endpoint( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + _isolate_trajectory_env(monkeypatch, tmp_path) + + with pytest.raises(ValidationError, match="trajectory_otlp_traces_endpoint"): + _ = ServerSettings(trajectory_enabled=True) + + +@pytest.mark.parametrize( + "endpoint", + [ + "http://user:pass@collector.example/v1/traces", + "http://collector.example/v1/traces#frag", + ], +) +def test_server_settings_rejects_trajectory_endpoint_with_credentials_or_fragment( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + endpoint: str, +) -> None: + _isolate_trajectory_env(monkeypatch, tmp_path) + + with pytest.raises(ValidationError, match="credentials or a fragment"): + _ = ServerSettings(trajectory_enabled=True, trajectory_otlp_traces_endpoint=endpoint) + + +def test_server_settings_reads_trajectory_batch_processor_limits_from_env( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + _isolate_trajectory_env(monkeypatch, tmp_path) + monkeypatch.setenv("DIFY_AGENT_TRAJECTORY_MAX_QUEUE_SIZE", "1024") + monkeypatch.setenv("DIFY_AGENT_TRAJECTORY_MAX_EXPORT_BATCH_SIZE", "128") + monkeypatch.setenv("DIFY_AGENT_TRAJECTORY_SCHEDULE_DELAY_MS", "250") + monkeypatch.setenv("DIFY_AGENT_TRAJECTORY_EXPORT_TIMEOUT_MS", "1500") + + settings = ServerSettings() + + assert settings.trajectory_max_queue_size == 1024 + assert settings.trajectory_max_export_batch_size == 128 + assert settings.trajectory_schedule_delay_ms == 250 + assert settings.trajectory_export_timeout_ms == 1500 + + +def test_server_settings_reads_trajectory_batch_processor_limits_from_dotenv( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + _isolate_trajectory_env(monkeypatch, tmp_path) + (tmp_path / ".env").write_text( + "DIFY_AGENT_TRAJECTORY_MAX_QUEUE_SIZE=1024\n" + "DIFY_AGENT_TRAJECTORY_MAX_EXPORT_BATCH_SIZE=128\n" + "DIFY_AGENT_TRAJECTORY_SCHEDULE_DELAY_MS=250\n" + "DIFY_AGENT_TRAJECTORY_EXPORT_TIMEOUT_MS=1500\n" + ) + + settings = ServerSettings() + + assert settings.trajectory_max_queue_size == 1024 + assert settings.trajectory_max_export_batch_size == 128 + assert settings.trajectory_schedule_delay_ms == 250 + assert settings.trajectory_export_timeout_ms == 1500 + + +@pytest.mark.parametrize( + "field", + [ + "trajectory_max_queue_size", + "trajectory_max_export_batch_size", + "trajectory_schedule_delay_ms", + "trajectory_export_timeout_ms", + ], +) +@pytest.mark.parametrize("bad_value", [0, -1]) +def test_server_settings_rejects_non_positive_trajectory_batch_processor_limits( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + field: str, + bad_value: int, +) -> None: + _isolate_trajectory_env(monkeypatch, tmp_path) + + with pytest.raises(ValidationError, match=field): + _ = ServerSettings.model_validate({field: bad_value}) + + +def test_server_settings_rejects_trajectory_batch_size_above_queue_size( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + _isolate_trajectory_env(monkeypatch, tmp_path) + + with pytest.raises(ValidationError, match="must not exceed"): + _ = ServerSettings(trajectory_max_queue_size=1024, trajectory_max_export_batch_size=1025) + + +def test_server_settings_accepts_equal_trajectory_batch_and_queue_size( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + _isolate_trajectory_env(monkeypatch, tmp_path) + + settings = ServerSettings(trajectory_max_queue_size=64, trajectory_max_export_batch_size=64) + + assert settings.trajectory_max_queue_size == 64 + assert settings.trajectory_max_export_batch_size == 64 + + +@pytest.mark.parametrize("mode", ["isolated", "shared"]) +def test_server_settings_reads_trajectory_trace_context_mode_from_env( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + mode: str, +) -> None: + _isolate_trajectory_env(monkeypatch, tmp_path) + monkeypatch.setenv("DIFY_AGENT_TRAJECTORY_TRACE_CONTEXT_MODE", mode) + + assert ServerSettings().trajectory_trace_context_mode == mode + + +def test_server_settings_reads_trajectory_trace_context_mode_from_dotenv( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + _isolate_trajectory_env(monkeypatch, tmp_path) + (tmp_path / ".env").write_text("DIFY_AGENT_TRAJECTORY_TRACE_CONTEXT_MODE=shared\n") + + assert ServerSettings().trajectory_trace_context_mode == "shared" + + +@pytest.mark.parametrize("bad_mode", ["auto", "SHARED", ""]) +def test_server_settings_rejects_unknown_trajectory_trace_context_mode( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + bad_mode: str, +) -> None: + _isolate_trajectory_env(monkeypatch, tmp_path) + + with pytest.raises(ValidationError, match="trajectory_trace_context_mode"): + _ = ServerSettings.model_validate({"trajectory_trace_context_mode": bad_mode}) + + +def test_server_settings_observability_dotenv_snapshot_only_captures_sdk_keys( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + monkeypatch.delenv("OTEL_EXPORTER_OTLP_ENDPOINT", raising=False) + monkeypatch.delenv("LOGFIRE_EMPTY", raising=False) + monkeypatch.delenv("LOGFIRE_TOKEN", raising=False) + dotenv = tmp_path / "otel.env" + dotenv.write_text( + "OTEL_EXPORTER_OTLP_ENDPOINT=http://snap:4318\n" + "LOGFIRE_EMPTY=\n" + "LOGFIRE_TOKEN=dotenv-test-only-secret\n" + "OTEL_UNSET\n" + "DIFY_AGENT_API_TOKEN=not-captured\n" + "UNRELATED=not-captured\n" + ) + + settings = ServerSettings(_env_file=dotenv) + + snapshot = settings.observability_dotenv + assert set(snapshot) == {"OTEL_EXPORTER_OTLP_ENDPOINT", "LOGFIRE_EMPTY", "LOGFIRE_TOKEN"} + assert snapshot["OTEL_EXPORTER_OTLP_ENDPOINT"].get_secret_value() == "http://snap:4318" + assert snapshot["LOGFIRE_EMPTY"].get_secret_value() == "" + assert snapshot["LOGFIRE_TOKEN"].get_secret_value() == "dotenv-test-only-secret" + assert "OTEL_EXPORTER_OTLP_ENDPOINT" not in os.environ + assert "LOGFIRE_TOKEN" not in os.environ + assert "observability_dotenv" not in settings.model_dump() + assert "observability_dotenv" not in settings.model_dump_json() + assert "http://snap:4318" not in repr(settings) + assert "http://snap:4318" not in repr(snapshot) + assert "dotenv-test-only-secret" not in repr(settings) + assert "dotenv-test-only-secret" not in repr(snapshot) + assert "dotenv-test-only-secret" not in settings.model_dump_json() + assert "dotenv-test-only-secret" not in repr(settings.model_dump()) + + +def test_server_settings_observability_dotenv_snapshots_are_independent(tmp_path: Path) -> None: + first = tmp_path / "first.env" + first.write_text("OTEL_EXPORTER_OTLP_ENDPOINT=http://first:4318\n") + second = tmp_path / "second.env" + second.write_text("OTEL_EXPORTER_OTLP_ENDPOINT=http://second:4318\n") + + settings_first = ServerSettings(_env_file=first) + settings_second = ServerSettings(_env_file=second) + settings_none = ServerSettings(_env_file=None) + + assert settings_first.observability_dotenv["OTEL_EXPORTER_OTLP_ENDPOINT"].get_secret_value() == "http://first:4318" + assert ( + settings_second.observability_dotenv["OTEL_EXPORTER_OTLP_ENDPOINT"].get_secret_value() == "http://second:4318" + ) + assert settings_none.observability_dotenv == {} + + +def test_server_settings_observability_dotenv_canonicalizes_names_by_case_sensitivity(tmp_path: Path) -> None: + dotenv = tmp_path / "otel.env" + dotenv.write_text("otel_exporter_otlp_endpoint=http://lower:4318\nLOGFIRE_SERVICE_NAME=upper\n") + + insensitive = ServerSettings(_env_file=dotenv) + assert set(insensitive.observability_dotenv) == {"OTEL_EXPORTER_OTLP_ENDPOINT", "LOGFIRE_SERVICE_NAME"} + + sensitive = ServerSettings(_env_file=dotenv, _case_sensitive=True) + assert set(sensitive.observability_dotenv) == {"LOGFIRE_SERVICE_NAME"} + + @pytest.mark.parametrize( "overrides", [{"runtime_backend": "local"}, {"e2b_api_key": None}, {"e2b_project_id": " "}, {"inner_api_key": None}], diff --git a/docker/envs/core-services/dify-agent.env.example b/docker/envs/core-services/dify-agent.env.example index b5fea249412..9e8ecd52343 100644 --- a/docker/envs/core-services/dify-agent.env.example +++ b/docker/envs/core-services/dify-agent.env.example @@ -29,6 +29,19 @@ DIFY_AGENT_OUTBOUND_HTTP_MAX_CONNECTIONS=100 DIFY_AGENT_OUTBOUND_HTTP_MAX_KEEPALIVE_CONNECTIONS=20 DIFY_AGENT_OUTBOUND_HTTP_KEEPALIVE_EXPIRY=30 +DIFY_AGENT_TRAJECTORY_ENABLED=false +DIFY_AGENT_TRAJECTORY_OTLP_TRACES_ENDPOINT= +DIFY_AGENT_TRAJECTORY_OTLP_HEADERS={} +DIFY_AGENT_TRAJECTORY_SERVICE_NAME=dify-agent-trajectory +DIFY_AGENT_TRAJECTORY_INCLUDE_CONTENT=false +DIFY_AGENT_TRAJECTORY_TRACE_CONTEXT_MODE=isolated +DIFY_AGENT_TRAJECTORY_MAX_QUEUE_SIZE=2048 +DIFY_AGENT_TRAJECTORY_MAX_EXPORT_BATCH_SIZE=512 +DIFY_AGENT_TRAJECTORY_SCHEDULE_DELAY_MS=5000 +DIFY_AGENT_TRAJECTORY_EXPORT_TIMEOUT_MS=5000 + +LOGFIRE_CONSOLE=false + # Leave empty to derive from PLUGIN_DAEMON_URL and PLUGIN_DAEMON_KEY in Docker Compose. DIFY_AGENT_PLUGIN_DAEMON_URL= DIFY_AGENT_PLUGIN_DAEMON_API_KEY=