core-labs/core
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

feat(adr-0059): correction-pass telemetry emission — backward perturbation auditable

Browse source 
`ChatRuntime.correct()` propagates a backward perturbation through the
session graph (per session/correction.py): each past turn whose output
versor has non-trivial CGA-alignment with the correction versor is
blended toward it (decayed by graph distance).  The forward regen turn
that followed already emitted a TurnEvent — but the backward
perturbation itself was invisible to the telemetry sink.

ADR-0059 closes that gap with a discriminated event line.

- chat/telemetry.py — adds `serialize_correction_event` +
  `format_correction_event_jsonl` emitting one JSONL line discriminated
  by `"type": "correction"`.  Payload: target_turn, records_count,
  turns_skipped, turn_idxs_affected, max_delta_norm, mean_delta_norm,
  SHA-256 correction_versor_digest, pack ids.  No raw versor coordinates.
- chat/runtime.py — `_emit_correction_event` (mirrors
  `_emit_turn_event`); called from `correct()` after the graph state
  is updated but before the forward regen turn.  No-op without sink.
- tests/test_correction_telemetry.py — 7 tests pin: no-op without
  sink, emission with sink, payload shape (required keys + types +
  ranges), SHA-256 digest shape, trust boundary (no versor
  coordinates leaked), determinism (byte-identical lines across
  runs), correction event and turn event coexist in the sink.

Trust boundary (per CLAUDE.md):
  - Metadata-only: only L2 deltas + SHA-256 digest.
  - No implicit wall-clock.
  - Deterministic: same CorrectionResult → byte-identical line.
  - Sink contract unchanged: `emit(line: str)`.
  - `versor_condition < 1e-6` invariant: untouched (telemetry-only).

Verification: smoke 67 / runtime 19 / correction telemetry 7 — green.
This commit is contained in:
Shay 2026-05-18 13:47:48 -07:00
parent fd80da6ac0
commit 29449f3775
5 changed files with 477 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

32
chat/runtime.py
View file

@ -26,7 +26,11 @@ from chat.refusal import (
inject_hedge,
should_inject_hedge,
)
from chat.telemetry import TurnEventSink, format_turn_event_jsonl
from chat.telemetry import (
TurnEventSink,
format_correction_event_jsonl,
format_turn_event_jsonl,
)
from chat.verdicts import TurnVerdicts
from teaching.discovery import (
extract_discovery_candidates,
@ -1144,11 +1148,37 @@ class ChatRuntime:
from_turn=target_turn,
)
self._context.apply_corrected_outputs(correction_result.records)
# ADR-0059 — emit a correction event before the regen turn so
# audit consumers can pair the backward perturbation with the
# forward turn event it produces. No-op when no sink attached.
self._emit_correction_event(correction_result, target_turn=target_turn)
regen_tokens = self._context.last_input_tokens
if not regen_tokens:
return self._stub_response(correction_state)
return self.chat(" ".join(regen_tokens), max_tokens=max_tokens)
def _emit_correction_event(
self, correction_result, *, target_turn: int,
) -> None:
"""ADR-0059 — emit one JSONL correction event to the telemetry sink.
Mirrors ``_emit_turn_event``: no-op when no sink is attached;
sink errors are intentionally NOT swallowed so a misconfigured
durable sink surfaces loudly rather than silently dropping
audit evidence.
"""
sink = self._telemetry_sink
if sink is None:
return
line = format_correction_event_jsonl(
correction_result,
target_turn=target_turn,
identity_pack_id=self.identity_pack_id,
safety_pack_id=getattr(self.safety_pack, "pack_id", ""),
ethics_pack_id=self.ethics_pack_id,
)
sink.emit(line)
def respond(self, text: str, max_tokens: int | None = None) -> str:
try:
return self.chat(text, max_tokens=max_tokens).surface

93
chat/telemetry.py
View file

