feat!: emit per-chunk streaming span events on backend spans

docs/docs/observability/telemetry.md

@@ -34,6 +34,7 @@ All telemetry is configured via environment variables:
 | `MELLEA_TRACES_OTLP` | Enable OTLP span exporter | `false` |
 | `MELLEA_TRACES_CONSOLE` | Print traces to console (debugging) | `false` |
 | `MELLEA_TRACES_CONTENT` | Capture prompt/response content on spans (may include PII) | `false` |
+| `MELLEA_GENERATION_CHUNK_EVENTS` | Emit a `chunk_processed` span event per streamed chunk on backend spans | `false` |
 | `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT` | Trace-specific OTLP endpoint (overrides general) | none |
 
 ### Metrics variables

docs/docs/observability/tracing.md

@@ -239,6 +239,10 @@ Mellea also adds context-specific attributes to backend spans:
 | `mellea.tool_calls_enabled` | Whether tool calling is enabled |
 | `mellea.num_actions` | Number of actions in batch (for `generate_from_raw`) |
 
+When `MELLEA_GENERATION_CHUNK_EVENTS=true`, backend spans also record a `chunk_processed`
+span event per streamed chunk, carrying its index and added text length. This is
+opt-in and off by default, since a long response produces one event per chunk.
+
 ### Span hierarchy
 
 Backend spans nest inside application spans:

mellea/core/base.py

@@ -23,6 +23,7 @@
 import datetime
 import enum
 import logging
+import os
 import threading
 from collections import OrderedDict
 from collections.abc import Callable, Coroutine, Iterable, Mapping
@@ -49,6 +50,7 @@
 
 from ..plugins.manager import has_plugins, invoke_hook
 from ..plugins.types import HookType
+from .utils import _parse_bool_env
 
 
 class CBlock:
@@ -751,6 +753,9 @@ class _GenerationState:
         chunk_size: Minimum number of chunks to stream at a single time.
         first_chunk_received: Whether the first streamed chunk has arrived
             (gates time-to-first-byte recording).
+        processed_chunk_index: Monotonic index of the next streamed chunk to be
+            processed, incremented once per chunk folded into the value across the
+            repeated `astream()` calls of one generation.
         generate: The task driving generation. Linked to `generate_type`.
         generate_type: Determines which functions can resolve the thunk's value.
         generate_extra: Auxiliary generation task; currently only used by hf.
@@ -770,6 +775,7 @@ class _GenerationState:
     queue: asyncio.Queue = field(default_factory=lambda: asyncio.Queue(maxsize=20))
     chunk_size: int = 3
     first_chunk_received: bool = False
+    processed_chunk_index: int = 0
     generate: asyncio.Task[None] | None = None
     generate_type: GenerateType = GenerateType.NONE
     generate_extra: asyncio.Task[Any] | None = None
@@ -853,6 +859,16 @@ def _record_ttfb(self) -> None:
             ).total_seconds() * 1000
             self._gen.first_chunk_received = True
 
