feat: enterprise otel exporter (#33138)

Co-authored-by: QuantumGhost <obelisk.reg+git@gmail.com>
Co-authored-by: Yunlu Wen <yunlu.wen@dify.ai>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

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,48 @@ 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
+
+    @property
+    def resolved_parent_context(self) -> tuple[str | None, str | None]:
+        """Resolve cross-workflow parent linking from metadata.
+
+        Extracts typed parent IDs from the untyped ``parent_trace_context``
+        metadata dict (set by tool_node when invoking nested workflows).
+
+        Returns:
+            (trace_correlation_override, parent_span_id_source) where
+            trace_correlation_override is the outer workflow_run_id and
+            parent_span_id_source is the outer node_execution_id.
+        """
+        parent_ctx = self.metadata.get("parent_trace_context")
+        if not isinstance(parent_ctx, dict):
+            return None, None
+        trace_override = parent_ctx.get("parent_workflow_run_id")
+        parent_span = parent_ctx.get("parent_node_execution_id")
+        return (
+            trace_override if isinstance(trace_override, str) else None,
+            parent_span if isinstance(parent_span, str) else None,
+        )
+
     @field_serializer("start_time", "end_time")
     def serialize_datetime(self, dt: datetime | None) -> str | None:
         if dt is None:
@@ -48,7 +90,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 +104,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 +151,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 +159,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 +246,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 +278,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

@@ -15,22 +15,32 @@ from sqlalchemy import select
 from sqlalchemy.orm import Session, sessionmaker
 
 from core.helper.encrypter import batch_decrypt_token, encrypt_token, obfuscated_token
-from core.ops.entities.config_entity import OPS_FILE_PATH, TracingProviderEnum
+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_database import db
 from extensions.ext_storage import storage
-from models.engine import db
+from models.account import Tenant
+from models.dataset import Dataset
 from models.model import App, AppModelConfig, Conversation, Message, MessageFile, TraceAppConfig
+from models.provider import Provider, ProviderCredential, ProviderModel, ProviderModelCredential, ProviderType
+from models.tools import ApiToolProvider, BuiltinToolProvider, MCPToolProvider, WorkflowToolProvider
 from models.workflow import WorkflowAppLog
 from tasks.ops_trace_task import process_trace_tasks
 
@@ -40,9 +50,144 @@ 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))  # type: ignore[attr-defined]
+        return str(name) if name else ""
+
+
+def _lookup_llm_credential_info(
+    tenant_id: str | None, provider: str | None, model: str | None, model_type: str | None = "llm"
+) -> tuple[str | None, str]:
+    """
+    Lookup LLM credential ID and name for the given provider and model.
+    Returns (credential_id, credential_name).
+
+    Handles async timing issues gracefully - if credential is deleted between lookups,
+    returns the ID but empty name rather than failing.
+    """
+    if not tenant_id or not provider:
+        return None, ""
+
+    try:
+        with Session(db.engine) as session:
+            # Try to find provider-level or model-level configuration
+            provider_record = session.scalar(
+                select(Provider).where(
+                    Provider.tenant_id == tenant_id,
+                    Provider.provider_name == provider,
+                    Provider.provider_type == ProviderType.CUSTOM,
+                )
+            )
+
+            if not provider_record:
+                return None, ""
+
+            # Check if there's a model-specific config
+            credential_id = None
+            credential_name = ""
+            is_model_level = False
+
+            if model:
+                # Try model-level first
+                model_record = session.scalar(
+                    select(ProviderModel).where(
+                        ProviderModel.tenant_id == tenant_id,
+                        ProviderModel.provider_name == provider,
+                        ProviderModel.model_name == model,
+                        ProviderModel.model_type == model_type,
+                    )
+                )
+
+                if model_record and model_record.credential_id:
+                    credential_id = model_record.credential_id
+                    is_model_level = True
+
+            if not credential_id and provider_record.credential_id:
+                # Fall back to provider-level credential
+                credential_id = provider_record.credential_id
+                is_model_level = False
+
+            # Lookup credential_name if we have credential_id
+            if credential_id:
+                try:
+                    if is_model_level:
+                        # Query ProviderModelCredential
+                        cred_name = session.scalar(
+                            select(ProviderModelCredential.credential_name).where(
+                                ProviderModelCredential.id == credential_id
+                            )
+                        )
+                    else:
+                        # Query ProviderCredential
+                        cred_name = session.scalar(
+                            select(ProviderCredential.credential_name).where(ProviderCredential.id == credential_id)
+                        )
+
+                    if cred_name:
+                        credential_name = str(cred_name)
+                except Exception as e:
+                    # Credential might have been deleted between lookups (async timing)
+                    # Return ID but empty name rather than failing
+                    logger.warning(
+                        "Failed to lookup credential name for credential_id=%s (provider=%s, model=%s): %s",
+                        credential_id,
+                        provider,
+                        model,
+                        str(e),
+                        exc_info=True,
+                    )
+
+            return credential_id, credential_name
+    except Exception as e:
+        # Database query failed or other unexpected error
+        # Return empty rather than propagating error to telemetry emission
+        logger.warning(
+            "Failed to lookup LLM credential info for tenant_id=%s, provider=%s, model=%s: %s",
+            tenant_id,
+            provider,
+            model,
+            str(e),
+            exc_info=True,
+        )
+        return None, ""
+
+
 class OpsTraceProviderConfigMap(collections.UserDict[str, dict[str, Any]]):
