7c80b791ec
The full lane carried 13 long-standing red tests whose premises were
invalidated by reviewed-corpus growth that landed in earlier commits.
None reflected runtime bugs — all four classes are corpus-state drift
where the test fixture became stale. Curated lanes were green, full
lane stayed quietly red. Closes that gap.
1. test_teaching_audit (2 tests).
* test_audit_real_corpus_runs_clean asserted dropped == () and
lines_on_disk == lines_loaded — premise written before any
supersession existed. Curriculum saturation v2 (commit a0edbb4)
ratified the wisdom_grounds_judgment → wisdom_requires_knowledge
supersession; the audit now correctly shows 1 dropped line.
Rewritten as the line-conservation invariant:
lines_loaded + len(dropped) == lines_on_disk
plus a typed-reason check on every dropped entry.
* test_default_superseded_by_is_null_in_loaded_entries asserted
ALL loaded entries have superseded_by == None. Wrong even by
ADR-0055 design: the replacement entry IS loaded and carries
the back-pointer to the retired chain. Rewritten as the
active-set invariant: any non-null superseded_by on a loaded
entry must reference a dropped (retired) chain id, never a live
one — no double-live state.
2. test_learning_loop_demo (7 tests).
The demo's headline prompt was "Why does thought exist?", and the
ADR-0057 demo trilogy (commit 82dac4b) chose (thought, cause) as
the cold cell. Cognition saturation v2 (commit a0edbb4) ratified
cause_thought_reveals_meaning into the active corpus — so the
cold turn now grounds, no discovery candidate is emitted, every
demo scene breaks. Rotated the cold subject to ``narrative``
(pack-resident, no chain, same thematic shape, same affirming
evidence pointer cause_creation_reveals_meaning). Demo headline,
evals/learning_loop/run_demo.py, core/cli.py preamble, and the
test assertions all updated together so the demo reads cleanly:
before: [none] I don't know — insufficient grounding...
after : [teaching] narrative — teaching-grounded ... narrative
reveals meaning ...
3. test_discovery_candidates (4 tests).
Test fixture used (judgment, CAUSE) as the still-cold pair.
Epistemology v1 (commit 2acf71f) ratified
cause_judgment_requires_wisdom — (judgment, cause) is no longer
cold. Rotated to ``principle`` (pack-resident, no chain on either
intent today). Added a pytest.skip self-guard so when a future
curriculum unit ratifies a (principle, *) chain the test rotates
cleanly instead of going red.
Full lane: 1892 passed, 2 skipped, 0 failed (was 4 failed pre-fix,
13 failed pre-ADR-0063). Cognition eval unchanged: public 100/100/
91.7/100, holdout 100/100/83.3/100.
311 lines
12 KiB
Python
311 lines
12 KiB
Python
"""ADR-0055 Phase A — teaching corpus audit + supersession + typed provenance.
|
|
|
|
Pinned contracts:
|
|
|
|
- ``teaching.audit.audit_corpus`` is pure, deterministic, and never
|
|
mutates the corpus or the pack.
|
|
- The active runtime loader (``chat.teaching_grounding._corpus_index``)
|
|
and the audit module agree on which entries are dropped and why.
|
|
- ``superseded_by`` retires an earlier chain from the active view but
|
|
leaves it on disk.
|
|
- Legacy ``"reviewed"`` provenance source token maps to
|
|
``"hand_authored"`` so the current corpus reports the typed enum
|
|
without a file rewrite.
|
|
- Pack-consistency drop is surfaced with the specific lemma name in
|
|
the reason.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from teaching.audit import audit_corpus
|
|
from teaching.provenance import Provenance, parse_provenance
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# parse_provenance — typed shape
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_parse_legacy_reviewed_token_maps_to_hand_authored() -> None:
|
|
p = parse_provenance("adr-0052:reviewed:2026-05-17")
|
|
assert p.adr_id == "adr-0052"
|
|
assert p.source == "hand_authored"
|
|
assert p.review_date == "2026-05-17"
|
|
assert p.raw == "adr-0052:reviewed:2026-05-17"
|
|
|
|
|
|
def test_parse_canonical_hand_authored() -> None:
|
|
p = parse_provenance("adr-9999:hand_authored:2099-12-31")
|
|
assert p.source == "hand_authored"
|
|
|
|
|
|
def test_parse_discovery_promoted() -> None:
|
|
p = parse_provenance("adr-9999:discovery_promoted:2099-12-31")
|
|
assert p.source == "discovery_promoted"
|
|
|
|
|
|
def test_parse_imported() -> None:
|
|
p = parse_provenance("adr-9999:imported:2099-12-31")
|
|
assert p.source == "imported"
|
|
|
|
|
|
def test_parse_unknown_source_token_falls_back() -> None:
|
|
p = parse_provenance("adr-9999:gibberish:2099-12-31")
|
|
assert p.source == "unknown"
|
|
# adr_id and review_date are still captured.
|
|
assert p.adr_id == "adr-9999"
|
|
assert p.review_date == "2099-12-31"
|
|
|
|
|
|
def test_parse_non_string_input_safe() -> None:
|
|
assert parse_provenance(None).source == "unknown"
|
|
assert parse_provenance(42).source == "unknown"
|
|
assert parse_provenance({"adr": "x"}).source == "unknown"
|
|
|
|
|
|
def test_parse_empty_string_safe() -> None:
|
|
p = parse_provenance("")
|
|
assert p == Provenance(adr_id=None, source="unknown", review_date=None, raw="")
|
|
|
|
|
|
def test_parse_short_string_no_crash() -> None:
|
|
p = parse_provenance("adr-0052")
|
|
assert p.source == "unknown"
|
|
assert p.raw == "adr-0052"
|
|
|
|
|
|
def test_parse_extra_trailing_colons_folded_into_date() -> None:
|
|
p = parse_provenance("adr-9999:hand_authored:2099-12-31:extra")
|
|
# Folding into review_date keeps drift safe — no crash, no silent loss.
|
|
assert p.review_date == "2099-12-31:extra"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# audit_corpus — real corpus, no mutations
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_audit_real_corpus_runs_clean() -> None:
|
|
"""The audit report's accounting invariant: every line on disk is
|
|
either loaded or accounted for as dropped (with a typed reason).
|
|
Hardcoding ``dropped == ()`` was the pre-supersession premise; the
|
|
curriculum saturation v2 (commit ``a0edbb4``) introduced one
|
|
ratified supersession (``verification_wisdom_grounds_judgment`` →
|
|
``verification_wisdom_requires_knowledge``). The invariant the
|
|
audit guarantees is line-conservation, not zero supersessions."""
|
|
report = audit_corpus()
|
|
assert report.lines_loaded + len(report.dropped) == report.lines_on_disk
|
|
assert len(report.loaded) >= 10
|
|
assert report.corpus_id == "cognition_chains_v1"
|
|
# Every dropped line must carry a typed reason so an operator can
|
|
# audit why it was retired without re-deriving the supersession.
|
|
for dropped in report.dropped:
|
|
assert dropped.reason.startswith("superseded_by:")
|
|
|
|
|
|
def test_audit_loaded_entries_have_typed_provenance() -> None:
|
|
report = audit_corpus()
|
|
for entry in report.loaded:
|
|
assert entry.provenance.source in {
|
|
"hand_authored", "discovery_promoted", "imported", "unknown",
|
|
}
|
|
|
|
|
|
def test_audit_is_deterministic() -> None:
|
|
a = audit_corpus()
|
|
b = audit_corpus()
|
|
assert a.as_dict() == b.as_dict()
|
|
|
|
|
|
def test_audit_as_dict_is_json_serialisable() -> None:
|
|
report = audit_corpus()
|
|
blob = json.dumps(report.as_dict(), sort_keys=True)
|
|
assert "cognition_chains_v1" in blob
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# audit_corpus — synthetic corpora exercising drop reasons
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _write_corpus(tmp_path: Path, lines: list[str]) -> Path:
|
|
p = tmp_path / "cognition_chains_v1.jsonl"
|
|
p.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
return p
|
|
|
|
|
|
def test_audit_surfaces_invalid_json(tmp_path: Path) -> None:
|
|
path = _write_corpus(tmp_path, ["not json at all"])
|
|
report = audit_corpus(path)
|
|
assert report.lines_on_disk == 1
|
|
assert report.lines_loaded == 0
|
|
assert len(report.dropped) == 1
|
|
assert report.dropped[0].reason == "invalid_json"
|
|
|
|
|
|
def test_audit_surfaces_unsupported_intent(tmp_path: Path) -> None:
|
|
path = _write_corpus(tmp_path, [json.dumps({
|
|
"chain_id": "x", "subject": "light", "intent": "definition",
|
|
"connective": "is", "object": "truth",
|
|
"provenance": "adr-test:hand_authored:2026-05-18",
|
|
})])
|
|
report = audit_corpus(path)
|
|
assert len(report.dropped) == 1
|
|
assert report.dropped[0].reason == "unsupported_intent:definition"
|
|
|
|
|
|
def test_audit_surfaces_pack_missing_subject(tmp_path: Path) -> None:
|
|
path = _write_corpus(tmp_path, [json.dumps({
|
|
"chain_id": "x", "subject": "zzznotalemma", "intent": "cause",
|
|
"connective": "reveals", "object": "truth",
|
|
"provenance": "adr-test:hand_authored:2026-05-18",
|
|
})])
|
|
report = audit_corpus(path)
|
|
assert len(report.dropped) == 1
|
|
assert report.dropped[0].reason == "pack_missing_subject:zzznotalemma"
|
|
|
|
|
|
def test_audit_surfaces_pack_missing_object(tmp_path: Path) -> None:
|
|
path = _write_corpus(tmp_path, [json.dumps({
|
|
"chain_id": "x", "subject": "light", "intent": "cause",
|
|
"connective": "reveals", "object": "zzznotalemma",
|
|
"provenance": "adr-test:hand_authored:2026-05-18",
|
|
})])
|
|
report = audit_corpus(path)
|
|
assert len(report.dropped) == 1
|
|
assert report.dropped[0].reason == "pack_missing_object:zzznotalemma"
|
|
|
|
|
|
def test_audit_surfaces_missing_required_field(tmp_path: Path) -> None:
|
|
path = _write_corpus(tmp_path, [json.dumps({
|
|
"chain_id": "x", "subject": "", "intent": "cause",
|
|
"connective": "reveals", "object": "truth",
|
|
})])
|
|
report = audit_corpus(path)
|
|
assert len(report.dropped) == 1
|
|
assert report.dropped[0].reason == "missing_required_field:subject"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Supersession — disk preserves history, active view drops superseded
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_supersession_drops_earlier_chain_from_active_view(tmp_path: Path) -> None:
|
|
older = {
|
|
"chain_id": "cause_light_reveals_truth",
|
|
"subject": "light", "intent": "cause", "connective": "reveals",
|
|
"object": "truth", "domains_subject_k": 2, "domains_object_k": 1,
|
|
"provenance": "adr-test:hand_authored:2026-05-18",
|
|
}
|
|
newer = {
|
|
"chain_id": "cause_light_grounds_truth",
|
|
"subject": "light", "intent": "cause", "connective": "grounds",
|
|
"object": "truth", "domains_subject_k": 2, "domains_object_k": 1,
|
|
"provenance": "adr-test:hand_authored:2026-05-19",
|
|
"superseded_by": "cause_light_reveals_truth",
|
|
}
|
|
path = _write_corpus(tmp_path, [json.dumps(older), json.dumps(newer)])
|
|
report = audit_corpus(path)
|
|
assert report.lines_on_disk == 2
|
|
assert report.lines_loaded == 1
|
|
# The newer entry retires the older one.
|
|
assert report.dropped[0].chain_id == "cause_light_reveals_truth"
|
|
assert report.dropped[0].reason == "superseded_by:cause_light_reveals_truth"
|
|
# Newer entry is active.
|
|
assert report.loaded[0].chain_id == "cause_light_grounds_truth"
|
|
|
|
|
|
def test_loaded_entries_with_supersession_point_to_dropped_chains() -> None:
|
|
"""The active-set invariant: a loaded entry may carry a non-null
|
|
``superseded_by`` (it is the replacement, pointing back at what it
|
|
retired), but the chain id it points to MUST itself be dropped
|
|
from the active set — otherwise the corpus would contain two
|
|
live versions of the same lemma's chain.
|
|
|
|
Was ``test_default_superseded_by_is_null_in_loaded_entries`` (an
|
|
overstrict premise written before any supersession existed); the
|
|
weaker invariant here is the one ADR-0055 Phase A actually
|
|
guarantees."""
|
|
report = audit_corpus()
|
|
dropped_ids = {entry.chain_id for entry in report.dropped}
|
|
live_ids = {entry.chain_id for entry in report.loaded}
|
|
for entry in report.loaded:
|
|
if entry.superseded_by is None:
|
|
continue
|
|
# Replacement entry — its back-pointer must reference a chain
|
|
# id that is dropped (retired), never a live one.
|
|
assert entry.superseded_by in dropped_ids, (
|
|
f"loaded entry {entry.chain_id!r} points to "
|
|
f"{entry.superseded_by!r} which is not in the dropped set "
|
|
f"— supersession invariant broken"
|
|
)
|
|
assert entry.superseded_by not in live_ids, (
|
|
f"loaded entry {entry.chain_id!r} supersedes a still-live "
|
|
f"chain {entry.superseded_by!r} — double-live state"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Runtime parity — runtime loader and audit agree
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_runtime_loader_and_audit_agree_on_active_chain_ids() -> None:
|
|
"""Whatever audit_corpus says is loaded must also be what the
|
|
runtime loader has indexed (modulo the keying — runtime keys by
|
|
(subject, intent); audit lists chain_ids)."""
|
|
from chat.teaching_grounding import _corpus_index
|
|
|
|
_corpus_index.cache_clear()
|
|
runtime_chains = {c.chain_id for c in _corpus_index().values()}
|
|
audit_chains = {c.chain_id for c in audit_corpus().loaded}
|
|
assert runtime_chains == audit_chains
|
|
|
|
|
|
def test_runtime_loader_honors_superseded_by(tmp_path: Path) -> None:
|
|
"""If a corpus on disk has supersession, the runtime loader's
|
|
active set must match the audit report."""
|
|
older = {
|
|
"chain_id": "cause_light_reveals_truth",
|
|
"subject": "light", "intent": "cause", "connective": "reveals",
|
|
"object": "truth",
|
|
"provenance": "adr-test:hand_authored:2026-05-18",
|
|
}
|
|
newer = {
|
|
"chain_id": "cause_light_grounds_truth",
|
|
"subject": "light", "intent": "cause", "connective": "grounds",
|
|
"object": "truth",
|
|
"provenance": "adr-test:hand_authored:2026-05-19",
|
|
"superseded_by": "cause_light_reveals_truth",
|
|
}
|
|
path = _write_corpus(tmp_path, [json.dumps(older), json.dumps(newer)])
|
|
|
|
from chat import teaching_grounding as tg
|
|
|
|
with patch.object(tg, "_CORPUS_PATH", path):
|
|
tg._corpus_index.cache_clear()
|
|
try:
|
|
index = tg._corpus_index()
|
|
active = {c.chain_id for c in index.values()}
|
|
assert active == {"cause_light_grounds_truth"}
|
|
finally:
|
|
tg._corpus_index.cache_clear()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Audit is read-only (trust boundary)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_audit_does_not_mutate_corpus_on_disk() -> None:
|
|
from chat.teaching_grounding import _CORPUS_PATH
|
|
|
|
before = _CORPUS_PATH.read_bytes()
|
|
audit_corpus()
|
|
after = _CORPUS_PATH.read_bytes()
|
|
assert before == after
|