feat(adr-0040): structured-logging sink for turn-event audit
Adds the canonical JSONL sink surface consuming TurnEvent records that ADR-0039 made uniform across main and stub paths. One deterministic line per turn; redact-by-default trust boundary; opt-in content emission; runtime auto-emits on attached sink. Trust boundary (CLAUDE.md): - Metadata-only by default — no surfaces or input tokens emitted. include_content=True opt-in at attachment time. - Path fixed at construction for JsonlFileSink; no user-controlled paths interpreted at emit time. - Sink errors propagate — telemetry failures should surface, not silently drop audit signal. Determinism: - sort_keys=True; compact separators. Same event → byte-identical line. - No implicit wall-clock; timestamps caller-provided. - Field set fixed; missing TurnEvent attrs fall back to safe defaults. API: - serialize_turn_event(event, **kwargs) -> dict (pure) - format_turn_event_jsonl(event, **kwargs) -> str (pure, deterministic) - TurnEventSink Protocol; JsonlBufferSink; JsonlFileSink - ChatRuntime.attach_telemetry_sink(sink, *, include_content=False) - _emit_turn_event invoked after both turn_log.append sites Wire format (alphabetised, always present): cycle_cost_total, dialogue_role, ethics_pack_id, ethics_runtime_checkable_count, ethics_upheld, ethics_violated, flagged, hedge_injected, identity_pack_id, refusal_emitted, safety_pack_id, safety_runtime_checkable_count, safety_upheld, safety_violated, stub_path, turn, vault_hits, versor_condition. Conditional: identity_* (when score present), surface / walk_surface / articulation_surface / input_tokens (when include_content=True), timestamp (when provided). Files: - chat/telemetry.py (new) — serializer, formatter, sinks - chat/runtime.py — attach + emit + post-append calls - tests/test_telemetry_sink.py (new) — 29 tests - docs/decisions/ADR-0040-telemetry-sink.md (new) Verification: - Combined pack-layer + telemetry suite: 199 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)
This commit is contained in:
Shay
parent
f3cc408f82
commit
226f14a941
4 changed files with 847 additions and 0 deletions
|
|
@ -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,
|
||||
|
|
|
|||
192
chat/telemetry.py
Normal file
192
chat/telemetry.py
Normal file
|
|
@ -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",
|
||||
]
|
||||
218
docs/decisions/ADR-0040-telemetry-sink.md
Normal file
218
docs/decisions/ADR-0040-telemetry-sink.md
Normal file
|
|
@ -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()`.
|
||||
387
tests/test_telemetry_sink.py
Normal file
387
tests/test_telemetry_sink.py
Normal file
|
|
@ -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
|
||||
Loading…
Reference in a new issue