refactor(tracing): one trace pipeline behind a minimal provider interface

.devcontainer/post_create_command.sh

@@ -7,7 +7,7 @@ cd web && pnpm install
 pipx install uv
 
 echo "alias start-api=\"cd $WORKSPACE_ROOT/api && uv run python -m flask run --host 0.0.0.0 --port=5001 --debug\"" >> ~/.bashrc
-echo "alias start-worker=\"cd $WORKSPACE_ROOT/api && uv run python -m celery -A app.celery worker -P threads -c 1 --loglevel INFO -Q dataset,dataset_summary,priority_dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_publisher,trigger_refresh_executor,retention\"" >> ~/.bashrc
+echo "alias start-worker=\"$WORKSPACE_ROOT/dev/start-worker --pool threads\"" >> ~/.bashrc
 echo "alias start-web=\"cd $WORKSPACE_ROOT/web && pnpm dev\"" >> ~/.bashrc
 echo "alias start-web-prod=\"cd $WORKSPACE_ROOT/web && pnpm build && pnpm start\"" >> ~/.bashrc
 echo "alias start-containers=\"cd $WORKSPACE_ROOT/docker && docker-compose -f docker-compose.middleware.yaml -p dify --env-file middleware.env up -d\"" >> ~/.bashrc

AGENTS.md

@@ -8,4 +8,4 @@ Before starting backend development, read the [architecture](docs/BACKEND_ARCHIT
 
 - Run backend commands through `uv run --project api <command>`.
 - Backend integration tests are CI-only and are not expected to run locally.
-- Keep `docker/.env.example` limited to variables required for a default Docker Compose deployment to start. Put optional and provider-specific settings in the matching `docker/envs/*.env.example` file; `docker/.env` overrides those service-specific env files.
+- Keep `docker/.env.example` limited to settings that require manual user configuration. Omit settings whose built-in defaults work unchanged. Put optional and provider-specific settings in the matching `docker/envs/*.env.example` file; `docker/.env` overrides those service-specific env files.

api/.env.example

@@ -118,10 +118,11 @@ CELERY_BROKER_URL=redis://:difyai123456@localhost:${REDIS_PORT}/1
 CELERY_BACKEND=redis
 
 # Ops trace configuration
-OPS_TRACE_UNIFIED_ENABLED=false
-OPS_TRACE_RETRYABLE_DISPATCH_MAX_RETRIES=780
-OPS_TRACE_RETRYABLE_DISPATCH_DELAY_SECONDS=5
-OPS_TRACE_PARENT_CONTEXT_TTL_SECONDS=3900
+# Keep new workers separate from legacy ops_trace workers during upgrades.
+OPS_TRACE_QUEUE=ops_trace_v2
+OPS_TRACE_EXPORT_MAX_ATTEMPTS=6
+# Daily cleanup of trace payloads an interrupted delivery left in storage.
+ENABLE_TRACE_PAYLOAD_SWEEP_TASK=true
 
 # Database configuration
 DB_TYPE=postgresql

api/.importlinter

@@ -6,6 +6,7 @@ root_packages =
     context
     configs
     controllers
+    domain
     extensions
     factories
     fields
@@ -17,6 +18,26 @@ root_packages =
     services
 include_external_packages = True
 
+[importlinter:contract:domain-boundary]
+name = Domain code depends on no application, persistence, framework, configuration, queue, transport, or legacy kernel code
+type = forbidden
+source_modules =
+    domain
+forbidden_modules =
+    celery
+    configs
+    controllers
+    core
+    enterprise
+    extensions
+    flask
+    models
+    repositories
+    services
+    sqlalchemy
+    tasks
+    werkzeug
+
 [importlinter:contract:backend-layers]
 name = Backend modules follow the transport, application, execution-kernel, and shared-library layers
 type = layers
@@ -1328,3 +1349,81 @@ forbidden_modules =
     sqlalchemy
     tasks
     werkzeug
+
+[importlinter:contract:tracing-application-boundary]
+name = Trace pipeline and Agent trace collection services and contracts are framework and persistence neutral
+type = forbidden
+source_modules =
+    services.tracing.canonical
+    services.tracing.collection
+    services.tracing.collection_ports
+    services.tracing.delivery
+    services.tracing.errors
+    services.tracing.hierarchy
+    services.tracing.provider_configs
+    services.tracing.providers
+    services.tracing.records
+    services.tracing.recovery
+    services.tracing.scopes
+    services.tracing.service
+forbidden_modules =
+    celery
+    configs
+    controllers
+    core
+    enterprise
+    extensions
+    flask
+    models
+    repositories
+    sqlalchemy
+    tasks
+    werkzeug
+    services.tracing.adapters
+    services.tracing.egress
+    services.tracing.registry
+    services.tracing.request_log_protection
+    services.tracing.runtime_bridge
+
+[importlinter:contract:tracing-controller-boundary]
+name = Tracing handlers reach persistence and tracing adapters only through the composition root
+type = forbidden
+source_modules =
+    controllers.console.agent.tracing
+    controllers.console.agent.tracing_runs
+    controllers.console.app.ops_trace
+    controllers.inner_api.agent.tracing
+forbidden_modules =
+    extensions.ext_database
+    models.application_trace
+    models.model
+    repositories
+    sqlalchemy
+    services.tracing.adapters
+    services.tracing.registry
+    services.tracing.runtime_bridge
+    services.tracing.service
+    domain
+# The composition root (extensions.ext_application_services) wires repositories and
+# adapters, so only direct imports are checked here.
+allow_indirect_imports = True
+
+[importlinter:contract:tracing-repository-boundary]
+name = Tracing persistence adapters only depend on the tracing contracts they implement
+type = forbidden
+source_modules =
+    repositories.app.tracing_config_repository
+    repositories.application_trace_repository
+    repositories.trace_record_repository
+    repositories.tracing_store_errors
+forbidden_modules =
+    controllers
+    extensions
+    flask
+    tasks
+    werkzeug
+    services.tracing.adapters
+    services.tracing.registry
+    services.tracing.runtime_bridge
+    services.tracing.service
+allow_indirect_imports = True

api/app_factory.py

@@ -186,7 +186,6 @@ def initialize_extensions(app: DifyApp):
         ext_database,
         ext_enterprise_telemetry,
         ext_fastopenapi,
-        ext_forward_refs,
         ext_hosting_provider,
         ext_import_modules,
         ext_key_provider,
@@ -218,7 +217,6 @@ def initialize_extensions(app: DifyApp):
         ext_warnings,
         ext_import_modules,
         ext_orjson,
-        ext_forward_refs,
         ext_compress,
         ext_code_based_extension,
         ext_database,

api/configs/feature/__init__.py

@@ -13,7 +13,6 @@
     PositiveInt,
     computed_field,
     field_validator,
-    model_validator,
 )
 from pydantic_settings import BaseSettings
 
