diff --git a/core/cli.py b/core/cli.py index 3185b1a9..aa2ba062 100644 --- a/core/cli.py +++ b/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 \n core teaching propose \n core teaching proposals --state pending\n core teaching review --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 \n core teaching propose \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 --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", diff --git a/teaching/contemplation.py b/teaching/contemplation.py index 6d8af375..179c9769 100644 --- a/teaching/contemplation.py +++ b/teaching/contemplation.py @@ -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:`` 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", ] diff --git a/teaching/exemplar_ingest.py b/teaching/exemplar_ingest.py new file mode 100644 index 00000000..f4e07d77 --- /dev/null +++ b/teaching/exemplar_ingest.py @@ -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", +] diff --git a/teaching/recognizer_synthesis.py b/teaching/recognizer_synthesis.py new file mode 100644 index 00000000..5d4645e7 --- /dev/null +++ b/teaching/recognizer_synthesis.py @@ -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", +] diff --git a/teaching/replay.py b/teaching/replay.py index d357a294..e499f93e 100644 --- a/teaching/replay.py +++ b/teaching/replay.py @@ -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", +] diff --git a/teaching/source.py b/teaching/source.py index a7cff216..60e5d72a 100644 --- a/teaching/source.py +++ b/teaching/source.py @@ -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)) diff --git a/tests/test_admissibility_replay_gate.py b/tests/test_admissibility_replay_gate.py new file mode 100644 index 00000000..90302600 --- /dev/null +++ b/tests/test_admissibility_replay_gate.py @@ -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 diff --git a/tests/test_exemplar_ingest.py b/tests/test_exemplar_ingest.py new file mode 100644 index 00000000..261969b2 --- /dev/null +++ b/tests/test_exemplar_ingest.py @@ -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" + ) diff --git a/tests/test_proposal_source.py b/tests/test_proposal_source.py index 70458f13..22cace2d 100644 --- a/tests/test_proposal_source.py +++ b/tests/test_proposal_source.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"} ) diff --git a/tests/test_propose_from_exemplars_cli.py b/tests/test_propose_from_exemplars_cli.py new file mode 100644 index 00000000..86d3c63c --- /dev/null +++ b/tests/test_propose_from_exemplars_cli.py @@ -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}" + ) diff --git a/tests/test_recognizer_synthesis.py b/tests/test_recognizer_synthesis.py new file mode 100644 index 00000000..649fe355 --- /dev/null +++ b/tests/test_recognizer_synthesis.py @@ -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" + )