-    def __getitem__(self, key: str) -> dict[str, Any]:
-        match key:
+    def __getitem__(self, provider: str) -> dict[str, Any]:
+        match provider:
             case TracingProviderEnum.LANGFUSE:
                 from core.ops.entities.config_entity import LangfuseConfig
                 from core.ops.langfuse_trace.langfuse_trace import LangFuseDataTrace
@@ -149,7 +294,7 @@ class OpsTraceProviderConfigMap(collections.UserDict[str, dict[str, Any]]):
                 }
 
             case _:
-                raise KeyError(f"Unsupported tracing provider: {key}")
+                raise KeyError(f"Unsupported tracing provider: {provider}")
 
 
 provider_config_map = OpsTraceProviderConfigMap()
@@ -314,6 +459,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:
@@ -466,8 +615,6 @@ class TraceTask:
 
     @classmethod
     def _get_workflow_run_repo(cls):
-        from repositories.factory import DifyAPIRepositoryFactory
-
         if cls._workflow_run_repo is None:
             with cls._repo_lock:
                 if cls._workflow_run_repo is None:
@@ -478,6 +625,77 @@ class TraceTask:
                     cls._workflow_run_repo = DifyAPIRepositoryFactory.create_api_workflow_run_repository(session_maker)
         return cls._workflow_run_repo
 
+    @classmethod
+    def _calculate_workflow_token_split(
+        cls, session: "Session", workflow_run_id: str, tenant_id: str
+    ) -> tuple[int, int]:
+        """Sum prompt/completion tokens across all node executions for a workflow run.
+
+        Reads from the ``outputs`` column (where LLM nodes store ``usage.prompt_tokens``
+        and ``usage.completion_tokens``) rather than ``execution_metadata``, which only
+        carries ``total_tokens``.  Projects only the ``outputs`` column to avoid loading
+        large JSON blobs unnecessarily.
+        """
+        import json
+
+        from models.workflow import WorkflowNodeExecutionModel
+
+        rows = (
+            session.execute(
+                select(WorkflowNodeExecutionModel.outputs).where(
+                    WorkflowNodeExecutionModel.tenant_id == tenant_id,
+                    WorkflowNodeExecutionModel.workflow_run_id == workflow_run_id,
+                )
+            )
+            .scalars()
+            .all()
+        )
+
+        total_prompt = 0
+        total_completion = 0
+
+        for raw in rows:
+            if not raw:
+                continue
+            try:
+                outputs = json.loads(raw) if isinstance(raw, str) else raw
+            except (ValueError, TypeError):
+                continue
+            if not isinstance(outputs, dict):
+                continue
+            usage = outputs.get("usage")
+            if not isinstance(usage, dict):
+                continue
+            prompt = usage.get("prompt_tokens")
+            if isinstance(prompt, (int, float)):
+                total_prompt += int(prompt)
+            completion = usage.get("completion_tokens")
+            if isinstance(completion, (int, float)):
+                total_completion += int(completion)
+
+        return (total_prompt, total_completion)
+
+    @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"
+
     def __init__(
         self,
         trace_type: Any,
@@ -491,6 +709,7 @@ class TraceTask:
         self.trace_type = trace_type
         self.message_id = message_id
         self.workflow_run_id = workflow_execution.id_ if workflow_execution else None
+        self.workflow_total_tokens: int | None = workflow_execution.total_tokens if workflow_execution else None
         self.conversation_id = conversation_id
         self.user_id = user_id
         self.timer = timer
@@ -498,6 +717,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
@@ -509,9 +730,12 @@ class TraceTask:
         preprocess_map = {
             TraceTaskName.CONVERSATION_TRACE: lambda: self.conversation_trace(**self.kwargs),
             TraceTaskName.WORKFLOW_TRACE: lambda: self.workflow_trace(
-                workflow_run_id=self.workflow_run_id, conversation_id=self.conversation_id, user_id=self.user_id
+                workflow_run_id=self.workflow_run_id,
+                conversation_id=self.conversation_id,
+                user_id=self.user_id,
+                total_tokens_override=self.workflow_total_tokens,
             ),
-            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 +751,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)()
@@ -541,6 +768,7 @@ class TraceTask:
         workflow_run_id: str | None,
         conversation_id: str | None,
         user_id: str | None,
+        total_tokens_override: int | None = None,
     ):
         if not workflow_run_id:
             return {}
