core-labs/core
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

feat(teaching): discovery gaps aggregator + auto-promotion queue (Phase 1.1+1.2)

Browse source 
Closes the corpus flywheel.  ADR-0055 Phase B emits DiscoveryCandidate
JSONL to the discovery sink, but until now there was no operator-facing
view: candidates accumulated to disk, no one grepped them, the system's
"I would have grounded this if I had a chain" signal went into a void.

P1.1 — Discovery aggregator (teaching/gaps.py).

  Pure reader over the discovery-sink monthly-rollover layout
  (<root>/<YYYY>/<YYYY-MM>.jsonl).  aggregate_gaps(root, since,
  sample_limit) groups emitted candidates by (subject, intent) cell
  and returns a deterministic ranked tuple of Gap records.

  - count: total emissions
  - boundary_clean_count: subset whose boundary_clean flag held
    (refusal/hedge-tainted emissions split out so operators can filter)
  - sample_candidate_ids: up to N retained ids per cell, sorted
  - months_seen: every month token where the cell appeared

  --since YYYY-MM filters by file naming convention (no timestamp
  dependency).  Malformed lines silently skipped.  Default root:
  teaching/discovery_log.

  CLI: core teaching gaps [--root PATH] [--since YYYY-MM] [--top N]
                          [--sample-limit N] [--json]

P1.2 — Auto-promotion queue (teaching/promotion.py).

  promote_gaps(gaps, threshold, include_tainted) lifts cells whose
  effective count meets the threshold into GapPromotion records.

  - Default mode: boundary_clean_count gates promotion.  Tainted-only
    cells (count > 0 but all emissions refusal/hedge-tainted) do not
    auto-promote — those may indicate the prompt hit a safety axis,
    not a curriculum gap.
  - include_tainted=True counts every emission (operator override).
  - Threshold must be >= 1 (zero threshold defeats the queue).
  - queue_id is stable + deterministic (gap:<intent>:<subject>@<N>).
  - No content synthesis — promotion never invents connective or
    object; only an operator can author a complete chain via the
    propose/replay/accept pipeline.

  CLI: core teaching queue [--threshold N] [--include-tainted]
                           [--root PATH] [--since YYYY-MM] [--json]

Operator workflow (closed loop):

  operator → core chat                            # asks question
           ← cold turn emits DiscoveryCandidate
  operator → core teaching gaps --top 10          # ranked gaps
  operator → core teaching queue --threshold 3    # auto-promoted
  operator → authors candidate JSONL
  operator → core teaching propose <path>         # replay gate runs
  operator → core teaching review <id> --accept   # corpus mutates

24 new tests (13 gaps + 11 promotion), all pure / no I/O dependencies,
fast (<1s combined).  Full lane: 1933 passed, 2 skipped.
This commit is contained in:
Shay 2026-05-18 16:04:39 -07:00
parent b5ba9b6d6f
commit 84e74eede8
5 changed files with 892 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

172
core/cli.py
View file

@ -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 <candidate-jsonl-path>\n core teaching proposals --state pending\n core teaching review <proposal_id> --accept --review-date 2026-05-18\n core teaching supersede cause_light_reveals_truth --subject light --intent cause --connective grounds --object truth --review-date 2026-05-18\n core teaching supersessions\n core teaching supersessions --json\n core test --suite fast -q\n core test --suite pulse -q\n core test --suite proof -q\n core test --suite cognition -q\n core test -- tests/test_alignment_graph.py -q\n core demo audit-tour\n core demo 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 <candidate-jsonl-path>\n core teaching proposals --state pending\n core teaching review <proposal_id> --accept --review-date 2026-05-18\n core teaching supersede cause_light_reveals_truth --subject light --intent cause --connective grounds --object truth --review-date 2026-05-18\n core teaching supersessions\n core teaching supersessions --json\n core test --suite fast -q\n core test --suite pulse -q\n core test --suite proof -q\n core test --suite cognition -q\n core test -- tests/test_alignment_graph.py -q\n core demo audit-tour\n core demo 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 <candidate-jsonl> "
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",

206
teaching/gaps.py Normal file
View file

@ -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
``<root>/<YYYY>/<YYYY-MM>.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
(``<YYYY>/<YYYY-MM>.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: ``<root>/<YYYY>/<YYYY-MM>.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"]

132
teaching/promotion.py Normal file
View file

@ -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"]

220
tests/test_teaching_gaps.py Normal file
View file

@ -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 ``<root>/<YYYY>/`` —
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]

163
tests/test_teaching_promotion.py Normal file
View file

@ -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]