core-labs/core
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions
core/tests/test_teaching_queue.py
Shay ec5d6f5ac7
feat(ADR-0161.1): core teaching queue list|show — read-only queue projection (#296) 
* docs(math): ADR-0163 — path to GSM8K mastery via candidate-graph admissibility (proposed)

Audit reframes the math roadmap entirely.

State of main: every named math capability axis (G1..G5, S1) passes
at 100% with wrong=0 on its controlled lane.  binding_graph,
math_versor_arithmetic, math_symbolic_equivalence, math_parser,
math_candidate_parser, math_solver, math_verifier, math_realizer,
math_problem_graph — all landed.  The worktrees on disk are stale
forks.

State of GSM8K (50-case train sample): correct=0, refused=50, wrong=0.
Every refusal reason is identical: "candidate_graph: no admissible
candidate for statement: <STATEMENT>".

The reframe: the gap is NOT in operator algebra, NOT in binding graph
internals, NOT in symbolic equivalence.  The gap is in
generate/math_candidate_graph.py — the admissibility surface that
turns a natural-language statement into a candidate the downstream
pipeline can consume.  The capability axes pass at 100% because they
test statement shapes the candidate-graph already admits.  GSM8K
refuses at 100% because its statements span shapes the candidate-graph
has never been taught.

Six-phase plan to lift GSM8K under the thesis "decodes, not generates":

A. Refusal taxonomy (measure before building)
B. Exemplar corpora per shape category (≤20 statements each, ≤3 per round)
C. Contemplation runner ingests exemplars; emits DerivedRecognizer
   proposals
D. Operator ratifies through ADR-0161 HITL queue (no new surface)
E. Re-baseline GSM8K train sample.  Round 1 exit: correct ≥ 10, wrong = 0.
   Round 2: ≥ 25.  Round 3: ≥ 35.
F. Scale to public/v1 (200 cases, target correct ≥ 100), then
   holdout (measurement-only — never tune against).

Three non-negotiables:
- wrong = 0 at every phase.  Auto-rejected by replay gate, not by
  operator vigilance.
- No hand-rolled recognizers in generate/.  Every recognizer lands
  via contemplation → proposal → review corridor.
- Active corpus mutation only via accept_proposal.

Status: proposed.  Implementation lands as three PRs starting with
Phase A scaffolding.

Scope discipline: docs-only.  No code, no eval changes, no corpus
mutation.

* feat(ADR-0161.1): core teaching queue list|show — read-only queue projection

* fix(ADR-0161.1): restore gap-queue CLI + rename new commands to hitl-queue + R1..R5 refinements
2026-05-26 11:42:51 -07:00

535 lines
16 KiB
Python
Raw Blame History

"""Tests for ADR-0161 human-in-the-loop review queue read-only commands."""
from __future__ import annotations
import json
from io import StringIO
from pathlib import Path
from unittest.mock import patch
import pytest
from core.cli import main
from teaching.discovery import DiscoveryCandidate, EvidencePointer
from teaching.proposals import ProposalLog, ReplayEvidence, build_proposal
from teaching.queue import derive_queue, find_queue_item
def make_candidate(candidate_id: str, subject: str, obj: str = "truth") -> DiscoveryCandidate:
return DiscoveryCandidate(
candidate_id=candidate_id,
proposed_chain={
"subject": subject,
"intent": "cause",
"connective": "reveals",
"object": obj,
},
trigger="would_have_grounded",
source_turn_trace="trace_1",
pack_consistent=True,
boundary_clean=True,
polarity="affirms",
claim_domain="factual",
evidence=(
EvidencePointer(
source="corpus",
ref="some_chain",
polarity="affirms",
epistemic_status="coherent",
),
),
)
def run_cli(args: list[str]) -> tuple[int, str, str]:
with patch("sys.argv", ["core"] + args), patch("sys.stdout", new_callable=StringIO) as out, patch("sys.stderr", new_callable=StringIO) as err:
try:
code = main()
except SystemExit as e:
code = e.code if isinstance(e.code, int) else 0
return code, out.getvalue(), err.getvalue()
# ---------------------------------------------------------------------------
# derive_queue purity and determinism
# ---------------------------------------------------------------------------
def test_derive_queue_purity(tmp_path: Path):
log_path = tmp_path / "proposals.jsonl"
log = ProposalLog(log_path)
runs_dir = tmp_path / "runs"
runs_dir.mkdir()
c1 = make_candidate("cand1", "light")
p1 = build_proposal(c1)
log.record_created(p1)
res1 = derive_queue(log, contemplation_runs_dir=runs_dir)
res2 = derive_queue(log, contemplation_runs_dir=runs_dir)
assert res1 == res2
def test_derive_queue_determinism(tmp_path: Path):
log_path = tmp_path / "proposals.jsonl"
log = ProposalLog(log_path)
runs_dir = tmp_path / "runs"
runs_dir.mkdir()
c1 = make_candidate("cand1", "light")
p1 = build_proposal(c1)
log.record_created(p1)
res1 = derive_queue(log, contemplation_runs_dir=runs_dir)
res2 = derive_queue(log, contemplation_runs_dir=runs_dir)
assert res1 == res2
# ---------------------------------------------------------------------------
# State derivation
# ---------------------------------------------------------------------------
def test_state_derivation(tmp_path: Path):
log_path = tmp_path / "proposals.jsonl"
log = ProposalLog(log_path)
runs_dir = tmp_path / "runs"
runs_dir.mkdir()
c1 = make_candidate("cand1", "light")
p1 = build_proposal(c1)
log.record_created(p1)
items = derive_queue(log, contemplation_runs_dir=runs_dir)
assert len(items) == 1
assert items[0].state == "pending"
log.record_transition(p1.proposal_id, "accepted", "review accepted")
items = derive_queue(log, contemplation_runs_dir=runs_dir)
assert len(items) == 1
assert items[0].state == "accepted"
log.record_transition(p1.proposal_id, "withdrawn", "review withdrawn")
items = derive_queue(log, contemplation_runs_dir=runs_dir)
assert len(items) == 1
assert items[0].state == "withdrawn"
# ---------------------------------------------------------------------------
# Age proposals
# ---------------------------------------------------------------------------
def test_age_proposals(tmp_path: Path):
log_path = tmp_path / "proposals.jsonl"
log = ProposalLog(log_path)
runs_dir = tmp_path / "runs"
runs_dir.mkdir()
c1 = make_candidate("cand1", "light")
p1 = build_proposal(c1)
c2 = make_candidate("cand2", "dark")
p2 = build_proposal(c2)
c3 = make_candidate("cand3", "shade")
p3 = build_proposal(c3)
log.record_created(p1)
log.record_created(p2)
log.record_created(p3)
items = derive_queue(log, contemplation_runs_dir=runs_dir)
assert len(items) == 3
assert items[0].proposal_id == p1.proposal_id
assert items[1].proposal_id == p2.proposal_id
assert items[2].proposal_id == p3.proposal_id
assert items[0].age_proposals == 2
assert items[1].age_proposals == 1
assert items[2].age_proposals == 0
log.record_transition(p1.proposal_id, "accepted", "accepting first")
items = derive_queue(log, contemplation_runs_dir=runs_dir)
assert items[0].age_proposals == 0
assert items[1].age_proposals == 1
assert items[2].age_proposals == 0
# ---------------------------------------------------------------------------
# Contemplation report path
# ---------------------------------------------------------------------------
def test_contemplation_report_path(tmp_path: Path):
log_path = tmp_path / "proposals.jsonl"
log = ProposalLog(log_path)
c1 = make_candidate("cand1", "light")
p1 = build_proposal(c1)
c2 = make_candidate("cand2", "dark")
p2 = build_proposal(c2)
c3 = make_candidate("cand3", "shade")
p3 = build_proposal(c3)
log.record_created(p1)
log.record_created(p2)
log.record_created(p3)
runs_dir = tmp_path / "runs"
runs_dir.mkdir()
run1_path = runs_dir / "run1.json"
run1_path.write_text(json.dumps({
"proposal_id": p1.proposal_id,
"scenes": []
}))
run2_path = runs_dir / "run2.json"
run2_path.write_text(json.dumps({
"scenes": [
{
"scene": "S3_engine_authored_proposal",
"detail": {
"proposal_id": p2.proposal_id
}
}
]
}))
items = derive_queue(log, contemplation_runs_dir=runs_dir)
assert len(items) == 3
assert items[0].proposal_id == p1.proposal_id
assert items[0].contemplation_report_path == str(run1_path.resolve())
assert items[1].proposal_id == p2.proposal_id
assert items[1].contemplation_report_path == str(run2_path.resolve())
assert items[2].proposal_id == p3.proposal_id
assert items[2].contemplation_report_path is None
# ---------------------------------------------------------------------------
# Review history
# ---------------------------------------------------------------------------
def test_review_history(tmp_path: Path):
log_path = tmp_path / "proposals.jsonl"
log = ProposalLog(log_path)
runs_dir = tmp_path / "runs"
runs_dir.mkdir()
c1 = make_candidate("cand1", "light")
p1 = build_proposal(c1)
log.record_created(p1)
log.record_transition(p1.proposal_id, "rejected", "regression check")
log.record_transition(p1.proposal_id, "pending", "re-evaluated")
log.record_transition(p1.proposal_id, "accepted", "approved")
items = derive_queue(log, contemplation_runs_dir=runs_dir)
assert len(items) == 1
item = items[0]
assert len(item.review_history) == 3
assert item.review_history[0]["to"] == "rejected"
assert item.review_history[0]["note"] == "regression check"
assert item.review_history[1]["to"] == "pending"
assert item.review_history[1]["note"] == "re-evaluated"
assert item.review_history[2]["to"] == "accepted"
assert item.review_history[2]["note"] == "approved"
# ---------------------------------------------------------------------------
# find_queue_item
# ---------------------------------------------------------------------------
def test_find_queue_item(tmp_path: Path):
log_path = tmp_path / "proposals.jsonl"
log = ProposalLog(log_path)
runs_dir = tmp_path / "runs"
runs_dir.mkdir()
c1 = make_candidate("cand1", "light")
p1 = build_proposal(c1)
c2 = make_candidate("cand2", "dark")
p2 = build_proposal(c2)
log.record_created(p1)
log.record_created(p2)
res = find_queue_item(log, p1.proposal_id, runs_dir)
assert res is not None
assert res.proposal_id == p1.proposal_id
prefix = p2.proposal_id[:12]
res_prefix = find_queue_item(log, prefix, runs_dir)
assert res_prefix is not None
assert res_prefix.proposal_id == p2.proposal_id
assert find_queue_item(log, "non_existent_id", runs_dir) is None
common_prefix = p1.proposal_id[0]
if p2.proposal_id.startswith(common_prefix):
assert find_queue_item(log, common_prefix, runs_dir) is None
# ---------------------------------------------------------------------------
# R2: contemplation_report_path caching
# ---------------------------------------------------------------------------
def test_contemplation_mapping_caching(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
log_path = tmp_path / "proposals.jsonl"
log = ProposalLog(log_path)
c1 = make_candidate("cand1", "light")
p1 = build_proposal(c1)
log.record_created(p1)
runs_dir = tmp_path / "runs"
runs_dir.mkdir()
run_file = runs_dir / "run1.json"
run_file.write_text(json.dumps({"proposal_id": p1.proposal_id}))
glob_called_count = 0
orig_glob = Path.glob
def mock_glob(self, pattern):
nonlocal glob_called_count
if str(self) == str(runs_dir.resolve()) or str(self) == str(runs_dir):
glob_called_count += 1
return orig_glob(self, pattern)
monkeypatch.setattr(Path, "glob", mock_glob)
# First call walks the directory
res1 = derive_queue(log, contemplation_runs_dir=runs_dir)
assert len(res1) == 1
assert glob_called_count == 1
# Second call hits cache (mtime is unchanged)
res2 = derive_queue(log, contemplation_runs_dir=runs_dir)
assert len(res2) == 1
assert glob_called_count == 1
# Change the runs directory modification time (simulated via touch/new file)
run_file2 = runs_dir / "run2.json"
run_file2.write_text(json.dumps({"proposal_id": "dummy"}))
# Third call re-evaluates
res3 = derive_queue(log, contemplation_runs_dir=runs_dir)
assert len(res3) == 1
assert glob_called_count == 2
# ---------------------------------------------------------------------------
# R4: empty string source_id normalization
# ---------------------------------------------------------------------------
def test_source_id_empty_string_normalization(tmp_path: Path):
log_path = tmp_path / "proposals.jsonl"
log = ProposalLog(log_path)
runs_dir = tmp_path / "runs"
runs_dir.mkdir()
# build_proposal returns proposal with source.source_id = "" by default
c1 = make_candidate("cand1", "light")
p1 = build_proposal(c1)
assert p1.source.source_id == ""
log.record_created(p1)
items = derive_queue(log, contemplation_runs_dir=runs_dir)
assert len(items) == 1
assert items[0].source_id is None
# ---------------------------------------------------------------------------
# CLI list and show commands
# ---------------------------------------------------------------------------
def test_cli_list_command(tmp_path: Path):
log_path = tmp_path / "proposals.jsonl"
log = ProposalLog(log_path)
c1 = make_candidate("cand1", "light")
p1 = build_proposal(c1)
log.record_created(p1)
log.record_replay(p1.proposal_id, ReplayEvidence(
baseline={}, candidate={}, regressed_metrics=(), replay_equivalent=True
))
runs_dir = tmp_path / "runs"
runs_dir.mkdir()
code, stdout, stderr = run_cli([
"teaching", "hitl-queue", "list",
"--log-path", str(log_path),
"--contemplation-runs-dir", str(runs_dir)
])
assert code == 0
assert "proposal_id" in stdout
assert "source_kind" in stdout
assert "state" in stdout
assert "age" in stdout
assert "replay" in stdout
assert p1.proposal_id[:12] in stdout
assert "ok" in stdout
code_j, stdout_j, stderr_j = run_cli([
"teaching", "hitl-queue", "list",
"--log-path", str(log_path),
"--contemplation-runs-dir", str(runs_dir),
"--json"
])
assert code_j == 0
parsed = json.loads(stdout_j)
assert isinstance(parsed, list)
assert len(parsed) == 1
assert parsed[0]["proposal_id"] == p1.proposal_id
assert parsed[0]["source_kind"] == "operator"
assert parsed[0]["state"] == "pending"
assert parsed[0]["age_proposals"] == 0
assert parsed[0]["replay_evidence"]["replay_equivalent"] is True
def test_cli_show_command(tmp_path: Path):
log_path = tmp_path / "proposals.jsonl"
log = ProposalLog(log_path)
c1 = make_candidate("cand1", "light")
p1 = build_proposal(c1)
c2 = make_candidate("cand2", "dark")
p2 = build_proposal(c2)
log.record_created(p1)
log.record_created(p2)
runs_dir = tmp_path / "runs"
runs_dir.mkdir()
code, stdout, stderr = run_cli([
"teaching", "hitl-queue", "show", p1.proposal_id,
"--log-path", str(log_path),
"--contemplation-runs-dir", str(runs_dir)
])
assert code == 0
assert f"Proposal ID: {p1.proposal_id}" in stdout
assert "ADR References:" in stdout
prefix = p2.proposal_id[:12]
code_p, stdout_p, stderr_p = run_cli([
"teaching", "hitl-queue", "show", prefix,
"--log-path", str(log_path),
"--contemplation-runs-dir", str(runs_dir)
])
assert code_p == 0
assert f"Proposal ID: {p2.proposal_id}" in stdout_p
code_m, stdout_m, stderr_m = run_cli([
"teaching", "hitl-queue", "show", "nonexistent_id",
"--log-path", str(log_path),
"--contemplation-runs-dir", str(runs_dir)
])
assert code_m != 0
assert "error:" in stderr_m
assert "matches zero queue items" in stderr_m
common_prefix = ""
for char1, char2 in zip(p1.proposal_id, p2.proposal_id):
if char1 == char2:
common_prefix += char1
else:
break
if common_prefix:
code_a, stdout_a, stderr_a = run_cli([
"teaching", "hitl-queue", "show", common_prefix,
"--log-path", str(log_path),
"--contemplation-runs-dir", str(runs_dir)
])
assert code_a != 0
assert "error:" in stderr_a
assert "ambiguous" in stderr_a
# ---------------------------------------------------------------------------
# Blocker: legacy queue --threshold 3 CLI command validation
# ---------------------------------------------------------------------------
def test_cli_legacy_queue_command(tmp_path: Path):
gaps_dir = tmp_path / "gaps"
sink = gaps_dir / "2026" / "2026-05.jsonl"
sink.parent.mkdir(parents=True, exist_ok=True)
for i in range(3):
entry = {
"candidate_id": f"cand-{i}",
"proposed_chain": {
"subject": "knowledge",
"intent": "cause",
"connective": None,
"object": None,
},
"trigger": "would_have_grounded",
"source_turn_trace": f"trace-{i}",
"pack_consistent": True,
"boundary_clean": True,
"review_state": "unreviewed",
}
with sink.open("a", encoding="utf-8") as fh:
fh.write(json.dumps(entry) + "\n")
code, stdout, stderr = run_cli([
"teaching", "queue",
"--root", str(gaps_dir),
"--threshold", "3",
])
assert code == 0
assert "rank" in stdout
assert "queue_id" in stdout
assert "count" in stdout
assert "clean" in stdout
assert "gap:cause:knowledge@3" in stdout
# ---------------------------------------------------------------------------
# Read-only invariant
# ---------------------------------------------------------------------------
def snapshot_dir(directory: Path) -> dict[Path, bytes]:
snapshot = {}
if not directory.exists():
return snapshot
for path in directory.glob("**/*"):
if path.is_file():
snapshot[path] = path.read_bytes()
return snapshot
def test_read_only_invariant():
project_root = Path(__file__).resolve().parent.parent
dirs = [
project_root / "teaching" / "proposals",
project_root / "packs",
project_root / "engine_state",
project_root / "contemplation" / "runs",
]
before_snapshots = {}
for d in dirs:
before_snapshots[d] = snapshot_dir(d)
run_cli(["teaching", "hitl-queue", "list"])
run_cli(["teaching", "hitl-queue", "show", "nonexistent"])
for d in dirs:
after_snapshot = snapshot_dir(d)
assert after_snapshot == before_snapshots[d], f"Directory {d} was mutated!"
Reference in a new issue View git blame Copy permalink