diff --git a/core/cli.py b/core/cli.py index 03ea9fe8..938f17ba 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 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 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 pack-measurements\n core demo long-context-comparison\n core demo anti-regression\n core demo learning-loop\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" +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 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 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 pack-measurements\n core demo long-context-comparison\n core demo anti-regression\n core demo learning-loop\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" _TEST_SUITES: dict[str, tuple[str, ...]] = { "fast": ( @@ -491,6 +491,126 @@ def cmd_teaching_audit(args: argparse.Namespace) -> int: return 0 +def cmd_teaching_gaps(args: argparse.Namespace) -> int: + """Phase 1.1 — rank (subject, intent) cells the runtime would have + grounded but couldn't, aggregated from emitted DiscoveryCandidates. + + Reads JSONL files written by + :class:`teaching.discovery_sink.DiscoveryMonthlyFileSink` under + *root* (default ``teaching/discovery_log``) and emits a ranked + table of cells ordered by emission count. + + Pure read — never mutates the sink. + """ + from teaching.gaps import _DEFAULT_ROOT, aggregate_gaps + + root = Path(args.root) if args.root else _DEFAULT_ROOT + try: + rows = aggregate_gaps( + root=root, + since=args.since, + sample_limit=max(1, int(args.sample_limit)), + ) + except ValueError as exc: + _die(str(exc), code=2) + + if args.top is not None and args.top > 0: + rows = rows[: args.top] + + if args.json: + payload = { + "root": str(root) if root is not None else None, + "since": args.since, + "total_cells": len(rows), + "gaps": [g.as_dict() for g in rows], + } + print(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True)) + return 0 if rows else 1 + + if not rows: + print("No discovery candidates found.") + if root is not None and not root.exists(): + print(f" (root path does not exist: {root})") + return 1 + + print(f"{'rank':>4} {'subject':<24}{'intent':<14}{'count':>6} {'clean':>6} months") + print("-" * 80) + for i, gap in enumerate(rows, 1): + months = ",".join(gap.months_seen) if gap.months_seen else "—" + print( + f"{i:>4} {gap.subject[:24]:<24}{gap.intent[:14]:<14}" + f"{gap.count:>6} {gap.boundary_clean_count:>6} {months}" + ) + return 0 + + +def cmd_teaching_queue(args: argparse.Namespace) -> int: + """Phase 1.2 — show the auto-promoted gap queue. + + Reads the discovery sink (same path as ``core teaching gaps``), + aggregates by cell, and emits cells whose boundary-clean + emission count meets ``--threshold``. + + Boundary-tainted emissions (refusal/hedge fired during the + contributing turn) are excluded by default; ``--include-tainted`` + counts every emission toward the threshold. Operators reach for + that flag deliberately, not by accident. + """ + from teaching.gaps import _DEFAULT_ROOT, aggregate_gaps + from teaching.promotion import promote_gaps + + root = Path(args.root) if args.root else _DEFAULT_ROOT + try: + gaps = aggregate_gaps( + root=root, + since=args.since, + sample_limit=5, + ) + except ValueError as exc: + _die(str(exc), code=2) + + if args.threshold < 1: + _die(f"--threshold must be >= 1 (got {args.threshold})", code=2) + + promoted = promote_gaps( + gaps, + threshold=args.threshold, + include_tainted=args.include_tainted, + ) + + if args.json: + payload = { + "root": str(root), + "since": args.since, + "threshold": args.threshold, + "include_tainted": args.include_tainted, + "total_promoted": len(promoted), + "queue": [p.as_dict() for p in promoted], + } + print(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True)) + return 0 if promoted else 1 + + if not promoted: + print(f"No cells met threshold {args.threshold}.") + return 1 + + print( + f"{'rank':>4} {'queue_id':<48}{'count':>6} {'clean':>6} months" + ) + print("-" * 96) + for i, p in enumerate(promoted, 1): + months = ",".join(p.months_seen) if p.months_seen else "—" + print( + f"{i:>4} {p.queue_id[:48]:<48}{p.count:>6} {p.boundary_clean_count:>6} {months}" + ) + print() + print( + f"Author chains with: core teaching propose " + f"(or hand-author + supersede)." + ) + return 0 + + def _load_candidate_jsonl(path: str) -> Any: """Read one enriched DiscoveryCandidate JSONL line from *path*.""" from teaching.discovery import DiscoveryCandidate, EvidencePointer, SubQuestion @@ -1819,6 +1939,56 @@ def build_parser() -> argparse.ArgumentParser: ) teaching_audit.set_defaults(func=cmd_teaching_audit) + teaching_queue = teaching_sub.add_parser( + "queue", + help="show auto-promoted high-priority gaps (cells crossing --threshold)", + ) + teaching_queue.add_argument( + "--root", default=None, + help="discovery-sink root (default: teaching/discovery_log)", + ) + teaching_queue.add_argument( + "--since", default=None, + help="lower-bound month token YYYY-MM", + ) + teaching_queue.add_argument( + "--threshold", type=int, default=3, + help="minimum (boundary-clean) emissions to promote a cell (default: 3)", + ) + teaching_queue.add_argument( + "--include-tainted", action="store_true", + help="count refusal/hedge-tainted emissions toward the threshold", + ) + teaching_queue.add_argument( + "--json", action="store_true", help="machine-readable output", + ) + teaching_queue.set_defaults(func=cmd_teaching_queue) + + teaching_gaps = teaching_sub.add_parser( + "gaps", + help="rank (subject, intent) cells discovery candidates would have grounded", + ) + teaching_gaps.add_argument( + "--root", default=None, + help="discovery-sink root (default: teaching/discovery_log)", + ) + teaching_gaps.add_argument( + "--since", default=None, + help="lower-bound month token YYYY-MM (default: include every available month)", + ) + teaching_gaps.add_argument( + "--top", type=int, default=None, + help="show only the top N cells by emission count", + ) + teaching_gaps.add_argument( + "--sample-limit", type=int, default=5, + help="max candidate_ids retained per cell as samples (default: 5)", + ) + teaching_gaps.add_argument( + "--json", action="store_true", help="machine-readable output", + ) + teaching_gaps.set_defaults(func=cmd_teaching_gaps) + teaching_propose = teaching_sub.add_parser( "propose", help="convert an enriched DiscoveryCandidate (JSONL) into a TeachingChainProposal", diff --git a/teaching/gaps.py b/teaching/gaps.py new file mode 100644 index 00000000..93c57714 --- /dev/null +++ b/teaching/gaps.py @@ -0,0 +1,206 @@ +"""teaching/gaps.py — Phase 1.1: aggregate emitted DiscoveryCandidates +into a ranked view of (subject, intent) cells the runtime would have +grounded if the corpus had a chain there. + +ADR-0055 Phase B emits ``DiscoveryCandidate`` JSONL lines to an +attached :class:`teaching.discovery_sink.DiscoveryCandidateSink`. +:class:`DiscoveryMonthlyFileSink` persists them under +``//.jsonl``. That stream answers "which +prompts couldn't ground today" — but it's append-only and operators +don't grep raw JSONL. + +This module is the **reader side** of the flywheel: it walks the +sink's persisted output and groups candidates by ``(subject, intent)`` +cell so operators can see at a glance which cells are most-asked +without resorting to ad-hoc shell pipelines. + +Design constraints (matching CLAUDE.md doctrine): + + - Pure reader. No mutation of any sink file. Aggregation is + derived state; the sink remains the source of truth. + - Deterministic ordering: cells are sorted by (count desc, subject, + intent) so the same input always produces the same view. + - Date filtering operates on the sink's file naming convention + (``/.jsonl``) — month-level granularity, no + timestamp dependency. + - Malformed lines are skipped silently; the aggregator never raises + on a single bad line. The point is a useful summary, not a + schema validator (use ``teaching audit`` for that). +""" + +from __future__ import annotations + +import json +import re +from dataclasses import dataclass +from pathlib import Path +from typing import Iterable + + +_DEFAULT_ROOT: Path = Path(__file__).resolve().parent / "discovery_log" + +# Sink file naming convention: ``//.jsonl``. +# Regex anchors to filename only — full path components are not +# matched so the same regex applies to nested-or-flat layouts that +# alternative sinks might use. +_MONTH_FILE_RE = re.compile(r"^(\d{4})-(\d{2})\.jsonl$") +_MONTH_TOKEN_RE = re.compile(r"^(\d{4})-(\d{2})$") + + +@dataclass(frozen=True, slots=True) +class Gap: + """One aggregated ``(subject, intent)`` cell. + + Fields: + - ``subject`` / ``intent``: the cell identity (lower-case). + - ``count``: total number of candidate emissions whose proposed + chain referenced this cell, across all aggregated months. + - ``boundary_clean_count``: subset of ``count`` whose + ``boundary_clean`` flag was True (refusal/hedge-tainted + candidates are still counted toward ``count`` but split out + here so operators can filter). + - ``sample_candidate_ids``: up to 5 candidate_ids contributing + to this cell, sorted for determinism — useful for spot-checks. + - ``months_seen``: sorted ``YYYY-MM`` months where this cell + appeared at least once. + """ + + subject: str + intent: str + count: int + boundary_clean_count: int + sample_candidate_ids: tuple[str, ...] + months_seen: tuple[str, ...] + + def as_dict(self) -> dict[str, object]: + return { + "subject": self.subject, + "intent": self.intent, + "count": self.count, + "boundary_clean_count": self.boundary_clean_count, + "sample_candidate_ids": list(self.sample_candidate_ids), + "months_seen": list(self.months_seen), + } + + +def _normalise_since(since: str | None) -> tuple[int, int] | None: + """Return ``(year, month)`` for *since*, or None if absent. + + Raises :class:`ValueError` on a malformed token (caller surfaces + a friendly CLI error). + """ + if since is None: + return None + match = _MONTH_TOKEN_RE.match(since.strip()) + if not match: + raise ValueError( + f"--since {since!r} is not a YYYY-MM token (e.g. '2026-05')" + ) + return int(match.group(1)), int(match.group(2)) + + +def _iter_candidate_files( + root: Path, *, since: tuple[int, int] | None +) -> Iterable[tuple[str, Path]]: + """Yield ``(month_token, path)`` for every JSONL file under *root* + whose filename matches the monthly-sink convention. + + Files outside the ``YYYY-MM.jsonl`` convention are skipped — the + sink can grow alternative names later without breaking the + aggregator's behavior on the canonical layout. + """ + if not root.exists() or not root.is_dir(): + return + for path in sorted(root.rglob("*.jsonl")): + m = _MONTH_FILE_RE.match(path.name) + if not m: + continue + year = int(m.group(1)) + month = int(m.group(2)) + if since is not None and (year, month) < since: + continue + yield f"{year:04d}-{month:02d}", path + + +def aggregate_gaps( + root: Path = _DEFAULT_ROOT, + *, + since: str | None = None, + sample_limit: int = 5, +) -> tuple[Gap, ...]: + """Aggregate every emitted ``DiscoveryCandidate`` under *root* into + a ranked tuple of :class:`Gap` records. + + ``since`` accepts a ``YYYY-MM`` token. When supplied, only + candidate files whose monthly token is ``>= since`` are read. + + The returned tuple is sorted by ``(count desc, subject asc, + intent asc)`` so identical inputs produce identical orderings — + important for deterministic CLI output that operators can diff + across invocations. + """ + since_tuple = _normalise_since(since) + counts: dict[tuple[str, str], int] = {} + clean_counts: dict[tuple[str, str], int] = {} + samples: dict[tuple[str, str], list[str]] = {} + months: dict[tuple[str, str], set[str]] = {} + + for month_token, path in _iter_candidate_files(root, since=since_tuple): + try: + text = path.read_text(encoding="utf-8") + except OSError: + continue + for line in text.splitlines(): + line = line.strip() + if not line: + continue + try: + entry = json.loads(line) + except json.JSONDecodeError: + continue + if not isinstance(entry, dict): + continue + chain = entry.get("proposed_chain") + if not isinstance(chain, dict): + continue + subject = chain.get("subject") + intent = chain.get("intent") + if not isinstance(subject, str) or not isinstance(intent, str): + continue + subject = subject.strip().lower() + intent = intent.strip().lower() + if not subject or not intent: + continue + key = (subject, intent) + counts[key] = counts.get(key, 0) + 1 + if entry.get("boundary_clean") is True: + clean_counts[key] = clean_counts.get(key, 0) + 1 + sample_list = samples.setdefault(key, []) + candidate_id = entry.get("candidate_id") + if ( + isinstance(candidate_id, str) + and candidate_id + and len(sample_list) < sample_limit + and candidate_id not in sample_list + ): + sample_list.append(candidate_id) + months.setdefault(key, set()).add(month_token) + + rows: list[Gap] = [] + for key, total in counts.items(): + subject, intent = key + rows.append( + Gap( + subject=subject, + intent=intent, + count=total, + boundary_clean_count=clean_counts.get(key, 0), + sample_candidate_ids=tuple(sorted(samples.get(key, ()))), + months_seen=tuple(sorted(months.get(key, ()))), + ) + ) + rows.sort(key=lambda g: (-g.count, g.subject, g.intent)) + return tuple(rows) + + +__all__ = ["Gap", "aggregate_gaps"] diff --git a/teaching/promotion.py b/teaching/promotion.py new file mode 100644 index 00000000..94c4d1ef --- /dev/null +++ b/teaching/promotion.py @@ -0,0 +1,132 @@ +"""teaching/promotion.py — Phase 1.2: auto-promote high-frequency +discovery cells into an operator-visible review queue. + +The discovery aggregator (:mod:`teaching.gaps`) ranks +``(subject, intent)`` cells by how many DiscoveryCandidate emissions +they accumulated. ADR-0055 emits structured evidence; the +aggregator surfaces frequency; **this module closes the loop**: +when a cell's emission count crosses a threshold, it becomes a +:class:`GapPromotion` — an explicit "author a chain for me" signal +that operators can act on without grepping raw aggregation tables. + +Design constraints: + + - **Pure function of the aggregator output.** Promotion is derived + state — no separate persistent queue, no double-bookkeeping. + Re-running ``promote_gaps`` on the same sink contents produces + the same result deterministically. + - **Threshold is explicit.** No magic defaults. Operators pick + a threshold appropriate to their workload (3 is the v1 baseline: + "three independent cold-start prompts hit this cell; author the + chain"). + - **No content synthesis.** A promotion records that authorship is + *needed*; it never invents connective or object. Only an + operator can author a complete chain — the trust boundary that + prevents stochastic chain-generation is preserved. + - **Boundary-clean filter on by default.** A gap whose only + contributing candidates carry ``boundary_clean=False`` (refusal- + or hedge-tainted) does not promote, since those signals may + indicate the prompt itself violated a safety/ethics axis rather + than a curriculum gap. The operator can opt in to including + tainted cells via ``include_tainted=True``. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Iterable + +from teaching.gaps import Gap + + +@dataclass(frozen=True, slots=True) +class GapPromotion: + """A ``(subject, intent)`` cell whose emission count met the + auto-promotion threshold. + + Fields mirror :class:`teaching.gaps.Gap` for traceability, plus + ``threshold`` so the operator can see *why* this cell promoted + (a count of 7 at threshold 3 reads differently than 7 at 7). + """ + + subject: str + intent: str + count: int + boundary_clean_count: int + sample_candidate_ids: tuple[str, ...] + months_seen: tuple[str, ...] + threshold: int + + @property + def queue_id(self) -> str: + """Stable, deterministic identifier for this cell promotion. + + The same cell at the same threshold always produces the same + ``queue_id`` so operators can diff queue states across + invocations. + """ + return f"gap:{self.intent}:{self.subject}@{self.threshold}" + + def as_dict(self) -> dict[str, object]: + return { + "queue_id": self.queue_id, + "subject": self.subject, + "intent": self.intent, + "count": self.count, + "boundary_clean_count": self.boundary_clean_count, + "sample_candidate_ids": list(self.sample_candidate_ids), + "months_seen": list(self.months_seen), + "threshold": self.threshold, + } + + +def promote_gaps( + gaps: Iterable[Gap], + *, + threshold: int = 3, + include_tainted: bool = False, +) -> tuple[GapPromotion, ...]: + """Return the subset of *gaps* whose effective count meets *threshold*. + + Effective count: + - When ``include_tainted=True`` (default False), the comparison + uses :attr:`Gap.count` (every emission counts). + - When ``include_tainted=False`` (default), the comparison uses + :attr:`Gap.boundary_clean_count` (refusal/hedge-tainted + emissions are excluded from the promotion decision). + + Threshold must be ``>= 1``; a threshold of zero would promote + every observed cell and defeats the purpose of a priority queue. + """ + if threshold < 1: + raise ValueError(f"threshold must be >= 1 (got {threshold!r})") + + promoted: list[GapPromotion] = [] + for gap in gaps: + effective_count = gap.count if include_tainted else gap.boundary_clean_count + if effective_count < threshold: + continue + promoted.append( + GapPromotion( + subject=gap.subject, + intent=gap.intent, + count=gap.count, + boundary_clean_count=gap.boundary_clean_count, + sample_candidate_ids=gap.sample_candidate_ids, + months_seen=gap.months_seen, + threshold=threshold, + ) + ) + # Deterministic order: highest effective count first, ties broken + # by subject then intent — mirrors :func:`aggregate_gaps`. + promoted.sort( + key=lambda p: ( + -(p.count if include_tainted else p.boundary_clean_count), + p.subject, + p.intent, + ) + ) + return tuple(promoted) + + +__all__ = ["GapPromotion", "promote_gaps"] diff --git a/tests/test_teaching_gaps.py b/tests/test_teaching_gaps.py new file mode 100644 index 00000000..98827510 --- /dev/null +++ b/tests/test_teaching_gaps.py @@ -0,0 +1,220 @@ +"""Phase 1.1 — discovery-gap aggregation tests. + +The contract these tests pin: + + - ``aggregate_gaps`` is a pure reader: never mutates sink files, + returns deterministic ordering, skips malformed lines silently. + - Filenames follow ``YYYY-MM.jsonl`` under ``//`` — + other names are ignored. + - ``--since YYYY-MM`` filters by month (inclusive lower bound). + - ``boundary_clean=false`` candidates are counted but split out so + operators can filter refusal/hedge-tainted cells separately. + - ``top`` truncation preserves order. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from teaching.gaps import Gap, aggregate_gaps + + +def _write_line(path: Path, payload: dict) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("a", encoding="utf-8") as fh: + fh.write(json.dumps(payload, sort_keys=True, separators=(",", ":"))) + fh.write("\n") + + +def _candidate( + *, + candidate_id: str, + subject: str, + intent: str, + boundary_clean: bool = True, +) -> dict: + return { + "candidate_id": candidate_id, + "proposed_chain": { + "subject": subject, + "intent": intent, + "connective": None, + "object": None, + }, + "trigger": "would_have_grounded", + "source_turn_trace": "trace-" + candidate_id, + "pack_consistent": True, + "boundary_clean": boundary_clean, + "review_state": "unreviewed", + } + + +# --------------------------------------------------------------------------- +# Aggregation — basic counts, ordering, sample retention +# --------------------------------------------------------------------------- + + +def test_aggregates_by_subject_intent(tmp_path: Path) -> None: + sink = tmp_path / "2026" / "2026-05.jsonl" + _write_line(sink, _candidate(candidate_id="a", subject="parent", intent="cause")) + _write_line(sink, _candidate(candidate_id="b", subject="parent", intent="cause")) + _write_line(sink, _candidate(candidate_id="c", subject="child", intent="verification")) + + rows = aggregate_gaps(tmp_path) + + assert len(rows) == 2 + parent_cause = next(g for g in rows if g.subject == "parent") + child_verif = next(g for g in rows if g.subject == "child") + assert parent_cause.intent == "cause" + assert parent_cause.count == 2 + assert parent_cause.boundary_clean_count == 2 + assert parent_cause.sample_candidate_ids == ("a", "b") + assert parent_cause.months_seen == ("2026-05",) + assert child_verif.count == 1 + + +def test_rank_order_count_desc_then_subject(tmp_path: Path) -> None: + sink = tmp_path / "2026" / "2026-05.jsonl" + # 3x parent-cause, 2x child-cause, 1x family-cause + for i in range(3): + _write_line(sink, _candidate(candidate_id=f"p{i}", subject="parent", intent="cause")) + for i in range(2): + _write_line(sink, _candidate(candidate_id=f"c{i}", subject="child", intent="cause")) + _write_line(sink, _candidate(candidate_id="f0", subject="family", intent="cause")) + + rows = aggregate_gaps(tmp_path) + + assert [g.subject for g in rows] == ["parent", "child", "family"] + assert [g.count for g in rows] == [3, 2, 1] + + +def test_top_truncation_preserves_order(tmp_path: Path) -> None: + sink = tmp_path / "2026" / "2026-05.jsonl" + for i in range(3): + _write_line(sink, _candidate(candidate_id=f"p{i}", subject="parent", intent="cause")) + _write_line(sink, _candidate(candidate_id="c0", subject="child", intent="cause")) + + rows = aggregate_gaps(tmp_path) + assert len(rows) == 2 + # Top-1 should yield the parent cell only. + assert rows[0].subject == "parent" + assert rows[0].count == 3 + + +# --------------------------------------------------------------------------- +# Boundary-clean accounting +# --------------------------------------------------------------------------- + + +def test_boundary_tainted_candidates_count_but_split(tmp_path: Path) -> None: + sink = tmp_path / "2026" / "2026-05.jsonl" + _write_line(sink, _candidate(candidate_id="clean", subject="parent", intent="cause")) + _write_line(sink, _candidate( + candidate_id="tainted", subject="parent", intent="cause", boundary_clean=False, + )) + + rows = aggregate_gaps(tmp_path) + assert len(rows) == 1 + assert rows[0].count == 2 + assert rows[0].boundary_clean_count == 1 + + +# --------------------------------------------------------------------------- +# --since month filter +# --------------------------------------------------------------------------- + + +def test_since_filter_excludes_earlier_months(tmp_path: Path) -> None: + _write_line(tmp_path / "2026" / "2026-04.jsonl", + _candidate(candidate_id="april", subject="parent", intent="cause")) + _write_line(tmp_path / "2026" / "2026-05.jsonl", + _candidate(candidate_id="may", subject="parent", intent="cause")) + + rows = aggregate_gaps(tmp_path, since="2026-05") + assert len(rows) == 1 + assert rows[0].count == 1 + assert rows[0].sample_candidate_ids == ("may",) + + +def test_since_filter_includes_boundary_month(tmp_path: Path) -> None: + _write_line(tmp_path / "2026" / "2026-05.jsonl", + _candidate(candidate_id="may", subject="parent", intent="cause")) + rows = aggregate_gaps(tmp_path, since="2026-05") + assert len(rows) == 1 + + +def test_since_rejects_malformed_token(tmp_path: Path) -> None: + with pytest.raises(ValueError): + aggregate_gaps(tmp_path, since="May 2026") + + +# --------------------------------------------------------------------------- +# Robustness — missing root, malformed JSONL, non-monthly filenames +# --------------------------------------------------------------------------- + + +def test_missing_root_returns_empty_tuple(tmp_path: Path) -> None: + rows = aggregate_gaps(tmp_path / "does_not_exist") + assert rows == () + + +def test_malformed_lines_silently_skipped(tmp_path: Path) -> None: + sink = tmp_path / "2026" / "2026-05.jsonl" + sink.parent.mkdir(parents=True, exist_ok=True) + sink.write_text( + "\n".join([ + "not json", + "{}", # missing proposed_chain + json.dumps({"proposed_chain": {"subject": ""}}), # empty subject + json.dumps(_candidate(candidate_id="ok", subject="parent", intent="cause")), + ]), + encoding="utf-8", + ) + rows = aggregate_gaps(tmp_path) + assert len(rows) == 1 + assert rows[0].subject == "parent" + assert rows[0].count == 1 + + +def test_non_monthly_filenames_ignored(tmp_path: Path) -> None: + bad = tmp_path / "2026" / "notes.jsonl" + good = tmp_path / "2026" / "2026-05.jsonl" + _write_line(bad, _candidate(candidate_id="bad", subject="parent", intent="cause")) + _write_line(good, _candidate(candidate_id="good", subject="parent", intent="cause")) + + rows = aggregate_gaps(tmp_path) + assert len(rows) == 1 + assert rows[0].count == 1 + assert rows[0].sample_candidate_ids == ("good",) + + +def test_aggregation_is_deterministic(tmp_path: Path) -> None: + sink = tmp_path / "2026" / "2026-05.jsonl" + for s in ("parent", "child", "ancestor"): + _write_line(sink, _candidate(candidate_id=f"id-{s}", subject=s, intent="cause")) + + a = aggregate_gaps(tmp_path) + b = aggregate_gaps(tmp_path) + assert a == b + assert [g.as_dict() for g in a] == [g.as_dict() for g in b] + + +def test_sample_limit_caps_retained_ids(tmp_path: Path) -> None: + sink = tmp_path / "2026" / "2026-05.jsonl" + for i in range(10): + _write_line(sink, _candidate(candidate_id=f"id-{i:02d}", subject="parent", intent="cause")) + rows = aggregate_gaps(tmp_path, sample_limit=3) + assert rows[0].count == 10 + assert len(rows[0].sample_candidate_ids) == 3 + + +def test_gap_dataclass_is_frozen() -> None: + gap = Gap( + subject="parent", intent="cause", count=1, + boundary_clean_count=1, sample_candidate_ids=("a",), months_seen=("2026-05",), + ) + with pytest.raises((AttributeError, TypeError)): + gap.count = 99 # type: ignore[misc] diff --git a/tests/test_teaching_promotion.py b/tests/test_teaching_promotion.py new file mode 100644 index 00000000..85a20448 --- /dev/null +++ b/tests/test_teaching_promotion.py @@ -0,0 +1,163 @@ +"""Phase 1.2 — gap auto-promotion tests. + +The contract these tests pin: + + - ``promote_gaps`` is a pure derivation from :class:`Gap` records. + - Default mode filters by ``boundary_clean_count``; tainted-only + cells never auto-promote unless ``include_tainted=True``. + - Threshold ``< 1`` raises — a zero threshold defeats the queue. + - Ordering: highest effective count first, ties broken by + (subject, intent) — matches :func:`aggregate_gaps`. + - ``queue_id`` is stable + deterministic. +""" + +from __future__ import annotations + +import pytest + +from teaching.gaps import Gap +from teaching.promotion import GapPromotion, promote_gaps + + +def _gap( + subject: str, + intent: str = "cause", + count: int = 3, + clean: int | None = None, + months: tuple[str, ...] = ("2026-05",), +) -> Gap: + return Gap( + subject=subject, + intent=intent, + count=count, + boundary_clean_count=count if clean is None else clean, + sample_candidate_ids=("a", "b"), + months_seen=months, + ) + + +# --------------------------------------------------------------------------- +# Default mode — boundary_clean_count gates the promotion +# --------------------------------------------------------------------------- + + +def test_clean_count_meets_threshold_promotes() -> None: + gaps = (_gap("parent", count=5, clean=5),) + promoted = promote_gaps(gaps, threshold=3) + assert len(promoted) == 1 + assert promoted[0].subject == "parent" + assert promoted[0].count == 5 + assert promoted[0].boundary_clean_count == 5 + assert promoted[0].threshold == 3 + + +def test_clean_count_below_threshold_does_not_promote() -> None: + gaps = (_gap("parent", count=5, clean=2),) + promoted = promote_gaps(gaps, threshold=3) + assert promoted == () + + +def test_tainted_only_cell_does_not_promote_by_default() -> None: + """A cell with count=5 but boundary_clean_count=0 means every + emission was refusal/hedge-tainted. Default mode must NOT + promote — those signals may indicate the *prompt* hit a safety + axis, not a curriculum gap.""" + gaps = (_gap("forbidden_thing", count=5, clean=0),) + promoted = promote_gaps(gaps, threshold=3) + assert promoted == () + + +def test_include_tainted_counts_every_emission() -> None: + gaps = (_gap("forbidden_thing", count=5, clean=0),) + promoted = promote_gaps(gaps, threshold=3, include_tainted=True) + assert len(promoted) == 1 + assert promoted[0].count == 5 + assert promoted[0].boundary_clean_count == 0 + + +# --------------------------------------------------------------------------- +# Threshold validation +# --------------------------------------------------------------------------- + + +def test_threshold_must_be_positive() -> None: + with pytest.raises(ValueError): + promote_gaps((_gap("parent"),), threshold=0) + + +# --------------------------------------------------------------------------- +# Ordering +# --------------------------------------------------------------------------- + + +def test_order_is_count_desc_then_subject() -> None: + gaps = ( + _gap("child", count=3, clean=3), + _gap("parent", count=5, clean=5), + _gap("ancestor", count=3, clean=3), + ) + promoted = promote_gaps(gaps, threshold=3) + assert [p.subject for p in promoted] == ["parent", "ancestor", "child"] + + +def test_order_is_stable_for_identical_counts() -> None: + gaps = ( + _gap("child", count=3), + _gap("ancestor", count=3), + _gap("parent", count=3), + ) + a = promote_gaps(gaps, threshold=3) + b = promote_gaps(gaps, threshold=3) + assert a == b + assert [p.subject for p in a] == ["ancestor", "child", "parent"] + + +# --------------------------------------------------------------------------- +# queue_id stability +# --------------------------------------------------------------------------- + + +def test_queue_id_format_and_stability() -> None: + gaps = (_gap("parent", intent="cause", count=3, clean=3),) + promoted = promote_gaps(gaps, threshold=3) + assert promoted[0].queue_id == "gap:cause:parent@3" + + # Same cell at a different threshold → different queue_id. + promoted2 = promote_gaps( + (_gap("parent", intent="cause", count=10, clean=10),), threshold=5 + ) + assert promoted2[0].queue_id == "gap:cause:parent@5" + + +def test_queue_id_distinguishes_intent() -> None: + promoted = promote_gaps( + ( + _gap("parent", intent="cause", count=3, clean=3), + _gap("parent", intent="verification", count=3, clean=3), + ), + threshold=3, + ) + queue_ids = {p.queue_id for p in promoted} + assert queue_ids == {"gap:cause:parent@3", "gap:verification:parent@3"} + + +# --------------------------------------------------------------------------- +# Promotion is a pure derivation — no side effects +# --------------------------------------------------------------------------- + + +def test_promotion_does_not_mutate_input() -> None: + gaps = (_gap("parent", count=3, clean=3),) + snapshot = gaps[0] + promote_gaps(gaps, threshold=3) + promote_gaps(gaps, threshold=2, include_tainted=True) + assert gaps[0] == snapshot + + +def test_promotion_is_frozen() -> None: + promo = GapPromotion( + subject="parent", intent="cause", count=3, boundary_clean_count=3, + sample_candidate_ids=("a",), months_seen=("2026-05",), threshold=3, + ) + with pytest.raises((AttributeError, TypeError)): + promo.count = 99 # type: ignore[misc]