@ -120,6 +120,97 @@ def format_turn_event_jsonl(event, **kwargs) -> str:
return json.dumps(payload, sort_keys=True, separators=(",", ":"))
# ---------- ADR-0059 — correction-event serializer ----------
def serialize_correction_event(
correction_result,
*,
target_turn: int,
identity_pack_id: str = "",
safety_pack_id: str = "",
ethics_pack_id: str = "",
timestamp: str | None = None,
) -> dict[str, object]:
"""Produce a JSON-safe audit dict from a ``CorrectionResult``.
Distinct from a turn event: this records the *backward* update to
a session graph triggered by ``ChatRuntime.correct()``. The
forward regen turn that follows still emits its own turn event;
this event documents the perturbation itself so audit consumers
can answer "which past turns moved, by how much, and toward what".
Trust boundary (per CLAUDE.md):
* **Metadata-only.** Versor coordinates are NOT emitted — only
the L2-delta-norm-per-record and a SHA-256 digest of the
correction versor's float32 bytes (deterministic identifier).
* **No implicit wall-clock.** ``timestamp`` is caller-provided.
* **Deterministic.** Same ``CorrectionResult`` → byte-identical
serialized line. ``records`` are traversed in their tuple
order (deterministic, matches insertion order from
``CorrectionPass.apply``).
"""
import hashlib
import math
records = getattr(correction_result, "records", ()) or ()
correction_versor = getattr(correction_result, "correction_versor", None)
# Per-record L2 deltas + the max across records.
deltas: list[float] = []
turn_idxs: list[int] = []
for r in records:
old_v = getattr(r, "old_versor", None)
new_v = getattr(r, "new_versor", None)
if old_v is None or new_v is None:
continue
# numpy.ndarray subtraction + norm; pure stdlib fallback would
# be slower but the runtime already imports numpy on this path.
import numpy as np
delta = float(np.linalg.norm(np.asarray(new_v) - np.asarray(old_v)))
if math.isfinite(delta):
deltas.append(delta)
turn_idxs.append(int(getattr(r, "turn_idx", 0)))
# SHA-256 digest of the correction versor's float32 bytes — gives
# a stable identifier for the perturbation without leaking
# coordinates. Falls back to empty string when missing.
digest = ""
if correction_versor is not None:
import numpy as np
digest = hashlib.sha256(
np.asarray(correction_versor, dtype=np.float32).tobytes()
).hexdigest()
out: dict[str, object] = {
"type": "correction",
"target_turn": int(target_turn),
"identity_pack_id": str(identity_pack_id),
"safety_pack_id": str(safety_pack_id),
"ethics_pack_id": str(ethics_pack_id),
"records_count": int(getattr(correction_result, "turns_affected", len(records))),
"turns_skipped": int(getattr(correction_result, "turns_skipped", 0)),
"turn_idxs_affected": sorted(turn_idxs),
"max_delta_norm": max(deltas) if deltas else 0.0,
"mean_delta_norm": (sum(deltas) / len(deltas)) if deltas else 0.0,
"correction_versor_digest": digest,
}
if timestamp is not None:
out["timestamp"] = str(timestamp)
return out
def format_correction_event_jsonl(correction_result, **kwargs) -> str:
"""Serialize a correction event as one deterministic JSONL line.
The ``"type": "correction"`` field discriminates this line from
turn events at consume time without changing the sink contract.
"""
payload = serialize_correction_event(correction_result, **kwargs)
return json.dumps(payload, sort_keys=True, separators=(",", ":"))
# ---------- sink protocol ----------
@ -267,7 +358,9 @@ __all__ = [
"JsonlBufferSink",
"JsonlFileSink",
"TurnEventSink",
"format_correction_event_jsonl",
"format_turn_event_jsonl",
"format_verdict_summary",
"serialize_correction_event",
"serialize_turn_event",
]

178
docs/decisions/ADR-0059-correction-pass-telemetry.md Normal file
View file