+    async def _emit_event(self, event_name: str, **data: Any) -> None:
+        """Fire a `generation_event` hook named `event_name` carrying `data`, if any plugin subscribes."""
+        if has_plugins(HookType.GENERATION_EVENT):
+            from ..plugins.hooks.generation import GenerationEventPayload
+
+            event_payload = GenerationEventPayload(
+                generation_id=self._call.generation_id, event_name=event_name, data=data
+            )
+            await invoke_hook(HookType.GENERATION_EVENT, event_payload)
+
     async def cancel_generation(self, error: Exception | None = None) -> None:
         """Cancel an in-progress streaming generation, drain the queue, and fire the `generation_error` hook.
 
@@ -1153,9 +1169,20 @@ async def astream(self) -> str:
 
             raise chunks[-1]
 
+        emit_chunk_events = self.generation.streaming and _parse_bool_env(
+            os.getenv("MELLEA_GENERATION_CHUNK_EVENTS", ""), default=False
+        )
         for chunk in chunks:
             assert self._gen.process is not None
+            prev_len = len(str(self._underlying_value or ""))
             await self._gen.process(self, chunk)
+            if emit_chunk_events:
+                await self._emit_event(
+                    "chunk_processed",
+                    chunk_index=self._gen.processed_chunk_index,
+                    chunk_text_length=len(str(self._underlying_value or "")) - prev_len,
+                )
+            self._gen.processed_chunk_index += 1
 
         if do_set_computed:
             assert self._underlying_value is not None

mellea/core/utils.py

@@ -69,9 +69,6 @@
 except ImportError:
     _OTEL_AVAILABLE = False
 
-from ..telemetry import get_otlp_log_handler
-from ..telemetry.context import _CONTEXT_VARS as _telemetry_vars, MelleaContextFilter
-
 # ---------------------------------------------------------------------------
 # Per-task/coroutine context fields (safe for asyncio — each Task gets its own copy)
 # ---------------------------------------------------------------------------
@@ -426,6 +423,8 @@ def _build_log_dict(self, record: logging.LogRecord) -> dict[str, Any]:
         # MelleaContextFilter stamps these onto the record before formatters run; read
         # them back off the record here so they appear in JSON output.  Fall back to the
         # ContextVar directly so the formatter still works without the filter attached.
+        from ..telemetry.context import _CONTEXT_VARS as _telemetry_vars
+
         for key, var in _telemetry_vars.items():
             value = getattr(record, key, var.get())
             if value is not None:
@@ -577,6 +576,8 @@ def configure_logging(logger: logging.Logger) -> None:
     Args:
         logger: The `logging.Logger` to configure.
     """
+    from ..telemetry import get_otlp_log_handler
+
     enabled_raw = os.environ.get("MELLEA_LOGS_ENABLED")
     if not _parse_bool_env(enabled_raw or "", default=True):
         return
@@ -701,6 +702,8 @@ def get_logger() -> logging.Logger:
         Returns:
             Configured logger instance.
         """
+        from ..telemetry.context import MelleaContextFilter
+
         if MelleaLogger.logger is None:
             with _logger_lock:
                 # Second check inside the lock: another thread may have finished

mellea/plugins/hooks/generation.py

@@ -83,6 +83,32 @@ class GenerationErrorPayload(MelleaBasePayload):
     generation_id: str | None = None
 
 
+class GenerationEventPayload(MelleaBasePayload):
+    """Payload for `generation_event` — a milestone event during a single generation.
+
+    A generic carrier for events emitted mid-generation. Subscribers branch on
+    `event_name` and read the keys `data` carries for that event.
+
+    Known events:
+        `chunk_processed`: emitted once per streamed chunk during `astream()`
+            (opt-in via `MELLEA_GENERATION_CHUNK_EVENTS`). `data` keys:
+            `chunk_index` (int), `chunk_text_length` (int).
+
+    Attributes:
+        generation_id: Mellea-side hook correlation ID matching the corresponding
+            pre_call payload, distinct from the provider-assigned
+            `GenerationMetadata.response_id`. `None` when the firing site did not
+            generate one.
+        event_name: Identifies the event. Subscribers dispatch on this.
+        data: Values for this event, keyed by name. The keys present depend on
+            `event_name` (see Known events above).
+    """
+
+    generation_id: str | None = None
+    event_name: str = ""
+    data: dict[str, Any] = {}
+
+
 class GenerationBatchPreCallPayload(MelleaBasePayload):
     """Payload for `generation_batch_pre_call` — fires once before a batch generation request.
 

mellea/plugins/policies.py

@@ -65,7 +65,7 @@ def _build_policies() -> dict[str, Any]:
         "generation_batch_pre_call": HookPayloadPolicy(
             writable_fields=frozenset({"model_options", "tool_calls", "format"})
         ),
-        # generation_post_call, generation_batch_post_call: observe-only
+        # generation_post_call, generation_batch_post_call, generation_event: observe-only
         # Validation
         "validation_pre_check": HookPayloadPolicy(
             writable_fields=frozenset({"requirements", "model_options"})

mellea/plugins/types.py

@@ -55,6 +55,7 @@ class HookType(StrEnum):
     GENERATION_BATCH_PRE_CALL = "generation_batch_pre_call"
     GENERATION_BATCH_POST_CALL = "generation_batch_post_call"
     GENERATION_BATCH_ERROR = "generation_batch_error"
+    GENERATION_EVENT = "generation_event"
 
     # Validation
     VALIDATION_PRE_CHECK = "validation_pre_check"
@@ -106,6 +107,7 @@ def _build_hook_registry() -> dict[str, tuple[type, type]]:
         GenerationBatchPostCallPayload,
         GenerationBatchPreCallPayload,
         GenerationErrorPayload,
+        GenerationEventPayload,
         GenerationPostCallPayload,
         GenerationPreCallPayload,
     )
