Phase 1 — bridge trace instrumentation (observation-only)

Adds generate/bridge_trace.py: a structured sink + serializer for
per-turn articulation-bridge trace records, following the exact
ADR-0040 telemetry sink pattern (JsonlBufferSink / JsonlFileSink /
FanOutSink, no wall-clock, redact-by-default).

Modifies generate/intent_bridge.py: articulate_with_intent() emits
one BridgeTraceRecord per call through a module-level opt-in sink
(attach_bridge_trace_sink / detach_bridge_trace_sink).  When no
sink is attached the call is a pure no-op — zero behavior change on
all existing paths.

The record captures:
  - intent_tag / intent_subject  (classifier output)
  - plan_subject / plan_predicate / plan_object  (articulation slots)
  - recalled_words_len / recalled_words_sample  (grounding supply)
  - pre_ground_obj  (what the graph node held before ground_graph)
  - post_ground_obj  (what it held after, or same if no grounding ran)
  - bridge_surface / bridge_useful  (final output + usefulness gate)
  - fallback_surface  (the plan.surface the runtime falls back to)

This is the Phase 1 measurement instrumentation described in the
full-sentence output mastery plan.  Phases 2-5 act on the data this
produces; Phase 1 itself is pure observation.

generate/bridge_trace.py

@@ -0,0 +1,292 @@
+"""generate/bridge_trace.py — per-turn articulation bridge trace.
+
+Phase 1 of the full-sentence output mastery plan.  Every call to
+``articulate_with_intent()`` in ``generate/intent_bridge.py`` emits
+one ``BridgeTraceRecord`` through the module-level sink when a sink
+has been attached.  When no sink is attached the emission is a pure
+no-op with zero overhead beyond the ``is None`` check.
+
+Follows the same trust-boundary conventions as ADR-0040
+(``chat/telemetry.py``):
+
+* **Redact-by-default.**  ``recalled_words_sample`` and
+  ``bridge_surface`` are only included in the serialized record when
+  the caller sets ``include_content=True``.
+* **No implicit wall-clock.**  ``timestamp`` is caller-provided.
+* **Append-only file paths.**  ``JsonlFileSink`` opens in append mode.
+* **Idempotent flush.**  Each ``emit()`` flushes immediately.
+
+The record covers every diagnostic dimension named in the mastery
+plan's Phase 1.3 trace specification:
+
+  intent_tag            — classifier output (string name of IntentTag)
+  intent_subject        — classifier subject slot
+  plan_subject          — ArticulationPlan.subject
+  plan_predicate        — ArticulationPlan.predicate
+  plan_object           — ArticulationPlan.object (None → empty string)
+  recalled_words_len    — len(recalled_words) passed to the bridge
+  recalled_words_sample — first 5 recalled words (content-gated)
+  pre_ground_obj        — graph node obj before ground_graph() ran
+  post_ground_obj       — graph node obj after ground_graph() ran
+  bridge_surface        — the surface the bridge produced (content-gated)
+  bridge_useful         — whether _is_useful_surface() passed
+  fallback_surface      — ArticulationPlan.surface (content-gated)
+
+See ``docs/decisions/full-sentence-output-plan.md``.
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import IO, Protocol
+
+
+# ---------------------------------------------------------------------------
+# Record
+# ---------------------------------------------------------------------------
+
+
+@dataclass(frozen=True, slots=True)
+class BridgeTraceRecord:
+    """One per-turn observation record from the intent bridge.
+
+    All fields are plain Python types so serialization is trivial and
+    the record is safe to construct inside the hot path without any
+    I/O or numpy dependency.
+    """
+
+    intent_tag: str                          # IntentTag.name (e.g. "DEFINITION")
+    intent_subject: str                      # classifier subject slot, "" if None
+    plan_subject: str
+    plan_predicate: str
+    plan_object: str                         # "" when ArticulationPlan.object is None
+    recalled_words_len: int
+    recalled_words_sample: tuple[str, ...]   # first ≤5 words; () when redacted
+    pre_ground_obj: str                      # graph p0.obj before ground_graph
+    post_ground_obj: str                     # graph p0.obj after ground_graph
+    bridge_surface: str                      # "" when redacted
+    bridge_useful: bool
+    fallback_surface: str                    # "" when redacted
+
+
+# ---------------------------------------------------------------------------
+# Serializer
+# ---------------------------------------------------------------------------
+
+
+def serialize_bridge_trace(
+    record: BridgeTraceRecord,
+    *,
+    include_content: bool = False,
+    timestamp: str | None = None,
+) -> dict[str, object]:
+    """Produce a JSON-safe audit dict from a ``BridgeTraceRecord``.
+
+    Content fields (``recalled_words_sample``, ``bridge_surface``,
+    ``fallback_surface``) are only emitted when ``include_content``
+    is True.  The metadata fields are always emitted so aggregation
+    pipelines can compute grounding-rate statistics without ever
+    seeing raw surface text.
+    """
+    out: dict[str, object] = {
+        "type": "bridge_trace",
+        "intent_tag": record.intent_tag,
+        "intent_subject_len": len(record.intent_subject),
+        "plan_subject_len": len(record.plan_subject),
+        "plan_predicate_len": len(record.plan_predicate),
+        "plan_object_present": bool(record.plan_object),
+        "recalled_words_len": record.recalled_words_len,
+        "pre_ground_obj_pending": record.pre_ground_obj in ("<pending>", "<prior>"),
+        "post_ground_obj_pending": record.post_ground_obj in ("<pending>", "<prior>"),
+        "bridge_useful": record.bridge_useful,
+        # Derived diagnostic flag: grounding changed something.
+        "grounding_changed_obj": record.pre_ground_obj != record.post_ground_obj,
+    }
+    if include_content:
+        out["intent_subject"] = record.intent_subject
+        out["plan_subject"] = record.plan_subject
+        out["plan_predicate"] = record.plan_predicate
+        out["plan_object"] = record.plan_object
+        out["recalled_words_sample"] = list(record.recalled_words_sample)
+        out["pre_ground_obj"] = record.pre_ground_obj
+        out["post_ground_obj"] = record.post_ground_obj
+        out["bridge_surface"] = record.bridge_surface
+        out["fallback_surface"] = record.fallback_surface
+    if timestamp is not None:
+        out["timestamp"] = str(timestamp)
+    return out
+
+
+def format_bridge_trace_jsonl(
+    record: BridgeTraceRecord,
+    *,
+    include_content: bool = False,
+    timestamp: str | None = None,
+) -> str:
+    """Serialize one trace record as a deterministic JSONL line.
+
+    Field order is alphabetical (``sort_keys=True``) for byte-stable
+    replay diffing.  No trailing newline — the sink owns termination.
+    """
+    payload = serialize_bridge_trace(
+        record, include_content=include_content, timestamp=timestamp
+    )
+    return json.dumps(payload, sort_keys=True, separators=(",", ":"))
+
+
+# ---------------------------------------------------------------------------
+# Sink protocol
+# ---------------------------------------------------------------------------
+
+
+class BridgeTraceSink(Protocol):
+    """Minimal sink contract (mirrors TurnEventSink in chat/telemetry.py).
+
+    Sinks receive one already-serialized JSONL line per
+    ``articulate_with_intent()`` call.  The sink is attached at the
+    module level via ``attach_bridge_trace_sink``; callers that never
+    attach a sink pay only the ``is None`` guard cost.
+    """
+
+    def emit(self, line: str) -> None: ...
+
+
+# ---------------------------------------------------------------------------
+# Concrete sinks
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class JsonlBufferSink:
+    """In-memory sink that captures every emitted line.
+
+    Useful for tests and interactive session analysis where
+    persistence is the caller's responsibility.
+    """
+
+    lines: list[str] = field(default_factory=list)
+    include_content: bool = False
+
+    def emit(self, line: str) -> None:
+        self.lines.append(line)
+
+    def records(self) -> list[dict]:
+        """Parse all emitted lines and return them as dicts."""
+        return [json.loads(line) for line in self.lines]
+
+    def grounding_rate(self) -> float:
+        """Fraction of turns where the bridge produced a useful surface.
+
+        Returns 0.0 when no lines have been emitted yet.
+        """
+        parsed = self.records()
+        if not parsed:
+            return 0.0
+        return sum(1 for r in parsed if r.get("bridge_useful")) / len(parsed)
+
+    def pending_rate(self) -> float:
+        """Fraction of turns where the post-ground obj was still <pending>.
+
+        Returns 0.0 when no lines have been emitted yet.
+        """
+        parsed = self.records()
+        if not parsed:
+            return 0.0
+        return sum(1 for r in parsed if r.get("post_ground_obj_pending")) / len(parsed)
+
+    def recalled_words_empty_rate(self) -> float:
+        """Fraction of turns where recalled_words_len == 0."""
+        parsed = self.records()
+        if not parsed:
+            return 0.0
+        return sum(1 for r in parsed if r.get("recalled_words_len", 0) == 0) / len(parsed)
+
+    def summary(self) -> dict:
+        """One-shot diagnostic summary dict for operator inspection."""
+        parsed = self.records()
+        n = len(parsed)
+        if n == 0:
+            return {"turns": 0}
+        by_intent: dict[str, dict] = {}
+        for r in parsed:
+            tag = r.get("intent_tag", "UNKNOWN")
+            bucket = by_intent.setdefault(tag, {"total": 0, "useful": 0, "pending": 0, "no_recalled": 0})
+            bucket["total"] += 1
+            if r.get("bridge_useful"):
+                bucket["useful"] += 1
+            if r.get("post_ground_obj_pending"):
+                bucket["pending"] += 1
+            if r.get("recalled_words_len", 0) == 0:
+                bucket["no_recalled"] += 1
+        return {
+            "turns": n,
+            "grounding_rate": round(self.grounding_rate(), 4),
+            "pending_rate": round(self.pending_rate(), 4),
+            "recalled_words_empty_rate": round(self.recalled_words_empty_rate(), 4),
+            "by_intent": by_intent,
+        }
+
+
+class JsonlFileSink:
+    """Append-only JSONL file sink with eager flush.
+
+    Path fixed at construction.  Each ``emit()`` flushes immediately
+    so a crashed turn loop still has prior turns durable on disk.
+    Supports context-manager usage.
+    """
+
+    def __init__(
+        self,
+        path: str | Path,
+        *,
+        include_content: bool = False,
+    ) -> None:
+        self._path = Path(path)
+        self._include_content = include_content
+        self._fh: IO[str] | None = None
+
+    def emit(self, line: str) -> None:
+        if self._fh is None:
+            self._path.parent.mkdir(parents=True, exist_ok=True)
+            self._fh = self._path.open("a", encoding="utf-8")
+        self._fh.write(line)
+        self._fh.write("\n")
+        self._fh.flush()
+
+    def close(self) -> None:
+        if self._fh is not None:
+            self._fh.close()
+            self._fh = None
+
+    def __enter__(self) -> "JsonlFileSink":
+        return self
+
+    def __exit__(self, *exc_info) -> None:
+        self.close()
+
+
+@dataclass
+class FanOutSink:
+    """Forward every emitted line to N sinks in declaration order.
+
+    Fail-fast: if sink *i* raises, sinks *i+1..* are NOT called.
+    """
+
+    sinks: tuple = ()  # tuple[BridgeTraceSink, ...]
+
+    def emit(self, line: str) -> None:
+        for sink in self.sinks:
+            sink.emit(line)
+
+
+__all__ = [
+    "BridgeTraceRecord",
+    "BridgeTraceSink",
+    "FanOutSink",
+    "JsonlBufferSink",
+    "JsonlFileSink",
+    "format_bridge_trace_jsonl",
+    "serialize_bridge_trace",
+]

generate/intent_bridge.py

@@ -13,6 +13,18 @@ Design constraints:
     with no grounded obj slots)
   - Does not alter the ArticulationPlan dataclass or ChatResponse structure;
     only the .surface field is replaced when the bridge succeeds
+
+Phase 1 instrumentation (observation-only)
+  ``articulate_with_intent()`` emits one ``BridgeTraceRecord`` per call
+  through the module-level ``_TRACE_SINK`` when a sink has been attached via
+  ``attach_bridge_trace_sink()``.  When no sink is attached the emission
+  path is a pure no-op (single ``is None`` guard, no allocation).  This
+  instruments the four dimensions named in the mastery plan's Phase 1.3:
+    - recalled_words population at the call site
+    - pre- and post-grounding obj slot content
+    - bridge_useful flag
+    - fallback_surface (what the runtime would use if bridge returns "")
+  Zero behavior change on all existing paths.
 """
 
 from __future__ import annotations