@ -0,0 +1,178 @@
# ADR-0059 — Correction-Pass Telemetry Emission
**Status:** Accepted
**Date:** 2026-05-18
**Author:** Shay
---
## Context
`ChatRuntime.correct(text, target_turn=-1)` propagates a backward
perturbation through the session graph: each past turn whose output
versor has non-trivial CGA-alignment with the correction versor is
blended toward it (decayed by graph distance). The mechanism is
documented in `session/correction.py`; `CorrectionPass.apply()`
returns a `CorrectionResult` carrying:
- `records: tuple[CorrectionRecord, ...]` — per-turn old/new versors,
alignment, decay, blend weight, distance from the correction-origin.
- `turns_affected: int`
- `turns_skipped: int`
The forward regen turn that follows `correct()` emits its own
`TurnEvent` via the existing telemetry sink. But the **backward
perturbation itself** — which past turns moved, by how much, and
toward what — was invisible to the telemetry sink.
That gap matters for the audit story:
- The session graph's state changed *between* turn events. A future
replay of the session would see two turn events whose deltas don't
add up unless the audit consumer knows a correction happened.
- The ADR-0055..0057 reviewed-corpus story rests on **every state
change** being inspectable. A backward correction that no sink
records is a structurally invisible mutation.
---
## Decision
Emit one JSONL line per `correct()` invocation, discriminated by
`"type": "correction"`, to the existing telemetry sink. Same sink
contract as `TurnEvent`; new event type alongside.
### Payload
```json
{
"type": "correction",
"target_turn": -1,
"identity_pack_id": "default_general_v1",
"safety_pack_id": "core_safety_axes_v1",
"ethics_pack_id": "default_general_ethics_v1",
"records_count": 1,
"turns_skipped": 0,
"turn_idxs_affected": [1],
"max_delta_norm": 1.3337411880493164,
"mean_delta_norm": 1.3337411880493164,
"correction_versor_digest": "e7e98af75ff6b4475807ed66f0672ec0cfab6f205896f8e626fc1490026ed597"
}
```
- `records_count` / `turns_skipped` — how many turns were perturbed
vs filtered by the minimum-alignment threshold.
- `turn_idxs_affected` — sorted tuple of affected turn indices.
Lets a downstream consumer correlate the correction with the
specific past turn events it perturbed.
- `max_delta_norm` / `mean_delta_norm` — L2 distance between
old and new versors per record, max and mean across records.
Magnitude of the backward perturbation without leaking
coordinates.
- `correction_versor_digest` — SHA-256 hex of the correction
versor's float32 bytes. Deterministic identifier for the
perturbation source; pairs identical corrections across runs
without exposing coordinates.
### Trust boundary (per CLAUDE.md)
- **Metadata-only.** No raw versor coordinates ever appear in the
payload — `old_versor`, `new_versor`, `correction_versor` are not
serialised. Only the L2 deltas and the SHA-256 digest.
- **No implicit wall-clock.** Timestamp is caller-provided (kwarg);
the runtime does not reach for `datetime.now()` on this path.
- **Deterministic.** Same `CorrectionResult` → byte-identical line.
`sort_keys=True` in the serialiser; record order matches
`CorrectionPass.apply` insertion order (deterministic).
- **No-op without a sink.** When no telemetry sink is attached,
`_emit_correction_event` returns immediately — `correct()`
behaviour is byte-identical to pre-ADR-0059.
- **Sink errors propagate.** Mirrors `_emit_turn_event`'s contract:
a misconfigured durable sink surfaces loudly rather than
silently dropping audit evidence.
---
## Why a discriminated event, not an extended turn event
Two options were considered:
1. **Add a `correction_event` block to `TurnEvent`.** Same line,
optional field. Lighter touch but pollutes the turn-event
schema with a payload that's only present on `correct()` calls,
and conflates two semantically distinct events (the backward
perturbation and the forward regen turn).
2. **Emit a separate event with a `"type": "correction"` discriminator.**
New line type alongside turn events. Sink contract unchanged
(still `emit(line: str)`); consumers discriminate by reading
the `"type"` field (absent on turn events today, present on
correction events).
Path (2) chosen because:
- It keeps the turn-event schema clean.
- It models the underlying event structure honestly: a `correct()`
call produces two events (the perturbation, then the regen turn),
not one composite turn event.
- It composes with the existing `FanOutSink` and `JsonlFileSink`
unchanged.
- A future `_emit_supersession_event`, `_emit_proposal_event`, etc.
can follow the same discriminator pattern.
---
## Consequences
### What changes
- `chat/telemetry.py` gains `serialize_correction_event` and
`format_correction_event_jsonl`. No changes to existing
serialisers.
- `chat/runtime.py` adds `_emit_correction_event` (mirrors
`_emit_turn_event`) and calls it from `correct()` after the
graph state is updated but before the forward regen turn.
### What does not change
- Sink contract: still `emit(line: str)`.
- Existing telemetry consumers that key off the absence of a
`"type"` field (treating every line as a turn event) keep
working — they'll see correction lines and either ignore them
or fail loudly, depending on their parser. Recommended consumer
pattern: dispatch on `payload.get("type", "turn")`.
- `CorrectionPass` itself is untouched; `CorrectionResult` already
carried everything the event needs.
- `versor_condition < 1e-6` invariant: this ADR adds telemetry
emission only; no field state is touched.
---
## Verification
```
tests/test_correction_telemetry.py 7 passed
- no_emission_without_sink
- emission_when_sink_attached
- correction_payload_shape (required keys, types, ranges)
- correction_versor_digest_is_sha256_prefix_or_empty
- no_versor_coordinates_in_payload (trust boundary)
- correction_event_is_deterministic_across_runs (byte-identical lines)
- turn_event_and_correction_event_coexist
Lanes (regression check):
core test --suite smoke 67 passed
core test --suite runtime 19 passed
```
---
## Cross-References
- [ADR-0040](./ADR-0040-telemetry-sink.md) — the structured-logging
sink this ADR adds an event type to.
- [ADR-0039](./ADR-0039-bundled-turn-verdicts.md) — the
`TurnVerdicts` bundle that made every turn event uniform.
- [ADR-0055](./ADR-0055-inter-session-memory-discovery-promotion.md)
— the broader inter-session-memory architecture where every state
change must be inspectable, of which correction events are now part.

