fix(telemetry): use URL scheme instead of API key for gRPC TLS detection

- Change insecure parameter from API key-based to URL scheme-based detection
- https:// endpoints now correctly use TLS (insecure=False)
- All other endpoints (http://, no scheme) use insecure=True
- Update tests to reflect URL scheme-based logic
- Remove incorrect documentation claiming API key controls TLS

api/README.md

@@ -42,7 +42,7 @@ The scripts resolve paths relative to their location, so you can run them from a
 
 1. Set up your application by visiting `http://localhost:3000`.
 
-1. Start the worker service (async and scheduler tasks, runs from `api`).
+1. Optional: start the worker service (async tasks, runs from `api`).
 
    ```bash
    ./dev/start-worker
@@ -54,6 +54,87 @@ The scripts resolve paths relative to their location, so you can run them from a
    ./dev/start-beat
    ```
 
+### Manual commands
+
+<details>
+<summary>Show manual setup and run steps</summary>
+
+These commands assume you start from the repository root.
+
+1. Start the docker-compose stack.
+
+   The backend requires middleware, including PostgreSQL, Redis, and Weaviate, which can be started together using `docker-compose`.
+
+   ```bash
+   cp docker/middleware.env.example docker/middleware.env
+   # Use mysql or another vector database profile if you are not using postgres/weaviate.
+   docker compose -f docker/docker-compose.middleware.yaml --profile postgresql --profile weaviate -p dify up -d
+   ```
+
+1. Copy env files.
+
+   ```bash
+   cp api/.env.example api/.env
+   cp web/.env.example web/.env.local
+   ```
+
+1. Install UV if needed.
+
+   ```bash
+   pip install uv
+   # Or on macOS
+   brew install uv
+   ```
+
+1. Install API dependencies.
+
+   ```bash
+   cd api
+   uv sync --group dev
+   ```
+
+1. Install web dependencies.
+
+   ```bash
+   cd web
+   pnpm install
+   cd ..
+   ```
+
+1. Start backend (runs migrations first, in a new terminal).
+
+   ```bash
+   cd api
+   uv run flask db upgrade
+   uv run flask run --host 0.0.0.0 --port=5001 --debug
+   ```
+
+1. Start Dify [web](../web) service (in a new terminal).
+
+   ```bash
+   cd web
+   pnpm dev:inspect
+   ```
+
+1. Set up your application by visiting `http://localhost:3000`.
+
+1. Optional: start the worker service (async tasks, in a new terminal).
+
+   ```bash
+   cd api
+   # Note: enterprise_telemetry queue is only used in Enterprise Edition
+   uv run celery -A app.celery worker -P threads -c 2 --loglevel INFO -Q dataset,priority_dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_executor,retention,enterprise_telemetry
+   ```
+
+1. Optional: start Celery Beat (scheduled tasks, in a new terminal).
+
+   ```bash
+   cd api
+   uv run celery -A app.celery beat
+   ```
+
+</details>
+
 ### Environment notes
 
 > [!IMPORTANT]

api/app_factory.py

@@ -81,6 +81,7 @@ def initialize_extensions(app: DifyApp):
         ext_commands,
         ext_compress,
         ext_database,
+        ext_enterprise_telemetry,
         ext_fastopenapi,
         ext_forward_refs,
         ext_hosting_provider,
@@ -131,6 +132,7 @@ def initialize_extensions(app: DifyApp):
         ext_commands,
         ext_fastopenapi,
         ext_otel,
+        ext_enterprise_telemetry,
         ext_request_logging,
         ext_session_factory,
     ]

api/configs/app_config.py

@@ -8,7 +8,7 @@ from pydantic_settings import BaseSettings, PydanticBaseSettingsSource, Settings
 from libs.file_utils import search_file_upwards
 
 from .deploy import DeploymentConfig
-from .enterprise import EnterpriseFeatureConfig
+from .enterprise import EnterpriseFeatureConfig, EnterpriseTelemetryConfig
 from .extra import ExtraServiceConfig
 from .feature import FeatureConfig
 from .middleware import MiddlewareConfig
@@ -73,6 +73,8 @@ class DifyConfig(
     # Enterprise feature configs
     # **Before using, please contact business@dify.ai by email to inquire about licensing matters.**
     EnterpriseFeatureConfig,
+    # Enterprise telemetry configs
+    EnterpriseTelemetryConfig,
 ):
     model_config = SettingsConfigDict(
         # read from dotenv format config file

api/configs/enterprise/__init__.py

@@ -18,3 +18,49 @@ class EnterpriseFeatureConfig(BaseSettings):
         description="Allow customization of the enterprise logo.",
         default=False,
     )
+
+
+class EnterpriseTelemetryConfig(BaseSettings):
+    """
+    Configuration for enterprise telemetry.
+    """
+
+    ENTERPRISE_TELEMETRY_ENABLED: bool = Field(
+        description="Enable enterprise telemetry collection (also requires ENTERPRISE_ENABLED=true).",
+        default=False,
+    )
+
+    ENTERPRISE_OTLP_ENDPOINT: str = Field(
+        description="Enterprise OTEL collector endpoint.",
+        default="",
+    )
+
+    ENTERPRISE_OTLP_HEADERS: str = Field(
+        description="Auth headers for OTLP export (key=value,key2=value2).",
+        default="",
+    )
+
+    ENTERPRISE_OTLP_PROTOCOL: str = Field(
+        description="OTLP protocol: 'http' or 'grpc' (default: http).",
+        default="http",
+    )
+
+    ENTERPRISE_OTLP_API_KEY: str = Field(
+        description="Bearer token for enterprise OTLP export authentication.",
+        default="",
+    )
+
+    ENTERPRISE_INCLUDE_CONTENT: bool = Field(
+        description="Include input/output content in traces (privacy toggle).",
+        default=True,
+    )
+
+    ENTERPRISE_SERVICE_NAME: str = Field(
+        description="Service name for OTEL resource.",
+        default="dify",
+    )
+
+    ENTERPRISE_OTEL_SAMPLING_RATE: float = Field(
+        description="Sampling rate for enterprise traces (0.0 to 1.0, default 1.0 = 100%).",
+        default=1.0,
+    )

api/core/app/apps/advanced_chat/generate_task_pipeline.py

@@ -67,6 +67,8 @@ from core.model_runtime.entities.llm_entities import LLMUsage
 from core.model_runtime.utils.encoders import jsonable_encoder
 from core.ops.ops_trace_manager import TraceQueueManager
 from core.repositories.human_input_repository import HumanInputFormRepositoryImpl
+from core.telemetry import TelemetryContext, TelemetryEvent, TraceTaskName
+from core.telemetry import emit as telemetry_emit
 from core.workflow.entities.pause_reason import HumanInputRequired
 from core.workflow.enums import WorkflowExecutionStatus
 from core.workflow.nodes import NodeType
@@ -622,8 +624,7 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
             )
 
             with self._database_session() as session:
-                # Save message
-                self._save_message(session=session, graph_runtime_state=resolved_state)
+                self._save_message(session=session, graph_runtime_state=resolved_state, trace_manager=trace_manager)
 
             yield workflow_finish_resp
         elif event.stopped_by in (
@@ -632,8 +633,7 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         ):
             # When hitting input-moderation or annotation-reply, the workflow will not start
             with self._database_session() as session:
-                # Save message
-                self._save_message(session=session)
+                self._save_message(session=session, trace_manager=trace_manager)
 
         yield self._message_end_to_stream_response()
 
@@ -642,6 +642,7 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         event: QueueAdvancedChatMessageEndEvent,
         *,
         graph_runtime_state: GraphRuntimeState | None = None,
+        trace_manager: TraceQueueManager | None = None,
         **kwargs,
     ) -> Generator[StreamResponse, None, None]:
         """Handle advanced chat message end events."""
@@ -660,7 +661,7 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         # Save message unless it has already been persisted on pause.
         if not self._message_saved_on_pause:
             with self._database_session() as session:
-                self._save_message(session=session, graph_runtime_state=resolved_state)
+                self._save_message(session=session, graph_runtime_state=resolved_state, trace_manager=trace_manager)
 
         yield self._message_end_to_stream_response()
 
@@ -877,7 +878,13 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         if self._conversation_name_generate_thread:
             logger.debug("Conversation name generation running as daemon thread")
 
-    def _save_message(self, *, session: Session, graph_runtime_state: GraphRuntimeState | None = None):
+    def _save_message(
+        self,
+        *,
+        session: Session,
+        graph_runtime_state: GraphRuntimeState | None = None,
+        trace_manager: TraceQueueManager | None = None,
+    ):
         message = self._get_message(session=session)
         if message is None:
             return
@@ -938,6 +945,22 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         ]
         session.add_all(message_files)
 
+        if trace_manager:
+            telemetry_emit(
+                TelemetryEvent(
+                    name=TraceTaskName.MESSAGE_TRACE,
+                    context=TelemetryContext(
+                        tenant_id=self._application_generate_entity.app_config.tenant_id,
+                        app_id=self._application_generate_entity.app_config.app_id,
+                    ),
+                    payload={
+                        "conversation_id": str(message.conversation_id),
+                        "message_id": str(message.id),
+                    },
+                ),
+                trace_manager=trace_manager,
+            )
+
     def _seed_graph_runtime_state_from_queue_manager(self) -> None:
         """Bootstrap the cached runtime state from the queue manager when present."""
         candidate = self._base_task_pipeline.queue_manager.graph_runtime_state

api/core/app/apps/workflow/app_generator.py

@@ -159,9 +159,12 @@ class WorkflowAppGenerator(BaseAppGenerator):
 
         inputs: Mapping[str, Any] = args["inputs"]
 
