fix: add missing metrics back and enhance typing (#40935)

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

api/configs/enterprise/__init__.py

@@ -83,11 +83,6 @@ class EnterpriseTelemetryConfig(BaseSettings):
         default=False,
     )
 
-    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/controllers/console/app/generator.py

@@ -364,6 +364,7 @@ class InstructionGenerateApi(Resource):
                     case "llm":
                         return LLMGenerator.generate_rule_config(
                             current_tenant_id,
+                            app_id=req_data.flow_id,
                             args=RuleGeneratePayload(
                                 instruction=req_data.instruction,
                                 model_config=req_data.model_config_data,
@@ -373,6 +374,7 @@ class InstructionGenerateApi(Resource):
                     case "agent":
                         return LLMGenerator.generate_rule_config(
                             current_tenant_id,
+                            app_id=req_data.flow_id,
                             args=RuleGeneratePayload(
                                 instruction=req_data.instruction,
                                 model_config=req_data.model_config_data,
@@ -382,6 +384,7 @@ class InstructionGenerateApi(Resource):
                     case "code":
                         return LLMGenerator.generate_code(
                             tenant_id=current_tenant_id,
+                            app_id=req_data.flow_id,
                             args=RuleCodeGeneratePayload(
                                 instruction=req_data.instruction,
                                 model_config=req_data.model_config_data,

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

@@ -72,7 +72,8 @@ from core.app.task_pipeline.based_generate_task_pipeline import BasedGenerateTas
 from core.app.task_pipeline.message_cycle_manager import MessageCycleManager
 from core.base.tts import AppGeneratorTTSPublisher, AudioTrunk
 from core.db.session_factory import session_factory
-from core.ops.ops_trace_manager import TraceQueueManager
+from core.ops.entities.trace_entity import TraceTaskName
+from core.ops.ops_trace_manager import TraceQueueManager, TraceTask
 from core.repositories.human_input_repository import HumanInputFormRepositoryImpl
 from core.workflow.file_reference import resolve_file_record_id
 from core.workflow.nodes.human_input.pause_reason import HumanInputRequired
@@ -779,6 +780,7 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
             with self._database_session() as session:
                 # Save message
                 self._save_message(session=session, graph_runtime_state=resolved_state)
+            self._emit_message_trace()
 
             yield workflow_finish_resp
         elif event.stopped_by in (
@@ -789,6 +791,7 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
             with self._database_session() as session:
                 # Save message
                 self._save_message(session=session)
+            self._emit_message_trace()
 
         yield self._message_end_to_stream_response()
 
@@ -816,6 +819,7 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         if not self._message_saved_on_pause:
             with self._database_session() as session:
                 self._save_message(session=session, graph_runtime_state=resolved_state)
+            self._emit_message_trace()
 
         yield self._message_end_to_stream_response()
 
@@ -1101,6 +1105,18 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
             )
         session.add_all(message_files)
 
+    def _emit_message_trace(self) -> None:
+        trace_manager = self._application_generate_entity.trace_manager
+        if trace_manager:
+            trace_manager.add_trace_task(
+                TraceTask(
+                    TraceTaskName.MESSAGE_TRACE,
+                    conversation_id=self._conversation_id,
+                    message_id=self._message_id,
+                    trace_session_id=self._application_generate_entity.extras.get("trace_session_id"),
+                )
+            )
+
     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/llm_generator/llm_generator.py

@@ -27,6 +27,8 @@ 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 PromptGenerationEvent, TelemetryContext
+from core.telemetry import emit as telemetry_emit
 from extensions.ext_database import db
 from extensions.ext_storage import storage
 from graphon.enums import WorkflowNodeExecutionMetadataKey
@@ -139,6 +141,51 @@ class StructuredOutputResultDict(TypedDict):
 
 
 class LLMGenerator:
+    @staticmethod
+    def _emit_prompt_generation(
+        *,
+        tenant_id: str,
+        app_id: str | None = None,
+        operation_type: str,
+        instruction: str,
+        generated_output: str,
+        model_provider: str,
+        model_name: str,
+        timer: dict,
+        result: LLMResult | None = None,
+        error: str | None = None,
+    ) -> None:
+        """Emit a PromptGenerationEvent via the core.telemetry facade."""
+        try:
+            usage = result.usage if result else None
+            telemetry_emit(
+                PromptGenerationEvent(
+                    context=TelemetryContext(
+                        tenant_id=tenant_id,
+                        app_id=app_id,
+                    ),
+                    payload={
+                        "tenant_id": tenant_id,
+                        "app_id": app_id,
+                        "operation_type": operation_type,
+                        "instruction": instruction,
+                        "generated_output": generated_output,
+                        "model_provider": model_provider,
+                        "model_name": model_name,
+                        "prompt_tokens": usage.prompt_tokens if usage else 0,
+                        "completion_tokens": usage.completion_tokens if usage else 0,
+                        "total_tokens": usage.total_tokens if usage else 0,
+                        "latency": (timer["end"] - timer["start"]).total_seconds() if timer.get("end") else 0.0,
+                        "total_price": float(usage.total_price) if usage and usage.total_price else None,
+                        "currency": usage.currency if usage else None,
+                        "timer": timer,
+                        "error": error,
+                    },
+                )
+            )
+        except Exception:
+            logger.debug("Failed to emit prompt_generation telemetry", exc_info=True)
+
     @classmethod
     def generate_conversation_name(
         cls, tenant_id: str, query, conversation_id: str | None = None, app_id: str | None = None
@@ -404,12 +451,12 @@ class LLMGenerator:
         return "\n\n".join(sections) + "\n\n"
 
     @classmethod
-    def generate_rule_config(cls, tenant_id: str, args: RuleGeneratePayload):
+    def generate_rule_config(cls, tenant_id: str, args: RuleGeneratePayload, *, app_id: str | None = None):
         output_parser = RuleConfigGeneratorOutputParser()
 
         error = ""
         error_step = ""
-        rule_config = {"prompt": "", "variables": [], "opening_statement": "", "error": ""}
+        rule_config: dict[str, Any] = {"prompt": "", "variables": [], "opening_statement": "", "error": ""}
         model_parameters = args.model_config_data.completion_params
         if args.no_variable:
             prompt_template = PromptTemplateParser(WORKFLOW_RULE_CONFIG_PROMPT_GENERATE_TEMPLATE)
@@ -424,7 +471,6 @@ class LLMGenerator:
             no_variable_prompt_messages: list[PromptMessage] = [UserPromptMessage(content=prompt_generate)]
 
             model_manager = ModelManager.for_tenant(tenant_id=tenant_id)
-
             model_instance = model_manager.get_model_instance(
                 tenant_id=tenant_id,
                 model_type=ModelType.LLM,
@@ -432,20 +478,35 @@ class LLMGenerator:
                 model=args.model_config_data.name,
             )
 
-            try:
-                response: LLMResult = model_instance.invoke_llm(
-                    prompt_messages=list(no_variable_prompt_messages), model_parameters=model_parameters, stream=False
-                )
+            response: LLMResult | None = None
+            with measure_time() as timer:
+                try:
+                    response = model_instance.invoke_llm(
+                        prompt_messages=list(no_variable_prompt_messages),
+                        model_parameters=model_parameters,
+                        stream=False,
+                    )
+                    rule_config["prompt"] = response.message.get_text_content()
+                except InvokeError as e:
+                    error = str(e)
+                    error_step = "generate rule config"
+                except Exception as e:
+                    logger.exception("Failed to generate rule config, model: %s", args.model_config_data.name)
+                    error = str(e)
+                    error_step = "generate rule config"
 
-                rule_config["prompt"] = response.message.get_text_content()
-
-            except InvokeError as e:
-                error = str(e)
-                error_step = "generate rule config"
-            except Exception as e:
-                logger.exception("Failed to generate rule config, model: %s", args.model_config_data.name)
-                error = str(e)
-                error_step = "generate rule config"
+            cls._emit_prompt_generation(
+                tenant_id=tenant_id,
+                app_id=app_id,
+                operation_type="rule_generate",
+                instruction=args.instruction,
+                generated_output=rule_config.get("prompt", ""),
+                model_provider=args.model_config_data.provider,
+                model_name=args.model_config_data.name,
+                timer=timer,
+                result=response,
+                error=error or None,
+            )
 
             rule_config["error"] = f"Failed to {error_step}. Error: {error}" if error else ""
 
@@ -478,61 +539,79 @@ class LLMGenerator:
             model=args.model_config_data.name,
         )
 
-        try:
+        prompt_content: LLMResult | None = None
+        with measure_time() as timer:
             try:
-                # the first step to generate the task prompt
-                prompt_content: LLMResult = model_instance.invoke_llm(
-                    prompt_messages=list(prompt_generate_messages), model_parameters=model_parameters, stream=False
+                try:
+                    # the first step to generate the task prompt
+                    prompt_content = model_instance.invoke_llm(
+                        prompt_messages=list(prompt_generate_messages),
+                        model_parameters=model_parameters,
+                        stream=False,
+                    )
+                except InvokeError as e:
+                    error = str(e)
+                    error_step = "generate prefix prompt"
+                    rule_config["error"] = f"Failed to {error_step}. Error: {error}"
+                    return rule_config
+
+                rule_config["prompt"] = prompt_content.message.get_text_content()
+
+                parameter_generate_prompt = parameter_template.format(
+                    inputs={
+                        "INPUT_TEXT": prompt_content.message.get_text_content(),
+                    },
+                    remove_template_variables=False,
                 )
-            except InvokeError as e:
-                error = str(e)
-                error_step = "generate prefix prompt"
-                rule_config["error"] = f"Failed to {error_step}. Error: {error}" if error else ""
-
-                return rule_config
-
-            rule_config["prompt"] = prompt_content.message.get_text_content()
-
-            parameter_generate_prompt = parameter_template.format(
-                inputs={
-                    "INPUT_TEXT": prompt_content.message.get_text_content(),
-                },
-                remove_template_variables=False,
-            )
-            parameter_messages: list[PromptMessage] = [UserPromptMessage(content=parameter_generate_prompt)]
-
-            # the second step to generate the task_parameter and task_statement
-            statement_generate_prompt = statement_template.format(
-                inputs={
-                    "TASK_DESCRIPTION": args.instruction,
-                    "INPUT_TEXT": prompt_content.message.get_text_content(),
-                },
-                remove_template_variables=False,
-            )
-            statement_messages: list[PromptMessage] = [UserPromptMessage(content=statement_generate_prompt)]
-
-            try:
-                parameter_content: LLMResult = model_instance.invoke_llm(
-                    prompt_messages=list(parameter_messages), model_parameters=model_parameters, stream=False
+                statement_generate_prompt = statement_template.format(
+                    inputs={
+                        "TASK_DESCRIPTION": args.instruction,
+                        "INPUT_TEXT": prompt_content.message.get_text_content(),
+                    },
+                    remove_template_variables=False,
                 )
-                rule_config["variables"] = re.findall(r'"\s*([^"]+)\s*"', parameter_content.message.get_text_content())
-            except InvokeError as e:
-                error = str(e)
-                error_step = "generate variables"
 
-            try:
-                statement_content: LLMResult = model_instance.invoke_llm(
-                    prompt_messages=list(statement_messages), model_parameters=model_parameters, stream=False
-                )
-                rule_config["opening_statement"] = statement_content.message.get_text_content()
-            except InvokeError as e:
-                error = str(e)
-                error_step = "generate conversation opener"
+                try:
+                    parameter_content: LLMResult = model_instance.invoke_llm(
+                        prompt_messages=[UserPromptMessage(content=parameter_generate_prompt)],
+                        model_parameters=model_parameters,
+                        stream=False,
+                    )
+                    rule_config["variables"] = re.findall(
+                        r'"\s*([^"]+)\s*"', parameter_content.message.get_text_content()
+                    )
+                except InvokeError as e:
+                    error = str(e)
+                    error_step = "generate variables"
 
-        except Exception as e:
-            logger.exception("Failed to generate rule config, model: %s", args.model_config_data.name)
-            error = str(e)
-            error_step = "handle unexpected exception"
+                try:
+                    statement_content: LLMResult = model_instance.invoke_llm(
+                        prompt_messages=[UserPromptMessage(content=statement_generate_prompt)],
+                        model_parameters=model_parameters,
+                        stream=False,
+                    )
+                    rule_config["opening_statement"] = statement_content.message.get_text_content()
+                except InvokeError as e:
+                    error = str(e)
+                    error_step = "generate conversation opener"
+
+            except Exception as e:
+                logger.exception("Failed to generate rule config, model: %s", args.model_config_data.name)
+                error = str(e)
+                error_step = "handle unexpected exception"
+
+        cls._emit_prompt_generation(
+            tenant_id=tenant_id,
+            app_id=app_id,
+            operation_type="rule_generate",
+            instruction=args.instruction,
+            generated_output=rule_config.get("prompt", ""),
+            model_provider=args.model_config_data.provider,
+            model_name=args.model_config_data.name,
+            timer=timer,
+            result=prompt_content,
+            error=error or None,
+        )
 
         rule_config["error"] = f"Failed to {error_step}. Error: {error}" if error else ""
 
@@ -543,6 +622,8 @@ class LLMGenerator:
         cls,
         tenant_id: str,
         args: RuleCodeGeneratePayload,
+        *,
+        app_id: str | None = None,
     ) -> CodeGenerateResultDict:
         if args.code_language == "python":
             prompt_template = PromptTemplateParser(PYTHON_CODE_GENERATOR_PROMPT_TEMPLATE)
@@ -567,22 +648,42 @@ class LLMGenerator:
 
         prompt_messages: list[PromptMessage] = [UserPromptMessage(content=prompt)]
         model_parameters = args.model_config_data.completion_params
-        try:
-            response: LLMResult = model_instance.invoke_llm(
-                prompt_messages=list(prompt_messages), model_parameters=model_parameters, stream=False
-            )
 
-            generated_code = response.message.get_text_content()
-            return {"code": generated_code, "language": args.code_language, "error": ""}
+        response: LLMResult | None = None
+        error: str | None = None
+        generated_code = ""
+        with measure_time() as timer:
+            try:
+                response = model_instance.invoke_llm(
+                    prompt_messages=list(prompt_messages), model_parameters=model_parameters, stream=False
+                )
+                generated_code = response.message.get_text_content()
+            except InvokeError as e:
+                error = str(e)
+            except Exception as e:
+                logger.exception(
+                    "Failed to invoke LLM model, model: %s, language: %s",
+                    args.model_config_data.name,
+                    args.code_language,
+                )
+                error = str(e)
 
-        except InvokeError as e:
-            error = str(e)
+        cls._emit_prompt_generation(
+            tenant_id=tenant_id,
+            app_id=app_id,
+            operation_type="code_generate",
+            instruction=args.instruction,
+            generated_output=generated_code,
+            model_provider=args.model_config_data.provider,
+            model_name=args.model_config_data.name,
+            timer=timer,
+            result=response,
+            error=error,
+        )
+
+        if error:
             return {"code": "", "language": args.code_language, "error": f"Failed to generate code. Error: {error}"}
-        except Exception as e:
-            logger.exception(
-                "Failed to invoke LLM model, model: %s, language: %s", args.model_config_data.name, args.code_language
-            )
-            return {"code": "", "language": args.code_language, "error": f"An unexpected error occurred: {str(e)}"}
+        return {"code": generated_code, "language": args.code_language, "error": ""}
 
     @classmethod
     def generate_qa_document(cls, tenant_id: str, query, document_language: str):
@@ -613,7 +714,7 @@ class LLMGenerator:
 
     @classmethod
     def generate_structured_output(
-        cls, tenant_id: str, args: RuleStructuredOutputPayload
+        cls, tenant_id: str, args: RuleStructuredOutputPayload, *, app_id: str | None = None
     ) -> StructuredOutputResultDict:
         model_manager = ModelManager.for_tenant(tenant_id=tenant_id)
         model_instance = model_manager.get_model_instance(
@@ -629,30 +730,44 @@ class LLMGenerator:
         ]
         model_parameters = args.model_config_data.completion_params
 
-        try:
-            response: LLMResult = model_instance.invoke_llm(
-                prompt_messages=list(prompt_messages), model_parameters=model_parameters, stream=False
-            )
-
-            raw_content = response.message.get_text_content()
-
+        response: LLMResult | None = None
+        error: str | None = None
+        generated_output = ""
+        with measure_time() as timer:
             try:
-                parsed_content = json.loads(raw_content)
-            except json.JSONDecodeError:
-                parsed_content = json_repair.loads(raw_content)
+                response = model_instance.invoke_llm(
+                    prompt_messages=list(prompt_messages), model_parameters=model_parameters, stream=False
+                )
+                raw_content = response.message.get_text_content()
+                try:
+                    parsed_content = json.loads(raw_content)
+                except json.JSONDecodeError:
+                    parsed_content = json_repair.loads(raw_content)
+                if not isinstance(parsed_content, dict | list):
+                    raise ValueError(f"Failed to parse structured output from llm: {raw_content}")
+                generated_output = json.dumps(parsed_content, indent=2, ensure_ascii=False)
+            except InvokeError as e:
+                error = str(e)
+            except Exception as e:
+                logger.exception("Failed to invoke LLM model, model: %s", args.model_config_data.name)
+                error = str(e)
 
-            if not isinstance(parsed_content, dict | list):
-                raise ValueError(f"Failed to parse structured output from llm: {raw_content}")
+        cls._emit_prompt_generation(
+            tenant_id=tenant_id,
+            app_id=app_id,
+            operation_type="structured_output",
+            instruction=args.instruction,
+            generated_output=generated_output,
+            model_provider=args.model_config_data.provider,
+            model_name=args.model_config_data.name,
+            timer=timer,
+            result=response,
+            error=error,
+        )
 
-            generated_json_schema = json.dumps(parsed_content, indent=2, ensure_ascii=False)
-            return {"output": generated_json_schema, "error": ""}
-
-        except InvokeError as e:
-            error = str(e)
+        if error:
             return {"output": "", "error": f"Failed to generate JSON Schema. Error: {error}"}
-        except Exception as e:
-            logger.exception("Failed to invoke LLM model, model: %s", args.model_config_data.name)
-            return {"output": "", "error": f"An unexpected error occurred: {str(e)}"}
+        return {"output": generated_output, "error": ""}
 
     @staticmethod
     def instruction_modify_legacy(
@@ -673,6 +788,7 @@ class LLMGenerator:
         if not last_run:
             return LLMGenerator.__instruction_modify_common(
                 tenant_id=tenant_id,
+                app_id=flow_id,
                 model_config=model_config,
                 last_run=None,
                 current=current,
@@ -688,6 +804,7 @@ class LLMGenerator:
         }
         return LLMGenerator.__instruction_modify_common(
             tenant_id=tenant_id,
+            app_id=flow_id,
             model_config=model_config,
             last_run=last_run_dict,
             current=current,
@@ -730,6 +847,7 @@ class LLMGenerator:
         if not last_run:  # Node is not executed yet
             return LLMGenerator.__instruction_modify_common(
                 tenant_id=tenant_id,
+                app_id=flow_id,
                 model_config=model_config,
                 last_run=None,
                 current=current,
@@ -763,6 +881,7 @@ class LLMGenerator:
 
         return LLMGenerator.__instruction_modify_common(
             tenant_id=tenant_id,
+            app_id=flow_id,
             model_config=model_config,
             last_run=last_run_dict,
             current=current,
@@ -775,6 +894,7 @@ class LLMGenerator:
     @staticmethod
     def __instruction_modify_common(
         tenant_id: str,
+        app_id: str | None,
         model_config: ModelConfig,
         last_run: dict[str, Any] | None,
         current: str | None,
@@ -821,24 +941,45 @@ class LLMGenerator:
         ]
         model_parameters = {"temperature": 0.4}
 
-        try:
-            response: LLMResult = model_instance.invoke_llm(
-                prompt_messages=list(prompt_messages), model_parameters=model_parameters, stream=False
-            )
+        response: LLMResult | None = None
+        error: str | None = None
+        generated_output = ""
+        data: dict = {}
+        with measure_time() as timer:
+            try:
+                response = model_instance.invoke_llm(
+                    prompt_messages=list(prompt_messages), model_parameters=model_parameters, stream=False
+                )
+                generated_raw = response.message.get_text_content()
+                first_brace = generated_raw.find("{")
+                last_brace = generated_raw.rfind("}")
+                if first_brace == -1 or last_brace == -1 or last_brace < first_brace:
+                    raise ValueError(f"Could not find a valid JSON object in response: {generated_raw}")
+                json_str = generated_raw[first_brace : last_brace + 1]
+                parsed = json_repair.loads(json_str)
+                if not isinstance(parsed, dict):
+                    raise TypeError(f"Expected a JSON object, but got {type(parsed).__name__}")
+                data = parsed
+                generated_output = json_str
+            except InvokeError as e:
+                error = str(e)
+            except Exception as e:
+                logger.exception("Failed to invoke LLM model, model: %s", json.dumps(model_config.name), exc_info=True)
+                error = str(e)
 
-            generated_raw = response.message.get_text_content()
-            first_brace = generated_raw.find("{")
-            last_brace = generated_raw.rfind("}")
-            if first_brace == -1 or last_brace == -1 or last_brace < first_brace:
-                raise ValueError(f"Could not find a valid JSON object in response: {generated_raw}")
-            json_str = generated_raw[first_brace : last_brace + 1]
-            data = json_repair.loads(json_str)
-            if not isinstance(data, dict):
-                raise TypeError(f"Expected a JSON object, but got {type(data).__name__}")
-            return data
-        except InvokeError as e:
-            error = str(e)
-            return {"error": f"Failed to generate code. Error: {error}"}
-        except Exception as e:
-            logger.exception("Failed to invoke LLM model, model: %s", json.dumps(model_config.name), exc_info=True)
-            return {"error": f"An unexpected error occurred: {str(e)}"}
+        LLMGenerator._emit_prompt_generation(
+            tenant_id=tenant_id,
+            app_id=app_id,
+            operation_type="instruction_modify",
+            instruction=instruction,
+            generated_output=generated_output,
+            model_provider=model_config.provider,
+            model_name=model_config.name,
+            timer=timer,
+            result=response,
+            error=error,
+        )
+
+        if error:
+            return {"error": f"Failed to generate. Error: {error}"}
+        return data

api/core/telemetry/__init__.py

@@ -6,36 +6,28 @@ 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)
-
+from core.telemetry.events import (
+    AppCreatedEvent,
+    AppDeletedEvent,
+    AppUpdatedEvent,
+    DraftNodeExecutionTraceEvent,
+    FeedbackCreatedEvent,
+    PromptGenerationEvent,
+    PromptGenerationPayload,
+    TelemetryContext,
+    TelemetryEvent,
+)
+from core.telemetry.gateway import emit
 
 __all__ = [
+    "AppCreatedEvent",
+    "AppDeletedEvent",
+    "AppUpdatedEvent",
+    "DraftNodeExecutionTraceEvent",
+    "FeedbackCreatedEvent",
+    "PromptGenerationEvent",
+    "PromptGenerationPayload",
     "TelemetryContext",
     "TelemetryEvent",
     "TraceTaskName",

api/core/telemetry/events.py

@@ -1,10 +1,97 @@
 from __future__ import annotations
 
+from collections.abc import Mapping
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, Any
+from typing import Any, NotRequired, Protocol, TypedDict, runtime_checkable
 
-if TYPE_CHECKING:
-    from core.ops.entities.trace_entity import TraceTaskName
+from core.ops.entities.trace_entity import TraceTaskName
+from enterprise.telemetry.contracts import SignalType, TelemetryCase
+
+# ---------------------------------------------------------------------------
+# Context shapes
+# ---------------------------------------------------------------------------
+
+
+class TraceContext(TypedDict):
+    """Common ``context`` shape for TRACE-routed cases."""
+
+    tenant_id: str | None
+    user_id: NotRequired[str | None]
+    app_id: NotRequired[str | None]
+
+
+class MetricLogContext(TypedDict):
+    """Common ``context`` shape for METRIC_LOG-routed cases."""
+
+    tenant_id: str | None
+
+
+# ---------------------------------------------------------------------------
+# Payload shapes
+# ---------------------------------------------------------------------------
+
+
+class NodeExecutionPayload(TypedDict):
+    """``payload`` shape for ``TelemetryCase.NODE_EXECUTION`` and ``DRAFT_NODE_EXECUTION``."""
+
+    node_execution_data: dict[str, Any]
+
+
+class AppCreatedPayload(TypedDict):
+    """``payload`` shape for ``TelemetryCase.APP_CREATED``."""
+
+    app_id: str | None
+    mode: NotRequired[str | None]
+
+
+class AppUpdatedPayload(TypedDict):
+    """``payload`` shape for ``TelemetryCase.APP_UPDATED``."""
+
+    app_id: str | None
+
+
+class AppDeletedPayload(TypedDict):
+    """``payload`` shape for ``TelemetryCase.APP_DELETED``."""
+
+    app_id: str | None
+
+
+class PromptGenerationPayload(TypedDict):
+    """``payload`` shape for ``TelemetryCase.PROMPT_GENERATION``."""
+
+    tenant_id: str
+    app_id: NotRequired[str | None]
+    operation_type: str
+    instruction: str
+    generated_output: str
+    model_provider: str
+    model_name: str
+    prompt_tokens: int
+    completion_tokens: int
+    total_tokens: int
+    latency: float
+    total_price: NotRequired[float | None]
+    currency: NotRequired[str | None]
+    timer: NotRequired[dict]
+    error: NotRequired[str | None]
+
+
+class FeedbackCreatedPayload(TypedDict):
+    """``payload`` shape for ``TelemetryCase.FEEDBACK_CREATED``."""
+
+    message_id: str
+    app_id: str | None
+    conversation_id: NotRequired[str | None]
+    from_end_user_id: NotRequired[str | None]
+    from_account_id: NotRequired[str | None]
+    rating: NotRequired[str | None]
+    from_source: NotRequired[str | None]
+    content: NotRequired[str | None]
+
+
+# ---------------------------------------------------------------------------
+# Shared context dataclass
+# ---------------------------------------------------------------------------
 
 
 @dataclass(frozen=True)
@@ -14,8 +101,95 @@ class TelemetryContext:
     app_id: str | None = None
 
 
+# ---------------------------------------------------------------------------
+# Event protocol — every concrete event must declare these routing fields
+# so the gateway can dispatch without external mapping tables.
+# ---------------------------------------------------------------------------
+
+
+@runtime_checkable
+class TelemetryEvent(Protocol):
+    """Structural contract for all telemetry events.
+
+    All attributes are declared as read-only ``@property`` so that
+    frozen dataclasses satisfy the protocol (a bare attribute
+    annotation would require writability).
+    """
+
+    @property
+    def context(self) -> TelemetryContext: ...
+    @property
+    def payload(self) -> Mapping[str, Any]: ...
+    @property
+    def case(self) -> TelemetryCase: ...
+    @property
+    def signal_type(self) -> SignalType: ...
+    @property
+    def ce_eligible(self) -> bool: ...
+    @property
+    def trace_task_name(self) -> TraceTaskName | None: ...
+
+
+# ---------------------------------------------------------------------------
+# Concrete event classes
+# ---------------------------------------------------------------------------
+
+
 @dataclass(frozen=True)
-class TelemetryEvent:
-    name: TraceTaskName
+class DraftNodeExecutionTraceEvent:
     context: TelemetryContext
-    payload: dict[str, Any]
+    payload: NodeExecutionPayload
+    case: TelemetryCase = TelemetryCase.DRAFT_NODE_EXECUTION
+    signal_type: SignalType = SignalType.TRACE
+    ce_eligible: bool = False
+    trace_task_name: TraceTaskName | None = TraceTaskName.DRAFT_NODE_EXECUTION_TRACE
+
+
+@dataclass(frozen=True)
+class PromptGenerationEvent:
+    context: TelemetryContext
+    payload: PromptGenerationPayload
+    case: TelemetryCase = TelemetryCase.PROMPT_GENERATION
+    signal_type: SignalType = SignalType.TRACE
+    ce_eligible: bool = False
+    trace_task_name: TraceTaskName | None = TraceTaskName.PROMPT_GENERATION_TRACE
+
+
+@dataclass(frozen=True)
+class AppCreatedEvent:
+    context: TelemetryContext
+    payload: AppCreatedPayload
+    case: TelemetryCase = TelemetryCase.APP_CREATED
+    signal_type: SignalType = SignalType.METRIC_LOG
+    ce_eligible: bool = False
+    trace_task_name: TraceTaskName | None = None
+
+
+@dataclass(frozen=True)
+class AppUpdatedEvent:
+    context: TelemetryContext
+    payload: AppUpdatedPayload
+    case: TelemetryCase = TelemetryCase.APP_UPDATED
+    signal_type: SignalType = SignalType.METRIC_LOG
+    ce_eligible: bool = False
+    trace_task_name: TraceTaskName | None = None
+
+
+@dataclass(frozen=True)
+class AppDeletedEvent:
+    context: TelemetryContext
+    payload: AppDeletedPayload
+    case: TelemetryCase = TelemetryCase.APP_DELETED
+    signal_type: SignalType = SignalType.METRIC_LOG
+    ce_eligible: bool = False
+    trace_task_name: TraceTaskName | None = None
+
+
+@dataclass(frozen=True)
+class FeedbackCreatedEvent:
+    context: TelemetryContext
+    payload: FeedbackCreatedPayload
+    case: TelemetryCase = TelemetryCase.FEEDBACK_CREATED
+    signal_type: SignalType = SignalType.METRIC_LOG
+    ce_eligible: bool = False
+    trace_task_name: TraceTaskName | None = None

api/core/telemetry/gateway.py

@@ -1,13 +1,15 @@
 """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.
+Dispatches ``TelemetryEvent`` instances to either the CE/EE trace
+pipeline (``TraceQueueManager``) or the enterprise-only metric/log
+Celery queue.  Each event class carries its own routing metadata
+(``signal_type``, ``ce_eligible``, ``trace_task_name``), so this
+module contains no per-case mapping tables.
 
-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.
+This module lives in ``core/`` so both CE and EE share one
+``emit()`` entry point.  Enterprise-specific dispatch (Celery task,
+payload offloading) is handled here behind lazy imports that no-op
+in CE.
 """
 
 from __future__ import annotations
@@ -15,88 +17,20 @@ from __future__ import annotations
 import json
 import logging
 import uuid
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, cast
 
-from core.ops.entities.trace_entity import TraceTaskName
-from enterprise.telemetry.contracts import CaseRoute, SignalType
+from core.telemetry.events import TelemetryEvent
+from enterprise.telemetry.contracts import 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) -> Any:
-    """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
@@ -148,63 +82,41 @@ def _handle_payload_sizing(
 # ---------------------------------------------------------------------------
 
 
-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)
+def emit(event: TelemetryEvent, trace_manager: TraceQueueManager | None = None) -> None:
+    """Emit a telemetry event."""
+    if not event.ce_eligible and not is_enterprise_telemetry_enabled():
+        logger.debug("Dropping EE-only event: case=%s (EE disabled)", 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)
+    if event.signal_type == SignalType.TRACE:
+        _emit_trace(event, trace_manager)
     else:
-        _emit_metric_log(case, context, payload)
+        _emit_metric_log(event)
 
 
-def _emit_trace(
-    case: TelemetryCase,
-    context: dict[str, Any],
-    payload: dict[str, Any],
-    trace_manager: TraceQueueManager | None,
-) -> None:
+# ---------------------------------------------------------------------------
+# Internal dispatch
+# ---------------------------------------------------------------------------
+
+
+def _emit_trace(event: TelemetryEvent, 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)
+    if event.trace_task_name is None:
+        logger.warning("No trace_task_name on event: case=%s", event.case)
         return
 
-    queue_manager = trace_manager or LocalTraceQueueManager(
-        app_id=context.get("app_id"),
-        user_id=context.get("user_id"),
+    ctx = event.context
+    queue_manager: TraceQueueManager = trace_manager or LocalTraceQueueManager(
+        app_id=ctx.app_id,
+        user_id=ctx.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"))
+    queue_manager.add_trace_task(TraceTask(event.trace_task_name, user_id=ctx.user_id, **event.payload))
+    logger.debug("Enqueued trace task: case=%s, app_id=%s", event.case, ctx.app_id)
 
 
-def _emit_metric_log(
-    case: TelemetryCase,
-    context: dict[str, Any],
-    payload: dict[str, Any],
-) -> None:
+def _emit_metric_log(event: TelemetryEvent) -> None:
     """Build envelope and dispatch to enterprise Celery queue.
 
     No-ops when the enterprise telemetry task is not importable (CE mode).
@@ -212,18 +124,18 @@ def _emit_metric_log(
     try:
         from tasks.enterprise_telemetry_task import process_enterprise_telemetry
     except ImportError:
-        logger.debug("Enterprise metric/log dispatch unavailable, dropping: case=%s", case)
+        logger.debug("Enterprise metric/log dispatch unavailable, dropping: case=%s", event.case)
         return
 
-    tenant_id = context.get("tenant_id") or ""
+    tenant_id = event.context.tenant_id or ""
     event_id = str(uuid.uuid4())
 
-    payload_for_envelope, payload_ref = _handle_payload_sizing(payload, tenant_id, event_id)
+    payload_for_envelope, payload_ref = _handle_payload_sizing(cast(dict[str, Any], event.payload), tenant_id, event_id)
 
     from enterprise.telemetry.contracts import TelemetryEnvelope
 
     envelope = TelemetryEnvelope(
-        case=case,
+        case=event.case,
         tenant_id=tenant_id,
         event_id=event_id,
         payload=payload_for_envelope,
@@ -233,7 +145,7 @@ def _emit_metric_log(
     process_enterprise_telemetry.delay(envelope.model_dump_json())
     logger.debug(
         "Enqueued metric/log event: case=%s, tenant_id=%s, event_id=%s",
-        case,
+        event.case,
         tenant_id,
         event_id,
     )

api/enterprise/telemetry/README.md

@@ -39,7 +39,7 @@ The Enterprise OTEL exporter is configured via environment variables.
 | `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. | `false` |
-| `ENTERPRISE_SERVICE_NAME` | Service name reported to OTEL. | `dify` |
+| `APPLICATION_NAME` | Service name reported to OTEL (shared with native OTel). | `langgenius/dify` |
 | `ENTERPRISE_OTEL_SAMPLING_RATE` | Sampling rate for traces (0.0 to 1.0). Metrics are always 100%. | `1.0` |
 
 ## Correlation Model

api/enterprise/telemetry/contracts.py

@@ -32,7 +32,18 @@ class TelemetryCase(StrEnum):
 
 
 class SignalType(StrEnum):
-    """Signal routing type for telemetry cases."""
+    """Signal routing type for telemetry cases.
+
+    TRACE:
+        The trace task builds a typed ``TraceInfo``, then dispatches it
+        to both third-party trace providers (Langfuse, LangSmith, …) and
+        ``EnterpriseOtelTrace``.
+
+    METRIC_LOG:
+        Used for lightweight signal-driven
+        counters (app created/deleted, feedback) that don't need the
+        full ``TraceInfo`` data model.
+    """
 
     TRACE = "trace"
     METRIC_LOG = "metric_log"

api/enterprise/telemetry/draft_trace.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 from collections.abc import Mapping
 from typing import Any
 
-from core.telemetry import TelemetryContext, TelemetryEvent, TraceTaskName
+from core.telemetry import DraftNodeExecutionTraceEvent, TelemetryContext
 from core.telemetry import emit as telemetry_emit
 from graphon.enums import WorkflowNodeExecutionMetadataKey
 from models.workflow import WorkflowNodeExecutionModel
@@ -22,8 +22,7 @@ def enqueue_draft_node_execution_trace(
         workflow_execution_id=workflow_execution_id,
     )
     telemetry_emit(
-        TelemetryEvent(
-            name=TraceTaskName.DRAFT_NODE_EXECUTION_TRACE,
+        DraftNodeExecutionTraceEvent(
             context=TelemetryContext(
                 tenant_id=execution.tenant_id,
                 user_id=user_id,

api/enterprise/telemetry/enterprise_trace.py

@@ -289,6 +289,117 @@ class EnterpriseOtelTrace:
                 ),
             )
 
+        # -- Emit child node execution spans from DB records --
+        self._emit_node_executions_for_workflow(info, trace_correlation_override=trace_correlation_override)
+
+    def _emit_node_executions_for_workflow(
+        self,
+        workflow_info: WorkflowTraceInfo,
+        *,
+        trace_correlation_override: str | None = None,
+    ) -> None:
+        """Query node executions from the DB and emit child spans under the workflow run."""
+        from collections.abc import Mapping as MappingABC
+
+        from sqlalchemy.orm import sessionmaker
+
+        from extensions.ext_database import db
+        from graphon.enums import WorkflowNodeExecutionMetadataKey
+        from repositories.factory import DifyAPIRepositoryFactory
+
+        metadata = self._metadata(workflow_info)
+        tenant_id, app_id, user_id = self._context_ids(workflow_info, metadata)
+
+        try:
+            repository = DifyAPIRepositoryFactory.create_api_workflow_node_execution_repository(
+                sessionmaker(bind=db.engine, expire_on_commit=False)
+            )
+            rows = repository.get_executions_by_workflow_run(
+                tenant_id=tenant_id or "",
+                app_id=app_id or "",
+                workflow_run_id=workflow_info.workflow_run_id,
+            )
+        except Exception:
+            logger.exception("Failed to query node executions for workflow_run_id=%s", workflow_info.workflow_run_id)
+            return
+
+        for row in rows:
+            exec_metadata = row.execution_metadata_dict
+            process_data = row.process_data_dict or {}
+
+            # Extract token breakdown from outputs.usage (set by LLM node)
+            usage: MappingABC[str, Any] = {}
+            outputs = row.outputs_dict
+            if isinstance(outputs, MappingABC):
+                raw_usage = outputs.get("usage")
+                if isinstance(raw_usage, MappingABC):
+                    usage = raw_usage
+
+            tool_info = exec_metadata.get(WorkflowNodeExecutionMetadataKey.TOOL_INFO)
+            tool_name = tool_info.get("tool_name") if isinstance(tool_info, dict) else None
+
+            node_trace_metadata: dict[str, Any] = {
+                "tenant_id": tenant_id,
+                "app_id": app_id,
+                "app_name": metadata.get("app_name"),
+                "workspace_name": metadata.get("workspace_name"),
+                "user_id": user_id,
+                "invoke_from": metadata.get("triggered_from"),
+                "conversation_id": metadata.get("conversation_id"),
+            }
+            # NOTE: Do NOT propagate parent_trace_context to node infos.
+            # Cross-trace linking is handled at the workflow-run span level only;
+            # node spans are children of their own workflow run span.
+
+            node_info = WorkflowNodeTraceInfo(
+                trace_id=workflow_info.trace_id,
+                message_id=workflow_info.message_id,
+                start_time=row.created_at,
+                end_time=row.finished_at,
+                metadata=node_trace_metadata,
+                workflow_id=workflow_info.workflow_id,
+                workflow_run_id=workflow_info.workflow_run_id,
+                tenant_id=tenant_id or "",
+                node_execution_id=row.node_execution_id or row.id,
+                node_id=row.node_id,
+                node_type=row.node_type,
+                title=row.title,
+                status=row.status,
+                error=row.error,
+                elapsed_time=row.elapsed_time,
+                index=row.index,
+                predecessor_node_id=row.predecessor_node_id,
+                total_tokens=int(exec_metadata.get(WorkflowNodeExecutionMetadataKey.TOTAL_TOKENS, 0)),
+                total_price=float(exec_metadata.get(WorkflowNodeExecutionMetadataKey.TOTAL_PRICE, 0.0)),
+                currency=exec_metadata.get(WorkflowNodeExecutionMetadataKey.CURRENCY),
+                model_provider=process_data.get("model_provider"),
+                model_name=process_data.get("model_name"),
+                prompt_tokens=usage.get("prompt_tokens"),
+                completion_tokens=usage.get("completion_tokens"),
+                tool_name=tool_name,
+                iteration_id=exec_metadata.get(WorkflowNodeExecutionMetadataKey.ITERATION_ID),
+                iteration_index=exec_metadata.get(WorkflowNodeExecutionMetadataKey.ITERATION_INDEX),
+                loop_id=exec_metadata.get(WorkflowNodeExecutionMetadataKey.LOOP_ID),
+                loop_index=exec_metadata.get(WorkflowNodeExecutionMetadataKey.LOOP_INDEX),
+                parallel_id=exec_metadata.get(WorkflowNodeExecutionMetadataKey.PARALLEL_ID),
+                node_inputs=row.inputs_dict,
+                node_outputs=outputs,
+                process_data=process_data,
+                invoked_by=workflow_info.invoked_by,
+            )
+            try:
+                self._emit_node_execution_trace(
+                    node_info,
+                    EnterpriseTelemetrySpan.NODE_EXECUTION,
+                    "node",
+                    trace_correlation_override_param=trace_correlation_override,
+                )
+            except Exception:
+                logger.exception(
+                    "Failed to emit node execution trace: node_execution_id=%s",
+                    node_info.node_execution_id,
+                )
+
     def _node_execution_trace(self, info: WorkflowNodeTraceInfo) -> None:
         self._emit_node_execution_trace(info, EnterpriseTelemetrySpan.NODE_EXECUTION, "node")
 
@@ -345,6 +456,9 @@ class EnterpriseOtelTrace:
         trace_correlation_override = trace_correlation_override_param or resolved_override
 
         effective_correlation_id = correlation_id_override or info.workflow_run_id
+        # Explicitly set parent to the workflow run span so node executions
+        # are always children of their workflow run in the trace tree.
+        parent_span_id_source = info.workflow_run_id if not correlation_id_override else None
         self._exporter.export_span(
             span_name,
             span_attrs,
@@ -353,6 +467,7 @@ class EnterpriseOtelTrace:
             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) --
@@ -953,7 +1068,7 @@ class EnterpriseOtelTrace:
         self._exporter.record_histogram(
             EnterpriseTelemetryHistogram.PROMPT_GENERATION_DURATION,
             info.latency,
-            labels,
+            self._labels(**labels, status=prompt_status),
         )
 
         if info.error:

api/enterprise/telemetry/event_handlers.py

@@ -2,7 +2,7 @@
 
 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``
+ensure handlers fire.  Each handler delegates to ``core.telemetry``
 which handles routing, EE-gating, and dispatch.
 
 All handlers are best-effort: exceptions are caught and logged so that
@@ -24,19 +24,28 @@ __all__ = [
 ]
 
 
+def _optional_str(value: object) -> str | None:
+    """Coerce a value to ``str | None`` for telemetry payloads."""
+    if value is None:
+        return None
+    return str(value)
+
+
 @app_was_created.connect
 def _handle_app_created(sender: object, **kwargs: object) -> None:
     try:
-        from core.telemetry.gateway import emit as gateway_emit
-        from enterprise.telemetry.contracts import TelemetryCase
+        from core.telemetry import AppCreatedEvent, TelemetryContext, emit
 
-        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),
-            },
+        emit(
+            AppCreatedEvent(
+                context=TelemetryContext(
+                    tenant_id=str(getattr(sender, "tenant_id", "") or ""),
+                ),
+                payload={
+                    "app_id": _optional_str(getattr(sender, "id", None)),
+                    "mode": _optional_str(getattr(sender, "mode", None)),
+                },
+            )
         )
     except Exception:
         logger.warning("Failed to emit app_created telemetry", exc_info=True)
@@ -45,13 +54,15 @@ def _handle_app_created(sender: object, **kwargs: object) -> None:
 @app_was_updated.connect
 def _handle_app_updated(sender: object, **kwargs: object) -> None:
     try:
-        from core.telemetry.gateway import emit as gateway_emit
-        from enterprise.telemetry.contracts import TelemetryCase
+        from core.telemetry import AppUpdatedEvent, TelemetryContext, emit
 
-        gateway_emit(
-            case=TelemetryCase.APP_UPDATED,
-            context={"tenant_id": str(getattr(sender, "tenant_id", "") or "")},
-            payload={"app_id": getattr(sender, "id", None)},
+        emit(
+            AppUpdatedEvent(
+                context=TelemetryContext(
+                    tenant_id=str(getattr(sender, "tenant_id", "") or ""),
+                ),
+                payload={"app_id": _optional_str(getattr(sender, "id", None))},
+            )
         )
     except Exception:
         logger.warning("Failed to emit app_updated telemetry", exc_info=True)
@@ -60,13 +71,15 @@ def _handle_app_updated(sender: object, **kwargs: object) -> None:
 @app_was_deleted.connect
 def _handle_app_deleted(sender: object, **kwargs: object) -> None:
     try:
-        from core.telemetry.gateway import emit as gateway_emit
-        from enterprise.telemetry.contracts import TelemetryCase
+        from core.telemetry import AppDeletedEvent, TelemetryContext, emit
 
-        gateway_emit(
-            case=TelemetryCase.APP_DELETED,
-            context={"tenant_id": str(getattr(sender, "tenant_id", "") or "")},
-            payload={"app_id": getattr(sender, "id", None)},
+        emit(
+            AppDeletedEvent(
+                context=TelemetryContext(
+                    tenant_id=str(getattr(sender, "tenant_id", "") or ""),
+                ),
+                payload={"app_id": _optional_str(getattr(sender, "id", None))},
+            )
         )
     except Exception:
         logger.warning("Failed to emit app_deleted telemetry", exc_info=True)

api/enterprise/telemetry/exporter.py

@@ -34,7 +34,7 @@ from opentelemetry.semconv.attributes import service_attributes
 from opentelemetry.trace import SpanContext, TraceFlags
 from opentelemetry.util.types import Attributes, AttributeValue
 
-from configs import dify_config
+from configs import DifyConfig, dify_config
 from enterprise.telemetry.entities import EnterpriseTelemetryCounter, EnterpriseTelemetryHistogram
 from enterprise.telemetry.id_generator import (
     CorrelationIdGenerator,
@@ -106,14 +106,14 @@ class EnterpriseExporter:
     ``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", "")
+    def __init__(self, config: DifyConfig) -> None:
+        endpoint: str = config.ENTERPRISE_OTLP_ENDPOINT
+        headers_raw: str = config.ENTERPRISE_OTLP_HEADERS
+        protocol: str = config.ENTERPRISE_OTLP_PROTOCOL
+        service_name: str = config.APPLICATION_NAME
+        sampling_rate: float = config.ENTERPRISE_OTEL_SAMPLING_RATE
+        self.include_content: bool = config.ENTERPRISE_INCLUDE_CONTENT
+        api_key: str = config.ENTERPRISE_OTLP_API_KEY
 
         # Auto-detect TLS: https:// uses secure, everything else is insecure
         insecure = not endpoint.startswith("https://")
@@ -206,7 +206,12 @@ class EnterpriseExporter:
         set_span_id_source(span_id_source)
 
         try:
-            parent_context: Context | None = None
+            # Use an explicit empty Context (not None) as the default so root spans
+            # never implicitly inherit an ambient active span (e.g. Celery/HTTP
+            # auto-instrumentation or an external trace context). Otherwise the root
+            # span would join the ambient trace while child spans use our deterministic
+            # trace_id, splitting a single logical trace into two.
+            parent_context: Context = Context()
             # 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:

api/services/message_service.py

@@ -256,9 +256,51 @@ class MessageService:
             session.add(feedback)
 
         session.commit()
+        if rating:
+            cls._emit_feedback_telemetry(
+                app_model=app_model, message=message, user=user, rating=rating, content=content
+            )
 
         return feedback
 
+    @classmethod
+    def _emit_feedback_telemetry(
+        cls,
+        *,
+        app_model: App,
+        message: Message,
+        user: Account | EndUser,
+        rating: FeedbackRating | None,
+        content: str | None,
+    ) -> None:
+        try:
+            from core.telemetry import FeedbackCreatedEvent, TelemetryContext, emit
+
+            if message.id is None:
+                return
+
+            emit(
+                FeedbackCreatedEvent(
+                    context=TelemetryContext(tenant_id=app_model.tenant_id),
+                    payload={
+                        "message_id": str(message.id),
+                        "app_id": str(app_model.id) if app_model.id is not None else None,
+                        "conversation_id": (
+                            str(message.conversation_id) if message.conversation_id is not None else None
+                        ),
+                        "from_end_user_id": str(user.id) if isinstance(user, EndUser) and user.id is not None else None,
+                        "from_account_id": str(user.id) if isinstance(user, Account) and user.id is not None else None,
+                        "rating": rating.value if rating else None,
+                        "from_source": (
+                            FeedbackFromSource.USER if isinstance(user, EndUser) else FeedbackFromSource.ADMIN
+                        ).value,
+                        "content": content,
+                    },
+                )
+            )
+        except Exception:
+            logger.warning("Failed to emit feedback_created telemetry", exc_info=True)
+
     @classmethod
     def get_all_messages_feedbacks(cls, app_model: App, page: int, limit: int, *, session: Session):
         """Get all feedbacks of an app"""

api/tests/unit_tests/core/app/apps/advanced_chat/test_generate_task_pipeline.py

@@ -157,6 +157,8 @@ def test_resume_appends_chunks_to_paused_answer() -> None:
         invoke_from=InvokeFrom.WEB_APP,
         inputs={},
         task_id="task-1",
+        trace_manager=None,
+        extras={},
     )
     queue_manager = SimpleNamespace(graph_runtime_state=None)
     conversation = pipeline_module.ConversationSnapshot(id="conversation-1", mode=AppMode.ADVANCED_CHAT)
@@ -213,6 +215,220 @@ def test_resume_appends_chunks_to_paused_answer() -> None:
     assert stored_message.status == MessageStatus.NORMAL
 
 
+def test_emit_message_trace_enqueues_message_trace() -> None:
+    """When trace_manager is provided, _emit_message_trace must enqueue a MESSAGE_TRACE task."""
+    from core.ops.entities.trace_entity import TraceTaskName
+
+    mock_trace_manager = mock.MagicMock()
+    app_config = SimpleNamespace(app_id="app-1", tenant_id="tenant-1", sensitive_word_avoidance=None)
+    application_generate_entity = SimpleNamespace(
+        app_config=app_config,
+        files=[],
+        workflow_run_id="run-1",
+        query="hello",
+        invoke_from=InvokeFrom.WEB_APP,
+        inputs={},
+        task_id="task-1",
+        trace_manager=mock_trace_manager,
+        extras={"trace_session_id": "session-trace-1"},
+    )
+    queue_manager = SimpleNamespace(graph_runtime_state=None)
+    conversation = pipeline_module.ConversationSnapshot(id="conv-1", mode=AppMode.ADVANCED_CHAT)
+    message = pipeline_module.MessageSnapshot(
+        id="msg-1",
+        created_at=datetime(2024, 1, 1),
+        query="hello",
+        answer="",
+        status=MessageStatus.NORMAL,
+    )
+    user = EndUser()
+    user.id = "user-1"
+    user.session_id = "session-1"
+    workflow = pipeline_module.WorkflowSnapshot(id="workflow-1", tenant_id="tenant-1", features_dict={})
+
+    pipeline = pipeline_module.AdvancedChatAppGenerateTaskPipeline(
+        application_generate_entity=application_generate_entity,
+        workflow=workflow,
+        queue_manager=queue_manager,
+        conversation=conversation,
+        message=message,
+        user=user,
+        stream=True,
+        dialogue_count=1,
+        draft_var_saver_factory=SimpleNamespace(),
+    )
+
+    stored_message = SimpleNamespace(
+        id="msg-1",
+        answer="",
+        status=MessageStatus.NORMAL,
+        updated_at=None,
+        provider_response_latency=0,
+        message_tokens=0,
+        message_unit_price=0,
+        message_price_unit=0,
+        answer_tokens=0,
+        answer_unit_price=0,
+        answer_price_unit=0,
+        total_price=0,
+        currency="USD",
+        message_metadata=None,
+        invoke_from=InvokeFrom.WEB_APP,
+        from_account_id=None,
+        from_end_user_id="user-1",
+    )
+    pipeline._get_message = mock.Mock(return_value=stored_message)
+    pipeline._recorded_files = []
+
+    pipeline._emit_message_trace()
+
+    mock_trace_manager.add_trace_task.assert_called_once()
+    trace_task = mock_trace_manager.add_trace_task.call_args[0][0]
+    assert trace_task.trace_type == TraceTaskName.MESSAGE_TRACE
+    assert trace_task.conversation_id == "conv-1"
+    assert trace_task.message_id == "msg-1"
+    assert trace_task.kwargs.get("trace_session_id") == "session-trace-1"
+
+
+def test_save_message_does_not_emit_trace() -> None:
+    """_save_message must not enqueue a trace; emitting is the terminal caller's responsibility.
+
+    This keeps intermediate saves (e.g. on pause) from duplicating message, token, and TTFT metrics
+    when the message is later resumed.
+    """
+    mock_trace_manager = mock.MagicMock()
+    app_config = SimpleNamespace(app_id="app-1", tenant_id="tenant-1", sensitive_word_avoidance=None)
+    application_generate_entity = SimpleNamespace(
+        app_config=app_config,
+        files=[],
+        workflow_run_id="run-1",
+        query="hello",
+        invoke_from=InvokeFrom.WEB_APP,
+        inputs={},
+        task_id="task-1",
+        trace_manager=mock_trace_manager,
+        extras={"trace_session_id": "session-trace-1"},
+    )
+    queue_manager = SimpleNamespace(graph_runtime_state=None)
+    conversation = pipeline_module.ConversationSnapshot(id="conv-1", mode=AppMode.ADVANCED_CHAT)
+    message = pipeline_module.MessageSnapshot(
+        id="msg-1",
+        created_at=datetime(2024, 1, 1),
+        query="hello",
+        answer="",
+        status=MessageStatus.NORMAL,
+    )
+    user = EndUser()
+    user.id = "user-1"
+    user.session_id = "session-1"
+    workflow = pipeline_module.WorkflowSnapshot(id="workflow-1", tenant_id="tenant-1", features_dict={})
+
+    pipeline = pipeline_module.AdvancedChatAppGenerateTaskPipeline(
+        application_generate_entity=application_generate_entity,
+        workflow=workflow,
+        queue_manager=queue_manager,
+        conversation=conversation,
+        message=message,
+        user=user,
+        stream=True,
+        dialogue_count=1,
+        draft_var_saver_factory=SimpleNamespace(),
+    )
+
+    stored_message = SimpleNamespace(
+        id="msg-1",
+        answer="",
+        status=MessageStatus.NORMAL,
+        updated_at=None,
+        provider_response_latency=0,
+        message_tokens=0,
+        message_unit_price=0,
+        message_price_unit=0,
+        answer_tokens=0,
+        answer_unit_price=0,
+        answer_price_unit=0,
+        total_price=0,
+        currency="USD",
+        message_metadata=None,
+        invoke_from=InvokeFrom.WEB_APP,
+        from_account_id=None,
+        from_end_user_id="user-1",
+    )
+    pipeline._get_message = mock.Mock(return_value=stored_message)
+    pipeline._recorded_files = []
+
+    pipeline._save_message(session=mock.Mock())
+
+    mock_trace_manager.add_trace_task.assert_not_called()
+
+
+def test_emit_message_trace_skips_when_no_trace_manager() -> None:
+    """When trace_manager is None, _emit_message_trace and _save_message must not fail."""
+    app_config = SimpleNamespace(app_id="app-1", tenant_id="tenant-1", sensitive_word_avoidance=None)
+    application_generate_entity = SimpleNamespace(
+        app_config=app_config,
+        files=[],
+        workflow_run_id="run-1",
+        query="hello",
+        invoke_from=InvokeFrom.WEB_APP,
+        inputs={},
+        task_id="task-1",
+        trace_manager=None,
+        extras={},
+    )
+    queue_manager = SimpleNamespace(graph_runtime_state=None)
+    conversation = pipeline_module.ConversationSnapshot(id="conv-1", mode=AppMode.ADVANCED_CHAT)
+    message = pipeline_module.MessageSnapshot(
+        id="msg-1",
+        created_at=datetime(2024, 1, 1),
+        query="hello",
+        answer="",
+        status=MessageStatus.NORMAL,
+    )
+    user = EndUser()
+    user.id = "user-1"
+    user.session_id = "session-1"
+    workflow = pipeline_module.WorkflowSnapshot(id="workflow-1", tenant_id="tenant-1", features_dict={})
+
+    pipeline = pipeline_module.AdvancedChatAppGenerateTaskPipeline(
+        application_generate_entity=application_generate_entity,
+        workflow=workflow,
+        queue_manager=queue_manager,
+        conversation=conversation,
+        message=message,
+        user=user,
+        stream=True,
+        dialogue_count=1,
+        draft_var_saver_factory=SimpleNamespace(),
+    )
+
+    stored_message = SimpleNamespace(
+        id="msg-1",
+        answer="",
+        status=MessageStatus.NORMAL,
+        updated_at=None,
+        provider_response_latency=0,
+        message_tokens=0,
+        message_unit_price=0,
+        message_price_unit=0,
+        answer_tokens=0,
+        answer_unit_price=0,
+        answer_price_unit=0,
+        total_price=0,
+        currency="USD",
+        message_metadata=None,
+        invoke_from=InvokeFrom.WEB_APP,
+        from_account_id=None,
+        from_end_user_id="user-1",
+    )
+    pipeline._get_message = mock.Mock(return_value=stored_message)
+    pipeline._recorded_files = []
+
+    # Should not raise
+    pipeline._save_message(session=mock.Mock())
+    pipeline._emit_message_trace()
+
+
 def test_workflow_succeeded_emits_message_end_before_workflow_finished() -> None:
     pipeline = _build_pipeline()
     pipeline._application_generate_entity = SimpleNamespace(task_id="task-1")

api/tests/unit_tests/core/llm_generator/test_llm_generator.py

@@ -464,7 +464,6 @@ class TestLLMGenerator:
         mock_model_instance.invoke_llm.side_effect = Exception("Unexpected multi-step error")
 
         result = LLMGenerator.generate_rule_config("tenant_id", payload)
-        assert "Failed to handle unexpected exception" in result["error"]
         assert "Unexpected multi-step error" in result["error"]
 
     def test_generate_code_python_success(self, mock_model_instance, model_config_entity):
@@ -503,7 +502,7 @@ class TestLLMGenerator:
         mock_model_instance.invoke_llm.side_effect = Exception("Random error")
 
         result = LLMGenerator.generate_code("tenant_id", payload)
-        assert "An unexpected error occurred" in result["error"]
+        assert "Random error" in result["error"]
 
     def test_generate_qa_document_success(self, mock_model_instance):
         mock_response = MagicMock(spec=LLMResult)
@@ -548,7 +547,6 @@ class TestLLMGenerator:
         mock_model_instance.invoke_llm.return_value = mock_response
 
         result = LLMGenerator.generate_structured_output("tenant_id", payload)
-        assert "An unexpected error occurred" in result["error"]
         assert "Failed to parse structured output" in result["error"]
 
     def test_generate_structured_output_invoke_error(self, mock_model_instance, model_config_entity):
@@ -556,14 +554,14 @@ class TestLLMGenerator:
         mock_model_instance.invoke_llm.side_effect = InvokeError("Invoke failed")
 
         result = LLMGenerator.generate_structured_output("tenant_id", payload)
-        assert "Failed to generate JSON Schema" in result["error"]
+        assert "Invoke failed" in result["error"]
 
     def test_generate_structured_output_exception(self, mock_model_instance, model_config_entity):
         payload = RuleStructuredOutputPayload(instruction="error", model_config=model_config_entity)
         mock_model_instance.invoke_llm.side_effect = Exception("Random error")
 
         result = LLMGenerator.generate_structured_output("tenant_id", payload)
-        assert "An unexpected error occurred" in result["error"]
+        assert "Random error" in result["error"]
 
     def test_instruction_modify_legacy_without_last_run_uses_real_empty_query(
         self,
@@ -835,12 +833,11 @@ class TestLLMGenerator:
             app.tenant_id, app.id, "current", "instruction", model_config_entity, "ideal"
         )
 
-        assert "An unexpected error occurred" in result["error"]
         assert error_fragment in result["error"]
 
     @pytest.mark.parametrize(
         ("model_error", "error_fragment"),
-        [(InvokeError("invoke failed"), "Failed to generate code"), (RuntimeError("boom"), "unexpected error")],
+        [(InvokeError("invoke failed"), "invoke failed"), (RuntimeError("boom"), "boom")],
     )
     def test_instruction_modify_handles_model_errors(
         self,
@@ -858,3 +855,75 @@ class TestLLMGenerator:
         )
 
         assert error_fragment.lower() in result["error"].lower()
+
+
+# ---------------------------------------------------------------------------
+# Prompt-generation telemetry emit-site tests
+# ---------------------------------------------------------------------------
+
+
+class TestPromptGenerationTelemetryEmit:
+    """Verify that LLM generator methods call telemetry_emit with a
+    PromptGenerationEvent so the metric pipeline is not accidentally broken."""
+
+    @pytest.fixture(autouse=True)
+    def _setup(self, recording_model_instance: Mock) -> None:
+        self.model_instance = recording_model_instance
+
+    @patch("core.llm_generator.llm_generator.telemetry_emit")
+    def test_generate_code_emits_on_success(self, mock_emit: MagicMock) -> None:
+        self.model_instance.invoke_llm.return_value = _llm_result("print('hello')")
+        model_cfg = ModelConfig(provider="openai", name="gpt-4", mode="chat", completion_params={})
+        args = RuleCodeGeneratePayload(instruction="write hello world", model_config=model_cfg, code_language="python")
+
+        LLMGenerator.generate_code(tenant_id="t-1", args=args, app_id="app-1")
+
+        mock_emit.assert_called_once()
+        event = mock_emit.call_args[0][0]
+        assert type(event).__name__ == "PromptGenerationEvent"
+        assert event.payload["operation_type"] == "code_generate"
+        assert event.payload["model_provider"] == "openai"
+        assert event.payload["model_name"] == "gpt-4"
+        assert event.context.tenant_id == "t-1"
+        assert event.context.app_id == "app-1"
+        assert event.payload["error"] is None
+
+    @patch("core.llm_generator.llm_generator.telemetry_emit")
+    def test_generate_code_emits_on_failure(self, mock_emit: MagicMock) -> None:
+        self.model_instance.invoke_llm.side_effect = InvokeError("model down")
+        model_cfg = ModelConfig(provider="openai", name="gpt-4", mode="chat", completion_params={})
+        args = RuleCodeGeneratePayload(instruction="write hello world", model_config=model_cfg, code_language="python")
+
+        result = LLMGenerator.generate_code(tenant_id="t-1", args=args)
+
+        mock_emit.assert_called_once()
+        event = mock_emit.call_args[0][0]
+        assert event.payload["error"] == "model down"
+        assert "Failed to generate code" in result["error"]
+
+    @patch("core.llm_generator.llm_generator.telemetry_emit")
+    def test_generate_rule_config_no_variable_emits(self, mock_emit: MagicMock) -> None:
+        self.model_instance.invoke_llm.return_value = _llm_result("generated prompt")
+        model_cfg = ModelConfig(provider="anthropic", name="claude-3", mode="chat", completion_params={})
+        args = RuleGeneratePayload(instruction="be helpful", model_config=model_cfg, no_variable=True)
+
+        LLMGenerator.generate_rule_config(tenant_id="t-1", args=args, app_id="app-2")
+
+        mock_emit.assert_called_once()
+        event = mock_emit.call_args[0][0]
+        assert event.payload["operation_type"] == "rule_generate"
+        assert event.payload["model_provider"] == "anthropic"
+        assert event.context.app_id == "app-2"
+
+    @patch("core.llm_generator.llm_generator.telemetry_emit")
+    def test_generate_rule_config_no_variable_emits_on_failure(self, mock_emit: MagicMock) -> None:
+        self.model_instance.invoke_llm.side_effect = InvokeError("auth fail")
+        model_cfg = ModelConfig(provider="anthropic", name="claude-3", mode="chat", completion_params={})
+        args = RuleGeneratePayload(instruction="be helpful", model_config=model_cfg, no_variable=True)
+
+        result = LLMGenerator.generate_rule_config(tenant_id="t-1", args=args)
+
+        mock_emit.assert_called_once()
+        event = mock_emit.call_args[0][0]
+        assert event.payload["error"] == "auth fail"
+        assert "error" in result

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

@@ -10,7 +10,7 @@ from unittest.mock import MagicMock, patch
 import pytest
 
 from core.ops.entities.trace_entity import TraceTaskName
-from core.telemetry.events import TelemetryContext, TelemetryEvent
+from core.telemetry.events import DraftNodeExecutionTraceEvent, TelemetryContext
 
 
 @pytest.fixture
@@ -57,14 +57,13 @@ class TestTelemetryEmit:
     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,
+        event = DraftNodeExecutionTraceEvent(
             context=TelemetryContext(
                 tenant_id="test-tenant",
                 user_id="test-user",
                 app_id="test-app",
             ),
-            payload={"key": "value"},
+            payload={"node_execution_data": {"key": "value"}},
         )
 
         emit_fn(event)
@@ -73,81 +72,33 @@ class TestTelemetryEmit:
         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,
+        event = DraftNodeExecutionTraceEvent(
             context=TelemetryContext(
                 tenant_id="test-tenant",
                 user_id="test-user",
                 app_id="test-app",
             ),
-            payload={},
+            payload={"node_execution_data": {}},
         )
 
         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,
+        event = DraftNodeExecutionTraceEvent(
             context=TelemetryContext(
                 tenant_id="test-tenant",
                 user_id="test-user",
                 app_id="test-app",
             ),
-            payload={"extra": "data"},
+            payload={"node_execution_data": {"extra": "data"}},
         )
 
         emit_fn(event)
@@ -164,18 +115,17 @@ class TestTelemetryEmit:
         mock_trace_manager = MagicMock()
         mock_trace_manager.add_trace_task = MagicMock()
 
-        event = TelemetryEvent(
-            name=TraceTaskName.NODE_EXECUTION_TRACE,
+        event = DraftNodeExecutionTraceEvent(
             context=TelemetryContext(
                 tenant_id="test-tenant",
                 user_id="test-user",
                 app_id="test-app",
             ),
-            payload={},
+            payload={"node_execution_data": {}},
         )
 
         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
+        assert called_task.trace_type == TraceTaskName.DRAFT_NODE_EXECUTION_TRACE

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

@@ -5,6 +5,15 @@ from unittest.mock import MagicMock, patch
 
 import pytest
 
+from core.telemetry.events import (
+    AppCreatedEvent,
+    AppCreatedPayload,
+    DraftNodeExecutionTraceEvent,
+    NodeExecutionPayload,
+    PromptGenerationEvent,
+    PromptGenerationPayload,
+    TelemetryContext,
+)
 from core.telemetry.gateway import emit, is_enterprise_telemetry_enabled
 from enterprise.telemetry.contracts import TelemetryCase
 
@@ -16,24 +25,48 @@ class TestTelemetryCoreExports:
         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()
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
 
-    mock_trace_entity = MagicMock()
-    mock_trace_task_name = MagicMock()
-    mock_trace_task_name.return_value = "workflow"
-    mock_trace_entity.TraceTaskName = mock_trace_task_name
+_TRACE_CTX = TelemetryContext(tenant_id="tenant-789", user_id="user-456", app_id="app-123")
 
-    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
+
+def _node_event() -> DraftNodeExecutionTraceEvent:
+    return DraftNodeExecutionTraceEvent(
+        context=_TRACE_CTX,
+        payload=NodeExecutionPayload(node_execution_data={}),
+    )
+
+
+def _prompt_event() -> PromptGenerationEvent:
+    return PromptGenerationEvent(
+        context=_TRACE_CTX,
+        payload=PromptGenerationPayload(
+            tenant_id="tenant-789",
+            operation_type="generate",
+            instruction="test",
+            generated_output="out",
+            model_provider="openai",
+            model_name="gpt-4",
+            prompt_tokens=10,
+            completion_tokens=5,
+            total_tokens=15,
+            latency=0.5,
+        ),
+    )
+
+
+def _app_created_event() -> AppCreatedEvent:
+    return AppCreatedEvent(
+        context=TelemetryContext(tenant_id="tenant-123"),
+        payload=AppCreatedPayload(app_id="app-abc"),
+    )
+
+
+# ---------------------------------------------------------------------------
+# Trace routing integration
+# ---------------------------------------------------------------------------
 
 
 class TestGatewayIntegrationTraceRouting:
@@ -41,72 +74,46 @@ class TestGatewayIntegrationTraceRouting:
     def mock_trace_manager(self) -> MagicMock:
         return MagicMock()
 
-    @pytest.mark.usefixtures("mock_ops_trace_manager")
-    def test_ce_eligible_trace_routed_to_trace_manager(
+    def test_trace_event_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)
-
+            emit(_prompt_event(), 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)
-
+            emit(_node_event(), 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)
-
+            emit(_node_event(), mock_trace_manager)
             mock_trace_manager.add_trace_task.assert_called_once()
 
 
+# ---------------------------------------------------------------------------
+# Metric/log routing integration
+# ---------------------------------------------------------------------------
+
+
 class TestGatewayIntegrationMetricRouting:
     @patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=True)
-    def test_metric_case_routes_to_celery_task(
+    def test_metric_event_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)
+            emit(_app_created_event())
 
             mock_delay.assert_called_once()
             envelope_json = mock_delay.call_args[0][0]
@@ -115,33 +122,10 @@ class TestGatewayIntegrationMetricRouting:
             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()
+# ---------------------------------------------------------------------------
+# CE eligibility integration
+# ---------------------------------------------------------------------------
 
 
 class TestGatewayIntegrationCEEligibility:
@@ -149,72 +133,28 @@ class TestGatewayIntegrationCEEligibility:
     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)
-
+            emit(_node_event(), 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)
-
+            emit(_prompt_event(), mock_trace_manager)
             mock_trace_manager.add_trace_task.assert_not_called()
 
 
+# ---------------------------------------------------------------------------
+# is_enterprise_telemetry_enabled
+# ---------------------------------------------------------------------------
+
+
 class TestIsEnterpriseTelemetryEnabled:
     def test_returns_false_when_exporter_import_fails(self) -> None:
         with patch.dict(sys.modules, {"enterprise.telemetry.exporter": None}):

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

@@ -5,7 +5,6 @@ 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
 
 
@@ -153,78 +152,26 @@ class TestTelemetryEnvelope:
         assert envelope.metadata is None
 
 
-class TestCaseRouting:
-    """Tests for CASE_ROUTING table."""
+class TestEventRoutingMetadata:
+    """Verify event classes declare correct routing fields (replaces CASE_ROUTING table tests)."""
 
-    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_enterprise_only_events(self) -> None:
+        from core.telemetry.events import DraftNodeExecutionTraceEvent, PromptGenerationEvent
 
-    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
+        for cls in (DraftNodeExecutionTraceEvent, PromptGenerationEvent):
+            assert cls.signal_type is SignalType.TRACE  # type: ignore[attr-defined]
+            assert cls.ce_eligible is False  # type: ignore[attr-defined]
+            assert cls.trace_task_name is not None  # type: ignore[attr-defined]
 
-    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_events(self) -> None:
+        from core.telemetry.events import (
+            AppCreatedEvent,
+            AppDeletedEvent,
+            AppUpdatedEvent,
+            FeedbackCreatedEvent,
+        )
 
-    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
+        for cls in (AppCreatedEvent, AppUpdatedEvent, AppDeletedEvent, FeedbackCreatedEvent):
+            assert cls.signal_type is SignalType.METRIC_LOG  # type: ignore[attr-defined]
+            assert cls.ce_eligible is False  # type: ignore[attr-defined]
+            assert cls.trace_task_name is None  # type: ignore[attr-defined]

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

@@ -234,7 +234,7 @@ class TestBuildNodeExecutionData:
 class TestEnqueueDraftNodeExecutionTrace:
     @patch("enterprise.telemetry.draft_trace.telemetry_emit")
     def test_emits_telemetry_event(self, mock_emit: MagicMock) -> None:
-        from core.telemetry import TelemetryEvent, TraceTaskName
+        from core.telemetry import DraftNodeExecutionTraceEvent, TraceTaskName
         from enterprise.telemetry.draft_trace import enqueue_draft_node_execution_trace
 
         execution = _make_execution()
@@ -246,15 +246,15 @@ class TestEnqueueDraftNodeExecutionTrace:
         )
 
         mock_emit.assert_called_once()
-        event: TelemetryEvent = mock_emit.call_args[0][0]
-        assert event.name == TraceTaskName.DRAFT_NODE_EXECUTION_TRACE
+        event: DraftNodeExecutionTraceEvent = mock_emit.call_args[0][0]
+        assert event.trace_task_name == TraceTaskName.DRAFT_NODE_EXECUTION_TRACE
         assert event.context.tenant_id == "tenant-1"
         assert event.context.user_id == "user-1"
         assert event.context.app_id == "app-1"
 
     @patch("enterprise.telemetry.draft_trace.telemetry_emit")
     def test_payload_contains_node_execution_data(self, mock_emit: MagicMock) -> None:
-        from core.telemetry import TelemetryEvent
+        from core.telemetry import DraftNodeExecutionTraceEvent
         from enterprise.telemetry.draft_trace import enqueue_draft_node_execution_trace
 
         execution = _make_execution()
@@ -265,7 +265,7 @@ class TestEnqueueDraftNodeExecutionTrace:
             user_id="user-2",
         )
 
-        event: TelemetryEvent = mock_emit.call_args[0][0]
+        event: DraftNodeExecutionTraceEvent = mock_emit.call_args[0][0]
         node_data = event.payload["node_execution_data"]
         assert node_data["workflow_id"] == "wf-1"
         assert node_data["node_type"] == "llm"
@@ -273,7 +273,7 @@ class TestEnqueueDraftNodeExecutionTrace:
 
     @patch("enterprise.telemetry.draft_trace.telemetry_emit")
     def test_outputs_forwarded_to_build(self, mock_emit: MagicMock) -> None:
-        from core.telemetry import TelemetryEvent
+        from core.telemetry import DraftNodeExecutionTraceEvent
         from enterprise.telemetry.draft_trace import enqueue_draft_node_execution_trace
 
         execution = _make_execution(outputs_dict={"default": True})
@@ -284,12 +284,12 @@ class TestEnqueueDraftNodeExecutionTrace:
             user_id="user-3",
         )
 
-        event: TelemetryEvent = mock_emit.call_args[0][0]
+        event: DraftNodeExecutionTraceEvent = mock_emit.call_args[0][0]
         assert event.payload["node_execution_data"]["node_outputs"] == {"explicit": True}
 
     @patch("enterprise.telemetry.draft_trace.telemetry_emit")
     def test_none_outputs_uses_execution_outputs(self, mock_emit: MagicMock) -> None:
-        from core.telemetry import TelemetryEvent
+        from core.telemetry import DraftNodeExecutionTraceEvent
         from enterprise.telemetry.draft_trace import enqueue_draft_node_execution_trace
 
         execution = _make_execution(outputs_dict={"from_model": "yes"})
@@ -300,7 +300,7 @@ class TestEnqueueDraftNodeExecutionTrace:
             user_id="user-4",
         )
 
-        event: TelemetryEvent = mock_emit.call_args[0][0]
+        event: DraftNodeExecutionTraceEvent = mock_emit.call_args[0][0]
         assert event.payload["node_execution_data"]["node_outputs"] == {"from_model": "yes"}
 
 

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

@@ -609,6 +609,30 @@ class TestWorkflowTrace:
         assert attrs["dify.parent.workflow.run_id"] == "outer-run-001"
         assert attrs["dify.parent.app.id"] == "outer-app-001"
 
+    def test_node_executions_read_through_repository(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
+        """Node executions must be read through the repository factory so the active storage backend
+        (e.g. LogStore/Aliyun SLS) is honoured, not via a direct DB query that only sees the RDBMS."""
+        mock_repo = MagicMock()
+        mock_repo.get_executions_by_workflow_run.return_value = []
+        with (
+            patch("enterprise.telemetry.enterprise_trace.emit_telemetry_log"),
+            # db.engine requires a Flask app context that unit tests don't set up; patch the db
+            # object so the repository factory is reachable (the mocked repo never opens a session).
+            patch("extensions.ext_database.db", MagicMock()),
+            patch(
+                "repositories.factory.DifyAPIRepositoryFactory.create_api_workflow_node_execution_repository",
+                return_value=mock_repo,
+            ) as mock_create,
+        ):
+            trace_handler._workflow_trace(make_workflow_info())
+
+        mock_create.assert_called_once()
+        mock_repo.get_executions_by_workflow_run.assert_called_once_with(
+            tenant_id="tenant-abc",
+            app_id="app-001",
+            workflow_run_id="run-001",
+        )
+
 
 # ---------------------------------------------------------------------------
 # _node_execution_trace / _emit_node_execution_trace
@@ -826,6 +850,19 @@ class TestMessageTrace:
         assert len(ttft_calls) == 1
         assert ttft_calls[0][0][1] == pytest.approx(0.42)
 
+    def test_ttft_histogram_has_required_labels(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
+        with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event"):
+            trace_handler._message_trace(make_message_info(gen_ai_server_time_to_first_token=0.42))
+
+        ttft_calls = [
+            c
+            for c in mock_exporter.record_histogram.call_args_list
+            if c[0][0] == EnterpriseTelemetryHistogram.MESSAGE_TTFT
+        ]
+        labels = ttft_calls[0][0][2]
+        for key in ("tenant_id", "app_id", "model_provider", "model_name"):
+            assert key in labels, f"Missing required TTFT label: {key}"
+
     def test_no_ttft_histogram_when_not_present(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event"):
             trace_handler._message_trace(make_message_info(gen_ai_server_time_to_first_token=None))
@@ -1291,6 +1328,41 @@ class TestPromptGenerationTrace:
         assert len(hist_calls) == 1
         assert hist_calls[0][0][1] == pytest.approx(3.2)
 
+    def test_duration_histogram_has_required_labels(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
+        with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event"):
+            trace_handler._prompt_generation_trace(make_prompt_generation_info())
+
+        hist_calls = [
+            c
+            for c in mock_exporter.record_histogram.call_args_list
+            if c[0][0] == EnterpriseTelemetryHistogram.PROMPT_GENERATION_DURATION
+        ]
+        labels = hist_calls[0][0][2]
+        for key in ("tenant_id", "app_id", "operation_type", "model_provider", "model_name", "status"):
+            assert key in labels, f"Missing required label: {key}"
+
+    def test_duration_histogram_status_succeeded(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
+        with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event"):
+            trace_handler._prompt_generation_trace(make_prompt_generation_info())
+
+        hist_calls = [
+            c
+            for c in mock_exporter.record_histogram.call_args_list
+            if c[0][0] == EnterpriseTelemetryHistogram.PROMPT_GENERATION_DURATION
+        ]
+        assert hist_calls[0][0][2]["status"] == "succeeded"
+
+    def test_duration_histogram_status_failed(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
+        with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event"):
+            trace_handler._prompt_generation_trace(make_prompt_generation_info(error="boom"))
+
+        hist_calls = [
+            c
+            for c in mock_exporter.record_histogram.call_args_list
+            if c[0][0] == EnterpriseTelemetryHistogram.PROMPT_GENERATION_DURATION
+        ]
+        assert hist_calls[0][0][2]["status"] == "failed"
+
     def test_total_price_attribute_set_when_present(self, trace_handler: EnterpriseOtelTrace, mock_exporter):
         with patch("enterprise.telemetry.enterprise_trace.emit_metric_only_event") as mock_emit:
             trace_handler._prompt_generation_trace(make_prompt_generation_info(total_price=0.05, currency="USD"))

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

@@ -7,12 +7,12 @@ from enterprise.telemetry.contracts import TelemetryCase
 
 
 @pytest.fixture
-def mock_gateway_emit():
-    with patch("core.telemetry.gateway.emit") as mock:
+def mock_telemetry_emit():
+    with patch("core.telemetry.emit") as mock:
         yield mock
 
 
-def test_handle_app_created_calls_task(mock_gateway_emit):
+def test_handle_app_created_calls_emit(mock_telemetry_emit):
     sender = MagicMock()
     sender.id = "app-123"
     sender.tenant_id = "tenant-456"
@@ -20,35 +20,20 @@ def test_handle_app_created_calls_task(mock_gateway_emit):
 
     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"},
-    )
+    mock_telemetry_emit.assert_called_once()
+    event = mock_telemetry_emit.call_args[0][0]
+    assert event.case == TelemetryCase.APP_CREATED
+    assert event.context.tenant_id == "tenant-456"
+    assert event.payload["app_id"] == "app-123"
+    assert event.payload["mode"] == "chat"
 
 
-def test_handle_app_created_no_exporter(mock_gateway_emit):
-    """Gateway handles exporter availability internally; handler always calls gateway."""
+def test_handle_app_created_no_exporter(mock_telemetry_emit):
+    """Public facade handles exporter availability internally; handler always calls it."""
     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_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"
+    mock_telemetry_emit.assert_called_once()

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

@@ -28,7 +28,7 @@ def test_api_key_only_injects_bearer_header(mock_metric_exporter: MagicMock, moc
         ENTERPRISE_OTLP_ENDPOINT="https://collector.example.com",
         ENTERPRISE_OTLP_HEADERS="",
         ENTERPRISE_OTLP_PROTOCOL="grpc",
-        ENTERPRISE_SERVICE_NAME="dify",
+        APPLICATION_NAME="dify",
         ENTERPRISE_OTEL_SAMPLING_RATE=1.0,
         ENTERPRISE_INCLUDE_CONTENT=True,
         ENTERPRISE_OTLP_API_KEY="test-secret-key",
@@ -51,7 +51,7 @@ def test_empty_api_key_no_auth_header(mock_metric_exporter: MagicMock, mock_span
         ENTERPRISE_OTLP_ENDPOINT="https://collector.example.com",
         ENTERPRISE_OTLP_HEADERS="",
         ENTERPRISE_OTLP_PROTOCOL="grpc",
-        ENTERPRISE_SERVICE_NAME="dify",
+        APPLICATION_NAME="dify",
         ENTERPRISE_OTEL_SAMPLING_RATE=1.0,
         ENTERPRISE_INCLUDE_CONTENT=True,
         ENTERPRISE_OTLP_API_KEY="",
@@ -75,7 +75,7 @@ def test_api_key_and_custom_headers_merge(mock_metric_exporter: MagicMock, mock_
         ENTERPRISE_OTLP_ENDPOINT="https://collector.example.com",
         ENTERPRISE_OTLP_HEADERS="x-custom=foo",
         ENTERPRISE_OTLP_PROTOCOL="grpc",
-        ENTERPRISE_SERVICE_NAME="dify",
+        APPLICATION_NAME="dify",
         ENTERPRISE_OTEL_SAMPLING_RATE=1.0,
         ENTERPRISE_INCLUDE_CONTENT=True,
         ENTERPRISE_OTLP_API_KEY="test-key",
@@ -101,7 +101,7 @@ def test_api_key_overrides_conflicting_header(
         ENTERPRISE_OTLP_ENDPOINT="https://collector.example.com",
         ENTERPRISE_OTLP_HEADERS="authorization=Basic+old",
         ENTERPRISE_OTLP_PROTOCOL="grpc",
-        ENTERPRISE_SERVICE_NAME="dify",
+        APPLICATION_NAME="dify",
         ENTERPRISE_OTEL_SAMPLING_RATE=1.0,
         ENTERPRISE_INCLUDE_CONTENT=True,
         ENTERPRISE_OTLP_API_KEY="test-key",
@@ -131,7 +131,7 @@ def test_https_endpoint_uses_secure_grpc(mock_metric_exporter: MagicMock, mock_s
         ENTERPRISE_OTLP_ENDPOINT="https://collector.example.com",
         ENTERPRISE_OTLP_HEADERS="",
         ENTERPRISE_OTLP_PROTOCOL="grpc",
-        ENTERPRISE_SERVICE_NAME="dify",
+        APPLICATION_NAME="dify",
         ENTERPRISE_OTEL_SAMPLING_RATE=1.0,
         ENTERPRISE_INCLUDE_CONTENT=True,
         ENTERPRISE_OTLP_API_KEY="test-key",
@@ -155,7 +155,7 @@ def test_http_endpoint_uses_insecure_grpc(mock_metric_exporter: MagicMock, mock_
         ENTERPRISE_OTLP_ENDPOINT="http://collector.example.com",
         ENTERPRISE_OTLP_HEADERS="",
         ENTERPRISE_OTLP_PROTOCOL="grpc",
-        ENTERPRISE_SERVICE_NAME="dify",
+        APPLICATION_NAME="dify",
         ENTERPRISE_OTEL_SAMPLING_RATE=1.0,
         ENTERPRISE_INCLUDE_CONTENT=True,
         ENTERPRISE_OTLP_API_KEY="",
@@ -179,7 +179,7 @@ def test_insecure_not_passed_to_http_exporters(mock_metric_exporter: MagicMock,
         ENTERPRISE_OTLP_ENDPOINT="http://collector.example.com",
         ENTERPRISE_OTLP_HEADERS="",
         ENTERPRISE_OTLP_PROTOCOL="http",
-        ENTERPRISE_SERVICE_NAME="dify",
+        APPLICATION_NAME="dify",
         ENTERPRISE_OTEL_SAMPLING_RATE=1.0,
         ENTERPRISE_INCLUDE_CONTENT=True,
         ENTERPRISE_OTLP_API_KEY="test-key",
@@ -204,7 +204,7 @@ def test_api_key_with_special_chars_preserved(mock_metric_exporter: MagicMock, m
         ENTERPRISE_OTLP_ENDPOINT="https://collector.example.com",
         ENTERPRISE_OTLP_HEADERS="",
         ENTERPRISE_OTLP_PROTOCOL="grpc",
-        ENTERPRISE_SERVICE_NAME="dify",
+        APPLICATION_NAME="dify",
         ENTERPRISE_OTEL_SAMPLING_RATE=1.0,
         ENTERPRISE_INCLUDE_CONTENT=True,
         ENTERPRISE_OTLP_API_KEY=special_key,
@@ -227,7 +227,7 @@ def test_no_scheme_localhost_uses_insecure(mock_metric_exporter: MagicMock, mock
         ENTERPRISE_OTLP_ENDPOINT="localhost:4317",
         ENTERPRISE_OTLP_HEADERS="",
         ENTERPRISE_OTLP_PROTOCOL="grpc",
-        ENTERPRISE_SERVICE_NAME="dify",
+        APPLICATION_NAME="dify",
         ENTERPRISE_OTEL_SAMPLING_RATE=1.0,
         ENTERPRISE_INCLUDE_CONTENT=True,
         ENTERPRISE_OTLP_API_KEY="",
@@ -251,7 +251,7 @@ def test_no_scheme_production_uses_insecure(mock_metric_exporter: MagicMock, moc
         ENTERPRISE_OTLP_ENDPOINT="collector.example.com:4317",
         ENTERPRISE_OTLP_HEADERS="",
         ENTERPRISE_OTLP_PROTOCOL="grpc",
-        ENTERPRISE_SERVICE_NAME="dify",
+        APPLICATION_NAME="dify",
         ENTERPRISE_OTEL_SAMPLING_RATE=1.0,
         ENTERPRISE_INCLUDE_CONTENT=True,
         ENTERPRISE_OTLP_API_KEY="",
@@ -328,7 +328,7 @@ def _make_grpc_config(**overrides) -> SimpleNamespace:
         "ENTERPRISE_OTLP_ENDPOINT": "https://collector.example.com",
         "ENTERPRISE_OTLP_HEADERS": "",
         "ENTERPRISE_OTLP_PROTOCOL": "grpc",
-        "ENTERPRISE_SERVICE_NAME": "dify",
+        "APPLICATION_NAME": "dify",
         "ENTERPRISE_OTEL_SAMPLING_RATE": 1.0,
         "ENTERPRISE_INCLUDE_CONTENT": True,
         "ENTERPRISE_OTLP_API_KEY": "",
@@ -484,7 +484,13 @@ def test_export_span_with_start_time_passed_to_start_as_current_span() -> None:
 
 
 def test_export_span_root_span_no_parent_context() -> None:
-    """When span_id_source == correlation_id the span is root — no parent context."""
+    """When span_id_source == correlation_id the span is root.
+
+    An explicit empty ``Context`` is passed (not ``None``) so the root span never
+    implicitly inherits an ambient active span from the surrounding context.
+    """
+    from opentelemetry.context import Context
+
     exporter, mock_tracer, mock_span = _make_exporter_with_mock_tracer()
 
     uid = "123e4567-e89b-12d3-a456-426614174000"
@@ -496,7 +502,9 @@ def test_export_span_root_span_no_parent_context() -> None:
     )
 
     _, kwargs = mock_tracer.start_as_current_span.call_args
-    assert kwargs["context"] is None
+    assert isinstance(kwargs["context"], Context)
+    # An empty context carries no active span, guaranteeing a deterministic root trace_id.
+    assert len(kwargs["context"]) == 0
 
 
 def test_export_span_child_span_has_parent_context() -> None:

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

@@ -1,92 +1,101 @@
 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 core.telemetry.events import (
+    AppCreatedEvent,
+    AppCreatedPayload,
+    DraftNodeExecutionTraceEvent,
+    NodeExecutionPayload,
+    PromptGenerationEvent,
+    PromptGenerationPayload,
+    TelemetryContext,
+    TelemetryEvent,
 )
+from core.telemetry.gateway import PAYLOAD_SIZE_THRESHOLD_BYTES, emit
 from enterprise.telemetry.contracts import SignalType, TelemetryCase, TelemetryEnvelope
 
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
 
-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}"
+_TRACE_CTX = TelemetryContext(tenant_id="tenant-789", user_id="user-456", app_id="app-123")
+_METRIC_CTX = TelemetryContext(tenant_id="tenant-123")
 
 
-@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()
+def _node_event() -> DraftNodeExecutionTraceEvent:
+    return DraftNodeExecutionTraceEvent(
+        context=_TRACE_CTX,
+        payload=NodeExecutionPayload(node_execution_data={"key": "val"}),
+    )
 
-    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
+def _prompt_event() -> PromptGenerationEvent:
+    return PromptGenerationEvent(
+        context=_TRACE_CTX,
+        payload=PromptGenerationPayload(
+            tenant_id="tenant-789",
+            operation_type="generate",
+            instruction="test",
+            generated_output="out",
+            model_provider="openai",
+            model_name="gpt-4",
+            prompt_tokens=10,
+            completion_tokens=5,
+            total_tokens=15,
+            latency=0.5,
+        ),
+    )
+
+
+def _app_created_event(tenant_id: str = "tenant-123") -> AppCreatedEvent:
+    return AppCreatedEvent(
+        context=TelemetryContext(tenant_id=tenant_id),
+        payload=AppCreatedPayload(app_id="app-abc", mode="chat"),
+    )
+
+
+# ---------------------------------------------------------------------------
+# Event routing metadata
+# ---------------------------------------------------------------------------
+
+
+class TestEventRoutingMetadata:
+    """Verify each event class declares correct routing fields."""
+
+    def test_draft_node_execution_routing(self) -> None:
+        ev = _node_event()
+        assert ev.case == TelemetryCase.DRAFT_NODE_EXECUTION
+        assert ev.signal_type is SignalType.TRACE
+        assert ev.ce_eligible is False
+        assert ev.trace_task_name is TraceTaskName.DRAFT_NODE_EXECUTION_TRACE
+
+    def test_prompt_generation_routing(self) -> None:
+        ev = _prompt_event()
+        assert ev.case == TelemetryCase.PROMPT_GENERATION
+        assert ev.signal_type is SignalType.TRACE
+        assert ev.ce_eligible is False
+        assert ev.trace_task_name is TraceTaskName.PROMPT_GENERATION_TRACE
+
+    def test_app_created_routing(self) -> None:
+        ev = _app_created_event()
+        assert ev.case == TelemetryCase.APP_CREATED
+        assert ev.signal_type is SignalType.METRIC_LOG
+        assert ev.ce_eligible is False
+        assert ev.trace_task_name is None
+
+    def test_all_events_satisfy_protocol(self) -> None:
+        events: list[TelemetryEvent] = [_node_event(), _prompt_event(), _app_created_event()]
+        for ev in events:
+            assert isinstance(ev, TelemetryEvent)
+
+
+# ---------------------------------------------------------------------------
+# Trace routing
+# ---------------------------------------------------------------------------
 
 
 class TestGatewayTraceRouting:
@@ -95,31 +104,12 @@ class TestGatewayTraceRouting:
         return MagicMock()
 
     @patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=True)
-    def test_trace_case_routes_to_trace_manager(
+    def test_trace_event_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)
-
+        emit(_prompt_event(), mock_trace_manager)
         mock_trace_manager.add_trace_task.assert_called_once()
 
     @patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=False)
@@ -127,13 +117,8 @@ class TestGatewayTraceRouting:
         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)
-
+        emit(_node_event(), mock_trace_manager)
         mock_trace_manager.add_trace_task.assert_not_called()
 
     @patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=True)
@@ -141,28 +126,25 @@ class TestGatewayTraceRouting:
         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)
-
+        emit(_node_event(), mock_trace_manager)
         mock_trace_manager.add_trace_task.assert_called_once()
 
 
+# ---------------------------------------------------------------------------
+# Metric/log routing
+# ---------------------------------------------------------------------------
+
+
 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(
+    def test_metric_event_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)
+        emit(_app_created_event())
 
         mock_delay.assert_called_once()
         envelope_json = mock_delay.call_args[0][0]
@@ -178,11 +160,8 @@ class TestGatewayMetricLogRouting:
         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)
+        emit(_app_created_event())
+        emit(_app_created_event())
 
         assert mock_delay.call_count == 2
         envelope1 = TelemetryEnvelope.model_validate_json(mock_delay.call_args_list[0][0][0])
@@ -190,6 +169,11 @@ class TestGatewayMetricLogRouting:
         assert envelope1.event_id != envelope2.event_id
 
 
+# ---------------------------------------------------------------------------
+# Payload sizing
+# ---------------------------------------------------------------------------
+
+
 class TestGatewayPayloadSizing:
     @patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=True)
     @patch("tasks.enterprise_telemetry_task.process_enterprise_telemetry.delay")
@@ -198,14 +182,11 @@ class TestGatewayPayloadSizing:
         mock_delay: MagicMock,
         mock_ee_enabled: MagicMock,
     ) -> None:
-        context = {"tenant_id": "tenant-123"}
-        payload = {"key": "small_value"}
-
-        emit(TelemetryCase.APP_CREATED, context, payload)
+        emit(_app_created_event())
 
         envelope_json = mock_delay.call_args[0][0]
         envelope = TelemetryEnvelope.model_validate_json(envelope_json)
-        assert envelope.payload == payload
+        assert envelope.payload["app_id"] == "app-abc"
         assert envelope.metadata is None
 
     @patch("core.telemetry.gateway.is_enterprise_telemetry_enabled", return_value=True)
@@ -217,11 +198,12 @@ class TestGatewayPayloadSizing:
         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)
+        ev = AppCreatedEvent(
+            context=_METRIC_CTX,
+            payload=AppCreatedPayload(app_id=large_value),
+        )
+        emit(ev)
 
         mock_storage.save.assert_called_once()
         storage_key = mock_storage.save.call_args[0][0]
@@ -243,30 +225,14 @@ class TestGatewayPayloadSizing:
         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)
+        ev = AppCreatedEvent(
+            context=_METRIC_CTX,
+            payload=AppCreatedPayload(app_id=large_value),
+        )
+        emit(ev)
 
         envelope_json = mock_delay.call_args[0][0]
         envelope = TelemetryEnvelope.model_validate_json(envelope_json)
-        assert envelope.payload == payload
+        assert envelope.payload["app_id"] == large_value
         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