@@ -560,7 +788,7 @@ class TraceTask:
         workflow_run_version = workflow_run.version
         error = workflow_run.error or ""
 
-        total_tokens = workflow_run.total_tokens
+        total_tokens = total_tokens_override if total_tokens_override is not None else workflow_run.total_tokens
 
         file_list = workflow_run_inputs.get("sys.file") or []
         query = workflow_run_inputs.get("query") or workflow_run_inputs.get("sys.query") or ""
@@ -581,8 +809,18 @@ class TraceTask:
                     Message.workflow_run_id == workflow_run_id,
                 )
                 message_id = session.scalar(message_data_stmt)
+            prompt_tokens, completion_tokens = self._calculate_workflow_token_split(
+                session, workflow_run_id=workflow_run_id, tenant_id=tenant_id
+            )
 
-        metadata = {
+        from core.telemetry.gateway import is_enterprise_telemetry_enabled
+
+        if is_enterprise_telemetry_enabled():
+            app_name, workspace_name = _lookup_app_and_workspace_names(workflow_run.app_id, tenant_id)
+        else:
+            app_name, workspace_name = "", ""
+
+        metadata: dict[str, Any] = {
             "workflow_id": workflow_id,
             "conversation_id": conversation_id,
             "workflow_run_id": workflow_run_id,
@@ -595,8 +833,14 @@ class TraceTask:
             "triggered_from": workflow_run.triggered_from,
             "user_id": user_id,
             "app_id": workflow_run.app_id,
+            "app_name": app_name,
+            "workspace_name": workspace_name,
         }
 
+        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,
             workflow_data=workflow_run.to_dict(),
@@ -611,6 +855,8 @@ 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,
             query=query,
             metadata=metadata,
@@ -618,10 +864,11 @@ class TraceTask:
             message_id=message_id,
             start_time=workflow_run.created_at,
             end_time=workflow_run.finished_at,
+            invoked_by=self._get_user_id_from_metadata(metadata),
         )
         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)
@@ -644,6 +891,19 @@ class TraceTask:
 
         streaming_metrics = self._extract_streaming_metrics(message_data)
 
+        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)
+
+        from core.telemetry.gateway import is_enterprise_telemetry_enabled
+
+        if is_enterprise_telemetry_enabled():
+            app_name, workspace_name = _lookup_app_and_workspace_names(message_data.app_id, tenant_id)
+        else:
+            app_name, workspace_name = "", ""
+
         metadata = {
             "conversation_id": message_data.conversation_id,
             "ls_provider": message_data.model_provider,
@@ -655,7 +915,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
 
@@ -672,7 +939,9 @@ class TraceTask:
             outputs=message_data.answer,
             file_list=file_list,
             start_time=created_at,
-            end_time=created_at + timedelta(seconds=message_data.provider_response_latency),
+            end_time=message_data.updated_at
+            if message_data.updated_at and message_data.updated_at > created_at
+            else created_at + timedelta(seconds=message_data.provider_response_latency),
             metadata=metadata,
             message_file_data=message_file_data,
             conversation_mode=conversation_mode,
@@ -697,6 +966,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 +1009,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 +1050,52 @@ 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)
+
+        from core.telemetry.gateway import is_enterprise_telemetry_enabled
+
+        if is_enterprise_telemetry_enabled():
+            app_name, workspace_name = _lookup_app_and_workspace_names(message_data.app_id, tenant_id)
+        else:
+            app_name, workspace_name = "", ""
+
+        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 +1106,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 +1165,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()
@@ -890,6 +1223,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,
@@ -904,6 +1239,182 @@ class TraceTask:
 
         return generate_name_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 {}