-        extras = {
+        extras: dict[str, Any] = {
             **extract_external_trace_id_from_args(args),
         }
+        parent_trace_context = args.get("_parent_trace_context")
+        if parent_trace_context:
+            extras["parent_trace_context"] = parent_trace_context
         workflow_run_id = str(workflow_run_id or uuid.uuid4())
         # FIXME (Yeuoly): we need to remove the SKIP_PREPARE_USER_INPUTS_KEY from the args
         # trigger shouldn't prepare user inputs

api/core/app/task_pipeline/easy_ui_based_generate_task_pipeline.py

@@ -52,10 +52,11 @@ from core.model_runtime.entities.message_entities import (
     TextPromptMessageContent,
 )
 from core.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
-from core.ops.entities.trace_entity import TraceTaskName
-from core.ops.ops_trace_manager import TraceQueueManager, TraceTask
+from core.ops.ops_trace_manager import TraceQueueManager
 from core.prompt.utils.prompt_message_util import PromptMessageUtil
 from core.prompt.utils.prompt_template_parser import PromptTemplateParser
+from core.telemetry import TelemetryContext, TelemetryEvent, TraceTaskName
+from core.telemetry import emit as telemetry_emit
 from core.tools.signature import sign_tool_file
 from core.workflow.file import helpers as file_helpers
 from core.workflow.file.enums import FileTransferMethod
@@ -412,10 +413,19 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
         message.message_metadata = self._task_state.metadata.model_dump_json()
 
         if trace_manager:
-            trace_manager.add_trace_task(
-                TraceTask(
-                    TraceTaskName.MESSAGE_TRACE, conversation_id=self._conversation_id, message_id=self._message_id
-                )
+            telemetry_emit(
+                TelemetryEvent(
+                    name=TraceTaskName.MESSAGE_TRACE,
+                    context=TelemetryContext(
+                        tenant_id=self._application_generate_entity.app_config.tenant_id,
+                        app_id=self._application_generate_entity.app_config.app_id,
+                    ),
+                    payload={
+                        "conversation_id": self._conversation_id,
+                        "message_id": self._message_id,
+                    },
+                ),
+                trace_manager=trace_manager,
             )
 
         message_was_created.send(

api/core/llm_generator/llm_generator.py

@@ -27,10 +27,10 @@ from core.model_runtime.entities.llm_entities import LLMResult
 from core.model_runtime.entities.message_entities import PromptMessage, SystemPromptMessage, UserPromptMessage
 from core.model_runtime.entities.model_entities import ModelType
 from core.model_runtime.errors.invoke import InvokeAuthorizationError, InvokeError
-from core.ops.entities.trace_entity import TraceTaskName
-from core.ops.ops_trace_manager import TraceQueueManager, TraceTask
 from core.ops.utils import measure_time
 from core.prompt.utils.prompt_template_parser import PromptTemplateParser
+from core.telemetry import TelemetryContext, TelemetryEvent, TraceTaskName
+from core.telemetry import emit as telemetry_emit
 from core.workflow.entities.workflow_node_execution import WorkflowNodeExecutionMetadataKey
 from extensions.ext_database import db
 from extensions.ext_storage import storage
@@ -95,16 +95,17 @@ class LLMGenerator:
         if len(name) > 75:
             name = name[:75] + "..."
 
-        # get tracing instance
-        trace_manager = TraceQueueManager(app_id=app_id)
-        trace_manager.add_trace_task(
-            TraceTask(
-                TraceTaskName.GENERATE_NAME_TRACE,
-                conversation_id=conversation_id,
-                generate_conversation_name=name,
-                inputs=prompt,
-                timer=timer,
-                tenant_id=tenant_id,
+        telemetry_emit(
+            TelemetryEvent(
+                name=TraceTaskName.GENERATE_NAME_TRACE,
+                context=TelemetryContext(tenant_id=tenant_id, app_id=app_id),
+                payload={
+                    "conversation_id": conversation_id,
+                    "generate_conversation_name": name,
+                    "inputs": prompt,
+                    "timer": timer,
+                    "tenant_id": tenant_id,
+                },
             )
         )
 

api/core/ops/entities/trace_entity.py

@@ -9,8 +9,8 @@ from pydantic import BaseModel, ConfigDict, field_serializer, field_validator
 class BaseTraceInfo(BaseModel):
     message_id: str | None = None
     message_data: Any | None = None
-    inputs: Union[str, dict[str, Any], list] | None = None
-    outputs: Union[str, dict[str, Any], list] | None = None
+    inputs: Union[str, dict[str, Any], list[Any]] | None = None
+    outputs: Union[str, dict[str, Any], list[Any]] | None = None
     start_time: datetime | None = None
     end_time: datetime | None = None
     metadata: dict[str, Any]
@@ -18,7 +18,7 @@ class BaseTraceInfo(BaseModel):
 
     @field_validator("inputs", "outputs")
     @classmethod
-    def ensure_type(cls, v):
+    def ensure_type(cls, v: str | dict[str, Any] | list[Any] | None) -> str | dict[str, Any] | list[Any] | None:
         if v is None:
             return None
         if isinstance(v, str | dict | list):
@@ -27,6 +27,26 @@ class BaseTraceInfo(BaseModel):
 
     model_config = ConfigDict(protected_namespaces=())
 
+    @property
+    def resolved_trace_id(self) -> str | None:
+        """Get trace_id with intelligent fallback.
+
+        Priority:
+        1. External trace_id (from X-Trace-Id header)
+        2. workflow_run_id (if this trace type has it)
+        3. message_id (as final fallback)
+        """
+        if self.trace_id:
+            return self.trace_id
+
+        # Try workflow_run_id (only exists on workflow-related traces)
+        workflow_run_id = getattr(self, "workflow_run_id", None)
+        if workflow_run_id:
+            return workflow_run_id
+
+        # Final fallback to message_id
+        return str(self.message_id) if self.message_id else None
+
     @field_serializer("start_time", "end_time")
     def serialize_datetime(self, dt: datetime | None) -> str | None:
         if dt is None:
@@ -48,7 +68,10 @@ class WorkflowTraceInfo(BaseTraceInfo):
     workflow_run_version: str
     error: str | None = None
     total_tokens: int
+    prompt_tokens: int | None = None
+    completion_tokens: int | None = None
     file_list: list[str]
+    invoked_by: str | None = None
     query: str
     metadata: dict[str, Any]
 
@@ -59,7 +82,7 @@ class MessageTraceInfo(BaseTraceInfo):
     answer_tokens: int
     total_tokens: int
     error: str | None = None
-    file_list: Union[str, dict[str, Any], list] | None = None
+    file_list: Union[str, dict[str, Any], list[Any]] | None = None
     message_file_data: Any | None = None
     conversation_mode: str
     gen_ai_server_time_to_first_token: float | None = None
@@ -106,7 +129,7 @@ class ToolTraceInfo(BaseTraceInfo):
     tool_config: dict[str, Any]
     time_cost: Union[int, float]
     tool_parameters: dict[str, Any]
-    file_url: Union[str, None, list] = None
+    file_url: Union[str, None, list[str]] = None
 
 
 class GenerateNameTraceInfo(BaseTraceInfo):
@@ -114,6 +137,79 @@ class GenerateNameTraceInfo(BaseTraceInfo):
     tenant_id: str
 
 
+class PromptGenerationTraceInfo(BaseTraceInfo):
+    """Trace information for prompt generation operations (rule-generate, code-generate, etc.)."""
+
+    tenant_id: str
+    user_id: str
+    app_id: str | None = None
+
+    operation_type: str
+    instruction: str
+
+    prompt_tokens: int
+    completion_tokens: int
+    total_tokens: int
+
+    model_provider: str
+    model_name: str
+
+    latency: float
+
+    total_price: float | None = None
+    currency: str | None = None
+
+    error: str | None = None
+
+    model_config = ConfigDict(protected_namespaces=())
+
+
+class WorkflowNodeTraceInfo(BaseTraceInfo):
+    workflow_id: str
+    workflow_run_id: str
+    tenant_id: str
+    node_execution_id: str
+    node_id: str
+    node_type: str
+    title: str
+
+    status: str
+    error: str | None = None
+    elapsed_time: float
+
+    index: int
+    predecessor_node_id: str | None = None
+
+    total_tokens: int = 0
+    total_price: float = 0.0
+    currency: str | None = None
+
+    model_provider: str | None = None
+    model_name: str | None = None
+    prompt_tokens: int | None = None
+    completion_tokens: int | None = None
+
+    tool_name: str | None = None
+
+    iteration_id: str | None = None
+    iteration_index: int | None = None
+    loop_id: str | None = None
+    loop_index: int | None = None
+    parallel_id: str | None = None
+
+    node_inputs: Mapping[str, Any] | None = None
+    node_outputs: Mapping[str, Any] | None = None
+    process_data: Mapping[str, Any] | None = None
+
+    invoked_by: str | None = None
+
+    model_config = ConfigDict(protected_namespaces=())
+
+
+class DraftNodeExecutionTrace(WorkflowNodeTraceInfo):
+    pass
+
+
 class TaskData(BaseModel):
     app_id: str
     trace_info_type: str
@@ -128,11 +224,31 @@ trace_info_info_map = {
     "DatasetRetrievalTraceInfo": DatasetRetrievalTraceInfo,
     "ToolTraceInfo": ToolTraceInfo,
     "GenerateNameTraceInfo": GenerateNameTraceInfo,
+    "PromptGenerationTraceInfo": PromptGenerationTraceInfo,
+    "WorkflowNodeTraceInfo": WorkflowNodeTraceInfo,
+    "DraftNodeExecutionTrace": DraftNodeExecutionTrace,
 }
 
 
+class OperationType(StrEnum):
+    """Operation type for token metric labels.
+
+    Used as a metric attribute on ``dify.tokens.input`` / ``dify.tokens.output``
+    counters so consumers can break down token usage by operation.
+    """
+
+    WORKFLOW = "workflow"
+    NODE_EXECUTION = "node_execution"
+    MESSAGE = "message"
+    RULE_GENERATE = "rule_generate"
+    CODE_GENERATE = "code_generate"
+    STRUCTURED_OUTPUT = "structured_output"
+    INSTRUCTION_MODIFY = "instruction_modify"
+
+
 class TraceTaskName(StrEnum):
     CONVERSATION_TRACE = "conversation"
+    DRAFT_NODE_EXECUTION_TRACE = "draft_node_execution"
     WORKFLOW_TRACE = "workflow"
     MESSAGE_TRACE = "message"
     MODERATION_TRACE = "moderation"
@@ -140,4 +256,6 @@ class TraceTaskName(StrEnum):
     DATASET_RETRIEVAL_TRACE = "dataset_retrieval"
     TOOL_TRACE = "tool"
     GENERATE_NAME_TRACE = "generate_conversation_name"
+    PROMPT_GENERATION_TRACE = "prompt_generation"
+    NODE_EXECUTION_TRACE = "node_execution"
     DATASOURCE_TRACE = "datasource"

api/core/ops/ops_trace_manager.py

@@ -18,19 +18,25 @@ from core.helper.encrypter import batch_decrypt_token, encrypt_token, obfuscated
 from core.ops.entities.config_entity import OPS_FILE_PATH, TracingProviderEnum
 from core.ops.entities.trace_entity import (
     DatasetRetrievalTraceInfo,
+    DraftNodeExecutionTrace,
     GenerateNameTraceInfo,
     MessageTraceInfo,
     ModerationTraceInfo,
+    PromptGenerationTraceInfo,
     SuggestedQuestionTraceInfo,
     TaskData,
     ToolTraceInfo,
     TraceTaskName,
+    WorkflowNodeTraceInfo,
     WorkflowTraceInfo,
 )
 from core.ops.utils import get_message_data
 from extensions.ext_storage import storage
+from models.account import Tenant
+from models.dataset import Dataset
 from models.engine import db
 from models.model import App, AppModelConfig, Conversation, Message, MessageFile, TraceAppConfig
+from models.tools import ApiToolProvider, BuiltinToolProvider, MCPToolProvider, WorkflowToolProvider
 from models.workflow import WorkflowAppLog
 from tasks.ops_trace_task import process_trace_tasks
 
@@ -40,6 +46,44 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 
+def _lookup_app_and_workspace_names(app_id: str | None, tenant_id: str | None) -> tuple[str, str]:
+    """Return (app_name, workspace_name) for the given IDs. Falls back to empty strings."""
+    app_name = ""
+    workspace_name = ""
+    if not app_id and not tenant_id:
+        return app_name, workspace_name
+    with Session(db.engine) as session:
+        if app_id:
+            name = session.scalar(select(App.name).where(App.id == app_id))
+            if name:
+                app_name = name
+        if tenant_id:
+            name = session.scalar(select(Tenant.name).where(Tenant.id == tenant_id))
+            if name:
+                workspace_name = name
+    return app_name, workspace_name
+
+
+_PROVIDER_TYPE_TO_MODEL: dict[str, type] = {
+    "builtin": BuiltinToolProvider,
+    "plugin": BuiltinToolProvider,
+    "api": ApiToolProvider,
+    "workflow": WorkflowToolProvider,
+    "mcp": MCPToolProvider,
+}
+
+
+def _lookup_credential_name(credential_id: str | None, provider_type: str | None) -> str:
+    if not credential_id:
+        return ""
+    model_cls = _PROVIDER_TYPE_TO_MODEL.get(provider_type or "")
+    if not model_cls:
+        return ""
+    with Session(db.engine) as session:
+        name = session.scalar(select(model_cls.name).where(model_cls.id == credential_id))
+        return str(name) if name else ""
+
+
 class OpsTraceProviderConfigMap(collections.UserDict[str, dict[str, Any]]):
     def __getitem__(self, key: str) -> dict[str, Any]:
         match key:
@@ -314,6 +358,10 @@ class OpsTraceManager:
         if app_id is None:
             return None
 
+        # Handle storage_id format (tenant-{uuid}) - not a real app_id
+        if isinstance(app_id, str) and app_id.startswith("tenant-"):
+            return None
+
         app: App | None = db.session.query(App).where(App.id == app_id).first()
 
         if app is None:
@@ -478,6 +526,56 @@ class TraceTask:
                     cls._workflow_run_repo = DifyAPIRepositoryFactory.create_api_workflow_run_repository(session_maker)
         return cls._workflow_run_repo
 
+    @classmethod
+    def _get_user_id_from_metadata(cls, metadata: dict[str, Any]) -> str:
+        """Extract user ID from metadata, prioritizing end_user over account.
+
+        Returns the actual user ID (end_user or account) who invoked the workflow,
+        regardless of invoke_from context.
+        """
+        # Priority 1: End user (external users via API/WebApp)
+        if user_id := metadata.get("from_end_user_id"):
+            return f"end_user:{user_id}"
+
+        # Priority 2: Account user (internal users via console/debugger)
+        if user_id := metadata.get("from_account_id"):
+            return f"account:{user_id}"
+
+        # Priority 3: User (internal users via console/debugger)
+        if user_id := metadata.get("user_id"):
+            return f"user:{user_id}"
+
+        return "anonymous"
+
+    @classmethod
+    def _calculate_workflow_token_split(cls, workflow_run_id: str, tenant_id: str) -> tuple[int, int]:
+        from core.workflow.enums import WorkflowNodeExecutionMetadataKey
+        from models.workflow import WorkflowNodeExecutionModel
+
+        with Session(db.engine) as session:
+            node_executions = session.scalars(
+                select(WorkflowNodeExecutionModel).where(
+                    WorkflowNodeExecutionModel.tenant_id == tenant_id,
+                    WorkflowNodeExecutionModel.workflow_run_id == workflow_run_id,
+                )
+            ).all()
+
+            total_prompt = 0
+            total_completion = 0
+
+            for node_exec in node_executions:
+                metadata = node_exec.execution_metadata_dict
+
+                prompt = metadata.get(WorkflowNodeExecutionMetadataKey.PROMPT_TOKENS)
+                if prompt is not None:
+                    total_prompt += prompt
+
+                completion = metadata.get(WorkflowNodeExecutionMetadataKey.COMPLETION_TOKENS)
+                if completion is not None:
+                    total_completion += completion
+
+            return (total_prompt, total_completion)
+
     def __init__(
         self,
         trace_type: Any,
@@ -498,6 +596,8 @@ class TraceTask:
         self.app_id = None
         self.trace_id = None
         self.kwargs = kwargs
+        if user_id is not None and "user_id" not in self.kwargs:
+            self.kwargs["user_id"] = user_id
         external_trace_id = kwargs.get("external_trace_id")
         if external_trace_id:
             self.trace_id = external_trace_id
@@ -511,7 +611,7 @@ class TraceTask:
             TraceTaskName.WORKFLOW_TRACE: lambda: self.workflow_trace(
                 workflow_run_id=self.workflow_run_id, conversation_id=self.conversation_id, user_id=self.user_id
             ),
-            TraceTaskName.MESSAGE_TRACE: lambda: self.message_trace(message_id=self.message_id),
+            TraceTaskName.MESSAGE_TRACE: lambda: self.message_trace(message_id=self.message_id, **self.kwargs),
             TraceTaskName.MODERATION_TRACE: lambda: self.moderation_trace(
                 message_id=self.message_id, timer=self.timer, **self.kwargs
             ),
@@ -527,6 +627,9 @@ class TraceTask:
             TraceTaskName.GENERATE_NAME_TRACE: lambda: self.generate_name_trace(
                 conversation_id=self.conversation_id, timer=self.timer, **self.kwargs
             ),
+            TraceTaskName.PROMPT_GENERATION_TRACE: lambda: self.prompt_generation_trace(**self.kwargs),
+            TraceTaskName.NODE_EXECUTION_TRACE: lambda: self.node_execution_trace(**self.kwargs),
+            TraceTaskName.DRAFT_NODE_EXECUTION_TRACE: lambda: self.draft_node_execution_trace(**self.kwargs),
         }
 
         return preprocess_map.get(self.trace_type, lambda: None)()
@@ -552,6 +655,9 @@ class TraceTask:
 
         workflow_id = workflow_run.workflow_id
         tenant_id = workflow_run.tenant_id
+        prompt_tokens, completion_tokens = self._calculate_workflow_token_split(
+            workflow_run_id=workflow_run_id, tenant_id=tenant_id
+        )
         workflow_run_id = workflow_run.id
         workflow_run_elapsed_time = workflow_run.elapsed_time
         workflow_run_status = workflow_run.status
@@ -582,7 +688,9 @@ class TraceTask:
                 )
                 message_id = session.scalar(message_data_stmt)
 
-        metadata = {
+        app_name, workspace_name = _lookup_app_and_workspace_names(workflow_run.app_id, tenant_id)
+
+        metadata: dict[str, Any] = {
             "workflow_id": workflow_id,
             "conversation_id": conversation_id,
             "workflow_run_id": workflow_run_id,
@@ -590,12 +698,17 @@ class TraceTask:
             "elapsed_time": workflow_run_elapsed_time,
             "status": workflow_run_status,
             "version": workflow_run_version,
+            "app_name": app_name,
+            "workspace_name": workspace_name,
             "total_tokens": total_tokens,
             "file_list": file_list,
             "triggered_from": workflow_run.triggered_from,
             "user_id": user_id,
             "app_id": workflow_run.app_id,
         }
+        parent_trace_context = self.kwargs.get("parent_trace_context")
+        if parent_trace_context:
+            metadata["parent_trace_context"] = parent_trace_context
 
         workflow_trace_info = WorkflowTraceInfo(
             trace_id=self.trace_id,
@@ -611,7 +724,10 @@ class TraceTask:
             workflow_run_version=workflow_run_version,
             error=error,
             total_tokens=total_tokens,
+            prompt_tokens=prompt_tokens,
+            completion_tokens=completion_tokens,
             file_list=file_list,
+            invoked_by=self._get_user_id_from_metadata(metadata),
             query=query,
             metadata=metadata,
             workflow_app_log_id=workflow_app_log_id,
@@ -621,12 +737,19 @@ class TraceTask:
         )
         return workflow_trace_info
 
-    def message_trace(self, message_id: str | None):
+    def message_trace(self, message_id: str | None, **kwargs):
         if not message_id:
             return {}
         message_data = get_message_data(message_id)
         if not message_data:
             return {}
+        tenant_id = ""
+        with Session(db.engine) as session:
+            tid = session.scalar(select(App.tenant_id).where(App.id == message_data.app_id))
+            if tid:
+                tenant_id = str(tid)
+
+        app_name, workspace_name = _lookup_app_and_workspace_names(message_data.app_id, tenant_id)
         conversation_mode_stmt = select(Conversation.mode).where(Conversation.id == message_data.conversation_id)
         conversation_mode = db.session.scalars(conversation_mode_stmt).all()
         if not conversation_mode or len(conversation_mode) == 0:
@@ -655,7 +778,14 @@ class TraceTask:
             "workflow_run_id": message_data.workflow_run_id,
             "from_source": message_data.from_source,
             "message_id": message_id,
+            "tenant_id": tenant_id,
+            "app_id": message_data.app_id,
+            "user_id": message_data.from_end_user_id or message_data.from_account_id,
+            "app_name": app_name,
+            "workspace_name": workspace_name,
         }
+        if node_execution_id := kwargs.get("node_execution_id"):
+            metadata["node_execution_id"] = node_execution_id
 
         message_tokens = message_data.message_tokens
 
@@ -697,6 +827,8 @@ class TraceTask:
             "preset_response": moderation_result.preset_response,
             "query": moderation_result.query,
         }
+        if node_execution_id := kwargs.get("node_execution_id"):
+            metadata["node_execution_id"] = node_execution_id
 
         # get workflow_app_log_id
         workflow_app_log_id = None
@@ -738,6 +870,8 @@ class TraceTask:
             "workflow_run_id": message_data.workflow_run_id,
             "from_source": message_data.from_source,
         }
+        if node_execution_id := kwargs.get("node_execution_id"):
+            metadata["node_execution_id"] = node_execution_id
 
         # get workflow_app_log_id
         workflow_app_log_id = None
@@ -777,6 +911,47 @@ class TraceTask:
         if not message_data:
             return {}
 
+        tenant_id = ""
+        with Session(db.engine) as session:
+            tid = session.scalar(select(App.tenant_id).where(App.id == message_data.app_id))
+            if tid:
+                tenant_id = str(tid)
+
+        app_name, workspace_name = _lookup_app_and_workspace_names(message_data.app_id, tenant_id)
+
+        doc_list = [doc.model_dump() for doc in documents] if documents else []
+        dataset_ids: set[str] = set()
+        for doc in doc_list:
+            doc_meta = doc.get("metadata") or {}
+            did = doc_meta.get("dataset_id")
+            if did:
+                dataset_ids.add(did)
+
+        embedding_models: dict[str, dict[str, str]] = {}
+        if dataset_ids:
+            with Session(db.engine) as session:
+                rows = session.execute(
+                    select(Dataset.id, Dataset.embedding_model, Dataset.embedding_model_provider).where(
+                        Dataset.id.in_(list(dataset_ids))
+                    )
+                ).all()
+                for row in rows:
+                    embedding_models[str(row[0])] = {
+                        "embedding_model": row[1] or "",
+                        "embedding_model_provider": row[2] or "",
+                    }
+
+        # Extract rerank model info from retrieval_model kwargs
+        rerank_model_provider = ""
+        rerank_model_name = ""
+        if "retrieval_model" in kwargs:
+            retrieval_model = kwargs["retrieval_model"]
+            if isinstance(retrieval_model, dict):
+                reranking_model = retrieval_model.get("reranking_model")
+                if isinstance(reranking_model, dict):
+                    rerank_model_provider = reranking_model.get("reranking_provider_name", "")
+                    rerank_model_name = reranking_model.get("reranking_model_name", "")
+
         metadata = {
             "message_id": message_id,
             "ls_provider": message_data.model_provider,
@@ -787,13 +962,23 @@ class TraceTask:
             "agent_based": message_data.agent_based,
             "workflow_run_id": message_data.workflow_run_id,
             "from_source": message_data.from_source,
+            "tenant_id": tenant_id,
+            "app_id": message_data.app_id,
+            "user_id": message_data.from_end_user_id or message_data.from_account_id,
+            "app_name": app_name,
+            "workspace_name": workspace_name,
+            "embedding_models": embedding_models,
+            "rerank_model_provider": rerank_model_provider,
+            "rerank_model_name": rerank_model_name,
         }
+        if node_execution_id := kwargs.get("node_execution_id"):
+            metadata["node_execution_id"] = node_execution_id
 
         dataset_retrieval_trace_info = DatasetRetrievalTraceInfo(
             trace_id=self.trace_id,
             message_id=message_id,
             inputs=message_data.query or message_data.inputs,
-            documents=[doc.model_dump() for doc in documents] if documents else [],
+            documents=doc_list,
             start_time=timer.get("start"),
             end_time=timer.get("end"),
             metadata=metadata,
@@ -836,6 +1021,10 @@ class TraceTask:
             "error": error,
             "tool_parameters": tool_parameters,
         }
+        if message_data.workflow_run_id:
+            metadata["workflow_run_id"] = message_data.workflow_run_id
+        if node_execution_id := kwargs.get("node_execution_id"):
+            metadata["node_execution_id"] = node_execution_id
 
         file_url = ""
         message_file_data = db.session.query(MessageFile).filter_by(message_id=message_id).first()
@@ -877,6 +1066,158 @@ class TraceTask:
 
         return tool_trace_info
 
+    def prompt_generation_trace(self, **kwargs) -> PromptGenerationTraceInfo | dict:
+        tenant_id = kwargs.get("tenant_id", "")
+        user_id = kwargs.get("user_id", "")
+        app_id = kwargs.get("app_id")
+        operation_type = kwargs.get("operation_type", "")
+        instruction = kwargs.get("instruction", "")
+        generated_output = kwargs.get("generated_output", "")
+
+        prompt_tokens = kwargs.get("prompt_tokens", 0)
+        completion_tokens = kwargs.get("completion_tokens", 0)
+        total_tokens = kwargs.get("total_tokens", 0)
+
+        model_provider = kwargs.get("model_provider", "")
+        model_name = kwargs.get("model_name", "")
+
+        latency = kwargs.get("latency", 0.0)
+
+        timer = kwargs.get("timer")
+        start_time = timer.get("start") if timer else None
+        end_time = timer.get("end") if timer else None
+
+        total_price = kwargs.get("total_price")
+        currency = kwargs.get("currency")
+
+        error = kwargs.get("error")
+
+        app_name = None
+        workspace_name = None
+        if app_id:
+            app_name, workspace_name = _lookup_app_and_workspace_names(app_id, tenant_id)
+
+        metadata = {
+            "tenant_id": tenant_id,
+            "user_id": user_id,
+            "app_id": app_id or "",
+            "app_name": app_name,
+            "workspace_name": workspace_name,
+            "operation_type": operation_type,
+            "model_provider": model_provider,
+            "model_name": model_name,
+        }
+        if node_execution_id := kwargs.get("node_execution_id"):
+            metadata["node_execution_id"] = node_execution_id
+
+        return PromptGenerationTraceInfo(
+            trace_id=self.trace_id,
+            inputs=instruction,
+            outputs=generated_output,
+            start_time=start_time,
+            end_time=end_time,
+            metadata=metadata,
+            tenant_id=tenant_id,
+            user_id=user_id,
+            app_id=app_id,
+            operation_type=operation_type,
+            instruction=instruction,
+            prompt_tokens=prompt_tokens,
+            completion_tokens=completion_tokens,
+            total_tokens=total_tokens,
+            model_provider=model_provider,
+            model_name=model_name,
+            latency=latency,
+            total_price=total_price,
+            currency=currency,
+            error=error,
+        )
+
+    def node_execution_trace(self, **kwargs) -> WorkflowNodeTraceInfo | dict:
+        node_data: dict = kwargs.get("node_execution_data", {})
+        if not node_data:
+            return {}
+
+        app_name, workspace_name = _lookup_app_and_workspace_names(node_data.get("app_id"), node_data.get("tenant_id"))
+
+        credential_name = _lookup_credential_name(
+            node_data.get("credential_id"), node_data.get("credential_provider_type")
+        )
+
+        metadata: dict[str, Any] = {
+            "tenant_id": node_data.get("tenant_id"),
+            "app_id": node_data.get("app_id"),
+            "app_name": app_name,
+            "workspace_name": workspace_name,
+            "user_id": node_data.get("user_id"),
+            "dataset_ids": node_data.get("dataset_ids"),
+            "dataset_names": node_data.get("dataset_names"),
+            "plugin_name": node_data.get("plugin_name"),
+            "credential_name": credential_name,
+        }
+
+        parent_trace_context = node_data.get("parent_trace_context")
+        if parent_trace_context:
+            metadata["parent_trace_context"] = parent_trace_context
+
+        message_id: str | None = None
+        conversation_id = node_data.get("conversation_id")
+        workflow_execution_id = node_data.get("workflow_execution_id")
+        if conversation_id and workflow_execution_id and not parent_trace_context:
+            with Session(db.engine) as session:
+                msg_id = session.scalar(
+                    select(Message.id).where(
+                        Message.conversation_id == conversation_id,
+                        Message.workflow_run_id == workflow_execution_id,
+                    )
+                )
+                if msg_id:
+                    message_id = str(msg_id)
+                    metadata["message_id"] = message_id
+
+        return WorkflowNodeTraceInfo(
+            trace_id=self.trace_id,
+            message_id=message_id,
+            start_time=node_data.get("created_at"),
+            end_time=node_data.get("finished_at"),
+            metadata=metadata,
+            workflow_id=node_data.get("workflow_id", ""),
+            workflow_run_id=node_data.get("workflow_execution_id", ""),
+            tenant_id=node_data.get("tenant_id", ""),
+            node_execution_id=node_data.get("node_execution_id", ""),
+            node_id=node_data.get("node_id", ""),
+            node_type=node_data.get("node_type", ""),
+            title=node_data.get("title", ""),
+            status=node_data.get("status", ""),
+            error=node_data.get("error"),
+            elapsed_time=node_data.get("elapsed_time", 0.0),
+            index=node_data.get("index", 0),
+            predecessor_node_id=node_data.get("predecessor_node_id"),
+            total_tokens=node_data.get("total_tokens", 0),
+            total_price=node_data.get("total_price", 0.0),
+            currency=node_data.get("currency"),
+            model_provider=node_data.get("model_provider"),
+            model_name=node_data.get("model_name"),
+            prompt_tokens=node_data.get("prompt_tokens"),
+            completion_tokens=node_data.get("completion_tokens"),
+            tool_name=node_data.get("tool_name"),
+            iteration_id=node_data.get("iteration_id"),
+            iteration_index=node_data.get("iteration_index"),
+            loop_id=node_data.get("loop_id"),
+            loop_index=node_data.get("loop_index"),
+            parallel_id=node_data.get("parallel_id"),
+            node_inputs=node_data.get("node_inputs"),
+            node_outputs=node_data.get("node_outputs"),
+            process_data=node_data.get("process_data"),
+            invoked_by=self._get_user_id_from_metadata(metadata),
+        )
+
+    def draft_node_execution_trace(self, **kwargs) -> DraftNodeExecutionTrace | dict:
+        node_trace = self.node_execution_trace(**kwargs)
+        if not node_trace or not isinstance(node_trace, WorkflowNodeTraceInfo):
+            return node_trace
+        return DraftNodeExecutionTrace(**node_trace.model_dump())
+
     def generate_name_trace(self, conversation_id, timer, **kwargs):
         generate_conversation_name = kwargs.get("generate_conversation_name")
         inputs = kwargs.get("inputs")
@@ -890,6 +1231,8 @@ class TraceTask:
             "conversation_id": conversation_id,
             "tenant_id": tenant_id,
         }
+        if node_execution_id := kwargs.get("node_execution_id"):
+            metadata["node_execution_id"] = node_execution_id
 
         generate_name_trace_info = GenerateNameTraceInfo(
             trace_id=self.trace_id,
@@ -937,13 +1280,16 @@ class TraceQueueManager:
         self.user_id = user_id
         self.trace_instance = OpsTraceManager.get_ops_trace_instance(app_id)
         self.flask_app = current_app._get_current_object()  # type: ignore
+        from core.telemetry.gateway import is_enterprise_telemetry_enabled
+
+        self._enterprise_telemetry_enabled = is_enterprise_telemetry_enabled()
         if trace_manager_timer is None:
             self.start_timer()
 
     def add_trace_task(self, trace_task: TraceTask):
         global trace_manager_timer, trace_manager_queue
         try:
-            if self.trace_instance:
+            if self._enterprise_telemetry_enabled or self.trace_instance:
                 trace_task.app_id = self.app_id
                 trace_manager_queue.put(trace_task)
         except Exception:
@@ -979,20 +1325,26 @@ class TraceQueueManager:
     def send_to_celery(self, tasks: list[TraceTask]):
         with self.flask_app.app_context():
             for task in tasks:
-                if task.app_id is None:
-                    continue
+                storage_id = task.app_id
+                if storage_id is None:
+                    tenant_id = task.kwargs.get("tenant_id")
+                    if tenant_id:
+                        storage_id = f"tenant-{tenant_id}"
+                    else:
+                        logger.warning("Skipping trace without app_id or tenant_id, trace_type: %s", task.trace_type)
+                        continue
                 file_id = uuid4().hex
                 trace_info = task.execute()
 
                 task_data = TaskData(
-                    app_id=task.app_id,
+                    app_id=storage_id,
                     trace_info_type=type(trace_info).__name__,
                     trace_info=trace_info.model_dump() if trace_info else None,
                 )
-                file_path = f"{OPS_FILE_PATH}{task.app_id}/{file_id}.json"
+                file_path = f"{OPS_FILE_PATH}{storage_id}/{file_id}.json"
                 storage.save(file_path, task_data.model_dump_json().encode("utf-8"))
                 file_info = {
                     "file_id": file_id,
-                    "app_id": task.app_id,
+                    "app_id": storage_id,
                 }
                 process_trace_tasks.delay(file_info)  # type: ignore

api/core/telemetry/__init__.py

@@ -0,0 +1,43 @@
+"""Telemetry facade.
+
+Thin public API for emitting telemetry events.  All routing logic
+lives in ``core.telemetry.gateway`` which is shared by both CE and EE.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from core.ops.entities.trace_entity import TraceTaskName
+from core.telemetry.events import TelemetryContext, TelemetryEvent
+from core.telemetry.gateway import TRACE_TASK_TO_CASE
+from core.telemetry.gateway import emit as gateway_emit
+
+if TYPE_CHECKING:
+    from core.ops.ops_trace_manager import TraceQueueManager
+
+
+def emit(event: TelemetryEvent, trace_manager: TraceQueueManager | None = None) -> None:
+    """Emit a telemetry event.
+
+    Translates the ``TelemetryEvent`` (keyed by ``TraceTaskName``) into a
+    ``TelemetryCase`` and delegates to ``core.telemetry.gateway.emit()``.
+    """
+    case = TRACE_TASK_TO_CASE.get(event.name)
+    if case is None:
+        return
+
+    context: dict[str, object] = {
+        "tenant_id": event.context.tenant_id,
+        "user_id": event.context.user_id,
+        "app_id": event.context.app_id,
+    }
+    gateway_emit(case, context, event.payload, trace_manager)
+
+
+__all__ = [
+    "TelemetryContext",
+    "TelemetryEvent",
+    "TraceTaskName",
+    "emit",
+]

api/core/telemetry/events.py

@@ -0,0 +1,21 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from core.ops.entities.trace_entity import TraceTaskName
+
+
+@dataclass(frozen=True)
+class TelemetryContext:
+    tenant_id: str | None = None
+    user_id: str | None = None
+    app_id: str | None = None
+
+
+@dataclass(frozen=True)
+class TelemetryEvent:
+    name: TraceTaskName
+    context: TelemetryContext
+    payload: dict[str, Any]

api/core/telemetry/gateway.py

@@ -0,0 +1,206 @@
+"""Telemetry gateway — single routing layer for all editions.
+
+Maps ``TelemetryCase`` → ``CaseRoute`` and dispatches events to either
+the CE/EE trace pipeline (``TraceQueueManager``) or the enterprise-only
+metric/log Celery queue.
+
+This module lives in ``core/`` so both CE and EE share one routing table
+and one ``emit()`` entry point.  No separate enterprise gateway module is
+needed — enterprise-specific dispatch (Celery task, payload offloading)
+is handled here behind lazy imports that no-op in CE.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import uuid
+from typing import TYPE_CHECKING, Any
+
+from core.ops.entities.trace_entity import TraceTaskName
+from enterprise.telemetry.contracts import CaseRoute, SignalType, TelemetryCase, TelemetryEnvelope
+from extensions.ext_storage import storage
+
+if TYPE_CHECKING:
+    from core.ops.ops_trace_manager import TraceQueueManager
+
+logger = logging.getLogger(__name__)
+
+PAYLOAD_SIZE_THRESHOLD_BYTES = 1 * 1024 * 1024
+
+# ---------------------------------------------------------------------------
+# Routing table — authoritative mapping for all editions
+# ---------------------------------------------------------------------------
+
+CASE_TO_TRACE_TASK: dict[TelemetryCase, TraceTaskName] = {
+    TelemetryCase.WORKFLOW_RUN: TraceTaskName.WORKFLOW_TRACE,
+    TelemetryCase.MESSAGE_RUN: TraceTaskName.MESSAGE_TRACE,
+    TelemetryCase.NODE_EXECUTION: TraceTaskName.NODE_EXECUTION_TRACE,
+    TelemetryCase.DRAFT_NODE_EXECUTION: TraceTaskName.DRAFT_NODE_EXECUTION_TRACE,
+    TelemetryCase.PROMPT_GENERATION: TraceTaskName.PROMPT_GENERATION_TRACE,
+    TelemetryCase.TOOL_EXECUTION: TraceTaskName.TOOL_TRACE,
+    TelemetryCase.MODERATION_CHECK: TraceTaskName.MODERATION_TRACE,
+    TelemetryCase.SUGGESTED_QUESTION: TraceTaskName.SUGGESTED_QUESTION_TRACE,
+    TelemetryCase.DATASET_RETRIEVAL: TraceTaskName.DATASET_RETRIEVAL_TRACE,
+    TelemetryCase.GENERATE_NAME: TraceTaskName.GENERATE_NAME_TRACE,
+}
+
+TRACE_TASK_TO_CASE: dict[TraceTaskName, TelemetryCase] = {v: k for k, v in CASE_TO_TRACE_TASK.items()}
+
+CASE_ROUTING: dict[TelemetryCase, CaseRoute] = {
+    # TRACE — CE-eligible (flow in both CE and EE)
+    TelemetryCase.WORKFLOW_RUN: CaseRoute(signal_type=SignalType.TRACE, ce_eligible=True),
+    TelemetryCase.MESSAGE_RUN: CaseRoute(signal_type=SignalType.TRACE, ce_eligible=True),
+    TelemetryCase.TOOL_EXECUTION: CaseRoute(signal_type=SignalType.TRACE, ce_eligible=True),
+    TelemetryCase.MODERATION_CHECK: CaseRoute(signal_type=SignalType.TRACE, ce_eligible=True),
+    TelemetryCase.SUGGESTED_QUESTION: CaseRoute(signal_type=SignalType.TRACE, ce_eligible=True),
+    TelemetryCase.DATASET_RETRIEVAL: CaseRoute(signal_type=SignalType.TRACE, ce_eligible=True),
+    TelemetryCase.GENERATE_NAME: CaseRoute(signal_type=SignalType.TRACE, ce_eligible=True),
+    # TRACE — enterprise-only
+    TelemetryCase.NODE_EXECUTION: CaseRoute(signal_type=SignalType.TRACE, ce_eligible=False),
+    TelemetryCase.DRAFT_NODE_EXECUTION: CaseRoute(signal_type=SignalType.TRACE, ce_eligible=False),
+    TelemetryCase.PROMPT_GENERATION: CaseRoute(signal_type=SignalType.TRACE, ce_eligible=False),
+    # METRIC_LOG — enterprise-only (signal-driven, not trace)
+    TelemetryCase.APP_CREATED: CaseRoute(signal_type=SignalType.METRIC_LOG, ce_eligible=False),
+    TelemetryCase.APP_UPDATED: CaseRoute(signal_type=SignalType.METRIC_LOG, ce_eligible=False),
+    TelemetryCase.APP_DELETED: CaseRoute(signal_type=SignalType.METRIC_LOG, ce_eligible=False),
+    TelemetryCase.FEEDBACK_CREATED: CaseRoute(signal_type=SignalType.METRIC_LOG, ce_eligible=False),
+}
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def is_enterprise_telemetry_enabled() -> bool:
+    try:
+        from enterprise.telemetry.exporter import is_enterprise_telemetry_enabled
+
+        return is_enterprise_telemetry_enabled()
+    except Exception:
+        return False
+
+
+def _handle_payload_sizing(
+    payload: dict[str, Any],
+    tenant_id: str,
+    event_id: str,
+) -> tuple[dict[str, Any], str | None]:
+    """Inline or offload payload based on size.
+
+    Returns ``(payload_for_envelope, storage_key | None)``.  Payloads
+    exceeding ``PAYLOAD_SIZE_THRESHOLD_BYTES`` are written to object
+    storage and replaced with an empty dict in the envelope.
+    """
+    try:
+        payload_json = json.dumps(payload)
+        payload_size = len(payload_json.encode("utf-8"))
+    except (TypeError, ValueError):
+        logger.warning("Failed to serialize payload for sizing: event_id=%s", event_id)
+        return payload, None
+
+    if payload_size <= PAYLOAD_SIZE_THRESHOLD_BYTES:
+        return payload, None
+
+    storage_key = f"telemetry/{tenant_id}/{event_id}.json"
+    try:
+        storage.save(storage_key, payload_json.encode("utf-8"))
+        logger.debug("Stored large payload to storage: key=%s, size=%d", storage_key, payload_size)
+        return {}, storage_key
+    except Exception:
+        logger.warning("Failed to store large payload, inlining instead: event_id=%s", event_id, exc_info=True)
+        return payload, None
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+
+def emit(
+    case: TelemetryCase,
+    context: dict[str, Any],
+    payload: dict[str, Any],
+    trace_manager: TraceQueueManager | None = None,
+) -> None:
+    """Route a telemetry event to the correct pipeline.
+
+    TRACE events are enqueued into ``TraceQueueManager`` (works in both CE
+    and EE).  Enterprise-only traces are silently dropped when EE is
+    disabled.
+
+    METRIC_LOG events are dispatched to the enterprise Celery queue;
+    silently dropped when enterprise telemetry is unavailable.
+    """
+    route = CASE_ROUTING.get(case)
+    if route is None:
+        logger.warning("Unknown telemetry case: %s, dropping event", case)
+        return
+
+    if not route.ce_eligible and not is_enterprise_telemetry_enabled():
+        logger.debug("Dropping EE-only event: case=%s (EE disabled)", case)
+        return
+
+    if route.signal_type is SignalType.TRACE:
+        _emit_trace(case, context, payload, trace_manager)
+    else:
+        _emit_metric_log(case, context, payload)
+
+
+def _emit_trace(
+    case: TelemetryCase,
+    context: dict[str, Any],
+    payload: dict[str, Any],
+    trace_manager: TraceQueueManager | None,
+) -> None:
+    from core.ops.ops_trace_manager import TraceQueueManager as LocalTraceQueueManager
+    from core.ops.ops_trace_manager import TraceTask
+
+    trace_task_name = CASE_TO_TRACE_TASK.get(case)
+    if trace_task_name is None:
+        logger.warning("No TraceTaskName mapping for case: %s", case)
+        return
+
+    queue_manager = trace_manager or LocalTraceQueueManager(
+        app_id=context.get("app_id"),
+        user_id=context.get("user_id"),
+    )
+    queue_manager.add_trace_task(TraceTask(trace_task_name, **payload))
+    logger.debug("Enqueued trace task: case=%s, app_id=%s", case, context.get("app_id"))
+
+
+def _emit_metric_log(
+    case: TelemetryCase,
+    context: dict[str, Any],
+    payload: dict[str, Any],
+) -> None:
+    """Build envelope and dispatch to enterprise Celery queue.
+
+    No-ops when the enterprise telemetry task is not importable (CE mode).
+    """
+    try:
+        from tasks.enterprise_telemetry_task import process_enterprise_telemetry
+    except ImportError:
+        logger.debug("Enterprise metric/log dispatch unavailable, dropping: case=%s", case)
+        return
+
+    tenant_id = context.get("tenant_id", "")
+    event_id = str(uuid.uuid4())
+
+    payload_for_envelope, payload_ref = _handle_payload_sizing(payload, tenant_id, event_id)
+
+    envelope = TelemetryEnvelope(
+        case=case,
+        tenant_id=tenant_id,
+        event_id=event_id,
+        payload=payload_for_envelope,
+        metadata={"payload_ref": payload_ref} if payload_ref else None,
+    )
+
+    process_enterprise_telemetry.delay(envelope.model_dump_json())
+    logger.debug(
+        "Enqueued metric/log event: case=%s, tenant_id=%s, event_id=%s",
+        case,
+        tenant_id,
+        event_id,
+    )

api/enterprise/telemetry/DATA_DICTIONARY.md

@@ -0,0 +1,452 @@
+# Dify Enterprise Telemetry Data Dictionary
+
+Quick reference for all telemetry signals emitted by Dify Enterprise. For configuration and architecture details, see [README.md](./README.md).
+
+## Resource Attributes
+
+Attached to every signal (Span, Metric, Log).
+
+| Attribute | Type | Example |
+|-----------|------|---------|
+| `service.name` | string | `dify` |
+| `host.name` | string | `dify-api-7f8b` |
+
+## Traces (Spans)
+
+### `dify.workflow.run`
+
+| Attribute | Type | Description |
+|-----------|------|-------------|
+| `dify.trace_id` | string | Business trace ID (Workflow Run ID) |
+| `dify.tenant_id` | string | Tenant identifier |
+| `dify.app_id` | string | Application identifier |
+| `dify.workflow.id` | string | Workflow definition ID |
+| `dify.workflow.run_id` | string | Unique ID for this run |
+| `dify.workflow.status` | string | `succeeded`, `failed`, `stopped`, etc. |
+| `dify.workflow.error` | string | Error message if failed |
+| `dify.workflow.elapsed_time` | float | Total execution time (seconds) |
+| `dify.invoke_from` | string | `api`, `webapp`, `debug` |
+| `dify.conversation.id` | string | Conversation ID (optional) |
+| `dify.message.id` | string | Message ID (optional) |
+| `dify.invoked_by` | string | User ID who triggered the run |
+| `dify.parent.trace_id` | string | Parent workflow trace ID (optional) |
+| `dify.parent.workflow.run_id` | string | Parent workflow run ID (optional) |
+| `dify.parent.node.execution_id` | string | Parent node execution ID (optional) |
+| `dify.parent.app.id` | string | Parent app ID (optional) |
+
+### `dify.node.execution`
+
+| Attribute | Type | Description |
+|-----------|------|-------------|
+| `dify.trace_id` | string | Business trace ID |
+| `dify.tenant_id` | string | Tenant identifier |
+| `dify.app_id` | string | Application identifier |
+| `dify.workflow.id` | string | Workflow definition ID |
+| `dify.workflow.run_id` | string | Workflow Run ID |
+| `dify.message.id` | string | Message ID (optional) |
+| `dify.conversation.id` | string | Conversation ID (optional) |
+| `dify.node.execution_id` | string | Unique node execution ID |
+| `dify.node.id` | string | Node ID in workflow graph |
+| `dify.node.type` | string | Node type (see appendix) |
+| `dify.node.title` | string | Display title |
+| `dify.node.status` | string | `succeeded`, `failed` |
+| `dify.node.error` | string | Error message if failed |
+| `dify.node.elapsed_time` | float | Execution time (seconds) |
+| `dify.node.index` | int | Execution order index |
+| `dify.node.predecessor_node_id` | string | Triggering node ID |
+| `dify.node.iteration_id` | string | Iteration ID (optional) |
+| `dify.node.loop_id` | string | Loop ID (optional) |
+| `dify.node.parallel_id` | string | Parallel branch ID (optional) |
+| `dify.node.invoked_by` | string | User ID who triggered execution |
+
+### `dify.node.execution.draft`
+
+Same attributes as `dify.node.execution`. Emitted during Preview/Debug runs.
+
+## Counters
+
+All counters are cumulative and emitted at 100% accuracy.
+
+### Token Counters
+
+| Metric | Unit | Description |
+|--------|------|-------------|
+| `dify.tokens.total` | `{token}` | Total tokens consumed |
+| `dify.tokens.input` | `{token}` | Input (prompt) tokens |
+| `dify.tokens.output` | `{token}` | Output (completion) tokens |
+
+**Labels:**
+- `tenant_id`, `app_id`, `operation_type`, `model_provider`, `model_name`, `node_type` (if node_execution)
+
+⚠️ **Warning:** `dify.tokens.total` at workflow level includes all node tokens. Filter by `operation_type` to avoid double-counting.
+
+### Request Counters
+
+| Metric | Unit | Description |
+|--------|------|-------------|
+| `dify.requests.total` | `{request}` | Total operations count |
+
+**Labels by type:**
+
+| `type` | Additional Labels |
+|--------|-------------------|
+| `workflow` | `tenant_id`, `app_id`, `status`, `invoke_from` |
+| `node` | `tenant_id`, `app_id`, `node_type`, `model_provider`, `model_name`, `status` |
+| `draft_node` | `tenant_id`, `app_id`, `node_type`, `model_provider`, `model_name`, `status` |
+| `message` | `tenant_id`, `app_id`, `model_provider`, `model_name`, `status`, `invoke_from` |
+| `tool` | `tenant_id`, `app_id`, `tool_name` |
+| `moderation` | `tenant_id`, `app_id` |
+| `suggested_question` | `tenant_id`, `app_id`, `model_provider`, `model_name` |
+| `dataset_retrieval` | `tenant_id`, `app_id` |
+| `generate_name` | `tenant_id`, `app_id` |
+| `prompt_generation` | `tenant_id`, `app_id`, `operation_type`, `model_provider`, `model_name`, `status` |
+
+### Error Counters
+
+| Metric | Unit | Description |
+|--------|------|-------------|
+| `dify.errors.total` | `{error}` | Total failed operations |
+
+**Labels by type:**
+
+| `type` | Additional Labels |
+|--------|-------------------|
+| `workflow` | `tenant_id`, `app_id` |
+| `node` | `tenant_id`, `app_id`, `node_type`, `model_provider`, `model_name` |
+| `draft_node` | `tenant_id`, `app_id`, `node_type`, `model_provider`, `model_name` |
+| `message` | `tenant_id`, `app_id`, `model_provider`, `model_name` |
+| `tool` | `tenant_id`, `app_id`, `tool_name` |
+| `prompt_generation` | `tenant_id`, `app_id`, `operation_type`, `model_provider`, `model_name` |
+
+### Other Counters
+
+| Metric | Unit | Labels |
+|--------|------|--------|
+| `dify.feedback.total` | `{feedback}` | `tenant_id`, `app_id`, `rating` |
+| `dify.dataset.retrievals.total` | `{retrieval}` | `tenant_id`, `app_id`, `dataset_id`, `embedding_model_provider`, `embedding_model`, `rerank_model_provider`, `rerank_model` |
+| `dify.app.created.total` | `{app}` | `tenant_id`, `app_id`, `mode` |
+| `dify.app.updated.total` | `{app}` | `tenant_id`, `app_id` |
+| `dify.app.deleted.total` | `{app}` | `tenant_id`, `app_id` |
+
+## Histograms
+
+| Metric | Unit | Labels |
+|--------|------|--------|
+| `dify.workflow.duration` | `s` | `tenant_id`, `app_id`, `status` |
+| `dify.node.duration` | `s` | `tenant_id`, `app_id`, `node_type`, `model_provider`, `model_name`, `plugin_name` |
+| `dify.message.duration` | `s` | `tenant_id`, `app_id`, `model_provider`, `model_name` |
+| `dify.message.time_to_first_token` | `s` | `tenant_id`, `app_id`, `model_provider`, `model_name` |
+| `dify.tool.duration` | `s` | `tenant_id`, `app_id`, `tool_name` |
+| `dify.prompt_generation.duration` | `s` | `tenant_id`, `app_id`, `operation_type`, `model_provider`, `model_name` |
+
+## Structured Logs
+
+### Span Companion Logs
+
+Logs that accompany spans. Signal type: `span_detail`
+
+#### `dify.workflow.run` Companion Log
+
+**Common attributes:** All span attributes (see Traces section) plus:
+
+| Additional Attribute | Type | Always Present | Description |
+|---------------------|------|----------------|-------------|
+| `dify.user.id` | string | No | User identifier |
+| `gen_ai.usage.total_tokens` | int | No | Total tokens (sum of all nodes) |
+| `dify.workflow.version` | string | Yes | Workflow version |
+| `dify.workflow.inputs` | string/JSON | Yes | Input parameters (content-gated) |
+| `dify.workflow.outputs` | string/JSON | Yes | Output results (content-gated) |
+| `dify.workflow.query` | string | No | User query text (content-gated) |
+
+**Event attributes:**
+- `dify.event.name`: `"dify.workflow.run"`
+- `dify.event.signal`: `"span_detail"`
+- `trace_id`, `span_id`, `tenant_id`, `user_id`
+
+#### `dify.node.execution` and `dify.node.execution.draft` Companion Logs
+
+**Common attributes:** All span attributes (see Traces section) plus:
+
+| Additional Attribute | Type | Always Present | Description |
+|---------------------|------|----------------|-------------|
+| `dify.user.id` | string | No | User identifier |
+| `gen_ai.provider.name` | string | No | LLM provider (LLM nodes only) |
+| `gen_ai.request.model` | string | No | LLM model (LLM nodes only) |
+| `gen_ai.usage.input_tokens` | int | No | Input tokens (LLM nodes only) |
+| `gen_ai.usage.output_tokens` | int | No | Output tokens (LLM nodes only) |
+| `gen_ai.usage.total_tokens` | int | No | Total tokens (LLM nodes only) |
+| `dify.node.total_price` | float | No | Cost (LLM nodes only) |
+| `dify.node.currency` | string | No | Currency code (LLM nodes only) |
+| `dify.node.plugin_name` | string | No | Plugin name (tool/knowledge nodes) |
+| `dify.node.plugin_id` | string | No | Plugin ID (tool/knowledge nodes) |
+| `dify.dataset.id` | string | No | Dataset ID (knowledge nodes) |
+| `dify.dataset.name` | string | No | Dataset name (knowledge nodes) |
+| `dify.node.inputs` | string/JSON | Yes | Node inputs (content-gated) |
+| `dify.node.outputs` | string/JSON | Yes | Node outputs (content-gated) |
+| `dify.node.process_data` | string/JSON | No | Processing data (content-gated) |
+
+**Event attributes:**
+- `dify.event.name`: `"dify.node.execution"` or `"dify.node.execution.draft"`
+- `dify.event.signal`: `"span_detail"`
+- `trace_id`, `span_id`, `tenant_id`, `user_id`
+
+### Standalone Logs
+
+Logs without structural spans. Signal type: `metric_only`
+
+#### `dify.message.run`
+
+| Attribute | Type | Description |
+|-----------|------|-------------|
+| `dify.event.name` | string | `"dify.message.run"` |
+| `dify.event.signal` | string | `"metric_only"` |
+| `trace_id` | string | OTEL trace ID (32-char hex) |
+| `span_id` | string | OTEL span ID (16-char hex) |
+| `tenant_id` | string | Tenant identifier |
+| `user_id` | string | User identifier (optional) |
+| `dify.app_id` | string | Application identifier |
+| `dify.message.id` | string | Message identifier |
+| `dify.conversation.id` | string | Conversation ID (optional) |
+| `dify.workflow.run_id` | string | Workflow run ID (optional) |
+| `dify.invoke_from` | string | `service-api`, `web-app`, `debugger`, `explore` |
+| `gen_ai.provider.name` | string | LLM provider |
+| `gen_ai.request.model` | string | LLM model |
+| `gen_ai.usage.input_tokens` | int | Input tokens |
+| `gen_ai.usage.output_tokens` | int | Output tokens |
+| `gen_ai.usage.total_tokens` | int | Total tokens |
+| `dify.message.status` | string | `succeeded`, `failed` |
+| `dify.message.error` | string | Error message (if failed) |
+| `dify.message.duration` | float | Duration (seconds) |
+| `dify.message.time_to_first_token` | float | TTFT (seconds) |
+| `dify.message.inputs` | string/JSON | Inputs (content-gated) |
+| `dify.message.outputs` | string/JSON | Outputs (content-gated) |
+
+#### `dify.tool.execution`
+
+| Attribute | Type | Description |
+|-----------|------|-------------|
+| `dify.event.name` | string | `"dify.tool.execution"` |
+| `dify.event.signal` | string | `"metric_only"` |
+| `trace_id` | string | OTEL trace ID |
+| `span_id` | string | OTEL span ID |
+| `tenant_id` | string | Tenant identifier |
+| `dify.app_id` | string | Application identifier |
+| `dify.message.id` | string | Message identifier |
+| `dify.tool.name` | string | Tool name |
+| `dify.tool.duration` | float | Duration (seconds) |
+| `dify.tool.status` | string | `succeeded`, `failed` |
+| `dify.tool.error` | string | Error message (if failed) |
+| `dify.tool.inputs` | string/JSON | Inputs (content-gated) |
+| `dify.tool.outputs` | string/JSON | Outputs (content-gated) |
+| `dify.tool.parameters` | string/JSON | Parameters (content-gated) |
+| `dify.tool.config` | string/JSON | Configuration (content-gated) |
+
+#### `dify.moderation.check`
+
+| Attribute | Type | Description |
+|-----------|------|-------------|
+| `dify.event.name` | string | `"dify.moderation.check"` |
+| `dify.event.signal` | string | `"metric_only"` |
+| `trace_id` | string | OTEL trace ID |
+| `span_id` | string | OTEL span ID |
+| `tenant_id` | string | Tenant identifier |
+| `dify.app_id` | string | Application identifier |
+| `dify.message.id` | string | Message identifier |
+| `dify.moderation.type` | string | `input`, `output` |
+| `dify.moderation.action` | string | `pass`, `block`, `flag` |
+| `dify.moderation.flagged` | boolean | Whether flagged |
+| `dify.moderation.categories` | JSON array | Flagged categories |
+| `dify.moderation.query` | string | Content (content-gated) |
+
+#### `dify.suggested_question.generation`
+
+| Attribute | Type | Description |
+|-----------|------|-------------|
+| `dify.event.name` | string | `"dify.suggested_question.generation"` |
+| `dify.event.signal` | string | `"metric_only"` |
+| `trace_id` | string | OTEL trace ID |
+| `span_id` | string | OTEL span ID |
+| `tenant_id` | string | Tenant identifier |
+| `dify.app_id` | string | Application identifier |
+| `dify.message.id` | string | Message identifier |
+| `dify.suggested_question.count` | int | Number of questions |
+| `dify.suggested_question.duration` | float | Duration (seconds) |
+| `dify.suggested_question.status` | string | `succeeded`, `failed` |
+| `dify.suggested_question.error` | string | Error message (if failed) |
+| `dify.suggested_question.questions` | JSON array | Questions (content-gated) |
+
+#### `dify.dataset.retrieval`
+
+| Attribute | Type | Description |
+|-----------|------|-------------|
+| `dify.event.name` | string | `"dify.dataset.retrieval"` |
+| `dify.event.signal` | string | `"metric_only"` |
+| `trace_id` | string | OTEL trace ID |
+| `span_id` | string | OTEL span ID |
+| `tenant_id` | string | Tenant identifier |
+| `dify.app_id` | string | Application identifier |
+| `dify.message.id` | string | Message identifier |
+| `dify.dataset.id` | string | Dataset identifier |
+| `dify.dataset.name` | string | Dataset name |
+| `dify.dataset.embedding_providers` | JSON array | Embedding model providers (one per dataset) |
+| `dify.dataset.embedding_models` | JSON array | Embedding models (one per dataset) |
+| `dify.retrieval.rerank_provider` | string | Rerank model provider |
+| `dify.retrieval.rerank_model` | string | Rerank model name |
+| `dify.retrieval.query` | string | Search query (content-gated) |
+| `dify.retrieval.document_count` | int | Documents retrieved |
+| `dify.retrieval.duration` | float | Duration (seconds) |
+| `dify.retrieval.status` | string | `succeeded`, `failed` |
+| `dify.retrieval.error` | string | Error message (if failed) |
+| `dify.dataset.documents` | JSON array | Documents (content-gated) |
+
+#### `dify.generate_name.execution`
+
+| Attribute | Type | Description |
+|-----------|------|-------------|
+| `dify.event.name` | string | `"dify.generate_name.execution"` |
+| `dify.event.signal` | string | `"metric_only"` |
+| `trace_id` | string | OTEL trace ID |
+| `span_id` | string | OTEL span ID |
+| `tenant_id` | string | Tenant identifier |
+| `dify.app_id` | string | Application identifier |
+| `dify.conversation.id` | string | Conversation identifier |
+| `dify.generate_name.duration` | float | Duration (seconds) |
+| `dify.generate_name.status` | string | `succeeded`, `failed` |
+| `dify.generate_name.error` | string | Error message (if failed) |
+| `dify.generate_name.inputs` | string/JSON | Inputs (content-gated) |
+| `dify.generate_name.outputs` | string | Generated name (content-gated) |
+
+#### `dify.prompt_generation.execution`
+
+| Attribute | Type | Description |
+|-----------|------|-------------|
+| `dify.event.name` | string | `"dify.prompt_generation.execution"` |
+| `dify.event.signal` | string | `"metric_only"` |
+| `trace_id` | string | OTEL trace ID |
+| `span_id` | string | OTEL span ID |
+| `tenant_id` | string | Tenant identifier |
+| `dify.app_id` | string | Application identifier |
+| `dify.prompt_generation.operation_type` | string | Operation type (see appendix) |
+| `gen_ai.provider.name` | string | LLM provider |
+| `gen_ai.request.model` | string | LLM model |
+| `gen_ai.usage.input_tokens` | int | Input tokens |
+| `gen_ai.usage.output_tokens` | int | Output tokens |
+| `gen_ai.usage.total_tokens` | int | Total tokens |
+| `dify.prompt_generation.duration` | float | Duration (seconds) |
+| `dify.prompt_generation.status` | string | `succeeded`, `failed` |
+| `dify.prompt_generation.error` | string | Error message (if failed) |
+| `dify.prompt_generation.instruction` | string | Instruction (content-gated) |
+| `dify.prompt_generation.output` | string/JSON | Output (content-gated) |
+
+#### `dify.app.created`
+
+| Attribute | Type | Description |
+|-----------|------|-------------|
+| `dify.event.name` | string | `"dify.app.created"` |
+| `dify.event.signal` | string | `"metric_only"` |
+| `tenant_id` | string | Tenant identifier |
+| `dify.app_id` | string | Application identifier |
+| `dify.app.mode` | string | `chat`, `completion`, `agent-chat`, `workflow` |
+| `dify.app.created_at` | string | Timestamp (ISO 8601) |
+
+#### `dify.app.updated`
+
+| Attribute | Type | Description |
+|-----------|------|-------------|
+| `dify.event.name` | string | `"dify.app.updated"` |
+| `dify.event.signal` | string | `"metric_only"` |
+| `tenant_id` | string | Tenant identifier |
+| `dify.app_id` | string | Application identifier |
+| `dify.app.updated_at` | string | Timestamp (ISO 8601) |
+
+#### `dify.app.deleted`
+
+| Attribute | Type | Description |
+|-----------|------|-------------|
+| `dify.event.name` | string | `"dify.app.deleted"` |
+| `dify.event.signal` | string | `"metric_only"` |
+| `tenant_id` | string | Tenant identifier |
+| `dify.app_id` | string | Application identifier |
+| `dify.app.deleted_at` | string | Timestamp (ISO 8601) |
+
+#### `dify.feedback.created`
+
+| Attribute | Type | Description |
+|-----------|------|-------------|
+| `dify.event.name` | string | `"dify.feedback.created"` |
+| `dify.event.signal` | string | `"metric_only"` |
+| `trace_id` | string | OTEL trace ID |
+| `span_id` | string | OTEL span ID |
+| `tenant_id` | string | Tenant identifier |
+| `dify.app_id` | string | Application identifier |
+| `dify.message.id` | string | Message identifier |
+| `dify.feedback.rating` | string | `like`, `dislike`, `null` |
+| `dify.feedback.content` | string | Feedback text (content-gated) |
+| `dify.feedback.created_at` | string | Timestamp (ISO 8601) |
+
+#### `dify.telemetry.rehydration_failed`
+
+Diagnostic event for telemetry system health monitoring.
+
+| Attribute | Type | Description |
+|-----------|------|-------------|
+| `dify.event.name` | string | `"dify.telemetry.rehydration_failed"` |
+| `dify.event.signal` | string | `"metric_only"` |
+| `tenant_id` | string | Tenant identifier |
+| `dify.telemetry.error` | string | Error message |
+| `dify.telemetry.payload_type` | string | Payload type (see appendix) |
+| `dify.telemetry.correlation_id` | string | Correlation ID |
+
+## Content-Gated Attributes
+
+When `ENTERPRISE_INCLUDE_CONTENT=false`, these attributes are replaced with reference strings (`ref:{id_type}={uuid}`).
+
+| Attribute | Signal |
+|-----------|--------|
+| `dify.workflow.inputs` | `dify.workflow.run` |
+| `dify.workflow.outputs` | `dify.workflow.run` |
+| `dify.workflow.query` | `dify.workflow.run` |
+| `dify.node.inputs` | `dify.node.execution` |
+| `dify.node.outputs` | `dify.node.execution` |
+| `dify.node.process_data` | `dify.node.execution` |
+| `dify.message.inputs` | `dify.message.run` |
+| `dify.message.outputs` | `dify.message.run` |
+| `dify.tool.inputs` | `dify.tool.execution` |
+| `dify.tool.outputs` | `dify.tool.execution` |
+| `dify.tool.parameters` | `dify.tool.execution` |
+| `dify.tool.config` | `dify.tool.execution` |
+| `dify.moderation.query` | `dify.moderation.check` |
+| `dify.suggested_question.questions` | `dify.suggested_question.generation` |
+| `dify.retrieval.query` | `dify.dataset.retrieval` |
+| `dify.dataset.documents` | `dify.dataset.retrieval` |
+| `dify.generate_name.inputs` | `dify.generate_name.execution` |
+| `dify.generate_name.outputs` | `dify.generate_name.execution` |
+| `dify.prompt_generation.instruction` | `dify.prompt_generation.execution` |
+| `dify.prompt_generation.output` | `dify.prompt_generation.execution` |
+| `dify.feedback.content` | `dify.feedback.created` |
+
+## Appendix
+
+### Operation Types
+
+- `workflow`, `node_execution`, `message`, `rule_generate`, `code_generate`, `structured_output`, `instruction_modify`
+
+### Node Types
+
+- `start`, `end`, `answer`, `llm`, `knowledge-retrieval`, `knowledge-index`, `if-else`, `code`, `template-transform`, `question-classifier`, `http-request`, `tool`, `datasource`, `variable-aggregator`, `loop`, `iteration`, `parameter-extractor`, `assigner`, `document-extractor`, `list-operator`, `agent`, `trigger-webhook`, `trigger-schedule`, `trigger-plugin`, `human-input`
+
+### Workflow Statuses
+
+- `running`, `succeeded`, `failed`, `stopped`, `partial-succeeded`, `paused`
+
+### Payload Types
+
+- `workflow`, `node`, `message`, `tool`, `moderation`, `suggested_question`, `dataset_retrieval`, `generate_name`, `prompt_generation`, `app`, `feedback`
+
+### Null Value Behavior
+
+**Spans:** Attributes with `null` values are omitted.
+
+**Logs:** Attributes with `null` values appear as `null` in JSON.
+
+**Content-Gated:** Replaced with reference strings, not set to `null`.

api/enterprise/telemetry/README.md

@@ -0,0 +1,116 @@
+# Dify Enterprise Telemetry
+
+This document provides an overview of the Dify Enterprise OpenTelemetry (OTEL) exporter and how to configure it for integration with observability stacks like Prometheus, Grafana, Jaeger, or Honeycomb.
+
+## Overview
+
+Dify Enterprise uses a "slim span + rich companion log" architecture to provide high-fidelity observability without overwhelming trace storage.
+
+- **Traces (Spans)**: Capture the structure, identity, and timing of high-level operations (Workflows and Nodes).
+- **Structured Logs**: Provide deep context (inputs, outputs, metadata) for every event, correlated to spans via `trace_id` and `span_id`.
+- **Metrics**: Provide 100% accurate counters and histograms for usage, performance, and error tracking.
+
+### Signal Architecture
+
+```mermaid
+graph TD
+    A[Workflow Run] -->|Span| B(dify.workflow.run)
+    A -->|Log| C(dify.workflow.run detail)
+    B ---|trace_id| C
+    
+    D[Node Execution] -->|Span| E(dify.node.execution)
+    D -->|Log| F(dify.node.execution detail)
+    E ---|span_id| F
+    
+    G[Message/Tool/etc] -->|Log| H(dify.* event)
+    G -->|Metric| I(dify.* counter/histogram)
+```
+
+## Configuration
+
+The Enterprise OTEL exporter is configured via environment variables.
+
+| Variable | Description | Default |
+|----------|-------------|---------|
+| `ENTERPRISE_ENABLED` | Master switch for all enterprise features. | `false` |
+| `ENTERPRISE_TELEMETRY_ENABLED` | Master switch for enterprise telemetry. | `false` |
+| `ENTERPRISE_OTLP_ENDPOINT` | OTLP collector endpoint (e.g., `http://otel-collector:4318`). | - |
+| `ENTERPRISE_OTLP_HEADERS` | Custom headers for OTLP requests (e.g., `x-scope-orgid=tenant1`). | - |
+| `ENTERPRISE_OTLP_PROTOCOL` | OTLP transport protocol (`http` or `grpc`). | `http` |
+| `ENTERPRISE_OTLP_API_KEY` | Bearer token for authentication. | - |
+| `ENTERPRISE_INCLUDE_CONTENT` | Whether to include sensitive content (inputs/outputs) in logs. | `true` |
+| `ENTERPRISE_SERVICE_NAME` | Service name reported to OTEL. | `dify` |
+| `ENTERPRISE_OTEL_SAMPLING_RATE` | Sampling rate for traces (0.0 to 1.0). Metrics are always 100%. | `1.0` |
+
+## Correlation Model
+
+Dify uses deterministic ID generation to ensure signals are correlated across different services and asynchronous tasks.
+
+### ID Generation Rules
+- `trace_id`: Derived from the correlation ID (workflow_run_id or node_execution_id for drafts) using `int(UUID(correlation_id))`
+- `span_id`: Derived from the source ID using `SHA256(source_id)[:8]`
+
+### Scenario A: Simple Workflow
+A single workflow run with multiple nodes. All spans and logs share the same `trace_id` (derived from `workflow_run_id`).
+
+```
+trace_id = UUID(workflow_run_id)
+├── [root span] dify.workflow.run (span_id = hash(workflow_run_id))
+│   ├── [child] dify.node.execution - "Start" (span_id = hash(node_exec_id_1))
+│   ├── [child] dify.node.execution - "LLM" (span_id = hash(node_exec_id_2))
+│   └── [child] dify.node.execution - "End" (span_id = hash(node_exec_id_3))
+```
+
+### Scenario B: Nested Sub-Workflow
+A workflow calling another workflow via a Tool or Sub-workflow node. The child workflow's spans are linked to the parent via `parent_span_id`. Both workflows share the same trace_id.
+
+```
+trace_id = UUID(outer_workflow_run_id)     ← shared across both workflows
+├── [root] dify.workflow.run (outer) (span_id = hash(outer_workflow_run_id))
+│   ├── dify.node.execution - "Start Node"
+│   ├── dify.node.execution - "Tool Node" (triggers sub-workflow)
+│   │   └── [child] dify.workflow.run (inner) (span_id = hash(inner_workflow_run_id))
+│   │       ├── dify.node.execution - "Inner Start"
+│   │       └── dify.node.execution - "Inner End"
+│   └── dify.node.execution - "End Node"
+```
+
+**Key attributes for nested workflows:**
+- Inner workflow's `dify.parent.trace_id` = outer `workflow_run_id`
+- Inner workflow's `dify.parent.node.execution_id` = tool node's `execution_id`
+- Inner workflow's `dify.parent.workflow.run_id` = outer `workflow_run_id`
+- Inner workflow's `dify.parent.app.id` = outer `app_id`
+
+### Scenario C: Draft Node Execution
+A single node run in isolation (debugger/preview mode). It creates its own trace where the node span is the root.
+
+```
+trace_id = UUID(node_execution_id)   ← own trace, NOT part of any workflow
+└── dify.node.execution.draft (span_id = hash(node_execution_id))
+```
+
+**Key difference:** Draft executions use `node_execution_id` as the correlation_id, so they are NOT children of any workflow trace.
+
+## Content Gating
+
+When `ENTERPRISE_INCLUDE_CONTENT` is set to `false`, sensitive content attributes (inputs, outputs, queries) are replaced with reference strings (e.g., `ref:workflow_run_id=...`) to prevent data leakage to the OTEL collector.
+
+**Reference String Format:**
+
+```
+ref:{id_type}={uuid}
+```
+
+**Examples:**
+
+```
+ref:workflow_run_id=550e8400-e29b-41d4-a716-446655440000
+ref:node_execution_id=660e8400-e29b-41d4-a716-446655440001
+ref:message_id=770e8400-e29b-41d4-a716-446655440002
+```
+
+To retrieve actual content when gating is enabled, query the Dify database using the provided UUID.
+
+## Reference
+
+For a complete list of telemetry signals, attributes, and data structures, see [DATA_DICTIONARY.md](./DATA_DICTIONARY.md).

api/enterprise/telemetry/contracts.py

@@ -0,0 +1,83 @@
+"""Telemetry gateway contracts and data structures.
+
+This module defines the envelope format for telemetry events and the routing
+configuration that determines how each event type is processed.
+"""
+
+from __future__ import annotations
+
+from enum import StrEnum
+from typing import Any
+
+from pydantic import BaseModel, field_validator
+
+
+class TelemetryCase(StrEnum):
+    """Enumeration of all known telemetry event cases."""
+
+    WORKFLOW_RUN = "workflow_run"
+    NODE_EXECUTION = "node_execution"
+    DRAFT_NODE_EXECUTION = "draft_node_execution"
+    MESSAGE_RUN = "message_run"
+    TOOL_EXECUTION = "tool_execution"
+    MODERATION_CHECK = "moderation_check"
+    SUGGESTED_QUESTION = "suggested_question"
+    DATASET_RETRIEVAL = "dataset_retrieval"
+    GENERATE_NAME = "generate_name"
+    PROMPT_GENERATION = "prompt_generation"
+    APP_CREATED = "app_created"
+    APP_UPDATED = "app_updated"
+    APP_DELETED = "app_deleted"
+    FEEDBACK_CREATED = "feedback_created"
+
+
+class SignalType(StrEnum):
+    """Signal routing type for telemetry cases."""
+
+    TRACE = "trace"
+    METRIC_LOG = "metric_log"
+
+
+class CaseRoute(BaseModel):
+    """Routing configuration for a telemetry case.
+
+    Attributes:
+        signal_type: The type of signal (trace or metric_log).
+        ce_eligible: Whether this case is eligible for community edition tracing.
+    """
+
+    signal_type: SignalType
+    ce_eligible: bool
+
+
+class TelemetryEnvelope(BaseModel):
+    """Envelope for telemetry events.
+
+    Attributes:
+        case: The telemetry case type.
+        tenant_id: The tenant identifier.
+        event_id: Unique event identifier for deduplication.
+        payload: The main event payload.
+        payload_fallback: Fallback payload (max 64KB).
+        metadata: Optional metadata dictionary.
+    """
+
+    case: TelemetryCase
+    tenant_id: str
+    event_id: str
+    payload: dict[str, Any]
+    payload_fallback: bytes | None = None
+    metadata: dict[str, Any] | None = None
+
+    @field_validator("payload_fallback")
+    @classmethod
+    def validate_payload_fallback_size(cls, v: bytes | None) -> bytes | None:
+        """Validate that payload_fallback does not exceed 64KB."""
+        if v is not None and len(v) > 65536:  # 64 * 1024
+            raise ValueError("payload_fallback must not exceed 64KB")
+        return v
+
+    class Config:
+        """Pydantic configuration."""
+
+        use_enum_values = False

api/enterprise/telemetry/draft_trace.py

@@ -0,0 +1,77 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+from core.telemetry import TelemetryContext, TelemetryEvent, TraceTaskName
+from core.telemetry import emit as telemetry_emit
+from core.workflow.enums import WorkflowNodeExecutionMetadataKey
+from models.workflow import WorkflowNodeExecutionModel
+
+
+def enqueue_draft_node_execution_trace(
+    *,
+    execution: WorkflowNodeExecutionModel,
+    outputs: Mapping[str, Any] | None,
+    workflow_execution_id: str | None,
+    user_id: str,
+) -> None:
+    node_data = _build_node_execution_data(
+        execution=execution,
+        outputs=outputs,
+        workflow_execution_id=workflow_execution_id,
+    )
+    telemetry_emit(
+        TelemetryEvent(
+            name=TraceTaskName.DRAFT_NODE_EXECUTION_TRACE,
+            context=TelemetryContext(
+                tenant_id=execution.tenant_id,
+                user_id=user_id,
+                app_id=execution.app_id,
+            ),
+            payload={"node_execution_data": node_data},
+        )
+    )
+
+
+def _build_node_execution_data(
+    *,
+    execution: WorkflowNodeExecutionModel,
+    outputs: Mapping[str, Any] | None,
+    workflow_execution_id: str | None,
+) -> dict[str, Any]:
+    metadata = execution.execution_metadata_dict
+    node_outputs = outputs if outputs is not None else execution.outputs_dict
+    execution_id = workflow_execution_id or execution.workflow_run_id or execution.id
+
+    return {
+        "workflow_id": execution.workflow_id,
+        "workflow_execution_id": execution_id,
+        "tenant_id": execution.tenant_id,
+        "app_id": execution.app_id,
+        "node_execution_id": execution.id,
+        "node_id": execution.node_id,
+        "node_type": execution.node_type,
+        "title": execution.title,
+        "status": execution.status,
+        "error": execution.error,
+        "elapsed_time": execution.elapsed_time,
+        "index": execution.index,
+        "predecessor_node_id": execution.predecessor_node_id,
+        "created_at": execution.created_at,
+        "finished_at": execution.finished_at,
+        "total_tokens": metadata.get(WorkflowNodeExecutionMetadataKey.TOTAL_TOKENS, 0),
+        "total_price": metadata.get(WorkflowNodeExecutionMetadataKey.TOTAL_PRICE, 0.0),
+        "currency": metadata.get(WorkflowNodeExecutionMetadataKey.CURRENCY),
+        "tool_name": (metadata.get(WorkflowNodeExecutionMetadataKey.TOOL_INFO) or {}).get("tool_name")
+        if isinstance(metadata.get(WorkflowNodeExecutionMetadataKey.TOOL_INFO), dict)
+        else None,
+        "iteration_id": metadata.get(WorkflowNodeExecutionMetadataKey.ITERATION_ID),
+        "iteration_index": metadata.get(WorkflowNodeExecutionMetadataKey.ITERATION_INDEX),
+        "loop_id": metadata.get(WorkflowNodeExecutionMetadataKey.LOOP_ID),
+        "loop_index": metadata.get(WorkflowNodeExecutionMetadataKey.LOOP_INDEX),
+        "parallel_id": metadata.get(WorkflowNodeExecutionMetadataKey.PARALLEL_ID),
+        "node_inputs": execution.inputs_dict,
+        "node_outputs": node_outputs,
+        "process_data": execution.process_data_dict,
+    }

api/enterprise/telemetry/enterprise_trace.py

@@ -0,0 +1,934 @@
+"""Enterprise trace handler — duck-typed, NOT a BaseTraceInstance subclass.
+
+Invoked directly in the Celery task, not through OpsTraceManager dispatch.
+Only requires a matching ``trace(trace_info)`` method signature.
+
+Signal strategy:
+- **Traces (spans)**: workflow run, node execution, draft node execution only.
+- **Metrics + structured logs**: all other event types.
+
+Token metric labels (unified structure):
+All token metrics (dify.tokens.input, dify.tokens.output, dify.tokens.total) use the
+same label set for consistent filtering and aggregation:
+- tenant_id: Tenant identifier
+- app_id: Application identifier
+- operation_type: Source of token usage (workflow | node_execution | message | rule_generate | etc.)
+- model_provider: LLM provider name (empty string if not applicable)
+- model_name: LLM model name (empty string if not applicable)
+- node_type: Workflow node type (empty string if not node_execution)
+
+This unified structure allows filtering by operation_type to separate:
+- Workflow-level aggregates (operation_type=workflow)
+- Individual node executions (operation_type=node_execution)
+- Direct message calls (operation_type=message)
+- Prompt generation operations (operation_type=rule_generate, code_generate, etc.)
+
+Without this, tokens are double-counted when querying totals (workflow totals include
+node totals, since workflow.total_tokens is the sum of all node tokens).
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from typing import Any, cast
+
+from opentelemetry.util.types import AttributeValue
+
+from core.ops.entities.trace_entity import (
+    BaseTraceInfo,
+    DatasetRetrievalTraceInfo,
+    DraftNodeExecutionTrace,
+    GenerateNameTraceInfo,
+    MessageTraceInfo,
+    ModerationTraceInfo,
+    OperationType,
+    PromptGenerationTraceInfo,
+    SuggestedQuestionTraceInfo,
+    ToolTraceInfo,
+    WorkflowNodeTraceInfo,
+    WorkflowTraceInfo,
+)
+from enterprise.telemetry.entities import (
+    EnterpriseTelemetryCounter,
+    EnterpriseTelemetryEvent,
+    EnterpriseTelemetryHistogram,
+    EnterpriseTelemetrySpan,
+    TokenMetricLabels,
+)
+from enterprise.telemetry.telemetry_log import emit_metric_only_event, emit_telemetry_log
+
+logger = logging.getLogger(__name__)
+
+
+class EnterpriseOtelTrace:
+    """Duck-typed enterprise trace handler.
+
+    ``*_trace`` methods emit spans (workflow/node only) or structured logs
+    (all other events), plus metrics at 100 % accuracy.
+    """
+
+    def __init__(self) -> None:
+        from extensions.ext_enterprise_telemetry import get_enterprise_exporter
+
+        exporter = get_enterprise_exporter()
+        if exporter is None:
+            raise RuntimeError("EnterpriseOtelTrace instantiated but exporter is not initialized")
+        self._exporter = exporter
+
+    def trace(self, trace_info: BaseTraceInfo) -> None:
+        if isinstance(trace_info, WorkflowTraceInfo):
+            self._workflow_trace(trace_info)
+        elif isinstance(trace_info, MessageTraceInfo):
+            self._message_trace(trace_info)
+        elif isinstance(trace_info, ToolTraceInfo):
+            self._tool_trace(trace_info)
+        elif isinstance(trace_info, DraftNodeExecutionTrace):
+            self._draft_node_execution_trace(trace_info)
+        elif isinstance(trace_info, WorkflowNodeTraceInfo):
+            self._node_execution_trace(trace_info)
+        elif isinstance(trace_info, ModerationTraceInfo):
+            self._moderation_trace(trace_info)
+        elif isinstance(trace_info, SuggestedQuestionTraceInfo):
+            self._suggested_question_trace(trace_info)
+        elif isinstance(trace_info, DatasetRetrievalTraceInfo):
+            self._dataset_retrieval_trace(trace_info)
+        elif isinstance(trace_info, GenerateNameTraceInfo):
+            self._generate_name_trace(trace_info)
+        elif isinstance(trace_info, PromptGenerationTraceInfo):
+            self._prompt_generation_trace(trace_info)
+
+    def _common_attrs(self, trace_info: BaseTraceInfo) -> dict[str, Any]:
+        metadata = self._metadata(trace_info)
+        tenant_id, app_id, user_id = self._context_ids(trace_info, metadata)
+        return {
+            "dify.trace_id": trace_info.resolved_trace_id,
+            "dify.tenant_id": tenant_id,
+            "dify.app_id": app_id,
+            "dify.app.name": metadata.get("app_name"),
+            "dify.workspace.name": metadata.get("workspace_name"),
+            "gen_ai.user.id": user_id,
+            "dify.message.id": trace_info.message_id,
+        }
+
+    def _metadata(self, trace_info: BaseTraceInfo) -> dict[str, Any]:
+        return trace_info.metadata
+
+    def _context_ids(
+        self,
+        trace_info: BaseTraceInfo,
+        metadata: dict[str, Any],
+    ) -> tuple[str | None, str | None, str | None]:
+        tenant_id = getattr(trace_info, "tenant_id", None) or metadata.get("tenant_id")
+        app_id = getattr(trace_info, "app_id", None) or metadata.get("app_id")
+        user_id = getattr(trace_info, "user_id", None) or metadata.get("user_id")
+        return tenant_id, app_id, user_id
+
+    def _labels(self, **values: AttributeValue) -> dict[str, AttributeValue]:
+        return dict(values)
+
+    def _safe_payload_value(self, value: Any) -> str | dict[str, Any] | list[object] | None:
+        if isinstance(value, str):
+            return value
+        if isinstance(value, dict):
+            return cast(dict[str, Any], value)
+        if isinstance(value, list):
+            items: list[object] = []
+            for item in cast(list[object], value):
+                items.append(item)
+            return items
+        return None
+
+    def _content_or_ref(self, value: Any, ref: str) -> Any:
+        if self._exporter.include_content:
+            return self._maybe_json(value)
+        return ref
+
+    def _maybe_json(self, value: Any) -> str | None:
+        if value is None:
+            return None
+        if isinstance(value, str):
+            return value
+        try:
+            return json.dumps(value, default=str)
+        except (TypeError, ValueError):
+            return str(value)
+
+    # ------------------------------------------------------------------
+    # SPAN-emitting handlers (workflow, node execution, draft node)
+    # ------------------------------------------------------------------
+
+    def _workflow_trace(self, info: WorkflowTraceInfo) -> None:
+        metadata = self._metadata(info)
+        tenant_id, app_id, user_id = self._context_ids(info, metadata)
+        # -- Slim span attrs: identity + structure + status + timing only --
+        span_attrs: dict[str, Any] = {
+            "dify.trace_id": info.resolved_trace_id,
+            "dify.tenant_id": tenant_id,
+            "dify.app_id": app_id,
+            "dify.workflow.id": info.workflow_id,
+            "dify.workflow.run_id": info.workflow_run_id,
+            "dify.workflow.status": info.workflow_run_status,
+            "dify.workflow.error": info.error,
+            "dify.workflow.elapsed_time": info.workflow_run_elapsed_time,
+            "dify.invoke_from": metadata.get("triggered_from"),
+            "dify.conversation.id": info.conversation_id,
+            "dify.message.id": info.message_id,
+            "dify.invoked_by": info.invoked_by,
+        }
+
+        trace_correlation_override: str | None = None
+        parent_span_id_source: str | None = None
+
+        parent_ctx = metadata.get("parent_trace_context")
+        if isinstance(parent_ctx, dict):
+            parent_ctx_dict = cast(dict[str, Any], parent_ctx)
+            span_attrs["dify.parent.trace_id"] = parent_ctx_dict.get("trace_id")
+            span_attrs["dify.parent.node.execution_id"] = parent_ctx_dict.get("parent_node_execution_id")
+            span_attrs["dify.parent.workflow.run_id"] = parent_ctx_dict.get("parent_workflow_run_id")
+            span_attrs["dify.parent.app.id"] = parent_ctx_dict.get("parent_app_id")
+
+            trace_override_value = parent_ctx_dict.get("parent_workflow_run_id")
+            if isinstance(trace_override_value, str):
+                trace_correlation_override = trace_override_value
+            parent_span_value = parent_ctx_dict.get("parent_node_execution_id")
+            if isinstance(parent_span_value, str):
+                parent_span_id_source = parent_span_value
+
+        self._exporter.export_span(
+            EnterpriseTelemetrySpan.WORKFLOW_RUN,
+            span_attrs,
+            correlation_id=info.workflow_run_id,
+            span_id_source=info.workflow_run_id,
+            start_time=info.start_time,
+            end_time=info.end_time,
+            trace_correlation_override=trace_correlation_override,
+            parent_span_id_source=parent_span_id_source,
+        )
+
+        # -- Companion log: ALL attrs (span + detail) for full picture --
+        log_attrs: dict[str, Any] = {**span_attrs}
+        log_attrs.update(
+            {
+                "dify.app.name": metadata.get("app_name"),
+                "dify.workspace.name": metadata.get("workspace_name"),
+                "gen_ai.user.id": user_id,
+                "gen_ai.usage.total_tokens": info.total_tokens,
+                "dify.workflow.version": info.workflow_run_version,
+            }
+        )
+
+        ref = f"ref:workflow_run_id={info.workflow_run_id}"
+        log_attrs["dify.workflow.inputs"] = self._content_or_ref(info.workflow_run_inputs, ref)
+        log_attrs["dify.workflow.outputs"] = self._content_or_ref(info.workflow_run_outputs, ref)
+        log_attrs["dify.workflow.query"] = self._content_or_ref(info.query, ref)
+
+        emit_telemetry_log(
+            event_name=EnterpriseTelemetryEvent.WORKFLOW_RUN,
+            attributes=log_attrs,
+            signal="span_detail",
+            trace_id_source=info.workflow_run_id,
+            span_id_source=info.workflow_run_id,
+            tenant_id=tenant_id,
+            user_id=user_id,
+        )
+
+        # -- Metrics --
+        labels = self._labels(
+            tenant_id=tenant_id or "",
+            app_id=app_id or "",
+        )
+        token_labels = TokenMetricLabels(
+            tenant_id=tenant_id or "",
+            app_id=app_id or "",
+            operation_type=OperationType.WORKFLOW,
+            model_provider="",
+            model_name="",
+            node_type="",
+        ).to_dict()
+        self._exporter.increment_counter(EnterpriseTelemetryCounter.TOKENS, info.total_tokens, token_labels)
+        if info.prompt_tokens is not None and info.prompt_tokens > 0:
+            self._exporter.increment_counter(EnterpriseTelemetryCounter.INPUT_TOKENS, info.prompt_tokens, token_labels)
+        if info.completion_tokens is not None and info.completion_tokens > 0:
+            self._exporter.increment_counter(
+                EnterpriseTelemetryCounter.OUTPUT_TOKENS, info.completion_tokens, token_labels
+            )
+        invoke_from = metadata.get("triggered_from", "")
+        self._exporter.increment_counter(
+            EnterpriseTelemetryCounter.REQUESTS,
+            1,
+            self._labels(
+                **labels,
+                type="workflow",
+                status=info.workflow_run_status,
+                invoke_from=invoke_from,
+            ),
+        )
+        self._exporter.record_histogram(
+            EnterpriseTelemetryHistogram.WORKFLOW_DURATION,
+            float(info.workflow_run_elapsed_time),
+            self._labels(
+                **labels,
+                status=info.workflow_run_status,
+            ),
+        )
+
+        if info.error:
+            self._exporter.increment_counter(
+                EnterpriseTelemetryCounter.ERRORS,
+                1,
+                self._labels(
+                    **labels,
+                    type="workflow",
+                ),
+            )
+
+    def _node_execution_trace(self, info: WorkflowNodeTraceInfo) -> None:
+        self._emit_node_execution_trace(info, EnterpriseTelemetrySpan.NODE_EXECUTION, "node")
+
+    def _draft_node_execution_trace(self, info: DraftNodeExecutionTrace) -> None:
+        self._emit_node_execution_trace(
+            info,
+            EnterpriseTelemetrySpan.DRAFT_NODE_EXECUTION,
+            "draft_node",
+            correlation_id_override=info.node_execution_id,
+            trace_correlation_override_param=info.workflow_run_id,
+        )
+
+    def _emit_node_execution_trace(
+        self,
+        info: WorkflowNodeTraceInfo,
+        span_name: EnterpriseTelemetrySpan,
+        request_type: str,
+        correlation_id_override: str | None = None,
+        trace_correlation_override_param: str | None = None,
+    ) -> None:
+        metadata = self._metadata(info)
+        tenant_id, app_id, user_id = self._context_ids(info, metadata)
+        # -- Slim span attrs: identity + structure + status + timing --
+        span_attrs: dict[str, Any] = {
+            "dify.trace_id": info.resolved_trace_id,
+            "dify.tenant_id": tenant_id,
+            "dify.app_id": app_id,
+            "dify.workflow.id": info.workflow_id,
+            "dify.workflow.run_id": info.workflow_run_id,
+            "dify.message.id": info.message_id,
+            "dify.conversation.id": metadata.get("conversation_id"),
+            "dify.node.execution_id": info.node_execution_id,
+            "dify.node.id": info.node_id,
+            "dify.node.type": info.node_type,
+            "dify.node.title": info.title,
+            "dify.node.status": info.status,
+            "dify.node.error": info.error,
+            "dify.node.elapsed_time": info.elapsed_time,
+            "dify.node.index": info.index,
+            "dify.node.predecessor_node_id": info.predecessor_node_id,
+            "dify.node.iteration_id": info.iteration_id,
+            "dify.node.loop_id": info.loop_id,
+            "dify.node.parallel_id": info.parallel_id,
+            "dify.node.invoked_by": info.invoked_by,
+        }
+
+        trace_correlation_override = trace_correlation_override_param
+        parent_ctx = metadata.get("parent_trace_context")
+        if isinstance(parent_ctx, dict):
+            parent_ctx_dict = cast(dict[str, Any], parent_ctx)
+            override_value = parent_ctx_dict.get("parent_workflow_run_id")
+            if isinstance(override_value, str):
+                trace_correlation_override = override_value
+
+        effective_correlation_id = correlation_id_override or info.workflow_run_id
+        self._exporter.export_span(
+            span_name,
+            span_attrs,
+            correlation_id=effective_correlation_id,
+            span_id_source=info.node_execution_id,
+            start_time=info.start_time,
+            end_time=info.end_time,
+            trace_correlation_override=trace_correlation_override,
+        )
+
+        # -- Companion log: ALL attrs (span + detail) --
+        log_attrs: dict[str, Any] = {**span_attrs}
+        log_attrs.update(
+            {
+                "dify.app.name": metadata.get("app_name"),
+                "dify.workspace.name": metadata.get("workspace_name"),
+                "dify.invoke_from": metadata.get("invoke_from"),
+                "gen_ai.user.id": user_id,
+                "gen_ai.usage.total_tokens": info.total_tokens,
+                "dify.node.total_price": info.total_price,
+                "dify.node.currency": info.currency,
+                "gen_ai.provider.name": info.model_provider,
+                "gen_ai.request.model": info.model_name,
+                "gen_ai.tool.name": info.tool_name,
+                "dify.node.iteration_index": info.iteration_index,
+                "dify.node.loop_index": info.loop_index,
+                "dify.plugin.name": metadata.get("plugin_name"),
+                "dify.credential.name": metadata.get("credential_name"),
+                "dify.credential.id": metadata.get("credential_id"),
+                "dify.dataset.ids": self._maybe_json(metadata.get("dataset_ids")),
+                "dify.dataset.names": self._maybe_json(metadata.get("dataset_names")),
+            }
+        )
+
+        ref = f"ref:node_execution_id={info.node_execution_id}"
+        log_attrs["dify.node.inputs"] = self._content_or_ref(info.node_inputs, ref)
+        log_attrs["dify.node.outputs"] = self._content_or_ref(info.node_outputs, ref)
+        log_attrs["dify.node.process_data"] = self._content_or_ref(info.process_data, ref)
+
+        emit_telemetry_log(
+            event_name=span_name.value,
+            attributes=log_attrs,
+            signal="span_detail",
+            trace_id_source=info.workflow_run_id,
+            span_id_source=info.node_execution_id,
+            tenant_id=tenant_id,
+            user_id=user_id,
+        )
+
+        # -- Metrics --
+        labels = self._labels(
+            tenant_id=tenant_id or "",
+            app_id=app_id or "",
+            node_type=info.node_type,
+            model_provider=info.model_provider or "",
+        )
+        if info.total_tokens:
+            token_labels = TokenMetricLabels(
+                tenant_id=tenant_id or "",
+                app_id=app_id or "",
+                operation_type=OperationType.NODE_EXECUTION,
+                model_provider=info.model_provider or "",
+                model_name=info.model_name or "",
+                node_type=info.node_type,
+            ).to_dict()
+            self._exporter.increment_counter(EnterpriseTelemetryCounter.TOKENS, info.total_tokens, token_labels)
+            if info.prompt_tokens is not None and info.prompt_tokens > 0:
+                self._exporter.increment_counter(
+                    EnterpriseTelemetryCounter.INPUT_TOKENS, info.prompt_tokens, token_labels
+                )
+            if info.completion_tokens is not None and info.completion_tokens > 0:
+                self._exporter.increment_counter(
+                    EnterpriseTelemetryCounter.OUTPUT_TOKENS, info.completion_tokens, token_labels
+                )
+        self._exporter.increment_counter(
+            EnterpriseTelemetryCounter.REQUESTS,
+            1,
+            self._labels(
+                **labels,
+                type=request_type,
+                status=info.status,
+                model_name=info.model_name or "",
+            ),
+        )
+        duration_labels = dict(labels)
+        duration_labels["model_name"] = info.model_name or ""
+        plugin_name = metadata.get("plugin_name")
+        if plugin_name and info.node_type in {"tool", "knowledge-retrieval"}:
+            duration_labels["plugin_name"] = plugin_name
+        self._exporter.record_histogram(EnterpriseTelemetryHistogram.NODE_DURATION, info.elapsed_time, duration_labels)
+
+        if info.error:
+            self._exporter.increment_counter(
+                EnterpriseTelemetryCounter.ERRORS,
+                1,
+                self._labels(
+                    **labels,
+                    type=request_type,
+                    model_name=info.model_name or "",
+                ),
+            )
+
+    # ------------------------------------------------------------------
+    # METRIC-ONLY handlers (structured log + counters/histograms)
+    # ------------------------------------------------------------------
+
+    def _message_trace(self, info: MessageTraceInfo) -> None:
+        metadata = self._metadata(info)
+        tenant_id, app_id, user_id = self._context_ids(info, metadata)
+        attrs = self._common_attrs(info)
+        attrs.update(
+            {
+                "dify.invoke_from": metadata.get("from_source"),
+                "dify.conversation.id": metadata.get("conversation_id"),
+                "dify.conversation.mode": info.conversation_mode,
+                "gen_ai.provider.name": metadata.get("ls_provider"),
+                "gen_ai.request.model": metadata.get("ls_model_name"),
+                "gen_ai.usage.input_tokens": info.message_tokens,
+                "gen_ai.usage.output_tokens": info.answer_tokens,
+                "gen_ai.usage.total_tokens": info.total_tokens,
+                "dify.message.status": metadata.get("status"),
+                "dify.message.error": info.error,
+                "dify.message.from_source": metadata.get("from_source"),
+                "dify.message.from_end_user_id": metadata.get("from_end_user_id"),
+                "dify.message.from_account_id": metadata.get("from_account_id"),
+                "dify.streaming": info.is_streaming_request,
+                "dify.message.time_to_first_token": info.gen_ai_server_time_to_first_token,
+                "dify.message.streaming_duration": info.llm_streaming_time_to_generate,
+                "dify.workflow.run_id": metadata.get("workflow_run_id"),
+            }
+        )
+        node_execution_id = metadata.get("node_execution_id")
+        if node_execution_id:
+            attrs["dify.node.execution_id"] = node_execution_id
+
+        ref = f"ref:message_id={info.message_id}"
+        inputs = self._safe_payload_value(info.inputs)
+        outputs = self._safe_payload_value(info.outputs)
+        attrs["dify.message.inputs"] = self._content_or_ref(inputs, ref)
+        attrs["dify.message.outputs"] = self._content_or_ref(outputs, ref)
+
+        emit_metric_only_event(
+            event_name=EnterpriseTelemetryEvent.MESSAGE_RUN,
+            attributes=attrs,
+            trace_id_source=metadata.get("workflow_run_id") or str(info.message_id) if info.message_id else None,
+            span_id_source=node_execution_id,
+            tenant_id=tenant_id,
+            user_id=user_id,
+        )
+
+        labels = self._labels(
+            tenant_id=tenant_id or "",
+            app_id=app_id or "",
+            model_provider=metadata.get("ls_provider", ""),
+            model_name=metadata.get("ls_model_name", ""),
+        )
+        token_labels = TokenMetricLabels(
+            tenant_id=tenant_id or "",
+            app_id=app_id or "",
+            operation_type=OperationType.MESSAGE,
+            model_provider=metadata.get("ls_provider", ""),
+            model_name=metadata.get("ls_model_name", ""),
+            node_type="",
+        ).to_dict()
+        self._exporter.increment_counter(EnterpriseTelemetryCounter.TOKENS, info.total_tokens, token_labels)
+        if info.message_tokens > 0:
+            self._exporter.increment_counter(EnterpriseTelemetryCounter.INPUT_TOKENS, info.message_tokens, token_labels)
+        if info.answer_tokens > 0:
+            self._exporter.increment_counter(EnterpriseTelemetryCounter.OUTPUT_TOKENS, info.answer_tokens, token_labels)
+        invoke_from = metadata.get("from_source", "")
+        self._exporter.increment_counter(
+            EnterpriseTelemetryCounter.REQUESTS,
+            1,
+            self._labels(
+                **labels,
+                type="message",
+                status=metadata.get("status", ""),
+                invoke_from=invoke_from,
+            ),
+        )
+
+        if info.start_time and info.end_time:
+            duration = (info.end_time - info.start_time).total_seconds()
+            self._exporter.record_histogram(EnterpriseTelemetryHistogram.MESSAGE_DURATION, duration, labels)
+
+        if info.gen_ai_server_time_to_first_token is not None:
+            self._exporter.record_histogram(
+                EnterpriseTelemetryHistogram.MESSAGE_TTFT, info.gen_ai_server_time_to_first_token, labels
+            )
+
+        if info.error:
+            self._exporter.increment_counter(
+                EnterpriseTelemetryCounter.ERRORS,
+                1,
+                self._labels(
+                    **labels,
+                    type="message",
+                ),
+            )
+
+    def _tool_trace(self, info: ToolTraceInfo) -> None:
+        metadata = self._metadata(info)
+        tenant_id, app_id, user_id = self._context_ids(info, metadata)
+        attrs = self._common_attrs(info)
+        attrs.update(
+            {
+                "gen_ai.tool.name": info.tool_name,
+                "dify.tool.time_cost": info.time_cost,
+                "dify.tool.error": info.error,
+                "dify.workflow.run_id": metadata.get("workflow_run_id"),
+            }
+        )
+        node_execution_id = metadata.get("node_execution_id")
+        if node_execution_id:
+            attrs["dify.node.execution_id"] = node_execution_id
+
+        ref = f"ref:message_id={info.message_id}"
+        attrs["dify.tool.inputs"] = self._content_or_ref(info.tool_inputs, ref)
+        attrs["dify.tool.outputs"] = self._content_or_ref(info.tool_outputs, ref)
+        attrs["dify.tool.parameters"] = self._content_or_ref(info.tool_parameters, ref)
+        attrs["dify.tool.config"] = self._content_or_ref(info.tool_config, ref)
+
+        emit_metric_only_event(
+            event_name=EnterpriseTelemetryEvent.TOOL_EXECUTION,
+            attributes=attrs,
+            trace_id_source=info.resolved_trace_id,
+            span_id_source=node_execution_id,
+            tenant_id=tenant_id,
+            user_id=user_id,
+        )
+
+        labels = self._labels(
+            tenant_id=tenant_id or "",
+            app_id=app_id or "",
+            tool_name=info.tool_name,
+        )
+        self._exporter.increment_counter(
+            EnterpriseTelemetryCounter.REQUESTS,
+            1,
+            self._labels(
+                **labels,
+                type="tool",
+            ),
+        )
+        self._exporter.record_histogram(EnterpriseTelemetryHistogram.TOOL_DURATION, float(info.time_cost), labels)
+
+        if info.error:
+            self._exporter.increment_counter(
+                EnterpriseTelemetryCounter.ERRORS,
+                1,
+                self._labels(
+                    **labels,
+                    type="tool",
+                ),
+            )
+
+    def _moderation_trace(self, info: ModerationTraceInfo) -> None:
+        metadata = self._metadata(info)
+        tenant_id, app_id, user_id = self._context_ids(info, metadata)
+        attrs = self._common_attrs(info)
+        attrs.update(
+            {
+                "dify.moderation.flagged": info.flagged,
+                "dify.moderation.action": info.action,
+                "dify.moderation.preset_response": info.preset_response,
+                "dify.workflow.run_id": metadata.get("workflow_run_id"),
+            }
+        )
+        node_execution_id = metadata.get("node_execution_id")
+        if node_execution_id:
+            attrs["dify.node.execution_id"] = node_execution_id
+
+        attrs["dify.moderation.query"] = self._content_or_ref(
+            info.query,
+            f"ref:message_id={info.message_id}",
+        )
+
+        emit_metric_only_event(
+            event_name=EnterpriseTelemetryEvent.MODERATION_CHECK,
+            attributes=attrs,
+            trace_id_source=info.resolved_trace_id,
+            span_id_source=node_execution_id,
+            tenant_id=tenant_id,
+            user_id=user_id,
+        )
+
+        labels = self._labels(
+            tenant_id=tenant_id or "",
+            app_id=app_id or "",
+        )
+        self._exporter.increment_counter(
+            EnterpriseTelemetryCounter.REQUESTS,
+            1,
+            self._labels(
+                **labels,
+                type="moderation",
+            ),
+        )
+
+    def _suggested_question_trace(self, info: SuggestedQuestionTraceInfo) -> None:
+        metadata = self._metadata(info)
+        tenant_id, app_id, user_id = self._context_ids(info, metadata)
+        attrs = self._common_attrs(info)
+        attrs.update(
+            {
+                "gen_ai.usage.total_tokens": info.total_tokens,
+                "dify.suggested_question.status": info.status,
+                "dify.suggested_question.error": info.error,
+                "gen_ai.provider.name": info.model_provider,
+                "gen_ai.request.model": info.model_id,
+                "dify.suggested_question.count": len(info.suggested_question),
+                "dify.workflow.run_id": metadata.get("workflow_run_id"),
+            }
+        )
+        node_execution_id = metadata.get("node_execution_id")
+        if node_execution_id:
+            attrs["dify.node.execution_id"] = node_execution_id
+
+        attrs["dify.suggested_question.questions"] = self._content_or_ref(
+            info.suggested_question,
+            f"ref:message_id={info.message_id}",
+        )
+
+        emit_metric_only_event(
+            event_name=EnterpriseTelemetryEvent.SUGGESTED_QUESTION_GENERATION,
+            attributes=attrs,
+            trace_id_source=info.resolved_trace_id,
+            span_id_source=node_execution_id,
+            tenant_id=tenant_id,
+            user_id=user_id,
+        )
+
+        labels = self._labels(
+            tenant_id=tenant_id or "",
+            app_id=app_id or "",
+        )
+        self._exporter.increment_counter(
+            EnterpriseTelemetryCounter.REQUESTS,
+            1,
+            self._labels(
+                **labels,
+                type="suggested_question",
+                model_provider=info.model_provider or "",
+                model_name=info.model_id or "",
+            ),
+        )
+
+    def _dataset_retrieval_trace(self, info: DatasetRetrievalTraceInfo) -> None:
+        metadata = self._metadata(info)
+        tenant_id, app_id, user_id = self._context_ids(info, metadata)
+        attrs = self._common_attrs(info)
+        attrs["dify.dataset.error"] = info.error
+        attrs["dify.workflow.run_id"] = metadata.get("workflow_run_id")
+        node_execution_id = metadata.get("node_execution_id")
+        if node_execution_id:
+            attrs["dify.node.execution_id"] = node_execution_id
+
+        docs: list[dict[str, Any]] = []
+        documents_any: Any = info.documents
+        documents_list: list[Any] = cast(list[Any], documents_any) if isinstance(documents_any, list) else []
+        for entry in documents_list:
+            if isinstance(entry, dict):
+                entry_dict: dict[str, Any] = cast(dict[str, Any], entry)
+                docs.append(entry_dict)
+        dataset_ids: list[str] = []
+        dataset_names: list[str] = []
+        structured_docs: list[dict[str, Any]] = []
+        for doc in docs:
+            meta_raw = doc.get("metadata")
+            meta: dict[str, Any] = cast(dict[str, Any], meta_raw) if isinstance(meta_raw, dict) else {}
+            did = meta.get("dataset_id")
+            dname = meta.get("dataset_name")
+            if did and did not in dataset_ids:
+                dataset_ids.append(did)
+            if dname and dname not in dataset_names:
+                dataset_names.append(dname)
+            structured_docs.append(
+                {
+                    "dataset_id": did,
+                    "document_id": meta.get("document_id"),
+                    "segment_id": meta.get("segment_id"),
+                    "score": meta.get("score"),
+                }
+            )
+
+        attrs["dify.dataset.ids"] = self._maybe_json(dataset_ids)
+        attrs["dify.dataset.names"] = self._maybe_json(dataset_names)
+        attrs["dify.retrieval.document_count"] = len(docs)
+
+        embedding_models_raw: Any = metadata.get("embedding_models")
+        embedding_models: dict[str, Any] = (
+            cast(dict[str, Any], embedding_models_raw) if isinstance(embedding_models_raw, dict) else {}
+        )
+        if embedding_models:
+            providers: list[str] = []
+            models: list[str] = []
+            for ds_info in embedding_models.values():
+                if isinstance(ds_info, dict):
+                    ds_info_dict: dict[str, Any] = cast(dict[str, Any], ds_info)
+                    p = ds_info_dict.get("embedding_model_provider", "")
+                    m = ds_info_dict.get("embedding_model", "")
+                    if p and p not in providers:
+                        providers.append(p)
+                    if m and m not in models:
+                        models.append(m)
+            attrs["dify.dataset.embedding_providers"] = self._maybe_json(providers)
+            attrs["dify.dataset.embedding_models"] = self._maybe_json(models)
+
+        # Add rerank model to logs
+        rerank_provider = metadata.get("rerank_model_provider", "")
+        rerank_model = metadata.get("rerank_model_name", "")
+        if rerank_provider or rerank_model:
+            attrs["dify.retrieval.rerank_provider"] = rerank_provider
+            attrs["dify.retrieval.rerank_model"] = rerank_model
+
+        ref = f"ref:message_id={info.message_id}"
+        retrieval_inputs = self._safe_payload_value(info.inputs)
+        attrs["dify.retrieval.query"] = self._content_or_ref(retrieval_inputs, ref)
+        attrs["dify.dataset.documents"] = self._content_or_ref(structured_docs, ref)
+
+        emit_metric_only_event(
+            event_name=EnterpriseTelemetryEvent.DATASET_RETRIEVAL,
+            attributes=attrs,
+            trace_id_source=metadata.get("workflow_run_id") or str(info.message_id) if info.message_id else None,
+            span_id_source=node_execution_id or (str(info.message_id) if info.message_id else None),
+            tenant_id=tenant_id,
+            user_id=user_id,
+        )
+
+        labels = self._labels(
+            tenant_id=tenant_id or "",
+            app_id=app_id or "",
+        )
+        self._exporter.increment_counter(
+            EnterpriseTelemetryCounter.REQUESTS,
+            1,
+            self._labels(
+                **labels,
+                type="dataset_retrieval",
+            ),
+        )
+
+        for did in dataset_ids:
+            # Get embedding model for this specific dataset
+            ds_embedding_info = embedding_models.get(did, {})
+            embedding_provider = ds_embedding_info.get("embedding_model_provider", "")
+            embedding_model = ds_embedding_info.get("embedding_model", "")
+
+            # Get rerank model (same for all datasets in this retrieval)
+            rerank_provider = metadata.get("rerank_model_provider", "")
+            rerank_model = metadata.get("rerank_model_name", "")
+
+            self._exporter.increment_counter(
+                EnterpriseTelemetryCounter.DATASET_RETRIEVALS,
+                1,
+                self._labels(
+                    **labels,
+                    dataset_id=did,
+                    embedding_model_provider=embedding_provider,
+                    embedding_model=embedding_model,
+                    rerank_model_provider=rerank_provider,
+                    rerank_model=rerank_model,
+                ),
+            )
+
+    def _generate_name_trace(self, info: GenerateNameTraceInfo) -> None:
+        metadata = self._metadata(info)
+        tenant_id, app_id, user_id = self._context_ids(info, metadata)
+        attrs = self._common_attrs(info)
+        attrs["dify.conversation.id"] = info.conversation_id
+        node_execution_id = metadata.get("node_execution_id")
+        if node_execution_id:
+            attrs["dify.node.execution_id"] = node_execution_id
+
+        ref = f"ref:conversation_id={info.conversation_id}"
+        inputs = self._safe_payload_value(info.inputs)
+        outputs = self._safe_payload_value(info.outputs)
+        attrs["dify.generate_name.inputs"] = self._content_or_ref(inputs, ref)
+        attrs["dify.generate_name.outputs"] = self._content_or_ref(outputs, ref)
+
+        emit_metric_only_event(
+            event_name=EnterpriseTelemetryEvent.GENERATE_NAME_EXECUTION,
+            attributes=attrs,
+            trace_id_source=info.resolved_trace_id,
+            span_id_source=node_execution_id,
+            tenant_id=tenant_id,
+            user_id=user_id,
+        )
+
+        labels = self._labels(
+            tenant_id=tenant_id or "",
+            app_id=app_id or "",
+        )
+        self._exporter.increment_counter(
+            EnterpriseTelemetryCounter.REQUESTS,
+            1,
+            self._labels(
+                **labels,
+                type="generate_name",
+            ),
+        )
+
+    def _prompt_generation_trace(self, info: PromptGenerationTraceInfo) -> None:
+        metadata = self._metadata(info)
+        tenant_id, app_id, user_id = self._context_ids(info, metadata)
+        attrs = {
+            "dify.trace_id": info.resolved_trace_id,
+            "dify.tenant_id": tenant_id,
+            "dify.user.id": user_id,
+            "dify.app.id": app_id or "",
+            "dify.app.name": metadata.get("app_name"),
+            "dify.workspace.name": metadata.get("workspace_name"),
+            "dify.operation.type": info.operation_type,
+            "gen_ai.provider.name": info.model_provider,
+            "gen_ai.request.model": info.model_name,
+            "gen_ai.usage.input_tokens": info.prompt_tokens,
+            "gen_ai.usage.output_tokens": info.completion_tokens,
+            "gen_ai.usage.total_tokens": info.total_tokens,
+            "dify.prompt_generation.latency": info.latency,
+            "dify.prompt_generation.error": info.error,
+        }
+        node_execution_id = metadata.get("node_execution_id")
+        if node_execution_id:
+            attrs["dify.node.execution_id"] = node_execution_id
+
+        if info.total_price is not None:
+            attrs["dify.prompt_generation.total_price"] = info.total_price
+            attrs["dify.prompt_generation.currency"] = info.currency
+
+        ref = f"ref:trace_id={info.trace_id}"
+        outputs = self._safe_payload_value(info.outputs)
+        attrs["dify.prompt_generation.instruction"] = self._content_or_ref(info.instruction, ref)
+        attrs["dify.prompt_generation.output"] = self._content_or_ref(outputs, ref)
+
+        emit_metric_only_event(
+            event_name=EnterpriseTelemetryEvent.PROMPT_GENERATION_EXECUTION,
+            attributes=attrs,
+            trace_id_source=info.resolved_trace_id,
+            span_id_source=node_execution_id,
+            tenant_id=tenant_id,
+            user_id=user_id,
+        )
+
+        token_labels = TokenMetricLabels(
+            tenant_id=tenant_id or "",
+            app_id=app_id or "",
+            operation_type=info.operation_type,
+            model_provider=info.model_provider,
+            model_name=info.model_name,
+            node_type="",
+        ).to_dict()
+
+        labels = self._labels(
+            tenant_id=tenant_id or "",
+            app_id=app_id or "",
+            operation_type=info.operation_type,
+            model_provider=info.model_provider,
+            model_name=info.model_name,
+        )
+
+        self._exporter.increment_counter(EnterpriseTelemetryCounter.TOKENS, info.total_tokens, token_labels)
+        if info.prompt_tokens > 0:
+            self._exporter.increment_counter(EnterpriseTelemetryCounter.INPUT_TOKENS, info.prompt_tokens, token_labels)
+        if info.completion_tokens > 0:
+            self._exporter.increment_counter(
+                EnterpriseTelemetryCounter.OUTPUT_TOKENS, info.completion_tokens, token_labels
+            )
+
+        status = "failed" if info.error else "success"
+        self._exporter.increment_counter(
+            EnterpriseTelemetryCounter.REQUESTS,
+            1,
+            self._labels(
+                **labels,
+                type="prompt_generation",
+                status=status,
+            ),
+        )
+
+        self._exporter.record_histogram(
+            EnterpriseTelemetryHistogram.PROMPT_GENERATION_DURATION,
+            info.latency,
+            labels,
+        )
+
+        if info.error:
+            self._exporter.increment_counter(
+                EnterpriseTelemetryCounter.ERRORS,
+                1,
+                self._labels(
+                    **labels,
+                    type="prompt_generation",
+                ),
+            )

api/enterprise/telemetry/entities/__init__.py

@@ -0,0 +1,121 @@
+from enum import StrEnum
+from typing import cast
+
+from opentelemetry.util.types import AttributeValue
+from pydantic import BaseModel, ConfigDict
+
+
+class EnterpriseTelemetrySpan(StrEnum):
+    WORKFLOW_RUN = "dify.workflow.run"
+    NODE_EXECUTION = "dify.node.execution"
+    DRAFT_NODE_EXECUTION = "dify.node.execution.draft"
+
+
+class EnterpriseTelemetryEvent(StrEnum):
+    """Event names for enterprise telemetry logs."""
+
+    APP_CREATED = "dify.app.created"
+    APP_UPDATED = "dify.app.updated"
+    APP_DELETED = "dify.app.deleted"
+    FEEDBACK_CREATED = "dify.feedback.created"
+    WORKFLOW_RUN = "dify.workflow.run"
+    MESSAGE_RUN = "dify.message.run"
+    TOOL_EXECUTION = "dify.tool.execution"
+    MODERATION_CHECK = "dify.moderation.check"
+    SUGGESTED_QUESTION_GENERATION = "dify.suggested_question.generation"
+    DATASET_RETRIEVAL = "dify.dataset.retrieval"
+    GENERATE_NAME_EXECUTION = "dify.generate_name.execution"
+    PROMPT_GENERATION_EXECUTION = "dify.prompt_generation.execution"
+    REHYDRATION_FAILED = "dify.telemetry.rehydration_failed"
+
+
+class EnterpriseTelemetryCounter(StrEnum):
+    TOKENS = "tokens"
+    INPUT_TOKENS = "input_tokens"
+    OUTPUT_TOKENS = "output_tokens"
+    REQUESTS = "requests"
+    ERRORS = "errors"
+    FEEDBACK = "feedback"
+    DATASET_RETRIEVALS = "dataset_retrievals"
+    APP_CREATED = "app_created"
+    APP_UPDATED = "app_updated"
+    APP_DELETED = "app_deleted"
+
+
+class EnterpriseTelemetryHistogram(StrEnum):
+    WORKFLOW_DURATION = "workflow_duration"
+    NODE_DURATION = "node_duration"
+    MESSAGE_DURATION = "message_duration"
+    MESSAGE_TTFT = "message_ttft"
+    TOOL_DURATION = "tool_duration"
+    PROMPT_GENERATION_DURATION = "prompt_generation_duration"
+
+
+class TokenMetricLabels(BaseModel):
+    """Unified label structure for all dify.token.* metrics.
+
+    All token counters (dify.tokens.input, dify.tokens.output, dify.tokens.total) MUST
+    use this exact label set to ensure consistent filtering and aggregation across
+    different operation types.
+
+    Attributes:
+        tenant_id: Tenant identifier.
+        app_id: Application identifier.
+        operation_type: Source of token usage (workflow | node_execution | message |
+            rule_generate | code_generate | structured_output | instruction_modify).
+        model_provider: LLM provider name. Empty string if not applicable (e.g., workflow-level).
+        model_name: LLM model name. Empty string if not applicable (e.g., workflow-level).
+        node_type: Workflow node type. Empty string unless operation_type=node_execution.
+
+    Usage:
+        labels = TokenMetricLabels(
+            tenant_id="tenant-123",
+            app_id="app-456",
+            operation_type=OperationType.WORKFLOW,
+            model_provider="",
+            model_name="",
+            node_type="",
+        )
+        exporter.increment_counter(
+            EnterpriseTelemetryCounter.INPUT_TOKENS,
+            100,
+            labels.to_dict()
+        )
+
+    Design rationale:
+        Without this unified structure, tokens get double-counted when querying totals
+        because workflow.total_tokens is already the sum of all node tokens. The
+        operation_type label allows filtering to separate workflow-level aggregates from
+        node-level detail, while keeping the same label cardinality for consistent queries.
+    """
+
+    tenant_id: str
+    app_id: str
+    operation_type: str
+    model_provider: str
+    model_name: str
+    node_type: str
+
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+    def to_dict(self) -> dict[str, AttributeValue]:
+        return cast(
+            dict[str, AttributeValue],
+            {
+                "tenant_id": self.tenant_id,
+                "app_id": self.app_id,
+                "operation_type": self.operation_type,
+                "model_provider": self.model_provider,
+                "model_name": self.model_name,
+                "node_type": self.node_type,
+            },
+        )
+
+
+__all__ = [
+    "EnterpriseTelemetryCounter",
+    "EnterpriseTelemetryEvent",
+    "EnterpriseTelemetryHistogram",
+    "EnterpriseTelemetrySpan",
+    "TokenMetricLabels",
+]

api/enterprise/telemetry/event_handlers.py

@@ -0,0 +1,84 @@
+"""Blinker signal handlers for enterprise telemetry.
+
+Registered at import time via ``@signal.connect`` decorators.
+Import must happen during ``ext_enterprise_telemetry.init_app()`` to
+ensure handlers fire.  Each handler delegates to ``core.telemetry.gateway``
+which handles routing, EE-gating, and dispatch.
+"""
+
+from __future__ import annotations
+
+import logging
+
+from events.app_event import app_was_created, app_was_deleted, app_was_updated
+from events.feedback_event import feedback_was_created
+
+logger = logging.getLogger(__name__)
+
+__all__ = [
+    "_handle_app_created",
+    "_handle_app_deleted",
+    "_handle_app_updated",
+    "_handle_feedback_created",
+]
+
+
+@app_was_created.connect
+def _handle_app_created(sender: object, **kwargs: object) -> None:
+    from core.telemetry.gateway import emit as gateway_emit
+    from enterprise.telemetry.contracts import TelemetryCase
+
+    gateway_emit(
+        case=TelemetryCase.APP_CREATED,
+        context={"tenant_id": str(getattr(sender, "tenant_id", "") or "")},
+        payload={
+            "app_id": getattr(sender, "id", None),
+            "mode": getattr(sender, "mode", None),
+        },
+    )
+
+
+@app_was_deleted.connect
+def _handle_app_deleted(sender: object, **kwargs: object) -> None:
+    from core.telemetry.gateway import emit as gateway_emit
+    from enterprise.telemetry.contracts import TelemetryCase
+
+    gateway_emit(
+        case=TelemetryCase.APP_DELETED,
+        context={"tenant_id": str(getattr(sender, "tenant_id", "") or "")},
+        payload={"app_id": getattr(sender, "id", None)},
+    )
+
+
+@app_was_updated.connect
+def _handle_app_updated(sender: object, **kwargs: object) -> None:
+    from core.telemetry.gateway import emit as gateway_emit
+    from enterprise.telemetry.contracts import TelemetryCase
+
+    gateway_emit(
+        case=TelemetryCase.APP_UPDATED,
+        context={"tenant_id": str(getattr(sender, "tenant_id", "") or "")},
+        payload={"app_id": getattr(sender, "id", None)},
+    )
+
+
+@feedback_was_created.connect
+def _handle_feedback_created(sender: object, **kwargs: object) -> None:
+    from core.telemetry.gateway import emit as gateway_emit
+    from enterprise.telemetry.contracts import TelemetryCase
+
+    tenant_id = str(kwargs.get("tenant_id", "") or "")
+    gateway_emit(
+        case=TelemetryCase.FEEDBACK_CREATED,
+        context={"tenant_id": tenant_id},
+        payload={
+            "message_id": getattr(sender, "message_id", None),
+            "app_id": getattr(sender, "app_id", None),
+            "conversation_id": getattr(sender, "conversation_id", None),
+            "from_end_user_id": getattr(sender, "from_end_user_id", None),
+            "from_account_id": getattr(sender, "from_account_id", None),
+            "rating": getattr(sender, "rating", None),
+            "from_source": getattr(sender, "from_source", None),
+            "content": getattr(sender, "content", None),
+        },
+    )

api/enterprise/telemetry/exporter.py

@@ -0,0 +1,267 @@
+"""Enterprise OTEL exporter — shared by EnterpriseOtelTrace, event handlers, and direct instrumentation.
+
+Uses dedicated TracerProvider and MeterProvider instances (configurable sampling,
+independent from ext_otel.py infrastructure).
+
+Initialized once during Flask extension init (single-threaded via ext_enterprise_telemetry.py).
+Accessed via ``ext_enterprise_telemetry.get_enterprise_exporter()`` from any thread/process.
+"""
+
+import logging
+import socket
+import uuid
+from datetime import datetime
+from typing import Any, cast
+
+from opentelemetry import trace
+from opentelemetry.context import Context
+from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter as GRPCMetricExporter
+from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter as GRPCSpanExporter
+from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter as HTTPMetricExporter
+from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter as HTTPSpanExporter
+from opentelemetry.sdk.metrics import MeterProvider
+from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
+from opentelemetry.sdk.resources import Resource
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import BatchSpanProcessor
+from opentelemetry.sdk.trace.sampling import ParentBasedTraceIdRatio
+from opentelemetry.semconv.resource import ResourceAttributes
+from opentelemetry.trace import SpanContext, TraceFlags
+from opentelemetry.util.types import Attributes, AttributeValue
+
+from configs import dify_config
+from enterprise.telemetry.entities import EnterpriseTelemetryCounter, EnterpriseTelemetryHistogram
+from enterprise.telemetry.id_generator import (
+    CorrelationIdGenerator,
+    compute_deterministic_span_id,
+    set_correlation_id,
+    set_span_id_source,
+)
+
+logger = logging.getLogger(__name__)
+
+
+def is_enterprise_telemetry_enabled() -> bool:
+    return bool(dify_config.ENTERPRISE_ENABLED and dify_config.ENTERPRISE_TELEMETRY_ENABLED)
+
+
+def _parse_otlp_headers(raw: str) -> dict[str, str]:
+    """Parse ``key=value,key2=value2`` into a dict."""
+    if not raw:
+        return {}
+    headers: dict[str, str] = {}
+    for pair in raw.split(","):
+        if "=" not in pair:
+            continue
+        k, v = pair.split("=", 1)
+        headers[k.strip()] = v.strip()
+    return headers
+
+
+def _datetime_to_ns(dt: datetime) -> int:
+    """Convert a datetime to nanoseconds since epoch (OTEL convention)."""
+    return int(dt.timestamp() * 1_000_000_000)
+
+
+class _ExporterFactory:
+    def __init__(self, protocol: str, endpoint: str, headers: dict[str, str], insecure: bool):
+        self._protocol = protocol
+        self._endpoint = endpoint
+        self._headers = headers
+        self._grpc_headers = tuple(headers.items()) if headers else None
+        self._http_headers = headers or None
+        self._insecure = insecure
+
+    def create_trace_exporter(self) -> HTTPSpanExporter | GRPCSpanExporter:
+        if self._protocol == "grpc":
+            return GRPCSpanExporter(
+                endpoint=self._endpoint or None,
+                headers=self._grpc_headers,
+                insecure=self._insecure,
+            )
+        trace_endpoint = f"{self._endpoint}/v1/traces" if self._endpoint else ""
+        return HTTPSpanExporter(endpoint=trace_endpoint or None, headers=self._http_headers)
+
+    def create_metric_exporter(self) -> HTTPMetricExporter | GRPCMetricExporter:
+        if self._protocol == "grpc":
+            return GRPCMetricExporter(
+                endpoint=self._endpoint or None,
+                headers=self._grpc_headers,
+                insecure=self._insecure,
+            )
+        metric_endpoint = f"{self._endpoint}/v1/metrics" if self._endpoint else ""
+        return HTTPMetricExporter(endpoint=metric_endpoint or None, headers=self._http_headers)
+
+
+class EnterpriseExporter:
+    """Shared OTEL exporter for all enterprise telemetry.
+
+    ``export_span`` creates spans with optional real timestamps, deterministic
+    span/trace IDs, and cross-workflow parent linking.
+    ``increment_counter`` / ``record_histogram`` emit OTEL metrics at 100% accuracy.
+    """
+
+    def __init__(self, config: object) -> None:
+        endpoint: str = getattr(config, "ENTERPRISE_OTLP_ENDPOINT", "")
+        headers_raw: str = getattr(config, "ENTERPRISE_OTLP_HEADERS", "")
+        protocol: str = (getattr(config, "ENTERPRISE_OTLP_PROTOCOL", "http") or "http").lower()
+        service_name: str = getattr(config, "ENTERPRISE_SERVICE_NAME", "dify")
+        sampling_rate: float = getattr(config, "ENTERPRISE_OTEL_SAMPLING_RATE", 1.0)
+        self.include_content: bool = getattr(config, "ENTERPRISE_INCLUDE_CONTENT", True)
+        api_key: str = getattr(config, "ENTERPRISE_OTLP_API_KEY", "")
+        
+        # Auto-detect TLS: https:// uses secure, everything else is insecure
+        insecure = not endpoint.startswith("https://")
+        
+        resource = Resource(
+            attributes={
+                ResourceAttributes.SERVICE_NAME: service_name,
+                ResourceAttributes.HOST_NAME: socket.gethostname(),
+            }
+        )
+        sampler = ParentBasedTraceIdRatio(sampling_rate)
+        id_generator = CorrelationIdGenerator()
+        self._tracer_provider = TracerProvider(resource=resource, sampler=sampler, id_generator=id_generator)
+
+        headers = _parse_otlp_headers(headers_raw)
+        if api_key:
+            if "authorization" in headers:
+                logger.warning(
+                    "ENTERPRISE_OTLP_API_KEY is set but ENTERPRISE_OTLP_HEADERS also contains "
+                    "'authorization'; the API key will take precedence."
+                )
+            headers["authorization"] = f"Bearer {api_key}"
+        factory = _ExporterFactory(protocol, endpoint, headers, insecure=insecure)
+
+        trace_exporter = factory.create_trace_exporter()
+        self._tracer_provider.add_span_processor(BatchSpanProcessor(trace_exporter))
+        self._tracer = self._tracer_provider.get_tracer("dify.enterprise")
+
+        metric_exporter = factory.create_metric_exporter()
+        self._meter_provider = MeterProvider(
+            resource=resource,
+            metric_readers=[PeriodicExportingMetricReader(metric_exporter)],
+        )
+        meter = self._meter_provider.get_meter("dify.enterprise")
+        self._counters = {
+            EnterpriseTelemetryCounter.TOKENS: meter.create_counter("dify.tokens.total", unit="{token}"),
+            EnterpriseTelemetryCounter.INPUT_TOKENS: meter.create_counter("dify.tokens.input", unit="{token}"),
+            EnterpriseTelemetryCounter.OUTPUT_TOKENS: meter.create_counter("dify.tokens.output", unit="{token}"),
+            EnterpriseTelemetryCounter.REQUESTS: meter.create_counter("dify.requests.total", unit="{request}"),
+            EnterpriseTelemetryCounter.ERRORS: meter.create_counter("dify.errors.total", unit="{error}"),
+            EnterpriseTelemetryCounter.FEEDBACK: meter.create_counter("dify.feedback.total", unit="{feedback}"),
+            EnterpriseTelemetryCounter.DATASET_RETRIEVALS: meter.create_counter(
+                "dify.dataset.retrievals.total", unit="{retrieval}"
+            ),
+            EnterpriseTelemetryCounter.APP_CREATED: meter.create_counter("dify.app.created.total", unit="{app}"),
+            EnterpriseTelemetryCounter.APP_UPDATED: meter.create_counter("dify.app.updated.total", unit="{app}"),
+            EnterpriseTelemetryCounter.APP_DELETED: meter.create_counter("dify.app.deleted.total", unit="{app}"),
+        }
+        self._histograms = {
+            EnterpriseTelemetryHistogram.WORKFLOW_DURATION: meter.create_histogram("dify.workflow.duration", unit="s"),
+            EnterpriseTelemetryHistogram.NODE_DURATION: meter.create_histogram("dify.node.duration", unit="s"),
+            EnterpriseTelemetryHistogram.MESSAGE_DURATION: meter.create_histogram("dify.message.duration", unit="s"),
+            EnterpriseTelemetryHistogram.MESSAGE_TTFT: meter.create_histogram(
+                "dify.message.time_to_first_token", unit="s"
+            ),
+            EnterpriseTelemetryHistogram.TOOL_DURATION: meter.create_histogram("dify.tool.duration", unit="s"),
+            EnterpriseTelemetryHistogram.PROMPT_GENERATION_DURATION: meter.create_histogram(
+                "dify.prompt_generation.duration", unit="s"
+            ),
+        }
+
+    def export_span(
+        self,
+        name: str,
+        attributes: dict[str, Any],
+        correlation_id: str | None = None,
+        span_id_source: str | None = None,
+        start_time: datetime | None = None,
+        end_time: datetime | None = None,
+        trace_correlation_override: str | None = None,
+        parent_span_id_source: str | None = None,
+    ) -> None:
+        """Export an OTEL span with optional deterministic IDs and real timestamps.
+
+        Args:
+            name: Span operation name.
+            attributes: Span attributes dict.
+            correlation_id: Source for trace_id derivation (groups spans in one trace).
+            span_id_source: Source for deterministic span_id (e.g. workflow_run_id or node_execution_id).
+            start_time: Real span start time. When None, uses current time.
+            end_time: Real span end time. When None, span ends immediately.
+            trace_correlation_override: Override trace_id source (for cross-workflow linking).
+                When set, trace_id is derived from this instead of ``correlation_id``.
+            parent_span_id_source: Override parent span_id source (for cross-workflow linking).
+                When set, parent span_id is derived from this value. When None and
+                ``correlation_id`` is set, parent is the workflow root span.
+        """
+        effective_trace_correlation = trace_correlation_override or correlation_id
+        set_correlation_id(effective_trace_correlation)
+        set_span_id_source(span_id_source)
+
+        try:
+            parent_context: Context | None = None
+            # A span is the "root" of its correlation group when span_id_source == correlation_id
+            # (i.e. a workflow root span).  All other spans are children.
+            if parent_span_id_source:
+                # Cross-workflow linking: parent is an explicit span (e.g. tool node in outer workflow)
+                parent_span_id = compute_deterministic_span_id(parent_span_id_source)
+                parent_trace_id = int(uuid.UUID(effective_trace_correlation)) if effective_trace_correlation else 0
+                if parent_trace_id:
+                    parent_span_context = SpanContext(
+                        trace_id=parent_trace_id,
+                        span_id=parent_span_id,
+                        is_remote=True,
+                        trace_flags=TraceFlags(TraceFlags.SAMPLED),
+                    )
+                    parent_context = trace.set_span_in_context(trace.NonRecordingSpan(parent_span_context))
+            elif correlation_id and correlation_id != span_id_source:
+                # Child span: parent is the correlation-group root (workflow root span)
+                parent_span_id = compute_deterministic_span_id(correlation_id)
+                parent_trace_id = int(uuid.UUID(effective_trace_correlation or correlation_id))
+                parent_span_context = SpanContext(
+                    trace_id=parent_trace_id,
+                    span_id=parent_span_id,
+                    is_remote=True,
+                    trace_flags=TraceFlags(TraceFlags.SAMPLED),
+                )
+                parent_context = trace.set_span_in_context(trace.NonRecordingSpan(parent_span_context))
+
+            span_start_time = _datetime_to_ns(start_time) if start_time is not None else None
+            span_end_on_exit = end_time is None
+
+            with self._tracer.start_as_current_span(
+                name,
+                context=parent_context,
+                start_time=span_start_time,
+                end_on_exit=span_end_on_exit,
+            ) as span:
+                for key, value in attributes.items():
+                    if value is not None:
+                        span.set_attribute(key, value)
+                if end_time is not None:
+                    span.end(end_time=_datetime_to_ns(end_time))
+        except Exception:
+            logger.exception("Failed to export span %s", name)
+        finally:
+            set_correlation_id(None)
+            set_span_id_source(None)
+
+    def increment_counter(
+        self, name: EnterpriseTelemetryCounter, value: int, labels: dict[str, AttributeValue]
+    ) -> None:
+        counter = self._counters.get(name)
+        if counter:
+            counter.add(value, cast(Attributes, labels))
+
+    def record_histogram(
+        self, name: EnterpriseTelemetryHistogram, value: float, labels: dict[str, AttributeValue]
+    ) -> None:
+        histogram = self._histograms.get(name)
+        if histogram:
+            histogram.record(value, cast(Attributes, labels))
+
+    def shutdown(self) -> None:
+        self._tracer_provider.shutdown()
+        self._meter_provider.shutdown()

api/enterprise/telemetry/id_generator.py

@@ -0,0 +1,76 @@
+"""Custom OTEL ID Generator for correlation-based trace/span ID derivation.
+
+Uses contextvars for thread-safe correlation_id -> trace_id mapping.
+When a span_id_source is set, the span_id is derived deterministically
+from that value, enabling any span to reference another as parent
+without depending on span creation order.
+"""
+
+import random
+import uuid
+from contextvars import ContextVar
+from typing import cast
+
+from opentelemetry.sdk.trace.id_generator import IdGenerator
+
+_correlation_id_context: ContextVar[str | None] = ContextVar("correlation_id", default=None)
+_span_id_source_context: ContextVar[str | None] = ContextVar("span_id_source", default=None)
+
+
+def set_correlation_id(correlation_id: str | None) -> None:
+    _correlation_id_context.set(correlation_id)
+
+
+def get_correlation_id() -> str | None:
+    return _correlation_id_context.get()
+
+
+def set_span_id_source(source_id: str | None) -> None:
+    """Set the source for deterministic span_id generation.
+
+    When set, ``generate_span_id()`` derives the span_id from this value
+    (lower 64 bits of the UUID).  Pass the ``workflow_run_id`` for workflow
+    root spans or ``node_execution_id`` for node spans.
+    """
+    _span_id_source_context.set(source_id)
+
+
+def compute_deterministic_span_id(source_id: str) -> int:
+    """Derive a deterministic span_id from any UUID string.
+
+    Uses the lower 64 bits of the UUID, guaranteeing non-zero output
+    (OTEL requires span_id != 0).
+    """
+    span_id = cast(int, uuid.UUID(source_id).int) & ((1 << 64) - 1)
+    return span_id if span_id != 0 else 1
+
+
+class CorrelationIdGenerator(IdGenerator):
+    """ID generator that derives trace_id and optionally span_id from context.
+
+    - trace_id: always derived from correlation_id (groups all spans in one trace)
+    - span_id: derived from span_id_source when set (enables deterministic
+      parent-child linking), otherwise random
+    """
+
+    def generate_trace_id(self) -> int:
+        correlation_id = _correlation_id_context.get()
+        if correlation_id:
+            try:
+                return cast(int, uuid.UUID(correlation_id).int)
+            except (ValueError, AttributeError):
+                pass
+        return random.getrandbits(128)
+
+    def generate_span_id(self) -> int:
+        source = _span_id_source_context.get()
+        if source:
+            try:
+                return compute_deterministic_span_id(source)
+            except (ValueError, AttributeError):
+                pass
+
+        span_id = random.getrandbits(64)
+        while span_id == 0:
+            span_id = random.getrandbits(64)
+        return span_id

api/enterprise/telemetry/metric_handler.py

@@ -0,0 +1,377 @@
+"""Enterprise metric/log event handler.
+
+This module processes metric and log telemetry events after they've been
+dequeued from the enterprise_telemetry Celery queue. It handles case routing,
+idempotency checking, and payload rehydration.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from enterprise.telemetry.contracts import TelemetryCase, TelemetryEnvelope
+from extensions.ext_redis import redis_client
+
+logger = logging.getLogger(__name__)
+
+
+class EnterpriseMetricHandler:
+    """Handler for enterprise metric and log telemetry events.
+
+    Processes envelopes from the enterprise_telemetry queue, routing each
+    case to the appropriate handler method. Implements idempotency checking
+    and payload rehydration with fallback.
+    """
+
+    def _increment_diagnostic_counter(self, counter_name: str, labels: dict[str, str] | None = None) -> None:
+        """Increment a diagnostic counter for operational monitoring.
+
+        Args:
+            counter_name: Name of the counter (e.g., 'processed_total', 'deduped_total').
+            labels: Optional labels for the counter.
+        """
+        try:
+            from extensions.ext_enterprise_telemetry import get_enterprise_exporter
+
+            exporter = get_enterprise_exporter()
+            if not exporter:
+                return
+
+            full_counter_name = f"enterprise_telemetry.handler.{counter_name}"
+            logger.debug(
+                "Diagnostic counter: %s, labels=%s",
+                full_counter_name,
+                labels or {},
+            )
+        except Exception:
+            logger.debug("Failed to increment diagnostic counter: %s", counter_name, exc_info=True)
+
+    def handle(self, envelope: TelemetryEnvelope) -> None:
+        """Main entry point for processing telemetry envelopes.
+
+        Args:
+            envelope: The telemetry envelope to process.
+        """
+        # Check for duplicate events
+        if self._is_duplicate(envelope):
+            logger.debug(
+                "Skipping duplicate event: tenant_id=%s, event_id=%s",
+                envelope.tenant_id,
+                envelope.event_id,
+            )
+            self._increment_diagnostic_counter("deduped_total")
+            return
+
+        # Route to appropriate handler based on case
+        case = envelope.case
+        if case == TelemetryCase.APP_CREATED:
+            self._on_app_created(envelope)
+            self._increment_diagnostic_counter("processed_total", {"case": "app_created"})
+        elif case == TelemetryCase.APP_UPDATED:
+            self._on_app_updated(envelope)
+            self._increment_diagnostic_counter("processed_total", {"case": "app_updated"})
+        elif case == TelemetryCase.APP_DELETED:
+            self._on_app_deleted(envelope)
+            self._increment_diagnostic_counter("processed_total", {"case": "app_deleted"})
+        elif case == TelemetryCase.FEEDBACK_CREATED:
+            self._on_feedback_created(envelope)
+            self._increment_diagnostic_counter("processed_total", {"case": "feedback_created"})
+        elif case == TelemetryCase.MESSAGE_RUN:
+            self._on_message_run(envelope)
+            self._increment_diagnostic_counter("processed_total", {"case": "message_run"})
+        elif case == TelemetryCase.TOOL_EXECUTION:
+            self._on_tool_execution(envelope)
+            self._increment_diagnostic_counter("processed_total", {"case": "tool_execution"})
+        elif case == TelemetryCase.MODERATION_CHECK:
+            self._on_moderation_check(envelope)
+            self._increment_diagnostic_counter("processed_total", {"case": "moderation_check"})
+        elif case == TelemetryCase.SUGGESTED_QUESTION:
+            self._on_suggested_question(envelope)
+            self._increment_diagnostic_counter("processed_total", {"case": "suggested_question"})
+        elif case == TelemetryCase.DATASET_RETRIEVAL:
+            self._on_dataset_retrieval(envelope)
+            self._increment_diagnostic_counter("processed_total", {"case": "dataset_retrieval"})
+        elif case == TelemetryCase.GENERATE_NAME:
+            self._on_generate_name(envelope)
+            self._increment_diagnostic_counter("processed_total", {"case": "generate_name"})
+        elif case == TelemetryCase.PROMPT_GENERATION:
+            self._on_prompt_generation(envelope)
+            self._increment_diagnostic_counter("processed_total", {"case": "prompt_generation"})
+        else:
+            logger.warning(
+                "Unknown telemetry case: %s (tenant_id=%s, event_id=%s)",
+                case,
+                envelope.tenant_id,
+                envelope.event_id,
+            )
+
+    def _is_duplicate(self, envelope: TelemetryEnvelope) -> bool:
+        """Check if this event has already been processed.
+
+        Uses Redis with TTL for deduplication. Returns True if duplicate,
+        False if first time seeing this event.
+
+        Args:
+            envelope: The telemetry envelope to check.
+
+        Returns:
+            True if this event_id has been seen before, False otherwise.
+        """
+        dedup_key = f"telemetry:dedup:{envelope.tenant_id}:{envelope.event_id}"
+
+        try:
+            # Atomic set-if-not-exists with 1h TTL
+            # Returns True if key was set (first time), None if already exists (duplicate)
+            was_set = redis_client.set(dedup_key, b"1", nx=True, ex=3600)
+            return was_set is None
+        except Exception:
+            # Fail open: if Redis is unavailable, process the event
+            # (prefer occasional duplicate over lost data)
+            logger.warning(
+                "Redis unavailable for deduplication check, processing event anyway: %s",
+                envelope.event_id,
+                exc_info=True,
+            )
+            return False
+
+    def _rehydrate(self, envelope: TelemetryEnvelope) -> dict[str, Any]:
+        """Rehydrate payload from reference or fallback.
+
+        Attempts to resolve payload_ref to full data. If that fails,
+        falls back to payload_fallback. If both fail, emits a degraded
+        event marker.
+
+        Args:
+            envelope: The telemetry envelope containing payload data.
+
+        Returns:
+            The rehydrated payload dictionary.
+        """
+        # For now, payload is directly in the envelope
+        # Future: implement payload_ref resolution from storage
+        payload = envelope.payload
+
+        if not payload and envelope.payload_fallback:
+            import pickle
+
+            try:
+                payload = pickle.loads(envelope.payload_fallback)  # noqa: S301
+                logger.debug("Used payload_fallback for event_id=%s", envelope.event_id)
+            except Exception:
+                logger.warning(
+                    "Failed to deserialize payload_fallback for event_id=%s",
+                    envelope.event_id,
+                    exc_info=True,
+                )
+
+        if not payload:
+            # Both ref and fallback failed - emit degraded event
+            logger.error(
+                "Payload rehydration failed for event_id=%s, tenant_id=%s, case=%s",
+                envelope.event_id,
+                envelope.tenant_id,
+                envelope.case,
+            )
+            # Emit degraded event marker
+            from enterprise.telemetry.entities import EnterpriseTelemetryEvent
+            from enterprise.telemetry.telemetry_log import emit_metric_only_event
+
+            emit_metric_only_event(
+                event_name=EnterpriseTelemetryEvent.REHYDRATION_FAILED,
+                attributes={
+                    "dify.tenant_id": envelope.tenant_id,
+                    "dify.event_id": envelope.event_id,
+                    "dify.case": envelope.case,
+                    "rehydration_failed": True,
+                },
+                tenant_id=envelope.tenant_id,
+            )
+            self._increment_diagnostic_counter("rehydration_failed_total")
+            return {}
+
+        return payload
+
+    # Stub methods for each metric/log case
+    # These will be implemented in later tasks with actual emission logic
+
+    def _on_app_created(self, envelope: TelemetryEnvelope) -> None:
+        """Handle app created event."""
+        from enterprise.telemetry.entities import EnterpriseTelemetryCounter, EnterpriseTelemetryEvent
+        from enterprise.telemetry.telemetry_log import emit_metric_only_event
+        from extensions.ext_enterprise_telemetry import get_enterprise_exporter
+
+        exporter = get_enterprise_exporter()
+        if not exporter:
+            logger.debug("No exporter available for APP_CREATED: event_id=%s", envelope.event_id)
+            return
+
+        payload = self._rehydrate(envelope)
+        if not payload:
+            return
+
+        attrs = {
+            "dify.app.id": payload.get("app_id"),
+            "dify.tenant_id": envelope.tenant_id,
+            "dify.event.id": envelope.event_id,
+            "dify.app.mode": payload.get("mode"),
+        }
+
+        emit_metric_only_event(
+            event_name=EnterpriseTelemetryEvent.APP_CREATED,
+            attributes=attrs,
+            tenant_id=envelope.tenant_id,
+        )
+        exporter.increment_counter(
+            EnterpriseTelemetryCounter.APP_CREATED,
+            1,
+            {
+                "tenant_id": envelope.tenant_id,
+                "app_id": str(payload.get("app_id", "")),
+                "mode": str(payload.get("mode", "")),
+            },
+        )
+
+    def _on_app_updated(self, envelope: TelemetryEnvelope) -> None:
+        """Handle app updated event."""
+        from enterprise.telemetry.entities import EnterpriseTelemetryCounter, EnterpriseTelemetryEvent
+        from enterprise.telemetry.telemetry_log import emit_metric_only_event
+        from extensions.ext_enterprise_telemetry import get_enterprise_exporter
+
+        exporter = get_enterprise_exporter()
+        if not exporter:
+            logger.debug("No exporter available for APP_UPDATED: event_id=%s", envelope.event_id)
+            return
+
+        payload = self._rehydrate(envelope)
+        if not payload:
+            return
+
+        attrs = {
+            "dify.app.id": payload.get("app_id"),
+            "dify.tenant_id": envelope.tenant_id,
+            "dify.event.id": envelope.event_id,
+        }
+
+        emit_metric_only_event(
+            event_name=EnterpriseTelemetryEvent.APP_UPDATED,
+            attributes=attrs,
+            tenant_id=envelope.tenant_id,
+        )
+        exporter.increment_counter(
+            EnterpriseTelemetryCounter.APP_UPDATED,
+            1,
+            {
+                "tenant_id": envelope.tenant_id,
+                "app_id": str(payload.get("app_id", "")),
+            },
+        )
+
+    def _on_app_deleted(self, envelope: TelemetryEnvelope) -> None:
+        """Handle app deleted event."""
+        from enterprise.telemetry.entities import EnterpriseTelemetryCounter, EnterpriseTelemetryEvent
+        from enterprise.telemetry.telemetry_log import emit_metric_only_event
+        from extensions.ext_enterprise_telemetry import get_enterprise_exporter
+
+        exporter = get_enterprise_exporter()
+        if not exporter:
+            logger.debug("No exporter available for APP_DELETED: event_id=%s", envelope.event_id)
+            return
+
+        payload = self._rehydrate(envelope)
+        if not payload:
+            return
+
+        attrs = {
+            "dify.app.id": payload.get("app_id"),
+            "dify.tenant_id": envelope.tenant_id,
+            "dify.event.id": envelope.event_id,
+        }
+
+        emit_metric_only_event(
+            event_name=EnterpriseTelemetryEvent.APP_DELETED,
+            attributes=attrs,
+            tenant_id=envelope.tenant_id,
+        )
+        exporter.increment_counter(
+            EnterpriseTelemetryCounter.APP_DELETED,
+            1,
+            {
+                "tenant_id": envelope.tenant_id,
+                "app_id": str(payload.get("app_id", "")),
+            },
+        )
+
+    def _on_feedback_created(self, envelope: TelemetryEnvelope) -> None:
+        """Handle feedback created event."""
+        from enterprise.telemetry.entities import EnterpriseTelemetryCounter, EnterpriseTelemetryEvent
+        from enterprise.telemetry.telemetry_log import emit_metric_only_event
+        from extensions.ext_enterprise_telemetry import get_enterprise_exporter
+
+        exporter = get_enterprise_exporter()
+        if not exporter:
+            logger.debug("No exporter available for FEEDBACK_CREATED: event_id=%s", envelope.event_id)
+            return
+
+        payload = self._rehydrate(envelope)
+        if not payload:
+            return
+
+        include_content = exporter.include_content
+        attrs: dict = {
+            "dify.message.id": payload.get("message_id"),
+            "dify.tenant_id": envelope.tenant_id,
+            "dify.event.id": envelope.event_id,
+            "dify.app_id": payload.get("app_id"),
+            "dify.conversation.id": payload.get("conversation_id"),
+            "gen_ai.user.id": payload.get("from_end_user_id") or payload.get("from_account_id"),
+            "dify.feedback.rating": payload.get("rating"),
+            "dify.feedback.from_source": payload.get("from_source"),
+        }
+        if include_content:
+            attrs["dify.feedback.content"] = payload.get("content")
+
+        user_id = payload.get("from_end_user_id") or payload.get("from_account_id")
+        emit_metric_only_event(
+            event_name=EnterpriseTelemetryEvent.FEEDBACK_CREATED,
+            attributes=attrs,
+            tenant_id=envelope.tenant_id,
+            user_id=str(user_id or ""),
+        )
+        exporter.increment_counter(
+            EnterpriseTelemetryCounter.FEEDBACK,
+            1,
+            {
+                "tenant_id": envelope.tenant_id,
+                "app_id": str(payload.get("app_id", "")),
+                "rating": str(payload.get("rating", "")),
+            },
+        )
+
+    def _on_message_run(self, envelope: TelemetryEnvelope) -> None:
+        """Handle message run event (stub)."""
+        logger.debug("Processing MESSAGE_RUN: event_id=%s", envelope.event_id)
+
+    def _on_tool_execution(self, envelope: TelemetryEnvelope) -> None:
+        """Handle tool execution event (stub)."""
+        logger.debug("Processing TOOL_EXECUTION: event_id=%s", envelope.event_id)
+
+    def _on_moderation_check(self, envelope: TelemetryEnvelope) -> None:
+        """Handle moderation check event (stub)."""
+        logger.debug("Processing MODERATION_CHECK: event_id=%s", envelope.event_id)
+
+    def _on_suggested_question(self, envelope: TelemetryEnvelope) -> None:
+        """Handle suggested question event (stub)."""
+        logger.debug("Processing SUGGESTED_QUESTION: event_id=%s", envelope.event_id)
+
+    def _on_dataset_retrieval(self, envelope: TelemetryEnvelope) -> None:
+        """Handle dataset retrieval event (stub)."""
+        logger.debug("Processing DATASET_RETRIEVAL: event_id=%s", envelope.event_id)
+
+    def _on_generate_name(self, envelope: TelemetryEnvelope) -> None:
+        """Handle generate name event (stub)."""
+        logger.debug("Processing GENERATE_NAME: event_id=%s", envelope.event_id)
+
+    def _on_prompt_generation(self, envelope: TelemetryEnvelope) -> None:
+        """Handle prompt generation event (stub)."""
+        logger.debug("Processing PROMPT_GENERATION: event_id=%s", envelope.event_id)

api/enterprise/telemetry/telemetry_log.py

@@ -0,0 +1,122 @@
+"""Structured-log emitter for enterprise telemetry events.
+
+Emits structured JSON log lines correlated with OTEL traces via trace_id.
+Picked up by ``StructuredJSONFormatter`` → stdout/Loki/Elastic.
+"""
+
+from __future__ import annotations
+
+import logging
+import uuid
+from functools import lru_cache
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from enterprise.telemetry.entities import EnterpriseTelemetryEvent
+
+logger = logging.getLogger("dify.telemetry")
+
+
+@lru_cache(maxsize=4096)
+def compute_trace_id_hex(uuid_str: str | None) -> str:
+    """Convert a business UUID string to a 32-hex OTEL-compatible trace_id.
+
+    Returns empty string when *uuid_str* is ``None`` or invalid.
+    """
+    if not uuid_str:
+        return ""
+    normalized = uuid_str.strip().lower()
+    if len(normalized) == 32 and all(ch in "0123456789abcdef" for ch in normalized):
+        return normalized
+    try:
+        return f"{uuid.UUID(normalized).int:032x}"
+    except (ValueError, AttributeError):
+        return ""
+
+
+@lru_cache(maxsize=4096)
+def compute_span_id_hex(uuid_str: str | None) -> str:
+    if not uuid_str:
+        return ""
+    normalized = uuid_str.strip().lower()
+    if len(normalized) == 16 and all(ch in "0123456789abcdef" for ch in normalized):
+        return normalized
+    try:
+        from enterprise.telemetry.id_generator import compute_deterministic_span_id
+
+        return f"{compute_deterministic_span_id(normalized):016x}"
+    except (ValueError, AttributeError):
+        return ""
+
+
+def emit_telemetry_log(
+    *,
+    event_name: str | EnterpriseTelemetryEvent,
+    attributes: dict[str, Any],
+    signal: str = "metric_only",
+    trace_id_source: str | None = None,
+    span_id_source: str | None = None,
+    tenant_id: str | None = None,
+    user_id: str | None = None,
+) -> None:
+    """Emit a structured log line for a telemetry event.
+
+    Parameters
+    ----------
+    event_name:
+        Canonical event name, e.g. ``"dify.workflow.run"``.
+    attributes:
+        All event-specific attributes (already built by the caller).
+    signal:
+        ``"metric_only"`` for events with no span, ``"span_detail"``
+        for detail logs accompanying a slim span.
+    trace_id_source:
+        A UUID string (e.g. ``workflow_run_id``) used to derive a 32-hex
+        trace_id for cross-signal correlation.
+    tenant_id:
+        Tenant identifier (for the ``IdentityContextFilter``).
+    user_id:
+        User identifier (for the ``IdentityContextFilter``).
+    """
+    if not logger.isEnabledFor(logging.INFO):
+        return
+    attrs = {
+        "dify.event.name": event_name,
+        "dify.event.signal": signal,
+        **attributes,
+    }
+
+    extra: dict[str, Any] = {"attributes": attrs}
+
+    trace_id_hex = compute_trace_id_hex(trace_id_source)
+    if trace_id_hex:
+        extra["trace_id"] = trace_id_hex
+    span_id_hex = compute_span_id_hex(span_id_source)
+    if span_id_hex:
+        extra["span_id"] = span_id_hex
+    if tenant_id:
+        extra["tenant_id"] = tenant_id
+    if user_id:
+        extra["user_id"] = user_id
+
+    logger.info("telemetry.%s", signal, extra=extra)
+
+
+def emit_metric_only_event(
+    *,
+    event_name: str | EnterpriseTelemetryEvent,
+    attributes: dict[str, Any],
+    trace_id_source: str | None = None,
+    span_id_source: str | None = None,
+    tenant_id: str | None = None,
+    user_id: str | None = None,
+) -> None:
+    emit_telemetry_log(
+        event_name=event_name,
+        attributes=attributes,
+        signal="metric_only",
+        trace_id_source=trace_id_source,
+        span_id_source=span_id_source,
+        tenant_id=tenant_id,
+        user_id=user_id,
+    )

api/events/app_event.py

@@ -3,6 +3,12 @@ from blinker import signal
 # sender: app
 app_was_created = signal("app-was-created")
 
+# sender: app
+app_was_deleted = signal("app-was-deleted")
+
+# sender: app
+app_was_updated = signal("app-was-updated")
+
 # sender: app, kwargs: app_model_config
 app_model_config_was_updated = signal("app-model-config-was-updated")
 

api/events/feedback_event.py

@@ -0,0 +1,4 @@
+from blinker import signal
+
+# sender: MessageFeedback, kwargs: tenant_id
+feedback_was_created = signal("feedback-was-created")

api/extensions/ext_enterprise_telemetry.py

@@ -0,0 +1,50 @@
+"""Flask extension for enterprise telemetry lifecycle management.
+
+Initializes the EnterpriseExporter singleton during ``create_app()``
+(single-threaded), registers blinker event handlers, and hooks atexit
+for graceful shutdown.
+
+Skipped entirely when ``ENTERPRISE_ENABLED`` and ``ENTERPRISE_TELEMETRY_ENABLED``
+are false (``is_enabled()`` gate).
+"""
+
+from __future__ import annotations
+
+import atexit
+import logging
+from typing import TYPE_CHECKING
+
+from configs import dify_config
+
+if TYPE_CHECKING:
+    from dify_app import DifyApp
+    from enterprise.telemetry.exporter import EnterpriseExporter
+
+logger = logging.getLogger(__name__)
+
+_exporter: EnterpriseExporter | None = None
+
+
+def is_enabled() -> bool:
+    return bool(dify_config.ENTERPRISE_ENABLED and dify_config.ENTERPRISE_TELEMETRY_ENABLED)
+
+
+def init_app(app: DifyApp) -> None:
+    global _exporter
+
+    if not is_enabled():
+        return
+
+    from enterprise.telemetry.exporter import EnterpriseExporter
+
+    _exporter = EnterpriseExporter(dify_config)
+    atexit.register(_exporter.shutdown)
+
+    # Import to trigger @signal.connect decorator registration
+    import enterprise.telemetry.event_handlers  # noqa: F401  # type: ignore[reportUnusedImport]
+
+    logger.info("Enterprise telemetry initialized")
+
+
+def get_enterprise_exporter() -> EnterpriseExporter | None:
+    return _exporter

api/extensions/ext_otel.py

@@ -78,16 +78,20 @@ def init_app(app: DifyApp):
     protocol = (dify_config.OTEL_EXPORTER_OTLP_PROTOCOL or "").lower()
     if dify_config.OTEL_EXPORTER_TYPE == "otlp":
         if protocol == "grpc":
+            # Auto-detect TLS: https:// uses secure, everything else is insecure
+            endpoint = dify_config.OTLP_BASE_ENDPOINT
+            insecure = not endpoint.startswith("https://")
+            
             exporter = GRPCSpanExporter(
-                endpoint=dify_config.OTLP_BASE_ENDPOINT,
+                endpoint=endpoint,
                 # Header field names must consist of lowercase letters, check RFC7540
-                headers=(("authorization", f"Bearer {dify_config.OTLP_API_KEY}"),),
-                insecure=True,
+                headers=((("authorization", f"Bearer {dify_config.OTLP_API_KEY}"),) if dify_config.OTLP_API_KEY else None),
+                insecure=insecure,
             )
             metric_exporter = GRPCMetricExporter(
-                endpoint=dify_config.OTLP_BASE_ENDPOINT,
-                headers=(("authorization", f"Bearer {dify_config.OTLP_API_KEY}"),),
-                insecure=True,
+                endpoint=endpoint,
+                headers=((("authorization", f"Bearer {dify_config.OTLP_API_KEY}"),) if dify_config.OTLP_API_KEY else None),
+                insecure=insecure,
             )
         else:
             headers = {"Authorization": f"Bearer {dify_config.OTLP_API_KEY}"} if dify_config.OTLP_API_KEY else None

api/extensions/otel/semconv/dify.py

@@ -21,3 +21,15 @@ class DifySpanAttributes:
 
     INVOKE_FROM = "dify.invoke_from"
     """Invocation source, e.g. SERVICE_API, WEB_APP, DEBUGGER."""
+
+    INVOKED_BY = "dify.invoked_by"
+    """Invoked by, e.g. end_user, account, user."""
+
+    USAGE_INPUT_TOKENS = "gen_ai.usage.input_tokens"
+    """Number of input tokens (prompt tokens) used."""
+
+    USAGE_OUTPUT_TOKENS = "gen_ai.usage.output_tokens"
+    """Number of output tokens (completion tokens) generated."""
+
+    USAGE_TOTAL_TOKENS = "gen_ai.usage.total_tokens"
+    """Total number of tokens used."""

api/services/account_service.py

@@ -289,6 +289,12 @@ class AccountService:
 
         TenantService.create_owner_tenant_if_not_exist(account=account)
 
+        # Enterprise-only: best-effort add the account to the default workspace (does not switch current workspace).
+        if dify_config.ENTERPRISE_ENABLED:
+            from services.enterprise.enterprise_service import try_join_default_workspace
+
+            try_join_default_workspace(str(account.id))
+
         return account
 
     @staticmethod
@@ -1407,6 +1413,12 @@ class RegisterService:
                 tenant_was_created.send(tenant)
 
             db.session.commit()
+
+            # Enterprise-only: best-effort add the account to the default workspace (does not switch current workspace).
+            if dify_config.ENTERPRISE_ENABLED:
+                from services.enterprise.enterprise_service import try_join_default_workspace
+
+                try_join_default_workspace(str(account.id))
         except WorkSpaceNotAllowedCreateError:
             db.session.rollback()
             logger.exception("Register failed")

api/services/enterprise/base.py

@@ -39,6 +39,9 @@ class BaseRequest:
         endpoint: str,
         json: Any | None = None,
         params: Mapping[str, Any] | None = None,
+        *,
+        timeout: float | httpx.Timeout | None = None,
+        raise_for_status: bool = False,
     ) -> Any:
         headers = {"Content-Type": "application/json", cls.secret_key_header: cls.secret_key}
         url = f"{cls.base_url}{endpoint}"
@@ -53,7 +56,16 @@ class BaseRequest:
             logger.debug("Failed to generate traceparent header", exc_info=True)
 
         with httpx.Client(mounts=mounts) as client:
-            response = client.request(method, url, json=json, params=params, headers=headers)
+            # IMPORTANT:
+            # - In httpx, passing timeout=None disables timeouts (infinite) and overrides the library default.
+            # - To preserve httpx's default timeout behavior for existing call sites, only pass the kwarg when set.
+            request_kwargs: dict[str, Any] = {"json": json, "params": params, "headers": headers}
+            if timeout is not None:
+                request_kwargs["timeout"] = timeout
+
+            response = client.request(method, url, **request_kwargs)
+            if raise_for_status:
+                response.raise_for_status()
         return response.json()
 
 

api/services/enterprise/enterprise_service.py

@@ -1,9 +1,16 @@
+import logging
+import uuid
 from datetime import datetime
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, ConfigDict, Field, model_validator
 
+from configs import dify_config
 from services.enterprise.base import EnterpriseRequest
 
+logger = logging.getLogger(__name__)
+
+DEFAULT_WORKSPACE_JOIN_TIMEOUT_SECONDS = 1.0
+
 
 class WebAppSettings(BaseModel):
     access_mode: str = Field(
@@ -30,6 +37,55 @@ class WorkspacePermission(BaseModel):
     )
 
 
+class DefaultWorkspaceJoinResult(BaseModel):
+    """
+    Result of ensuring an account is a member of the enterprise default workspace.
+
+    - joined=True is idempotent (already a member also returns True)
+    - joined=False means enterprise default workspace is not configured or invalid/archived
+    """
+
+    workspace_id: str = Field(default="", alias="workspaceId")
+    joined: bool
+    message: str
+
+    model_config = ConfigDict(extra="forbid", populate_by_name=True)
+
+    @model_validator(mode="after")
+    def _check_workspace_id_when_joined(self) -> "DefaultWorkspaceJoinResult":
+        if self.joined and not self.workspace_id:
+            raise ValueError("workspace_id must be non-empty when joined is True")
+        return self
+
+
+def try_join_default_workspace(account_id: str) -> None:
+    """
+    Enterprise-only side-effect: ensure account is a member of the default workspace.
+
+    This is a best-effort integration. Failures must not block user registration.
+    """
+
+    if not dify_config.ENTERPRISE_ENABLED:
+        return
+
+    try:
+        result = EnterpriseService.join_default_workspace(account_id=account_id)
+        if result.joined:
+            logger.info(
+                "Joined enterprise default workspace for account %s (workspace_id=%s)",
+                account_id,
+                result.workspace_id,
+            )
+        else:
+            logger.info(
+                "Skipped joining enterprise default workspace for account %s (message=%s)",
+                account_id,
+                result.message,
+            )
+    except Exception:
+        logger.warning("Failed to join enterprise default workspace for account %s", account_id, exc_info=True)
+
+
 class EnterpriseService:
     @classmethod
     def get_info(cls):
@@ -39,6 +95,34 @@ class EnterpriseService:
     def get_workspace_info(cls, tenant_id: str):
         return EnterpriseRequest.send_request("GET", f"/workspace/{tenant_id}/info")
 
+    @classmethod
+    def join_default_workspace(cls, *, account_id: str) -> DefaultWorkspaceJoinResult:
+        """
+        Call enterprise inner API to add an account to the default workspace.
+
+        NOTE: EnterpriseRequest.base_url is expected to already include the `/inner/api` prefix,
+        so the endpoint here is `/default-workspace/members`.
+        """
+
+        # Ensure we are sending a UUID-shaped string (enterprise side validates too).
+        try:
+            uuid.UUID(account_id)
+        except ValueError as e:
+            raise ValueError(f"account_id must be a valid UUID: {account_id}") from e
+
+        data = EnterpriseRequest.send_request(
+            "POST",
+            "/default-workspace/members",
+            json={"account_id": account_id},
+            timeout=DEFAULT_WORKSPACE_JOIN_TIMEOUT_SECONDS,
+            raise_for_status=True,
+        )
+        if not isinstance(data, dict):
+            raise ValueError("Invalid response format from enterprise default workspace API")
+        if "joined" not in data or "message" not in data:
+            raise ValueError("Invalid response payload from enterprise default workspace API")
+        return DefaultWorkspaceJoinResult.model_validate(data)
+
     @classmethod
     def get_app_sso_settings_last_update_time(cls) -> datetime:
         data = EnterpriseRequest.send_request("GET", "/sso/app/last-update-time")

api/tasks/enterprise_telemetry_task.py

@@ -0,0 +1,52 @@
+"""Celery worker for enterprise metric/log telemetry events.
+
+This module defines the Celery task that processes telemetry envelopes
+from the enterprise_telemetry queue. It deserializes envelopes and
+dispatches them to the EnterpriseMetricHandler.
+"""
+
+import json
+import logging
+
+from celery import shared_task
+
+from enterprise.telemetry.contracts import TelemetryEnvelope
+from enterprise.telemetry.metric_handler import EnterpriseMetricHandler
+
+logger = logging.getLogger(__name__)
+
+
+@shared_task(queue="enterprise_telemetry")
+def process_enterprise_telemetry(envelope_json: str) -> None:
+    """Process enterprise metric/log telemetry envelope.
+
+    This task is enqueued by the TelemetryGateway for metric/log-only
+    events. It deserializes the envelope and dispatches to the handler.
+
+    Best-effort processing: logs errors but never raises, to avoid
+    failing user requests due to telemetry issues.
+
+    Args:
+        envelope_json: JSON-serialized TelemetryEnvelope.
+    """
+    try:
+        # Deserialize envelope
+        envelope_dict = json.loads(envelope_json)
+        envelope = TelemetryEnvelope.model_validate(envelope_dict)
+
+        # Process through handler
+        handler = EnterpriseMetricHandler()
+        handler.handle(envelope)
+
+        logger.debug(
+            "Successfully processed telemetry envelope: tenant_id=%s, event_id=%s, case=%s",
+            envelope.tenant_id,
+            envelope.event_id,
+            envelope.case,
+        )
+    except Exception:
+        # Best-effort: log and drop on error, never fail user request
+        logger.warning(
+            "Failed to process enterprise telemetry envelope, dropping event",
+            exc_info=True,
+        )

api/tasks/ops_trace_task.py

@@ -39,12 +39,24 @@ def process_trace_tasks(file_info):
         trace_info["documents"] = [Document.model_validate(doc) for doc in trace_info["documents"]]
 
     try:
+        trace_type = trace_info_info_map.get(trace_info_type)
+        if trace_type:
+            trace_info = trace_type(**trace_info)
+
+        from extensions.ext_enterprise_telemetry import is_enabled as is_ee_telemetry_enabled
+
+        if is_ee_telemetry_enabled():
+            from enterprise.telemetry.enterprise_trace import EnterpriseOtelTrace
+
+            try:
+                EnterpriseOtelTrace().trace(trace_info)
+            except Exception:
+                logger.warning("Enterprise trace failed for app_id: %s", app_id, exc_info=True)
+
         if trace_instance:
             with current_app.app_context():
-                trace_type = trace_info_info_map.get(trace_info_type)
-                if trace_type:
-                    trace_info = trace_type(**trace_info)
                 trace_instance.trace(trace_info)
+
         logger.info("Processing trace tasks success, app_id: %s", app_id)
     except Exception as e:
         logger.info("error:\n\n\n%s\n\n\n\n", e)
@@ -52,4 +64,12 @@ def process_trace_tasks(file_info):
         redis_client.incr(failed_key)
         logger.info("Processing trace tasks failed, app_id: %s", app_id)
     finally:
-        storage.delete(file_path)
+        try:
+            storage.delete(file_path)
+        except Exception as e:
+            logger.warning(
+                "Failed to delete trace file %s for app_id %s: %s",
+                file_path,
+                app_id,
+                e,
+            )

api/tests/unit_tests/core/ops/test_trace_queue_manager.py

@@ -0,0 +1,200 @@
+"""Unit tests for TraceQueueManager telemetry guard.
+
+This test suite verifies that TraceQueueManager correctly drops trace tasks
+when telemetry is disabled, proving Bug 1 from code review is a false positive.
+
+The guard logic moved from persistence.py to TraceQueueManager.add_trace_task()
+at line 1282 of ops_trace_manager.py:
+    if self._enterprise_telemetry_enabled or self.trace_instance:
+        trace_task.app_id = self.app_id
+        trace_manager_queue.put(trace_task)
+
+Tasks are only enqueued if EITHER:
+- Enterprise telemetry is enabled (_enterprise_telemetry_enabled=True), OR
+- A third-party trace instance (Langfuse, etc.) is configured
+
+When BOTH are false, tasks are silently dropped (correct behavior).
+"""
+
+import queue
+import sys
+import types
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+@pytest.fixture
+def trace_queue_manager_and_task(monkeypatch):
+    """Fixture to provide TraceQueueManager and TraceTask with delayed imports."""
+    module_name = "core.ops.ops_trace_manager"
+    if module_name not in sys.modules:
+        ops_stub = types.ModuleType(module_name)
+
+        class StubTraceTask:
+            def __init__(self, trace_type):
+                self.trace_type = trace_type
+                self.app_id = None
+
+        class StubTraceQueueManager:
+            def __init__(self, app_id=None):
+                self.app_id = app_id
+                from core.telemetry.gateway import is_enterprise_telemetry_enabled
+
+                self._enterprise_telemetry_enabled = is_enterprise_telemetry_enabled()
+                self.trace_instance = StubOpsTraceManager.get_ops_trace_instance(app_id)
+
+            def add_trace_task(self, trace_task):
+                if self._enterprise_telemetry_enabled or self.trace_instance:
+                    trace_task.app_id = self.app_id
+                    from core.ops.ops_trace_manager import trace_manager_queue
+
+                    trace_manager_queue.put(trace_task)
+
+        class StubOpsTraceManager:
+            @staticmethod
+            def get_ops_trace_instance(app_id):
+                return None
+
+        ops_stub.TraceQueueManager = StubTraceQueueManager
+        ops_stub.TraceTask = StubTraceTask
+        ops_stub.OpsTraceManager = StubOpsTraceManager
+        ops_stub.trace_manager_queue = MagicMock(spec=queue.Queue)
+        monkeypatch.setitem(sys.modules, module_name, ops_stub)
+
+    from core.ops.entities.trace_entity import TraceTaskName
+
+    ops_module = __import__(module_name, fromlist=["TraceQueueManager", "TraceTask"])
+    TraceQueueManager = ops_module.TraceQueueManager
+    TraceTask = ops_module.TraceTask
+
+    return TraceQueueManager, TraceTask, TraceTaskName
+
+
+class TestTraceQueueManagerTelemetryGuard:
+    """Test TraceQueueManager's telemetry guard in add_trace_task()."""
+
+    def test_task_not_enqueued_when_telemetry_disabled_and_no_trace_instance(self, trace_queue_manager_and_task):
+        """Verify task is NOT enqueued when telemetry disabled and no trace instance.
+
+        This is the core guard: when _enterprise_telemetry_enabled=False AND
+        trace_instance=None, the task should be silently dropped.
+        """
+        TraceQueueManager, TraceTask, TraceTaskName = trace_queue_manager_and_task
+
+        mock_queue = MagicMock(spec=queue.Queue)
+
+        trace_task = TraceTask(trace_type=TraceTaskName.WORKFLOW_TRACE)
+
+        with (
+            patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=False),
+            patch("core.ops.ops_trace_manager.OpsTraceManager.get_ops_trace_instance", return_value=None),
+            patch("core.ops.ops_trace_manager.trace_manager_queue", mock_queue),
+        ):
+            manager = TraceQueueManager(app_id="test-app-id")
+            manager.add_trace_task(trace_task)
+
+            mock_queue.put.assert_not_called()
+
+    def test_task_enqueued_when_telemetry_enabled(self, trace_queue_manager_and_task):
+        """Verify task IS enqueued when enterprise telemetry is enabled.
+
+        When _enterprise_telemetry_enabled=True, the task should be enqueued
+        regardless of trace_instance state.
+        """
+        TraceQueueManager, TraceTask, TraceTaskName = trace_queue_manager_and_task
+
+        mock_queue = MagicMock(spec=queue.Queue)
+
+        trace_task = TraceTask(trace_type=TraceTaskName.WORKFLOW_TRACE)
+
+        with (
+            patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=True),
+            patch("core.ops.ops_trace_manager.OpsTraceManager.get_ops_trace_instance", return_value=None),
+            patch("core.ops.ops_trace_manager.trace_manager_queue", mock_queue),
+        ):
+            manager = TraceQueueManager(app_id="test-app-id")
+            manager.add_trace_task(trace_task)
+
+            mock_queue.put.assert_called_once()
+            called_task = mock_queue.put.call_args[0][0]
+            assert called_task.app_id == "test-app-id"
+
+    def test_task_enqueued_when_trace_instance_configured(self, trace_queue_manager_and_task):
+        """Verify task IS enqueued when third-party trace instance is configured.
+
+        When trace_instance is not None (e.g., Langfuse configured), the task
+        should be enqueued even if enterprise telemetry is disabled.
+        """
+        TraceQueueManager, TraceTask, TraceTaskName = trace_queue_manager_and_task
+
+        mock_queue = MagicMock(spec=queue.Queue)
+
+        mock_trace_instance = MagicMock()
+
+        trace_task = TraceTask(trace_type=TraceTaskName.WORKFLOW_TRACE)
+
+        with (
+            patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=False),
+            patch(
+                "core.ops.ops_trace_manager.OpsTraceManager.get_ops_trace_instance", return_value=mock_trace_instance
+            ),
+            patch("core.ops.ops_trace_manager.trace_manager_queue", mock_queue),
+        ):
+            manager = TraceQueueManager(app_id="test-app-id")
+            manager.add_trace_task(trace_task)
+
+            mock_queue.put.assert_called_once()
+            called_task = mock_queue.put.call_args[0][0]
+            assert called_task.app_id == "test-app-id"
+
+    def test_task_enqueued_when_both_telemetry_and_trace_instance_enabled(self, trace_queue_manager_and_task):
+        """Verify task IS enqueued when both telemetry and trace instance are enabled.
+
+        When both _enterprise_telemetry_enabled=True AND trace_instance is set,
+        the task should definitely be enqueued.
+        """
+        TraceQueueManager, TraceTask, TraceTaskName = trace_queue_manager_and_task
+
+        mock_queue = MagicMock(spec=queue.Queue)
+
+        mock_trace_instance = MagicMock()
+
+        trace_task = TraceTask(trace_type=TraceTaskName.WORKFLOW_TRACE)
+
+        with (
+            patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=True),
+            patch(
+                "core.ops.ops_trace_manager.OpsTraceManager.get_ops_trace_instance", return_value=mock_trace_instance
+            ),
+            patch("core.ops.ops_trace_manager.trace_manager_queue", mock_queue),
+        ):
+            manager = TraceQueueManager(app_id="test-app-id")
+            manager.add_trace_task(trace_task)
+
+            mock_queue.put.assert_called_once()
+            called_task = mock_queue.put.call_args[0][0]
+            assert called_task.app_id == "test-app-id"
+
+    def test_app_id_set_before_enqueue(self, trace_queue_manager_and_task):
+        """Verify app_id is set on the task before enqueuing.
+
+        The guard logic sets trace_task.app_id = self.app_id before calling
+        trace_manager_queue.put(trace_task). This test verifies that behavior.
+        """
+        TraceQueueManager, TraceTask, TraceTaskName = trace_queue_manager_and_task
+
+        mock_queue = MagicMock(spec=queue.Queue)
+
+        trace_task = TraceTask(trace_type=TraceTaskName.WORKFLOW_TRACE)
+
+        with (
+            patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=True),
+            patch("core.ops.ops_trace_manager.OpsTraceManager.get_ops_trace_instance", return_value=None),
+            patch("core.ops.ops_trace_manager.trace_manager_queue", mock_queue),
+        ):
+            manager = TraceQueueManager(app_id="expected-app-id")
+            manager.add_trace_task(trace_task)
+
+            called_task = mock_queue.put.call_args[0][0]
+            assert called_task.app_id == "expected-app-id"

