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
This commit is contained in:
Shay
GitHub
parent
dbeb1b2f00
commit
00c3968937
4 changed files with 230 additions and 51 deletions
|
|
@ -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)
|
||||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
Loading…
Reference in a new issue