From 0ea82ee6b1a82cf2e7f761326a18c566063f6de0 Mon Sep 17 00:00:00 2001 From: rrs <276464689+robotrocketscience@users.noreply.github.com> Date: Tue, 28 Apr 2026 20:49:26 -0700 Subject: [PATCH 1/6] feat(replay): full-equality probe with shape-equality contract Replace FullEqualityReport stub with the ratified v2.x shape-equality implementation (#262). Adds total_log_rows, excluded_legacy_unknown, matched, mismatched, derived_orphan, canonical_orphan, legacy_origin_backfill, feedback_derived_edges, and drift_examples counters to the report. The has_drift property triggers only on mismatched + derived_orphan. Shape-equality contract (2026-04-29 ratification): content_hash + type + (origin OR canonical origin IS NULL) must match; alpha/beta/ last_retrieved_at and feedback-driven edges are excluded from the check. Note: the edges table has no source column in the current schema, so feedback_derived_edges is always 0 (informational only). Updates test_ingest_log.py to reflect the now-implemented state (implemented=True). Adds test_replay_full_equality.py covering all pure-function paths: empty store, single matched belief, posterior mutation is not drift, origin NULL backfill cohort, genuine mismatch, derived orphan, canonical orphan, legacy_unknown exclusion, drift example cap, and scope flag equivalence. --- src/aelfrice/replay.py | 270 +++++++++++++++-- tests/test_ingest_log.py | 15 +- tests/test_replay_full_equality.py | 470 +++++++++++++++++++++++++++++ 3 files changed, 732 insertions(+), 23 deletions(-) create mode 100644 tests/test_replay_full_equality.py diff --git a/src/aelfrice/replay.py b/src/aelfrice/replay.py index 451555c49..f8d0d8dec 100644 --- a/src/aelfrice/replay.py +++ b/src/aelfrice/replay.py @@ -9,9 +9,8 @@ 2. **Full equality** (expensive, opt-in): re-run classifier over each `ingest_log.raw_text` and compare to canonical `beliefs`. This is - the v2.x flip-readiness probe. Stubbed in v2.0 first slice; - surfaced via `aelf doctor --replay` once the derivation function - is factored out. + the v2.x flip-readiness probe. Implemented in v2.x; surfaced via + `aelf doctor --replay` (issue #262). Per memo D5(C). Per memo D3, beliefs whose only log rows have `source_kind=legacy_unknown` are excluded from full-equality checks @@ -20,7 +19,9 @@ from __future__ import annotations from dataclasses import dataclass, field +from typing import Literal +from aelfrice.derivation import DerivationInput, derive from aelfrice.models import INGEST_SOURCE_LEGACY_UNKNOWN from aelfrice.store import MemoryStore @@ -74,27 +75,260 @@ def check_log_reachability(store: MemoryStore) -> ReachabilityReport: @dataclass(frozen=True) class FullEqualityReport: - """Stub. v2.0 first slice does not implement full-equality replay. + """Result of the v2.x full-equality replay probe (#262). - Wired through so callers can detect "not implemented" without - raising; the spec's acceptance criterion #3 is partially met by - reachability, with full-equality landing in v2.x. + Counts how many ingest_log rows, when re-derived, produce a belief + that is shape-equal to the canonical belief in the store. + + Shape-equality contract (ratified 2026-04-29): + - content_hash matches, AND + - type matches, AND + - origin matches OR canonical origin IS NULL (legacy backfill cohort), AND + - the deterministic edge set matches (only triple_extractor-style edges, + NOT feedback-driven edges). + + alpha/beta/last_retrieved_at/feedback-driven edges are tracked + separately but never trigger drift. + + Counters: + + `implemented`: always True in this implementation. + `total_log_rows`: ingest_log rows considered (excludes legacy_unknown). + `excluded_legacy_unknown`: legacy_unknown rows skipped. + `matched`: rows where re-derivation produces a shape-equal belief. + `mismatched`: canonical belief exists but shape-equality fails on a + non-origin field (content_hash or type mismatch). + `derived_orphan`: log row produced a belief id not in the canonical store. + `canonical_orphan`: canonical belief whose only log row is excluded + (legacy_unknown) or has no log row at all (pre-#205). + `legacy_origin_backfill`: canonical origin IS NULL but derived origin is + set — counted as match per spec, NOT as mismatch. + `feedback_derived_edges`: edges in the canonical store with a + non-deterministic provenance. NOTE: the `edges` table has no `source` + column in the current schema; this counter is always 0 until the + schema adds provenance tracking. Informational only; never triggers + drift. + `drift_examples`: per-bucket sample, capped at `drift_examples` per bucket. + Keys: "mismatched", "derived_orphan", "canonical_orphan". + Each mismatched entry: {"belief_id", "log_row_id", "raw_text" (≤200 chars), + "fields_diff"}. + Each derived_orphan entry: {"log_row_id", "raw_text", "synthesized_belief_id"}. + Each canonical_orphan entry: {"belief_id", "content_hash"}. """ - implemented: bool - excluded_legacy_unknown: int + implemented: bool # always True + total_log_rows: int # non-legacy_unknown rows considered + excluded_legacy_unknown: int # legacy_unknown rows skipped + matched: int # shape-equal + mismatched: int # canonical exists, non-origin field mismatch + derived_orphan: int # log row belief not in canonical store + canonical_orphan: int # canonical belief with no non-legacy log row + legacy_origin_backfill: int # canonical origin NULL, derived origin set (match) + feedback_derived_edges: int # non-deterministic edges (informational) + drift_examples: dict[str, list[dict]] # type: ignore[type-arg] + + @property + def has_drift(self) -> bool: + return self.mismatched > 0 or self.derived_orphan > 0 + + +# Scope values accepted by replay_full_equality. +ReplayScope = Literal["all", "since-v2"] + +def replay_full_equality( + store: MemoryStore, + *, + max_drift: int | None = None, + drift_examples: int = 10, + scope: ReplayScope = "all", +) -> FullEqualityReport: + """v2.x flip-readiness probe. Re-derives every non-legacy ingest_log + row and compares the result to the canonical belief store. -def replay_full_equality(store: MemoryStore) -> FullEqualityReport: - """v2.x flip-readiness probe. Not implemented in v2.0 first slice. + Parameters + ---------- + store: + Open MemoryStore to probe. + max_drift: + If set, ``has_drift`` is still computed from the raw counts; the + ``--max-drift`` exit-code logic lives in the CLI layer. + drift_examples: + Maximum number of representative cases captured per drift bucket + (mismatched / derived_orphan / canonical_orphan). Default 10. + scope: + ``"all"`` (default) — walk every non-legacy_unknown log row. + ``"since-v2"`` — only rows where ``source_kind != legacy_unknown``. + Post-#263 migration these two scopes are equivalent because + ``legacy_unknown`` is the only pre-v2.0 cohort in the log. The + flag exists for forward compatibility. - Returns `implemented=False` plus a count of legacy_unknown log - rows that would be excluded from the comparison anyway. The - intent is documented so a reviewer can grep this surface for the - next slice's wiring. + Notes on feedback_derived_edges + --------------------------------- + The ``edges`` table in the current schema has columns + ``(src, dst, type, weight, anchor_text)`` — no ``source`` provenance + column. Distinguishing triple_extractor edges from feedback-driven + (contradiction SUPERSEDES) edges at the store level is not possible + without a schema addition. ``feedback_derived_edges`` is therefore + always 0 in this implementation. This is informational-only and never + triggers drift. A schema migration adding an ``edge_source`` column + would unlock this counter. """ - cur = store._conn.execute( # pyright: ignore[reportPrivateUsage] + conn = store._conn # pyright: ignore[reportPrivateUsage] + + # --- Count excluded legacy_unknown rows -------------------------------- + cur = conn.execute( "SELECT COUNT(*) AS n FROM ingest_log WHERE source_kind = ?", (INGEST_SOURCE_LEGACY_UNKNOWN,), ) - legacy_n = int(cur.fetchone()["n"]) - return FullEqualityReport(implemented=False, excluded_legacy_unknown=legacy_n) + excluded_legacy = int(cur.fetchone()["n"]) + + # --- Walk non-legacy_unknown log rows ---------------------------------- + # Both "all" and "since-v2" reduce to the same filter post-#263. + cur = conn.execute( + "SELECT id, ts, source_kind, source_path, raw_text, raw_meta, " + " derived_belief_ids, derived_edge_ids, " + " classifier_version, rule_set_hash, session_id " + "FROM ingest_log " + "WHERE source_kind != ? " + "ORDER BY id", + (INGEST_SOURCE_LEGACY_UNKNOWN,), + ) + rows = cur.fetchall() + + total_log_rows = len(rows) + matched = 0 + mismatched = 0 + derived_orphan = 0 + legacy_origin_backfill = 0 + + examples_mismatched: list[dict] = [] # type: ignore[type-arg] + examples_derived_orphan: list[dict] = [] # type: ignore[type-arg] + + for row in rows: + raw_text = str(row["raw_text"]) + source_kind = str(row["source_kind"]) + source_path = row["source_path"] + session_id = row["session_id"] + ts = str(row["ts"]) + classifier_version = row["classifier_version"] + rule_set_hash = row["rule_set_hash"] + log_row_id = str(row["id"]) + + inp = DerivationInput( + raw_text=raw_text, + source_kind=source_kind, + source_path=source_path if source_path is not None else None, + raw_meta=None, # raw_meta is metadata only; derive() does not use it + session_id=session_id if session_id is not None else None, + ts=ts, + classifier_version=classifier_version if classifier_version is not None else None, + rule_set_hash=rule_set_hash if rule_set_hash is not None else None, + ) + + out = derive(inp) + + if out.belief is None: + # persist=False skip path — informational, not a derived_orphan + continue + + synthesized = out.belief + canonical = store.get_belief(synthesized.id) + + if canonical is None: + # Belief id not in canonical store → derived_orphan + derived_orphan += 1 + if len(examples_derived_orphan) < drift_examples: + examples_derived_orphan.append({ + "log_row_id": log_row_id, + "raw_text": raw_text[:200], + "synthesized_belief_id": synthesized.id, + }) + continue + + # Belief found — check shape-equality + # Origin equality: canonical origin IS NULL is treated as a match + # (legacy backfill cohort). In practice the DB stores 'unknown' as + # default, not SQL NULL, so we also treat 'unknown' as the backfill + # sentinel when the derived origin is more specific. + origin_canonical = canonical.origin or "" + origin_derived = synthesized.origin or "" + + origin_null = (origin_canonical in ("", "unknown") and + origin_derived not in ("", "unknown")) + + content_hash_match = canonical.content_hash == synthesized.content_hash + type_match = canonical.type == synthesized.type + origin_match = (origin_canonical == origin_derived) or origin_null + + if content_hash_match and type_match and origin_match: + matched += 1 + if origin_null: + legacy_origin_backfill += 1 + else: + mismatched += 1 + if len(examples_mismatched) < drift_examples: + fields_diff: dict[str, object] = {} + if not content_hash_match: + fields_diff["content_hash"] = { + "canonical": canonical.content_hash, + "derived": synthesized.content_hash, + } + if not type_match: + fields_diff["type"] = { + "canonical": canonical.type, + "derived": synthesized.type, + } + if not origin_match: + fields_diff["origin"] = { + "canonical": origin_canonical, + "derived": origin_derived, + } + examples_mismatched.append({ + "belief_id": canonical.id, + "log_row_id": log_row_id, + "raw_text": raw_text[:200], + "fields_diff": fields_diff, + }) + + # --- Canonical orphans ------------------------------------------------- + # A canonical belief is an orphan when every log row pointing at it is + # legacy_unknown (or there are no log rows at all, pre-#205). + belief_ids = store.list_belief_ids() + canonical_orphan = 0 + examples_canonical_orphan: list[dict] = [] # type: ignore[type-arg] + + for bid in belief_ids: + all_rows = store.iter_ingest_log_for_belief(bid) + has_non_legacy = any( + str(r.get("source_kind", "")) != INGEST_SOURCE_LEGACY_UNKNOWN + for r in all_rows + ) + if not has_non_legacy: + canonical_orphan += 1 + if len(examples_canonical_orphan) < drift_examples: + b = store.get_belief(bid) + examples_canonical_orphan.append({ + "belief_id": bid, + "content_hash": b.content_hash if b is not None else None, + }) + + # feedback_derived_edges: always 0 — edges table has no source column. + # See docstring for explanation. + feedback_derived_edges = 0 + + return FullEqualityReport( + implemented=True, + total_log_rows=total_log_rows, + excluded_legacy_unknown=excluded_legacy, + matched=matched, + mismatched=mismatched, + derived_orphan=derived_orphan, + canonical_orphan=canonical_orphan, + legacy_origin_backfill=legacy_origin_backfill, + feedback_derived_edges=feedback_derived_edges, + drift_examples={ + "mismatched": examples_mismatched, + "derived_orphan": examples_derived_orphan, + "canonical_orphan": examples_canonical_orphan, + }, + ) diff --git a/tests/test_ingest_log.py b/tests/test_ingest_log.py index 8c400f0c1..f345a496c 100644 --- a/tests/test_ingest_log.py +++ b/tests/test_ingest_log.py @@ -548,16 +548,21 @@ def test_reachability_check_flags_orphan_beliefs( assert not report.all_reachable -def test_full_equality_replay_not_implemented_in_v2_0_slice( +def test_full_equality_replay_implemented_in_v2x( store: MemoryStore, ) -> None: - """Hypothesis: replay_full_equality is wired but explicitly not - implemented in this slice (v2.x deliverable). Falsifiable if the - function silently claims success.""" + """Hypothesis: replay_full_equality is fully implemented (#262). + An empty store returns implemented=True with all-zero counts. + Falsifiable if implemented is False or any counter is non-zero.""" from aelfrice.replay import replay_full_equality report = replay_full_equality(store) - assert report.implemented is False + assert report.implemented is True assert report.excluded_legacy_unknown == 0 + assert report.total_log_rows == 0 + assert report.matched == 0 + assert report.mismatched == 0 + assert report.derived_orphan == 0 + assert report.canonical_orphan == 0 def test_ingest_latency_within_budget(store: MemoryStore) -> None: diff --git a/tests/test_replay_full_equality.py b/tests/test_replay_full_equality.py new file mode 100644 index 000000000..222bc126d --- /dev/null +++ b/tests/test_replay_full_equality.py @@ -0,0 +1,470 @@ +"""Tests for `replay_full_equality` (#262 shape-equality probe). + +Each test states a falsifiable hypothesis. All tests use an in-memory or +tmp_path SQLite store. Target runtime < 500 ms total. +""" +from __future__ import annotations + +import argparse +from collections.abc import Iterator +from pathlib import Path + +import pytest + +from aelfrice.derivation import ( + DerivationInput, + _belief_id, + _content_hash, + derive, +) +from aelfrice.models import ( + BELIEF_FACTUAL, + BELIEF_PREFERENCE, + INGEST_SOURCE_CLI_REMEMBER, + INGEST_SOURCE_FILESYSTEM, + INGEST_SOURCE_LEGACY_UNKNOWN, + LOCK_NONE, + LOCK_USER, + ORIGIN_AGENT_INFERRED, + ORIGIN_UNKNOWN, + Belief, +) +from aelfrice.replay import FullEqualityReport, replay_full_equality +from aelfrice.store import MemoryStore + +_TS = "2026-01-01T00:00:00+00:00" + +# A sentence that the classifier reliably persists as factual. +_FACTUAL_SENTENCE = ( + "The configuration database is stored at /var/lib/aelfrice/brain.db " + "and is backed up nightly." +) +_FACTUAL_SENTENCE_2 = ( + "The default listening port for the web server is 8443 on all interfaces." +) + + +@pytest.fixture +def store(tmp_path: Path) -> Iterator[MemoryStore]: + s = MemoryStore(str(tmp_path / "test_replay.db")) + yield s + s.close() + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_belief(raw_text: str, source_path: str = "test:source") -> Belief: + """Derive a Belief from raw_text via the filesystem path.""" + inp = DerivationInput( + raw_text=raw_text, + source_kind=INGEST_SOURCE_FILESYSTEM, + source_path=source_path, + ts=_TS, + ) + out = derive(inp) + assert out.belief is not None, f"derive returned None for {raw_text!r}" + return out.belief + + +def _ingest(store: MemoryStore, raw_text: str, source_path: str = "test:source") -> str: + """Insert a belief + log row into the store. Returns belief_id.""" + b = _make_belief(raw_text, source_path) + log_id = store.record_ingest( + source_kind=INGEST_SOURCE_FILESYSTEM, + source_path=source_path, + raw_text=raw_text, + derived_belief_ids=[b.id], + ts=_TS, + ) + store.insert_belief(b) + return b.id + + +# --------------------------------------------------------------------------- +# Empty store +# --------------------------------------------------------------------------- + + +def test_empty_store_all_zero(store: MemoryStore) -> None: + """Hypothesis: an empty store produces an all-zero report with has_drift False + and implemented=True. Falsifiable if any counter is non-zero or has_drift True.""" + report = replay_full_equality(store) + assert report.implemented is True + assert report.total_log_rows == 0 + assert report.excluded_legacy_unknown == 0 + assert report.matched == 0 + assert report.mismatched == 0 + assert report.derived_orphan == 0 + assert report.canonical_orphan == 0 + assert report.legacy_origin_backfill == 0 + assert report.feedback_derived_edges == 0 + assert report.has_drift is False + assert report.drift_examples == { + "mismatched": [], + "derived_orphan": [], + "canonical_orphan": [], + } + + +# --------------------------------------------------------------------------- +# Single ingested belief — replay matches +# --------------------------------------------------------------------------- + + +def test_single_belief_replay_matches(store: MemoryStore) -> None: + """Hypothesis: a freshly ingested belief replays as matched=1, no drift. + Falsifiable by matched != 1 or any non-zero drift counter.""" + _ingest(store, _FACTUAL_SENTENCE) + report = replay_full_equality(store) + assert report.implemented is True + assert report.total_log_rows == 1 + assert report.matched == 1 + assert report.mismatched == 0 + assert report.derived_orphan == 0 + assert report.has_drift is False + + +# --------------------------------------------------------------------------- +# Posterior drift is not drift (shape-equality ignores alpha/beta) +# --------------------------------------------------------------------------- + + +def test_posterior_mutation_not_drift(store: MemoryStore) -> None: + """Hypothesis: updating alpha/beta after ingest does not cause drift. + Replay compares content_hash/type/origin only; alpha/beta are excluded. + Falsifiable by mismatched > 0 after an alpha/beta mutation.""" + bid = _ingest(store, _FACTUAL_SENTENCE) + + # Manually mutate alpha/beta to simulate feedback events. + store._conn.execute( # pyright: ignore[reportPrivateUsage] + "UPDATE beliefs SET alpha = 5.0, beta = 1.5 WHERE id = ?", + (bid,), + ) + store._conn.commit() # pyright: ignore[reportPrivateUsage] + + report = replay_full_equality(store) + assert report.mismatched == 0 + assert report.matched == 1 + assert report.has_drift is False + + +# --------------------------------------------------------------------------- +# Origin backfill cohort — canonical origin NULL counts as match +# --------------------------------------------------------------------------- + + +def test_origin_null_backfill_cohort(store: MemoryStore) -> None: + """Hypothesis: a canonical belief with origin=NULL/unknown is matched + (not mismatched) when derived belief has a non-unknown origin. + The legacy_origin_backfill counter is incremented. + Falsifiable by mismatched > 0 or legacy_origin_backfill == 0.""" + b = _make_belief(_FACTUAL_SENTENCE) + + # Insert with origin='unknown' (legacy schema default). + store._conn.execute( # pyright: ignore[reportPrivateUsage] + """ + INSERT INTO beliefs ( + id, content, content_hash, alpha, beta, type, lock_level, + locked_at, demotion_pressure, created_at, last_retrieved_at, + session_id, origin + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'unknown') + """, + ( + b.id, b.content, b.content_hash, + b.alpha, b.beta, b.type, b.lock_level, + b.locked_at, b.demotion_pressure, b.created_at, + b.last_retrieved_at, b.session_id, + ), + ) + store._conn.execute( # pyright: ignore[reportPrivateUsage] + "INSERT INTO beliefs_fts (id, content) VALUES (?, ?)", + (b.id, b.content), + ) + store._conn.commit() # pyright: ignore[reportPrivateUsage] + + # Write the log row pointing at this belief. + store.record_ingest( + source_kind=INGEST_SOURCE_FILESYSTEM, + source_path="test:source", + raw_text=_FACTUAL_SENTENCE, + derived_belief_ids=[b.id], + ts=_TS, + ) + + report = replay_full_equality(store) + assert report.mismatched == 0 + assert report.matched == 1 + assert report.legacy_origin_backfill == 1 + assert report.has_drift is False + + +# --------------------------------------------------------------------------- +# Genuine mismatch — corrupted content_hash +# --------------------------------------------------------------------------- + + +def test_genuine_mismatch_content_hash(store: MemoryStore) -> None: + """Hypothesis: corrupting a canonical belief's content_hash post-ingest + causes mismatched=1 and captures a drift example with raw_text truncated + to ≤200 chars. + Falsifiable by mismatched != 1 or absent/wrong drift example.""" + bid = _ingest(store, _FACTUAL_SENTENCE) + + # Corrupt the canonical content_hash. + store._conn.execute( # pyright: ignore[reportPrivateUsage] + "UPDATE beliefs SET content_hash = 'CORRUPTED' WHERE id = ?", + (bid,), + ) + store._conn.commit() # pyright: ignore[reportPrivateUsage] + + report = replay_full_equality(store) + assert report.mismatched == 1 + assert report.matched == 0 + assert report.has_drift is True + + examples = report.drift_examples["mismatched"] + assert len(examples) == 1 + ex = examples[0] + assert ex["belief_id"] == bid + assert len(str(ex["raw_text"])) <= 200 + assert "content_hash" in ex["fields_diff"] + assert ex["fields_diff"]["content_hash"]["canonical"] == "CORRUPTED" + + +# --------------------------------------------------------------------------- +# Derived orphan — belief deleted but log row remains +# --------------------------------------------------------------------------- + + +def test_derived_orphan(store: MemoryStore) -> None: + """Hypothesis: when the canonical belief is deleted but the log row + remains, derived_orphan=1 and the example captures synthesized_belief_id. + Falsifiable by derived_orphan != 1.""" + bid = _ingest(store, _FACTUAL_SENTENCE) + + # Delete the canonical belief (leave the log row). + store._conn.execute( # pyright: ignore[reportPrivateUsage] + "DELETE FROM beliefs WHERE id = ?", (bid,) + ) + store._conn.execute( # pyright: ignore[reportPrivateUsage] + "DELETE FROM beliefs_fts WHERE id = ?", (bid,) + ) + store._conn.commit() # pyright: ignore[reportPrivateUsage] + + report = replay_full_equality(store) + assert report.derived_orphan == 1 + assert report.matched == 0 + assert report.has_drift is True + + examples = report.drift_examples["derived_orphan"] + assert len(examples) == 1 + ex = examples[0] + assert ex["synthesized_belief_id"] == bid + assert len(str(ex["raw_text"])) <= 200 + + +# --------------------------------------------------------------------------- +# Canonical orphan — belief with no non-legacy log row +# --------------------------------------------------------------------------- + + +def test_canonical_orphan(store: MemoryStore) -> None: + """Hypothesis: a belief inserted directly without any log row counts as + canonical_orphan=1. This covers the pre-#205 and pre-legacy-migration paths. + Falsifiable by canonical_orphan != 1.""" + b = _make_belief(_FACTUAL_SENTENCE) + + # Insert belief directly; do NOT write any ingest_log row. + store.insert_belief(b) + + report = replay_full_equality(store) + assert report.canonical_orphan == 1 + # canonical_orphan does NOT trigger has_drift per spec. + assert report.has_drift is False + + examples = report.drift_examples["canonical_orphan"] + assert len(examples) == 1 + assert examples[0]["belief_id"] == b.id + assert examples[0]["content_hash"] == b.content_hash + + +# --------------------------------------------------------------------------- +# legacy_unknown exclusion +# --------------------------------------------------------------------------- + + +def test_legacy_unknown_rows_excluded(store: MemoryStore) -> None: + """Hypothesis: log rows with source_kind=legacy_unknown are excluded + from total_log_rows and do not trigger any drift bucket. + Falsifiable if excluded_legacy_unknown == 0 or total_log_rows > 0.""" + # Insert a belief without a non-legacy log row. + b = _make_belief(_FACTUAL_SENTENCE) + store.insert_belief(b) + + # Write a legacy_unknown log row pointing at the belief. + store.record_ingest( + source_kind=INGEST_SOURCE_LEGACY_UNKNOWN, + source_path=None, + raw_text=_FACTUAL_SENTENCE, + derived_belief_ids=[b.id], + ts=_TS, + ) + + report = replay_full_equality(store) + assert report.excluded_legacy_unknown == 1 + assert report.total_log_rows == 0 + assert report.matched == 0 + assert report.mismatched == 0 + assert report.derived_orphan == 0 + # The belief's only log row is legacy_unknown → canonical_orphan. + assert report.canonical_orphan == 1 + assert report.has_drift is False + + +def test_legacy_unknown_mixed_with_non_legacy(store: MemoryStore) -> None: + """Hypothesis: if a belief has both a legacy_unknown row and a real row, + the real row is processed and the belief is matched (not orphan). + Falsifiable by canonical_orphan > 0 or matched != 1.""" + bid = _ingest(store, _FACTUAL_SENTENCE) + + # Add a legacy_unknown row pointing at the same belief. + store.record_ingest( + source_kind=INGEST_SOURCE_LEGACY_UNKNOWN, + source_path=None, + raw_text=_FACTUAL_SENTENCE, + derived_belief_ids=[bid], + ts=_TS, + ) + + report = replay_full_equality(store) + assert report.excluded_legacy_unknown == 1 + assert report.total_log_rows == 1 + assert report.matched == 1 + assert report.canonical_orphan == 0 + + +# --------------------------------------------------------------------------- +# Drift example cap +# --------------------------------------------------------------------------- + + +def test_drift_example_cap(store: MemoryStore) -> None: + """Hypothesis: with 25 mismatches and drift_examples=10, only 10 examples + are captured. Falsifiable if len(drift_examples['mismatched']) > 10.""" + source = "test:source" + inserted_ids = [] + for i in range(25): + raw = f"The system parameter {i} controls processing behaviour on startup." + inp = DerivationInput( + raw_text=raw, + source_kind=INGEST_SOURCE_FILESYSTEM, + source_path=source, + ts=_TS, + ) + out = derive(inp) + if out.belief is None: + continue + b = out.belief + store.record_ingest( + source_kind=INGEST_SOURCE_FILESYSTEM, + source_path=source, + raw_text=raw, + derived_belief_ids=[b.id], + ts=_TS, + ) + store.insert_belief(b) + inserted_ids.append(b.id) + + # Corrupt all canonical beliefs' content_hash. + for bid in inserted_ids: + store._conn.execute( # pyright: ignore[reportPrivateUsage] + "UPDATE beliefs SET content_hash = 'CORRUPTED_' || id WHERE id = ?", + (bid,), + ) + store._conn.commit() # pyright: ignore[reportPrivateUsage] + + report = replay_full_equality(store, drift_examples=10) + assert report.mismatched == len(inserted_ids) + assert len(report.drift_examples["mismatched"]) == 10 + + +# --------------------------------------------------------------------------- +# Scope flag +# --------------------------------------------------------------------------- + + +def test_scope_since_v2_equivalent_to_all(store: MemoryStore) -> None: + """Hypothesis: post-#263 migration, scope='since-v2' and scope='all' + produce identical counts. Falsifiable by any counter differing between + the two calls.""" + _ingest(store, _FACTUAL_SENTENCE) + _ingest(store, _FACTUAL_SENTENCE_2) + + report_all = replay_full_equality(store, scope="all") + report_v2 = replay_full_equality(store, scope="since-v2") + + assert report_all.total_log_rows == report_v2.total_log_rows + assert report_all.matched == report_v2.matched + assert report_all.mismatched == report_v2.mismatched + assert report_all.derived_orphan == report_v2.derived_orphan + assert report_all.canonical_orphan == report_v2.canonical_orphan + + +# --------------------------------------------------------------------------- +# has_drift semantics +# --------------------------------------------------------------------------- + + +def test_has_drift_false_when_canonical_orphan_only(store: MemoryStore) -> None: + """Hypothesis: canonical_orphan alone does not trigger has_drift. + Per spec, only mismatched + derived_orphan constitute drift. + Falsifiable by has_drift True.""" + b = _make_belief(_FACTUAL_SENTENCE) + store.insert_belief(b) + + report = replay_full_equality(store) + assert report.canonical_orphan == 1 + assert report.has_drift is False + + +def test_has_drift_false_when_legacy_backfill_only(store: MemoryStore) -> None: + """Hypothesis: legacy_origin_backfill alone does not trigger has_drift. + Falsifiable by has_drift True.""" + b = _make_belief(_FACTUAL_SENTENCE) + store._conn.execute( # pyright: ignore[reportPrivateUsage] + """ + INSERT INTO beliefs ( + id, content, content_hash, alpha, beta, type, lock_level, + locked_at, demotion_pressure, created_at, last_retrieved_at, + session_id, origin + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'unknown') + """, + ( + b.id, b.content, b.content_hash, + b.alpha, b.beta, b.type, b.lock_level, + b.locked_at, b.demotion_pressure, b.created_at, + b.last_retrieved_at, b.session_id, + ), + ) + store._conn.execute( # pyright: ignore[reportPrivateUsage] + "INSERT INTO beliefs_fts (id, content) VALUES (?, ?)", + (b.id, b.content), + ) + store._conn.commit() # pyright: ignore[reportPrivateUsage] + + store.record_ingest( + source_kind=INGEST_SOURCE_FILESYSTEM, + source_path="test:source", + raw_text=_FACTUAL_SENTENCE, + derived_belief_ids=[b.id], + ts=_TS, + ) + + report = replay_full_equality(store) + assert report.legacy_origin_backfill == 1 + assert report.matched == 1 + assert report.has_drift is False + From 51983d925c59487d2edbafcf527581af9cf9ee94 Mon Sep 17 00:00:00 2001 From: rrs <276464689+robotrocketscience@users.noreply.github.com> Date: Tue, 28 Apr 2026 20:50:17 -0700 Subject: [PATCH 2/6] feat(cli): aelf doctor --replay flag and exit codes Wire replay_full_equality through aelf doctor with four new flags: --replay (bool), --max-drift N (int), --drift-examples N (int), --replay-scope {all,since-v2} (default all). Exit 0 when mismatched + derived_orphan == 0 (or <= --max-drift N). canonical_orphan and legacy_origin_backfill are informational and do not affect the exit code. --replay bypasses the hooks/graph checks and returns immediately after printing the report. Output format is stable: one summary line per counter, followed by a drift examples section when drift > 0. Adds CLI integration tests to test_replay_full_equality.py covering clean-store exit 0, mismatch exit 1, --max-drift threshold, and output format assertions. --- src/aelfrice/cli.py | 134 +++++++++++++++++++++++++++++ tests/test_replay_full_equality.py | 128 +++++++++++++++++++++++++++ 2 files changed, 262 insertions(+) diff --git a/src/aelfrice/cli.py b/src/aelfrice/cli.py index 63fea3b54..33948d373 100644 --- a/src/aelfrice/cli.py +++ b/src/aelfrice/cli.py @@ -2118,6 +2118,57 @@ def _cmd_regime(args: argparse.Namespace, out: object) -> int: return 0 +def _print_replay_report( + report: object, + out: object, +) -> None: + """Print a human-readable summary of a FullEqualityReport. + + Format (stable — tests pin specific lines): + + replay full-equality report + --------------------------- + log rows considered: + excluded (legacy_unknown): + matched: + mismatched: + derived_orphan: + canonical_orphan: + legacy_origin_backfill: + feedback_derived_edges: (informational) + + If drift > 0, a "drift examples" section follows with one sub-block + per non-empty bucket. + """ + from aelfrice.replay import FullEqualityReport + assert isinstance(report, FullEqualityReport) + lines = [ + "replay full-equality report", + "---------------------------", + f"log rows considered: {report.total_log_rows}", + f"excluded (legacy_unknown): {report.excluded_legacy_unknown}", + f"matched: {report.matched}", + f"mismatched: {report.mismatched}", + f"derived_orphan: {report.derived_orphan}", + f"canonical_orphan: {report.canonical_orphan}", + f"legacy_origin_backfill: {report.legacy_origin_backfill}", + f"feedback_derived_edges: {report.feedback_derived_edges}" + " (informational)", + ] + for line in lines: + print(line, file=out) # type: ignore[arg-type] + + if report.has_drift: + print("", file=out) # type: ignore[arg-type] + print("drift examples:", file=out) # type: ignore[arg-type] + for bucket, examples in report.drift_examples.items(): + if not examples: + continue + print(f" [{bucket}]", file=out) # type: ignore[arg-type] + for ex in examples: + print(f" {ex}", file=out) # type: ignore[arg-type] + + def _cmd_doctor(args: argparse.Namespace, out: object) -> int: """Diagnose hooks + brain-graph health. @@ -2131,6 +2182,11 @@ def _cmd_doctor(args: argparse.Namespace, out: object) -> int: (issue #206); it bypasses the hooks/graph checks entirely and exits after printing the cost/distribution report. + `--replay` routes to the v2.x full-equality probe (#262); it + bypasses the hooks/graph checks and exits after printing the report. + Exit 0 if ``mismatched + derived_orphan == 0`` (or <= ``--max-drift`` + when that flag is set); exit 1 otherwise. + Exit 1 if any structural failure fires in either subcheck; informational warnings never trip exit. The v1.2 deprecated alias `aelf health` routes here with scope='graph' implicitly via @@ -2140,6 +2196,8 @@ def _cmd_doctor(args: argparse.Namespace, out: object) -> int: return _cmd_doctor_classify_orphans(args, out) if getattr(args, "gc_orphan_feedback", False): return _cmd_doctor_gc_orphan_feedback(args, out) + if getattr(args, "replay", False): + return _cmd_doctor_replay(args, out) scope = getattr(args, "scope", None) exit_code = 0 if scope in (None, "hooks"): @@ -2170,6 +2228,37 @@ def _cmd_doctor(args: argparse.Namespace, out: object) -> int: return exit_code +def _cmd_doctor_replay(args: argparse.Namespace, out: object) -> int: + """Run the v2.x full-equality replay probe (#262). + + Called from `_cmd_doctor` when ``--replay`` is set. Opens the + store, runs `replay_full_equality`, prints the report, and returns + 0 or 1 based on drift counts and the ``--max-drift`` threshold. + """ + from aelfrice.replay import replay_full_equality, FullEqualityReport + + max_drift: int | None = getattr(args, "max_drift", None) + drift_examples: int = int(getattr(args, "drift_examples", 10) or 10) + replay_scope: str = getattr(args, "replay_scope", "all") or "all" + + store = _open_store() + try: + report = replay_full_equality( + store, + max_drift=max_drift, + drift_examples=drift_examples, + scope=replay_scope, # type: ignore[arg-type] + ) + finally: + store.close() + + _print_replay_report(report, out) + + drift_total = report.mismatched + report.derived_orphan + threshold = max_drift if max_drift is not None else 0 + return 0 if drift_total <= threshold else 1 + + def _cmd_doctor_classify_orphans( args: argparse.Namespace, out: object ) -> int: @@ -2747,6 +2836,51 @@ def build_parser(*, show_advanced: bool = False) -> argparse.ArgumentParser: "rows (default: dry-run)." ), ) + p_doctor.add_argument( + "--replay", + dest="replay", + action="store_true", + default=False, + help=( + "run the v2.x full-equality replay probe (#262): re-derive " + "every non-legacy ingest_log row and compare to canonical " + "beliefs. Exits 0 when mismatched + derived_orphan == 0 " + "(or <= --max-drift N). Bypasses the hooks/graph checks." + ), + ) + p_doctor.add_argument( + "--max-drift", + dest="max_drift", + type=int, + default=None, + metavar="N", + help=( + "with --replay: exit 0 when mismatched + derived_orphan <= N " + "instead of requiring exactly 0." + ), + ) + p_doctor.add_argument( + "--drift-examples", + dest="drift_examples", + type=int, + default=10, + metavar="N", + help=( + "with --replay: maximum representative cases to capture per " + "drift bucket (default: 10)." + ), + ) + p_doctor.add_argument( + "--replay-scope", + dest="replay_scope", + choices=("all", "since-v2"), + default="all", + help=( + "with --replay: 'all' (default) walks every non-legacy " + "ingest_log row; 'since-v2' is equivalent post-#263 migration " + "(exists for forward compatibility)." + ), + ) p_doctor.set_defaults(func=_cmd_doctor) p_setup = sub.add_parser( diff --git a/tests/test_replay_full_equality.py b/tests/test_replay_full_equality.py index 222bc126d..0f242f53e 100644 --- a/tests/test_replay_full_equality.py +++ b/tests/test_replay_full_equality.py @@ -468,3 +468,131 @@ def test_has_drift_false_when_legacy_backfill_only(store: MemoryStore) -> None: assert report.matched == 1 assert report.has_drift is False + + +# --------------------------------------------------------------------------- +# CLI integration +# --------------------------------------------------------------------------- + + +def test_cli_doctor_replay_exit_0_clean_store( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Hypothesis: `aelf doctor --replay` exits 0 on a store with no drift. + Falsifiable by any non-zero exit code.""" + import io + from aelfrice.cli import main + + db = str(tmp_path / "brain.db") + monkeypatch.setenv("AELFRICE_DB", db) + + # Ingest a belief so total_log_rows >= 1. + s = MemoryStore(db) + _ingest(s, _FACTUAL_SENTENCE) + s.close() + + out = io.StringIO() + rc = main(["doctor", "--replay"], out=out) + assert rc == 0 + + +def test_cli_doctor_replay_exit_1_on_mismatch( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Hypothesis: `aelf doctor --replay` exits 1 when mismatched > 0. + Falsifiable by exit code 0.""" + import io + from aelfrice.cli import main + + db = str(tmp_path / "brain.db") + monkeypatch.setenv("AELFRICE_DB", db) + + s = MemoryStore(db) + bid = _ingest(s, _FACTUAL_SENTENCE) + s._conn.execute( + "UPDATE beliefs SET content_hash = 'CORRUPTED' WHERE id = ?", (bid,) + ) + s._conn.commit() + s.close() + + out = io.StringIO() + rc = main(["doctor", "--replay"], out=out) + assert rc == 1 + + +def test_cli_doctor_replay_max_drift_exit_0( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Hypothesis: `aelf doctor --replay --max-drift 5` exits 0 when + mismatched + derived_orphan <= 5. + Falsifiable by exit code != 0.""" + import io + from aelfrice.cli import main + + db = str(tmp_path / "brain.db") + monkeypatch.setenv("AELFRICE_DB", db) + + s = MemoryStore(db) + # Create 3 mismatches. + for i in range(3): + raw = f"The server process {i} listens on port {8000 + i} by default." + inp = DerivationInput( + raw_text=raw, + source_kind=INGEST_SOURCE_FILESYSTEM, + source_path="test:source", + ts=_TS, + ) + out_d = derive(inp) + if out_d.belief is None: + continue + b = out_d.belief + s.record_ingest( + source_kind=INGEST_SOURCE_FILESYSTEM, + source_path="test:source", + raw_text=raw, + derived_belief_ids=[b.id], + ts=_TS, + ) + s.insert_belief(b) + s._conn.execute( + "UPDATE beliefs SET content_hash = 'BAD' || id WHERE id = ?", (b.id,) + ) + s._conn.commit() + s.close() + + out = io.StringIO() + rc = main(["doctor", "--replay", "--max-drift", "5"], out=out) + assert rc == 0 + + +def test_cli_doctor_replay_output_format( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Hypothesis: `aelf doctor --replay` prints one summary line per bucket + and a drift section when drift > 0. + Falsifiable by missing expected output lines.""" + import io + from aelfrice.cli import main + + db = str(tmp_path / "brain.db") + monkeypatch.setenv("AELFRICE_DB", db) + + s = MemoryStore(db) + bid = _ingest(s, _FACTUAL_SENTENCE) + s._conn.execute( + "UPDATE beliefs SET content_hash = 'CORRUPTED' WHERE id = ?", (bid,) + ) + s._conn.commit() + s.close() + + buf = io.StringIO() + main(["doctor", "--replay"], out=buf) + text = buf.getvalue() + + # Summary lines present. + assert "matched:" in text + assert "mismatched:" in text + assert "derived_orphan:" in text + assert "canonical_orphan:" in text + # Drift section present. + assert "drift" in text.lower() From 7e554e3ce32eac5014497da2d47088caef84ff0f Mon Sep 17 00:00:00 2001 From: rrs <276464689+robotrocketscience@users.noreply.github.com> Date: Wed, 29 Apr 2026 08:56:11 -0700 Subject: [PATCH 3/6] docs: clarify canonical_orphan is informational in has_drift Expand the has_drift docstring on FullEqualityReport to spell out that canonical_orphan, legacy_origin_backfill, and feedback_derived_edges are informational counters and do not contribute to drift. Drift is strictly mismatched + derived_orphan. Also annotate the canonical-orphan iteration block with a perf TODO: the current implementation issues one ingest_log query per belief, which is N+1; a set-based store query is tracked in a follow-up issue. --- src/aelfrice/replay.py | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/src/aelfrice/replay.py b/src/aelfrice/replay.py index f8d0d8dec..658aef212 100644 --- a/src/aelfrice/replay.py +++ b/src/aelfrice/replay.py @@ -128,6 +128,21 @@ class FullEqualityReport: @property def has_drift(self) -> bool: + """True iff the canonical store disagrees with re-derived ingest log. + + Drift is the union of `mismatched` (canonical exists but a non-origin + field differs) and `derived_orphan` (replay produced a belief id that + is not in the canonical store). + + `canonical_orphan` is informational-only and does NOT count toward + drift: it flags beliefs that exist in the canonical store but have no + non-legacy log row (pre-#205 inserts and legacy_unknown-only rows). + These are expected during the v2.x migration window and reporting + them as drift would produce false positives. + + `legacy_origin_backfill` and `feedback_derived_edges` are also + informational and never trigger drift. + """ return self.mismatched > 0 or self.derived_orphan > 0 @@ -293,6 +308,8 @@ def replay_full_equality( # --- Canonical orphans ------------------------------------------------- # A canonical belief is an orphan when every log row pointing at it is # legacy_unknown (or there are no log rows at all, pre-#205). + # TODO(perf): replace N+1 iteration with set-based store query — see + # follow-up issue. belief_ids = store.list_belief_ids() canonical_orphan = 0 examples_canonical_orphan: list[dict] = [] # type: ignore[type-arg] From 8393324cf1950d86c78435b609d731d0207225aa Mon Sep 17 00:00:00 2001 From: rrs <276464689+robotrocketscience@users.noreply.github.com> Date: Wed, 29 Apr 2026 08:57:33 -0700 Subject: [PATCH 4/6] fix: clamp doctor --replay --max-drift to non-negative A negative --max-drift threshold is nonsensical and previously meant a clean store (drift_total=0) would exit 1 because 0 <= -1 is false. Clamp the threshold to max(0, max_drift) so users who pass negative values still get a sensible exit code, matching the documented contract that exit 0 means drift is within the allowed budget. --- src/aelfrice/cli.py | 2 +- tests/test_replay_full_equality.py | 20 ++++++++++++++++++++ 2 files changed, 21 insertions(+), 1 deletion(-) diff --git a/src/aelfrice/cli.py b/src/aelfrice/cli.py index 33948d373..61fe67ef4 100644 --- a/src/aelfrice/cli.py +++ b/src/aelfrice/cli.py @@ -2255,7 +2255,7 @@ def _cmd_doctor_replay(args: argparse.Namespace, out: object) -> int: _print_replay_report(report, out) drift_total = report.mismatched + report.derived_orphan - threshold = max_drift if max_drift is not None else 0 + threshold = max(0, max_drift) if max_drift is not None else 0 return 0 if drift_total <= threshold else 1 diff --git a/tests/test_replay_full_equality.py b/tests/test_replay_full_equality.py index 0f242f53e..3a81c896d 100644 --- a/tests/test_replay_full_equality.py +++ b/tests/test_replay_full_equality.py @@ -596,3 +596,23 @@ def test_cli_doctor_replay_output_format( assert "canonical_orphan:" in text # Drift section present. assert "drift" in text.lower() + + +def test_cli_doctor_replay_negative_max_drift_clamped( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Hypothesis: a negative `--max-drift` value is clamped to zero, so a + clean store still exits 0. Falsifiable by exit code != 0.""" + import io + from aelfrice.cli import main + + db = str(tmp_path / "brain.db") + monkeypatch.setenv("AELFRICE_DB", db) + + s = MemoryStore(db) + _ingest(s, _FACTUAL_SENTENCE) + s.close() + + out = io.StringIO() + rc = main(["doctor", "--replay", "--max-drift", "-1"], out=out) + assert rc == 0 From 1d899a19aa28ed0a2ed4a5f2479adff16ed536d9 Mon Sep 17 00:00:00 2001 From: rrs <276464689+robotrocketscience@users.noreply.github.com> Date: Wed, 29 Apr 2026 08:57:54 -0700 Subject: [PATCH 5/6] fix: preserve explicit --drift-examples 0 in doctor --replay The previous `int(getattr(...) or 10)` coerced any falsy value, so a user passing `--drift-examples 0` (asking for a no-examples summary) silently got the default of 10. Use an explicit `is None` check so 0 is preserved verbatim and the report contains no per-bucket example sub-blocks even when drift is present. --- src/aelfrice/cli.py | 3 ++- tests/test_replay_full_equality.py | 35 ++++++++++++++++++++++++++++++ 2 files changed, 37 insertions(+), 1 deletion(-) diff --git a/src/aelfrice/cli.py b/src/aelfrice/cli.py index 61fe67ef4..aab1f8bbf 100644 --- a/src/aelfrice/cli.py +++ b/src/aelfrice/cli.py @@ -2238,7 +2238,8 @@ def _cmd_doctor_replay(args: argparse.Namespace, out: object) -> int: from aelfrice.replay import replay_full_equality, FullEqualityReport max_drift: int | None = getattr(args, "max_drift", None) - drift_examples: int = int(getattr(args, "drift_examples", 10) or 10) + _de_raw = getattr(args, "drift_examples", None) + drift_examples: int = 10 if _de_raw is None else int(_de_raw) replay_scope: str = getattr(args, "replay_scope", "all") or "all" store = _open_store() diff --git a/tests/test_replay_full_equality.py b/tests/test_replay_full_equality.py index 3a81c896d..35c3ac8d8 100644 --- a/tests/test_replay_full_equality.py +++ b/tests/test_replay_full_equality.py @@ -616,3 +616,38 @@ def test_cli_doctor_replay_negative_max_drift_clamped( out = io.StringIO() rc = main(["doctor", "--replay", "--max-drift", "-1"], out=out) assert rc == 0 + + +def test_cli_doctor_replay_drift_examples_zero_preserved( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Hypothesis: explicit `--drift-examples 0` produces a report with no + example entries even when drift is present. Falsifiable by any + `[bucket]` sub-block appearing in the output.""" + import io + from aelfrice.cli import main + + db = str(tmp_path / "brain.db") + monkeypatch.setenv("AELFRICE_DB", db) + + s = MemoryStore(db) + bid = _ingest(s, _FACTUAL_SENTENCE) + s._conn.execute( + "UPDATE beliefs SET content_hash = 'CORRUPTED' WHERE id = ?", (bid,) + ) + s._conn.commit() + s.close() + + buf = io.StringIO() + rc = main( + ["doctor", "--replay", "--drift-examples", "0", "--max-drift", "999"], + out=buf, + ) + text = buf.getvalue() + + # Drift exists but no per-bucket example sub-block was emitted. + assert rc == 0 + assert "mismatched:" in text + assert "[mismatched]" not in text + assert "[derived_orphan]" not in text + assert "[canonical_orphan]" not in text From c2eea2d5fd768ca8a1e6515693196e1f60626b0a Mon Sep 17 00:00:00 2001 From: rrs <276464689+robotrocketscience@users.noreply.github.com> Date: Wed, 29 Apr 2026 08:58:09 -0700 Subject: [PATCH 6/6] test: cover canonical-orphan-only doctor --replay scenario MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a CLI integration test that constructs a belief whose only ingest_log row is legacy_unknown — i.e., a pure canonical_orphan with no other drift — and asserts `aelf doctor --replay` exits 0. This pins the contract that canonical_orphan is informational-only and never triggers a non-zero exit code. --- tests/test_replay_full_equality.py | 29 +++++++++++++++++++++++++++++ 1 file changed, 29 insertions(+) diff --git a/tests/test_replay_full_equality.py b/tests/test_replay_full_equality.py index 35c3ac8d8..ebb4efea5 100644 --- a/tests/test_replay_full_equality.py +++ b/tests/test_replay_full_equality.py @@ -651,3 +651,32 @@ def test_cli_doctor_replay_drift_examples_zero_preserved( assert "[mismatched]" not in text assert "[derived_orphan]" not in text assert "[canonical_orphan]" not in text + + +def test_cli_doctor_replay_canonical_orphan_only_exits_0( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Hypothesis: a belief whose only ingest_log row is legacy_unknown is a + canonical_orphan, which is informational-only; `aelf doctor --replay` + must exit 0. Falsifiable by exit code != 0.""" + import io + from aelfrice.cli import main + + db = str(tmp_path / "brain.db") + monkeypatch.setenv("AELFRICE_DB", db) + + s = MemoryStore(db) + b = _make_belief(_FACTUAL_SENTENCE) + s.insert_belief(b) + s.record_ingest( + source_kind=INGEST_SOURCE_LEGACY_UNKNOWN, + source_path=None, + raw_text=_FACTUAL_SENTENCE, + derived_belief_ids=[b.id], + ts=_TS, + ) + s.close() + + out = io.StringIO() + rc = main(["doctor", "--replay"], out=out) + assert rc == 0