api/tests/unit_tests/core/telemetry/test_facade.py

@@ -0,0 +1,181 @@
+"""Unit tests for core.telemetry.emit() routing and enterprise-only filtering."""
+
+from __future__ import annotations
+
+import queue
+import sys
+import types
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from core.ops.entities.trace_entity import TraceTaskName
+from core.telemetry.events import TelemetryContext, TelemetryEvent
+
+
+@pytest.fixture
+def telemetry_test_setup(monkeypatch):
+    module_name = "core.ops.ops_trace_manager"
+    ops_stub = types.ModuleType(module_name)
+
+    class StubTraceTask:
+        def __init__(self, trace_type, **kwargs):
+            self.trace_type = trace_type
+            self.app_id = None
+            self.kwargs = kwargs
+
+    class StubTraceQueueManager:
+        def __init__(self, app_id=None, user_id=None):
+            self.app_id = app_id
+            self.user_id = user_id
+            self.trace_instance = StubOpsTraceManager.get_ops_trace_instance(app_id)
+
+        def add_trace_task(self, trace_task):
+            trace_task.app_id = self.app_id
+            from core.ops.ops_trace_manager import trace_manager_queue
+
+            trace_manager_queue.put(trace_task)
+
+    class StubOpsTraceManager:
+        @staticmethod
+        def get_ops_trace_instance(app_id):
+            return None
+
+    ops_stub.TraceQueueManager = StubTraceQueueManager
+    ops_stub.TraceTask = StubTraceTask
+    ops_stub.OpsTraceManager = StubOpsTraceManager
+    ops_stub.trace_manager_queue = MagicMock(spec=queue.Queue)
+    monkeypatch.setitem(sys.modules, module_name, ops_stub)
+
+    from core.telemetry import emit
+
+    return emit, ops_stub.trace_manager_queue
+
+
+class TestTelemetryEmit:
+    @patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=True)
+    def test_emit_enterprise_trace_creates_trace_task(self, _mock_ee, telemetry_test_setup):
+        emit_fn, mock_queue = telemetry_test_setup
+
+        event = TelemetryEvent(
+            name=TraceTaskName.DRAFT_NODE_EXECUTION_TRACE,
+            context=TelemetryContext(
+                tenant_id="test-tenant",
+                user_id="test-user",
+                app_id="test-app",
+            ),
+            payload={"key": "value"},
+        )
+
+        emit_fn(event)
+
+        mock_queue.put.assert_called_once()
+        called_task = mock_queue.put.call_args[0][0]
+        assert called_task.trace_type == TraceTaskName.DRAFT_NODE_EXECUTION_TRACE
+
+    def test_emit_community_trace_enqueued(self, telemetry_test_setup):
+        emit_fn, mock_queue = telemetry_test_setup
+
+        event = TelemetryEvent(
+            name=TraceTaskName.WORKFLOW_TRACE,
+            context=TelemetryContext(
+                tenant_id="test-tenant",
+                user_id="test-user",
+                app_id="test-app",
+            ),
+            payload={},
+        )
+
+        emit_fn(event)
+
+        mock_queue.put.assert_called_once()
+
+    def test_emit_enterprise_only_trace_dropped_when_ee_disabled(self, telemetry_test_setup):
+        emit_fn, mock_queue = telemetry_test_setup
+
+        event = TelemetryEvent(
+            name=TraceTaskName.DRAFT_NODE_EXECUTION_TRACE,
+            context=TelemetryContext(
+                tenant_id="test-tenant",
+                user_id="test-user",
+                app_id="test-app",
+            ),
+            payload={},
+        )
+
+        emit_fn(event)
+
+        mock_queue.put.assert_not_called()
+
+    @patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=True)
+    def test_emit_all_enterprise_only_traces_allowed_when_ee_enabled(self, _mock_ee, telemetry_test_setup):
+        emit_fn, mock_queue = telemetry_test_setup
+
+        enterprise_only_traces = [
+            TraceTaskName.DRAFT_NODE_EXECUTION_TRACE,
+            TraceTaskName.NODE_EXECUTION_TRACE,
+            TraceTaskName.PROMPT_GENERATION_TRACE,
+        ]
+
+        for trace_name in enterprise_only_traces:
+            mock_queue.reset_mock()
+
+            event = TelemetryEvent(
+                name=trace_name,
+                context=TelemetryContext(
+                    tenant_id="test-tenant",
+                    user_id="test-user",
+                    app_id="test-app",
+                ),
+                payload={},
+            )
+
+            emit_fn(event)
+
+            mock_queue.put.assert_called_once()
+            called_task = mock_queue.put.call_args[0][0]
+            assert called_task.trace_type == trace_name
+
+    @patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=True)
+    def test_emit_passes_name_directly_to_trace_task(self, _mock_ee, telemetry_test_setup):
+        emit_fn, mock_queue = telemetry_test_setup
+
+        event = TelemetryEvent(
+            name=TraceTaskName.DRAFT_NODE_EXECUTION_TRACE,
+            context=TelemetryContext(
+                tenant_id="test-tenant",
+                user_id="test-user",
+                app_id="test-app",
+            ),
+            payload={"extra": "data"},
+        )
+
+        emit_fn(event)
+
+        mock_queue.put.assert_called_once()
+        called_task = mock_queue.put.call_args[0][0]
+        assert called_task.trace_type == TraceTaskName.DRAFT_NODE_EXECUTION_TRACE
+        assert isinstance(called_task.trace_type, TraceTaskName)
+
+    @patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=True)
+    def test_emit_with_provided_trace_manager(self, _mock_ee, telemetry_test_setup):
+        emit_fn, mock_queue = telemetry_test_setup
+
+        mock_trace_manager = MagicMock()
+        mock_trace_manager.add_trace_task = MagicMock()
+
+        event = TelemetryEvent(
+            name=TraceTaskName.NODE_EXECUTION_TRACE,
+            context=TelemetryContext(
+                tenant_id="test-tenant",
+                user_id="test-user",
+                app_id="test-app",
+            ),
+            payload={},
+        )
+
+        emit_fn(event, trace_manager=mock_trace_manager)
+
+        mock_trace_manager.add_trace_task.assert_called_once()
+        called_task = mock_trace_manager.add_trace_task.call_args[0][0]
+        assert called_task.trace_type == TraceTaskName.NODE_EXECUTION_TRACE

