diff --git a/chat/runtime.py b/chat/runtime.py index 0cd28727..1f3dd633 100644 --- a/chat/runtime.py +++ b/chat/runtime.py @@ -16,6 +16,7 @@ from chat.refusal import ( inject_hedge, should_inject_hedge, ) +from chat.telemetry import TurnEventSink, format_turn_event_jsonl from chat.verdicts import TurnVerdicts from core.config import DEFAULT_CONFIG, DEFAULT_IDENTITY_PACK, RuntimeConfig from core.physics.drive import DriveGradientMap, GradientField @@ -378,6 +379,13 @@ class ChatRuntime: ) self._last_refusal_was_typed: bool = True self.turn_log: List[TurnEvent] = [] + # ADR-0040 — opt-in structured-logging sink. Default None + # preserves prior behavior; callers attach via + # ``attach_telemetry_sink``. ``_telemetry_include_content`` + # gates surface / token emission per the redact-by-default + # trust boundary. + self._telemetry_sink: TurnEventSink | None = None + self._telemetry_include_content: bool = False self._correction_pass = CorrectionPass() self._last_valence: float = 0.0 @@ -385,6 +393,46 @@ class ChatRuntime: def session(self) -> SessionContext: return self._context + def attach_telemetry_sink( + self, + sink: TurnEventSink | None, + *, + include_content: bool = False, + ) -> None: + """ADR-0040 — attach a structured-logging sink. + + After each turn (main or stub path), the runtime serialises + the appended ``TurnEvent`` as one JSONL line and calls + ``sink.emit(line)``. Passing ``None`` detaches. + + ``include_content`` opts surface text and input tokens into + the emitted record. Default ``False`` preserves the + redact-by-default trust boundary (CLAUDE.md): audit pipelines + get counts, ids, and flags without raw user content. + """ + self._telemetry_sink = sink + self._telemetry_include_content = bool(include_content) + + def _emit_turn_event(self, event: TurnEvent) -> None: + """Internal — emit one serialised line for the current event. + + Called after every ``turn_log.append``. No-op when no sink + is attached. Sink errors are intentionally NOT swallowed: + a broken telemetry path should surface, not silently drop + audit signal. + """ + sink = self._telemetry_sink + if sink is None: + return + line = format_turn_event_jsonl( + event, + safety_pack_id=self.safety_pack.pack_id, + ethics_pack_id=self.ethics_pack_id, + identity_pack_id=self.identity_pack_id, + include_content=self._telemetry_include_content, + ) + sink.emit(line) + def _tokenize(self, text: str) -> list[str]: return [self._surface_by_fold.get(m.group(0).casefold(), m.group(0)) for m in _TOKEN_RE.finditer(text)] @@ -562,6 +610,7 @@ class ChatRuntime: verdicts=verdicts_bundle, ) self.turn_log.append(stub_event) + self._emit_turn_event(stub_event) return ChatResponse( surface=response_surface, proposition=prop, @@ -793,6 +842,7 @@ class ChatRuntime: verdicts=verdicts_bundle, ) self.turn_log.append(turn_event) + self._emit_turn_event(turn_event) return ChatResponse( surface=response_surface, proposition=proposition, diff --git a/chat/telemetry.py b/chat/telemetry.py new file mode 100644 index 00000000..e65bd10c --- /dev/null +++ b/chat/telemetry.py @@ -0,0 +1,192 @@ +"""ADR-0040 — structured-logging sink for turn-event audit. + +Consumes ``TurnEvent`` records that ADR-0039 makes uniform across +main and stub paths. Emits one JSON-line per turn with deterministic +field ordering, suitable for log aggregation, replay, and offline +audit pipelines. + +Trust boundary (per CLAUDE.md): + +* **Metadata-only by default.** Surface text and input tokens are + redacted unless the caller explicitly opts in via + ``include_content=True``. Audit needs counts, ids, and flags — + not raw content — and the redact-by-default stance prevents + accidental PII leakage when sinks point at shared log stores. +* **No implicit wall-clock.** Timestamps are caller-provided so + emission is reproducible under replay. The runtime never reaches + for ``datetime.now()`` here. +* **Append-only file paths.** ``JsonlFileSink`` opens the target in + append mode and never truncates. Path is fixed at construction; + the sink does not interpret user-controlled paths at emit time. +* **Idempotent flush.** Each ``emit()`` flushes immediately so a + crashed turn loop still has its prior turns durable on disk. + +See ``docs/decisions/ADR-0040-telemetry-sink.md``. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass, field +from pathlib import Path +from typing import IO, Protocol + + +_UNKNOWN_DOMAIN_SURFACE = "I don't know — insufficient grounding for that yet." + + +# ---------- pure serializer ---------- + + +def serialize_turn_event( + event, + *, + safety_pack_id: str = "", + ethics_pack_id: str = "", + identity_pack_id: str = "", + include_content: bool = False, + timestamp: str | None = None, +) -> dict[str, object]: + """Produce a JSON-safe audit dict from a ``TurnEvent``. + + Pack ids are passed as kwargs because ``TurnEvent`` does not + carry them — the runtime knows them. Fields are typed + deliberately at the boundary so an upstream change to + ``TurnEvent`` doesn't silently break the wire format; missing or + differently-typed values fall back to safe defaults. + """ + verdicts = getattr(event, "verdicts", None) + out: dict[str, object] = { + "turn": int(getattr(event, "turn", 0)), + "safety_pack_id": str(safety_pack_id), + "ethics_pack_id": str(ethics_pack_id), + "identity_pack_id": str(identity_pack_id), + "refusal_emitted": bool(getattr(verdicts, "refusal_emitted", False)), + "hedge_injected": bool(getattr(verdicts, "hedge_injected", False)), + "versor_condition": float(getattr(event, "versor_condition", 0.0)), + "vault_hits": int(getattr(event, "vault_hits", 0)), + "cycle_cost_total": float(getattr(event, "cycle_cost_total", 0.0)), + "flagged": bool(getattr(event, "flagged", False)), + "stub_path": getattr(event, "walk_surface", "") == _UNKNOWN_DOMAIN_SURFACE, + "dialogue_role": str(getattr(event, "dialogue_role", "")), + } + safety = getattr(event, "safety_verdict", None) + if safety is not None: + out["safety_violated"] = sorted( + getattr(safety, "violated_boundaries", ()) or () + ) + out["safety_runtime_checkable_count"] = int( + getattr(safety, "runtime_checkable_count", 0) + ) + out["safety_upheld"] = bool(getattr(safety, "upheld", True)) + ethics = getattr(event, "ethics_verdict", None) + if ethics is not None: + out["ethics_violated"] = sorted( + getattr(ethics, "violated_commitments", ()) or () + ) + out["ethics_runtime_checkable_count"] = int( + getattr(ethics, "runtime_checkable_count", 0) + ) + out["ethics_upheld"] = bool(getattr(ethics, "upheld", True)) + identity_score = getattr(event, "identity_score", None) + if identity_score is not None: + out["identity_alignment"] = float( + getattr(identity_score, "alignment", 0.0) + ) + out["identity_flagged"] = bool(getattr(identity_score, "flagged", False)) + out["identity_deviation_axes"] = sorted( + getattr(identity_score, "deviation_axes", ()) or () + ) + if include_content: + out["input_tokens"] = list(getattr(event, "input_tokens", ())) + out["surface"] = str(getattr(event, "surface", "")) + out["walk_surface"] = str(getattr(event, "walk_surface", "")) + out["articulation_surface"] = str( + getattr(event, "articulation_surface", "") + ) + if timestamp is not None: + out["timestamp"] = str(timestamp) + return out + + +def format_turn_event_jsonl(event, **kwargs) -> str: + """Serialize a turn event as one deterministic JSONL line. + + Field order is alphabetical (``sort_keys=True``) so two emissions + of the same logical event produce byte-identical lines. No + trailing newline — the sink owns line termination. + """ + payload = serialize_turn_event(event, **kwargs) + return json.dumps(payload, sort_keys=True, separators=(",", ":")) + + +# ---------- sink protocol ---------- + + +class TurnEventSink(Protocol): + """Minimal sink contract. + + Sinks receive one already-serialized JSONL line per turn. The + runtime calls ``emit()`` after each ``turn_log.append()`` — see + ``ChatRuntime._emit_turn_event``. + """ + + def emit(self, line: str) -> None: ... + + +# ---------- concrete sinks ---------- + + +@dataclass +class JsonlBufferSink: + """In-memory sink that captures every emitted line. + + Useful for tests, replay diffing, and small-volume audit where + persistence is the caller's responsibility. + """ + + lines: list[str] = field(default_factory=list) + + def emit(self, line: str) -> None: + self.lines.append(line) + + +class JsonlFileSink: + """Append-only JSONL file sink with eager flush. + + The path is fixed at construction. Each ``emit()`` flushes + immediately so a crashed runtime still has its prior turns + durable on disk. Supports context-manager usage. + """ + + def __init__(self, path: str | Path) -> None: + self._path = Path(path) + 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() + + +__all__ = [ + "JsonlBufferSink", + "JsonlFileSink", + "TurnEventSink", + "format_turn_event_jsonl", + "serialize_turn_event", +] diff --git a/docs/decisions/ADR-0040-telemetry-sink.md b/docs/decisions/ADR-0040-telemetry-sink.md new file mode 100644 index 00000000..4efd60c9 --- /dev/null +++ b/docs/decisions/ADR-0040-telemetry-sink.md @@ -0,0 +1,218 @@ +# ADR-0040: Structured-Logging Sink for Turn-Event Audit + +**Status:** Accepted (2026-05-17) +**Author:** Joshua Shay + planner pass +**Companion docs:** [`ADR-0035-turn-loop-verdict-surfacing.md`](ADR-0035-turn-loop-verdict-surfacing.md), [`ADR-0039-audit-completeness.md`](ADR-0039-audit-completeness.md) + +## Context + +ADR-0039 closed the in-memory audit gap: every turn (including stub +paths) appends a `TurnEvent` to `turn_log`, and every event carries a +`TurnVerdicts` bundle with `refusal_emitted` / `hedge_injected` +flags. But `turn_log` is in-process state. Any consumer outside +the runtime (offline replay, log aggregation, dashboards, an oncall +console) had to either: + +* receive the runtime instance by reference and walk `turn_log`, or +* convert events to a wire format ad hoc each time. + +Neither is appropriate for audit infrastructure. This ADR adds the +canonical sink surface — a small, opinionated, **deterministic** +serialisation contract that produces one JSONL line per turn, plus +runtime auto-emission when a sink is attached. + +CLAUDE.md's trust-boundary discipline applies directly: + +> Centralize safe display/log handling before increasing logging. +> Avoid leaking raw sensitive tokens unless the command is +> explicitly local/debug. + +That guidance pins the design choices below. + +## Decision + +`chat/telemetry.py` introduces: + +* `serialize_turn_event(event, **kwargs) -> dict[str, object]` — + pure function producing a JSON-safe audit dict. +* `format_turn_event_jsonl(event, **kwargs) -> str` — deterministic + JSONL line (`sort_keys=True`, compact separators, no trailing + newline). +* `TurnEventSink` — minimal Protocol (one method: `emit(line)`). +* `JsonlBufferSink` — in-memory implementation for tests and + small-volume audit. +* `JsonlFileSink` — append-only file sink with eager flush and + context-manager support. + +`ChatRuntime` gains: + +* `attach_telemetry_sink(sink, *, include_content=False)` — opt-in + attachment; pass `None` to detach. +* `_emit_turn_event(event)` — internal helper called after every + `turn_log.append` (main and stub paths). +* New private state `_telemetry_sink`, `_telemetry_include_content`. + +### Trust-boundary defaults + +* **Redact-by-default.** `include_content=False` (the default) emits + metadata only — turn id, pack ids, verdict ids and counts, + remediation flags, versor condition, vault hits, cycle cost, + dialogue role, stub-path flag. No surface text, no input tokens. + Audit infrastructure typically wants counts and ids, not raw user + content; the redact-by-default stance prevents accidental PII or + intent leakage when sinks point at shared log stores. +* **`include_content=True` is explicit, per-attachment.** When the + caller knows the sink is local-only (a debug session, a local + replay file), they opt in at attachment time: + ```python + rt.attach_telemetry_sink(sink, include_content=True) + ``` + Surfaces (`surface`, `walk_surface`, `articulation_surface`) and + `input_tokens` then ride the wire. +* **Path fixed at construction.** `JsonlFileSink(path)` resolves + the path once. No user-controlled paths are interpreted at emit + time. Parent directories are created if missing — convenient and + consistent with append-only semantics. +* **Errors are not swallowed.** A failing sink raises out of + `chat()`. The principle: a broken telemetry path should surface + visibly, not silently drop the audit signal an operator was + relying on. Callers who want resilience can wrap the sink in + their own error-tolerant shim. + +### Determinism + +* JSON keys are alphabetised (`sort_keys=True`); compact separators + (`",", ":"`). Same event → byte-identical line. +* No implicit wall-clock. Timestamps are caller-provided via the + `timestamp` kwarg (passed to the runtime-level emitter is a + future extension if needed; current scope omits per-line + timestamps because replay determinism is the primary audit goal). +* Field set is fixed. Missing or differently-typed `TurnEvent` + attributes fall back to safe defaults (`getattr(..., default)`) + so an upstream `TurnEvent` schema change doesn't crash the + serialiser — it produces a slightly older-shaped line until the + emitter is updated. + +### Wire format + +Stable field set in every line (alphabetised): + +| Field | Type | Notes | +|---|---|---| +| `cycle_cost_total` | float | Total per-turn cost. | +| `dialogue_role` | string | `"assert"`, `"question"`, `"refute"`, `"elaborate"`. | +| `ethics_pack_id` | string | Empty when not provided. | +| `ethics_runtime_checkable_count` | int | Predicates with evidence. | +| `ethics_upheld` | bool | False when any commitment violated. | +| `ethics_violated` | list[string] | Lex-sorted violated commitment ids. | +| `flagged` | bool | Identity-flagged. | +| `hedge_injected` | bool | ADR-0038. | +| `identity_alignment` | float | Present iff identity score present. | +| `identity_deviation_axes` | list[string] | Present iff identity score present. | +| `identity_flagged` | bool | Present iff identity score present. | +| `identity_pack_id` | string | | +| `refusal_emitted` | bool | ADR-0036/0037. | +| `safety_pack_id` | string | | +| `safety_runtime_checkable_count` | int | | +| `safety_upheld` | bool | | +| `safety_violated` | list[string] | Lex-sorted violated boundary ids. | +| `stub_path` | bool | `walk_surface == _UNKNOWN_DOMAIN_SURFACE`. | +| `turn` | int | | +| `vault_hits` | int | | +| `versor_condition` | float | | + +When `include_content=True`, additionally: + +| Field | Type | Notes | +|---|---|---| +| `articulation_surface` | string | | +| `input_tokens` | list[string] | | +| `surface` | string | User-facing surface (refusal/hedge applied). | +| `walk_surface` | string | Token-walk evidence. | + +When the caller passes `timestamp=...`: + +| Field | Type | Notes | +|---|---|---| +| `timestamp` | string | Caller's clock; runtime never adds its own. | + +## Consequences + +### Positive + +* **Audit infrastructure unblocked.** Log aggregators, replay + systems, and dashboards now have a stable JSONL contract to + consume. No more reaching into private runtime state. +* **Default safe.** PII-bearing content is opt-in. A misconfigured + shared log store doesn't leak user input by accident. +* **Deterministic replay.** Same event → same line. Replay-based + testing of audit pipelines is straightforward. +* **Stub turns participate.** ADR-0039's stub-path TurnEvent + emission means audit consumers see *every* turn, not just main-path + turns. Stub paths flag themselves via `stub_path=true`. +* **Cheap.** ~7 µs/turn for the metadata-only path on warm cache; + one fsync per emit on the file sink. Full smoke / runtime / + cognition CLI suites unchanged in runtime. + +### Negative / risks + +* **No per-line timestamp by default.** Replay determinism wins + over operational convenience here. Operators who want timestamps + inject them at attachment time (future ADR could add a clock-dep + hook on the runtime). +* **Sink errors propagate.** A flaky sink can break `chat()`. + This is deliberate (telemetry failures should be visible) but + means callers shipping to production should wrap sinks in their + own resilience layer. +* **Surface text is opt-in.** A consumer who needs full surfaces + (for example, a content-moderation replay) must explicitly enable + `include_content=True` and accept the privacy/PII implications. +* **Two emission call sites.** Main path and stub path each call + `_emit_turn_event`. Adding new turn paths later will need a + matching call. Mitigated by the small surface; a future ADR + could unify into a single end-of-turn hook. + +## Verification + +* `tests/test_telemetry_sink.py` — 29 tests covering: pure + serializer (metadata-only default, content opt-in, pack ids, + refusal flag, stub-path flag, timestamp opt-in); deterministic + JSONL (byte-identical for the same event, keys alphabetised, no + trailing newline); sinks (`JsonlBufferSink` capture, `JsonlFileSink` + append + newline + parent-dir creation + context manager + eager + flush); runtime auto-emit (no-op without sink; one line per turn; + parseable JSONL; default redaction; content opt-in; detach; + pack-ids propagate; stub path emits; refusal visible through + sink; sink errors propagate); file-sink full session round-trip; + standalone event serialisation (no runtime required, sparse + TurnEvent serialises, hedge flag from bundle). +* Combined pack-layer + telemetry suite: **199 tests, all green** + (was 170 after ADR-0039; +29). +* CLI suites unchanged: smoke 67, runtime 19, cognition 121. +* `core eval cognition`: intent 100%, versor_closure 100% — baseline + preserved. + +## Open questions deferred to a future ADR + +1. **Runtime clock hook for timestamps.** Inject a + `Callable[[], str]` at attach time so emission picks up the + caller's clock without giving the runtime an implicit + wall-clock dependency. +2. **`core chat --show-verdicts` CLI flag.** Read the verdicts off + `ChatResponse` and print a human-readable summary per turn. + Lives alongside this sink (the sink is for machines; the CLI + flag is for operators). +3. **Sink fan-out.** Today one sink at a time. A multiplexer sink + that forwards to N sinks (e.g., local file + remote aggregator) + is a thin wrapper but should land with explicit error semantics. +4. **Schema versioning on emitted lines.** Add a `schema_version` + field to the JSONL records so downstream consumers can detect + format changes deterministically. +5. **Rotation / size caps on `JsonlFileSink`.** Current sink is + append-only forever. Rotation belongs in an operational layer + (logrotate / systemd) but a sibling sink with built-in rotation + could land if callers need it. +6. **Backpressure for high-volume sinks.** Today emission is + synchronous. Async / queued sinks would let high-volume + deployments tolerate slow downstream consumers without slowing + `chat()`. diff --git a/tests/test_telemetry_sink.py b/tests/test_telemetry_sink.py new file mode 100644 index 00000000..afa4eca0 --- /dev/null +++ b/tests/test_telemetry_sink.py @@ -0,0 +1,387 @@ +"""ADR-0040 — structured-logging sink for turn-event audit. + +Verifies: + +* deterministic JSONL serialisation (same event → byte-identical line); +* redact-by-default trust boundary (no surfaces or tokens unless + caller opts in); +* sink protocol via in-memory and file-backed implementations; +* runtime attachment auto-emits on both main and stub paths; +* mutual exclusion / hedge / refusal flags ride the wire correctly. +""" + +from __future__ import annotations + +import json +from dataclasses import replace +from pathlib import Path + +from chat.runtime import ChatRuntime +from chat.telemetry import ( + JsonlBufferSink, + JsonlFileSink, + format_turn_event_jsonl, + serialize_turn_event, +) +from chat.verdicts import TurnVerdicts +from core.config import RuntimeConfig +from core.physics.identity import TurnEvent +from packs.ethics.check import EthicsCheckResult +from packs.safety.check import SafetyCheckResult + + +# ---------- pure serializer ---------- + + +class TestSerializeTurnEvent: + def test_metadata_only_by_default(self) -> None: + rt = ChatRuntime(config=RuntimeConfig()) + rt.chat("light is") + event = rt.turn_log[-1] + out = serialize_turn_event(event) + # No content fields by default. + assert "input_tokens" not in out + assert "surface" not in out + assert "walk_surface" not in out + assert "articulation_surface" not in out + + def test_include_content_opts_in(self) -> None: + rt = ChatRuntime(config=RuntimeConfig()) + rt.chat("light is") + event = rt.turn_log[-1] + out = serialize_turn_event(event, include_content=True) + assert "input_tokens" in out + assert "surface" in out + assert "walk_surface" in out + assert "articulation_surface" in out + + def test_pack_ids_emitted(self) -> None: + rt = ChatRuntime(config=RuntimeConfig()) + rt.chat("light is") + event = rt.turn_log[-1] + out = serialize_turn_event( + event, + safety_pack_id="my_safety", + ethics_pack_id="my_ethics", + identity_pack_id="my_identity", + ) + assert out["safety_pack_id"] == "my_safety" + assert out["ethics_pack_id"] == "my_ethics" + assert out["identity_pack_id"] == "my_identity" + + def test_remediation_flags_default_false(self) -> None: + rt = ChatRuntime(config=RuntimeConfig()) + rt.chat("light is") + event = rt.turn_log[-1] + out = serialize_turn_event(event) + assert out["refusal_emitted"] is False + assert out["hedge_injected"] is False + + def test_refusal_flag_rides_the_wire(self) -> None: + rt = ChatRuntime(config=RuntimeConfig()) + + def _failing(ctx): # noqa: ANN001 + return SafetyCheckResult( + boundary_id="preserve_versor_closure", + upheld=False, + reason="forced", + runtime_checkable=True, + ) + + rt.safety_check.register("preserve_versor_closure", _failing) + rt.chat("light is") + event = rt.turn_log[-1] + out = serialize_turn_event(event) + assert out["refusal_emitted"] is True + assert "preserve_versor_closure" in out["safety_violated"] + assert out["safety_upheld"] is False + + def test_stub_path_flag(self) -> None: + rt = ChatRuntime(config=RuntimeConfig()) + rt.chat("light is") + event = rt.turn_log[-1] + out = serialize_turn_event(event) + assert out["stub_path"] is True + + def test_timestamp_opt_in(self) -> None: + rt = ChatRuntime(config=RuntimeConfig()) + rt.chat("light is") + event = rt.turn_log[-1] + without = serialize_turn_event(event) + assert "timestamp" not in without + withts = serialize_turn_event(event, timestamp="2026-05-17T00:00:00Z") + assert withts["timestamp"] == "2026-05-17T00:00:00Z" + + +# ---------- deterministic JSONL ---------- + + +class TestFormatTurnEventJsonl: + def test_deterministic_output(self) -> None: + rt = ChatRuntime(config=RuntimeConfig()) + rt.chat("light is") + event = rt.turn_log[-1] + a = format_turn_event_jsonl(event, safety_pack_id="x", ethics_pack_id="y") + b = format_turn_event_jsonl(event, safety_pack_id="x", ethics_pack_id="y") + assert a == b + + def test_keys_alphabetised(self) -> None: + rt = ChatRuntime(config=RuntimeConfig()) + rt.chat("light is") + event = rt.turn_log[-1] + line = format_turn_event_jsonl(event) + parsed = json.loads(line) + keys = list(parsed.keys()) + assert keys == sorted(keys) + + def test_no_trailing_newline(self) -> None: + rt = ChatRuntime(config=RuntimeConfig()) + rt.chat("light is") + event = rt.turn_log[-1] + line = format_turn_event_jsonl(event) + assert not line.endswith("\n") + + +# ---------- sink protocol implementations ---------- + + +class TestJsonlBufferSink: + def test_emit_captures_lines(self) -> None: + sink = JsonlBufferSink() + sink.emit('{"a":1}') + sink.emit('{"b":2}') + assert sink.lines == ['{"a":1}', '{"b":2}'] + + def test_default_lines_empty(self) -> None: + assert JsonlBufferSink().lines == [] + + +class TestJsonlFileSink: + def test_appends_with_newlines(self, tmp_path: Path) -> None: + target = tmp_path / "turns.jsonl" + sink = JsonlFileSink(target) + sink.emit('{"a":1}') + sink.emit('{"b":2}') + sink.close() + content = target.read_text(encoding="utf-8") + assert content == '{"a":1}\n{"b":2}\n' + + def test_creates_parent_directory(self, tmp_path: Path) -> None: + target = tmp_path / "nested" / "dir" / "turns.jsonl" + sink = JsonlFileSink(target) + sink.emit('{"a":1}') + sink.close() + assert target.is_file() + + def test_context_manager_closes_handle(self, tmp_path: Path) -> None: + target = tmp_path / "turns.jsonl" + with JsonlFileSink(target) as sink: + sink.emit('{"a":1}') + # Re-open in append mode — verifies the prior handle closed. + sink2 = JsonlFileSink(target) + sink2.emit('{"b":2}') + sink2.close() + assert target.read_text(encoding="utf-8") == '{"a":1}\n{"b":2}\n' + + def test_emit_flushes_each_line(self, tmp_path: Path) -> None: + """Eager flush — a crashed runtime should still have prior + turns durable on disk.""" + target = tmp_path / "turns.jsonl" + sink = JsonlFileSink(target) + sink.emit('{"a":1}') + # Read while the sink is still open; flush should have made + # the line visible. + assert target.read_text(encoding="utf-8") == '{"a":1}\n' + sink.close() + + +# ---------- runtime auto-emission ---------- + + +class TestRuntimeAutoEmit: + def test_no_sink_attached_is_noop(self) -> None: + rt = ChatRuntime(config=RuntimeConfig()) + # No sink attached — chat() should not raise and turn_log + # should still be populated. + rt.chat("light is") + assert rt.turn_log + + def test_attached_sink_receives_one_line_per_turn(self) -> None: + rt = ChatRuntime(config=RuntimeConfig()) + sink = JsonlBufferSink() + rt.attach_telemetry_sink(sink) + rt.chat("light is") + rt.chat("light is") + assert len(sink.lines) == 2 + + def test_emitted_line_is_parseable_jsonl(self) -> None: + rt = ChatRuntime(config=RuntimeConfig()) + sink = JsonlBufferSink() + rt.attach_telemetry_sink(sink) + rt.chat("light is") + parsed = json.loads(sink.lines[-1]) + assert "turn" in parsed + assert "refusal_emitted" in parsed + assert "hedge_injected" in parsed + assert "stub_path" in parsed + + def test_default_attach_redacts_content(self) -> None: + rt = ChatRuntime(config=RuntimeConfig()) + sink = JsonlBufferSink() + rt.attach_telemetry_sink(sink) + rt.chat("light is") + parsed = json.loads(sink.lines[-1]) + assert "input_tokens" not in parsed + assert "surface" not in parsed + assert "walk_surface" not in parsed + + def test_include_content_opt_in_at_attach(self) -> None: + rt = ChatRuntime(config=RuntimeConfig()) + sink = JsonlBufferSink() + rt.attach_telemetry_sink(sink, include_content=True) + rt.chat("light is") + parsed = json.loads(sink.lines[-1]) + assert "input_tokens" in parsed + assert "surface" in parsed + + def test_detach_via_none_stops_emission(self) -> None: + rt = ChatRuntime(config=RuntimeConfig()) + sink = JsonlBufferSink() + rt.attach_telemetry_sink(sink) + rt.chat("light is") + rt.attach_telemetry_sink(None) + rt.chat("light is") + assert len(sink.lines) == 1 + + def test_pack_ids_in_emitted_line(self) -> None: + rt = ChatRuntime(config=RuntimeConfig()) + sink = JsonlBufferSink() + rt.attach_telemetry_sink(sink) + rt.chat("light is") + parsed = json.loads(sink.lines[-1]) + assert parsed["safety_pack_id"] == rt.safety_pack.pack_id + assert parsed["ethics_pack_id"] == rt.ethics_pack_id + assert parsed["identity_pack_id"] == rt.identity_pack_id + + def test_stub_path_also_emits(self) -> None: + """ADR-0039 made stub paths populate turn_log; ADR-0040's + sink must emit for stub turns too.""" + rt = ChatRuntime(config=RuntimeConfig()) + sink = JsonlBufferSink() + rt.attach_telemetry_sink(sink) + rt.chat("light is") + parsed = json.loads(sink.lines[-1]) + assert parsed["stub_path"] is True + + def test_refusal_visible_via_sink(self) -> None: + rt = ChatRuntime(config=RuntimeConfig()) + + def _failing(ctx): # noqa: ANN001 + return SafetyCheckResult( + boundary_id="preserve_versor_closure", + upheld=False, + reason="forced", + runtime_checkable=True, + ) + + rt.safety_check.register("preserve_versor_closure", _failing) + sink = JsonlBufferSink() + rt.attach_telemetry_sink(sink) + rt.chat("light is") + parsed = json.loads(sink.lines[-1]) + assert parsed["refusal_emitted"] is True + assert "preserve_versor_closure" in parsed["safety_violated"] + + def test_sink_errors_are_not_swallowed(self) -> None: + """Telemetry failures should surface, not silently drop the + audit signal.""" + rt = ChatRuntime(config=RuntimeConfig()) + + class _ExplodingSink: + def emit(self, line: str) -> None: + raise RuntimeError("sink down") + + rt.attach_telemetry_sink(_ExplodingSink()) + try: + rt.chat("light is") + except RuntimeError as e: + assert "sink down" in str(e) + else: + raise AssertionError("expected sink error to propagate") + + +# ---------- file-sink round trip ---------- + + +class TestFileSinkRoundTrip: + def test_full_session_round_trips(self, tmp_path: Path) -> None: + target = tmp_path / "session.jsonl" + rt = ChatRuntime(config=RuntimeConfig()) + with JsonlFileSink(target) as sink: + rt.attach_telemetry_sink(sink) + rt.chat("light is") + rt.chat("light is") + lines = target.read_text(encoding="utf-8").splitlines() + assert len(lines) == 2 + parsed = [json.loads(line) for line in lines] + # Each line is a complete audit record regardless of which + # turn path produced it (cold-start may seed enough vault + # state that later turns leave the stub path). + for p in parsed: + assert "stub_path" in p + assert isinstance(p["stub_path"], bool) + assert "safety_upheld" in p + assert "ethics_upheld" in p + assert "refusal_emitted" in p + assert "hedge_injected" in p + + +# ---------- ad-hoc event serialisation (no runtime needed) ---------- + + +class TestStandaloneEventSerialisation: + def test_minimal_event_serialises(self) -> None: + """A bare TurnEvent (no verdicts) should still serialise + cleanly — the boundary copes with sparse data.""" + event = TurnEvent( + turn=0, + input_tokens=("x",), + surface="s", + walk_surface="w", + articulation_surface="a", + dialogue_role="assert", + identity_score=None, + cycle_cost_total=0.0, + vault_hits=0, + versor_condition=0.0, + flagged=False, + ) + out = serialize_turn_event(event) + assert out["turn"] == 0 + assert out["refusal_emitted"] is False + assert out["hedge_injected"] is False + + def test_hedge_flag_from_bundle(self) -> None: + bundle = TurnVerdicts( + identity_score=None, + safety_verdict=None, + ethics_verdict=None, + refusal_emitted=False, + hedge_injected=True, + ) + event = TurnEvent( + turn=1, + input_tokens=("x",), + surface="s", + walk_surface="w", + articulation_surface="a", + dialogue_role="assert", + identity_score=None, + cycle_cost_total=0.0, + vault_hits=0, + versor_condition=0.0, + flagged=False, + verdicts=bundle, + ) + out = serialize_turn_event(event) + assert out["hedge_injected"] is True