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

feat(W-024): reboot_event audit trail entry (L10b.3, ADR-0158) (#284)

Browse source 
* feat(W-024): reboot_event audit trail entry (L10b.3, ADR-0158)

L10 scope §Sub-question 3: a reboot_event analog of TurnEvent, written
to the telemetry JSONL, lets future audit reconstruct when this engine
instance lost and regained its lifetime.

- serialize_reboot_event / format_reboot_event_jsonl in chat/telemetry.py
  emit type="reboot" with restored_turn_count, stored/current revisions,
  revision_matched, recognizers_count, candidates_count
- ChatRuntime._load_engine_state() buffers the JSONL line in
  _pending_reboot_payload (str|None); ChatRuntime.attach_telemetry_sink()
  flushes it exactly once when a sink is first attached
- Reboot event precedes all turn events in the session audit stream
- Pinned by 11 tests: serializer structure, determinism, revision_matched
  logic, runtime integration (emit-once, no-checkpoint, no-load-state,
  revision match, ordering)

Closes L10b: W-022 (atomic writes) + W-023 (revision warning) + W-024
together satisfy ADR-0146's atomic/observable/auditable checkpoint triad.

* fix(W-024): expose cached public git revision helper
This commit is contained in:
Shay 2026-05-25 20:37:00 -07:00 committed by GitHub
parent 26237f2335
commit 5045700484
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 458 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

23
chat/runtime.py
View file

@ -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.

71
chat/telemetry.py
View file

@ -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",
]

97
docs/decisions/ADR-0158-reboot-event-audit.md Normal file
View file

@ -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).

21
engine_state/__init__.py
View file

@ -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"]

253
tests/test_adr_0158_reboot_event.py Normal file
View file

@ -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"