api/tests/unit_tests/core/telemetry/test_gateway_integration.py

@@ -0,0 +1,225 @@
+from __future__ import annotations
+
+import sys
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from core.telemetry.gateway import emit, is_enterprise_telemetry_enabled
+from enterprise.telemetry.contracts import TelemetryCase
+
+
+class TestTelemetryCoreExports:
+    def test_is_enterprise_telemetry_enabled_exported(self) -> None:
+        from core.telemetry.gateway import is_enterprise_telemetry_enabled as exported_func
+
+        assert callable(exported_func)
+
+
+@pytest.fixture
+def mock_ops_trace_manager():
+    mock_module = MagicMock()
+    mock_trace_task_class = MagicMock()
+    mock_trace_task_class.return_value = MagicMock()
+    mock_module.TraceTask = mock_trace_task_class
+    mock_module.TraceQueueManager = MagicMock()
+
+    mock_trace_entity = MagicMock()
+    mock_trace_task_name = MagicMock()
+    mock_trace_task_name.return_value = "workflow"
+    mock_trace_entity.TraceTaskName = mock_trace_task_name
+
+    with (
+        patch.dict(sys.modules, {"core.ops.ops_trace_manager": mock_module}),
+        patch.dict(sys.modules, {"core.ops.entities.trace_entity": mock_trace_entity}),
+    ):
+        yield mock_module, mock_trace_entity
+
+
+class TestGatewayIntegrationTraceRouting:
+    @pytest.fixture
+    def mock_trace_manager(self) -> MagicMock:
+        return MagicMock()
+
+    @pytest.mark.usefixtures("mock_ops_trace_manager")
+    def test_ce_eligible_trace_routed_to_trace_manager(
+        self,
+        mock_trace_manager: MagicMock,
+    ) -> None:
+        with patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=True):
+            context = {"app_id": "app-123", "user_id": "user-456", "tenant_id": "tenant-789"}
+            payload = {"workflow_run_id": "run-abc"}
+
+            emit(TelemetryCase.WORKFLOW_RUN, context, payload, mock_trace_manager)
+
+            mock_trace_manager.add_trace_task.assert_called_once()
+
+    @pytest.mark.usefixtures("mock_ops_trace_manager")
+    def test_ce_eligible_trace_routed_when_ee_disabled(
+        self,
+        mock_trace_manager: MagicMock,
+    ) -> None:
+        with patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=False):
+            context = {"app_id": "app-123", "user_id": "user-456"}
+            payload = {"workflow_run_id": "run-abc"}
+
+            emit(TelemetryCase.WORKFLOW_RUN, context, payload, mock_trace_manager)
+
+            mock_trace_manager.add_trace_task.assert_called_once()
+
+    @pytest.mark.usefixtures("mock_ops_trace_manager")
+    def test_enterprise_only_trace_dropped_when_ee_disabled(
+        self,
+        mock_trace_manager: MagicMock,
+    ) -> None:
+        with patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=False):
+            context = {"app_id": "app-123", "user_id": "user-456"}
+            payload = {"node_id": "node-abc"}
+
+            emit(TelemetryCase.NODE_EXECUTION, context, payload, mock_trace_manager)
+
+            mock_trace_manager.add_trace_task.assert_not_called()
+
+    @pytest.mark.usefixtures("mock_ops_trace_manager")
+    def test_enterprise_only_trace_routed_when_ee_enabled(
+        self,
+        mock_trace_manager: MagicMock,
+    ) -> None:
+        with patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=True):
+            context = {"app_id": "app-123", "user_id": "user-456"}
+            payload = {"node_id": "node-abc"}
+
+            emit(TelemetryCase.NODE_EXECUTION, context, payload, mock_trace_manager)
+
+            mock_trace_manager.add_trace_task.assert_called_once()
+
+
+class TestGatewayIntegrationMetricRouting:
+    @patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=True)
+    def test_metric_case_routes_to_celery_task(
+        self,
+        _mock_ee_enabled: MagicMock,
+    ) -> None:
+        from enterprise.telemetry.contracts import TelemetryEnvelope
+
+        with patch("tasks.enterprise_telemetry_task.process_enterprise_telemetry.delay") as mock_delay:
+            context = {"tenant_id": "tenant-123"}
+            payload = {"app_id": "app-abc", "name": "My App"}
+
+            emit(TelemetryCase.APP_CREATED, context, payload)
+
+            mock_delay.assert_called_once()
+            envelope_json = mock_delay.call_args[0][0]
+            envelope = TelemetryEnvelope.model_validate_json(envelope_json)
+            assert envelope.case == TelemetryCase.APP_CREATED
+            assert envelope.tenant_id == "tenant-123"
+            assert envelope.payload["app_id"] == "app-abc"
+
+    @pytest.mark.usefixtures("mock_ops_trace_manager")
+    @patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=True)
+    def test_tool_execution_trace_routed(
+        self,
+        _mock_ee_enabled: MagicMock,
+    ) -> None:
+        mock_trace_manager = MagicMock()
+        context = {"tenant_id": "tenant-123", "app_id": "app-123"}
+        payload = {"tool_name": "test_tool", "tool_inputs": {}, "tool_outputs": "result"}
+
+        emit(TelemetryCase.TOOL_EXECUTION, context, payload, mock_trace_manager)
+
+        mock_trace_manager.add_trace_task.assert_called_once()
+
+    @pytest.mark.usefixtures("mock_ops_trace_manager")
+    @patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=True)
+    def test_moderation_check_trace_routed(
+        self,
+        _mock_ee_enabled: MagicMock,
+    ) -> None:
+        mock_trace_manager = MagicMock()
+        context = {"tenant_id": "tenant-123", "app_id": "app-123"}
+        payload = {"message_id": "msg-123", "moderation_result": {"flagged": False}}
+
+        emit(TelemetryCase.MODERATION_CHECK, context, payload, mock_trace_manager)
+
+        mock_trace_manager.add_trace_task.assert_called_once()
+
+
+class TestGatewayIntegrationCEEligibility:
+    @pytest.fixture
+    def mock_trace_manager(self) -> MagicMock:
+        return MagicMock()
+
+    @pytest.mark.usefixtures("mock_ops_trace_manager")
+    def test_workflow_run_is_ce_eligible(
+        self,
+        mock_trace_manager: MagicMock,
+    ) -> None:
+        with patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=False):
+            context = {"app_id": "app-123", "user_id": "user-456"}
+            payload = {"workflow_run_id": "run-abc"}
+
+            emit(TelemetryCase.WORKFLOW_RUN, context, payload, mock_trace_manager)
+
+            mock_trace_manager.add_trace_task.assert_called_once()
+
+    @pytest.mark.usefixtures("mock_ops_trace_manager")
+    def test_message_run_is_ce_eligible(
+        self,
+        mock_trace_manager: MagicMock,
+    ) -> None:
+        with patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=False):
+            context = {"app_id": "app-123", "user_id": "user-456"}
+            payload = {"message_id": "msg-abc", "conversation_id": "conv-123"}
+
+            emit(TelemetryCase.MESSAGE_RUN, context, payload, mock_trace_manager)
+
+            mock_trace_manager.add_trace_task.assert_called_once()
+
+    @pytest.mark.usefixtures("mock_ops_trace_manager")
+    def test_node_execution_not_ce_eligible(
+        self,
+        mock_trace_manager: MagicMock,
+    ) -> None:
+        with patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=False):
+            context = {"app_id": "app-123", "user_id": "user-456"}
+            payload = {"node_id": "node-abc"}
+
+            emit(TelemetryCase.NODE_EXECUTION, context, payload, mock_trace_manager)
+
+            mock_trace_manager.add_trace_task.assert_not_called()
+
+    @pytest.mark.usefixtures("mock_ops_trace_manager")
+    def test_draft_node_execution_not_ce_eligible(
+        self,
+        mock_trace_manager: MagicMock,
+    ) -> None:
+        with patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=False):
+            context = {"app_id": "app-123", "user_id": "user-456"}
+            payload = {"node_execution_data": {}}
+
+            emit(TelemetryCase.DRAFT_NODE_EXECUTION, context, payload, mock_trace_manager)
+
+            mock_trace_manager.add_trace_task.assert_not_called()
+
+    @pytest.mark.usefixtures("mock_ops_trace_manager")
+    def test_prompt_generation_not_ce_eligible(
+        self,
+        mock_trace_manager: MagicMock,
+    ) -> None:
+        with patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=False):
+            context = {"app_id": "app-123", "user_id": "user-456", "tenant_id": "tenant-789"}
+            payload = {"operation_type": "generate", "instruction": "test"}
+
+            emit(TelemetryCase.PROMPT_GENERATION, context, payload, mock_trace_manager)
+
+            mock_trace_manager.add_trace_task.assert_not_called()
+
+
+class TestIsEnterpriseTelemetryEnabled:
+    def test_returns_false_when_exporter_import_fails(self) -> None:
+        with patch.dict(sys.modules, {"enterprise.telemetry.exporter": None}):
+            result = is_enterprise_telemetry_enabled()
+            assert result is False
+
+    def test_function_is_callable(self) -> None:
+        assert callable(is_enterprise_telemetry_enabled)