+
+        from core.telemetry.gateway import is_enterprise_telemetry_enabled
+
+        if is_enterprise_telemetry_enabled():
+            app_name, workspace_name = _lookup_app_and_workspace_names(
+                node_data.get("app_id"), node_data.get("tenant_id")
+            )
+        else:
+            app_name, workspace_name = "", ""
+
+        # Try tool credential lookup first
+        credential_id = node_data.get("credential_id")
+        if is_enterprise_telemetry_enabled():
+            credential_name = _lookup_credential_name(credential_id, node_data.get("credential_provider_type"))
+            # If no credential_id found (e.g., LLM nodes), try LLM credential lookup
+            if not credential_id:
+                llm_cred_id, llm_cred_name = _lookup_llm_credential_info(
+                    tenant_id=node_data.get("tenant_id"),
+                    provider=node_data.get("model_provider"),
+                    model=node_data.get("model_name"),
+                    model_type="llm",
+                )
+                if llm_cred_id:
+                    credential_id = llm_cred_id
+                    credential_name = llm_cred_name
+        else:
+            credential_name = ""
+        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"),
+            "invoke_from": node_data.get("invoke_from"),
+            "credential_id": credential_id,
+            "credential_name": credential_name,
+            "dataset_ids": node_data.get("dataset_ids"),
+            "dataset_names": node_data.get("dataset_names"),
+            "plugin_name": node_data.get("plugin_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
+            if conversation_id:
+                metadata["conversation_id"] = conversation_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 isinstance(node_trace, WorkflowNodeTraceInfo):
+            return node_trace
+        return DraftNodeExecutionTrace(**node_trace.model_dump())
+
     def _extract_streaming_metrics(self, message_data) -> dict:
         if not message_data.message_metadata:
             return {}
@@ -937,13 +1448,17 @@ 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 +1494,27 @@ 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 emit as gateway_emit
+from core.telemetry.gateway import get_trace_task_to_case
+
+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 = get_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,239 @@
+"""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
+from extensions.ext_storage import storage
+
+if TYPE_CHECKING:
+    from core.ops.ops_trace_manager import TraceQueueManager
+    from enterprise.telemetry.contracts import TelemetryCase
+
+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] | None = None
+_case_routing: dict[TelemetryCase, CaseRoute] | None = None
+
+
+def _get_case_to_trace_task() -> dict[TelemetryCase, TraceTaskName]:
+    global _case_to_trace_task
+    if _case_to_trace_task is None:
+        from enterprise.telemetry.contracts import TelemetryCase
+
+        _case_to_trace_task = {
+            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,
+        }
+    return _case_to_trace_task
+
+
+def get_trace_task_to_case() -> dict[TraceTaskName, TelemetryCase]:
+    """Return TraceTaskName → TelemetryCase (inverse of _get_case_to_trace_task)."""
+    return {v: k for k, v in _get_case_to_trace_task().items()}
+
+
+def _get_case_routing() -> dict[TelemetryCase, CaseRoute]:
+    global _case_routing
+    if _case_routing is None:
+        from enterprise.telemetry.contracts import CaseRoute, SignalType, TelemetryCase
+
+        _case_routing = {
+            # 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),
+        }
+    return _case_routing
+
+
+def __getattr__(name: str) -> dict:
+    """Lazy module-level access to routing tables."""
+    if name == "CASE_ROUTING":
+        return _get_case_routing()
+    if name == "CASE_TO_TRACE_TASK":
+        return _get_case_to_trace_task()
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+
+
+# ---------------------------------------------------------------------------
+# 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 = _get_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 == 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 = _get_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, user_id=context.get("user_id"), **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") or ""
+    event_id = str(uuid.uuid4())
+
+    payload_for_envelope, payload_ref = _handle_payload_sizing(payload, tenant_id, event_id)
+
+    from enterprise.telemetry.contracts import TelemetryEnvelope
+
+    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,
+    )