@@ -166,6 +168,7 @@ def _build_hook_registry() -> dict[str, tuple[type, type]]:
             GenerationBatchErrorPayload,
             PluginResult,
         ),
+        HookType.GENERATION_EVENT.value: (GenerationEventPayload, PluginResult),
         # Validation
         HookType.VALIDATION_PRE_CHECK.value: (ValidationPreCheckPayload, PluginResult),
         HookType.VALIDATION_POST_CHECK.value: (

mellea/telemetry/tracing_plugins.py

@@ -7,7 +7,8 @@
 pipelines to automatically emit spans when tracing is enabled:
 
 - BackendTracingPlugin: Emits Gen-AI semconv backend spans for every LLM
-  generation, on both chat and raw (batch) paths.
+  generation, on both chat and raw (batch) paths, plus mid-generation span
+  events from the generation_event hook.
 - ComponentTracingPlugin: Emits application-level spans tracking component
   execution.
 - StreamingTracingPlugin: Emits an application-level orchestration span and
@@ -44,6 +45,7 @@
         GenerationBatchPostCallPayload,
         GenerationBatchPreCallPayload,
         GenerationErrorPayload,
+        GenerationEventPayload,
         GenerationPostCallPayload,
         GenerationPreCallPayload,
     )
@@ -73,10 +75,13 @@ class BackendTracingPlugin(Plugin, name="backend_tracing", priority=1040):
     This plugin hooks into the generation pre-call, post-call, and error
     events on both the chat and raw (batch) paths to automatically emit one
     span per LLM call. Spans are started on pre-call and ended on post-call
-    or error, correlated across hooks via generation_id.
+    or error, correlated across hooks via generation_id. It also records
+    mid-generation span events from the `generation_event` hook onto the
+    in-flight span.
 
     All hooks run SEQUENTIAL so the OTel context token attached in pre-call
