feat(ADR-0161.2): HITL queue backpressure — pending-count cap + queue_full reports (#311)
This commit is contained in:
Shay
GitHub
parent
ac77b88864
commit
76032db9a0
6 changed files with 489 additions and 16 deletions
30
core/cli.py
30
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,
|
||||
|
|
|
|||
|
|
@ -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/<timestamp>_queue_full.json` instead of a new proposal
|
||||
when the cap is reached, returning `RefusedAtCapacity`.
|
||||
- `contemplation/runs/<timestamp>.json` schema extended with
|
||||
`report_kind ∈ {"learning_arc", "queue_full"}` (default
|
||||
`"learning_arc"` for back-compat).
|
||||
|
|
|
|||
55
docs/hitl-backpressure.md
Normal file
55
docs/hitl-backpressure.md
Normal file
|
|
@ -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/<ISO-8601-UTC>_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 <proposal_id> --accept --review-date 2026-05-26
|
||||
|
||||
# Or withdraw a proposal
|
||||
core teaching review <proposal_id> --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`.
|
||||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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())
|
||||
|
|
|
|||
298
tests/test_hitl_queue_backpressure.py
Normal file
298
tests/test_hitl_queue_backpressure.py
Normal file
|
|
@ -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!"
|
||||
Loading…
Reference in a new issue