1
docs/decisions/README.md
View file

@ -68,6 +68,7 @@ ADRs record significant architectural decisions: what was decided, why, what alt
| [ADR-0056](ADR-0056-contemplation-loop-c1.md) | Contemplation loop (Phase C1): question decomposition, polarity (affirms/falsifies/undetermined), claim_domain typing (factual/relational/evaluative), sync-only by design | **Accepted** (2026-05-18, implemented `4eecf73`) |
| [ADR-0057](ADR-0057-teaching-chain-proposal-review.md) | Teaching-chain proposal + review + replay-equivalence gate (Phase C2): the only path to active-corpus extension; eligibility predicate; auto-reject on metric regression; operator accept/reject/withdraw; append-only proposal log | **Accepted** (2026-05-18) |
| [ADR-0058](ADR-0058-forward-graph-constraint-status.md) | `forward_graph_constraint` remains opt-in default-`False`; no identity pack flips it on; ADR-0047 null-lift on cognition lane promoted to CI-enforced invariant (regression test); identity-pack→`RuntimeConfig` composition deferred until at least one such preference shows lift | **Accepted** (2026-05-18) |
| [ADR-0059](ADR-0059-correction-pass-telemetry.md) | `ChatRuntime.correct()` emits a discriminated `"type": "correction"` JSONL event to the existing telemetry sink with `target_turn`, `records_count`, `turn_idxs_affected`, `max_delta_norm`, `mean_delta_norm`, SHA-256 correction-versor digest, pack ids — no raw versor coordinates; deterministic; no-op without sink | **Accepted** (2026-05-18) |
---

174
tests/test_correction_telemetry.py Normal file
View file

