From 76032db9a0b8b30b6897aa171c01c43b7a0adcca Mon Sep 17 00:00:00 2001 From: Shay Date: Tue, 26 May 2026 16:16:08 -0700 Subject: [PATCH] =?UTF-8?q?feat(ADR-0161.2):=20HITL=20queue=20backpressure?= =?UTF-8?q?=20=E2=80=94=20pending-count=20cap=20+=20queue=5Ffull=20reports?= =?UTF-8?q?=20(#311)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- core/cli.py | 30 ++ docs/decisions/ADR-0161-hitl-async-queue.md | 7 +- docs/hitl-backpressure.md | 55 ++++ teaching/proposals.py | 81 +++++- teaching/queue.py | 34 ++- tests/test_hitl_queue_backpressure.py | 298 ++++++++++++++++++++ 6 files changed, 489 insertions(+), 16 deletions(-) create mode 100644 docs/hitl-backpressure.md create mode 100644 tests/test_hitl_queue_backpressure.py diff --git a/core/cli.py b/core/cli.py index aa2ba062..9349dbc4 100644 --- a/core/cli.py +++ b/core/cli.py @@ -1364,6 +1364,21 @@ def cmd_teaching_propose(args: argparse.Namespace) -> int: ) except ProposalError as exc: _die(f"ineligible: {exc}", code=1) + + from teaching.proposals import RefusedAtCapacity + if isinstance(proposal, RefusedAtCapacity): + try: + rel_path = proposal.report_path.relative_to(_REPO_ROOT) + except ValueError: + try: + rel_path = proposal.report_path.relative_to(Path.cwd()) + except ValueError: + rel_path = proposal.report_path + print(f"queue_full: pending={proposal.pending_count}, cap={proposal.cap}") + print("candidates_skipped: 1") + print(f"report_written: {rel_path}") + return 1 + rec = log.find(proposal.proposal_id) print(f"proposal_id : {proposal.proposal_id}") print(f"state : {rec['state']}") @@ -1453,6 +1468,21 @@ def cmd_teaching_propose_from_exemplars(args: argparse.Namespace) -> int: f"ineligible candidate for {corpus.shape_category.value}: {exc}", code=1, ) + + from teaching.proposals import RefusedAtCapacity + if isinstance(proposal, RefusedAtCapacity): + try: + rel_path = proposal.report_path.relative_to(_REPO_ROOT) + except ValueError: + try: + rel_path = proposal.report_path.relative_to(Path.cwd()) + except ValueError: + rel_path = proposal.report_path + print(f"queue_full: pending={proposal.pending_count}, cap={proposal.cap}") + print("candidates_skipped: 1") + print(f"report_written: {rel_path}") + return 1 + rec = log.find(proposal.proposal_id) result = { "shape_category": corpus.shape_category.value, diff --git a/docs/decisions/ADR-0161-hitl-async-queue.md b/docs/decisions/ADR-0161-hitl-async-queue.md index fb7498ec..220d506b 100644 --- a/docs/decisions/ADR-0161-hitl-async-queue.md +++ b/docs/decisions/ADR-0161-hitl-async-queue.md @@ -364,9 +364,10 @@ existing recorded queue history: ### Step 2 — Backpressure (pending-count cap) -- `propose_from_candidate` consults pending count via - `teaching/queue.derive_queue` and emits a `queue_full` report - instead of a new proposal when the cap is reached. +- `propose_from_candidate` in `teaching/proposals.py` consults pending count via + `teaching.queue.derive_queue` and writes a `queue_full` report at + `contemplation/runs/_queue_full.json` instead of a new proposal + when the cap is reached, returning `RefusedAtCapacity`. - `contemplation/runs/.json` schema extended with `report_kind ∈ {"learning_arc", "queue_full"}` (default `"learning_arc"` for back-compat). diff --git a/docs/hitl-backpressure.md b/docs/hitl-backpressure.md new file mode 100644 index 00000000..cf15b58c --- /dev/null +++ b/docs/hitl-backpressure.md @@ -0,0 +1,55 @@ +# HITL review queue backpressure + +To prevent reviewed proposal queues from growing beyond human attention limits, the queue enforces a pending count cap (ADR-0161 §4). + +## Configuration + +- **Default Cap**: 256 pending proposals. +- **Environment Override**: To temporarily raise the cap, set the `CORE_HITL_PENDING_CAP` environment variable: + ```bash + export CORE_HITL_PENDING_CAP=512 + ``` + +Accepted, rejected, and withdrawn proposals do not count toward the cap. + +## Queue Full Reports + +When the pending count is at or above the cap, `propose_from_candidate` will refuse to create a new proposal and instead emit a `queue_full` report to the contemplation runs directory: +`contemplation/runs/_queue_full.json` + +Example `queue_full` report shape: +```json +{ + "report_kind": "queue_full", + "emitted_at_revision": "9a738a16...", + "pending_count": 256, + "cap": 256, + "candidates_skipped": [ + { + "candidate_id": "8a9b2c3d...", + "shape_category": "factual", + "reason": "queue_full" + } + ] +} +``` + +The CLI exits with code `1` when capacity is refused, alerting CI runners and operators of the backpressure event. + +## Operator Clearance Loop + +When the queue is full and proposals are skipped: + +1. **Review Pending Proposals**: Inspect the current queue using: + ```bash + core teaching hitl-queue list --state pending + ``` +2. **Clear Space**: Ratify or withdraw pending proposals: + ```bash + # Accept a proposal + core teaching review --accept --review-date 2026-05-26 + + # Or withdraw a proposal + core teaching review --withdraw + ``` +3. **Re-run Propose**: Once pending count falls below the cap, re-run the propose command. Skipped candidates will land successfully as fresh proposals with the same deterministic `proposal_id`. diff --git a/teaching/proposals.py b/teaching/proposals.py index c68f53a8..a07fa60c 100644 --- a/teaching/proposals.py +++ b/teaching/proposals.py @@ -45,6 +45,10 @@ if TYPE_CHECKING: DEFAULT_PROPOSAL_LOG_PATH: Path = ( Path(__file__).resolve().parent / "proposals" / "proposals.jsonl" ) +DEFAULT_PENDING_CAP: int = 256 +DEFAULT_CONTEMPLATION_RUNS_DIR: Path = ( + Path(__file__).resolve().parent.parent / "contemplation" / "runs" +) ReviewState = Literal["pending", "accepted", "rejected", "withdrawn"] @@ -116,6 +120,15 @@ class TeachingChainProposal: } +@dataclass(frozen=True, slots=True) +class RefusedAtCapacity: + candidate_id: str + shape_category: str + pending_count: int + cap: int + report_path: Path + + class ProposalError(ValueError): """Raised when a candidate fails an eligibility gate or when a review action is attempted in a state that does not allow it.""" @@ -453,7 +466,8 @@ def propose_from_candidate( run_replay: Any = None, allow_evaluative: bool = False, source: ProposalSource | None = None, -) -> TeachingChainProposal: + cap: int | None = None, +) -> TeachingChainProposal | RefusedAtCapacity: """End-to-end: build proposal, run replay-equivalence gate, auto-reject on regression, otherwise leave pending. @@ -473,6 +487,69 @@ def propose_from_candidate( existing = log.find(proposal.proposal_id) if existing is not None: return proposal + + if log.path == DEFAULT_PROPOSAL_LOG_PATH: + contemplation_runs_dir = DEFAULT_CONTEMPLATION_RUNS_DIR + else: + if (log.path.parent / "runs").exists(): + contemplation_runs_dir = log.path.parent / "runs" + elif (log.path.parent / "contemplation" / "runs").exists(): + contemplation_runs_dir = log.path.parent / "contemplation" / "runs" + else: + contemplation_runs_dir = log.path.parent / "runs" + + from teaching.queue import derive_queue + queue_items = derive_queue(log, contemplation_runs_dir) + pending_count = sum(1 for item in queue_items if item.state == "pending") + + resolved_cap = cap + if resolved_cap is None: + import os + env_val = os.environ.get("CORE_HITL_PENDING_CAP") + if env_val is not None: + try: + resolved_cap = int(env_val) + except ValueError: + resolved_cap = DEFAULT_PENDING_CAP + else: + resolved_cap = DEFAULT_PENDING_CAP + + if pending_count >= resolved_cap: + shape_category = ( + candidate.proposed_chain.get("recognizer_spec", {}).get("shape_category") + if candidate.proposed_chain + else None + ) or candidate.claim_domain + + from datetime import datetime, timezone + stamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H%M%SZ") + report_path = contemplation_runs_dir / f"{stamp}_queue_full.json" + + report_data = { + "report_kind": "queue_full", + "emitted_at_revision": _current_revision(), + "pending_count": pending_count, + "cap": resolved_cap, + "candidates_skipped": [ + { + "candidate_id": candidate.candidate_id, + "shape_category": shape_category, + "reason": "queue_full" + } + ] + } + contemplation_runs_dir.mkdir(parents=True, exist_ok=True) + report_path.write_text( + json.dumps(report_data, ensure_ascii=False, indent=2, sort_keys=True) + "\n", + encoding="utf-8" + ) + return RefusedAtCapacity( + candidate_id=candidate.candidate_id, + shape_category=shape_category, + pending_count=pending_count, + cap=resolved_cap, + report_path=report_path, + ) log.record_created(proposal) if run_replay is None: @@ -557,9 +634,11 @@ def withdraw_proposal( __all__ = [ + "DEFAULT_PENDING_CAP", "DEFAULT_PROPOSAL_LOG_PATH", "ProposalError", "ProposalLog", + "RefusedAtCapacity", "ReplayEvidence", "ReviewState", "TeachingChainProposal", diff --git a/teaching/queue.py b/teaching/queue.py index fe4eb58e..17677d06 100644 --- a/teaching/queue.py +++ b/teaching/queue.py @@ -45,19 +45,29 @@ def _load_contemplation_mapping(runs_dir: Path, runs_dir_mtime: float) -> dict[s continue pids: set[str] = set() - top_pid = data.get("proposal_id") - if isinstance(top_pid, str): - pids.add(top_pid) + report_kind = data.get("report_kind", "learning_arc") + if report_kind == "learning_arc": + top_pid = data.get("proposal_id") + if isinstance(top_pid, str): + pids.add(top_pid) - scenes = data.get("scenes") - if isinstance(scenes, list): - for scene in scenes: - if isinstance(scene, dict): - detail = scene.get("detail") - if isinstance(detail, dict): - scene_pid = detail.get("proposal_id") - if isinstance(scene_pid, str): - pids.add(scene_pid) + scenes = data.get("scenes") + if isinstance(scenes, list): + for scene in scenes: + if isinstance(scene, dict): + detail = scene.get("detail") + if isinstance(detail, dict): + scene_pid = detail.get("proposal_id") + if isinstance(scene_pid, str): + pids.add(scene_pid) + elif report_kind == "queue_full": + candidates_skipped = data.get("candidates_skipped", []) + if isinstance(candidates_skipped, list): + for cand in candidates_skipped: + if isinstance(cand, dict): + cand_id = cand.get("candidate_id") + if isinstance(cand_id, str): + pids.add(cand_id) for pid in pids: mapping[pid] = str(path.resolve()) diff --git a/tests/test_hitl_queue_backpressure.py b/tests/test_hitl_queue_backpressure.py new file mode 100644 index 00000000..d5abe0b6 --- /dev/null +++ b/tests/test_hitl_queue_backpressure.py @@ -0,0 +1,298 @@ +"""Tests for ADR-0161 Step 2: HITL review queue backpressure (cap + reports).""" + +from __future__ import annotations + +import json +import os +from datetime import datetime, timezone +from pathlib import Path +from unittest.mock import patch + +import pytest + +from teaching.discovery import DiscoveryCandidate, EvidencePointer +from teaching.proposals import ( + DEFAULT_PENDING_CAP, + ProposalLog, + RefusedAtCapacity, + ReplayEvidence, + build_proposal, + propose_from_candidate, +) +from teaching.queue import derive_queue + + +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 _fake_replay_equivalent(chain): + return ReplayEvidence( + baseline={"intent_accuracy": 1.0}, + candidate={"intent_accuracy": 1.0}, + regressed_metrics=(), + replay_equivalent=True, + ) + + +def test_default_pending_cap_is_pinned(): + """ADR-0161 §4: default pending cap is strictly 256. + + Any changes to this default require an ADR amendment. + """ + assert DEFAULT_PENDING_CAP == 256 + + +def test_capacity_boundaries(tmp_path: Path): + log_path = tmp_path / "proposals.jsonl" + log = ProposalLog(log_path) + runs_dir = tmp_path / "runs" + runs_dir.mkdir() + + # Pre-populate log with 2 pending proposals + for i in range(2): + c = make_candidate(f"cand_{i}", f"subject_{i}") + proposal = build_proposal(c) + log.record_created(proposal) + + # cap = 3. Current pending is 2 (which is cap - 1) + # The 3rd candidate should land successfully + c3 = make_candidate("cand_2", "subject_2") + res3 = propose_from_candidate( + c3, log=log, run_replay=_fake_replay_equivalent, cap=3 + ) + assert not isinstance(res3, RefusedAtCapacity) + assert log.find(res3.proposal_id)["state"] == "pending" + + # Current pending is now 3 (which is at cap). + # The 4th candidate should be refused + c4 = make_candidate("cand_3", "subject_3") + res4 = propose_from_candidate( + c4, log=log, run_replay=_fake_replay_equivalent, cap=3 + ) + assert isinstance(res4, RefusedAtCapacity) + assert res4.candidate_id == "cand_3" + assert res4.pending_count == 3 + assert res4.cap == 3 + assert res4.report_path.exists() + + # Verify queue_full report contents + report = json.loads(res4.report_path.read_text(encoding="utf-8")) + assert report["report_kind"] == "queue_full" + assert report["pending_count"] == 3 + assert report["cap"] == 3 + assert len(report["candidates_skipped"]) == 1 + assert report["candidates_skipped"][0]["candidate_id"] == "cand_3" + assert report["candidates_skipped"][0]["shape_category"] == "factual" + assert report["candidates_skipped"][0]["reason"] == "queue_full" + + # Verify that the 4th proposal was NOT written to the log + assert log.find(build_proposal(c4).proposal_id) is None + + +def test_over_cap_refuses_consistently(tmp_path: Path): + log_path = tmp_path / "proposals.jsonl" + log = ProposalLog(log_path) + + # Pre-populate log with 5 pending proposals + for i in range(5): + c = make_candidate(f"cand_{i}", f"subject_{i}") + proposal = build_proposal(c) + log.record_created(proposal) + + # With cap = 3 (current pending = 5), we are already over capacity + # It should refuse consistently + c_new = make_candidate("cand_new", "subject_new") + res = propose_from_candidate( + c_new, log=log, run_replay=_fake_replay_equivalent, cap=3 + ) + assert isinstance(res, RefusedAtCapacity) + assert res.pending_count == 5 + assert res.cap == 3 + + +def test_core_hitl_pending_cap_env_var_override(tmp_path: Path): + log_path = tmp_path / "proposals.jsonl" + log = ProposalLog(log_path) + + # Pre-populate log with 2 pending proposals + for i in range(2): + c = make_candidate(f"cand_{i}", f"subject_{i}") + proposal = build_proposal(c) + log.record_created(proposal) + + # Set env var CORE_HITL_PENDING_CAP = 2. Current pending = 2 + # The 3rd candidate should be refused + c3 = make_candidate("cand_2", "subject_2") + with patch.dict(os.environ, {"CORE_HITL_PENDING_CAP": "2"}): + res = propose_from_candidate( + c3, log=log, run_replay=_fake_replay_equivalent + ) + assert isinstance(res, RefusedAtCapacity) + assert res.cap == 2 + assert res.pending_count == 2 + + +def test_explicit_cap_kwarg_overrides_env_var(tmp_path: Path): + log_path = tmp_path / "proposals.jsonl" + log = ProposalLog(log_path) + + for i in range(2): + c = make_candidate(f"cand_{i}", f"subject_{i}") + proposal = build_proposal(c) + log.record_created(proposal) + + c3 = make_candidate("cand_2", "subject_2") + # env var is 2, but kwarg is 5. Should successfully propose + with patch.dict(os.environ, {"CORE_HITL_PENDING_CAP": "2"}): + res = propose_from_candidate( + c3, log=log, run_replay=_fake_replay_equivalent, cap=5 + ) + assert not isinstance(res, RefusedAtCapacity) + assert log.find(res.proposal_id)["state"] == "pending" + + +def test_repropose_same_candidate_post_clearance(tmp_path: Path): + log_path = tmp_path / "proposals.jsonl" + log = ProposalLog(log_path) + + # 1. Propose candidate when space is available (cap = 2, current pending = 0) + c1 = make_candidate("cand_1", "subject_1") + res1 = propose_from_candidate( + c1, log=log, run_replay=_fake_replay_equivalent, cap=2 + ) + assert not isinstance(res1, RefusedAtCapacity) + proposal_id_original = res1.proposal_id + + # 2. Add another pending to hit cap + c2 = make_candidate("cand_2", "subject_2") + propose_from_candidate( + c2, log=log, run_replay=_fake_replay_equivalent, cap=2 + ) + + # Now pending count is 2, cap is 2. + # 3. Re-proposing candidate 1 should return the existing proposal record (idempotency) + # even though we are technically at capacity, because it is already in the log. + res1_again = propose_from_candidate( + c1, log=log, run_replay=_fake_replay_equivalent, cap=2 + ) + assert not isinstance(res1_again, RefusedAtCapacity) + assert res1_again.proposal_id == proposal_id_original + + # 4. Transition candidate 1 to accepted (clearance) + log.record_transition(proposal_id_original, "accepted", "ratified") + + # Now pending count is 1 (candidate 2 is still pending, candidate 1 is accepted). + # Since pending (1) < cap (2), we can propose a new candidate or re-propose candidate 1 + # as a fresh proposal. Wait, re-proposing same candidate post-clearance: + # Under ADR-0151, proposal_id is content-deterministic. + # It lands as a fresh proposal with the SAME proposal_id. + res1_post_clear = propose_from_candidate( + c1, log=log, run_replay=_fake_replay_equivalent, cap=2 + ) + assert not isinstance(res1_post_clear, RefusedAtCapacity) + assert res1_post_clear.proposal_id == proposal_id_original + + +def test_queue_full_report_byte_stability(tmp_path: Path): + log_path = tmp_path / "proposals.jsonl" + log = ProposalLog(log_path) + runs_dir = tmp_path / "runs" + runs_dir.mkdir() + + # Pre-populate log to hit cap + c = make_candidate("cand_0", "subject_0") + proposal = build_proposal(c) + log.record_created(proposal) + + # Force a known revision and timestamp for byte-stability + with patch("teaching.proposals._current_revision", return_value="f00ba2"): + c1 = make_candidate("cand_1", "subject_1") + res1 = propose_from_candidate( + c1, log=log, run_replay=_fake_replay_equivalent, cap=1 + ) + assert isinstance(res1, RefusedAtCapacity) + report_text1 = res1.report_path.read_text(encoding="utf-8") + + # Delete the report and re-run with same parameters to verify byte-stable output + res1.report_path.unlink() + with patch("teaching.proposals._current_revision", return_value="f00ba2"): + res2 = propose_from_candidate( + c1, log=log, run_replay=_fake_replay_equivalent, cap=1 + ) + assert isinstance(res2, RefusedAtCapacity) + report_text2 = res2.report_path.read_text(encoding="utf-8") + + # The JSON string must be exactly identical + assert report_text1 == report_text2 + + # Check key ordering + parsed = json.loads(report_text1) + keys = list(parsed.keys()) + assert keys == sorted(keys) + + +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_capacity_refusal_does_not_mutate_system(): + """Verify capacity refusal does not mutate packs, corpus, or recognizer registry.""" + project_root = Path(__file__).resolve().parent.parent + dirs = [ + project_root / "teaching" / "cognition_chains", + project_root / "packs", + project_root / "language_packs" / "data", + ] + + before_snapshots = {} + for d in dirs: + before_snapshots[d] = snapshot_dir(d) + + # Trigger a capacity refusal + tmp_log = ProposalLog(project_root / "teaching" / "proposals" / "temp_proposals.jsonl") + c = make_candidate("cand_x", "subject_x") + res = propose_from_candidate( + c, log=tmp_log, run_replay=_fake_replay_equivalent, cap=0 + ) + assert isinstance(res, RefusedAtCapacity) + + # Clean up temp files if created in the directory + if tmp_log.path.exists(): + tmp_log.path.unlink() + # Clean up generated runs file + if res.report_path.exists(): + res.report_path.unlink() + + # Assert no mutations occurred in system folders + for d in dirs: + after_snapshot = snapshot_dir(d) + assert after_snapshot == before_snapshots[d], f"Directory {d} was mutated!"