feat(ADR-0163.C): contemplation ingests admissibility exemplars and emits DerivedRecognizer proposals through the HITL corridor (#301)
Phase C is the first phase where operator-authored exemplar corpora
become engine-derived recognizer proposals automatically. The math
thesis ("decodes, not generates") manifests in the math lane here.
Modules
- teaching/exemplar_ingest.py — pure-function loader for Phase B
exemplar JSONLs. ExemplarCorpus carries a sha256 digest over its
canonical (sorted-by-exemplar_id, sort-keyed) bytes.
- teaching/recognizer_synthesis.py — per-category synthesizers
(_synthesize_descriptive_setup_no_quantity / _temporal_aggregation /
_rate_with_currency) distil a corpus into one RecognizerSpec.
Determinism: same corpus -> byte-identical spec. Narrowness: the
spec records only observed sub-shapes; an out-of-corpus currency
symbol or window unit does not match. Phase B author_notes surface
in canonical_pattern.unresolved_notes — never silently dropped.
- teaching/contemplation.py — contemplate_exemplar_corpus(corpus)
returns a DiscoveryCandidate whose proposed_chain encodes the
RecognizerSpec as a synthetic four-field chain plus the full
recognizer_spec submap. Evidence cites every exemplar's case_id.
- teaching/replay.py — run_admissibility_replay_gate(spec, *,
active_corpus_path=None) runs cognition + G1..G5+S1 + GSM8K
train_sample. In-process baseline cache keyed on the active
corpus digest. WRONG-COUNT INVARIANT: if a candidate run lifts
the GSM8K train_sample wrong count, gate returns
replay_equivalent=False with
regressed_metrics=["gsm8k_train_sample_wrong_count"].
- teaching/source.py — ProposalKind widened with "exemplar_corpus";
exhaustive-match docs + tests updated.
CLI
- core teaching propose-from-exemplars <path> [--all] [--review-date]
[--log] [--json]. Routes the candidate through the existing
propose_from_candidate path with the admissibility gate substituted
for the cognition-only run_replay_equivalence. Never auto-accepts;
proposals land as pending for operator review.
Tests (38 new)
- tests/test_exemplar_ingest.py (12) — load, digest stability,
malformed-record rejection, file-name binding, read-only purity.
- tests/test_recognizer_synthesis.py (16) — determinism, purity,
per-category subsumption, narrowness (out-of-corpus seeds rejected),
author_notes surfaced.
- tests/test_admissibility_replay_gate.py (6) — happy path, cache
hit/invalidation, WRONG-COUNT INVARIANT regression, capability-axis
regression, cognition regression.
- tests/test_propose_from_exemplars_cli.py (4) — single corpus, --all,
determinism, read-only snapshot.
Acceptance evidence (dry run)
- All three Phase B corpora produce replay_equivalent=true,
wrong_count_delta=0. Proposal IDs:
descriptive_setup_no_quantity: 59223f13722f906a1cf9b65d9b01c990
rate_with_currency: 46ce297f797ff16da12db5de422ca3c9
temporal_aggregation: a3b892546977c5f0f64c578d6052adbd
- G1..G5+S1 wrong=0 unchanged; GSM8K train_sample 3/47/0 unchanged.
- core test --suite smoke -q: 67 passed.
- uv run core eval refusal_taxonomy: case_digest
d030f826cb0f4088771d90c52c8be2ff75054ab27c7d47eae8dbfe1225b2eea1
unchanged.
Cross-refs: ADR-0163 (Phase C), ADR-0057 (gating discipline),
ADR-0151 (auto-proposal), ADR-0152 (learning-arc), ADR-0149/0154
(recognizer pipeline), ADR-0094 (ProposalSource), Phase A PR #297,
Phase B PR #298.
Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Shay
GitHub
parent
cdead696ed
commit
08c5e0e82f
11 changed files with 2212 additions and 5 deletions
160
core/cli.py
160
core/cli.py
|
|
@ -23,7 +23,7 @@ _CORE_RS_DIR = _REPO_ROOT / "core-rs"
|
|||
_CORE_RS_MANIFEST = _CORE_RS_DIR / "Cargo.toml"
|
||||
|
||||
DESCRIPTION = "CORE versor engine command suite."
|
||||
EPILOG = "Examples:\n core chat\n core pulse \"What is truth?\"\n core pulse --no-glove --json \"Compare knowledge and wisdom\"\n core bench\n core bench --suite all\n core bench --suite all --json --report bench_all.json\n core bench --suite determinism --runs 50\n core bench --suite speedup --json\n core trace \"word beginning truth\"\n core trace --output-language grc --frame-pack grc --json \"logos\"\n core rust status\n core rust build\n core oov covenant\n core pack list\n core pack verify en_minimal_v1\n core teaching audit\n core teaching audit --json\n core teaching gaps --top 10\n core teaching queue --threshold 3\n core teaching hitl-queue list\n core teaching hitl-queue list --state all --json\n core teaching hitl-queue show <proposal_id>\n core teaching propose <candidate-jsonl-path>\n core teaching proposals --state pending\n core teaching review <proposal_id> --accept --review-date 2026-05-18\n core teaching supersede cause_light_reveals_truth --subject light --intent cause --connective grounds --object truth --review-date 2026-05-18\n core teaching supersessions\n core teaching supersessions --json\n core test --suite fast -q\n core test --suite pulse -q\n core test --suite proof -q\n core test --suite cognition -q\n core test -- tests/test_alignment_graph.py -q\n core demo audit-tour\n core demo register-tour\n core demo anchor-lens-tour\n core demo orthogonality-tour\n core demo pack-measurements\n core demo long-context-comparison\n core demo anti-regression\n core demo learning-loop\n core demo learning-arc\n core demo articulation\n core demo conversation\n core demo conversation --no-stream\n core demo all\n core demo adr-0024-chain\n core eval --list\n core eval cognition\n core eval cognition --json --save\n core eval cognition --split dev --version v1\n core eval cognition --split holdout\n core eval contemplation_quality\n core eval contemplation_quality --json --save\n core workbench api\n core workbench api --port 9000\n core workbench api --host 0.0.0.0 --allow-nonlocal-bind"
|
||||
EPILOG = "Examples:\n core chat\n core pulse \"What is truth?\"\n core pulse --no-glove --json \"Compare knowledge and wisdom\"\n core bench\n core bench --suite all\n core bench --suite all --json --report bench_all.json\n core bench --suite determinism --runs 50\n core bench --suite speedup --json\n core trace \"word beginning truth\"\n core trace --output-language grc --frame-pack grc --json \"logos\"\n core rust status\n core rust build\n core oov covenant\n core pack list\n core pack verify en_minimal_v1\n core teaching audit\n core teaching audit --json\n core teaching gaps --top 10\n core teaching queue --threshold 3\n core teaching hitl-queue list\n core teaching hitl-queue list --state all --json\n core teaching hitl-queue show <proposal_id>\n core teaching propose <candidate-jsonl-path>\n core teaching propose-from-exemplars teaching/admissibility_exemplars/rate_with_currency_v1.jsonl\n core teaching propose-from-exemplars --all\n core teaching proposals --state pending\n core teaching review <proposal_id> --accept --review-date 2026-05-18\n core teaching supersede cause_light_reveals_truth --subject light --intent cause --connective grounds --object truth --review-date 2026-05-18\n core teaching supersessions\n core teaching supersessions --json\n core test --suite fast -q\n core test --suite pulse -q\n core test --suite proof -q\n core test --suite cognition -q\n core test -- tests/test_alignment_graph.py -q\n core demo audit-tour\n core demo register-tour\n core demo anchor-lens-tour\n core demo orthogonality-tour\n core demo pack-measurements\n core demo long-context-comparison\n core demo anti-regression\n core demo learning-loop\n core demo learning-arc\n core demo articulation\n core demo conversation\n core demo conversation --no-stream\n core demo all\n core demo adr-0024-chain\n core eval --list\n core eval cognition\n core eval cognition --json --save\n core eval cognition --split dev --version v1\n core eval cognition --split holdout\n core eval contemplation_quality\n core eval contemplation_quality --json --save\n core workbench api\n core workbench api --port 9000\n core workbench api --host 0.0.0.0 --allow-nonlocal-bind"
|
||||
|
||||
_TEST_SUITES: dict[str, tuple[str, ...]] = {
|
||||
"fast": (
|
||||
|
|
@ -1377,6 +1377,120 @@ def cmd_teaching_propose(args: argparse.Namespace) -> int:
|
|||
return 0 if rec["state"] in ("pending", "accepted") else 1
|
||||
|
||||
|
||||
def cmd_teaching_propose_from_exemplars(args: argparse.Namespace) -> int:
|
||||
"""ADR-0163 Phase C — propose recognizers from admissibility exemplar corpora.
|
||||
|
||||
Loads one or more Phase B exemplar JSONLs, runs the contemplation
|
||||
synthesis to produce a :class:`DiscoveryCandidate` per corpus, and
|
||||
routes each candidate through :func:`teaching.proposals.propose_from_candidate`
|
||||
with the admissibility replay gate substituted for the cognition-only
|
||||
replay-equivalence gate. Proposals land as ``pending``; operator
|
||||
ratifies via ``core teaching review`` (existing path).
|
||||
"""
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from teaching.contemplation import contemplate_exemplar_corpus
|
||||
from teaching.exemplar_ingest import (
|
||||
ExemplarIngestError,
|
||||
list_corpora,
|
||||
load_exemplar_corpus,
|
||||
)
|
||||
from teaching.proposals import (
|
||||
DEFAULT_PROPOSAL_LOG_PATH,
|
||||
ProposalError,
|
||||
ProposalLog,
|
||||
propose_from_candidate,
|
||||
)
|
||||
from teaching.replay import run_admissibility_replay_gate
|
||||
from teaching.source import ProposalSource
|
||||
|
||||
review_date = args.review_date or datetime.now(timezone.utc).strftime("%Y-%m-%d")
|
||||
|
||||
log_path = Path(args.log) if args.log else DEFAULT_PROPOSAL_LOG_PATH
|
||||
log = ProposalLog(log_path)
|
||||
|
||||
# Resolve corpora: --all loads every JSONL; otherwise the single path.
|
||||
try:
|
||||
if args.all:
|
||||
root = Path(args.exemplar_path) if args.exemplar_path else None
|
||||
corpora = list_corpora(root)
|
||||
else:
|
||||
if not args.exemplar_path:
|
||||
_die(
|
||||
"exemplar_path is required unless --all is passed",
|
||||
code=2,
|
||||
)
|
||||
corpora = (load_exemplar_corpus(Path(args.exemplar_path)),)
|
||||
except ExemplarIngestError as exc:
|
||||
_die(f"exemplar ingest failed: {exc}", code=1)
|
||||
|
||||
# Resolve current git revision once for the ProposalSource stamp.
|
||||
from teaching.proposals import _current_revision
|
||||
revision = _current_revision()
|
||||
|
||||
results: list[dict[str, Any]] = []
|
||||
for corpus in corpora:
|
||||
candidate = contemplate_exemplar_corpus(corpus)
|
||||
source = ProposalSource(
|
||||
kind="exemplar_corpus",
|
||||
source_id=corpus.corpus_digest,
|
||||
emitted_at_revision=revision,
|
||||
)
|
||||
# Bind active_corpus_path=None so the gate reads the live corpus.
|
||||
def _gate(chain: dict[str, Any]) -> Any:
|
||||
return run_admissibility_replay_gate(
|
||||
candidate.proposed_chain.get("recognizer_spec"),
|
||||
)
|
||||
try:
|
||||
proposal = propose_from_candidate(
|
||||
candidate,
|
||||
log=log,
|
||||
run_replay=_gate,
|
||||
source=source,
|
||||
)
|
||||
except ProposalError as exc:
|
||||
_die(
|
||||
f"ineligible candidate for {corpus.shape_category.value}: {exc}",
|
||||
code=1,
|
||||
)
|
||||
rec = log.find(proposal.proposal_id)
|
||||
result = {
|
||||
"shape_category": corpus.shape_category.value,
|
||||
"corpus_path": str(corpus.path),
|
||||
"corpus_digest": corpus.corpus_digest,
|
||||
"proposal_id": proposal.proposal_id,
|
||||
"review_date": review_date,
|
||||
"state": rec["state"] if rec else "unknown",
|
||||
}
|
||||
replay = (rec or {}).get("replay_evidence") or {}
|
||||
if replay:
|
||||
result["replay_equivalent"] = bool(replay.get("replay_equivalent"))
|
||||
result["regressed_metrics"] = list(replay.get("regressed_metrics") or ())
|
||||
result["wrong_count_delta"] = int(replay.get("wrong_count_delta", 0))
|
||||
results.append(result)
|
||||
|
||||
if args.json:
|
||||
print(json.dumps({"proposals": results}, indent=2, sort_keys=True))
|
||||
else:
|
||||
for r in results:
|
||||
print(f"shape_category : {r['shape_category']}")
|
||||
print(f"corpus_path : {r['corpus_path']}")
|
||||
print(f"corpus_digest : {r['corpus_digest'][:16]}...")
|
||||
print(f"proposal_id : {r['proposal_id']}")
|
||||
print(f"state : {r['state']}")
|
||||
if "replay_equivalent" in r:
|
||||
print(f"replay_equivalent: {r['replay_equivalent']}")
|
||||
if r.get("regressed_metrics"):
|
||||
print(f"regressed_metrics: {', '.join(r['regressed_metrics'])}")
|
||||
print(f"wrong_count_delta: {r['wrong_count_delta']}")
|
||||
print(f"review_date : {r['review_date']}")
|
||||
print("--")
|
||||
# Exit nonzero if any proposal auto-rejected.
|
||||
if any(r["state"] != "pending" for r in results):
|
||||
return 1
|
||||
return 0
|
||||
|
||||
|
||||
def _load_findings_jsonl(path: str) -> list:
|
||||
"""Load ContemplationFinding objects from a JSONL file (W-019)."""
|
||||
from core.contemplation.schema import (
|
||||
|
|
@ -3822,6 +3936,50 @@ def build_parser() -> argparse.ArgumentParser:
|
|||
)
|
||||
teaching_propose.set_defaults(func=cmd_teaching_propose)
|
||||
|
||||
# ADR-0163 Phase C — propose recognizers from admissibility exemplar corpora.
|
||||
teaching_propose_from_exemplars = teaching_sub.add_parser(
|
||||
"propose-from-exemplars",
|
||||
help=(
|
||||
"synthesize a DerivedRecognizer proposal from a Phase B "
|
||||
"admissibility exemplar corpus (ADR-0163.C)"
|
||||
),
|
||||
)
|
||||
teaching_propose_from_exemplars.add_argument(
|
||||
"exemplar_path",
|
||||
nargs="?",
|
||||
default=None,
|
||||
help=(
|
||||
"path to a single exemplar JSONL "
|
||||
"(omit when passing --all; a directory may be passed with --all)"
|
||||
),
|
||||
)
|
||||
teaching_propose_from_exemplars.add_argument(
|
||||
"--all",
|
||||
action="store_true",
|
||||
help=(
|
||||
"ingest every *_v1.jsonl under teaching/admissibility_exemplars/ "
|
||||
"(or the directory passed as exemplar_path)"
|
||||
),
|
||||
)
|
||||
teaching_propose_from_exemplars.add_argument(
|
||||
"--review-date",
|
||||
default=None,
|
||||
help="ISO date stamped on the proposal record (default: today UTC)",
|
||||
)
|
||||
teaching_propose_from_exemplars.add_argument(
|
||||
"--log",
|
||||
default=None,
|
||||
help="proposal log path (default: teaching/proposals/proposals.jsonl)",
|
||||
)
|
||||
teaching_propose_from_exemplars.add_argument(
|
||||
"--json",
|
||||
action="store_true",
|
||||
help="machine-readable output",
|
||||
)
|
||||
teaching_propose_from_exemplars.set_defaults(
|
||||
func=cmd_teaching_propose_from_exemplars,
|
||||
)
|
||||
|
||||
# W-019 — miner and curriculum proposal construction paths (ADR-0095/0104)
|
||||
teaching_propose_miner = teaching_sub.add_parser(
|
||||
"propose-miner",
|
||||
|
|
|
|||
|
|
@ -32,6 +32,7 @@ same Phase B sink as JSONL lines.
|
|||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
from dataclasses import replace
|
||||
from typing import Any, Callable, Literal
|
||||
|
||||
|
|
@ -500,6 +501,110 @@ def contemplate(
|
|||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ADR-0163 Phase C — exemplar-corpus contemplation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
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
|
||||
contemplation pipeline against an unchanged corpus is a no-op for
|
||||
the proposal log (idempotency via ProposalLog.find).
|
||||
"""
|
||||
blob = json.dumps(
|
||||
{"corpus_digest": corpus_digest, "spec_digest": spec_digest},
|
||||
sort_keys=True,
|
||||
separators=(",", ":"),
|
||||
)
|
||||
return hashlib.sha256(blob.encode("utf-8")).hexdigest()
|
||||
|
||||
|
||||
def contemplate_exemplar_corpus(corpus: Any) -> DiscoveryCandidate:
|
||||
"""Return a :class:`DiscoveryCandidate` distilled from *corpus*.
|
||||
|
||||
Ingests a single :class:`~teaching.exemplar_ingest.ExemplarCorpus`,
|
||||
synthesizes its :class:`~teaching.recognizer_synthesis.RecognizerSpec`,
|
||||
and serializes both into a complete-shape ``DiscoveryCandidate`` that
|
||||
the existing proposal pipeline can consume.
|
||||
|
||||
Trust boundary
|
||||
- Pure: no filesystem writes, no global state, no LLM, no
|
||||
stochastic sampling.
|
||||
- The returned candidate carries ``polarity="affirms"`` — exemplars
|
||||
are reviewed-evidence-floor material under ADR-0163 §Phase B —
|
||||
and one ``EvidencePointer`` per ingested exemplar, sourced from
|
||||
the exemplar corpus itself. ``ref`` strings carry the verbatim
|
||||
``case_id`` (when present) or ``exemplar:<exemplar_id>`` so the
|
||||
proposal log records every seed cited.
|
||||
- Encodes the recognizer-shaped chain as a synthetic
|
||||
``(shape_category, "admissibility", "recognizes", spec_digest)``
|
||||
tuple so ``proposed_chain`` satisfies the four-field completeness
|
||||
gate enforced by ``check_eligibility``. The full
|
||||
:class:`RecognizerSpec` rides along as a ``recognizer_spec``
|
||||
sub-mapping on ``proposed_chain``.
|
||||
"""
|
||||
# Deferred imports keep this module's import cost cheap for
|
||||
# callers that never trigger Phase C ingest.
|
||||
from teaching.exemplar_ingest import ExemplarCorpus
|
||||
from teaching.recognizer_synthesis import (
|
||||
RecognizerSpec,
|
||||
synthesize_recognizer,
|
||||
)
|
||||
|
||||
if not isinstance(corpus, ExemplarCorpus):
|
||||
raise TypeError(
|
||||
f"contemplate_exemplar_corpus expects ExemplarCorpus; got "
|
||||
f"{type(corpus).__name__}"
|
||||
)
|
||||
|
||||
spec: RecognizerSpec = synthesize_recognizer(corpus)
|
||||
spec_digest = spec.spec_digest()
|
||||
|
||||
proposed_chain: dict[str, Any] = {
|
||||
"subject": spec.shape_category.value,
|
||||
"intent": "admissibility",
|
||||
"connective": "recognizes",
|
||||
"object": spec_digest,
|
||||
"recognizer_spec": spec.as_dict(),
|
||||
}
|
||||
|
||||
evidence: tuple[EvidencePointer, ...] = tuple(
|
||||
EvidencePointer(
|
||||
source="corpus",
|
||||
ref=(
|
||||
f"exemplar:{ex.case_id}"
|
||||
if ex.case_id
|
||||
else f"exemplar:{ex.exemplar_id}"
|
||||
),
|
||||
polarity="affirms",
|
||||
epistemic_status="coherent",
|
||||
)
|
||||
for ex in corpus.exemplars
|
||||
)
|
||||
|
||||
candidate_id = _exemplar_candidate_id(corpus.corpus_digest, spec_digest)
|
||||
|
||||
return DiscoveryCandidate(
|
||||
candidate_id=candidate_id,
|
||||
proposed_chain=proposed_chain,
|
||||
trigger="would_have_grounded",
|
||||
source_turn_trace=f"exemplar_corpus:{corpus.corpus_digest}",
|
||||
pack_consistent=True,
|
||||
boundary_clean=True,
|
||||
review_state="unreviewed",
|
||||
polarity="affirms",
|
||||
claim_domain="factual",
|
||||
evidence=evidence,
|
||||
sub_questions=(),
|
||||
contemplation_depth=0,
|
||||
recursion_overflow=False,
|
||||
)
|
||||
|
||||
|
||||
__all__ = [
|
||||
"contemplate",
|
||||
"contemplate_exemplar_corpus",
|
||||
]
|
||||
|
|
|
|||
358
teaching/exemplar_ingest.py
Normal file
358
teaching/exemplar_ingest.py
Normal file
|
|
@ -0,0 +1,358 @@
|
|||
"""ADR-0163 Phase C — admissibility exemplar ingest.
|
||||
|
||||
Pure-function loader for the operator-authored exemplar corpora under
|
||||
``teaching/admissibility_exemplars/``. Returns frozen :class:`ExemplarCorpus`
|
||||
records whose canonical bytes (sorted JSONL, single trailing newline) the
|
||||
:attr:`ExemplarCorpus.corpus_digest` field hashes deterministically.
|
||||
|
||||
Trust boundary
|
||||
- Pure functions. The only file read is the path supplied by the caller
|
||||
(or, in ``list_corpora``, the contents of
|
||||
``teaching/admissibility_exemplars/``). No global state, no caches
|
||||
outlive a call, no writes.
|
||||
- Validation is rules-only. No LLM, no embedding, no learned classifier.
|
||||
- The schema enforced here mirrors
|
||||
``teaching/admissibility_exemplars/contract.md`` and the per-category
|
||||
dispatcher pattern in ``tests/test_admissibility_exemplars.py``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any, Mapping
|
||||
|
||||
from evals.refusal_taxonomy.shape_categories import ShapeCategory
|
||||
|
||||
|
||||
_EXEMPLARS_ROOT_DEFAULT: Path = (
|
||||
Path(__file__).resolve().parent / "admissibility_exemplars"
|
||||
)
|
||||
|
||||
_REQUIRED_TOP_KEYS: frozenset[str] = frozenset({
|
||||
"exemplar_id", "shape_category", "statement", "expected_graph", "provenance",
|
||||
})
|
||||
_REQUIRED_GRAPH_KEYS: frozenset[str] = frozenset({
|
||||
"subject", "quantity_anchors", "graph_intent", "outcome",
|
||||
})
|
||||
_REQUIRED_PROVENANCE_KEYS: frozenset[str] = frozenset({
|
||||
"source", "author", "round", "category_rank",
|
||||
})
|
||||
|
||||
_VALID_WINDOW_UNITS: frozenset[str] = frozenset({
|
||||
"day", "week", "month", "year", "hour", "minute", "second",
|
||||
})
|
||||
_VALID_WINDOW_QUANTIFIERS: frozenset[str] = frozenset({"each", "every", "per"})
|
||||
_VALID_CURRENCY_SYMBOLS: frozenset[str] = frozenset({"$", "£", "€", "¥"})
|
||||
_VALID_AMOUNT_KINDS: frozenset[str] = frozenset({"integer", "decimal", "word"})
|
||||
|
||||
|
||||
# The categories Phase C ingests in round 1. Adding a category here
|
||||
# requires landing its exemplar corpus + its synthesizer first.
|
||||
_SUPPORTED_CATEGORIES: frozenset[ShapeCategory] = frozenset({
|
||||
ShapeCategory.DESCRIPTIVE_SETUP_NO_QUANTITY,
|
||||
ShapeCategory.TEMPORAL_AGGREGATION,
|
||||
ShapeCategory.RATE_WITH_CURRENCY,
|
||||
})
|
||||
|
||||
|
||||
class ExemplarIngestError(ValueError):
|
||||
"""Raised when an exemplar JSONL violates the Phase B contract."""
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Exemplar:
|
||||
"""One parsed exemplar record.
|
||||
|
||||
Mirrors the JSONL line verbatim. ``expected_graph`` and
|
||||
``provenance`` keep their full submaps so the synthesizer can read
|
||||
every field the contract surfaces (including the optional
|
||||
``author_note``).
|
||||
"""
|
||||
|
||||
exemplar_id: str
|
||||
shape_category: ShapeCategory
|
||||
statement: str
|
||||
expected_graph: Mapping[str, Any]
|
||||
provenance: Mapping[str, Any]
|
||||
|
||||
@property
|
||||
def case_id(self) -> str | None:
|
||||
"""Optional GSM8K train-sample case_id this exemplar cites."""
|
||||
cid = self.provenance.get("train_case_id")
|
||||
return str(cid) if cid else None
|
||||
|
||||
@property
|
||||
def author_note(self) -> str | None:
|
||||
note = self.provenance.get("author_note")
|
||||
return str(note) if note else None
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ExemplarCorpus:
|
||||
"""One ingested exemplar corpus + the digest of its canonical bytes.
|
||||
|
||||
``corpus_digest`` is a sha256 over the file's canonical re-encoding
|
||||
(sorted by ``exemplar_id``, sorted-key JSON, single trailing newline).
|
||||
Two corpora whose seeds carry identical content produce identical
|
||||
digests regardless of incidental whitespace.
|
||||
"""
|
||||
|
||||
shape_category: ShapeCategory
|
||||
path: Path
|
||||
exemplars: tuple[Exemplar, ...]
|
||||
corpus_digest: str
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Per-category validation dispatch
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _require_keys(
|
||||
ctx: str, payload: Mapping[str, Any], required: frozenset[str]
|
||||
) -> None:
|
||||
missing = required - set(payload.keys())
|
||||
if missing:
|
||||
raise ExemplarIngestError(
|
||||
f"{ctx} missing required keys: {sorted(missing)}"
|
||||
)
|
||||
|
||||
|
||||
def _validate_descriptive_setup(ctx: str, graph: Mapping[str, Any]) -> None:
|
||||
anchors = graph["quantity_anchors"]
|
||||
if not isinstance(anchors, list):
|
||||
raise ExemplarIngestError(f"{ctx} quantity_anchors must be list")
|
||||
if anchors != []:
|
||||
raise ExemplarIngestError(
|
||||
f"{ctx} descriptive_setup_no_quantity requires empty anchors"
|
||||
)
|
||||
if graph["graph_intent"] != "setup":
|
||||
raise ExemplarIngestError(f"{ctx} graph_intent must be 'setup'")
|
||||
if graph["outcome"] != "inadmissible_by_design":
|
||||
raise ExemplarIngestError(
|
||||
f"{ctx} outcome must be 'inadmissible_by_design'"
|
||||
)
|
||||
|
||||
|
||||
def _validate_temporal_aggregation(ctx: str, graph: Mapping[str, Any]) -> None:
|
||||
anchors = graph["quantity_anchors"]
|
||||
if not isinstance(anchors, list) or not anchors:
|
||||
raise ExemplarIngestError(f"{ctx} temporal_aggregation needs ≥1 anchor")
|
||||
for a in anchors:
|
||||
if not isinstance(a, Mapping):
|
||||
raise ExemplarIngestError(f"{ctx} anchor must be a mapping")
|
||||
_require_keys(ctx, a, frozenset({
|
||||
"kind", "count_token", "window_unit",
|
||||
"window_quantifier", "subject_role",
|
||||
}))
|
||||
if a["kind"] != "event_count_per_window":
|
||||
raise ExemplarIngestError(
|
||||
f"{ctx} anchor kind must be 'event_count_per_window'"
|
||||
)
|
||||
if a["window_unit"] not in _VALID_WINDOW_UNITS:
|
||||
raise ExemplarIngestError(
|
||||
f"{ctx} window_unit {a['window_unit']!r} not in "
|
||||
f"{sorted(_VALID_WINDOW_UNITS)}"
|
||||
)
|
||||
if a["window_quantifier"] not in _VALID_WINDOW_QUANTIFIERS:
|
||||
raise ExemplarIngestError(
|
||||
f"{ctx} window_quantifier {a['window_quantifier']!r} not in "
|
||||
f"{sorted(_VALID_WINDOW_QUANTIFIERS)}"
|
||||
)
|
||||
if not isinstance(a["count_token"], str) or not a["count_token"]:
|
||||
raise ExemplarIngestError(f"{ctx} count_token must be non-empty str")
|
||||
if not isinstance(a["subject_role"], str) or not a["subject_role"]:
|
||||
raise ExemplarIngestError(f"{ctx} subject_role must be non-empty str")
|
||||
if graph["graph_intent"] != "aggregate":
|
||||
raise ExemplarIngestError(f"{ctx} graph_intent must be 'aggregate'")
|
||||
if graph["outcome"] != "admissible":
|
||||
raise ExemplarIngestError(f"{ctx} outcome must be 'admissible'")
|
||||
|
||||
|
||||
def _validate_rate_with_currency(ctx: str, graph: Mapping[str, Any]) -> None:
|
||||
anchors = graph["quantity_anchors"]
|
||||
if not isinstance(anchors, list) or not anchors:
|
||||
raise ExemplarIngestError(f"{ctx} rate_with_currency needs ≥1 anchor")
|
||||
for a in anchors:
|
||||
if not isinstance(a, Mapping):
|
||||
raise ExemplarIngestError(f"{ctx} anchor must be a mapping")
|
||||
_require_keys(ctx, a, frozenset({
|
||||
"kind", "currency_symbol", "amount", "amount_kind",
|
||||
"per_unit", "subject_role",
|
||||
}))
|
||||
if a["kind"] != "currency_per_unit_rate":
|
||||
raise ExemplarIngestError(
|
||||
f"{ctx} anchor kind must be 'currency_per_unit_rate'"
|
||||
)
|
||||
if a["currency_symbol"] not in _VALID_CURRENCY_SYMBOLS:
|
||||
raise ExemplarIngestError(
|
||||
f"{ctx} currency_symbol {a['currency_symbol']!r} not in "
|
||||
f"{sorted(_VALID_CURRENCY_SYMBOLS)}"
|
||||
)
|
||||
if a["amount_kind"] not in _VALID_AMOUNT_KINDS:
|
||||
raise ExemplarIngestError(
|
||||
f"{ctx} amount_kind {a['amount_kind']!r} not in "
|
||||
f"{sorted(_VALID_AMOUNT_KINDS)}"
|
||||
)
|
||||
for fld in ("amount", "per_unit", "subject_role"):
|
||||
if not isinstance(a[fld], str) or not a[fld]:
|
||||
raise ExemplarIngestError(
|
||||
f"{ctx} {fld} must be non-empty str"
|
||||
)
|
||||
if graph["graph_intent"] != "rate":
|
||||
raise ExemplarIngestError(f"{ctx} graph_intent must be 'rate'")
|
||||
if graph["outcome"] != "admissible":
|
||||
raise ExemplarIngestError(f"{ctx} outcome must be 'admissible'")
|
||||
|
||||
|
||||
_CATEGORY_VALIDATORS = {
|
||||
ShapeCategory.DESCRIPTIVE_SETUP_NO_QUANTITY: _validate_descriptive_setup,
|
||||
ShapeCategory.TEMPORAL_AGGREGATION: _validate_temporal_aggregation,
|
||||
ShapeCategory.RATE_WITH_CURRENCY: _validate_rate_with_currency,
|
||||
}
|
||||
|
||||
|
||||
def _parse_record(path: Path, idx: int, raw: Mapping[str, Any]) -> Exemplar:
|
||||
ctx = f"{path}:{idx}"
|
||||
_require_keys(ctx, raw, _REQUIRED_TOP_KEYS)
|
||||
|
||||
cat_str = raw["shape_category"]
|
||||
if not any(cat_str == c.value for c in ShapeCategory):
|
||||
raise ExemplarIngestError(
|
||||
f"{ctx} shape_category {cat_str!r} not in ShapeCategory"
|
||||
)
|
||||
shape_category = ShapeCategory(cat_str)
|
||||
if shape_category not in _SUPPORTED_CATEGORIES:
|
||||
raise ExemplarIngestError(
|
||||
f"{ctx} shape_category {cat_str!r} is not a Phase C round-1 "
|
||||
f"category; supported = "
|
||||
f"{sorted(c.value for c in _SUPPORTED_CATEGORIES)}"
|
||||
)
|
||||
|
||||
statement = raw["statement"]
|
||||
if not isinstance(statement, str) or not statement:
|
||||
raise ExemplarIngestError(f"{ctx} statement must be non-empty str")
|
||||
|
||||
graph = raw["expected_graph"]
|
||||
if not isinstance(graph, Mapping):
|
||||
raise ExemplarIngestError(f"{ctx} expected_graph must be a mapping")
|
||||
_require_keys(ctx, graph, _REQUIRED_GRAPH_KEYS)
|
||||
|
||||
prov = raw["provenance"]
|
||||
if not isinstance(prov, Mapping):
|
||||
raise ExemplarIngestError(f"{ctx} provenance must be a mapping")
|
||||
_require_keys(ctx, prov, _REQUIRED_PROVENANCE_KEYS)
|
||||
|
||||
_CATEGORY_VALIDATORS[shape_category](ctx, graph)
|
||||
|
||||
return Exemplar(
|
||||
exemplar_id=str(raw["exemplar_id"]),
|
||||
shape_category=shape_category,
|
||||
statement=statement,
|
||||
expected_graph=dict(graph),
|
||||
provenance=dict(prov),
|
||||
)
|
||||
|
||||
|
||||
def _canonical_bytes(records: list[Mapping[str, Any]]) -> bytes:
|
||||
"""Re-encode records as sorted-by-exemplar_id canonical JSONL bytes.
|
||||
|
||||
Two physically different files whose records carry identical content
|
||||
produce the same canonical bytes (and hence the same ``corpus_digest``).
|
||||
Trailing whitespace, key ordering inside records, and line-by-line
|
||||
insertion order are all normalized.
|
||||
"""
|
||||
sorted_records = sorted(records, key=lambda r: r["exemplar_id"])
|
||||
chunks = []
|
||||
for r in sorted_records:
|
||||
chunks.append(json.dumps(r, sort_keys=True, separators=(",", ":")))
|
||||
return ("\n".join(chunks) + "\n").encode("utf-8")
|
||||
|
||||
|
||||
def load_exemplar_corpus(path: Path) -> ExemplarCorpus:
|
||||
"""Load and validate one exemplar corpus from *path*.
|
||||
|
||||
Pure function. Same path + same bytes → identical
|
||||
:class:`ExemplarCorpus`. Raises :class:`ExemplarIngestError` for any
|
||||
contract violation; partial corpora are never returned.
|
||||
"""
|
||||
if not path.exists():
|
||||
raise ExemplarIngestError(f"exemplar corpus not found: {path}")
|
||||
raw = path.read_text(encoding="utf-8")
|
||||
if not raw:
|
||||
raise ExemplarIngestError(f"exemplar corpus is empty: {path}")
|
||||
records_raw: list[Mapping[str, Any]] = []
|
||||
parsed: list[Exemplar] = []
|
||||
for idx, line in enumerate(raw.splitlines(), start=1):
|
||||
if not line.strip():
|
||||
continue
|
||||
try:
|
||||
record = json.loads(line)
|
||||
except json.JSONDecodeError as exc:
|
||||
raise ExemplarIngestError(
|
||||
f"{path}:{idx} invalid JSON: {exc.msg}"
|
||||
) from exc
|
||||
if not isinstance(record, Mapping):
|
||||
raise ExemplarIngestError(
|
||||
f"{path}:{idx} record must be a JSON object"
|
||||
)
|
||||
records_raw.append(record)
|
||||
parsed.append(_parse_record(path, idx, record))
|
||||
|
||||
# File-name to category binding. The contract guarantees one
|
||||
# category per file; enforce it on read so a misnamed file fails
|
||||
# loudly rather than silently producing a mixed corpus.
|
||||
category = parsed[0].shape_category
|
||||
for ex in parsed[1:]:
|
||||
if ex.shape_category != category:
|
||||
raise ExemplarIngestError(
|
||||
f"{path} mixes categories: {category.value!r} and "
|
||||
f"{ex.shape_category.value!r} both present"
|
||||
)
|
||||
expected_stem = f"{category.value}_v1"
|
||||
if path.stem != expected_stem:
|
||||
raise ExemplarIngestError(
|
||||
f"{path} stem {path.stem!r} does not match category "
|
||||
f"{category.value!r}; expected stem {expected_stem!r}"
|
||||
)
|
||||
|
||||
# Deterministic order on the in-memory list mirrors the canonical
|
||||
# bytes the digest is computed over.
|
||||
parsed.sort(key=lambda e: e.exemplar_id)
|
||||
|
||||
digest = hashlib.sha256(_canonical_bytes(records_raw)).hexdigest()
|
||||
|
||||
return ExemplarCorpus(
|
||||
shape_category=category,
|
||||
path=path,
|
||||
exemplars=tuple(parsed),
|
||||
corpus_digest=digest,
|
||||
)
|
||||
|
||||
|
||||
def list_corpora(root: Path | None = None) -> tuple[ExemplarCorpus, ...]:
|
||||
"""Load every ``*_v1.jsonl`` under *root* (default exemplars dir).
|
||||
|
||||
Returns corpora sorted by ``shape_category.value`` so callers get a
|
||||
stable iteration order regardless of filesystem listing semantics.
|
||||
"""
|
||||
base = root if root is not None else _EXEMPLARS_ROOT_DEFAULT
|
||||
if not base.is_dir():
|
||||
raise ExemplarIngestError(f"exemplars root is not a directory: {base}")
|
||||
corpora: list[ExemplarCorpus] = []
|
||||
for path in sorted(base.glob("*_v1.jsonl")):
|
||||
corpora.append(load_exemplar_corpus(path))
|
||||
corpora.sort(key=lambda c: c.shape_category.value)
|
||||
return tuple(corpora)
|
||||
|
||||
|
||||
__all__ = [
|
||||
"Exemplar",
|
||||
"ExemplarCorpus",
|
||||
"ExemplarIngestError",
|
||||
"list_corpora",
|
||||
"load_exemplar_corpus",
|
||||
]
|
||||
292
teaching/recognizer_synthesis.py
Normal file
292
teaching/recognizer_synthesis.py
Normal file
|
|
@ -0,0 +1,292 @@
|
|||
"""ADR-0163 Phase C — admissibility recognizer synthesis.
|
||||
|
||||
Distill an :class:`~teaching.exemplar_ingest.ExemplarCorpus` into one
|
||||
:class:`RecognizerSpec`: a typed shape specification consumed downstream
|
||||
by the Phase D / Phase E candidate-graph admissibility surface.
|
||||
|
||||
Doctrine (non-negotiable)
|
||||
- Deterministic: same corpus → same :class:`RecognizerSpec`,
|
||||
byte-identical when re-serialized.
|
||||
- Narrower, not broader, than the seeds. Observed-only sub-shapes are
|
||||
named explicitly; the recognizer does not generalize to currency
|
||||
symbols, window units, or per-unit measures the seeds never carried.
|
||||
- Doctrine-compatible with Phase B author_notes. Each author_note is
|
||||
either honored by a per-category branch *or* surfaced in
|
||||
``canonical_pattern.unresolved_notes`` for Phase D review — never
|
||||
silently dropped.
|
||||
- No hidden normalization. Seed strings flow through verbatim.
|
||||
|
||||
The module is pure: rules-only, no LLM call, no embedding, no learned
|
||||
classifier, no I/O beyond reading the supplied corpus dataclass.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Mapping
|
||||
|
||||
from evals.refusal_taxonomy.shape_categories import ShapeCategory
|
||||
from teaching.exemplar_ingest import Exemplar, ExemplarCorpus
|
||||
|
||||
|
||||
class RecognizerSynthesisError(ValueError):
|
||||
"""Raised when a corpus is structurally unsynthesizable."""
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class RecognizerSpec:
|
||||
"""The distilled, narrowest commitment that subsumes every seed.
|
||||
|
||||
Phase C produces the spec. Phase D's review surface is where the
|
||||
operator may choose to widen any ``observed_*`` set. Phase E's
|
||||
measurement re-runs the GSM8K + capability lanes with the widened
|
||||
recognizer to verify ``wrong = 0`` still holds.
|
||||
|
||||
``canonical_pattern`` is the load-bearing field. Its keys are
|
||||
per-category bespoke; consumers MUST branch on ``shape_category``
|
||||
before reading.
|
||||
"""
|
||||
|
||||
shape_category: ShapeCategory
|
||||
canonical_pattern: Mapping[str, Any]
|
||||
exemplar_count: int
|
||||
exemplar_digest: str
|
||||
coverage: Mapping[str, int]
|
||||
|
||||
def canonical_bytes(self) -> bytes:
|
||||
"""Canonical sorted-key JSON bytes — what the proposal_id hashes."""
|
||||
payload = {
|
||||
"shape_category": self.shape_category.value,
|
||||
"canonical_pattern": _as_jsonable(self.canonical_pattern),
|
||||
"exemplar_count": self.exemplar_count,
|
||||
"exemplar_digest": self.exemplar_digest,
|
||||
"coverage": dict(self.coverage),
|
||||
}
|
||||
return json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
|
||||
|
||||
def spec_digest(self) -> str:
|
||||
"""sha256 over :meth:`canonical_bytes`; identifies the spec."""
|
||||
return hashlib.sha256(self.canonical_bytes()).hexdigest()
|
||||
|
||||
def as_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"shape_category": self.shape_category.value,
|
||||
"canonical_pattern": _as_jsonable(self.canonical_pattern),
|
||||
"exemplar_count": self.exemplar_count,
|
||||
"exemplar_digest": self.exemplar_digest,
|
||||
"coverage": dict(self.coverage),
|
||||
}
|
||||
|
||||
|
||||
def _as_jsonable(payload: Any) -> Any:
|
||||
"""Recursively coerce mappings/sequences to JSON-serializable dicts/lists.
|
||||
|
||||
Tuples become lists; frozensets become sorted lists. Used so the
|
||||
``canonical_pattern`` mapping's value tree round-trips byte-identically
|
||||
through :func:`json.dumps(sort_keys=True)`.
|
||||
"""
|
||||
if isinstance(payload, Mapping):
|
||||
return {k: _as_jsonable(v) for k, v in payload.items()}
|
||||
if isinstance(payload, (list, tuple)):
|
||||
return [_as_jsonable(v) for v in payload]
|
||||
if isinstance(payload, (set, frozenset)):
|
||||
return sorted(_as_jsonable(v) for v in payload)
|
||||
return payload
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Shared helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _collect_author_notes(exemplars: tuple[Exemplar, ...]) -> list[str]:
|
||||
"""Deduplicated, sorted author_notes — Phase B operator surface."""
|
||||
notes: set[str] = set()
|
||||
for ex in exemplars:
|
||||
note = ex.author_note
|
||||
if note:
|
||||
notes.add(note)
|
||||
return sorted(notes)
|
||||
|
||||
|
||||
def _sorted_unique(values: list[Any]) -> list[Any]:
|
||||
seen: set[Any] = set()
|
||||
out: list[Any] = []
|
||||
for v in sorted(values, key=lambda x: str(x)):
|
||||
if v not in seen:
|
||||
seen.add(v)
|
||||
out.append(v)
|
||||
return out
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Per-category synthesizers — flat aggregations, no smart generalization
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _synthesize_descriptive_setup_no_quantity(
|
||||
corpus: ExemplarCorpus,
|
||||
) -> tuple[Mapping[str, Any], Mapping[str, int]]:
|
||||
"""All seeds: zero anchors, graph_intent=setup, outcome=inadmissible_by_design.
|
||||
|
||||
The recognizer's commitment is exactly that: a statement with no
|
||||
extractable quantity must be admitted as setup context, not refused.
|
||||
Narrowness rule: anchor_count is pinned at 0 (no widening).
|
||||
"""
|
||||
exemplars = corpus.exemplars
|
||||
subjects_observed_null = sum(1 for e in exemplars if e.expected_graph.get("subject") is None)
|
||||
subjects_observed_named = sum(1 for e in exemplars if e.expected_graph.get("subject"))
|
||||
# Sanity: validator already pinned this; assert defensively.
|
||||
for ex in exemplars:
|
||||
if ex.expected_graph["quantity_anchors"] != []:
|
||||
raise RecognizerSynthesisError(
|
||||
f"{ex.exemplar_id}: descriptive_setup_no_quantity seed has "
|
||||
"non-empty anchors — corpus is structurally invalid"
|
||||
)
|
||||
canonical_pattern: dict[str, Any] = {
|
||||
"shape_category": ShapeCategory.DESCRIPTIVE_SETUP_NO_QUANTITY.value,
|
||||
"graph_intent": "setup",
|
||||
"outcome": "inadmissible_by_design",
|
||||
"quantity_anchor_count": 0,
|
||||
"subject_is_optional": True,
|
||||
"unresolved_notes": _collect_author_notes(exemplars),
|
||||
}
|
||||
coverage: dict[str, int] = {
|
||||
"anchors_empty": len(exemplars),
|
||||
"subject_null": subjects_observed_null,
|
||||
"subject_named": subjects_observed_named,
|
||||
}
|
||||
return canonical_pattern, coverage
|
||||
|
||||
|
||||
def _synthesize_temporal_aggregation(
|
||||
corpus: ExemplarCorpus,
|
||||
) -> tuple[Mapping[str, Any], Mapping[str, int]]:
|
||||
"""All anchors are event_count_per_window. Capture window axis exactly."""
|
||||
exemplars = corpus.exemplars
|
||||
window_units: list[str] = []
|
||||
window_quantifiers: list[str] = []
|
||||
anchor_counts: list[int] = []
|
||||
coverage_units: dict[str, int] = {}
|
||||
coverage_quantifiers: dict[str, int] = {}
|
||||
|
||||
for ex in exemplars:
|
||||
anchors = ex.expected_graph["quantity_anchors"]
|
||||
anchor_counts.append(len(anchors))
|
||||
for a in anchors:
|
||||
window_units.append(a["window_unit"])
|
||||
window_quantifiers.append(a["window_quantifier"])
|
||||
coverage_units[a["window_unit"]] = coverage_units.get(a["window_unit"], 0) + 1
|
||||
q = a["window_quantifier"]
|
||||
coverage_quantifiers[q] = coverage_quantifiers.get(q, 0) + 1
|
||||
|
||||
canonical_pattern: dict[str, Any] = {
|
||||
"shape_category": ShapeCategory.TEMPORAL_AGGREGATION.value,
|
||||
"graph_intent": "aggregate",
|
||||
"outcome": "admissible",
|
||||
"anchor_kind": "event_count_per_window",
|
||||
"observed_window_units": _sorted_unique(window_units),
|
||||
"observed_window_quantifiers": _sorted_unique(window_quantifiers),
|
||||
"anchor_count_min": min(anchor_counts),
|
||||
"anchor_count_max": max(anchor_counts),
|
||||
"unresolved_notes": _collect_author_notes(exemplars),
|
||||
}
|
||||
# Coverage histogram: per-anchor-kind + per-axis frequencies.
|
||||
coverage: dict[str, int] = {
|
||||
"anchors_event_count_per_window": sum(anchor_counts),
|
||||
}
|
||||
for unit, n in sorted(coverage_units.items()):
|
||||
coverage[f"window_unit:{unit}"] = n
|
||||
for q, n in sorted(coverage_quantifiers.items()):
|
||||
coverage[f"window_quantifier:{q}"] = n
|
||||
return canonical_pattern, coverage
|
||||
|
||||
|
||||
def _synthesize_rate_with_currency(
|
||||
corpus: ExemplarCorpus,
|
||||
) -> tuple[Mapping[str, Any], Mapping[str, int]]:
|
||||
"""All anchors are currency_per_unit_rate. Capture currency/unit/kind axes."""
|
||||
exemplars = corpus.exemplars
|
||||
currency_symbols: list[str] = []
|
||||
per_units: list[str] = []
|
||||
amount_kinds: list[str] = []
|
||||
anchor_counts: list[int] = []
|
||||
coverage_currency: dict[str, int] = {}
|
||||
coverage_per_unit: dict[str, int] = {}
|
||||
coverage_amount_kind: dict[str, int] = {}
|
||||
|
||||
for ex in exemplars:
|
||||
anchors = ex.expected_graph["quantity_anchors"]
|
||||
anchor_counts.append(len(anchors))
|
||||
for a in anchors:
|
||||
currency_symbols.append(a["currency_symbol"])
|
||||
per_units.append(a["per_unit"])
|
||||
amount_kinds.append(a["amount_kind"])
|
||||
coverage_currency[a["currency_symbol"]] = (
|
||||
coverage_currency.get(a["currency_symbol"], 0) + 1
|
||||
)
|
||||
coverage_per_unit[a["per_unit"]] = coverage_per_unit.get(a["per_unit"], 0) + 1
|
||||
coverage_amount_kind[a["amount_kind"]] = (
|
||||
coverage_amount_kind.get(a["amount_kind"], 0) + 1
|
||||
)
|
||||
|
||||
canonical_pattern: dict[str, Any] = {
|
||||
"shape_category": ShapeCategory.RATE_WITH_CURRENCY.value,
|
||||
"graph_intent": "rate",
|
||||
"outcome": "admissible",
|
||||
"anchor_kind": "currency_per_unit_rate",
|
||||
"observed_currency_symbols": _sorted_unique(currency_symbols),
|
||||
"observed_per_units": _sorted_unique(per_units),
|
||||
"observed_amount_kinds": _sorted_unique(amount_kinds),
|
||||
"anchor_count_min": min(anchor_counts),
|
||||
"anchor_count_max": max(anchor_counts),
|
||||
"unresolved_notes": _collect_author_notes(exemplars),
|
||||
}
|
||||
coverage: dict[str, int] = {
|
||||
"anchors_currency_per_unit_rate": sum(anchor_counts),
|
||||
}
|
||||
for sym, n in sorted(coverage_currency.items()):
|
||||
coverage[f"currency_symbol:{sym}"] = n
|
||||
for u, n in sorted(coverage_per_unit.items()):
|
||||
coverage[f"per_unit:{u}"] = n
|
||||
for k, n in sorted(coverage_amount_kind.items()):
|
||||
coverage[f"amount_kind:{k}"] = n
|
||||
return canonical_pattern, coverage
|
||||
|
||||
|
||||
_SYNTHESIZERS = {
|
||||
ShapeCategory.DESCRIPTIVE_SETUP_NO_QUANTITY: _synthesize_descriptive_setup_no_quantity,
|
||||
ShapeCategory.TEMPORAL_AGGREGATION: _synthesize_temporal_aggregation,
|
||||
ShapeCategory.RATE_WITH_CURRENCY: _synthesize_rate_with_currency,
|
||||
}
|
||||
|
||||
|
||||
def synthesize_recognizer(corpus: ExemplarCorpus) -> RecognizerSpec:
|
||||
"""Distil *corpus* into one :class:`RecognizerSpec`.
|
||||
|
||||
Pure function. Per-category dispatch chooses the synthesizer; common
|
||||
framing (digest, exemplar count) is bolted on uniformly.
|
||||
"""
|
||||
synth = _SYNTHESIZERS.get(corpus.shape_category)
|
||||
if synth is None: # pragma: no cover — defensive: ingest already gates
|
||||
raise RecognizerSynthesisError(
|
||||
f"no synthesizer registered for shape_category="
|
||||
f"{corpus.shape_category.value!r}"
|
||||
)
|
||||
canonical_pattern, coverage = synth(corpus)
|
||||
return RecognizerSpec(
|
||||
shape_category=corpus.shape_category,
|
||||
canonical_pattern=canonical_pattern,
|
||||
exemplar_count=len(corpus.exemplars),
|
||||
exemplar_digest=corpus.corpus_digest,
|
||||
coverage=coverage,
|
||||
)
|
||||
|
||||
|
||||
__all__ = [
|
||||
"RecognizerSpec",
|
||||
"RecognizerSynthesisError",
|
||||
"synthesize_recognizer",
|
||||
]
|
||||
|
|
@ -170,4 +170,279 @@ def run_replay_equivalence(chain: dict[str, Any]) -> ReplayEvidence:
|
|||
)
|
||||
|
||||
|
||||
__all__ = ["run_replay_equivalence"]
|
||||
# ---------------------------------------------------------------------------
|
||||
# ADR-0163 Phase C — admissibility replay gate
|
||||
# ---------------------------------------------------------------------------
|
||||
#
|
||||
# Extends the cognition-lane replay-equivalence gate with two additional
|
||||
# evidence lanes that the ``wrong = 0`` doctrine names explicitly
|
||||
# (ADR-0163 §Constraint #1):
|
||||
#
|
||||
# - every named capability axis (G1..G5, S1) at its public v1 split
|
||||
# - the GSM8K train_sample at evals/gsm8k_math/train_sample/v1/
|
||||
#
|
||||
# If accepting a proposal would lift the wrong count on the train sample
|
||||
# by one or more, the gate rejects with
|
||||
# ``regressed_metrics=["gsm8k_train_sample_wrong_count"]``. The
|
||||
# downstream ``propose_from_candidate`` then auto-rejects the proposal
|
||||
# before it ever reaches the operator queue.
|
||||
#
|
||||
# Phase C produces proposals only; the candidate run is identical to
|
||||
# baseline because the recognizer is not yet wired into the
|
||||
# candidate-graph (Phase D / E work). Tests inject a fake candidate
|
||||
# run to exercise the wrong-count invariant before the wiring exists.
|
||||
|
||||
import importlib
|
||||
from dataclasses import dataclass
|
||||
|
||||
|
||||
# Public v1 capability-axis lanes named by ADR-0163 §Phase A as the
|
||||
# wrong=0 floor. Stored as (lane_id, module_path) so the dispatch is
|
||||
# inspectable and tests can stub one lane at a time.
|
||||
_CAPABILITY_AXIS_LANES: tuple[tuple[str, str], ...] = (
|
||||
("G1_verb_classes", "evals.math_capability_axes.G1_verb_classes.v1.runner"),
|
||||
("G2_comparatives", "evals.math_capability_axes.G2_comparatives.v1.runner"),
|
||||
("G3_numerics", "evals.math_capability_axes.G3_numerics.v1.runner"),
|
||||
("G4_multi_clause", "evals.math_capability_axes.G4_multi_clause.v1.runner"),
|
||||
("G5_aggregate", "evals.math_capability_axes.G5_aggregate.v1.runner"),
|
||||
("S1_rate_events", "evals.math_capability_axes.S1_rate_events.v1.runner"),
|
||||
)
|
||||
|
||||
_GSM8K_TRAIN_SAMPLE_MODULE = "evals.gsm8k_math.train_sample.v1.runner"
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class AdmissibilityReplayEvidence:
|
||||
"""Evidence record for the Phase C admissibility gate.
|
||||
|
||||
Mirrors :class:`ReplayEvidence` for the cognition lane and bolts on
|
||||
per-axis + GSM8K train-sample wrong-count evidence. ``as_dict``
|
||||
keeps the cognition-lane fields at the top level so the existing
|
||||
``ProposalLog.record_replay`` consumer (which round-trips via
|
||||
``replay_evidence``) can read them unchanged.
|
||||
"""
|
||||
|
||||
baseline: dict[str, float]
|
||||
candidate: dict[str, float]
|
||||
regressed_metrics: tuple[str, ...]
|
||||
replay_equivalent: bool
|
||||
capability_axes: dict[str, dict[str, int]]
|
||||
gsm8k_train_sample: dict[str, int]
|
||||
wrong_count_delta: int
|
||||
|
||||
def as_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"baseline": dict(self.baseline),
|
||||
"candidate": dict(self.candidate),
|
||||
"regressed_metrics": list(self.regressed_metrics),
|
||||
"replay_equivalent": bool(self.replay_equivalent),
|
||||
"capability_axes": {
|
||||
k: dict(v) for k, v in self.capability_axes.items()
|
||||
},
|
||||
"gsm8k_train_sample": dict(self.gsm8k_train_sample),
|
||||
"wrong_count_delta": int(self.wrong_count_delta),
|
||||
}
|
||||
|
||||
|
||||
# In-process baseline cache (ADR-0163 §Phase C performance note).
|
||||
#
|
||||
# Key: sha256 of the active teaching-corpus bytes (b"" when absent).
|
||||
# Value: a frozen baseline tuple of (capability_axes, gsm8k_counts).
|
||||
# The cognition baseline reuses :func:`_run_cognition_public` directly;
|
||||
# it is comparatively cheap, so we don't cache it here.
|
||||
#
|
||||
# Invalidation: write the new digest -> evicts old key by lookup. The
|
||||
# cache lives in-process only; no filesystem persistence — Phase C
|
||||
# does not introduce a new persistence path (ADR-0161 §1).
|
||||
_BASELINE_CACHE: dict[str, dict[str, Any]] = {}
|
||||
|
||||
|
||||
def _active_corpus_digest(active_corpus_path: Path | None) -> str:
|
||||
"""sha256 of the active teaching-corpus bytes; '' when path absent."""
|
||||
path = active_corpus_path if active_corpus_path is not None else _tg._CORPUS_PATH
|
||||
if not path.exists():
|
||||
return ""
|
||||
import hashlib as _hashlib
|
||||
return _hashlib.sha256(path.read_bytes()).hexdigest()
|
||||
|
||||
|
||||
def _normalize_report_counts(axis_id: str, report: dict[str, Any]) -> dict[str, int]:
|
||||
"""Coerce a per-axis report to a uniform {correct,wrong,refused} dict.
|
||||
|
||||
Each axis runner emits its own dialect of metrics:
|
||||
|
||||
- G1 reports a top-level ``counts`` dict directly.
|
||||
- G2 / G4 / G5 / S1 report ``metrics={passed, wrong, cases_total, ...}``;
|
||||
``correct`` maps to ``passed`` and ``refused`` is the remainder.
|
||||
- G3 reports ``metrics={solved_correct, solved_wrong, refused_as_expected, ...}``.
|
||||
|
||||
The wrong count is the load-bearing field — the gate's invariant
|
||||
reads ``wrong`` only — but ``correct`` and ``refused`` round out
|
||||
the record so the evidence is auditable.
|
||||
"""
|
||||
if "counts" in report:
|
||||
c = report["counts"]
|
||||
return {
|
||||
"correct": int(c.get("correct", 0)),
|
||||
"wrong": int(c.get("wrong", 0)),
|
||||
"refused": int(c.get("refused", 0)),
|
||||
}
|
||||
m = report.get("metrics", {})
|
||||
if "solved_wrong" in m or "solved_correct" in m:
|
||||
return {
|
||||
"correct": int(m.get("solved_correct", 0)),
|
||||
"wrong": int(m.get("solved_wrong", 0)),
|
||||
"refused": int(m.get("refused_as_expected", 0)),
|
||||
}
|
||||
cases_total = int(m.get("cases_total", 0))
|
||||
passed = int(m.get("passed", 0))
|
||||
wrong = int(m.get("wrong", 0))
|
||||
refused = max(0, cases_total - passed - wrong)
|
||||
return {"correct": passed, "wrong": wrong, "refused": refused}
|
||||
|
||||
|
||||
def _run_capability_axes() -> dict[str, dict[str, int]]:
|
||||
"""Run every capability-axis lane; return {axis_id: counts}.
|
||||
|
||||
Each runner module exposes ``_load_cases`` and ``build_report``; we
|
||||
call them directly to avoid the report-on-disk side effect of the
|
||||
runner ``main()`` entrypoint. The capability lanes are deterministic
|
||||
against the current commit SHA.
|
||||
"""
|
||||
out: dict[str, dict[str, int]] = {}
|
||||
for axis_id, module_path in _CAPABILITY_AXIS_LANES:
|
||||
mod = importlib.import_module(module_path)
|
||||
lc_args = mod._load_cases.__code__.co_argcount
|
||||
br_args = mod.build_report.__code__.co_argcount
|
||||
cases = mod._load_cases(mod._CASES_PATH) if lc_args == 1 else mod._load_cases()
|
||||
report = mod.build_report(cases) if br_args >= 1 else mod.build_report()
|
||||
out[axis_id] = _normalize_report_counts(axis_id, report)
|
||||
return out
|
||||
|
||||
|
||||
def _run_gsm8k_train_sample() -> dict[str, int]:
|
||||
"""Run the GSM8K train-sample lane; return counts."""
|
||||
mod = importlib.import_module(_GSM8K_TRAIN_SAMPLE_MODULE)
|
||||
cases = mod._load_cases(mod._CASES_PATH)
|
||||
report = mod.build_report(cases)
|
||||
return _normalize_report_counts("gsm8k_train_sample", report)
|
||||
|
||||
|
||||
def _wrong_count_delta(
|
||||
baseline: dict[str, int], candidate: dict[str, int]
|
||||
) -> int:
|
||||
"""Positive iff the candidate increased the wrong count."""
|
||||
return int(candidate.get("wrong", 0)) - int(baseline.get("wrong", 0))
|
||||
|
||||
|
||||
def run_admissibility_replay_gate(
|
||||
spec: Any,
|
||||
*,
|
||||
active_corpus_path: Path | None = None,
|
||||
_capability_axes_runner: Any = None,
|
||||
_gsm8k_runner: Any = None,
|
||||
_cognition_runner: Any = None,
|
||||
) -> "AdmissibilityReplayEvidence":
|
||||
"""Run the Phase C admissibility gate against *spec*.
|
||||
|
||||
The gate runs three evidence lanes:
|
||||
|
||||
1. The cognition lane (inherited from
|
||||
:func:`run_replay_equivalence`).
|
||||
2. Every capability axis (G1..G5, S1) at its public v1 split.
|
||||
3. The GSM8K train_sample at v1.
|
||||
|
||||
For each lane the BASELINE run is cached in-process keyed on the
|
||||
active teaching-corpus digest. The first proposal pays the full
|
||||
baseline cost; subsequent proposals against the same corpus reuse
|
||||
it. The CANDIDATE run is computed live every time — no candidate
|
||||
caching.
|
||||
|
||||
Phase C wiring of the recognizer into the candidate-graph has not
|
||||
landed (that is Phase D / E work). Until it does, the candidate
|
||||
run produces the same counts as the baseline. The wrong-count
|
||||
invariant is therefore enforceable by simulating an elevated
|
||||
candidate count, which is how the regression test in
|
||||
``test_admissibility_replay_gate.py`` exercises this path.
|
||||
|
||||
Test hooks ``_capability_axes_runner``, ``_gsm8k_runner``, and
|
||||
``_cognition_runner`` exist for unit tests to inject baseline or
|
||||
candidate counts without re-running real eval lanes. They are
|
||||
private and not part of the public contract.
|
||||
|
||||
``replay_equivalent`` is True iff:
|
||||
- the cognition lane's ``regressed_metrics`` is empty,
|
||||
- every capability axis reports ``wrong == 0``,
|
||||
- the GSM8K train_sample's ``wrong`` count did not increase.
|
||||
"""
|
||||
capability_axes_runner = _capability_axes_runner or _run_capability_axes
|
||||
gsm8k_runner = _gsm8k_runner or _run_gsm8k_train_sample
|
||||
cognition_runner = _cognition_runner or _run_cognition_public
|
||||
|
||||
digest = _active_corpus_digest(active_corpus_path)
|
||||
cached = _BASELINE_CACHE.get(digest)
|
||||
if cached is None:
|
||||
baseline_capability = capability_axes_runner()
|
||||
baseline_gsm8k = gsm8k_runner()
|
||||
_BASELINE_CACHE[digest] = {
|
||||
"capability_axes": baseline_capability,
|
||||
"gsm8k_train_sample": baseline_gsm8k,
|
||||
}
|
||||
else:
|
||||
baseline_capability = cached["capability_axes"]
|
||||
baseline_gsm8k = cached["gsm8k_train_sample"]
|
||||
|
||||
# Cognition lane runs live (its baseline is cheap and its caches
|
||||
# are managed by chat.teaching_grounding).
|
||||
_tg.clear_teaching_caches()
|
||||
cognition_baseline = cognition_runner()
|
||||
|
||||
# Candidate runs. Phase C ships no candidate-graph wiring, so
|
||||
# the live candidate run produces baseline-equivalent counts.
|
||||
candidate_capability = capability_axes_runner()
|
||||
candidate_gsm8k = gsm8k_runner()
|
||||
cognition_candidate = cognition_runner()
|
||||
|
||||
# Cognition regression detection — same logic as run_replay_equivalence.
|
||||
regressed: list[str] = []
|
||||
for metric in _WATCHED_METRICS.metrics:
|
||||
b = cognition_baseline.get(metric)
|
||||
c = cognition_candidate.get(metric)
|
||||
if b is None or c is None:
|
||||
continue
|
||||
if c < b:
|
||||
regressed.append(metric)
|
||||
|
||||
# Wrong-count invariant on GSM8K train_sample.
|
||||
wrong_delta = _wrong_count_delta(baseline_gsm8k, candidate_gsm8k)
|
||||
if wrong_delta > 0:
|
||||
regressed.append("gsm8k_train_sample_wrong_count")
|
||||
|
||||
# Capability-axis wrong floor. Any axis whose candidate wrong>0
|
||||
# is a regression. G3 numerics already carries 6 expected-refusal
|
||||
# cases that count as "correct" in the runner's verdict map, so
|
||||
# this guard reads the wrong count directly.
|
||||
capability_wrong_axes: list[str] = []
|
||||
for axis_id, counts in candidate_capability.items():
|
||||
if counts["wrong"] > 0:
|
||||
capability_wrong_axes.append(axis_id)
|
||||
if capability_wrong_axes:
|
||||
for axis_id in capability_wrong_axes:
|
||||
regressed.append(f"capability_axis_wrong:{axis_id}")
|
||||
|
||||
return AdmissibilityReplayEvidence(
|
||||
baseline=cognition_baseline,
|
||||
candidate=cognition_candidate,
|
||||
regressed_metrics=tuple(sorted(set(regressed))),
|
||||
replay_equivalent=not regressed,
|
||||
capability_axes=candidate_capability,
|
||||
gsm8k_train_sample=candidate_gsm8k,
|
||||
wrong_count_delta=wrong_delta,
|
||||
)
|
||||
|
||||
|
||||
__all__ = [
|
||||
"AdmissibilityReplayEvidence",
|
||||
"run_admissibility_replay_gate",
|
||||
"run_replay_equivalence",
|
||||
]
|
||||
|
|
|
|||
|
|
@ -22,6 +22,8 @@ Consumers must branch on :attr:`ProposalSource.kind` using exhaustive
|
|||
...
|
||||
case "contemplation":
|
||||
...
|
||||
case "exemplar_corpus":
|
||||
...
|
||||
case _: # pragma: no cover - exhaustiveness
|
||||
assert_never(proposal.source.kind)
|
||||
"""
|
||||
|
|
@ -32,7 +34,9 @@ from dataclasses import dataclass
|
|||
from typing import Any, Literal, Mapping, get_args
|
||||
|
||||
|
||||
ProposalKind = Literal["operator", "miner", "curriculum", "contemplation"]
|
||||
ProposalKind = Literal[
|
||||
"operator", "miner", "curriculum", "contemplation", "exemplar_corpus"
|
||||
]
|
||||
ALLOWED_KINDS: frozenset[str] = frozenset(get_args(ProposalKind))
|
||||
|
||||
|
||||
|
|
|
|||
260
tests/test_admissibility_replay_gate.py
Normal file
260
tests/test_admissibility_replay_gate.py
Normal file
|
|
@ -0,0 +1,260 @@
|
|||
"""ADR-0163 Phase C — admissibility replay-gate tests.
|
||||
|
||||
Pins:
|
||||
- the helper runs the cognition + capability-axis + GSM8K train_sample lanes
|
||||
- baseline cache hit: a second call against the same corpus digest does NOT
|
||||
re-run the baselines
|
||||
- cache invalidation: changing the corpus digest re-runs baselines
|
||||
- WRONG-COUNT INVARIANT: a candidate run that lifts the GSM8K train_sample
|
||||
wrong count is rejected with the typed regressed_metrics entry
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from teaching.exemplar_ingest import load_exemplar_corpus
|
||||
from teaching.recognizer_synthesis import synthesize_recognizer
|
||||
import teaching.replay as replay_mod
|
||||
from teaching.replay import (
|
||||
AdmissibilityReplayEvidence,
|
||||
run_admissibility_replay_gate,
|
||||
)
|
||||
|
||||
|
||||
_REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
_EXEMPLAR = (
|
||||
_REPO_ROOT / "teaching" / "admissibility_exemplars" / "rate_with_currency_v1.jsonl"
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _clean_baseline_cache() -> Any:
|
||||
"""Each test starts with a clean baseline cache."""
|
||||
replay_mod._BASELINE_CACHE.clear()
|
||||
yield
|
||||
replay_mod._BASELINE_CACHE.clear()
|
||||
|
||||
|
||||
def _stub_capability_axes() -> dict[str, dict[str, int]]:
|
||||
return {
|
||||
"G1_verb_classes": {"correct": 20, "wrong": 0, "refused": 0},
|
||||
"G2_comparatives": {"correct": 29, "wrong": 0, "refused": 0},
|
||||
"G3_numerics": {"correct": 20, "wrong": 0, "refused": 6},
|
||||
"G4_multi_clause": {"correct": 32, "wrong": 0, "refused": 0},
|
||||
"G5_aggregate": {"correct": 20, "wrong": 0, "refused": 0},
|
||||
"S1_rate_events": {"correct": 20, "wrong": 0, "refused": 0},
|
||||
}
|
||||
|
||||
|
||||
def _stub_gsm8k() -> dict[str, int]:
|
||||
return {"correct": 3, "wrong": 0, "refused": 47}
|
||||
|
||||
|
||||
def _stub_cognition() -> dict[str, float]:
|
||||
return {
|
||||
"intent_accuracy": 1.0,
|
||||
"surface_groundedness": 1.0,
|
||||
"term_capture_rate": 1.0,
|
||||
"versor_closure_rate": 1.0,
|
||||
}
|
||||
|
||||
|
||||
def _spec() -> Any:
|
||||
return synthesize_recognizer(load_exemplar_corpus(_EXEMPLAR))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Happy path: every lane wrong=0 → replay_equivalent=True
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_gate_passes_when_no_lane_regresses() -> None:
|
||||
ev = run_admissibility_replay_gate(
|
||||
_spec(),
|
||||
_capability_axes_runner=_stub_capability_axes,
|
||||
_gsm8k_runner=_stub_gsm8k,
|
||||
_cognition_runner=_stub_cognition,
|
||||
)
|
||||
assert isinstance(ev, AdmissibilityReplayEvidence)
|
||||
assert ev.replay_equivalent is True
|
||||
assert ev.regressed_metrics == ()
|
||||
assert ev.wrong_count_delta == 0
|
||||
assert ev.capability_axes["G1_verb_classes"]["wrong"] == 0
|
||||
assert ev.gsm8k_train_sample == {"correct": 3, "wrong": 0, "refused": 47}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Cache hit + invalidation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_baseline_cache_hit_on_second_call(tmp_path: Path) -> None:
|
||||
"""Second call with the same active corpus digest reuses baselines."""
|
||||
active = tmp_path / "active_corpus.jsonl"
|
||||
active.write_text("{}\n", encoding="utf-8")
|
||||
|
||||
cap_calls: list[int] = []
|
||||
gsm_calls: list[int] = []
|
||||
|
||||
def _cap() -> dict[str, dict[str, int]]:
|
||||
cap_calls.append(1)
|
||||
return _stub_capability_axes()
|
||||
|
||||
def _gsm() -> dict[str, int]:
|
||||
gsm_calls.append(1)
|
||||
return _stub_gsm8k()
|
||||
|
||||
run_admissibility_replay_gate(
|
||||
_spec(),
|
||||
active_corpus_path=active,
|
||||
_capability_axes_runner=_cap,
|
||||
_gsm8k_runner=_gsm,
|
||||
_cognition_runner=_stub_cognition,
|
||||
)
|
||||
first_cap = len(cap_calls)
|
||||
first_gsm = len(gsm_calls)
|
||||
# Each first call runs the baseline AND the candidate -> 2 runs each.
|
||||
assert first_cap >= 2 and first_gsm >= 2
|
||||
|
||||
run_admissibility_replay_gate(
|
||||
_spec(),
|
||||
active_corpus_path=active,
|
||||
_capability_axes_runner=_cap,
|
||||
_gsm8k_runner=_gsm,
|
||||
_cognition_runner=_stub_cognition,
|
||||
)
|
||||
# On the second call only the CANDIDATE run fires; baseline is cached.
|
||||
assert len(cap_calls) == first_cap + 1
|
||||
assert len(gsm_calls) == first_gsm + 1
|
||||
|
||||
|
||||
def test_baseline_cache_invalidates_on_corpus_change(tmp_path: Path) -> None:
|
||||
active_a = tmp_path / "corpus_a.jsonl"
|
||||
active_a.write_text("a\n", encoding="utf-8")
|
||||
active_b = tmp_path / "corpus_b.jsonl"
|
||||
active_b.write_text("b\n", encoding="utf-8")
|
||||
|
||||
cap_calls: list[int] = []
|
||||
gsm_calls: list[int] = []
|
||||
|
||||
def _cap() -> dict[str, dict[str, int]]:
|
||||
cap_calls.append(1)
|
||||
return _stub_capability_axes()
|
||||
|
||||
def _gsm() -> dict[str, int]:
|
||||
gsm_calls.append(1)
|
||||
return _stub_gsm8k()
|
||||
|
||||
run_admissibility_replay_gate(
|
||||
_spec(),
|
||||
active_corpus_path=active_a,
|
||||
_capability_axes_runner=_cap,
|
||||
_gsm8k_runner=_gsm,
|
||||
_cognition_runner=_stub_cognition,
|
||||
)
|
||||
a_cap, a_gsm = len(cap_calls), len(gsm_calls)
|
||||
# Different corpus digest -> baseline re-runs.
|
||||
run_admissibility_replay_gate(
|
||||
_spec(),
|
||||
active_corpus_path=active_b,
|
||||
_capability_axes_runner=_cap,
|
||||
_gsm8k_runner=_gsm,
|
||||
_cognition_runner=_stub_cognition,
|
||||
)
|
||||
# The second call runs baseline + candidate again because the cache
|
||||
# was invalidated by the digest change.
|
||||
assert len(cap_calls) >= a_cap + 2
|
||||
assert len(gsm_calls) >= a_gsm + 2
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# WRONG-COUNT INVARIANT (the load-bearing test for ADR-0163 §Constraint #1)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_wrong_count_invariant_auto_rejects_gsm8k_regression() -> None:
|
||||
"""If the candidate lifts the GSM8K wrong count by ≥ 1, the gate
|
||||
rejects with the typed regressed_metrics entry — Phase D / E's
|
||||
wiring never reaches the operator review."""
|
||||
|
||||
baseline_gsm = {"correct": 3, "wrong": 0, "refused": 47}
|
||||
candidate_gsm = {"correct": 3, "wrong": 1, "refused": 46}
|
||||
|
||||
# Pre-populate the baseline cache so the runner returns the
|
||||
# candidate's elevated counts. This mirrors a Phase D wiring that
|
||||
# mis-admits one previously-refused case as a wrong answer.
|
||||
call_count = {"n": 0}
|
||||
|
||||
def _alternating_gsm() -> dict[str, int]:
|
||||
# First call: baseline. Second call (live candidate): elevated.
|
||||
call_count["n"] += 1
|
||||
return baseline_gsm if call_count["n"] == 1 else candidate_gsm
|
||||
|
||||
ev = run_admissibility_replay_gate(
|
||||
_spec(),
|
||||
_capability_axes_runner=_stub_capability_axes,
|
||||
_gsm8k_runner=_alternating_gsm,
|
||||
_cognition_runner=_stub_cognition,
|
||||
)
|
||||
assert ev.replay_equivalent is False
|
||||
assert "gsm8k_train_sample_wrong_count" in ev.regressed_metrics
|
||||
assert ev.wrong_count_delta == 1
|
||||
|
||||
|
||||
def test_capability_axis_wrong_count_also_rejects() -> None:
|
||||
"""Any capability axis whose candidate wrong>0 is a regression.
|
||||
|
||||
G1..G5+S1 are wrong=0 today; a candidate that flips any to >0
|
||||
must be rejected.
|
||||
"""
|
||||
elevated = _stub_capability_axes()
|
||||
elevated["G3_numerics"] = {"correct": 19, "wrong": 1, "refused": 6}
|
||||
|
||||
call_count = {"n": 0}
|
||||
|
||||
def _alt_caps() -> dict[str, dict[str, int]]:
|
||||
call_count["n"] += 1
|
||||
return _stub_capability_axes() if call_count["n"] == 1 else elevated
|
||||
|
||||
ev = run_admissibility_replay_gate(
|
||||
_spec(),
|
||||
_capability_axes_runner=_alt_caps,
|
||||
_gsm8k_runner=_stub_gsm8k,
|
||||
_cognition_runner=_stub_cognition,
|
||||
)
|
||||
assert ev.replay_equivalent is False
|
||||
assert any(
|
||||
m.startswith("capability_axis_wrong:") for m in ev.regressed_metrics
|
||||
)
|
||||
|
||||
|
||||
def test_cognition_lane_regression_also_rejects() -> None:
|
||||
"""The cognition-lane regression detection from the older
|
||||
run_replay_equivalence path is preserved verbatim."""
|
||||
|
||||
baseline = {
|
||||
"intent_accuracy": 1.0,
|
||||
"surface_groundedness": 1.0,
|
||||
"term_capture_rate": 1.0,
|
||||
"versor_closure_rate": 1.0,
|
||||
}
|
||||
candidate = {**baseline, "intent_accuracy": 0.9}
|
||||
|
||||
call_count = {"n": 0}
|
||||
|
||||
def _alt_cog() -> dict[str, float]:
|
||||
call_count["n"] += 1
|
||||
return baseline if call_count["n"] == 1 else candidate
|
||||
|
||||
ev = run_admissibility_replay_gate(
|
||||
_spec(),
|
||||
_capability_axes_runner=_stub_capability_axes,
|
||||
_gsm8k_runner=_stub_gsm8k,
|
||||
_cognition_runner=_alt_cog,
|
||||
)
|
||||
assert ev.replay_equivalent is False
|
||||
assert "intent_accuracy" in ev.regressed_metrics
|
||||
211
tests/test_exemplar_ingest.py
Normal file
211
tests/test_exemplar_ingest.py
Normal file
|
|
@ -0,0 +1,211 @@
|
|||
"""ADR-0163 Phase C — exemplar_ingest tests.
|
||||
|
||||
Pins:
|
||||
- load_exemplar_corpus parses each Phase B JSONL without loss
|
||||
- corpus_digest is byte-stable across runs
|
||||
- malformed exemplars raise ExemplarIngestError
|
||||
- the module performs no I/O beyond the supplied path
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import builtins
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from evals.refusal_taxonomy.shape_categories import ShapeCategory
|
||||
from teaching.exemplar_ingest import (
|
||||
Exemplar,
|
||||
ExemplarCorpus,
|
||||
ExemplarIngestError,
|
||||
list_corpora,
|
||||
load_exemplar_corpus,
|
||||
)
|
||||
|
||||
|
||||
_REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
_EXEMPLARS_ROOT = _REPO_ROOT / "teaching" / "admissibility_exemplars"
|
||||
_ROUND_1 = (
|
||||
("descriptive_setup_no_quantity_v1.jsonl", ShapeCategory.DESCRIPTIVE_SETUP_NO_QUANTITY),
|
||||
("temporal_aggregation_v1.jsonl", ShapeCategory.TEMPORAL_AGGREGATION),
|
||||
("rate_with_currency_v1.jsonl", ShapeCategory.RATE_WITH_CURRENCY),
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("filename", "category"), _ROUND_1)
|
||||
def test_loads_phase_b_corpus_without_loss(filename: str, category: ShapeCategory) -> None:
|
||||
path = _EXEMPLARS_ROOT / filename
|
||||
corpus = load_exemplar_corpus(path)
|
||||
assert isinstance(corpus, ExemplarCorpus)
|
||||
assert corpus.shape_category is category
|
||||
assert corpus.path == path
|
||||
assert len(corpus.exemplars) == 20
|
||||
# Every exemplar carries the supported category.
|
||||
for ex in corpus.exemplars:
|
||||
assert isinstance(ex, Exemplar)
|
||||
assert ex.shape_category is category
|
||||
# Internal ordering matches the canonical sort by exemplar_id.
|
||||
ids = [ex.exemplar_id for ex in corpus.exemplars]
|
||||
assert ids == sorted(ids)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("filename", "_category"), _ROUND_1)
|
||||
def test_corpus_digest_is_byte_stable(filename: str, _category: ShapeCategory) -> None:
|
||||
path = _EXEMPLARS_ROOT / filename
|
||||
a = load_exemplar_corpus(path)
|
||||
b = load_exemplar_corpus(path)
|
||||
assert a.corpus_digest == b.corpus_digest
|
||||
assert len(a.corpus_digest) == 64 # sha256 hex
|
||||
|
||||
|
||||
def test_list_corpora_loads_every_round_1_file() -> None:
|
||||
corpora = list_corpora(_EXEMPLARS_ROOT)
|
||||
cats = {c.shape_category for c in corpora}
|
||||
assert cats == {cat for _, cat in _ROUND_1}
|
||||
# Stable iteration order.
|
||||
again = list_corpora(_EXEMPLARS_ROOT)
|
||||
assert [c.corpus_digest for c in corpora] == [c.corpus_digest for c in again]
|
||||
|
||||
|
||||
def test_rejects_unknown_shape_category(tmp_path: Path) -> None:
|
||||
bad = tmp_path / "uncategorized_v1.jsonl"
|
||||
bad.write_text(
|
||||
json.dumps({
|
||||
"exemplar_id": "x-0001",
|
||||
"shape_category": "uncategorized",
|
||||
"statement": "test",
|
||||
"expected_graph": {
|
||||
"subject": None,
|
||||
"quantity_anchors": [],
|
||||
"graph_intent": "setup",
|
||||
"outcome": "inadmissible_by_design",
|
||||
},
|
||||
"provenance": {
|
||||
"source": "phase_b_seed",
|
||||
"author": "test",
|
||||
"round": 1,
|
||||
"category_rank": 9,
|
||||
},
|
||||
}, separators=(",", ":")) + "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
with pytest.raises(ExemplarIngestError, match="not a Phase C round-1 category"):
|
||||
load_exemplar_corpus(bad)
|
||||
|
||||
|
||||
def test_rejects_mismatched_anchor_shape(tmp_path: Path) -> None:
|
||||
# rate_with_currency JSONL but with a missing currency_symbol.
|
||||
bad = tmp_path / "rate_with_currency_v1.jsonl"
|
||||
bad.write_text(
|
||||
json.dumps({
|
||||
"exemplar_id": "rwc-bad-0001",
|
||||
"shape_category": "rate_with_currency",
|
||||
"statement": "test",
|
||||
"expected_graph": {
|
||||
"subject": "x",
|
||||
"quantity_anchors": [
|
||||
{
|
||||
"kind": "currency_per_unit_rate",
|
||||
# currency_symbol intentionally missing
|
||||
"amount": "10",
|
||||
"amount_kind": "integer",
|
||||
"per_unit": "hour",
|
||||
"subject_role": "x",
|
||||
},
|
||||
],
|
||||
"graph_intent": "rate",
|
||||
"outcome": "admissible",
|
||||
},
|
||||
"provenance": {
|
||||
"source": "phase_b_seed",
|
||||
"author": "test",
|
||||
"round": 1,
|
||||
"category_rank": 3,
|
||||
},
|
||||
}, separators=(",", ":")) + "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
with pytest.raises(ExemplarIngestError, match="missing required keys"):
|
||||
load_exemplar_corpus(bad)
|
||||
|
||||
|
||||
def test_rejects_file_name_category_mismatch(tmp_path: Path) -> None:
|
||||
# Stem says temporal_aggregation_v1 but record says rate_with_currency.
|
||||
bad = tmp_path / "temporal_aggregation_v1.jsonl"
|
||||
bad.write_text(
|
||||
json.dumps({
|
||||
"exemplar_id": "rwc-mismatch-0001",
|
||||
"shape_category": "rate_with_currency",
|
||||
"statement": "test",
|
||||
"expected_graph": {
|
||||
"subject": "x",
|
||||
"quantity_anchors": [
|
||||
{
|
||||
"kind": "currency_per_unit_rate",
|
||||
"currency_symbol": "$",
|
||||
"amount": "10",
|
||||
"amount_kind": "integer",
|
||||
"per_unit": "hour",
|
||||
"subject_role": "x",
|
||||
},
|
||||
],
|
||||
"graph_intent": "rate",
|
||||
"outcome": "admissible",
|
||||
},
|
||||
"provenance": {
|
||||
"source": "phase_b_seed",
|
||||
"author": "test",
|
||||
"round": 1,
|
||||
"category_rank": 3,
|
||||
},
|
||||
}, separators=(",", ":")) + "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
with pytest.raises(ExemplarIngestError, match="does not match category"):
|
||||
load_exemplar_corpus(bad)
|
||||
|
||||
|
||||
def test_load_reads_only_supplied_path(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""The ingest module is pure — only the supplied path is opened.
|
||||
|
||||
Wrap ``builtins.open`` to record every absolute path opened during
|
||||
a load. Only the supplied JSONL may appear (the module reads no
|
||||
config, no caches, no sibling files).
|
||||
"""
|
||||
real_open = builtins.open
|
||||
opened: list[str] = []
|
||||
|
||||
def _tracking_open(file: Any, *args: Any, **kwargs: Any) -> Any:
|
||||
opened.append(str(file))
|
||||
return real_open(file, *args, **kwargs)
|
||||
|
||||
monkeypatch.setattr(builtins, "open", _tracking_open)
|
||||
target = _EXEMPLARS_ROOT / "rate_with_currency_v1.jsonl"
|
||||
# Read_text() bypasses builtins.open in CPython 3.13, so the tracker
|
||||
# may legitimately catch nothing. The load completes; assert the
|
||||
# only paths that DID surface (if any) are the target itself.
|
||||
load_exemplar_corpus(target)
|
||||
for path in opened:
|
||||
# Allow read of the target; nothing else.
|
||||
assert str(target) in path or path.endswith(".jsonl"), (
|
||||
f"unexpected file opened during ingest: {path}"
|
||||
)
|
||||
|
||||
|
||||
def test_module_imports_no_llm_or_ml() -> None:
|
||||
"""Phase C synthesis is rules-only. No transformer / embedding / ML dep."""
|
||||
import teaching.exemplar_ingest as m
|
||||
module_file = m.__file__
|
||||
assert module_file is not None
|
||||
src = Path(module_file).read_text(encoding="utf-8")
|
||||
for forbidden in (
|
||||
"transformers", "torch", "tensorflow", "openai",
|
||||
"anthropic", "sklearn", "numpy.random",
|
||||
# No "import nltk" etc.
|
||||
):
|
||||
assert forbidden not in src, (
|
||||
f"forbidden import {forbidden!r} in exemplar_ingest.py"
|
||||
)
|
||||
|
|
@ -139,6 +139,8 @@ class TestExhaustiveMatchPattern:
|
|||
return f"c:{src.source_id}"
|
||||
case "contemplation":
|
||||
return f"q:{src.source_id}"
|
||||
case "exemplar_corpus":
|
||||
return f"e:{src.source_id}"
|
||||
case _: # pragma: no cover - exhaustiveness
|
||||
assert_never(src.kind)
|
||||
|
||||
|
|
@ -161,9 +163,20 @@ class TestExhaustiveMatchPattern:
|
|||
)
|
||||
assert self._describe(src) == "q:frontier_compare"
|
||||
|
||||
def test_kinds_sealed_at_four(self) -> None:
|
||||
def test_covers_exemplar_corpus(self) -> None:
|
||||
src = ProposalSource(
|
||||
kind="exemplar_corpus",
|
||||
source_id="rate_with_currency_v1_digest",
|
||||
emitted_at_revision="x",
|
||||
)
|
||||
assert self._describe(src) == "e:rate_with_currency_v1_digest"
|
||||
|
||||
def test_kinds_sealed_at_five(self) -> None:
|
||||
# ADR-0163.C widened the sealed set with "exemplar_corpus" so
|
||||
# Phase C admissibility proposals carry typed provenance distinct
|
||||
# from autonomous contemplation.
|
||||
assert ALLOWED_KINDS == frozenset(
|
||||
{"operator", "miner", "curriculum", "contemplation"}
|
||||
{"operator", "miner", "curriculum", "contemplation", "exemplar_corpus"}
|
||||
)
|
||||
|
||||
|
||||
|
|
|
|||
217
tests/test_propose_from_exemplars_cli.py
Normal file
217
tests/test_propose_from_exemplars_cli.py
Normal file
|
|
@ -0,0 +1,217 @@
|
|||
"""ADR-0163 Phase C — propose-from-exemplars CLI tests.
|
||||
|
||||
Pins:
|
||||
- the CLI loads a real Phase B JSONL and produces a pending proposal
|
||||
- --all produces three pending proposals (one per Phase B corpus)
|
||||
- proposal_id is deterministic across runs with the same corpus_digest
|
||||
- the CLI does NOT mutate any corpus, pack, recognizer registry, or
|
||||
eval lane file outside the supplied tmp paths
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
import teaching.replay as replay_mod
|
||||
from teaching.proposals import ProposalLog
|
||||
|
||||
|
||||
_REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
_EXEMPLARS = _REPO_ROOT / "teaching" / "admissibility_exemplars"
|
||||
_ACTIVE_CORPUS = (
|
||||
_REPO_ROOT / "teaching" / "cognition_chains" / "cognition_chains_v1.jsonl"
|
||||
)
|
||||
_GSM8K_TRAIN_REPORT = (
|
||||
_REPO_ROOT / "evals" / "gsm8k_math" / "train_sample" / "v1" / "report.json"
|
||||
)
|
||||
|
||||
|
||||
def _stub_capability_axes() -> dict[str, dict[str, int]]:
|
||||
return {
|
||||
"G1_verb_classes": {"correct": 20, "wrong": 0, "refused": 0},
|
||||
"G2_comparatives": {"correct": 29, "wrong": 0, "refused": 0},
|
||||
"G3_numerics": {"correct": 20, "wrong": 0, "refused": 6},
|
||||
"G4_multi_clause": {"correct": 32, "wrong": 0, "refused": 0},
|
||||
"G5_aggregate": {"correct": 20, "wrong": 0, "refused": 0},
|
||||
"S1_rate_events": {"correct": 20, "wrong": 0, "refused": 0},
|
||||
}
|
||||
|
||||
|
||||
def _stub_gsm8k() -> dict[str, int]:
|
||||
return {"correct": 3, "wrong": 0, "refused": 47}
|
||||
|
||||
|
||||
def _stub_cognition() -> dict[str, float]:
|
||||
return {
|
||||
"intent_accuracy": 1.0,
|
||||
"surface_groundedness": 1.0,
|
||||
"term_capture_rate": 1.0,
|
||||
"versor_closure_rate": 1.0,
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _stub_eval_lanes(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Stub the heavy eval lanes so CLI tests run in milliseconds.
|
||||
|
||||
The CLI invokes :func:`teaching.replay.run_admissibility_replay_gate`
|
||||
via the existing :func:`propose_from_candidate` path. Substituting
|
||||
the lane runners at module scope is enough; the gate calls them by
|
||||
name (``_run_capability_axes``, ``_run_gsm8k_train_sample``,
|
||||
``_run_cognition_public``).
|
||||
"""
|
||||
replay_mod._BASELINE_CACHE.clear()
|
||||
monkeypatch.setattr(replay_mod, "_run_capability_axes", _stub_capability_axes)
|
||||
monkeypatch.setattr(replay_mod, "_run_gsm8k_train_sample", _stub_gsm8k)
|
||||
monkeypatch.setattr(replay_mod, "_run_cognition_public", _stub_cognition)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# In-process CLI invocation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _invoke_cli(args: list[str]) -> tuple[int, str, str]:
|
||||
"""Run the CLI in-process by calling ``core.cli.main``.
|
||||
|
||||
Captures argv + stdout/stderr; returns (exit_code, stdout, stderr).
|
||||
"""
|
||||
import io
|
||||
import contextlib
|
||||
|
||||
from core import cli as core_cli
|
||||
|
||||
saved_argv = sys.argv
|
||||
saved_stdout = sys.stdout
|
||||
saved_stderr = sys.stderr
|
||||
out_buf = io.StringIO()
|
||||
err_buf = io.StringIO()
|
||||
try:
|
||||
sys.argv = ["core", *args]
|
||||
with (
|
||||
contextlib.redirect_stdout(out_buf),
|
||||
contextlib.redirect_stderr(err_buf),
|
||||
):
|
||||
try:
|
||||
code = core_cli.main()
|
||||
except SystemExit as exc:
|
||||
code = int(exc.code) if exc.code is not None else 0
|
||||
finally:
|
||||
sys.argv = saved_argv
|
||||
sys.stdout = saved_stdout
|
||||
sys.stderr = saved_stderr
|
||||
return code, out_buf.getvalue(), err_buf.getvalue()
|
||||
|
||||
|
||||
def test_cli_single_corpus_produces_pending_proposal(tmp_path: Path) -> None:
|
||||
log_path = tmp_path / "proposals.jsonl"
|
||||
code, out, _ = _invoke_cli([
|
||||
"teaching", "propose-from-exemplars",
|
||||
str(_EXEMPLARS / "rate_with_currency_v1.jsonl"),
|
||||
"--log", str(log_path),
|
||||
"--json",
|
||||
])
|
||||
assert code == 0, f"CLI exited {code}; stdout={out!r}"
|
||||
payload = json.loads(out)
|
||||
assert len(payload["proposals"]) == 1
|
||||
p = payload["proposals"][0]
|
||||
assert p["shape_category"] == "rate_with_currency"
|
||||
assert p["state"] == "pending"
|
||||
assert p["replay_equivalent"] is True
|
||||
assert p["wrong_count_delta"] == 0
|
||||
# The proposal exists in the log.
|
||||
log = ProposalLog(log_path)
|
||||
rec = log.find(p["proposal_id"])
|
||||
assert rec is not None
|
||||
assert rec["state"] == "pending"
|
||||
assert rec["proposal"]["source"]["kind"] == "exemplar_corpus"
|
||||
|
||||
|
||||
def test_cli_all_flag_proposes_three_corpora(tmp_path: Path) -> None:
|
||||
log_path = tmp_path / "proposals.jsonl"
|
||||
code, out, _ = _invoke_cli([
|
||||
"teaching", "propose-from-exemplars",
|
||||
"--all",
|
||||
"--log", str(log_path),
|
||||
"--json",
|
||||
])
|
||||
assert code == 0
|
||||
payload = json.loads(out)
|
||||
cats = {p["shape_category"] for p in payload["proposals"]}
|
||||
assert cats == {
|
||||
"descriptive_setup_no_quantity",
|
||||
"rate_with_currency",
|
||||
"temporal_aggregation",
|
||||
}
|
||||
for p in payload["proposals"]:
|
||||
assert p["state"] == "pending"
|
||||
|
||||
|
||||
def test_proposal_id_is_deterministic_for_same_corpus(tmp_path: Path) -> None:
|
||||
log_a = tmp_path / "log_a.jsonl"
|
||||
log_b = tmp_path / "log_b.jsonl"
|
||||
code_a, out_a, _ = _invoke_cli([
|
||||
"teaching", "propose-from-exemplars",
|
||||
str(_EXEMPLARS / "rate_with_currency_v1.jsonl"),
|
||||
"--log", str(log_a),
|
||||
"--json",
|
||||
])
|
||||
code_b, out_b, _ = _invoke_cli([
|
||||
"teaching", "propose-from-exemplars",
|
||||
str(_EXEMPLARS / "rate_with_currency_v1.jsonl"),
|
||||
"--log", str(log_b),
|
||||
"--json",
|
||||
])
|
||||
assert code_a == code_b == 0
|
||||
pid_a = json.loads(out_a)["proposals"][0]["proposal_id"]
|
||||
pid_b = json.loads(out_b)["proposals"][0]["proposal_id"]
|
||||
assert pid_a == pid_b
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Read-only snapshot — the CLI mutates nothing outside the supplied paths
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _digest(path: Path) -> str:
|
||||
if not path.exists():
|
||||
return ""
|
||||
return hashlib.sha256(path.read_bytes()).hexdigest()
|
||||
|
||||
|
||||
def _snapshot_paths() -> list[Path]:
|
||||
"""Files the CLI MUST NOT mutate."""
|
||||
out: list[Path] = []
|
||||
out.extend(sorted(_EXEMPLARS.glob("*_v1.jsonl")))
|
||||
if _ACTIVE_CORPUS.exists():
|
||||
out.append(_ACTIVE_CORPUS)
|
||||
if _GSM8K_TRAIN_REPORT.exists():
|
||||
out.append(_GSM8K_TRAIN_REPORT)
|
||||
# Capability axis reports are touched by the runners when run via
|
||||
# write_report; the CLI gate calls build_report() directly so
|
||||
# report.json must remain byte-identical.
|
||||
for report in (_REPO_ROOT / "evals" / "math_capability_axes").rglob("v1/report.json"):
|
||||
out.append(report)
|
||||
return out
|
||||
|
||||
|
||||
def test_cli_does_not_mutate_input_files(tmp_path: Path) -> None:
|
||||
snapshot_before = {p: _digest(p) for p in _snapshot_paths()}
|
||||
log_path = tmp_path / "proposals.jsonl"
|
||||
code, _, _ = _invoke_cli([
|
||||
"teaching", "propose-from-exemplars",
|
||||
"--all",
|
||||
"--log", str(log_path),
|
||||
"--json",
|
||||
])
|
||||
assert code == 0
|
||||
snapshot_after = {p: _digest(p) for p in _snapshot_paths()}
|
||||
for path in snapshot_before:
|
||||
assert snapshot_before[path] == snapshot_after[path], (
|
||||
f"CLI mutated read-only file: {path}"
|
||||
)
|
||||
314
tests/test_recognizer_synthesis.py
Normal file
314
tests/test_recognizer_synthesis.py
Normal file
|
|
@ -0,0 +1,314 @@
|
|||
"""ADR-0163 Phase C — recognizer_synthesis tests.
|
||||
|
||||
Pins:
|
||||
- synthesize_recognizer is deterministic (same corpus -> same spec bytes)
|
||||
- synthesize_recognizer is pure (no I/O, no global state)
|
||||
- per-category canonical_pattern subsumes every seed
|
||||
- the pattern is NARROWER than a generic any-shape (an out-of-corpus
|
||||
seed must not match)
|
||||
- author_notes are honored or surfaced — never silently dropped
|
||||
- the module performs no LLM / embedding / ML import
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import builtins
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from evals.refusal_taxonomy.shape_categories import ShapeCategory
|
||||
from teaching.exemplar_ingest import (
|
||||
Exemplar,
|
||||
ExemplarCorpus,
|
||||
load_exemplar_corpus,
|
||||
)
|
||||
from teaching.recognizer_synthesis import (
|
||||
RecognizerSpec,
|
||||
synthesize_recognizer,
|
||||
)
|
||||
|
||||
|
||||
_REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
_EXEMPLARS_ROOT = _REPO_ROOT / "teaching" / "admissibility_exemplars"
|
||||
_ROUND_1: tuple[tuple[str, ShapeCategory], ...] = (
|
||||
("descriptive_setup_no_quantity_v1.jsonl", ShapeCategory.DESCRIPTIVE_SETUP_NO_QUANTITY),
|
||||
("temporal_aggregation_v1.jsonl", ShapeCategory.TEMPORAL_AGGREGATION),
|
||||
("rate_with_currency_v1.jsonl", ShapeCategory.RATE_WITH_CURRENCY),
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def corpora() -> dict[ShapeCategory, ExemplarCorpus]:
|
||||
out: dict[ShapeCategory, ExemplarCorpus] = {}
|
||||
for filename, cat in _ROUND_1:
|
||||
out[cat] = load_exemplar_corpus(_EXEMPLARS_ROOT / filename)
|
||||
return out
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Determinism + purity
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("_filename", "category"), _ROUND_1)
|
||||
def test_synthesis_is_deterministic(
|
||||
_filename: str,
|
||||
category: ShapeCategory,
|
||||
corpora: dict[ShapeCategory, ExemplarCorpus],
|
||||
) -> None:
|
||||
corpus = corpora[category]
|
||||
a = synthesize_recognizer(corpus)
|
||||
b = synthesize_recognizer(corpus)
|
||||
assert a.canonical_bytes() == b.canonical_bytes()
|
||||
assert a.spec_digest() == b.spec_digest()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("_filename", "category"), _ROUND_1)
|
||||
def test_synthesis_is_pure_no_io(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
_filename: str,
|
||||
category: ShapeCategory,
|
||||
corpora: dict[ShapeCategory, ExemplarCorpus],
|
||||
) -> None:
|
||||
corpus = corpora[category]
|
||||
real_open = builtins.open
|
||||
|
||||
def _no_open(*args: Any, **kwargs: Any) -> Any:
|
||||
raise AssertionError(
|
||||
f"synthesize_recognizer opened a file: args={args}"
|
||||
)
|
||||
|
||||
monkeypatch.setattr(builtins, "open", _no_open)
|
||||
try:
|
||||
spec = synthesize_recognizer(corpus)
|
||||
finally:
|
||||
monkeypatch.setattr(builtins, "open", real_open)
|
||||
assert isinstance(spec, RecognizerSpec)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Subsumption + narrowness
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _matches(spec: RecognizerSpec, ex: Exemplar) -> bool:
|
||||
"""Mechanical predicate: does *spec* subsume *ex*?
|
||||
|
||||
The recognizer's canonical_pattern is bespoke per category, so the
|
||||
matcher is bespoke too. Each branch checks every axis the spec
|
||||
constrains. Used only in tests to assert (a) every seed matches
|
||||
and (b) an out-of-corpus seed does not.
|
||||
"""
|
||||
p = spec.canonical_pattern
|
||||
graph = ex.expected_graph
|
||||
if spec.shape_category is ShapeCategory.DESCRIPTIVE_SETUP_NO_QUANTITY:
|
||||
return (
|
||||
graph["graph_intent"] == p["graph_intent"]
|
||||
and graph["outcome"] == p["outcome"]
|
||||
and len(graph["quantity_anchors"]) == p["quantity_anchor_count"]
|
||||
)
|
||||
if spec.shape_category is ShapeCategory.TEMPORAL_AGGREGATION:
|
||||
if graph["graph_intent"] != p["graph_intent"]:
|
||||
return False
|
||||
if graph["outcome"] != p["outcome"]:
|
||||
return False
|
||||
anchors = graph["quantity_anchors"]
|
||||
if not (p["anchor_count_min"] <= len(anchors) <= p["anchor_count_max"]):
|
||||
return False
|
||||
observed_units = set(p["observed_window_units"])
|
||||
observed_quants = set(p["observed_window_quantifiers"])
|
||||
for a in anchors:
|
||||
if a["kind"] != p["anchor_kind"]:
|
||||
return False
|
||||
if a["window_unit"] not in observed_units:
|
||||
return False
|
||||
if a["window_quantifier"] not in observed_quants:
|
||||
return False
|
||||
return True
|
||||
if spec.shape_category is ShapeCategory.RATE_WITH_CURRENCY:
|
||||
if graph["graph_intent"] != p["graph_intent"]:
|
||||
return False
|
||||
if graph["outcome"] != p["outcome"]:
|
||||
return False
|
||||
anchors = graph["quantity_anchors"]
|
||||
if not (p["anchor_count_min"] <= len(anchors) <= p["anchor_count_max"]):
|
||||
return False
|
||||
observed_curr = set(p["observed_currency_symbols"])
|
||||
observed_pu = set(p["observed_per_units"])
|
||||
observed_ak = set(p["observed_amount_kinds"])
|
||||
for a in anchors:
|
||||
if a["kind"] != p["anchor_kind"]:
|
||||
return False
|
||||
if a["currency_symbol"] not in observed_curr:
|
||||
return False
|
||||
if a["per_unit"] not in observed_pu:
|
||||
return False
|
||||
if a["amount_kind"] not in observed_ak:
|
||||
return False
|
||||
return True
|
||||
raise AssertionError(f"no matcher for {spec.shape_category!r}")
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("_filename", "category"), _ROUND_1)
|
||||
def test_canonical_pattern_subsumes_every_seed(
|
||||
_filename: str,
|
||||
category: ShapeCategory,
|
||||
corpora: dict[ShapeCategory, ExemplarCorpus],
|
||||
) -> None:
|
||||
corpus = corpora[category]
|
||||
spec = synthesize_recognizer(corpus)
|
||||
for ex in corpus.exemplars:
|
||||
assert _matches(spec, ex), (
|
||||
f"{ex.exemplar_id}: synthesized spec does NOT subsume its own seed"
|
||||
)
|
||||
|
||||
|
||||
def _ex(category: ShapeCategory, graph: dict[str, Any]) -> Exemplar:
|
||||
return Exemplar(
|
||||
exemplar_id="out-of-corpus-0001",
|
||||
shape_category=category,
|
||||
statement="test",
|
||||
expected_graph=graph,
|
||||
provenance={"source": "phase_b_seed", "author": "test", "round": 1, "category_rank": 0},
|
||||
)
|
||||
|
||||
|
||||
def test_descriptive_pattern_rejects_seed_with_anchor(
|
||||
corpora: dict[ShapeCategory, ExemplarCorpus],
|
||||
) -> None:
|
||||
"""A descriptive-setup recognizer must not match a statement carrying
|
||||
an anchor — that would mean admitting quantitative shapes as setup."""
|
||||
spec = synthesize_recognizer(corpora[ShapeCategory.DESCRIPTIVE_SETUP_NO_QUANTITY])
|
||||
fake = _ex(
|
||||
ShapeCategory.DESCRIPTIVE_SETUP_NO_QUANTITY,
|
||||
{
|
||||
"subject": "x",
|
||||
"quantity_anchors": [
|
||||
{
|
||||
"kind": "currency_per_unit_rate",
|
||||
"currency_symbol": "$",
|
||||
"amount": "1",
|
||||
"amount_kind": "integer",
|
||||
"per_unit": "hour",
|
||||
"subject_role": "x",
|
||||
},
|
||||
],
|
||||
"graph_intent": "setup",
|
||||
"outcome": "inadmissible_by_design",
|
||||
},
|
||||
)
|
||||
assert not _matches(spec, fake)
|
||||
|
||||
|
||||
def test_temporal_pattern_rejects_unseen_window_unit(
|
||||
corpora: dict[ShapeCategory, ExemplarCorpus],
|
||||
) -> None:
|
||||
"""If the seeds never carry a millisecond window, the recognizer
|
||||
must not generalize to it. Phase D's review can widen; synthesis
|
||||
does not."""
|
||||
spec = synthesize_recognizer(corpora[ShapeCategory.TEMPORAL_AGGREGATION])
|
||||
observed_units = set(spec.canonical_pattern["observed_window_units"])
|
||||
# Find any window unit NOT in the observed set. The Phase B
|
||||
# vocabulary covers second..year, but seeds may use a subset.
|
||||
all_units = {"day", "week", "month", "year", "hour", "minute", "second"}
|
||||
unseen = all_units - observed_units
|
||||
assert unseen, "no unseen window unit available — corpus covers vocabulary"
|
||||
fake_unit = sorted(unseen)[0]
|
||||
fake = _ex(
|
||||
ShapeCategory.TEMPORAL_AGGREGATION,
|
||||
{
|
||||
"subject": "x",
|
||||
"quantity_anchors": [
|
||||
{
|
||||
"kind": "event_count_per_window",
|
||||
"count_token": "1",
|
||||
"window_unit": fake_unit,
|
||||
"window_quantifier": "each",
|
||||
"subject_role": "x",
|
||||
},
|
||||
],
|
||||
"graph_intent": "aggregate",
|
||||
"outcome": "admissible",
|
||||
},
|
||||
)
|
||||
assert not _matches(spec, fake), (
|
||||
f"recognizer wrongly generalized to unseen window_unit={fake_unit!r}"
|
||||
)
|
||||
|
||||
|
||||
def test_rate_pattern_rejects_unseen_currency(
|
||||
corpora: dict[ShapeCategory, ExemplarCorpus],
|
||||
) -> None:
|
||||
"""Same narrowness rule for currencies: the seeds cite a subset of
|
||||
{$, £, €, ¥}. Currencies outside that subset must not match."""
|
||||
spec = synthesize_recognizer(corpora[ShapeCategory.RATE_WITH_CURRENCY])
|
||||
observed = set(spec.canonical_pattern["observed_currency_symbols"])
|
||||
all_sym = {"$", "£", "€", "¥"}
|
||||
unseen = all_sym - observed
|
||||
if not unseen:
|
||||
# Every currency in the vocabulary appeared. Fall back to a
|
||||
# synthetic currency not in the vocabulary at all.
|
||||
fake_sym = "₿" # bitcoin sign — not in _VALID_CURRENCY_SYMBOLS
|
||||
else:
|
||||
fake_sym = sorted(unseen)[0]
|
||||
fake = _ex(
|
||||
ShapeCategory.RATE_WITH_CURRENCY,
|
||||
{
|
||||
"subject": "x",
|
||||
"quantity_anchors": [
|
||||
{
|
||||
"kind": "currency_per_unit_rate",
|
||||
"currency_symbol": fake_sym,
|
||||
"amount": "10",
|
||||
"amount_kind": "integer",
|
||||
"per_unit": list(spec.canonical_pattern["observed_per_units"])[0],
|
||||
"subject_role": "x",
|
||||
},
|
||||
],
|
||||
"graph_intent": "rate",
|
||||
"outcome": "admissible",
|
||||
},
|
||||
)
|
||||
assert not _matches(spec, fake), (
|
||||
f"recognizer wrongly generalized to unseen currency={fake_sym!r}"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Author_notes are honored or surfaced — never silently dropped
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("_filename", "category"), _ROUND_1)
|
||||
def test_author_notes_surface_in_unresolved_notes(
|
||||
_filename: str,
|
||||
category: ShapeCategory,
|
||||
corpora: dict[ShapeCategory, ExemplarCorpus],
|
||||
) -> None:
|
||||
corpus = corpora[category]
|
||||
spec = synthesize_recognizer(corpus)
|
||||
unresolved = set(spec.canonical_pattern["unresolved_notes"])
|
||||
for ex in corpus.exemplars:
|
||||
note = ex.author_note
|
||||
if not note:
|
||||
continue
|
||||
assert note in unresolved, (
|
||||
f"{ex.exemplar_id}: author_note silently dropped: {note!r}"
|
||||
)
|
||||
|
||||
|
||||
def test_module_imports_no_llm_or_ml() -> None:
|
||||
"""Phase C synthesis is rules-only. No transformer / embedding."""
|
||||
import teaching.recognizer_synthesis as m
|
||||
module_file = m.__file__
|
||||
assert module_file is not None
|
||||
src = Path(module_file).read_text(encoding="utf-8")
|
||||
for forbidden in (
|
||||
"transformers", "torch", "tensorflow", "openai",
|
||||
"anthropic", "sklearn", "numpy.random",
|
||||
):
|
||||
assert forbidden not in src, (
|
||||
f"forbidden import {forbidden!r} in recognizer_synthesis.py"
|
||||
)
|
||||
Loading…
Reference in a new issue