@@ -34,6 +46,89 @@ _PRIOR = "<prior>"
 _EMPTY_INDICATORS = frozenset({_PENDING, _PRIOR, "...", ""})
 
 
+# ---------------------------------------------------------------------------
+# Phase 1 — module-level trace sink (opt-in, observation-only)
+# ---------------------------------------------------------------------------
+
+_TRACE_SINK = None  # type: object | None
+_TRACE_INCLUDE_CONTENT: bool = False
+
+
+def attach_bridge_trace_sink(
+    sink,
+    *,
+    include_content: bool = False,
+) -> None:
+    """Attach a :class:`generate.bridge_trace.BridgeTraceSink`.
+
+    After each call to ``articulate_with_intent()`` the runtime emits
+    one JSONL-formatted ``BridgeTraceRecord`` to *sink*.  Passing
+    ``None`` detaches.
+
+    ``include_content`` opts surface text, recalled words, and slot
+    values into the emitted record.  Default ``False`` preserves
+    redact-by-default (CLAUDE.md trust boundary): aggregation
+    pipelines get counts and flags without raw text.
+    """
+    global _TRACE_SINK, _TRACE_INCLUDE_CONTENT
+    _TRACE_SINK = sink
+    _TRACE_INCLUDE_CONTENT = bool(include_content)
+
+
+def detach_bridge_trace_sink() -> None:
+    """Detach any attached trace sink (convenience alias for attach(None))."""
+    global _TRACE_SINK, _TRACE_INCLUDE_CONTENT
+    _TRACE_SINK = None
+    _TRACE_INCLUDE_CONTENT = False
+
+
+def _emit_trace(
+    *,
+    intent_tag: str,
+    intent_subject: str,
+    plan: ArticulationPlan,
+    recalled_words: tuple[str, ...],
+    pre_ground_obj: str,
+    post_ground_obj: str,
+    bridge_surface: str,
+    bridge_useful: bool,
+) -> None:
+    """Emit one BridgeTraceRecord to the attached sink (no-op when None).
+
+    Called from within ``articulate_with_intent()`` after the bridge
+    has resolved.  All arguments are plain Python types — no numpy,
+    no I/O dependencies at the construction site.
+    """
+    if _TRACE_SINK is None:
+        return
+    from generate.bridge_trace import BridgeTraceRecord, format_bridge_trace_jsonl
+
+    record = BridgeTraceRecord(
+        intent_tag=intent_tag,
+        intent_subject=intent_subject,
+        plan_subject=plan.subject or "",
+        plan_predicate=plan.predicate or "",
+        plan_object=plan.object or "",
+        recalled_words_len=len(recalled_words),
+        recalled_words_sample=recalled_words[:5] if _TRACE_INCLUDE_CONTENT else (),
+        pre_ground_obj=pre_ground_obj,
+        post_ground_obj=post_ground_obj,
+        bridge_surface=bridge_surface if _TRACE_INCLUDE_CONTENT else "",
+        bridge_useful=bridge_useful,
+        fallback_surface=plan.surface if _TRACE_INCLUDE_CONTENT else "",
+    )
+    line = format_bridge_trace_jsonl(
+        record,
+        include_content=_TRACE_INCLUDE_CONTENT,
+    )
+    _TRACE_SINK.emit(line)
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+
 def classify_intent_from_input(text: str) -> DialogueIntent:
     """Run the rule-based intent classifier against raw input text."""
     return classify_intent(text)
