feat(telemetry): CloudEvents + Langfuse tracing as projections over a captured LLM stream (#845)

* feat(telemetry): CloudEvents + Langfuse tracing as projections over a captured LLM stream

Capture each LLM call once (CapturedLLMCall) and fan it out to multiple
exporters -- "one data model, two projections": a CloudEvents trace stream
(llm.call.traced / trace.content) and a Langfuse projection, both reconstructing
trace -> run -> step -> generation from the same source of truth.

- Capture seam (src/llm/capture.py): one canonicalization + content-addressed
  hashing point, with an O(N) per-span memo so repeated context isn't re-hashed.
- Session correlation threaded telemetry -> captured call -> exporters,
  namespaced only at the Langfuse export boundary.
- Span identity consolidated onto LLMTelemetryContext; dropped TRACE_ENDPOINT.
- Canonical generation/step names; dreamer branches nest under one dream trace;
  tool calls become spans under their step.
- LANGFUSE_EXPORTER_MODE toggle ("exporter" default; "inline" kept one release
  for side-by-side validation), centralized into computed settings predicates.
- Per-run/per-trace dedup registries (trace_session, langfuse_session) bounded
  by an LRU so dedup and span grouping survive long-running workers.
- Embedding-call tracing; deterministic high-volume event sampling.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* fix(telemetry): address trace-review findings (span/step_seq collisions, test, logging)

- Dreamer specialists mint a distinct span_id per execution (trace_id stays the
  shared dream run_id), so their CloudEvents trace resource ids no longer collide
  between deduction and induction.
- Tool-loop no-tool early-return streams the tail with the next ordinal
  (iteration+2) instead of reusing the in-loop call's step_seq, avoiding a
  colliding trace resource id; mirrors the synthesis path.
- Tighten test_clips_oversized_string to assert output stays within TRACE_MAX_BYTES.
- emit_trace logs the swallowed exception with exc_info for debuggability.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* fix(telemetry): silence exporter-mode Langfuse warning + drop summarizer run_id placeholder

Two CloudEvents/Langfuse correctness fixes, independent of the trace viewer.

Langfuse exporter-mode gating: annotate_current_generation_io (and its two
executor.py call-site guards) were gated on LANGFUSE_PUBLIC_KEY instead of
langfuse_inline_enabled. In the default `exporter` mode they called
get_client().update_current_generation() with no active @observe span, logging
"No active span in current context" (~14 per dialectic run) and building
throwaway model_dump payloads on every LLM call. The LangfuseExporter projects
I/O from the captured stream, so these helpers must no-op in exporter mode.
Gated all three on langfuse_inline_enabled; added a regression test; fixed a
stale conditional_observe docstring.

Summarizer run_id placeholder: AgentToolSummaryCreatedEvent hardcoded
run_id="deriver"/iteration=0 because summarization is a single LLM call, not an
agentic run. That placeholder pollutes run_id grouping in the CloudEvents stream
(any consumer that groups by run_id sees a phantom "deriver" run). Made
run_id/iteration optional (None) and re-keyed get_resource_id on
message_id:summary_type (the real per-summary identity; run_id/iteration can no
longer identify it); bumped schema_version 2->3. Xatu ingestion stores only the
CloudEvent envelope, so the field/resource_id/version changes are transparent to it.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* docs: update docstrings to be less verbose

* fix(telemetry): address PR review on captured-stream tracing

- embedding traces get a fresh span_id under parent_span_id=run_id, so
  sibling embeddings in one run no longer share a span/idempotency key
- capture the provider finish_reason from stream chunks instead of
  hardcoding "stop" on a successful drain
- gate the Langfuse exporter behind TELEMETRY.ENABLED (master switch) so
  disabling telemetry sends no traces at all
- rename _emit_derived_content -> _emit_hashed_content
- inline the _emit_trace wrapper; drop unused trace_session.end_run

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* refactor: rename TELEMETRY_TRACE_PAYLOADS to TELEMETRY_TRACE_PAYLOADS_ENABLED

* fix(telemetry): capture provider tool calls in trace stream

The captured trace stream dropped assistant tool calls for openai/gemini:
build_captured_messages only read {role, content, tool_call_id}, but those
providers keep tool calls outside content (openai's tool_calls, gemini's
parts), so replayed tool-call turns landed as empty content and gemini lost
its text and tool results entirely. Anthropic (tool_use in content) was fine.

Normalize each input message per provider into a unified tool_calls
[{id, name, input}] field on CapturedMessage/TraceContentEvent, recovering
gemini text/results along the way, and fold tool_calls into
compute_content_hash so empty-content openai turns no longer collide in the
dedup store. langfuse_exporter._input now surfaces the calls.

Also fix a silent serialization drop: gemini thought_signature is bytes, so
model_dump(mode="json") on the traced event raised UnicodeDecodeError and
emit_trace swallowed it -- dropping the whole tool-calling iteration from the
trace stream (billing and Langfuse were unaffected). base64-encode the
signature on the telemetry path; replay keeps the raw bytes.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(telemetry): type replay tool-call dict for bytes signature

thought_signature widened to str | bytes | None, but
_tool_call_result_to_dict's literal was inferred as
dict[str, str | dict[str, Any]], so the bytes assignment failed project-wide
basedpyright (the per-file pre-commit hook didn't catch it). Annotate the
dict as dict[str, Any]; the replay path keeps the raw bytes unchanged.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* test: remove 3 tests

---------

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

Author: akattelu

.env.template

@@ -276,6 +276,11 @@ LLM_OPENAI_API_KEY=your-api-key-here
 # TELEMETRY_MAX_BUFFER_SIZE=10000
 # TELEMETRY_NAMESPACE=honcho  # Inherits from NAMESPACE if not set
 
+# Full-fidelity payload tracing (llm.call.traced / trace.content). Default-off
+# TELEMETRY_TRACE_PAYLOADS_ENABLED=false  # Trace events ship to TELEMETRY_ENDPOINT
+# TELEMETRY_TRACE_MAX_BYTES=262144  # Per-message cap; oversized content is clipped
+# TELEMETRY_TRACE_PURPOSES=[]  # JSON list of CallPurpose values to capture; empty = all
+
 # =============================================================================
 # Cache
 # =============================================================================

src/config.py

@@ -1176,6 +1176,19 @@ class TelemetrySettings(HonchoSettings):
     # that join high-volume events to aggregate envelopes first.
     HIGH_VOLUME_SAMPLE_RATE: Annotated[float, Field(default=1.0, ge=0.0, le=1.0)] = 1.0
 
+    # --- Full-fidelity payload tracing (llm.call.traced / trace.content) ---
+    # Master toggle for replay-grade content capture. Default-off.
+    TRACE_PAYLOADS_ENABLED: bool = False
+
+    # Per-message cap (bytes) for captured content; oversized string content is
+    # clipped (with a marker) and the call is flagged was_truncated.
+    TRACE_MAX_BYTES: Annotated[int, Field(default=262144, gt=0)] = 262144
+
+    # Allowlist of CallPurpose values to capture; empty = all. Typed as str to
+    # keep the enum out of config (validated against CallPurpose at the producer,
+    # same pattern as LLMTelemetryContext.call_purpose).
+    TRACE_PURPOSES: list[str] = Field(default_factory=list)
+
 
 class CacheSettings(HonchoSettings):
     model_config = SettingsConfigDict(env_prefix="CACHE_", extra="ignore")  # pyright: ignore
@@ -1345,6 +1358,17 @@ def _require_api_key_for_turbopuffer(self) -> "VectorStoreSettings":
         return self
 
 
+class TraceViewerSettings(HonchoSettings):
+    model_config = SettingsConfigDict(env_prefix="TRACE_VIEWER_", extra="ignore")  # pyright: ignore
+
+    ENABLED: bool = False
+    HOST: str = "127.0.0.1"
+    PORT: int = 8002
+    STORAGE_DIR: str = "./traces"
+    MAX_REQUEST_BYTES: int = 10 * 1024 * 1024  # 10 MB
+    VENDOR_CDN_BASE: str = "https://cdn.jsdelivr.net/npm"
+
+
 class AppSettings(HonchoSettings):
     # No env_prefix for app-level settings
     model_config = SettingsConfigDict(  # pyright: ignore
@@ -1364,6 +1388,29 @@ class AppSettings(HonchoSettings):
     EMBED_MESSAGES: bool = True
     LANGFUSE_HOST: str | None = None
     LANGFUSE_PUBLIC_KEY: str | None = None
+    # How Langfuse traces are produced:
+    #   "exporter" (default) — Langfuse is a projection over the captured
+    #     CapturedLLMCall stream (LangfuseExporter), the same source of truth as
+    #     the CloudEvents trace stream.
+    #   "inline" — legacy live instrumentation (@observe + propagate_attributes
+    #     spans during execution). Kept one release for side-by-side validation.
+    LANGFUSE_EXPORTER_MODE: Literal["inline", "exporter"] = "exporter"
+
+    @property
+    def langfuse_inline_enabled(self) -> bool:
+        """True when the legacy inline Langfuse instrumentation is active
+        (keys configured + ``LANGFUSE_EXPORTER_MODE == "inline"``)."""
+        return (
+            bool(self.LANGFUSE_PUBLIC_KEY) and self.LANGFUSE_EXPORTER_MODE == "inline"
+        )
+
+    @property
+    def langfuse_exporter_enabled(self) -> bool:
+        """True when the Langfuse exporter (a projection over the captured call
+        stream) is active (keys configured + ``LANGFUSE_EXPORTER_MODE == "exporter"``)."""
+        return (
+            bool(self.LANGFUSE_PUBLIC_KEY) and self.LANGFUSE_EXPORTER_MODE == "exporter"
+        )
 
     # Origins allowed by the FastAPI CORSMiddleware
     CORS_ORIGINS: list[str] = [
@@ -1394,6 +1441,7 @@ class AppSettings(HonchoSettings):
     CACHE: CacheSettings = Field(default_factory=CacheSettings)
     DREAM: DreamSettings = Field(default_factory=DreamSettings)
     VECTOR_STORE: VectorStoreSettings = Field(default_factory=VectorStoreSettings)
+    TRACE_VIEWER: TraceViewerSettings = Field(default_factory=TraceViewerSettings)
 
     @field_validator("LOG_LEVEL")
     def validate_log_level(cls, v: str) -> str:

src/deriver/deriver.py

@@ -1,6 +1,8 @@
 import logging
 import time
 
+from nanoid import generate as generate_nanoid
+
 from src import crud
 from src.config import ConfiguredModelSettings, settings
 from src.crud.representation import RepresentationManager
@@ -142,6 +144,7 @@ async def process_representation_tasks_batch(
     model_config = base_model_config
 
     # Single LLM call
+    trace_id = generate_nanoid()
     llm_start = time.perf_counter()
     response = await honcho_llm_call(
         model_config=model_config,
@@ -159,6 +162,8 @@ async def process_representation_tasks_batch(
             parent_category="representation",
             observed=observed,
             track_name="Minimal Deriver",
+            trace_id=trace_id,
+            span_id=trace_id,
         ),
     )
     llm_duration = (time.perf_counter() - llm_start) * 1000

src/dialectic/chat.py

@@ -50,6 +50,9 @@ async def agentic_chat(
             session = await crud.get_session(
                 db, workspace_name=workspace_name, session_name=session_name
             )
+        # Read the opaque Session.id while the instance is still bound; the ORM
+        # object detaches once this read-only session closes below.
+        session_id = session.id if session else None
         workspace = await crud.get_workspace(db, workspace_name=workspace_name)
         configuration = get_configuration(None, session, workspace)
 
@@ -68,6 +71,7 @@ async def agentic_chat(
     agent = DialecticAgent(
         workspace_name=workspace_name,
         session_name=session_name,
+        session_id=session_id,
         observer=observer,
         observed=observed,
         observer_peer_card=observer_peer_card,
@@ -111,6 +115,9 @@ async def agentic_chat_stream(
             session = await crud.get_session(
                 db, workspace_name=workspace_name, session_name=session_name
             )
+        # Read the opaque Session.id while the instance is still bound; the ORM
+        # object detaches once this read-only session closes below.
+        session_id = session.id if session else None
         workspace = await crud.get_workspace(db, workspace_name=workspace_name)
         configuration = get_configuration(None, session, workspace)
 
@@ -129,6 +136,7 @@ async def agentic_chat_stream(
     agent = DialecticAgent(
         workspace_name=workspace_name,
         session_name=session_name,
+        session_id=session_id,
         observer=observer,
         observed=observed,
         observer_peer_card=observer_peer_card,

src/dialectic/core.py

@@ -68,6 +68,7 @@ def __init__(
         observed_peer_card: list[str] | None = None,
         metric_key: str | None = None,
         reasoning_level: ReasoningLevel = "low",
+        session_id: str | None = None,
     ):
         """
         Initialize the dialectic agent.
@@ -81,9 +82,11 @@ def __init__(
             observed_peer_card: Biographical information about the observed peer
             metric_key: Optional key for logging metrics (if provided, agent won't log separately)
             reasoning_level: Level of reasoning to apply
+            session_id: ID used for grouping traces (not session_name)
         """
         self.workspace_name: str = workspace_name
         self.session_name: str | None = session_name
+        self.session_id: str | None = session_id
         self.observer: str = observer
         self.observed: str = observed
         self.observer_peer_card: list[str] | None = observer_peer_card
@@ -179,6 +182,7 @@ async def _prefetch_relevant_observations(self, query: str) -> str | None:
                 workspace_name=self.workspace_name,
                 run_id=self._run_id,
                 parent_category="dialectic",
+                session_id=self.session_id,
             ):
                 query_embedding = await embedding_client.embed(query)
 
@@ -316,6 +320,9 @@ def _telemetry_context(self, track_name: str | None = None) -> LLMTelemetryConte
             parent_category="dialectic",
             agent_type="dialectic",
             run_id=self._run_id,
+            trace_id=self._run_id,
+            span_id=self._run_id,
+            session_id=self.session_id,
             peer_name=self.observed,
             track_name=track_name,
         )

src/dreamer/specialists.py

@@ -169,6 +169,10 @@ async def run(
             SpecialistResult with metrics and content
         """
         run_id = parent_run_id or generate_nanoid()
+        # Specialists sharing the orchestrator's run_id (one dream trace) each get a
+        # distinct span_id so their CloudEvents trace resource ids don't collide;
+        # trace_id stays run_id so Langfuse still groups them (keyed by agent_type).
+        span_id = generate_nanoid() if parent_run_id is not None else run_id
         task_name = f"dreamer_{self.name}_{run_id}"
         start_time = time.perf_counter()
 
@@ -292,6 +296,11 @@ async def run(
                     parent_category="dream",
                     agent_type=self.name,
                     run_id=run_id,
+                    # Root span per specialist run (distinct span_id, see above).
+                    # parent_span_id stays None for now; wiring specialists as
+                    # children of a dream-level trace is forking (out of scope).
+                    trace_id=run_id,
+                    span_id=span_id,
                     observer=observer,
                     observed=observed,
                     track_name=f"Dreamer/{self.name}",

src/embedding_client.py

@@ -9,6 +9,7 @@
 import tiktoken
 from google import genai
 from google.genai import types as genai_types
+from nanoid import generate as generate_nanoid
 from openai import AsyncOpenAI
 
 from .config import EmbeddingModelConfig, resolve_embedding_model_config, settings
@@ -88,6 +89,7 @@ def _publish_embedding_event(
             get_embedding_call_purpose,
             get_embedding_parent_category,
             get_embedding_run_id,
+            get_embedding_session_id,
             get_embedding_workspace_name,
         )
 
@@ -121,6 +123,30 @@ def _publish_embedding_event(
                 run_id=get_embedding_run_id(),
             )
         )
+
+        # Trace stream (ground-truth) — gated on payload tracing. Each embedding
+        # gets its own span nested under the driving agent run (parent_span_id =
+        # run_id), so multiple embeddings in one run don't share a span id.
+        if settings.TELEMETRY.TRACE_PAYLOADS_ENABLED:
+            from src.telemetry.events import EmbeddingCallTracedEvent, emit_trace
+
+            run_id = get_embedding_run_id()
+            span_id = generate_nanoid()
+            emit_trace(
+                EmbeddingCallTracedEvent(
+                    trace_id=run_id or span_id,
+                    span_id=span_id,
+                    parent_span_id=run_id,
+                    session_id=get_embedding_session_id(),
+                    call_purpose=purpose_slug,
+                    parent_category=get_embedding_parent_category(),
+                    provider=provider,
+                    model=model,
+                    provider_input_tokens=input_tokens_estimate,
+                    provider_output_tokens=0,
+                    input_count=input_count,
+                )
+            )
     except Exception:  # pragma: no cover - telemetry must not raise
         logger.debug("Failed to emit EmbeddingCallCompletedEvent", exc_info=True)
 

src/llm/backend.py

@@ -14,7 +14,8 @@ class ToolCallResult:
     id: str
     name: str
     input: dict[str, Any]
-    thought_signature: str | None = None
+    # Gemini returns this as raw bytes; other providers omit it.
+    thought_signature: str | bytes | None = None
 
 
 @dataclass(slots=True)