feat(agent): export agent trajectory with a separated tracer (#42651)

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

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`:
 

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

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"]

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",
+]

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:

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:

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"])

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",
+]

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:

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")
 

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)

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()

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:

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:

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
 

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}]

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)

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}],

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=