api/tests/unit_tests/enterprise/telemetry/test_contracts.py

@@ -0,0 +1,259 @@
+"""Unit tests for telemetry gateway contracts."""
+
+from __future__ import annotations
+
+import pytest
+from pydantic import ValidationError
+
+from core.telemetry.gateway import CASE_ROUTING
+from enterprise.telemetry.contracts import CaseRoute, SignalType, TelemetryCase, TelemetryEnvelope
+
+
+class TestTelemetryCase:
+    """Tests for TelemetryCase enum."""
+
+    def test_all_cases_defined(self) -> None:
+        """Verify all 14 telemetry cases are defined."""
+        expected_cases = {
+            "WORKFLOW_RUN",
+            "NODE_EXECUTION",
+            "DRAFT_NODE_EXECUTION",
+            "MESSAGE_RUN",
+            "TOOL_EXECUTION",
+            "MODERATION_CHECK",
+            "SUGGESTED_QUESTION",
+            "DATASET_RETRIEVAL",
+            "GENERATE_NAME",
+            "PROMPT_GENERATION",
+            "APP_CREATED",
+            "APP_UPDATED",
+            "APP_DELETED",
+            "FEEDBACK_CREATED",
+        }
+        actual_cases = {case.name for case in TelemetryCase}
+        assert actual_cases == expected_cases
+
+    def test_case_values(self) -> None:
+        """Verify case enum values are correct."""
+        assert TelemetryCase.WORKFLOW_RUN.value == "workflow_run"
+        assert TelemetryCase.NODE_EXECUTION.value == "node_execution"
+        assert TelemetryCase.DRAFT_NODE_EXECUTION.value == "draft_node_execution"
+        assert TelemetryCase.MESSAGE_RUN.value == "message_run"
+        assert TelemetryCase.TOOL_EXECUTION.value == "tool_execution"
+        assert TelemetryCase.MODERATION_CHECK.value == "moderation_check"
+        assert TelemetryCase.SUGGESTED_QUESTION.value == "suggested_question"
+        assert TelemetryCase.DATASET_RETRIEVAL.value == "dataset_retrieval"
+        assert TelemetryCase.GENERATE_NAME.value == "generate_name"
+        assert TelemetryCase.PROMPT_GENERATION.value == "prompt_generation"
+        assert TelemetryCase.APP_CREATED.value == "app_created"
+        assert TelemetryCase.APP_UPDATED.value == "app_updated"
+        assert TelemetryCase.APP_DELETED.value == "app_deleted"
+        assert TelemetryCase.FEEDBACK_CREATED.value == "feedback_created"
+
+
+class TestCaseRoute:
+    """Tests for CaseRoute model."""
+
+    def test_valid_trace_route(self) -> None:
+        """Verify valid trace route creation."""
+        route = CaseRoute(signal_type=SignalType.TRACE, ce_eligible=True)
+        assert route.signal_type == SignalType.TRACE
+        assert route.ce_eligible is True
+
+    def test_valid_metric_log_route(self) -> None:
+        """Verify valid metric_log route creation."""
+        route = CaseRoute(signal_type=SignalType.METRIC_LOG, ce_eligible=False)
+        assert route.signal_type == SignalType.METRIC_LOG
+        assert route.ce_eligible is False
+
+    def test_invalid_signal_type(self) -> None:
+        """Verify invalid signal_type is rejected."""
+        with pytest.raises(ValidationError):
+            CaseRoute(signal_type="invalid", ce_eligible=True)
+
+
+class TestTelemetryEnvelope:
+    """Tests for TelemetryEnvelope model."""
+
+    def test_valid_envelope_minimal(self) -> None:
+        """Verify valid minimal envelope creation."""
+        envelope = TelemetryEnvelope(
+            case=TelemetryCase.WORKFLOW_RUN,
+            tenant_id="tenant-123",
+            event_id="event-456",
+            payload={"key": "value"},
+        )
+        assert envelope.case == TelemetryCase.WORKFLOW_RUN
+        assert envelope.tenant_id == "tenant-123"
+        assert envelope.event_id == "event-456"
+        assert envelope.payload == {"key": "value"}
+        assert envelope.payload_fallback is None
+        assert envelope.metadata is None
+
+    def test_valid_envelope_full(self) -> None:
+        """Verify valid envelope with all fields."""
+        metadata = {"source": "api"}
+        fallback = b"fallback data"
+        envelope = TelemetryEnvelope(
+            case=TelemetryCase.MESSAGE_RUN,
+            tenant_id="tenant-789",
+            event_id="event-012",
+            payload={"message": "hello"},
+            payload_fallback=fallback,
+            metadata=metadata,
+        )
+        assert envelope.case == TelemetryCase.MESSAGE_RUN
+        assert envelope.tenant_id == "tenant-789"
+        assert envelope.event_id == "event-012"
+        assert envelope.payload == {"message": "hello"}
+        assert envelope.payload_fallback == fallback
+        assert envelope.metadata == metadata
+
+    def test_missing_required_case(self) -> None:
+        """Verify missing case field is rejected."""
+        with pytest.raises(ValidationError):
+            TelemetryEnvelope(
+                tenant_id="tenant-123",
+                event_id="event-456",
+                payload={"key": "value"},
+            )
+
+    def test_missing_required_tenant_id(self) -> None:
+        """Verify missing tenant_id field is rejected."""
+        with pytest.raises(ValidationError):
+            TelemetryEnvelope(
+                case=TelemetryCase.WORKFLOW_RUN,
+                event_id="event-456",
+                payload={"key": "value"},
+            )
+
+    def test_missing_required_event_id(self) -> None:
+        """Verify missing event_id field is rejected."""
+        with pytest.raises(ValidationError):
+            TelemetryEnvelope(
+                case=TelemetryCase.WORKFLOW_RUN,
+                tenant_id="tenant-123",
+                payload={"key": "value"},
+            )
+
+    def test_missing_required_payload(self) -> None:
+        """Verify missing payload field is rejected."""
+        with pytest.raises(ValidationError):
+            TelemetryEnvelope(
+                case=TelemetryCase.WORKFLOW_RUN,
+                tenant_id="tenant-123",
+                event_id="event-456",
+            )
+
+    def test_payload_fallback_within_limit(self) -> None:
+        """Verify payload_fallback within 64KB limit is accepted."""
+        fallback = b"x" * 65536
+        envelope = TelemetryEnvelope(
+            case=TelemetryCase.WORKFLOW_RUN,
+            tenant_id="tenant-123",
+            event_id="event-456",
+            payload={"key": "value"},
+            payload_fallback=fallback,
+        )
+        assert envelope.payload_fallback == fallback
+
+    def test_payload_fallback_exceeds_limit(self) -> None:
+        """Verify payload_fallback exceeding 64KB is rejected."""
+        fallback = b"x" * 65537
+        with pytest.raises(ValidationError) as exc_info:
+            TelemetryEnvelope(
+                case=TelemetryCase.WORKFLOW_RUN,
+                tenant_id="tenant-123",
+                event_id="event-456",
+                payload={"key": "value"},
+                payload_fallback=fallback,
+            )
+        assert "64KB" in str(exc_info.value)
+
+    def test_payload_fallback_none(self) -> None:
+        """Verify payload_fallback can be None."""
+        envelope = TelemetryEnvelope(
+            case=TelemetryCase.WORKFLOW_RUN,
+            tenant_id="tenant-123",
+            event_id="event-456",
+            payload={"key": "value"},
+            payload_fallback=None,
+        )
+        assert envelope.payload_fallback is None
+
+
+class TestCaseRouting:
+    """Tests for CASE_ROUTING table."""
+
+    def test_all_cases_routed(self) -> None:
+        """Verify all 14 cases have routing entries."""
+        assert len(CASE_ROUTING) == 14
+        for case in TelemetryCase:
+            assert case in CASE_ROUTING
+
+    def test_trace_ce_eligible_cases(self) -> None:
+        """Verify trace cases with CE eligibility."""
+        ce_eligible_trace_cases = {
+            TelemetryCase.WORKFLOW_RUN,
+            TelemetryCase.MESSAGE_RUN,
+        }
+        for case in ce_eligible_trace_cases:
+            route = CASE_ROUTING[case]
+            assert route.signal_type == SignalType.TRACE
+            assert route.ce_eligible is True
+
+    def test_trace_enterprise_only_cases(self) -> None:
+        """Verify trace cases that are enterprise-only."""
+        enterprise_only_trace_cases = {
+            TelemetryCase.NODE_EXECUTION,
+            TelemetryCase.DRAFT_NODE_EXECUTION,
+            TelemetryCase.PROMPT_GENERATION,
+        }
+        for case in enterprise_only_trace_cases:
+            route = CASE_ROUTING[case]
+            assert route.signal_type == SignalType.TRACE
+            assert route.ce_eligible is False
+
+    def test_metric_log_cases(self) -> None:
+        """Verify metric/log-only cases."""
+        metric_log_cases = {
+            TelemetryCase.APP_CREATED,
+            TelemetryCase.APP_UPDATED,
+            TelemetryCase.APP_DELETED,
+            TelemetryCase.FEEDBACK_CREATED,
+        }
+        for case in metric_log_cases:
+            route = CASE_ROUTING[case]
+            assert route.signal_type == SignalType.METRIC_LOG
+            assert route.ce_eligible is False
+
+    def test_routing_table_completeness(self) -> None:
+        """Verify routing table covers all cases with correct types."""
+        trace_cases = {
+            TelemetryCase.WORKFLOW_RUN,
+            TelemetryCase.MESSAGE_RUN,
+            TelemetryCase.NODE_EXECUTION,
+            TelemetryCase.DRAFT_NODE_EXECUTION,
+            TelemetryCase.PROMPT_GENERATION,
+            TelemetryCase.TOOL_EXECUTION,
+            TelemetryCase.MODERATION_CHECK,
+            TelemetryCase.SUGGESTED_QUESTION,
+            TelemetryCase.DATASET_RETRIEVAL,
+            TelemetryCase.GENERATE_NAME,
+        }
+        metric_log_cases = {
+            TelemetryCase.APP_CREATED,
+            TelemetryCase.APP_UPDATED,
+            TelemetryCase.APP_DELETED,
+            TelemetryCase.FEEDBACK_CREATED,
+        }
+
+        all_cases = trace_cases | metric_log_cases
+        assert len(all_cases) == 14
+        assert all_cases == set(TelemetryCase)
+
+        for case in trace_cases:
+            assert CASE_ROUTING[case].signal_type == SignalType.TRACE
+
+        for case in metric_log_cases:
+            assert CASE_ROUTING[case].signal_type == SignalType.METRIC_LOG

