From 417f71917cbe6bd97ed48627310fc9dab2e4cb3d Mon Sep 17 00:00:00 2001 From: Shay Date: Sun, 17 May 2026 21:47:47 -0700 Subject: [PATCH] feat(adr-0041): core chat --show-verdicts + FanOutSink MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two thin layers closing the audit story end-to-end: - core chat --show-verdicts prints format_verdict_summary(verdicts) to stderr after each turn. Stdout stays clean for piped consumers. Format is dense and terse; designed to skim, not machine-parseable (the JSONL sink owns that contract). - FanOutSink forwards every emitted line to N sinks in declaration order. Fail-fast on first error — consistent with ADR-0040's single-sink contract (audit failures surface). Composes with any combination of JsonlFileSink / JsonlBufferSink / future sinks. Two formatters, one bundle: format_turn_event_jsonl (machine, ADR-0040) and format_verdict_summary (operator, ADR-0041) both consume the same TurnVerdicts. No risk of drift. Summary format: [identity=0.83 safety=ok ethics=VIOLATED:foo refusal=- hedge=YES] Audit story now reads end-to-end: - TurnVerdicts bundle (ADR-0039) - Machine JSONL sink (ADR-0040) - Fan-out + operator CLI (ADR-0041) Files: - chat/telemetry.py — FanOutSink dataclass, format_verdict_summary, _format_verdict_short helper - core/cli.py — --show-verdicts on chat subparser; cmd_chat prints summary to stderr when set - tests/test_telemetry_fanout_and_summary.py (new) — 13 tests - docs/decisions/ADR-0041-cli-verdicts-and-fanout.md (new) Verification: - Combined pack-layer + telemetry suite: 212 green (was 199; +13) - CLI suites unchanged: smoke 67, runtime 19, cognition 121 - core eval cognition: intent 100%, versor_closure 100% (baseline) - Manual smoke: echo "light is" | core chat --show-verdicts prints expected bracketed audit line to stderr alongside response. --- chat/telemetry.py | 81 +++++++ core/cli.py | 19 ++ .../ADR-0041-cli-verdicts-and-fanout.md | 174 ++++++++++++++ tests/test_telemetry_fanout_and_summary.py | 220 ++++++++++++++++++ 4 files changed, 494 insertions(+) create mode 100644 docs/decisions/ADR-0041-cli-verdicts-and-fanout.md create mode 100644 tests/test_telemetry_fanout_and_summary.py diff --git a/chat/telemetry.py b/chat/telemetry.py index e65bd10c..d3dc6ec3 100644 --- a/chat/telemetry.py +++ b/chat/telemetry.py @@ -183,10 +183,91 @@ class JsonlFileSink: self.close() +# ---------- fan-out ---------- + + +@dataclass +class FanOutSink: + """Forward every emitted line to N sinks in declaration order. + + ADR-0041. Composes with any combination of sinks — typically + ``JsonlFileSink`` (durable) + ``JsonlBufferSink`` (in-memory + audit), or two file sinks (local + shadow). + + **Error semantics:** fail-fast. If sink *i* raises, sinks *i+1..* + are NOT called and the exception propagates to the caller. This + is consistent with the single-sink contract: telemetry failures + surface, never silently drop audit signal. Callers wanting + partial-success semantics wrap individual sinks in their own + error-tolerant shim. + """ + + sinks: tuple = () # tuple[TurnEventSink, ...] + + def emit(self, line: str) -> None: + for sink in self.sinks: + sink.emit(line) + + +# ---------- operator-facing summary formatter ---------- + + +def format_verdict_summary(verdicts) -> str: + """ADR-0041 — one-line human-readable summary of a TurnVerdicts bundle. + + Used by ``core chat --show-verdicts`` to print a per-turn audit + line to the operator. Distinct from ``format_turn_event_jsonl`` + (machine-facing): this is dense, terse, and skims the high-signal + fields. Empty string when ``verdicts`` is None. + + Format:: + + [identity=0.83 safety=ok ethics=ok refusal=- hedge=-] + [identity=0.42 safety=VIOLATED:preserve_versor_closure ethics=ok refusal=YES hedge=-] + """ + if verdicts is None: + return "" + parts: list[str] = [] + identity = getattr(verdicts, "identity_score", None) + if identity is not None: + alignment = float(getattr(identity, "alignment", 0.0)) + parts.append(f"identity={alignment:.2f}") + else: + parts.append("identity=-") + safety = getattr(verdicts, "safety_verdict", None) + parts.append(_format_verdict_short( + safety, "safety", id_attr="violated_boundaries", + )) + ethics = getattr(verdicts, "ethics_verdict", None) + parts.append(_format_verdict_short( + ethics, "ethics", id_attr="violated_commitments", + )) + parts.append( + "refusal=YES" if getattr(verdicts, "refusal_emitted", False) + else "refusal=-" + ) + parts.append( + "hedge=YES" if getattr(verdicts, "hedge_injected", False) + else "hedge=-" + ) + return "[" + " ".join(parts) + "]" + + +def _format_verdict_short(verdict, label: str, *, id_attr: str) -> str: + if verdict is None: + return f"{label}=-" + violated = sorted(getattr(verdict, id_attr, ()) or ()) + if not violated: + return f"{label}=ok" + return f"{label}=VIOLATED:{','.join(violated)}" + + __all__ = [ + "FanOutSink", "JsonlBufferSink", "JsonlFileSink", "TurnEventSink", "format_turn_event_jsonl", + "format_verdict_summary", "serialize_turn_event", ] diff --git a/core/cli.py b/core/cli.py index f258749f..1f5d5c83 100644 --- a/core/cli.py +++ b/core/cli.py @@ -207,10 +207,14 @@ def cmd_chat(args: argparse.Namespace) -> int: return _print_identity_packs(use_json=getattr(args, "json", False)) try: from chat.runtime import ChatRuntime + # ADR-0041 — operator-facing verdict readout. Imported lazily + # so a broken telemetry module doesn't block REPL startup. + from chat.telemetry import format_verdict_summary except Exception as exc: # pragma: no cover - exercised by CLI in broken envs _print_runtime_import_hint(exc) runtime = ChatRuntime(config=_runtime_config_from_args(args)) + show_verdicts = bool(getattr(args, "show_verdicts", False)) while True: try: text = input("> ").strip() @@ -227,6 +231,13 @@ def cmd_chat(args: argparse.Namespace) -> int: print(f"[{exc}]", file=sys.stderr) continue print(response.surface) + if show_verdicts: + # ADR-0041 — print the verdict bundle to stderr so the + # response surface on stdout stays parseable by tooling + # that pipes through ``core chat``. + summary = format_verdict_summary(response.verdicts) + if summary: + print(summary, file=sys.stderr) return 0 @@ -1145,6 +1156,14 @@ def build_parser() -> argparse.ArgumentParser: action="store_true", help="emit machine-readable JSON (with --list-identity-packs)", ) + chat.add_argument( + "--show-verdicts", + action="store_true", + help=( + "after each turn, print the TurnVerdicts bundle summary to " + "stderr (ADR-0041 operator-facing audit readout)" + ), + ) chat.set_defaults(func=cmd_chat) test = subparsers.add_parser("test", help="run pytest with curated suite aliases or direct passthrough") diff --git a/docs/decisions/ADR-0041-cli-verdicts-and-fanout.md b/docs/decisions/ADR-0041-cli-verdicts-and-fanout.md new file mode 100644 index 00000000..5743aaf2 --- /dev/null +++ b/docs/decisions/ADR-0041-cli-verdicts-and-fanout.md @@ -0,0 +1,174 @@ +# ADR-0041: `core chat --show-verdicts` + Sink Fan-Out + +**Status:** Accepted (2026-05-17) +**Author:** Joshua Shay + planner pass +**Companion docs:** [`ADR-0039-audit-completeness.md`](ADR-0039-audit-completeness.md), [`ADR-0040-telemetry-sink.md`](ADR-0040-telemetry-sink.md) + +## Context + +ADR-0040 landed the machine-facing audit surface: deterministic JSONL +lines on an attached sink. Two follow-ups were left explicit: + +* **Operator-facing readout.** A human-readable per-turn summary + printed alongside the chat response so an operator can debug + refusals/hedges/violations interactively without parsing JSONL. +* **Sink fan-out.** Today one sink at a time. Operators often + want both a durable local file *and* a remote aggregator (or two + sinks pointing at different log stores) simultaneously. + +Both are thin layers on top of the existing surface. Bundling them +into one ADR keeps the audit story coherent. + +## Decision + +### `core chat --show-verdicts` + +New CLI flag on the `chat` subparser. When set, after each turn the +REPL prints the verdict bundle summary to **stderr**: + +```text +> light is +I don't know — insufficient grounding for that yet. +[identity=- safety=ok ethics=VIOLATED:acknowledge_uncertainty refusal=- hedge=-] +``` + +Design choices: + +* **Summary goes to stderr.** The chat response goes to stdout + (unchanged). Tooling that pipes `core chat` through a filter + doesn't see verdict noise interleaved with the response; humans + watching the terminal see both. +* **Format is dense and terse.** One bracketed line per turn, + fixed field order: `identity`, `safety`, `ethics`, `refusal`, + `hedge`. Designed to skim, not to be machine-parsed (the JSONL + sink owns that contract). +* **Distinct from the machine-facing JSONL.** ADR-0040's + `format_turn_event_jsonl` is for log aggregators; the new + `format_verdict_summary` is for operators. Two formatters, one + underlying bundle — no risk of drift. + +### `FanOutSink` + +New sink in `chat/telemetry.py`: + +```python +@dataclass +class FanOutSink: + sinks: tuple = () # tuple[TurnEventSink, ...] + def emit(self, line: str) -> None: + for sink in self.sinks: + sink.emit(line) +``` + +* **Fail-fast.** First sink that raises propagates the exception; + subsequent sinks are NOT called. Consistent with ADR-0040's + single-sink contract: telemetry failures surface visibly, never + silently drop audit signal. +* **Composable.** Any combination of sinks (file + buffer, + multiple files, file + a future remote sink) works. Order is + preserved. +* **Stateless.** No internal buffering; emission is synchronous + and immediate to each child sink. + +A `ResilientSink` wrapper (swallows errors per-sink) is intentionally +**not** included. Callers who want partial-success semantics wrap +individual sinks in their own error-tolerant shim. The default +contract should reflect the doctrine "audit failures are visible"; +the resilient wrapper is its own future ADR if needed at scale. + +### `format_verdict_summary` formatter + +New pure function in `chat/telemetry.py`: + +```python +def format_verdict_summary(verdicts) -> str: ... +``` + +* Returns `""` for `None` input. +* Pulls fields off the bundle using `getattr` with safe defaults — + same boundary discipline as `serialize_turn_event`. +* Renders identity alignment to two decimals (`identity=0.83`) or + `identity=-` when no score is available (stub turns). +* Safety / ethics: `ok` when no violations, else + `VIOLATED:,` in lex order. +* Remediation flags: `refusal=YES`/`refusal=-` and `hedge=YES`/`hedge=-`. + +## Consequences + +### Positive + +* **Operator path complete.** The audit story now reads end-to-end + for both humans and machines: + - Per-turn TurnVerdicts bundle (ADR-0039) + - Machine-facing JSONL sink (ADR-0040) + - Fan-out across multiple sinks (ADR-0041) + - Operator-facing CLI summary (ADR-0041) +* **No new core runtime surface.** Both additions are pure layers + on top of `ChatResponse.verdicts` and the existing sink protocol. +* **Single source of truth.** Operator and machine formatters share + the underlying bundle; no risk of one going stale while the other + evolves. +* **Composable fan-out.** Tested end-to-end: runtime attaches a + `FanOutSink` and gets atomic distribution to file + buffer (or any + N sinks) without changes to the runtime. + +### Negative / risks + +* **`format_verdict_summary` format is human-stable, not + machine-stable.** Callers should not parse it. The JSONL sink + remains the only machine-stable wire format. Documented in code + comments but not enforced at the type level. +* **Fan-out is synchronous.** Slow sinks slow the turn loop. + Acceptable today (the only sinks are in-memory and local file); + async / queued sinks are deferred to a future ADR alongside + backpressure. +* **CLI summary is on stderr.** Tooling reading stderr separately + works fine; tooling that merges streams sees the summary + interleaved. This is the standard Unix split between content + and metadata — the tradeoff was accepted to keep stdout clean + for piped consumers. + +## Verification + +* `tests/test_telemetry_fanout_and_summary.py` — 13 tests covering: + fan-out forwarding to all sinks, emission-order preservation, + empty-sinks no-op, fail-fast on first error (downstream sinks NOT + called), composition with file sink; runtime-with-fan-out + end-to-end; verdict summary (None → empty, clean turn, safety + violation, ethics violation, multiple violations lex-sorted, no + identity score, real ChatResponse formats without error). +* CLI smoke (manual): + ```text + $ echo "light is" | core chat --show-verdicts + > [identity=- safety=ok ethics=VIOLATED:acknowledge_uncertainty refusal=- hedge=-] + I don't know — insufficient grounding for that yet. + ``` + Correct: cold-start stub turn has no identity score; ethics + flags `acknowledge_uncertainty` because alignment_score=0.0 + falls below `hedge_threshold_soft`; default pack opt-in lists are + empty so no refusal/hedge fires. +* Combined pack-layer + telemetry suite: **212 tests, all green** + (was 199 after ADR-0040; +13). +* 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. **Resilient (error-tolerant) sink wrapper.** Wraps any sink and + swallows emit-time errors with optional callback. Sibling to + `FanOutSink`; lands if real deployments need it. +2. **Async / queued sinks.** Synchronous fan-out becomes a + bottleneck when one sink is a slow remote aggregator. A queued + wrapper with backpressure semantics is the natural next step. +3. **`--show-verdicts` granularity flags.** Today the summary is + a single one-liner. Operators may want `--verdicts-verbose` for + per-predicate detail or `--verdicts-only-on-violation` for + high-volume sessions. +4. **Schema versioning on the JSONL wire format.** Was deferred + from ADR-0040. Now that operators have a stable readout, the + machine wire format can evolve more aggressively if needed. +5. **Sink registry.** Today sinks are constructed by the caller. + A registry that resolves sinks from config strings + (`"file:/var/log/core/audit.jsonl"`, `"buffer"`) would simplify + declarative deployment configuration. diff --git a/tests/test_telemetry_fanout_and_summary.py b/tests/test_telemetry_fanout_and_summary.py new file mode 100644 index 00000000..1b21944d --- /dev/null +++ b/tests/test_telemetry_fanout_and_summary.py @@ -0,0 +1,220 @@ +"""ADR-0041 — sink fan-out and operator-facing verdict summary. + +Two sibling additions: + +* ``FanOutSink`` — forwards every emitted line to N sinks; fail-fast + on the first sink that raises (same error contract as a single + sink). +* ``format_verdict_summary`` — one-line operator-facing readout of a + ``TurnVerdicts`` bundle. Used by ``core chat --show-verdicts``. +""" + +from __future__ import annotations + +from dataclasses import replace +from pathlib import Path + +from chat.runtime import ChatRuntime +from chat.telemetry import ( + FanOutSink, + JsonlBufferSink, + JsonlFileSink, + format_verdict_summary, +) +from chat.verdicts import TurnVerdicts +from core.config import RuntimeConfig +from core.physics.identity import IdentityScore +from packs.ethics.check import EthicsCheckResult, EthicsVerdict +from packs.safety.check import SafetyCheckResult, SafetyVerdict + + +# ---------- FanOutSink ---------- + + +class TestFanOutSink: + def test_forwards_to_all_sinks(self) -> None: + a = JsonlBufferSink() + b = JsonlBufferSink() + fan = FanOutSink(sinks=(a, b)) + fan.emit('{"x":1}') + fan.emit('{"y":2}') + assert a.lines == ['{"x":1}', '{"y":2}'] + assert b.lines == ['{"x":1}', '{"y":2}'] + + def test_preserves_emission_order(self) -> None: + a = JsonlBufferSink() + fan = FanOutSink(sinks=(a,)) + for i in range(5): + fan.emit(f'{{"i":{i}}}') + assert a.lines == [f'{{"i":{i}}}' for i in range(5)] + + def test_empty_sinks_tuple_is_noop(self) -> None: + fan = FanOutSink(sinks=()) + # Must not raise. + fan.emit('{"x":1}') + + def test_fail_fast_on_first_sink_error(self) -> None: + """First-error semantics: when sink i raises, sinks i+1.. + do not receive the line. This preserves audit-signal + visibility — telemetry failures surface.""" + a = JsonlBufferSink() + b = JsonlBufferSink() + + class _Boom: + def emit(self, line: str) -> None: + raise RuntimeError("boom") + + fan = FanOutSink(sinks=(a, _Boom(), b)) + try: + fan.emit('{"x":1}') + except RuntimeError as e: + assert "boom" in str(e) + else: + raise AssertionError("expected RuntimeError to propagate") + # First sink saw the line; downstream sink after the failure + # did NOT. + assert a.lines == ['{"x":1}'] + assert b.lines == [] + + def test_composes_with_file_sink(self, tmp_path: Path) -> None: + target = tmp_path / "audit.jsonl" + buf = JsonlBufferSink() + with JsonlFileSink(target) as fsink: + fan = FanOutSink(sinks=(buf, fsink)) + fan.emit('{"x":1}') + fan.emit('{"y":2}') + assert buf.lines == ['{"x":1}', '{"y":2}'] + assert target.read_text(encoding="utf-8") == '{"x":1}\n{"y":2}\n' + + +# ---------- FanOutSink integrated with runtime ---------- + + +class TestRuntimeWithFanOut: + def test_runtime_attached_fanout_distributes(self, tmp_path: Path) -> None: + rt = ChatRuntime(config=RuntimeConfig()) + buf = JsonlBufferSink() + target = tmp_path / "session.jsonl" + with JsonlFileSink(target) as fsink: + rt.attach_telemetry_sink(FanOutSink(sinks=(buf, fsink))) + rt.chat("light is") + rt.chat("light is") + assert len(buf.lines) == 2 + assert len(target.read_text(encoding="utf-8").splitlines()) == 2 + + +# ---------- format_verdict_summary ---------- + + +class TestFormatVerdictSummary: + def test_none_returns_empty_string(self) -> None: + assert format_verdict_summary(None) == "" + + def test_clean_turn_summary(self) -> None: + bundle = _bundle( + identity_alignment=0.83, + refusal_emitted=False, + hedge_injected=False, + ) + out = format_verdict_summary(bundle) + assert out.startswith("[") and out.endswith("]") + assert "identity=0.83" in out + assert "safety=ok" in out + assert "ethics=ok" in out + assert "refusal=-" in out + assert "hedge=-" in out + + def test_safety_violation_summary(self) -> None: + bundle = _bundle( + safety_violated=("preserve_versor_closure",), + refusal_emitted=True, + ) + out = format_verdict_summary(bundle) + assert "safety=VIOLATED:preserve_versor_closure" in out + assert "refusal=YES" in out + + def test_ethics_violation_summary(self) -> None: + bundle = _bundle( + ethics_violated=("acknowledge_uncertainty",), + hedge_injected=True, + ) + out = format_verdict_summary(bundle) + assert "ethics=VIOLATED:acknowledge_uncertainty" in out + assert "hedge=YES" in out + + def test_multiple_violations_lex_sorted(self) -> None: + bundle = _bundle( + safety_violated=("zzz_late", "aaa_early"), + ) + out = format_verdict_summary(bundle) + assert "safety=VIOLATED:aaa_early,zzz_late" in out + + def test_no_identity_score_shows_dash(self) -> None: + bundle = TurnVerdicts( + identity_score=None, + safety_verdict=None, + ethics_verdict=None, + refusal_emitted=False, + hedge_injected=False, + ) + out = format_verdict_summary(bundle) + assert "identity=-" in out + + def test_response_from_runtime_formats(self) -> None: + """End-to-end: a real ChatResponse.verdicts bundle formats + without error.""" + rt = ChatRuntime(config=RuntimeConfig()) + resp = rt.chat("light is") + out = format_verdict_summary(resp.verdicts) + # Stub-path turn has no identity_score but valid verdicts. + assert out.startswith("[") and out.endswith("]") + assert "refusal=" in out + assert "hedge=" in out + + +# ---------- helpers ---------- + + +def _bundle( + *, + identity_alignment: float | None = 1.0, + safety_violated: tuple = (), + ethics_violated: tuple = (), + refusal_emitted: bool = False, + hedge_injected: bool = False, +) -> TurnVerdicts: + identity_score = None + if identity_alignment is not None: + # IdentityScore.alignment forces 1.0 when deviation_axes is + # empty; pass a non-empty set so the formatter sees the + # requested alignment value. + deviation = ( + frozenset() if identity_alignment >= 1.0 else frozenset({"_test_axis"}) + ) + identity_score = IdentityScore( + score=identity_alignment, + flagged=False, + deviation_axes=deviation, + trajectory_id="test", + ) + safety = SafetyVerdict( + pack_id="test_safety", + results=(), + upheld=not safety_violated, + violated_boundaries=frozenset(safety_violated), + runtime_checkable_count=0, + ) + ethics = EthicsVerdict( + pack_id="test_ethics", + results=(), + upheld=not ethics_violated, + violated_commitments=frozenset(ethics_violated), + runtime_checkable_count=0, + ) + return TurnVerdicts( + identity_score=identity_score, + safety_verdict=safety, + ethics_verdict=ethics, + refusal_emitted=refusal_emitted, + hedge_injected=hedge_injected, + )