From b9778b85dfd63a60e188af42ec2fbc97c1fbf8cd Mon Sep 17 00:00:00 2001 From: Shay Date: Mon, 18 May 2026 18:04:57 -0700 Subject: [PATCH] =?UTF-8?q?Phase=201=20=E2=80=94=20bridge=20trace=20instru?= =?UTF-8?q?mentation=20(observation-only)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 292 ++++++++++++++++++++++++++++++++++++++ generate/intent_bridge.py | 132 ++++++++++++++++- 2 files changed, 423 insertions(+), 1 deletion(-) create mode 100644 generate/bridge_trace.py diff --git a/generate/bridge_trace.py b/generate/bridge_trace.py new file mode 100644 index 00000000..eebae7c7 --- /dev/null +++ b/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 ("", ""), + "post_ground_obj_pending": record.post_ground_obj in ("", ""), + "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 . + + 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", +] diff --git a/generate/intent_bridge.py b/generate/intent_bridge.py index 1f09c61c..73804b30 100644 --- a/generate/intent_bridge.py +++ b/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 = "" _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