From 00c3968937060865681486250d65b75f39f6d20d Mon Sep 17 00:00:00 2001 From: Shay Date: Wed, 27 May 2026 09:43:16 -0700 Subject: [PATCH] fix(ADR-0167): route contemplation and proposal replay by candidate domain (#363) * fix(teaching): select proposal replay gate from candidate domain * test(teaching): pin domain-selected proposal replay gates * fix(teaching): make contemplation probes domain-aware * test(teaching): pin domain-aware contemplation partition --- teaching/contemplation.py | 102 ++++++++++++++++++++++++------- teaching/proposals.py | 62 +++++++++++-------- tests/test_contemplation.py | 57 ++++++++++++++++- tests/test_teaching_proposals.py | 60 +++++++++++++++++- 4 files changed, 230 insertions(+), 51 deletions(-) diff --git a/teaching/contemplation.py b/teaching/contemplation.py index 179c9769..ac22537c 100644 --- a/teaching/contemplation.py +++ b/teaching/contemplation.py @@ -19,14 +19,14 @@ grounded this turn?") and returns an *enriched* candidate with: contemplation never silently truncates. The loop is a pure function of the candidate, the reviewed teaching -corpus, the ratified cognition pack, and an optional vault probe -hook. No clock-time, no LLM, no stochastic sampling, no concurrency -— ADR-0056 Call 4 (sync, not async). +corpus, the ratified domain pack, and an optional vault probe hook. No +clock-time, no LLM, no stochastic sampling, no concurrency — ADR-0056 +Call 4 (sync, not async). -Trust boundary: this module reads ``_pack_index()`` and -``_corpus_index()`` only. It NEVER writes to the corpus, the pack, -or runtime state. Output enriched candidates flow back through the -same Phase B sink as JSONL lines. +Trust boundary: this module reads domain-selected pack/corpus indices +only. It NEVER writes to the corpus, the pack, or runtime state. Output +enriched candidates flow back through the same Phase B sink as JSONL +lines. """ from __future__ import annotations @@ -34,7 +34,8 @@ from __future__ import annotations import hashlib import json from dataclasses import replace -from typing import Any, Callable, Literal +from pathlib import Path +from typing import Any, Callable, Literal, Mapping from chat.pack_grounding import _pack_index from chat.teaching_grounding import _corpus_index @@ -68,6 +69,55 @@ loop. ``None`` means "no vault probe in this contemplation pass." """ _DEFAULT_MAX_DEPTH: int = 8 +_MATH_PACK_PATH = ( + Path(__file__).resolve().parent.parent + / "language_packs" + / "data" + / "en_core_math_v1" +) + + +# --------------------------------------------------------------------------- +# Domain index resolution +# --------------------------------------------------------------------------- + + +def _pack_index_for_domain(domain: str) -> dict[str, tuple[str, ...]]: + """Return the read-only pack index for *domain*. + + Cognition preserves the legacy ``chat.pack_grounding._pack_index`` + semantics. Math reads ``en_core_math_v1`` through the operational + lexicon loader and exposes category membership as shape-level pack + evidence. Unknown domains fail closed. + """ + if domain == "cognition": + return _pack_index() + if domain == "math": + from generate.comprehension.lexicon import load_lexicon + + lexicon = load_lexicon(_MATH_PACK_PATH) + return { + surface: (entry.category,) + for surface, entry in lexicon.by_surface.items() + } + return {} + + +def _corpus_index_for_domain(domain: str) -> Mapping[Any, Any]: + """Return the reviewed corpus index for *domain*. + + The cognition domain keeps the ADR-0056 reviewed teaching corpus. + Math currently has no reviewed TeachingChain-style corpus for + contemplation; returning an empty mapping is deliberate fail-closed + behavior that prevents math candidates from borrowing cognition + evidence. ADR-0167 FOLLOWUPS §5a can tighten this once a math corpus + exists. + """ + if domain == "cognition": + return _corpus_index() + if domain == "math": + return {} + return {} # --------------------------------------------------------------------------- @@ -97,9 +147,14 @@ def _sub_id(parent_candidate_id: str, index: int, payload: dict[str, Any]) -> st def _probe_corpus_direct( - subject: str, intent: str, connective: str | None, obj: str | None + subject: str, + intent: str, + connective: str | None, + obj: str | None, + *, + domain: str = "cognition", ) -> tuple[EvidencePointer, ...]: - """Look in the active reviewed corpus for affirming/falsifying chains. + """Look in the domain-selected reviewed corpus for direct evidence. - Exact match on ``(subject, intent, connective, object)`` is affirming evidence (the proposed chain already exists). @@ -110,7 +165,7 @@ def _probe_corpus_direct( reviewed memory). """ out: list[EvidencePointer] = [] - corpus = _corpus_index() + corpus = _corpus_index_for_domain(domain) chain = corpus.get((subject, intent)) if chain is None: return () @@ -144,7 +199,9 @@ def _probe_corpus_direct( return tuple(out) -def _probe_pack(subject: str, obj: str | None) -> tuple[EvidencePointer, ...]: +def _probe_pack( + subject: str, obj: str | None, *, domain: str = "cognition" +) -> tuple[EvidencePointer, ...]: """Pack lemma residency is shape-level affirming evidence. A pack-resident subject means the subject is grounded; if both @@ -153,7 +210,7 @@ def _probe_pack(subject: str, obj: str | None) -> tuple[EvidencePointer, ...]: falsify (pack ``semantic_domains`` don't express negation — Call 2 of ADR-0056). """ - pack = _pack_index() + pack = _pack_index_for_domain(domain) out: list[EvidencePointer] = [] if subject in pack: out.append(EvidencePointer( @@ -194,10 +251,8 @@ def _decompose( """Return decomposed sub-question payloads. For a Phase B partial chain ``(subject, intent, None, None)``, - enumerate every reviewed object the corpus has used with the - same ``intent`` and treat each as a candidate match for - ``subject``. This is the deterministic, pack-grounded analogue - of "what could this relation be about?" + enumerate every reviewed object the domain corpus has used with the + same ``intent`` and treat each as a candidate match for ``subject``. Returns an empty tuple when no decomposition is possible — the parent records the gap (Call 1 of ADR-0056) and stops. @@ -209,7 +264,7 @@ def _decompose( if obj is not None: # Already has a concrete object — no further decomposition. return () - corpus = _corpus_index() + corpus = _corpus_index_for_domain(candidate.domain) # Deterministic order: sort by object lemma. seen_objects: list[tuple[str, str]] = [] for key, chain in corpus.items(): @@ -352,7 +407,7 @@ def _materialise_sub_candidate( def _probe( - chain: dict[str, Any], vault_probe: _VaultProbe | None + chain: dict[str, Any], vault_probe: _VaultProbe | None, *, domain: str = "cognition" ) -> tuple[EvidencePointer, ...]: """Canonical probe order: vault → pack → corpus. @@ -368,8 +423,8 @@ def _probe( out: list[EvidencePointer] = [] out.extend(_probe_vault(subject, obj, vault_probe)) - out.extend(_probe_pack(subject, obj)) - out.extend(_probe_corpus_direct(subject, intent, connective, obj)) + out.extend(_probe_pack(subject, obj, domain=domain)) + out.extend(_probe_corpus_direct(subject, intent, connective, obj, domain=domain)) return tuple(out) @@ -421,7 +476,7 @@ def contemplate( ) # Direct probe on the parent chain. - direct_evidence = _probe(candidate.proposed_chain, vault_probe) + direct_evidence = _probe(candidate.proposed_chain, vault_probe, domain=candidate.domain) # Decompose into sub-questions. sub_payloads = _decompose(candidate) @@ -510,7 +565,7 @@ def _exemplar_candidate_id(corpus_digest: str, spec_digest: str) -> str: """Deterministic candidate id for an exemplar-derived contemplation. Hash over the corpus digest + the spec digest: identical corpora - yield identical specs yield identical candidate ids. Re-running the + yield identical specs yield identical candidate ids. Re-running the contemplation pipeline against an unchanged corpus is a no-op for the proposal log (idempotency via ProposalLog.find). """ @@ -594,6 +649,7 @@ def contemplate_exemplar_corpus(corpus: Any) -> DiscoveryCandidate: source_turn_trace=f"exemplar_corpus:{corpus.corpus_digest}", pack_consistent=True, boundary_clean=True, + domain="math", review_state="unreviewed", polarity="affirms", claim_domain="factual", diff --git a/teaching/proposals.py b/teaching/proposals.py index ac36e77e..6f4d7c1b 100644 --- a/teaching/proposals.py +++ b/teaching/proposals.py @@ -23,7 +23,7 @@ import hashlib import json from dataclasses import asdict, dataclass from pathlib import Path -from typing import TYPE_CHECKING, Any, Literal +from typing import TYPE_CHECKING, Any, Callable, Literal from teaching.provenance import Provenance from teaching.source import ProposalSource @@ -52,6 +52,7 @@ DEFAULT_CONTEMPLATION_RUNS_DIR: Path = ( ReviewState = Literal["pending", "accepted", "rejected", "withdrawn"] +ReplayGate = Callable[[dict[str, Any]], Any] @dataclass(frozen=True, slots=True) @@ -310,7 +311,7 @@ class ProposalLog: def record_created(self, proposal: TeachingChainProposal) -> None: self._append({"event": "created", "proposal": proposal.as_dict()}) - def record_replay(self, proposal_id: str, evidence: ReplayEvidence) -> None: + def record_replay(self, proposal_id: str, evidence: Any) -> None: self._append({ "event": "replay", "proposal_id": proposal_id, @@ -474,6 +475,23 @@ def append_chain_to_corpus( # --------------------------------------------------------------------------- +def _replay_gate_for_domain(domain: str) -> ReplayGate: + """Return the replay gate for a candidate domain. + + Cognition candidates keep the ADR-0057 cognition replay-equivalence + gate. Math candidates use the ADR-0163 admissibility gate so wrong=0 + capability axes and GSM8K train-sample evidence are checked by + default instead of depending on each caller to pass an override. + """ + if domain == "cognition": + from teaching.replay import run_replay_equivalence + return run_replay_equivalence + if domain == "math": + from teaching.replay import run_admissibility_replay_gate + return run_admissibility_replay_gate + raise ProposalError(f"unsupported proposal domain: {domain!r}") + + def propose_from_candidate( candidate: DiscoveryCandidate, *, @@ -486,10 +504,10 @@ def propose_from_candidate( """End-to-end: build proposal, run replay-equivalence gate, auto-reject on regression, otherwise leave pending. - ``run_replay`` is the replay function (``teaching.replay. - run_replay_equivalence`` by default); accepting it as a kwarg - keeps tests fast — they can pass a fake that returns a stub - ``ReplayEvidence`` without booting the cognition lane. + ``run_replay`` overrides the domain-selected replay function for + tests or specialised callers. When omitted, the gate is selected + from ``candidate.domain``: cognition → ``run_replay_equivalence``; + math → ``run_admissibility_replay_gate``. Submission-time checks fire in this order (ADR-0161 §3): 1. Capacity (Step 2) — queue_full if pending_count >= cap @@ -536,11 +554,11 @@ def propose_from_candidate( if candidate.proposed_chain else None ) or candidate.claim_domain - + from datetime import datetime, timezone stamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H%M%SZ") report_path = contemplation_runs_dir / f"{stamp}_queue_full.json" - + report_data = { "report_kind": "queue_full", "emitted_at_revision": _current_revision(), @@ -604,9 +622,8 @@ def propose_from_candidate( log.record_created(proposal) - if run_replay is None: - from teaching.replay import run_replay_equivalence as run_replay - evidence = run_replay(proposal.proposed_chain) + replay = run_replay if run_replay is not None else _replay_gate_for_domain(candidate.domain) + evidence = replay(proposal.proposed_chain) log.record_replay(proposal.proposal_id, evidence) if not evidence.replay_equivalent: @@ -662,12 +679,12 @@ def accept_proposal( def reject_proposal( proposal_id: str, *, log: ProposalLog, operator_note: str = "" ) -> None: - record = log.find(proposal_id) - if record is None: + rec = log.find(proposal_id) + if rec is None: raise ProposalError(f"proposal not found: {proposal_id}") - if record["state"] != "pending": + if rec["state"] != "pending": raise ProposalError( - f"proposal {proposal_id} is {record['state']!r}, not pending" + f"proposal {proposal_id} is {rec['state']!r}, not pending" ) log.record_transition(proposal_id, "rejected", operator_note) @@ -675,26 +692,23 @@ def reject_proposal( def withdraw_proposal( proposal_id: str, *, log: ProposalLog, operator_note: str = "" ) -> None: - record = log.find(proposal_id) - if record is None: + rec = log.find(proposal_id) + if rec is None: raise ProposalError(f"proposal not found: {proposal_id}") - if record["state"] != "pending": + if rec["state"] != "pending": raise ProposalError( - f"proposal {proposal_id} is {record['state']!r}, not pending" + f"proposal {proposal_id} is {rec['state']!r}, not pending" ) log.record_transition(proposal_id, "withdrawn", operator_note) __all__ = [ - "DEFAULT_PENDING_CAP", - "DEFAULT_PROPOSAL_LOG_PATH", "ProposalError", "ProposalLog", - "RefusedAsDependent", + "RefusedAsCapacity", "RefusedAsDuplicate", - "RefusedAtCapacity", + "RefusedAsDependent", "ReplayEvidence", - "ReviewState", "TeachingChainProposal", "accept_proposal", "append_chain_to_corpus", diff --git a/tests/test_contemplation.py b/tests/test_contemplation.py index 95d8e4ae..efae81e9 100644 --- a/tests/test_contemplation.py +++ b/tests/test_contemplation.py @@ -35,6 +35,7 @@ CORPUS_BYTES_BEFORE = _CORPUS_PATH.read_bytes() if _CORPUS_PATH.exists() else b" def _phase_b_candidate( *, subject: str = "wisdom", intent: str = "cause", candidate_id: str = "cand_abc", trace: str = "trace_xyz", + domain: str = "cognition", ) -> DiscoveryCandidate: return DiscoveryCandidate( candidate_id=candidate_id, @@ -48,6 +49,7 @@ def _phase_b_candidate( source_turn_trace=trace, pack_consistent=True, boundary_clean=True, + domain=domain, ) @@ -93,8 +95,8 @@ def test_empty_pack_and_corpus_terminates_with_gap(monkeypatch): """No pack, no corpus ⇒ every probe fails, parent gap-records.""" from teaching import contemplation as contemp_mod - monkeypatch.setattr(contemp_mod, "_pack_index", lambda: {}) - monkeypatch.setattr(contemp_mod, "_corpus_index", lambda: {}) + monkeypatch.setattr(contemp_mod, "_pack_index_for_domain", lambda _domain: {}) + monkeypatch.setattr(contemp_mod, "_corpus_index_for_domain", lambda _domain: {}) cand = _phase_b_candidate() out = contemplate(cand) @@ -106,6 +108,55 @@ def test_empty_pack_and_corpus_terminates_with_gap(monkeypatch): assert out.recursion_overflow is False +# --------------------------------------------------------------------------- +# Domain-aware partition +# --------------------------------------------------------------------------- + + +def test_math_contemplation_does_not_borrow_cognition_corpus(): + """Math candidates fail closed instead of using cognition corpus evidence.""" + cand = DiscoveryCandidate( + candidate_id="cand_math_no_cognition_leak", + proposed_chain={ + "subject": "light", + "intent": "cause", + "connective": "reveals", + "object": "truth", + }, + trigger="would_have_grounded", + source_turn_trace="t_math", + pack_consistent=True, + boundary_clean=True, + domain="math", + ) + out = contemplate(cand) + assert out.domain == "math" + assert out.polarity == "undetermined" + assert not any(e.source == "corpus" for e in out.evidence) + + +def test_math_contemplation_uses_math_pack_residency(): + """Math candidates can receive math-pack evidence without corpus leakage.""" + cand = DiscoveryCandidate( + candidate_id="cand_math_pack", + proposed_chain={ + "subject": "does", + "intent": "admissibility", + "connective": "recognizes", + "object": "does", + }, + trigger="would_have_grounded", + source_turn_trace="t_math_pack", + pack_consistent=True, + boundary_clean=True, + domain="math", + ) + out = contemplate(cand) + assert out.domain == "math" + assert any(e.source == "pack" and e.ref == "does" for e in out.evidence) + assert not any(e.source == "corpus" for e in out.evidence) + + # --------------------------------------------------------------------------- # Factual affirming evidence # --------------------------------------------------------------------------- @@ -172,7 +223,7 @@ def test_mixed_evidence_upgrades_claim_domain(monkeypatch): """Mixed affirm + falsify ⇒ undetermined AND domain upgrades one tier.""" from teaching import contemplation as contemp_mod - def fake_corpus_probe(subject, intent, connective, obj): + def fake_corpus_probe(subject, intent, connective, obj, *, domain="cognition"): return ( EvidencePointer( source="corpus", ref="chain_aff", polarity="affirms", diff --git a/tests/test_teaching_proposals.py b/tests/test_teaching_proposals.py index 73a62828..7727f2f0 100644 --- a/tests/test_teaching_proposals.py +++ b/tests/test_teaching_proposals.py @@ -44,7 +44,7 @@ CORPUS_BYTES_BEFORE = _CORPUS_PATH.read_bytes() if _CORPUS_PATH.exists() else b" def _enriched(*, polarity="affirms", claim_domain="factual", connective="reveals", obj="truth", subject="light", - evidence=None, boundary_clean=True): + evidence=None, boundary_clean=True, domain="cognition"): if evidence is None: evidence = ( EvidencePointer( @@ -62,6 +62,7 @@ def _enriched(*, polarity="affirms", claim_domain="factual", source_turn_trace="trace_1", pack_consistent=True, boundary_clean=boundary_clean, + domain=domain, polarity=polarity, claim_domain=claim_domain, evidence=evidence, @@ -190,6 +191,63 @@ def test_propose_from_candidate_auto_rejects_on_regression(tmp_path: Path): assert "surface_groundedness" in rec["operator_note"] +def test_propose_selects_replay_gate_by_candidate_domain(monkeypatch, tmp_path: Path): + calls: list[str] = [] + + def fake_cognition_gate(chain): + calls.append("cognition") + return _fake_replay_equivalent(chain) + + def fake_math_gate(chain): + calls.append("math") + return _fake_replay_equivalent(chain) + + monkeypatch.setattr( + "teaching.replay.run_replay_equivalence", + fake_cognition_gate, + ) + monkeypatch.setattr( + "teaching.replay.run_admissibility_replay_gate", + fake_math_gate, + ) + + log_cognition = ProposalLog(tmp_path / "cognition" / "proposals.jsonl") + propose_from_candidate(_enriched(domain="cognition"), log=log_cognition) + + log_math = ProposalLog(tmp_path / "math" / "proposals.jsonl") + propose_from_candidate( + _enriched(domain="math", subject="sees", connective="recognizes", obj="drain"), + log=log_math, + ) + + assert calls == ["cognition", "math"] + + +def test_explicit_replay_override_wins_over_domain(monkeypatch, tmp_path: Path): + calls: list[str] = [] + + def forbidden_math_gate(chain): + raise AssertionError("domain-selected math gate should not run") + + def override_gate(chain): + calls.append("override") + return _fake_replay_equivalent(chain) + + monkeypatch.setattr( + "teaching.replay.run_admissibility_replay_gate", + forbidden_math_gate, + ) + + log = ProposalLog(tmp_path / "proposals.jsonl") + propose_from_candidate( + _enriched(domain="math", subject="sees", connective="recognizes", obj="drain"), + log=log, + run_replay=override_gate, + ) + + assert calls == ["override"] + + def test_propose_is_idempotent(tmp_path: Path): log = ProposalLog(tmp_path / "proposals.jsonl") c = _enriched()