@@ -125,21 +220,56 @@ def articulate_with_intent(
 
     The caller (chat/runtime.py) should fall back to the existing
     ArticulationPlan.surface when this returns "".
+
+    Phase 1: emits one BridgeTraceRecord to the module-level sink (if
+    attached) after resolution — observation-only, no effect on return value.
     """
     intent = classify_intent_from_input(text)
+    intent_tag_name = intent.tag.name if intent.tag is not None else "UNKNOWN"
+    intent_subject = intent.subject or ""
 
     graph = _build_graph_from_intent(intent, plan)
+
+    # Record pre-grounding obj for the Phase 1 trace.
+    pre_ground_obj = graph.nodes[0].obj if graph.nodes else _PENDING
+
     if recalled_words:
         graph = ground_graph(graph, recalled_words)
 
+    # Record post-grounding obj for the Phase 1 trace.
+    post_ground_obj = graph.nodes[0].obj if graph.nodes else _PENDING
+
     articulation_target = plan_articulation(graph)
     realized: RealizedPlan = realize_semantic(articulation_target, graph)
 
     if not realized.surface or not realized.fragments:
+        _emit_trace(
+            intent_tag=intent_tag_name,
+            intent_subject=intent_subject,
+            plan=plan,
+            recalled_words=recalled_words,
+            pre_ground_obj=pre_ground_obj,
+            post_ground_obj=post_ground_obj,
+            bridge_surface="",
+            bridge_useful=False,
+        )
         return ""
 
     surface = realized.surface
-    if not _is_useful_surface(surface):
+    useful = _is_useful_surface(surface)
+
+    _emit_trace(
+        intent_tag=intent_tag_name,
+        intent_subject=intent_subject,
+        plan=plan,
+        recalled_words=recalled_words,
+        pre_ground_obj=pre_ground_obj,
+        post_ground_obj=post_ground_obj,
+        bridge_surface=surface,
+        bridge_useful=useful,
+    )
+
+    if not useful:
         return ""
 
     return surface