@ -0,0 +1,174 @@
"""ADR-0059 — correction-pass telemetry emission.
The backward perturbation triggered by ``ChatRuntime.correct()`` must
be visible to audit consumers. Today the forward regen turn that
follows ``correct()`` emits a turn event, but the correction delta
itself — which past turns moved, how far, and toward what — was
invisible to the telemetry sink.
ADR-0059 closes that gap with a dedicated correction-event JSONL line
discriminated by ``"type": "correction"``. Same sink contract; new
event type alongside ``TurnEvent`` lines.
These tests pin:
- No emission when no sink is attached.
- Emission when a sink is attached AND ``correct()`` triggered at
least one record.
- Payload shape: type discriminator, target_turn, records_count,
turns_skipped, turn_idxs_affected, max_delta_norm,
mean_delta_norm, correction_versor_digest (SHA-256), pack ids.
- Trust boundary: no versor coordinates in the line, only L2 deltas
and a deterministic versor digest.
- Determinism: same input → byte-identical line.
"""
from __future__ import annotations
import json
from chat.runtime import ChatRuntime
from chat.telemetry import JsonlBufferSink
def _seed_two_turns(rt: ChatRuntime) -> None:
"""Populate the session graph with enough turns for backward walk."""
rt.chat("Why does light exist?")
rt.chat("Why does thought exist?")
def test_no_emission_without_sink() -> None:
rt = ChatRuntime()
_seed_two_turns(rt)
response = rt.correct("light reveals truth")
assert response.surface != "" # regen still runs
def test_emission_when_sink_attached() -> None:
rt = ChatRuntime()
sink = JsonlBufferSink()
rt.attach_telemetry_sink(sink)
_seed_two_turns(rt)
n_before = len(sink.lines)
rt.correct("light reveals truth")
n_after = len(sink.lines)
assert n_after > n_before
correction_lines = [
json.loads(line) for line in sink.lines
if json.loads(line).get("type") == "correction"
]
assert len(correction_lines) == 1
def test_correction_payload_shape() -> None:
rt = ChatRuntime()
sink = JsonlBufferSink()
rt.attach_telemetry_sink(sink)
_seed_two_turns(rt)
rt.correct("light reveals truth")
correction_lines = [
json.loads(line) for line in sink.lines
if json.loads(line).get("type") == "correction"
]
assert len(correction_lines) == 1
payload = correction_lines[0]
required_keys = {
"type",
"target_turn",
"identity_pack_id",
"safety_pack_id",
"ethics_pack_id",
"records_count",
"turns_skipped",
"turn_idxs_affected",
"max_delta_norm",
"mean_delta_norm",
"correction_versor_digest",
}
assert required_keys.issubset(payload.keys())
assert payload["type"] == "correction"
assert payload["target_turn"] == -1
assert isinstance(payload["records_count"], int)
assert isinstance(payload["turn_idxs_affected"], list)
assert all(isinstance(t, int) for t in payload["turn_idxs_affected"])
assert isinstance(payload["max_delta_norm"], (int, float))
assert payload["max_delta_norm"] >= 0.0
assert isinstance(payload["mean_delta_norm"], (int, float))
assert payload["mean_delta_norm"] >= 0.0
def test_correction_versor_digest_is_sha256_prefix_or_empty() -> None:
rt = ChatRuntime()
sink = JsonlBufferSink()
rt.attach_telemetry_sink(sink)
_seed_two_turns(rt)
rt.correct("light reveals truth")
correction_lines = [
json.loads(line) for line in sink.lines
if json.loads(line).get("type") == "correction"
]
digest = correction_lines[0]["correction_versor_digest"]
assert isinstance(digest, str)
# SHA-256 hex digest = 64 chars; emit empty only if versor absent.
assert digest == "" or (len(digest) == 64 and all(c in "0123456789abcdef" for c in digest))
def test_no_versor_coordinates_in_payload() -> None:
"""Trust boundary — only L2 deltas + SHA-256 digest, never coords."""
rt = ChatRuntime()
sink = JsonlBufferSink()
rt.attach_telemetry_sink(sink)
_seed_two_turns(rt)
rt.correct("light reveals truth")
payload = next(
json.loads(line) for line in sink.lines
if json.loads(line).get("type") == "correction"
)
forbidden_keys = {"old_versor", "new_versor", "correction_versor", "versor", "records"}
leaked = forbidden_keys & set(payload.keys())
assert not leaked, (
f"correction event leaked raw versor field(s) {leaked!r}: {payload}"
)
def test_correction_event_is_deterministic_across_runs() -> None:
"""Two runtimes seeded identically and corrected identically must
emit byte-identical correction-event lines."""
def _run() -> str:
rt = ChatRuntime()
sink = JsonlBufferSink()
rt.attach_telemetry_sink(sink)
_seed_two_turns(rt)
rt.correct("light reveals truth")
return next(
line for line in sink.lines
if json.loads(line).get("type") == "correction"
)
a = _run()
b = _run()
assert a == b
def test_turn_event_and_correction_event_coexist() -> None:
"""``correct()`` emits a correction event AND the regen turn emits
its own turn event. Both types appear in the sink."""
rt = ChatRuntime()
sink = JsonlBufferSink()
rt.attach_telemetry_sink(sink)
_seed_two_turns(rt)
n_before = len(sink.lines)
rt.correct("light reveals truth")
new_lines = [json.loads(line) for line in sink.lines[n_before:]]
types_seen = {payload.get("type", "turn") for payload in new_lines}
# Correction event discriminated; turn event has no "type" key by today's
# serializer, so it presents as the default "turn".
assert "correction" in types_seen