api/tests/unit_tests/enterprise/telemetry/test_event_handlers.py

@@ -0,0 +1,121 @@
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from enterprise.telemetry import event_handlers
+from enterprise.telemetry.contracts import TelemetryCase
+
+
+@pytest.fixture
+def mock_gateway_emit():
+    with patch("core.telemetry.gateway.emit") as mock:
+        yield mock
+
+
+def test_handle_app_created_calls_task(mock_gateway_emit):
+    sender = MagicMock()
+    sender.id = "app-123"
+    sender.tenant_id = "tenant-456"
+    sender.mode = "chat"
+
+    event_handlers._handle_app_created(sender)
+
+    mock_gateway_emit.assert_called_once_with(
+        case=TelemetryCase.APP_CREATED,
+        context={"tenant_id": "tenant-456"},
+        payload={"app_id": "app-123", "mode": "chat"},
+    )
+
+
+def test_handle_app_created_no_exporter(mock_gateway_emit):
+    """Gateway handles exporter availability internally; handler always calls gateway."""
+    sender = MagicMock()
+    sender.id = "app-123"
+    sender.tenant_id = "tenant-456"
+
+    event_handlers._handle_app_created(sender)
+
+    mock_gateway_emit.assert_called_once()
+
+
+def test_handle_app_updated_calls_task(mock_gateway_emit):
+    sender = MagicMock()
+    sender.id = "app-123"
+    sender.tenant_id = "tenant-456"
+
+    event_handlers._handle_app_updated(sender)
+
+    mock_gateway_emit.assert_called_once_with(
+        case=TelemetryCase.APP_UPDATED,
+        context={"tenant_id": "tenant-456"},
+        payload={"app_id": "app-123"},
+    )
+
+
+def test_handle_app_deleted_calls_task(mock_gateway_emit):
+    sender = MagicMock()
+    sender.id = "app-123"
+    sender.tenant_id = "tenant-456"
+
+    event_handlers._handle_app_deleted(sender)
+
+    mock_gateway_emit.assert_called_once_with(
+        case=TelemetryCase.APP_DELETED,
+        context={"tenant_id": "tenant-456"},
+        payload={"app_id": "app-123"},
+    )
+
+
+def test_handle_feedback_created_calls_task(mock_gateway_emit):
+    sender = MagicMock()
+    sender.message_id = "msg-123"
+    sender.app_id = "app-456"
+    sender.conversation_id = "conv-789"
+    sender.from_end_user_id = "user-001"
+    sender.from_account_id = None
+    sender.rating = "like"
+    sender.from_source = "api"
+    sender.content = "Great response!"
+
+    event_handlers._handle_feedback_created(sender, tenant_id="tenant-456")
+
+    mock_gateway_emit.assert_called_once_with(
+        case=TelemetryCase.FEEDBACK_CREATED,
+        context={"tenant_id": "tenant-456"},
+        payload={
+            "message_id": "msg-123",
+            "app_id": "app-456",
+            "conversation_id": "conv-789",
+            "from_end_user_id": "user-001",
+            "from_account_id": None,
+            "rating": "like",
+            "from_source": "api",
+            "content": "Great response!",
+        },
+    )
+
+
+def test_handle_feedback_created_no_exporter(mock_gateway_emit):
+    """Gateway handles exporter availability internally; handler always calls gateway."""
+    sender = MagicMock()
+    sender.message_id = "msg-123"
+
+    event_handlers._handle_feedback_created(sender, tenant_id="tenant-456")
+
+    mock_gateway_emit.assert_called_once()
+
+
+def test_handlers_create_valid_envelopes(mock_gateway_emit):
+    """Verify handlers pass correct TelemetryCase and payload structure."""
+    sender = MagicMock()
+    sender.id = "app-123"
+    sender.tenant_id = "tenant-456"
+    sender.mode = "chat"
+
+    event_handlers._handle_app_created(sender)
+
+    call_kwargs = mock_gateway_emit.call_args[1]
+    assert call_kwargs["case"] == TelemetryCase.APP_CREATED
+    assert call_kwargs["context"]["tenant_id"] == "tenant-456"
+    assert call_kwargs["payload"]["app_id"] == "app-123"
+    assert call_kwargs["payload"]["mode"] == "chat"

