diff --git a/chat/runtime.py b/chat/runtime.py index 310709fe..20bea64c 100644 --- a/chat/runtime.py +++ b/chat/runtime.py @@ -40,6 +40,7 @@ from core.epistemic_state import ( from chat.telemetry import ( TurnEventSink, format_correction_event_jsonl, + format_reboot_event_jsonl, format_turn_event_jsonl, ) from chat.verdicts import TurnVerdicts @@ -661,6 +662,10 @@ class ChatRuntime: self._pending_recognizer_examples: list[ tuple[tuple[str, ...], FeatureBundle] ] = [] + # W-024 / ADR-0158 — reboot event JSONL line buffered here when a + # checkpoint is loaded, flushed to the sink on attach_telemetry_sink. + # None means no reboot was detected this session. + self._pending_reboot_payload: str | None = None if self._engine_state_store is not None and self._engine_state_store.exists(): self._load_engine_state() @@ -668,12 +673,20 @@ class ChatRuntime: store = self._engine_state_store if store is None: return - self._recognizer_registry = RecognizerRegistry.from_recognizers( - store.load_recognizers() - ) + recognizers = store.load_recognizers() + self._recognizer_registry = RecognizerRegistry.from_recognizers(recognizers) self._pending_candidates = store.load_discovery_candidates() manifest = store.load_manifest() or {} self._turn_count = int(manifest.get("turn_count", 0)) + # W-024 / ADR-0158 — buffer reboot event for emission when sink attaches. + from engine_state import _git_revision + self._pending_reboot_payload = format_reboot_event_jsonl( + restored_turn_count=self._turn_count, + stored_revision=str(manifest.get("written_at_revision", "unknown")), + current_revision=_git_revision(), + recognizers_count=len(recognizers), + candidates_count=len(self._pending_candidates), + ) if self.config.auto_proposal_enabled and self._pending_candidates: _auto_propose_from_candidates(self._pending_candidates) @@ -813,6 +826,10 @@ class ChatRuntime: """ADR-0040 — attach a structured-logging sink.""" self._telemetry_sink = sink self._telemetry_include_content = bool(include_content) + # W-024 / ADR-0158 — flush buffered reboot event now that sink is live. + if sink is not None and self._pending_reboot_payload is not None: + sink.emit(self._pending_reboot_payload) + self._pending_reboot_payload = None def attach_articulation_sink(self, sink: Any | None) -> None: """Phase 5 — attach a sink for per-turn articulation observations. diff --git a/chat/telemetry.py b/chat/telemetry.py index 1529bb4e..815bbb8c 100644 --- a/chat/telemetry.py +++ b/chat/telemetry.py @@ -256,6 +256,75 @@ def format_correction_event_jsonl(correction_result, **kwargs) -> str: return json.dumps(payload, sort_keys=True, separators=(",", ":")) +# ---------- ADR-0158 — reboot-event serializer ---------- + + +def serialize_reboot_event( + *, + restored_turn_count: int, + stored_revision: str, + current_revision: str, + recognizers_count: int, + candidates_count: int, + timestamp: str | None = None, +) -> dict[str, object]: + """Produce a JSON-safe audit dict for an engine-state reboot recovery. + + Written to the telemetry JSONL when ``ChatRuntime`` loads an existing + checkpoint at startup (L10 scope §Sub-question 3). Lets future audit + reconstruct when this engine instance lost and regained its lifetime. + + ``revision_matched`` is True iff both revisions are known and equal, + indicating the checkpoint was written by the same code that is running. + False means the operator should verify the loaded state is still valid + (the W-023 ``RuntimeWarning`` will already have fired). + + Trust boundary: no surface text or tokens; metadata only. + """ + revision_matched = bool( + stored_revision not in ("unknown", "") + and current_revision not in ("unknown", "") + and stored_revision == current_revision + ) + out: dict[str, object] = { + "type": "reboot", + "restored_turn_count": int(restored_turn_count), + "stored_revision": str(stored_revision), + "current_revision": str(current_revision), + "revision_matched": revision_matched, + "recognizers_count": int(recognizers_count), + "candidates_count": int(candidates_count), + } + if timestamp is not None: + out["timestamp"] = str(timestamp) + return out + + +def format_reboot_event_jsonl( + *, + restored_turn_count: int, + stored_revision: str, + current_revision: str, + recognizers_count: int, + candidates_count: int, + timestamp: str | None = None, +) -> str: + """Serialize a reboot event as one deterministic JSONL line. + + ``"type": "reboot"`` discriminates this line from turn and correction + events at consume time without changing the sink contract. + """ + payload = serialize_reboot_event( + restored_turn_count=restored_turn_count, + stored_revision=stored_revision, + current_revision=current_revision, + recognizers_count=recognizers_count, + candidates_count=candidates_count, + timestamp=timestamp, + ) + return json.dumps(payload, sort_keys=True, separators=(",", ":")) + + # ---------- sink protocol ---------- @@ -404,8 +473,10 @@ __all__ = [ "JsonlFileSink", "TurnEventSink", "format_correction_event_jsonl", + "format_reboot_event_jsonl", "format_turn_event_jsonl", "format_verdict_summary", "serialize_correction_event", + "serialize_reboot_event", "serialize_turn_event", ] diff --git a/docs/decisions/ADR-0158-reboot-event-audit.md b/docs/decisions/ADR-0158-reboot-event-audit.md new file mode 100644 index 00000000..7cfd45fd --- /dev/null +++ b/docs/decisions/ADR-0158-reboot-event-audit.md @@ -0,0 +1,97 @@ +# ADR-0158 — reboot_event audit trail entry (W-024 / L10b.3) + +Status: accepted +Date: 2026-05-26 + +## Context + +L10 scope §Sub-question 3 asked: + +> **What does reboot record?** A `reboot_event` analog of `TurnEvent`, +> written to the audit trail, that lets future audit reconstruct the fact +> that this engine instance lost and regained its lifetime here. + +ADR-0156 §"Out of scope" deferred this to L10b.3 / W-024. After W-022 +(atomic writes) and W-023 (revision-mismatch warning), the checkpoint path +is now correct and observable. The missing piece is the audit record itself: +without it, the telemetry JSONL cannot distinguish a fresh start from a +reboot recovery, making the engine's lifetime history uninterpretable. + +## Decision + +### Serializer (`chat/telemetry.py`) + +Add `serialize_reboot_event(...)` and `format_reboot_event_jsonl(...)`, +following the same pattern as `serialize_correction_event` / ADR-0059: + +```jsonc +{ + "type": "reboot", // discriminator + "restored_turn_count": 5, // manifest["turn_count"] + "stored_revision": "abc123", // manifest["written_at_revision"] + "current_revision": "def456", // _git_revision() at load time + "revision_matched": false, // true iff both known and equal + "recognizers_count": 2, // DerivedRecognizers restored + "candidates_count": 1, // DiscoveryCandidates restored + "timestamp": "..." // optional, caller-provided +} +``` + +`revision_matched` is `False` when either side is `"unknown"` or `""` — +the same guard as W-023's warning suppression, keeping the two features +consistent. + +### Runtime wiring (`chat/runtime.py`) + +The telemetry sink is `None` at `__init__` time — attached later via +`attach_telemetry_sink()`. Two-step buffered emission: + +1. **`_load_engine_state()`** — builds and stores the JSONL string in + `self._pending_reboot_payload: str | None`. The string is the fully + serialized line; no further computation required at flush time. + +2. **`attach_telemetry_sink(sink)`** — if `_pending_reboot_payload` is + set and `sink` is not None, emits the line and clears the field. + Subsequent calls to `attach_telemetry_sink` (e.g. replacing the sink) + do NOT re-emit; the payload is consumed exactly once. + +This placement guarantees the reboot event appears **before any turn +events** in the audit stream for the session, matching the semantic intent +of "lifetime boundary record". + +### Trust boundary + +Metadata only — no surface text, no tokens, no versor coordinates. +`_git_revision()` returns a short SHA derived from the local git repo; +already used for the manifest and W-023 warning. + +## Invariants pinned by tests + +`tests/test_adr_0158_reboot_event.py` (11 tests): + +**Serializer:** +- Structure, fields, and `revision_matched` logic (match / mismatch / unknown) +- Optional `timestamp` field +- Byte-identical output for identical inputs (determinism) + +**Runtime integration:** +- Reboot event emitted to sink on `attach_telemetry_sink` after checkpoint load +- Emitted exactly once — second `attach_telemetry_sink` call does not re-emit +- No event when no checkpoint directory exists (fresh start) +- No event when `no_load_state=True` +- `revision_matched: true` when revisions agree +- Reboot event precedes synthetic turn events in the stream + +## L10b closure + +With W-024, the L10b sequence is complete: + +| Chunk | Work item | ADR | What it does | +|---|---|---|---| +| L10b.1 | W-022 | ADR-0156 | Atomic checkpoint writes (write-temp + fsync + rename) | +| L10b.2 | W-023 | ADR-0157 | Revision-mismatch `RuntimeWarning` on load | +| L10b.3 | W-024 | ADR-0158 | `reboot_event` audit trail entry | + +The checkpoint path now satisfies all three of ADR-0146's runtime-model +commitments: **atomic**, **observable** (warning on staleness), and +**auditable** (reboot recorded in the telemetry stream). diff --git a/engine_state/__init__.py b/engine_state/__init__.py index 1758a12c..1d15b1c6 100644 --- a/engine_state/__init__.py +++ b/engine_state/__init__.py @@ -18,6 +18,7 @@ import stat import subprocess import tempfile import warnings +from functools import lru_cache from pathlib import Path from typing import Sequence @@ -78,7 +79,14 @@ _DEFAULT_DIR = ( ) -def _git_revision() -> str: +@lru_cache(maxsize=1) +def get_git_revision() -> str: + """Return the current short git revision once per process. + + Public helper for runtime audit surfaces that need the same revision + value used by engine-state manifests and revision-mismatch warnings. + Cached to avoid duplicate subprocess calls during startup. + """ try: return ( subprocess.run( @@ -93,6 +101,11 @@ def _git_revision() -> str: return "unknown" +def _git_revision() -> str: + """Backward-compatible private alias; use get_git_revision() in new code.""" + return get_git_revision() + + class EngineStateStore: def __init__(self, path: Path | None = None) -> None: self.path = path or _DEFAULT_DIR @@ -141,7 +154,7 @@ class EngineStateStore: manifest = { "schema_version": _SCHEMA_VERSION, "turn_count": turn_count, - "written_at_revision": _git_revision(), + "written_at_revision": get_git_revision(), } _atomic_write_text( self.path / "manifest.json", @@ -159,7 +172,7 @@ class EngineStateStore: # W-023 / ADR-0157 — revision-mismatch warning per ADR-0146 §Risks line 127. # Never refuse to load; reboot is recovery, not control flow. stored_rev = manifest.get("written_at_revision", "unknown") - current_rev = _git_revision() + current_rev = get_git_revision() if stored_rev not in ("unknown", "") and current_rev not in ("unknown", "") and stored_rev != current_rev: warnings.warn( f"engine_state checkpoint was written at revision {stored_rev!r} " @@ -175,4 +188,4 @@ class EngineStateStore: return (self.path / "manifest.json").exists() -__all__ = ["EngineStateStore"] +__all__ = ["EngineStateStore", "get_git_revision"] diff --git a/tests/test_adr_0158_reboot_event.py b/tests/test_adr_0158_reboot_event.py new file mode 100644 index 00000000..c20bdaae --- /dev/null +++ b/tests/test_adr_0158_reboot_event.py @@ -0,0 +1,253 @@ +"""ADR-0158 (W-024) — reboot_event audit trail entry on engine-state load. + +L10 scope §Sub-question 3: + A ``reboot_event`` analog of ``TurnEvent``, written to the audit trail, + that lets future audit reconstruct the fact that this engine instance + lost and regained its lifetime here. +""" + +from __future__ import annotations + +import json +from unittest.mock import patch + +import pytest + +from chat.runtime import ChatRuntime +from chat.telemetry import ( + JsonlBufferSink, + format_reboot_event_jsonl, + serialize_reboot_event, +) +from engine_state import EngineStateStore +from recognition.anti_unifier import Constant, DerivedRecognizer, TypedSlot +from teaching.discovery import DiscoveryCandidate, EvidencePointer, SubQuestion + + +# ---------- fixtures ---------- + + +def _recognizer() -> DerivedRecognizer: + return DerivedRecognizer( + pattern=( + Constant("light"), + TypedSlot( + feature_name="object", + slot_type="noun", + min_width=1, + max_width=2, + ignored_prefix_tokens=("the",), + ), + ), + teaching_set_id="set-1", + constant_features={"intent": "definition"}, + absent_features={"negated": 0}, + ) + + +def _candidate() -> DiscoveryCandidate: + ev = EvidencePointer( + source="pack", + ref="en_core_cognition_v1:light", + polarity="affirms", + epistemic_status="ratified", + ) + sq = SubQuestion( + sub_id="sub-1", + proposed_subject="light", + proposed_intent="verification", + outcome="grounded", + evidence=(ev,), + ) + return DiscoveryCandidate( + candidate_id="cand-1", + proposed_chain={"subject": "light", "intent": "verification"}, + trigger="would_have_grounded", + source_turn_trace="trace-1", + pack_consistent=True, + boundary_clean=True, + polarity="affirms", + claim_domain="factual", + evidence=(ev,), + sub_questions=(sq,), + contemplation_depth=1, + ) + + +def _populated_store(tmp_path, *, turn_count: int = 5, revision: str = "abc123") -> EngineStateStore: + store = EngineStateStore(tmp_path) + store.save_recognizers([_recognizer()]) + store.save_discovery_candidates([_candidate()]) + store.save_manifest(turn_count=turn_count) + # Overwrite manifest to inject a controlled revision for testing. + import json as _json + manifest = { + "schema_version": 1, + "turn_count": turn_count, + "written_at_revision": revision, + } + (tmp_path / "manifest.json").write_text( + _json.dumps(manifest, sort_keys=True, indent=2), encoding="utf-8" + ) + return store + + +# ---------- serializer unit tests ---------- + + +def test_serialize_reboot_event_structure() -> None: + payload = serialize_reboot_event( + restored_turn_count=7, + stored_revision="abc123", + current_revision="abc123", + recognizers_count=2, + candidates_count=1, + ) + + assert payload["type"] == "reboot" + assert payload["restored_turn_count"] == 7 + assert payload["stored_revision"] == "abc123" + assert payload["current_revision"] == "abc123" + assert payload["revision_matched"] is True + assert payload["recognizers_count"] == 2 + assert payload["candidates_count"] == 1 + assert "timestamp" not in payload + + +def test_serialize_reboot_event_mismatch_revision() -> None: + payload = serialize_reboot_event( + restored_turn_count=3, + stored_revision="old000", + current_revision="new111", + recognizers_count=0, + candidates_count=0, + ) + + assert payload["revision_matched"] is False + + +def test_serialize_reboot_event_unknown_revision_not_matched() -> None: + payload = serialize_reboot_event( + restored_turn_count=1, + stored_revision="unknown", + current_revision="abc123", + recognizers_count=0, + candidates_count=0, + ) + + assert payload["revision_matched"] is False + + +def test_serialize_reboot_event_with_timestamp() -> None: + payload = serialize_reboot_event( + restored_turn_count=0, + stored_revision="a", + current_revision="a", + recognizers_count=0, + candidates_count=0, + timestamp="2026-05-26T00:00:00Z", + ) + + assert payload["timestamp"] == "2026-05-26T00:00:00Z" + + +def test_format_reboot_event_jsonl_is_deterministic() -> None: + line1 = format_reboot_event_jsonl( + restored_turn_count=4, + stored_revision="rev1", + current_revision="rev2", + recognizers_count=1, + candidates_count=0, + ) + line2 = format_reboot_event_jsonl( + restored_turn_count=4, + stored_revision="rev1", + current_revision="rev2", + recognizers_count=1, + candidates_count=0, + ) + + assert line1 == line2 + parsed = json.loads(line1) + assert parsed["type"] == "reboot" + + +# ---------- runtime integration tests ---------- + + +def test_reboot_event_emitted_when_sink_attached_after_load(tmp_path) -> None: + _populated_store(tmp_path, turn_count=5, revision="stored_rev") + sink = JsonlBufferSink() + + with patch("engine_state._git_revision", return_value="current_rev"): + runtime = ChatRuntime(engine_state_path=tmp_path) + runtime.attach_telemetry_sink(sink) + + assert len(sink.lines) == 1 + event = json.loads(sink.lines[0]) + assert event["type"] == "reboot" + assert event["restored_turn_count"] == 5 + assert event["stored_revision"] == "stored_rev" + assert event["current_revision"] == "current_rev" + assert event["revision_matched"] is False + assert event["recognizers_count"] == 1 + assert event["candidates_count"] == 1 + + +def test_reboot_event_emitted_once_not_twice(tmp_path) -> None: + _populated_store(tmp_path) + sink = JsonlBufferSink() + + runtime = ChatRuntime(engine_state_path=tmp_path) + runtime.attach_telemetry_sink(sink) + # Attaching a second sink should NOT re-emit the reboot event. + sink2 = JsonlBufferSink() + runtime.attach_telemetry_sink(sink2) + + assert len(sink.lines) == 1 + assert len(sink2.lines) == 0 + + +def test_no_reboot_event_when_no_checkpoint(tmp_path) -> None: + sink = JsonlBufferSink() + + runtime = ChatRuntime(engine_state_path=tmp_path) + runtime.attach_telemetry_sink(sink) + + assert sink.lines == [] + + +def test_no_reboot_event_when_no_load_state(tmp_path) -> None: + _populated_store(tmp_path) + sink = JsonlBufferSink() + + runtime = ChatRuntime(no_load_state=True, engine_state_path=tmp_path) + runtime.attach_telemetry_sink(sink) + + assert sink.lines == [] + + +def test_reboot_event_revision_matched_true(tmp_path) -> None: + _populated_store(tmp_path, revision="matchedrev") + sink = JsonlBufferSink() + + with patch("engine_state._git_revision", return_value="matchedrev"): + runtime = ChatRuntime(engine_state_path=tmp_path) + runtime.attach_telemetry_sink(sink) + + event = json.loads(sink.lines[0]) + assert event["revision_matched"] is True + + +def test_reboot_event_precedes_turn_events(tmp_path) -> None: + _populated_store(tmp_path, turn_count=2) + sink = JsonlBufferSink() + + runtime = ChatRuntime(engine_state_path=tmp_path) + runtime.attach_telemetry_sink(sink) + # Emit a synthetic turn event stub directly to verify ordering. + sink.emit('{"type":"turn","turn":0}') + + assert len(sink.lines) == 2 + first = json.loads(sink.lines[0]) + assert first["type"] == "reboot"