@@ -1438,6 +1437,11 @@ class MultiModalTransferConfig(BaseSettings):
 
 
 class OpsTraceConfig(BaseSettings):
+    OPS_TRACE_QUEUE: str = Field(
+        default="ops_trace_v2",
+        pattern=r"^[A-Za-z0-9_.-]+$",
+        description="Celery queue for tracing and sandbox metering; isolate new workers during rolling upgrades.",
+    )
     OPS_AGENT_TRACE_ENABLED: bool = Field(
         default=False, description="Enable application-scoped Agent tracing collection and delivery."
     )
@@ -1463,37 +1467,11 @@ class OpsTraceConfig(BaseSettings):
         default=7, description="Retention of trace payloads and encrypted destination snapshots."
     )
 
-    OPS_TRACE_UNIFIED_ENABLED: bool = Field(
-        description="Enable unified ops tracing for providers registered in the unified registry.",
-        default=False,
-    )
-
-    # Include scheduling and export grace after the parent workflow's maximum execution time.
-    # Recommended: max_retries >= ceil((WORKFLOW_MAX_EXECUTION_TIME + grace_seconds) / delay_seconds).
-    OPS_TRACE_RETRYABLE_DISPATCH_MAX_RETRIES: PositiveInt = Field(
-        description="Maximum retry attempts for transient ops trace provider dispatch failures.",
-        default=780,
-    )
-
-    OPS_TRACE_RETRYABLE_DISPATCH_DELAY_SECONDS: PositiveInt = Field(
-        description="Delay in seconds between transient ops trace provider dispatch retry attempts.",
-        default=5,
-    )
-
-    OPS_TRACE_PARENT_CONTEXT_TTL_SECONDS: PositiveInt = Field(
-        description="Retention in seconds for unified tracing parent contexts.",
-        default=3900,
+    OPS_TRACE_EXPORT_MAX_ATTEMPTS: PositiveInt = Field(
+        default=6,
+        description="Attempts to send one trace to its provider when the provider is temporarily unavailable.",
     )
 
-    @model_validator(mode="after")
-    def validate_parent_context_retention(self) -> "OpsTraceConfig":
-        if not self.OPS_TRACE_UNIFIED_ENABLED:
-            return self
-        retry_window = self.OPS_TRACE_RETRYABLE_DISPATCH_MAX_RETRIES * self.OPS_TRACE_RETRYABLE_DISPATCH_DELAY_SECONDS
-        if retry_window > self.OPS_TRACE_PARENT_CONTEXT_TTL_SECONDS:
-            raise ValueError("OPS_TRACE_PARENT_CONTEXT_TTL_SECONDS must cover the retry window")
-        return self
-
 
 class CeleryBeatConfig(BaseSettings):
     CELERY_BEAT_SCHEDULER_TIME: int = Field(
@@ -1503,6 +1481,10 @@ class CeleryBeatConfig(BaseSettings):
 
 
 class CeleryScheduleTasksConfig(BaseSettings):
+    ENABLE_TRACE_PAYLOAD_SWEEP_TASK: bool = Field(
+        description="Enable the daily cleanup of trace payloads that an interrupted delivery left in storage",
+        default=True,
+    )
     ENABLE_APPLICATION_TRACE_RECONCILE_TASK: bool = Field(
         description="Enable periodic application trace recovery and retention cleanup when Agent tracing is enabled",
         default=True,

api/controllers/console/agent/tracing.py

@@ -1,6 +1,7 @@
 """Agent tracing settings address Agent resources, never hidden inline Apps."""
 
 from collections.abc import Callable
+from dataclasses import asdict
 from typing import Literal
 from uuid import UUID
 
@@ -9,7 +10,6 @@
 from pydantic import Field
 from werkzeug.exceptions import NotFound
 
-from configs import dify_config
 from controllers.common.rbac import AgentId, PlainApp, RBACCheck
 from controllers.common.schema import query_params_from_model, register_response_schema_models
 from controllers.console import console_ns
@@ -206,22 +206,22 @@ def delete(self, context: RequestContext, agent_id: UUID):
 
 
 def _provider_capabilities():
-    from services.tracing.provider_dispatch import provider_capabilities
-
     query = TraceProviderQuery.model_validate(request.args.to_dict(flat=True))
     try:
-        capabilities = provider_capabilities(query.tracing_provider)
-    except ValueError as error:
+        capabilities = application_services().tracing.provider_capabilities.get(query.tracing_provider)
+    except AppTracingConfigInvalidProviderError as error:
         raise UnsupportedTracingProviderError() from error
+    except AppTracingConfigProviderUnavailableError as error:
+        raise TracingProviderUnavailableError() from error
     return dump_response(
         TraceProviderCapabilitiesResponse,
-        {**capabilities, "api_collection_enabled": dify_config.OPS_AGENT_TRACE_ENABLED},
+        {**asdict(capabilities.support), "api_collection_enabled": capabilities.api_collection_enabled},
     )
 
 
 @console_ns.route("/agent/<uuid:agent_id>/trace-capabilities")
 class AgentTraceCapabilitiesApi(Resource):
-    @error_responses(UnsupportedTracingProviderError)
+    @error_responses(TracingProviderUnavailableError, UnsupportedTracingProviderError)
     @console_ns.doc("get_agent_trace_capabilities", params=query_params_from_model(TraceProviderQuery))
     @console_ns.response(
         200, "Provider support declarations", console_ns.models[TraceProviderCapabilitiesResponse.__name__]
@@ -234,7 +234,7 @@ def get(self, context: RequestContext, agent_id: UUID):
 
 @console_ns.route("/apps/<uuid:app_id>/trace-capabilities")
 class AppTraceCapabilitiesApi(Resource):
-    @error_responses(UnsupportedTracingProviderError)
+    @error_responses(TracingProviderUnavailableError, UnsupportedTracingProviderError)
     @console_ns.doc("get_app_trace_capabilities", params=query_params_from_model(TraceProviderQuery))
     @console_ns.response(
         200, "Provider support declarations", console_ns.models[TraceProviderCapabilitiesResponse.__name__]

api/controllers/console/agent/tracing_runs.py

@@ -23,9 +23,8 @@
 from libs.helper import dump_response
 from machinery.context import RequestContext
 from models.account import TenantAccountRole
-from repositories.application_trace_repository import TraceCollectionError
 from services.app.agent_app_contracts import AgentAppNotFoundError
-from services.tracing.service import application_tracing_service
+from services.tracing.errors import TraceCollectionError
 
 type TraceSourceKind = Literal["message", "workflow", "node"]
 _source_kind_adapter = TypeAdapter(TraceSourceKind)
@@ -155,7 +154,7 @@ def _agent_app_id(context: RequestContext, agent_id: UUID) -> str:
 def _status(context: RequestContext, app_id: str, kind: TraceSourceKind, source_id: str):
     node_id = _query_node_id()
     result = _run_call(
-        lambda: application_tracing_service().status(
+        lambda: application_services().tracing.agent_traces.status(
             tenant_id=context.active_workspace_id,
             app_id=app_id,
             source_kind=kind,
@@ -170,7 +169,7 @@ def _retry(context: RequestContext, app_id: str, kind: TraceSourceKind, source_i
     node_id = _query_node_id()
     payload = TraceRunRetryPayload.model_validate(console_ns.payload or {})
     result = _run_call(
-        lambda: application_tracing_service().retry(
+        lambda: application_services().tracing.agent_traces.retry(
             tenant_id=context.active_workspace_id,
             app_id=app_id,
             source_kind=kind,

api/controllers/console/app/ops_trace.py

@@ -73,7 +73,12 @@ class TraceAppConfigResponse(ResponseModel):
     id: str | None = None
     app_id: str | None = None
     tracing_provider: str | None = None
-    tracing_config: dict[str, str | None] | None = Field(default=None)
+    tracing_config: dict[str, str | None] | None = Field(
+        default=None, description="Non-secret fields; secrets are write-only and never returned"
+    )
+    configured_secrets: list[str] | None = Field(
+        default=None, description="Secret fields that hold a value; submit them empty to keep that value"
+    )
     is_active: bool | None = None
     created_at: str | None = None
     updated_at: str | None = None
@@ -85,6 +90,7 @@ def from_record(cls, record: AppTracingConfigRecord) -> "TraceAppConfigResponse"
             app_id=record.app_id,
             tracing_provider=record.tracing_provider,
             tracing_config=record.tracing_config,
+            configured_secrets=list(record.configured_secrets),
             is_active=record.is_active,
             created_at=str(record.created_at),
             updated_at=str(record.updated_at),

api/controllers/inner_api/agent/tracing.py

@@ -13,9 +13,14 @@
 from controllers.common.schema import register_response_schema_models, register_schema_models
 from controllers.inner_api import inner_api_ns
 from controllers.inner_api.wraps import agent_inner_api_only
+from extensions.ext_application_services import application_services
 from fields.base import ResponseModel
-from repositories.application_trace_repository import TraceCollectionError
-from services.tracing.service import application_tracing_service
+from services.tracing.errors import (
+    TraceCollectionConflictError,
+    TraceCollectionError,
+    TraceCollectionLimitError,
+    TraceCollectionNotFoundError,
+)
 
 
 class AgentTraceBatchResponse(ResponseModel):
@@ -38,12 +43,17 @@ def post(self):
         request.max_content_length = dify_config.OPS_AGENT_TRACE_BATCH_MAX_BYTES
         try:
             payload = AgentTraceBatch.model_validate(inner_api_ns.payload or {})
-            accepted = application_tracing_service().ingest(payload)
+            accepted = application_services().tracing.agent_traces.ingest(payload)
         except ValidationError as error:
             raise BadRequest("Invalid Agent trace batch") from error
+        except TraceCollectionNotFoundError as error:
+            raise NotFound(error.code) from error
+        except TraceCollectionConflictError as error:
+            raise Conflict(error.code) from error
+        except TraceCollectionLimitError as error:
+            raise RequestEntityTooLarge(error.code) from error
         except TraceCollectionError as error:
-            error_type = {404: NotFound, 409: Conflict, 413: RequestEntityTooLarge}.get(error.status, BadRequest)
-            raise error_type(error.code) from error
+            raise BadRequest(error.code) from error
         return AgentTraceBatchResponse(result="success", batch_id=payload.batch_id, accepted_spans=accepted).model_dump(
             mode="json"
         )