api/tests/unit_tests/enterprise/telemetry/test_exporter.py

@@ -0,0 +1,261 @@
+"""Unit tests for EnterpriseExporter and _ExporterFactory."""
+
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from configs.enterprise import EnterpriseTelemetryConfig
+from enterprise.telemetry.exporter import EnterpriseExporter
+
+
+def test_config_api_key_default_empty():
+    """Test that ENTERPRISE_OTLP_API_KEY defaults to empty string."""
+    config = EnterpriseTelemetryConfig()
+    assert config.ENTERPRISE_OTLP_API_KEY == ""
+
+
+@patch("enterprise.telemetry.exporter.GRPCSpanExporter")
+@patch("enterprise.telemetry.exporter.GRPCMetricExporter")
+def test_api_key_only_injects_bearer_header(mock_metric_exporter: MagicMock, mock_span_exporter: MagicMock) -> None:
+    """Test that API key alone injects Bearer authorization header."""
+    mock_config = SimpleNamespace(
+        ENTERPRISE_OTLP_ENDPOINT="https://collector.example.com",
+        ENTERPRISE_OTLP_HEADERS="",
+        ENTERPRISE_OTLP_PROTOCOL="grpc",
+        ENTERPRISE_SERVICE_NAME="dify",
+        ENTERPRISE_OTEL_SAMPLING_RATE=1.0,
+        ENTERPRISE_INCLUDE_CONTENT=True,
+        ENTERPRISE_OTLP_API_KEY="test-secret-key",
+    )
+
+    EnterpriseExporter(mock_config)
+
+    # Verify span exporter was called with Bearer header
+    assert mock_span_exporter.call_args is not None
+    headers = mock_span_exporter.call_args.kwargs.get("headers")
+    assert headers is not None
+    assert ("authorization", "Bearer test-secret-key") in headers
+
+
+@patch("enterprise.telemetry.exporter.GRPCSpanExporter")
+@patch("enterprise.telemetry.exporter.GRPCMetricExporter")
+def test_empty_api_key_no_auth_header(mock_metric_exporter: MagicMock, mock_span_exporter: MagicMock) -> None:
+    """Test that empty API key does not inject authorization header."""
+    mock_config = SimpleNamespace(
+        ENTERPRISE_OTLP_ENDPOINT="https://collector.example.com",
+        ENTERPRISE_OTLP_HEADERS="",
+        ENTERPRISE_OTLP_PROTOCOL="grpc",
+        ENTERPRISE_SERVICE_NAME="dify",
+        ENTERPRISE_OTEL_SAMPLING_RATE=1.0,
+        ENTERPRISE_INCLUDE_CONTENT=True,
+        ENTERPRISE_OTLP_API_KEY="",
+    )
+
+    EnterpriseExporter(mock_config)
+
+    # Verify span exporter was called without authorization header
+    assert mock_span_exporter.call_args is not None
+    headers = mock_span_exporter.call_args.kwargs.get("headers")
+    # Headers should be None or not contain authorization
+    if headers is not None:
+        assert not any(key == "authorization" for key, _ in headers)
+
+
+@patch("enterprise.telemetry.exporter.GRPCSpanExporter")
+@patch("enterprise.telemetry.exporter.GRPCMetricExporter")
+def test_api_key_and_custom_headers_merge(mock_metric_exporter: MagicMock, mock_span_exporter: MagicMock) -> None:
+    """Test that API key and custom headers are merged correctly."""
+    mock_config = SimpleNamespace(
+        ENTERPRISE_OTLP_ENDPOINT="https://collector.example.com",
+        ENTERPRISE_OTLP_HEADERS="x-custom=foo",
+        ENTERPRISE_OTLP_PROTOCOL="grpc",
+        ENTERPRISE_SERVICE_NAME="dify",
+        ENTERPRISE_OTEL_SAMPLING_RATE=1.0,
+        ENTERPRISE_INCLUDE_CONTENT=True,
+        ENTERPRISE_OTLP_API_KEY="test-key",
+    )
+
+    EnterpriseExporter(mock_config)
+
+    # Verify both headers are present
+    assert mock_span_exporter.call_args is not None
+    headers = mock_span_exporter.call_args.kwargs.get("headers")
+    assert headers is not None
+    assert ("authorization", "Bearer test-key") in headers
+    assert ("x-custom", "foo") in headers
+
+
+@patch("enterprise.telemetry.exporter.logger")
+@patch("enterprise.telemetry.exporter.GRPCSpanExporter")
+@patch("enterprise.telemetry.exporter.GRPCMetricExporter")
+def test_api_key_overrides_conflicting_header(
+    mock_metric_exporter: MagicMock, mock_span_exporter: MagicMock, mock_logger: MagicMock
+) -> None:
+    """Test that API key overrides conflicting authorization header and logs warning."""
+    mock_config = SimpleNamespace(
+        ENTERPRISE_OTLP_ENDPOINT="https://collector.example.com",
+        ENTERPRISE_OTLP_HEADERS="authorization=Basic old",
+        ENTERPRISE_OTLP_PROTOCOL="grpc",
+        ENTERPRISE_SERVICE_NAME="dify",
+        ENTERPRISE_OTEL_SAMPLING_RATE=1.0,
+        ENTERPRISE_INCLUDE_CONTENT=True,
+        ENTERPRISE_OTLP_API_KEY="test-key",
+    )
+
+    EnterpriseExporter(mock_config)
+
+    # Verify Bearer header takes precedence
+    assert mock_span_exporter.call_args is not None
+    headers = mock_span_exporter.call_args.kwargs.get("headers")
+    assert headers is not None
+    assert ("authorization", "Bearer test-key") in headers
+    # Verify old authorization header is not present
+    assert ("authorization", "Basic old") not in headers
+
+    # Verify warning was logged
+    mock_logger.warning.assert_called_once()
+    assert mock_logger.warning.call_args is not None
+    warning_message = mock_logger.warning.call_args[0][0]
+    assert "ENTERPRISE_OTLP_API_KEY is set" in warning_message
+    assert "authorization" in warning_message
+
+
+@patch("enterprise.telemetry.exporter.GRPCSpanExporter")
+@patch("enterprise.telemetry.exporter.GRPCMetricExporter")
+def test_https_endpoint_uses_secure_grpc(mock_metric_exporter: MagicMock, mock_span_exporter: MagicMock) -> None:
+    """Test that https:// endpoint enables TLS (insecure=False) for gRPC."""
+    mock_config = SimpleNamespace(
+        ENTERPRISE_OTLP_ENDPOINT="https://collector.example.com",
+        ENTERPRISE_OTLP_HEADERS="",
+        ENTERPRISE_OTLP_PROTOCOL="grpc",
+        ENTERPRISE_SERVICE_NAME="dify",
+        ENTERPRISE_OTEL_SAMPLING_RATE=1.0,
+        ENTERPRISE_INCLUDE_CONTENT=True,
+        ENTERPRISE_OTLP_API_KEY="test-key",
+    )
+
+    EnterpriseExporter(mock_config)
+
+    # Verify insecure=False for both exporters (https:// scheme)
+    assert mock_span_exporter.call_args is not None
+    assert mock_span_exporter.call_args.kwargs["insecure"] is False
+
+    assert mock_metric_exporter.call_args is not None
+    assert mock_metric_exporter.call_args.kwargs["insecure"] is False
+
+@patch("enterprise.telemetry.exporter.GRPCSpanExporter")
+@patch("enterprise.telemetry.exporter.GRPCMetricExporter")
+def test_http_endpoint_uses_insecure_grpc(mock_metric_exporter: MagicMock, mock_span_exporter: MagicMock) -> None:
+    """Test that http:// endpoint uses insecure gRPC (insecure=True)."""
+    mock_config = SimpleNamespace(
+        ENTERPRISE_OTLP_ENDPOINT="http://collector.example.com",
+        ENTERPRISE_OTLP_HEADERS="",
+        ENTERPRISE_OTLP_PROTOCOL="grpc",
+        ENTERPRISE_SERVICE_NAME="dify",
+        ENTERPRISE_OTEL_SAMPLING_RATE=1.0,
+        ENTERPRISE_INCLUDE_CONTENT=True,
+        ENTERPRISE_OTLP_API_KEY="",
+    )
+
+    EnterpriseExporter(mock_config)
+
+    # Verify insecure=True for both exporters (http:// scheme)
+    assert mock_span_exporter.call_args is not None
+    assert mock_span_exporter.call_args.kwargs["insecure"] is True
+
+    assert mock_metric_exporter.call_args is not None
+    assert mock_metric_exporter.call_args.kwargs["insecure"] is True
+
+@patch("enterprise.telemetry.exporter.HTTPSpanExporter")
+@patch("enterprise.telemetry.exporter.HTTPMetricExporter")
+def test_insecure_not_passed_to_http_exporters(mock_metric_exporter: MagicMock, mock_span_exporter: MagicMock) -> None:
+    """Test that insecure parameter is not passed to HTTP exporters."""
+    mock_config = SimpleNamespace(
+        ENTERPRISE_OTLP_ENDPOINT="http://collector.example.com",
+        ENTERPRISE_OTLP_HEADERS="",
+        ENTERPRISE_OTLP_PROTOCOL="http",
+        ENTERPRISE_SERVICE_NAME="dify",
+        ENTERPRISE_OTEL_SAMPLING_RATE=1.0,
+        ENTERPRISE_INCLUDE_CONTENT=True,
+        ENTERPRISE_OTLP_API_KEY="test-key",
+    )
+
+    EnterpriseExporter(mock_config)
+
+    # Verify insecure kwarg is NOT in HTTP exporter calls
+    assert mock_span_exporter.call_args is not None
+    assert "insecure" not in mock_span_exporter.call_args.kwargs
+
+    assert mock_metric_exporter.call_args is not None
+    assert "insecure" not in mock_metric_exporter.call_args.kwargs
+
+
+@patch("enterprise.telemetry.exporter.GRPCSpanExporter")
+@patch("enterprise.telemetry.exporter.GRPCMetricExporter")
+def test_api_key_with_special_chars_preserved(mock_metric_exporter: MagicMock, mock_span_exporter: MagicMock) -> None:
+    """Test that API key with special characters is preserved without mangling."""
+    special_key = "abc+def/ghi=jkl=="
+    mock_config = SimpleNamespace(
+        ENTERPRISE_OTLP_ENDPOINT="https://collector.example.com",
+        ENTERPRISE_OTLP_HEADERS="",
+        ENTERPRISE_OTLP_PROTOCOL="grpc",
+        ENTERPRISE_SERVICE_NAME="dify",
+        ENTERPRISE_OTEL_SAMPLING_RATE=1.0,
+        ENTERPRISE_INCLUDE_CONTENT=True,
+        ENTERPRISE_OTLP_API_KEY=special_key,
+    )
+
+    EnterpriseExporter(mock_config)
+
+    # Verify special characters are preserved in Bearer header
+    assert mock_span_exporter.call_args is not None
+    headers = mock_span_exporter.call_args.kwargs.get("headers")
+    assert headers is not None
+    assert ("authorization", f"Bearer {special_key}") in headers
+
+
+@patch("enterprise.telemetry.exporter.GRPCSpanExporter")
+@patch("enterprise.telemetry.exporter.GRPCMetricExporter")
+def test_no_scheme_localhost_uses_insecure(mock_metric_exporter: MagicMock, mock_span_exporter: MagicMock) -> None:
+    """Test that endpoint without scheme defaults to insecure for localhost."""
+    mock_config = SimpleNamespace(
+        ENTERPRISE_OTLP_ENDPOINT="localhost:4317",
+        ENTERPRISE_OTLP_HEADERS="",
+        ENTERPRISE_OTLP_PROTOCOL="grpc",
+        ENTERPRISE_SERVICE_NAME="dify",
+        ENTERPRISE_OTEL_SAMPLING_RATE=1.0,
+        ENTERPRISE_INCLUDE_CONTENT=True,
+        ENTERPRISE_OTLP_API_KEY="",
+    )
+
+    EnterpriseExporter(mock_config)
+
+    # Verify insecure=True for localhost without scheme
+    assert mock_span_exporter.call_args is not None
+    assert mock_span_exporter.call_args.kwargs["insecure"] is True
+
+    assert mock_metric_exporter.call_args is not None
+    assert mock_metric_exporter.call_args.kwargs["insecure"] is True
+
+
+@patch("enterprise.telemetry.exporter.GRPCSpanExporter")
+@patch("enterprise.telemetry.exporter.GRPCMetricExporter")
+def test_no_scheme_production_uses_insecure(mock_metric_exporter: MagicMock, mock_span_exporter: MagicMock) -> None:
+    """Test that endpoint without scheme defaults to insecure (not https://)."""
+    mock_config = SimpleNamespace(
+        ENTERPRISE_OTLP_ENDPOINT="collector.example.com:4317",
+        ENTERPRISE_OTLP_HEADERS="",
+        ENTERPRISE_OTLP_PROTOCOL="grpc",
+        ENTERPRISE_SERVICE_NAME="dify",
+        ENTERPRISE_OTEL_SAMPLING_RATE=1.0,
+        ENTERPRISE_INCLUDE_CONTENT=True,
+        ENTERPRISE_OTLP_API_KEY="",
+    )
+
+    EnterpriseExporter(mock_config)
+
+    # Verify insecure=True for any endpoint without https:// scheme
+    assert mock_span_exporter.call_args is not None
+    assert mock_span_exporter.call_args.kwargs["insecure"] is True
+
+    assert mock_metric_exporter.call_args is not None
+    assert mock_metric_exporter.call_args.kwargs["insecure"] is True

api/tests/unit_tests/enterprise/telemetry/test_gateway.py

@@ -0,0 +1,272 @@
+from __future__ import annotations
+
+import sys
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from core.ops.entities.trace_entity import TraceTaskName
+from core.telemetry.gateway import (
+    CASE_ROUTING,
+    CASE_TO_TRACE_TASK,
+    PAYLOAD_SIZE_THRESHOLD_BYTES,
+    emit,
+)
+from enterprise.telemetry.contracts import SignalType, TelemetryCase, TelemetryEnvelope
+
+
+class TestCaseRoutingTable:
+    def test_all_cases_have_routing(self) -> None:
+        for case in TelemetryCase:
+            assert case in CASE_ROUTING, f"Missing routing for {case}"
+
+    def test_trace_cases(self) -> None:
+        trace_cases = [
+            TelemetryCase.WORKFLOW_RUN,
+            TelemetryCase.MESSAGE_RUN,
+            TelemetryCase.NODE_EXECUTION,
+            TelemetryCase.DRAFT_NODE_EXECUTION,
+            TelemetryCase.PROMPT_GENERATION,
+        ]
+        for case in trace_cases:
+            assert CASE_ROUTING[case].signal_type is SignalType.TRACE, f"{case} should be trace"
+
+    def test_metric_log_cases(self) -> None:
+        metric_log_cases = [
+            TelemetryCase.APP_CREATED,
+            TelemetryCase.APP_UPDATED,
+            TelemetryCase.APP_DELETED,
+            TelemetryCase.FEEDBACK_CREATED,
+        ]
+        for case in metric_log_cases:
+            assert CASE_ROUTING[case].signal_type is SignalType.METRIC_LOG, f"{case} should be metric_log"
+
+    def test_ce_eligible_cases(self) -> None:
+        ce_eligible_cases = [
+            TelemetryCase.WORKFLOW_RUN,
+            TelemetryCase.MESSAGE_RUN,
+            TelemetryCase.TOOL_EXECUTION,
+            TelemetryCase.MODERATION_CHECK,
+            TelemetryCase.SUGGESTED_QUESTION,
+            TelemetryCase.DATASET_RETRIEVAL,
+            TelemetryCase.GENERATE_NAME,
+        ]
+        for case in ce_eligible_cases:
+            assert CASE_ROUTING[case].ce_eligible is True, f"{case} should be CE eligible"
+
+    def test_enterprise_only_cases(self) -> None:
+        enterprise_only_cases = [
+            TelemetryCase.NODE_EXECUTION,
+            TelemetryCase.DRAFT_NODE_EXECUTION,
+            TelemetryCase.PROMPT_GENERATION,
+        ]
+        for case in enterprise_only_cases:
+            assert CASE_ROUTING[case].ce_eligible is False, f"{case} should be enterprise-only"
+
+    def test_trace_cases_have_task_name_mapping(self) -> None:
+        trace_cases = [c for c in TelemetryCase if CASE_ROUTING[c].signal_type is SignalType.TRACE]
+        for case in trace_cases:
+            assert case in CASE_TO_TRACE_TASK, f"Missing TraceTaskName mapping for {case}"
+
+
+@pytest.fixture
+def mock_ops_trace_manager():
+    mock_module = MagicMock()
+    mock_trace_task_class = MagicMock()
+    mock_trace_task_class.return_value = MagicMock()
+    mock_module.TraceTask = mock_trace_task_class
+    mock_module.TraceQueueManager = MagicMock()
+
+    mock_trace_entity = MagicMock()
+    mock_trace_task_name = MagicMock()
+    mock_trace_task_name.return_value = "workflow"
+    mock_trace_entity.TraceTaskName = mock_trace_task_name
+
+    with (
+        patch.dict(sys.modules, {"core.ops.ops_trace_manager": mock_module}),
+        patch.dict(sys.modules, {"core.ops.entities.trace_entity": mock_trace_entity}),
+    ):
+        yield mock_module, mock_trace_entity
+
+
+class TestGatewayTraceRouting:
+    @pytest.fixture
+    def mock_trace_manager(self) -> MagicMock:
+        return MagicMock()
+
+    @patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=True)
+    def test_trace_case_routes_to_trace_manager(
+        self,
+        _mock_ee_enabled: MagicMock,
+        mock_trace_manager: MagicMock,
+        mock_ops_trace_manager: tuple[MagicMock, MagicMock],
+    ) -> None:
+        context = {"app_id": "app-123", "user_id": "user-456", "tenant_id": "tenant-789"}
+        payload = {"workflow_run_id": "run-abc"}
+
+        emit(TelemetryCase.WORKFLOW_RUN, context, payload, mock_trace_manager)
+
+        mock_trace_manager.add_trace_task.assert_called_once()
+
+    @patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=False)
+    def test_ce_eligible_trace_enqueued_when_ee_disabled(
+        self,
+        _mock_ee_enabled: MagicMock,
+        mock_trace_manager: MagicMock,
+        mock_ops_trace_manager: tuple[MagicMock, MagicMock],
+    ) -> None:
+        context = {"app_id": "app-123", "user_id": "user-456"}
+        payload = {"workflow_run_id": "run-abc"}
+
+        emit(TelemetryCase.WORKFLOW_RUN, context, payload, mock_trace_manager)
+
+        mock_trace_manager.add_trace_task.assert_called_once()
+
+    @patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=False)
+    def test_enterprise_only_trace_dropped_when_ee_disabled(
+        self,
+        _mock_ee_enabled: MagicMock,
+        mock_trace_manager: MagicMock,
+        mock_ops_trace_manager: tuple[MagicMock, MagicMock],
+    ) -> None:
+        context = {"app_id": "app-123", "user_id": "user-456"}
+        payload = {"node_id": "node-abc"}
+
+        emit(TelemetryCase.NODE_EXECUTION, context, payload, mock_trace_manager)
+
+        mock_trace_manager.add_trace_task.assert_not_called()
+
+    @patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=True)
+    def test_enterprise_only_trace_enqueued_when_ee_enabled(
+        self,
+        _mock_ee_enabled: MagicMock,
+        mock_trace_manager: MagicMock,
+        mock_ops_trace_manager: tuple[MagicMock, MagicMock],
+    ) -> None:
+        context = {"app_id": "app-123", "user_id": "user-456"}
+        payload = {"node_id": "node-abc"}
+
+        emit(TelemetryCase.NODE_EXECUTION, context, payload, mock_trace_manager)
+
+        mock_trace_manager.add_trace_task.assert_called_once()
+
+
+class TestGatewayMetricLogRouting:
+    @patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=True)
+    @patch("tasks.enterprise_telemetry_task.process_enterprise_telemetry.delay")
+    def test_metric_case_routes_to_celery_task(
+        self,
+        mock_delay: MagicMock,
+        _mock_ee_enabled: MagicMock,
+    ) -> None:
+        context = {"tenant_id": "tenant-123"}
+        payload = {"app_id": "app-abc", "name": "My App"}
+
+        emit(TelemetryCase.APP_CREATED, context, payload)
+
+        mock_delay.assert_called_once()
+        envelope_json = mock_delay.call_args[0][0]
+        envelope = TelemetryEnvelope.model_validate_json(envelope_json)
+        assert envelope.case == TelemetryCase.APP_CREATED
+        assert envelope.tenant_id == "tenant-123"
+        assert envelope.payload["app_id"] == "app-abc"
+
+    @patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=True)
+    @patch("tasks.enterprise_telemetry_task.process_enterprise_telemetry.delay")
+    def test_envelope_has_unique_event_id(
+        self,
+        mock_delay: MagicMock,
+        _mock_ee_enabled: MagicMock,
+    ) -> None:
+        context = {"tenant_id": "tenant-123"}
+        payload = {"app_id": "app-abc"}
+
+        emit(TelemetryCase.APP_CREATED, context, payload)
+        emit(TelemetryCase.APP_CREATED, context, payload)
+
+        assert mock_delay.call_count == 2
+        envelope1 = TelemetryEnvelope.model_validate_json(mock_delay.call_args_list[0][0][0])
+        envelope2 = TelemetryEnvelope.model_validate_json(mock_delay.call_args_list[1][0][0])
+        assert envelope1.event_id != envelope2.event_id
+
+
+class TestGatewayPayloadSizing:
+    @patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=True)
+    @patch("tasks.enterprise_telemetry_task.process_enterprise_telemetry.delay")
+    def test_small_payload_inlined(
+        self,
+        mock_delay: MagicMock,
+        _mock_ee_enabled: MagicMock,
+    ) -> None:
+        context = {"tenant_id": "tenant-123"}
+        payload = {"key": "small_value"}
+
+        emit(TelemetryCase.APP_CREATED, context, payload)
+
+        envelope_json = mock_delay.call_args[0][0]
+        envelope = TelemetryEnvelope.model_validate_json(envelope_json)
+        assert envelope.payload == payload
+        assert envelope.metadata is None
+
+    @patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=True)
+    @patch("core.telemetry.gateway.storage")
+    @patch("tasks.enterprise_telemetry_task.process_enterprise_telemetry.delay")
+    def test_large_payload_stored(
+        self,
+        mock_delay: MagicMock,
+        mock_storage: MagicMock,
+        _mock_ee_enabled: MagicMock,
+    ) -> None:
+        context = {"tenant_id": "tenant-123"}
+        large_value = "x" * (PAYLOAD_SIZE_THRESHOLD_BYTES + 1000)
+        payload = {"key": large_value}
+
+        emit(TelemetryCase.APP_CREATED, context, payload)
+
+        mock_storage.save.assert_called_once()
+        storage_key = mock_storage.save.call_args[0][0]
+        assert storage_key.startswith("telemetry/tenant-123/")
+
+        envelope_json = mock_delay.call_args[0][0]
+        envelope = TelemetryEnvelope.model_validate_json(envelope_json)
+        assert envelope.payload == {}
+        assert envelope.metadata is not None
+        assert envelope.metadata["payload_ref"] == storage_key
+
+    @patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=True)
+    @patch("core.telemetry.gateway.storage")
+    @patch("tasks.enterprise_telemetry_task.process_enterprise_telemetry.delay")
+    def test_large_payload_fallback_on_storage_error(
+        self,
+        mock_delay: MagicMock,
+        mock_storage: MagicMock,
+        _mock_ee_enabled: MagicMock,
+    ) -> None:
+        mock_storage.save.side_effect = Exception("Storage failure")
+        context = {"tenant_id": "tenant-123"}
+        large_value = "x" * (PAYLOAD_SIZE_THRESHOLD_BYTES + 1000)
+        payload = {"key": large_value}
+
+        emit(TelemetryCase.APP_CREATED, context, payload)
+
+        envelope_json = mock_delay.call_args[0][0]
+        envelope = TelemetryEnvelope.model_validate_json(envelope_json)
+        assert envelope.payload == payload
+        assert envelope.metadata is None
+
+
+class TestTraceTaskNameMapping:
+    def test_workflow_run_mapping(self) -> None:
+        assert CASE_TO_TRACE_TASK[TelemetryCase.WORKFLOW_RUN] is TraceTaskName.WORKFLOW_TRACE
+
+    def test_message_run_mapping(self) -> None:
+        assert CASE_TO_TRACE_TASK[TelemetryCase.MESSAGE_RUN] is TraceTaskName.MESSAGE_TRACE
+
+    def test_node_execution_mapping(self) -> None:
+        assert CASE_TO_TRACE_TASK[TelemetryCase.NODE_EXECUTION] is TraceTaskName.NODE_EXECUTION_TRACE
+
+    def test_draft_node_execution_mapping(self) -> None:
+        assert CASE_TO_TRACE_TASK[TelemetryCase.DRAFT_NODE_EXECUTION] is TraceTaskName.DRAFT_NODE_EXECUTION_TRACE
+
+    def test_prompt_generation_mapping(self) -> None:
+        assert CASE_TO_TRACE_TASK[TelemetryCase.PROMPT_GENERATION] is TraceTaskName.PROMPT_GENERATION_TRACE