-    can be detached on the same task in post-call / error.
+    can be detached on the same task in post-call / error, and so
+    `generation_event` appends to the span before post-call ends it.
     """
 
     # --- Chat hooks ---
@@ -137,6 +142,27 @@ async def on_error(
             gen=gen,
         )
 
+    @hook("generation_event")
+    async def on_generation_event(
+        self, payload: GenerationEventPayload, context: dict[str, Any]
+    ) -> None:
+        """Record a span event on the in-flight backend span for one `generation_event`."""
+        if not payload.generation_id:
+            return
+        from mellea.telemetry.tracing import add_span_event
+
+        if payload.event_name == "chunk_processed":
+            add_span_event(
+                payload.generation_id,
+                event_name="chunk_processed",
+                attributes={
+                    "mellea.generation.chunk_index": payload.data.get("chunk_index"),
+                    "mellea.generation.chunk_text_length": payload.data.get(
+                        "chunk_text_length"
+                    ),
+                },
+            )
+
     # --- Batch hooks ---
 
     @hook("generation_batch_pre_call")
@@ -355,34 +381,44 @@ async def on_streaming_event(
                 payload.streaming_id,
                 event_name="quick_check",
                 attributes={
-                    "chunk_index": ev.chunk_index,
-                    "passed": ev.passed,
-                    "requirement_count": len(ev.results),
+                    "mellea.streaming.chunk_index": ev.chunk_index,
+                    "mellea.validation.passed": ev.passed,
+                    "mellea.validation.requirement_count": len(ev.results),
                 },
             )
         elif isinstance(ev, ChunkEvent):
             add_span_event(
                 payload.streaming_id,
                 event_name="chunk",
-                attributes={"chunk_index": ev.chunk_index, "text_length": len(ev.text)},
+                attributes={
+                    "mellea.streaming.chunk_index": ev.chunk_index,
+                    "mellea.streaming.chunk_text_length": len(ev.text),
+                },
             )
         elif isinstance(ev, StreamingDoneEvent):
             add_span_event(
                 payload.streaming_id,
                 event_name="streaming_done",
-                attributes={"full_text_length": len(ev.full_text)},
+                attributes={"mellea.streaming.full_text_length": len(ev.full_text)},
             )
         elif isinstance(ev, FullValidationEvent):
             add_span_event(
                 payload.streaming_id,
                 event_name="full_validation",
-                attributes={"passed": ev.passed, "requirement_count": len(ev.results)},
+                attributes={
+                    "mellea.validation.passed": ev.passed,
+                    "mellea.validation.requirement_count": len(ev.results),
+                },
             )
         elif isinstance(ev, ErrorEvent):
+            # Not OTel's reserved `exception.*`: `detail` isn't always the plain message.
             add_span_event(
                 payload.streaming_id,
                 event_name="error",
-                attributes={"exception_type": ev.exception_type, "detail": ev.detail},
+                attributes={
+                    "mellea.error.type": ev.exception_type,
+                    "mellea.error.detail": ev.detail,
+                },
             )
 
     @hook("streaming_end")
@@ -398,8 +434,8 @@ async def on_streaming_end(
             payload.streaming_id,
             event_name="completed",
             attributes={
-                "success": payload.success,
-                "full_text_length": payload.full_text_length,
+                "mellea.streaming.success": payload.success,
+                "mellea.streaming.full_text_length": payload.full_text_length,
             },
         )
         finish_streaming_span(
@@ -508,10 +544,10 @@ async def on_iteration(
             payload.sampling_id,
             event_name="iteration",
             attributes={
-                "iteration": payload.iteration,
-                "all_validations_passed": payload.all_validations_passed,
-                "valid_count": payload.valid_count,
-                "total_count": payload.total_count,
+                "mellea.sampling.iteration": payload.iteration,
+                "mellea.sampling.all_validations_passed": payload.all_validations_passed,
+                "mellea.validation.valid_count": payload.valid_count,
+                "mellea.validation.requirement_count": payload.total_count,
             },
         )
 
@@ -528,9 +564,9 @@ async def on_repair(
             payload.sampling_id,
             event_name="repair",
             attributes={
-                "repair_iteration": payload.repair_iteration,
-                "repair_type": payload.repair_type,
-                "failed_count": len(payload.failed_validations),
+                "mellea.sampling.repair_iteration": payload.repair_iteration,
+                "mellea.sampling.repair_type": payload.repair_type,
+                "mellea.validation.failed_count": len(payload.failed_validations),
             },
         )
 

test/telemetry/test_tracing_backend.py

@@ -114,13 +114,20 @@ async def fake_chat(*args, **kwargs):
 
 @pytest.mark.integration
 @pytest.mark.asyncio
-async def test_streaming_span_creates_and_closes_span(span_exporter):
+@pytest.mark.parametrize("emit", [True, False], ids=["emit_on", "emit_off_default"])
+async def test_streaming_span_creates_and_closes_span(span_exporter, monkeypatch, emit):
     """Streaming backend call creates a chat span that closes after the stream completes.
 
     Uses a mocked Ollama client so no server is needed.  Verifies the core
     TracingPlugin invariant: the span must remain open for the full duration of
-    streaming and close only once all chunks are consumed.
+    streaming and close only once all chunks are consumed. `chunk_processed`
+    events are emitted only when `MELLEA_GENERATION_CHUNK_EVENTS` is on; with the env
+    unset (the default) the span carries no such events.
     """
+    if emit:
+        monkeypatch.setenv("MELLEA_GENERATION_CHUNK_EVENTS", "true")
+    else:
+        monkeypatch.delenv("MELLEA_GENERATION_CHUNK_EVENTS", raising=False)
 
     async def fake_chat_stream(*args, **kwargs):
         for content in ["1", " 2", " 3"]:
@@ -180,6 +187,16 @@ async def fake_chat_stream(*args, **kwargs):
         "the streaming delay, suggesting the span did not stay open for the full stream"
     )
 
+    chunk_events = [e for e in backend_span.events if e.name == "chunk_processed"]
+    if emit:
+        assert chunk_events, (
+            "expected chunk_processed events with MELLEA_GENERATION_CHUNK_EVENTS on"
+        )
+    else:
+        assert not chunk_events, (
+            "expected no chunk_processed events when MELLEA_GENERATION_CHUNK_EVENTS is unset"
+        )
+
 
 @pytest.mark.integration
 @pytest.mark.asyncio