api/tests/unit_tests/enterprise/telemetry/test_metric_handler.py

@@ -0,0 +1,477 @@
+"""Unit tests for EnterpriseMetricHandler."""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from enterprise.telemetry.contracts import TelemetryCase, TelemetryEnvelope
+from enterprise.telemetry.metric_handler import EnterpriseMetricHandler
+
+
+@pytest.fixture
+def mock_redis():
+    with patch("enterprise.telemetry.metric_handler.redis_client") as mock:
+        yield mock
+
+
+@pytest.fixture
+def sample_envelope():
+    return TelemetryEnvelope(
+        case=TelemetryCase.APP_CREATED,
+        tenant_id="test-tenant",
+        event_id="test-event-123",
+        payload={"app_id": "app-123", "name": "Test App"},
+    )
+
+
+def test_dispatch_app_created(sample_envelope, mock_redis):
+    mock_redis.set.return_value = True
+
+    handler = EnterpriseMetricHandler()
+    with patch.object(handler, "_on_app_created") as mock_handler:
+        handler.handle(sample_envelope)
+        mock_handler.assert_called_once_with(sample_envelope)
+
+
+def test_dispatch_app_updated(mock_redis):
+    mock_redis.set.return_value = True
+    envelope = TelemetryEnvelope(
+        case=TelemetryCase.APP_UPDATED,
+        tenant_id="test-tenant",
+        event_id="test-event-456",
+        payload={},
+    )
+
+    handler = EnterpriseMetricHandler()
+    with patch.object(handler, "_on_app_updated") as mock_handler:
+        handler.handle(envelope)
+        mock_handler.assert_called_once_with(envelope)
+
+
+def test_dispatch_app_deleted(mock_redis):
+    mock_redis.set.return_value = True
+    envelope = TelemetryEnvelope(
+        case=TelemetryCase.APP_DELETED,
+        tenant_id="test-tenant",
+        event_id="test-event-789",
+        payload={},
+    )
+
+    handler = EnterpriseMetricHandler()
+    with patch.object(handler, "_on_app_deleted") as mock_handler:
+        handler.handle(envelope)
+        mock_handler.assert_called_once_with(envelope)
+
+
+def test_dispatch_feedback_created(mock_redis):
+    mock_redis.set.return_value = True
+    envelope = TelemetryEnvelope(
+        case=TelemetryCase.FEEDBACK_CREATED,
+        tenant_id="test-tenant",
+        event_id="test-event-abc",
+        payload={},
+    )
+
+    handler = EnterpriseMetricHandler()
+    with patch.object(handler, "_on_feedback_created") as mock_handler:
+        handler.handle(envelope)
+        mock_handler.assert_called_once_with(envelope)
+
+
+def test_dispatch_message_run(mock_redis):
+    mock_redis.set.return_value = True
+    envelope = TelemetryEnvelope(
+        case=TelemetryCase.MESSAGE_RUN,
+        tenant_id="test-tenant",
+        event_id="test-event-msg",
+        payload={},
+    )
+
+    handler = EnterpriseMetricHandler()
+    with patch.object(handler, "_on_message_run") as mock_handler:
+        handler.handle(envelope)
+        mock_handler.assert_called_once_with(envelope)
+
+
+def test_dispatch_tool_execution(mock_redis):
+    mock_redis.set.return_value = True
+    envelope = TelemetryEnvelope(
+        case=TelemetryCase.TOOL_EXECUTION,
+        tenant_id="test-tenant",
+        event_id="test-event-tool",
+        payload={},
+    )
+
+    handler = EnterpriseMetricHandler()
+    with patch.object(handler, "_on_tool_execution") as mock_handler:
+        handler.handle(envelope)
+        mock_handler.assert_called_once_with(envelope)
+
+
+def test_dispatch_moderation_check(mock_redis):
+    mock_redis.set.return_value = True
+    envelope = TelemetryEnvelope(
+        case=TelemetryCase.MODERATION_CHECK,
+        tenant_id="test-tenant",
+        event_id="test-event-mod",
+        payload={},
+    )
+
+    handler = EnterpriseMetricHandler()
+    with patch.object(handler, "_on_moderation_check") as mock_handler:
+        handler.handle(envelope)
+        mock_handler.assert_called_once_with(envelope)
+
+
+def test_dispatch_suggested_question(mock_redis):
+    mock_redis.set.return_value = True
+    envelope = TelemetryEnvelope(
+        case=TelemetryCase.SUGGESTED_QUESTION,
+        tenant_id="test-tenant",
+        event_id="test-event-sq",
+        payload={},
+    )
+
+    handler = EnterpriseMetricHandler()
+    with patch.object(handler, "_on_suggested_question") as mock_handler:
+        handler.handle(envelope)
+        mock_handler.assert_called_once_with(envelope)
+
+
+def test_dispatch_dataset_retrieval(mock_redis):
+    mock_redis.set.return_value = True
+    envelope = TelemetryEnvelope(
+        case=TelemetryCase.DATASET_RETRIEVAL,
+        tenant_id="test-tenant",
+        event_id="test-event-ds",
+        payload={},
+    )
+
+    handler = EnterpriseMetricHandler()
+    with patch.object(handler, "_on_dataset_retrieval") as mock_handler:
+        handler.handle(envelope)
+        mock_handler.assert_called_once_with(envelope)
+
+
+def test_dispatch_generate_name(mock_redis):
+    mock_redis.set.return_value = True
+    envelope = TelemetryEnvelope(
+        case=TelemetryCase.GENERATE_NAME,
+        tenant_id="test-tenant",
+        event_id="test-event-gn",
+        payload={},
+    )
+
+    handler = EnterpriseMetricHandler()
+    with patch.object(handler, "_on_generate_name") as mock_handler:
+        handler.handle(envelope)
+        mock_handler.assert_called_once_with(envelope)
+
+
+def test_dispatch_prompt_generation(mock_redis):
+    mock_redis.set.return_value = True
+    envelope = TelemetryEnvelope(
+        case=TelemetryCase.PROMPT_GENERATION,
+        tenant_id="test-tenant",
+        event_id="test-event-pg",
+        payload={},
+    )
+
+    handler = EnterpriseMetricHandler()
+    with patch.object(handler, "_on_prompt_generation") as mock_handler:
+        handler.handle(envelope)
+        mock_handler.assert_called_once_with(envelope)
+
+
+def test_all_known_cases_have_handlers(mock_redis):
+    mock_redis.set.return_value = True
+    handler = EnterpriseMetricHandler()
+
+    for case in TelemetryCase:
+        envelope = TelemetryEnvelope(
+            case=case,
+            tenant_id="test-tenant",
+            event_id=f"test-{case.value}",
+            payload={},
+        )
+        handler.handle(envelope)
+
+
+def test_idempotency_duplicate(sample_envelope, mock_redis):
+    mock_redis.set.return_value = None
+
+    handler = EnterpriseMetricHandler()
+    with patch.object(handler, "_on_app_created") as mock_handler:
+        handler.handle(sample_envelope)
+        mock_handler.assert_not_called()
+
+
+def test_idempotency_first_seen(sample_envelope, mock_redis):
+    mock_redis.set.return_value = True
+
+    handler = EnterpriseMetricHandler()
+    is_dup = handler._is_duplicate(sample_envelope)
+
+    assert is_dup is False
+    mock_redis.set.assert_called_once_with(
+        "telemetry:dedup:test-tenant:test-event-123",
+        b"1",
+        nx=True,
+        ex=3600,
+    )
+
+
+def test_idempotency_redis_failure_fails_open(sample_envelope, mock_redis, caplog):
+    mock_redis.set.side_effect = Exception("Redis unavailable")
+
+    handler = EnterpriseMetricHandler()
+    is_dup = handler._is_duplicate(sample_envelope)
+
+    assert is_dup is False
+    assert "Redis unavailable for deduplication check" in caplog.text
+
+
+def test_rehydration_uses_payload(sample_envelope):
+    handler = EnterpriseMetricHandler()
+    payload = handler._rehydrate(sample_envelope)
+
+    assert payload == {"app_id": "app-123", "name": "Test App"}
+
+
+def test_rehydration_fallback():
+    import pickle
+
+    fallback_data = {"fallback": "data"}
+    envelope = TelemetryEnvelope(
+        case=TelemetryCase.APP_CREATED,
+        tenant_id="test-tenant",
+        event_id="test-event-fb",
+        payload={},
+        payload_fallback=pickle.dumps(fallback_data),
+    )
+
+    handler = EnterpriseMetricHandler()
+    payload = handler._rehydrate(envelope)
+
+    assert payload == fallback_data
+
+
+def test_rehydration_emits_degraded_event_on_failure():
+    envelope = TelemetryEnvelope(
+        case=TelemetryCase.APP_CREATED,
+        tenant_id="test-tenant",
+        event_id="test-event-fail",
+        payload={},
+        payload_fallback=None,
+    )
+
+    handler = EnterpriseMetricHandler()
+    with patch("enterprise.telemetry.telemetry_log.emit_metric_only_event") as mock_emit:
+        payload = handler._rehydrate(envelope)
+
+        from enterprise.telemetry.entities import EnterpriseTelemetryEvent
+
+        assert payload == {}
+        mock_emit.assert_called_once()
+        call_args = mock_emit.call_args
+        assert call_args[1]["event_name"] == EnterpriseTelemetryEvent.REHYDRATION_FAILED
+        assert call_args[1]["attributes"]["rehydration_failed"] is True
+
+
+def test_on_app_created_emits_correct_event(mock_redis):
+    mock_redis.set.return_value = True
+    envelope = TelemetryEnvelope(
+        case=TelemetryCase.APP_CREATED,
+        tenant_id="tenant-123",
+        event_id="event-456",
+        payload={"app_id": "app-789", "mode": "chat"},
+    )
+
+    handler = EnterpriseMetricHandler()
+    with (
+        patch("extensions.ext_enterprise_telemetry.get_enterprise_exporter") as mock_get_exporter,
+        patch("enterprise.telemetry.telemetry_log.emit_metric_only_event") as mock_emit,
+    ):
+        mock_exporter = MagicMock()
+        mock_get_exporter.return_value = mock_exporter
+
+        handler._on_app_created(envelope)
+
+        from enterprise.telemetry.entities import EnterpriseTelemetryEvent
+
+        mock_emit.assert_called_once_with(
+            event_name=EnterpriseTelemetryEvent.APP_CREATED,
+            attributes={
+                "dify.app.id": "app-789",
+                "dify.tenant_id": "tenant-123",
+                "dify.event.id": "event-456",
+                "dify.app.mode": "chat",
+            },
+            tenant_id="tenant-123",
+        )
+        from enterprise.telemetry.entities import EnterpriseTelemetryCounter
+
+        mock_exporter.increment_counter.assert_called_once()
+        call_args = mock_exporter.increment_counter.call_args
+        assert call_args[0][0] == EnterpriseTelemetryCounter.APP_CREATED
+        assert call_args[0][1] == 1
+        assert call_args[0][2]["tenant_id"] == "tenant-123"
+        assert call_args[0][2]["app_id"] == "app-789"
+        assert call_args[0][2]["mode"] == "chat"
+
+
+def test_on_app_updated_emits_correct_event(mock_redis):
+    mock_redis.set.return_value = True
+    envelope = TelemetryEnvelope(
+        case=TelemetryCase.APP_UPDATED,
+        tenant_id="tenant-123",
+        event_id="event-456",
+        payload={"app_id": "app-789"},
+    )
+
+    handler = EnterpriseMetricHandler()
+    with (
+        patch("extensions.ext_enterprise_telemetry.get_enterprise_exporter") as mock_get_exporter,
+        patch("enterprise.telemetry.telemetry_log.emit_metric_only_event") as mock_emit,
+    ):
+        mock_exporter = MagicMock()
+        mock_get_exporter.return_value = mock_exporter
+
+        handler._on_app_updated(envelope)
+
+        from enterprise.telemetry.entities import EnterpriseTelemetryEvent
+
+        mock_emit.assert_called_once_with(
+            event_name=EnterpriseTelemetryEvent.APP_UPDATED,
+            attributes={
+                "dify.app.id": "app-789",
+                "dify.tenant_id": "tenant-123",
+                "dify.event.id": "event-456",
+            },
+            tenant_id="tenant-123",
+        )
+        from enterprise.telemetry.entities import EnterpriseTelemetryCounter
+
+        mock_exporter.increment_counter.assert_called_once()
+        call_args = mock_exporter.increment_counter.call_args
+        assert call_args[0][0] == EnterpriseTelemetryCounter.APP_UPDATED
+        assert call_args[0][1] == 1
+        assert call_args[0][2]["tenant_id"] == "tenant-123"
+        assert call_args[0][2]["app_id"] == "app-789"
+
+
+def test_on_app_deleted_emits_correct_event(mock_redis):
+    mock_redis.set.return_value = True
+    envelope = TelemetryEnvelope(
+        case=TelemetryCase.APP_DELETED,
+        tenant_id="tenant-123",
+        event_id="event-456",
+        payload={"app_id": "app-789"},
+    )
+
+    handler = EnterpriseMetricHandler()
+    with (
+        patch("extensions.ext_enterprise_telemetry.get_enterprise_exporter") as mock_get_exporter,
+        patch("enterprise.telemetry.telemetry_log.emit_metric_only_event") as mock_emit,
+    ):
+        mock_exporter = MagicMock()
+        mock_get_exporter.return_value = mock_exporter
+
+        handler._on_app_deleted(envelope)
+
+        from enterprise.telemetry.entities import EnterpriseTelemetryEvent
+
+        mock_emit.assert_called_once_with(
+            event_name=EnterpriseTelemetryEvent.APP_DELETED,
+            attributes={
+                "dify.app.id": "app-789",
+                "dify.tenant_id": "tenant-123",
+                "dify.event.id": "event-456",
+            },
+            tenant_id="tenant-123",
+        )
+        from enterprise.telemetry.entities import EnterpriseTelemetryCounter
+
+        mock_exporter.increment_counter.assert_called_once()
+        call_args = mock_exporter.increment_counter.call_args
+        assert call_args[0][0] == EnterpriseTelemetryCounter.APP_DELETED
+        assert call_args[0][1] == 1
+        assert call_args[0][2]["tenant_id"] == "tenant-123"
+        assert call_args[0][2]["app_id"] == "app-789"
+
+
+def test_on_feedback_created_emits_correct_event(mock_redis):
+    mock_redis.set.return_value = True
+    envelope = TelemetryEnvelope(
+        case=TelemetryCase.FEEDBACK_CREATED,
+        tenant_id="tenant-123",
+        event_id="event-456",
+        payload={
+            "message_id": "msg-001",
+            "app_id": "app-789",
+            "conversation_id": "conv-123",
+            "from_end_user_id": "user-456",
+            "from_account_id": None,
+            "rating": "like",
+            "from_source": "api",
+            "content": "Great!",
+        },
+    )
+
+    handler = EnterpriseMetricHandler()
+    with (
+        patch("extensions.ext_enterprise_telemetry.get_enterprise_exporter") as mock_get_exporter,
+        patch("enterprise.telemetry.telemetry_log.emit_metric_only_event") as mock_emit,
+    ):
+        mock_exporter = MagicMock()
+        mock_exporter.include_content = True
+        mock_get_exporter.return_value = mock_exporter
+
+        handler._on_feedback_created(envelope)
+
+        mock_emit.assert_called_once()
+        call_args = mock_emit.call_args
+        assert call_args[1]["event_name"] == "dify.feedback.created"
+        assert call_args[1]["attributes"]["dify.message.id"] == "msg-001"
+        assert call_args[1]["attributes"]["dify.feedback.content"] == "Great!"
+        assert call_args[1]["tenant_id"] == "tenant-123"
+        assert call_args[1]["user_id"] == "user-456"
+
+        mock_exporter.increment_counter.assert_called_once()
+        counter_args = mock_exporter.increment_counter.call_args
+        assert counter_args[0][2]["app_id"] == "app-789"
+        assert counter_args[0][2]["rating"] == "like"
+
+
+def test_on_feedback_created_without_content(mock_redis):
+    mock_redis.set.return_value = True
+    envelope = TelemetryEnvelope(
+        case=TelemetryCase.FEEDBACK_CREATED,
+        tenant_id="tenant-123",
+        event_id="event-456",
+        payload={
+            "message_id": "msg-001",
+            "app_id": "app-789",
+            "conversation_id": "conv-123",
+            "from_end_user_id": "user-456",
+            "from_account_id": None,
+            "rating": "like",
+            "from_source": "api",
+            "content": "Great!",
+        },
+    )
+
+    handler = EnterpriseMetricHandler()
+    with (
+        patch("extensions.ext_enterprise_telemetry.get_enterprise_exporter") as mock_get_exporter,
+        patch("enterprise.telemetry.telemetry_log.emit_metric_only_event") as mock_emit,
+    ):
+        mock_exporter = MagicMock()
+        mock_exporter.include_content = False
+        mock_get_exporter.return_value = mock_exporter
+
+        handler._on_feedback_created(envelope)
+
+        mock_emit.assert_called_once()
+        call_args = mock_emit.call_args
+        assert "dify.feedback.content" not in call_args[1]["attributes"]

api/tests/unit_tests/services/enterprise/test_enterprise_service.py

@@ -0,0 +1,141 @@
+"""Unit tests for enterprise service integrations.
+
+This module covers the enterprise-only default workspace auto-join behavior:
+- Enterprise mode disabled: no external calls
+- Successful join / skipped join: no errors
+- Failures (network/invalid response/invalid UUID): soft-fail wrapper must not raise
+"""
+
+from unittest.mock import patch
+
+import pytest
+
+from services.enterprise.enterprise_service import (
+    DefaultWorkspaceJoinResult,
+    EnterpriseService,
+    try_join_default_workspace,
+)
+
+
+class TestJoinDefaultWorkspace:
+    def test_join_default_workspace_success(self):
+        account_id = "11111111-1111-1111-1111-111111111111"
+        response = {"workspace_id": "22222222-2222-2222-2222-222222222222", "joined": True, "message": "ok"}
+
+        with patch("services.enterprise.enterprise_service.EnterpriseRequest.send_request") as mock_send_request:
+            mock_send_request.return_value = response
+
+            result = EnterpriseService.join_default_workspace(account_id=account_id)
+
+            assert isinstance(result, DefaultWorkspaceJoinResult)
+            assert result.workspace_id == response["workspace_id"]
+            assert result.joined is True
+            assert result.message == "ok"
+
+            mock_send_request.assert_called_once_with(
+                "POST",
+                "/default-workspace/members",
+                json={"account_id": account_id},
+                timeout=1.0,
+                raise_for_status=True,
+            )
+
+    def test_join_default_workspace_invalid_response_format_raises(self):
+        account_id = "11111111-1111-1111-1111-111111111111"
+
+        with patch("services.enterprise.enterprise_service.EnterpriseRequest.send_request") as mock_send_request:
+            mock_send_request.return_value = "not-a-dict"
+
+            with pytest.raises(ValueError, match="Invalid response format"):
+                EnterpriseService.join_default_workspace(account_id=account_id)
+
+    def test_join_default_workspace_invalid_account_id_raises(self):
+        with pytest.raises(ValueError):
+            EnterpriseService.join_default_workspace(account_id="not-a-uuid")
+
+    def test_join_default_workspace_missing_required_fields_raises(self):
+        account_id = "11111111-1111-1111-1111-111111111111"
+        response = {"workspace_id": "", "message": "ok"}  # missing "joined"
+
+        with patch("services.enterprise.enterprise_service.EnterpriseRequest.send_request") as mock_send_request:
+            mock_send_request.return_value = response
+
+            with pytest.raises(ValueError, match="Invalid response payload"):
+                EnterpriseService.join_default_workspace(account_id=account_id)
+
+    def test_join_default_workspace_joined_without_workspace_id_raises(self):
+        with pytest.raises(ValueError, match="workspace_id must be non-empty when joined is True"):
+            DefaultWorkspaceJoinResult(workspace_id="", joined=True, message="ok")
+
+
+class TestTryJoinDefaultWorkspace:
+    def test_try_join_default_workspace_enterprise_disabled_noop(self):
+        with (
+            patch("services.enterprise.enterprise_service.dify_config") as mock_config,
+            patch("services.enterprise.enterprise_service.EnterpriseService.join_default_workspace") as mock_join,
+        ):
+            mock_config.ENTERPRISE_ENABLED = False
+
+            try_join_default_workspace("11111111-1111-1111-1111-111111111111")
+
+            mock_join.assert_not_called()
+
+    def test_try_join_default_workspace_successful_join_does_not_raise(self):
+        account_id = "11111111-1111-1111-1111-111111111111"
+
+        with (
+            patch("services.enterprise.enterprise_service.dify_config") as mock_config,
+            patch("services.enterprise.enterprise_service.EnterpriseService.join_default_workspace") as mock_join,
+        ):
+            mock_config.ENTERPRISE_ENABLED = True
+            mock_join.return_value = DefaultWorkspaceJoinResult(
+                workspace_id="22222222-2222-2222-2222-222222222222",
+                joined=True,
+                message="ok",
+            )
+
+            # Should not raise
+            try_join_default_workspace(account_id)
+
+            mock_join.assert_called_once_with(account_id=account_id)
+
+    def test_try_join_default_workspace_skipped_join_does_not_raise(self):
+        account_id = "11111111-1111-1111-1111-111111111111"
+
+        with (
+            patch("services.enterprise.enterprise_service.dify_config") as mock_config,
+            patch("services.enterprise.enterprise_service.EnterpriseService.join_default_workspace") as mock_join,
+        ):
+            mock_config.ENTERPRISE_ENABLED = True
+            mock_join.return_value = DefaultWorkspaceJoinResult(
+                workspace_id="",
+                joined=False,
+                message="no default workspace configured",
+            )
+
+            # Should not raise
+            try_join_default_workspace(account_id)
+
+            mock_join.assert_called_once_with(account_id=account_id)
+
+    def test_try_join_default_workspace_api_failure_soft_fails(self):
+        account_id = "11111111-1111-1111-1111-111111111111"
+
+        with (
+            patch("services.enterprise.enterprise_service.dify_config") as mock_config,
+            patch("services.enterprise.enterprise_service.EnterpriseService.join_default_workspace") as mock_join,
+        ):
+            mock_config.ENTERPRISE_ENABLED = True
+            mock_join.side_effect = Exception("network failure")
+
+            # Should not raise
+            try_join_default_workspace(account_id)
+
+            mock_join.assert_called_once_with(account_id=account_id)
+
+    def test_try_join_default_workspace_invalid_account_id_soft_fails(self):
+        with patch("services.enterprise.enterprise_service.dify_config") as mock_config:
+            mock_config.ENTERPRISE_ENABLED = True
+
+            # Should not raise even though UUID parsing fails inside join_default_workspace
+            try_join_default_workspace("not-a-uuid")

api/tests/unit_tests/services/test_account_service.py

@@ -1064,6 +1064,67 @@ class TestRegisterService:
 
     # ==================== Registration Tests ====================
 
+    def test_create_account_and_tenant_calls_default_workspace_join_when_enterprise_enabled(
+        self, mock_db_dependencies, mock_external_service_dependencies, monkeypatch
+    ):
+        """Enterprise-only side effect should be invoked when ENTERPRISE_ENABLED is True."""
+        monkeypatch.setattr(dify_config, "ENTERPRISE_ENABLED", True, raising=False)
+
+        mock_external_service_dependencies["feature_service"].get_system_features.return_value.is_allow_register = True
+        mock_external_service_dependencies["billing_service"].is_email_in_freeze.return_value = False
+
+        mock_account = TestAccountAssociatedDataFactory.create_account_mock(
+            account_id="11111111-1111-1111-1111-111111111111"
+        )
+
+        with (
+            patch("services.account_service.AccountService.create_account") as mock_create_account,
+            patch("services.account_service.TenantService.create_owner_tenant_if_not_exist") as mock_create_workspace,
+            patch("services.enterprise.enterprise_service.try_join_default_workspace") as mock_join_default_workspace,
+        ):
+            mock_create_account.return_value = mock_account
+
+            result = AccountService.create_account_and_tenant(
+                email="test@example.com",
+                name="Test User",
+                interface_language="en-US",
+                password=None,
+            )
+
+            assert result == mock_account
+            mock_create_workspace.assert_called_once_with(account=mock_account)
+            mock_join_default_workspace.assert_called_once_with(str(mock_account.id))
+
+    def test_create_account_and_tenant_does_not_call_default_workspace_join_when_enterprise_disabled(
+        self, mock_db_dependencies, mock_external_service_dependencies, monkeypatch
+    ):
+        """Enterprise-only side effect should not be invoked when ENTERPRISE_ENABLED is False."""
+        monkeypatch.setattr(dify_config, "ENTERPRISE_ENABLED", False, raising=False)
+
+        mock_external_service_dependencies["feature_service"].get_system_features.return_value.is_allow_register = True
+        mock_external_service_dependencies["billing_service"].is_email_in_freeze.return_value = False
+
+        mock_account = TestAccountAssociatedDataFactory.create_account_mock(
+            account_id="11111111-1111-1111-1111-111111111111"
+        )
+
+        with (
+            patch("services.account_service.AccountService.create_account") as mock_create_account,
+            patch("services.account_service.TenantService.create_owner_tenant_if_not_exist") as mock_create_workspace,
+            patch("services.enterprise.enterprise_service.try_join_default_workspace") as mock_join_default_workspace,
+        ):
+            mock_create_account.return_value = mock_account
+
+            AccountService.create_account_and_tenant(
+                email="test@example.com",
+                name="Test User",
+                interface_language="en-US",
+                password=None,
+            )
+
+            mock_create_workspace.assert_called_once_with(account=mock_account)
+            mock_join_default_workspace.assert_not_called()
+
     def test_register_success(self, mock_db_dependencies, mock_external_service_dependencies):
         """Test successful account registration."""
         # Setup mocks
@@ -1115,6 +1176,65 @@ class TestRegisterService:
                 mock_event.send.assert_called_once_with(mock_tenant)
                 self._assert_database_operations_called(mock_db_dependencies["db"])
 
+    def test_register_calls_default_workspace_join_when_enterprise_enabled(
+        self, mock_db_dependencies, mock_external_service_dependencies, monkeypatch
+    ):
+        """Enterprise-only side effect should be invoked after successful register commit."""
+        monkeypatch.setattr(dify_config, "ENTERPRISE_ENABLED", True, raising=False)
+
+        mock_external_service_dependencies["feature_service"].get_system_features.return_value.is_allow_register = True
+        mock_external_service_dependencies["billing_service"].is_email_in_freeze.return_value = False
+
+        mock_account = TestAccountAssociatedDataFactory.create_account_mock(
+            account_id="11111111-1111-1111-1111-111111111111"
+        )
+
+        with (
+            patch("services.account_service.AccountService.create_account") as mock_create_account,
+            patch("services.enterprise.enterprise_service.try_join_default_workspace") as mock_join_default_workspace,
+        ):
+            mock_create_account.return_value = mock_account
+
+            result = RegisterService.register(
+                email="test@example.com",
+                name="Test User",
+                password="password123",
+                language="en-US",
+                create_workspace_required=False,
+            )
+
+            assert result == mock_account
+            mock_join_default_workspace.assert_called_once_with(str(mock_account.id))
+
+    def test_register_does_not_call_default_workspace_join_when_enterprise_disabled(
+        self, mock_db_dependencies, mock_external_service_dependencies, monkeypatch
+    ):
+        """Enterprise-only side effect should not be invoked when ENTERPRISE_ENABLED is False."""
+        monkeypatch.setattr(dify_config, "ENTERPRISE_ENABLED", False, raising=False)
+
+        mock_external_service_dependencies["feature_service"].get_system_features.return_value.is_allow_register = True
+        mock_external_service_dependencies["billing_service"].is_email_in_freeze.return_value = False
+
+        mock_account = TestAccountAssociatedDataFactory.create_account_mock(
+            account_id="11111111-1111-1111-1111-111111111111"
+        )
+
+        with (
+            patch("services.account_service.AccountService.create_account") as mock_create_account,
+            patch("services.enterprise.enterprise_service.try_join_default_workspace") as mock_join_default_workspace,
+        ):
+            mock_create_account.return_value = mock_account
+
+            RegisterService.register(
+                email="test@example.com",
+                name="Test User",
+                password="password123",
+                language="en-US",
+                create_workspace_required=False,
+            )
+
+            mock_join_default_workspace.assert_not_called()
+
     def test_register_with_oauth(self, mock_db_dependencies, mock_external_service_dependencies):
         """Test account registration with OAuth integration."""
         # Setup mocks

api/tests/unit_tests/tasks/test_enterprise_telemetry_task.py

@@ -0,0 +1,69 @@
+"""Unit tests for enterprise telemetry Celery task."""
+
+import json
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from enterprise.telemetry.contracts import TelemetryCase, TelemetryEnvelope
+from tasks.enterprise_telemetry_task import process_enterprise_telemetry
+
+
+@pytest.fixture
+def sample_envelope_json():
+    envelope = TelemetryEnvelope(
+        case=TelemetryCase.APP_CREATED,
+        tenant_id="test-tenant",
+        event_id="test-event-123",
+        payload={"app_id": "app-123"},
+    )
+    return envelope.model_dump_json()
+
+
+def test_process_enterprise_telemetry_success(sample_envelope_json):
+    with patch("tasks.enterprise_telemetry_task.EnterpriseMetricHandler") as mock_handler_class:
+        mock_handler = MagicMock()
+        mock_handler_class.return_value = mock_handler
+
+        process_enterprise_telemetry(sample_envelope_json)
+
+        mock_handler.handle.assert_called_once()
+        call_args = mock_handler.handle.call_args[0][0]
+        assert isinstance(call_args, TelemetryEnvelope)
+        assert call_args.case == TelemetryCase.APP_CREATED
+        assert call_args.tenant_id == "test-tenant"
+        assert call_args.event_id == "test-event-123"
+
+
+def test_process_enterprise_telemetry_invalid_json(caplog):
+    invalid_json = "not valid json"
+
+    process_enterprise_telemetry(invalid_json)
+
+    assert "Failed to process enterprise telemetry envelope" in caplog.text
+
+
+def test_process_enterprise_telemetry_handler_exception(sample_envelope_json, caplog):
+    with patch("tasks.enterprise_telemetry_task.EnterpriseMetricHandler") as mock_handler_class:
+        mock_handler = MagicMock()
+        mock_handler.handle.side_effect = Exception("Handler error")
+        mock_handler_class.return_value = mock_handler
+
+        process_enterprise_telemetry(sample_envelope_json)
+
+        assert "Failed to process enterprise telemetry envelope" in caplog.text
+
+
+def test_process_enterprise_telemetry_validation_error(caplog):
+    invalid_envelope = json.dumps(
+        {
+            "case": "INVALID_CASE",
+            "tenant_id": "test-tenant",
+            "event_id": "test-event",
+            "payload": {},
+        }
+    )
+
+    process_enterprise_telemetry(invalid_envelope)
+
+    assert "Failed to process enterprise telemetry envelope" in caplog.text