From 72fac590293f675597553f10be8e40e6fdb4249e Mon Sep 17 00:00:00 2001 From: Shay Date: Tue, 26 May 2026 16:46:25 -0700 Subject: [PATCH] =?UTF-8?q?feat(ADR-0161.3):=20submission-time=20invariant?= =?UTF-8?q?s=20=E2=80=94=20duplicate=20+=20dependent=5Fon=5Fpending=20auto?= =?UTF-8?q?-reject=20(#313)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds two pre-gate checks to propose_from_candidate that fire after the Step 2 capacity check and before the replay gate. No log entry is written on either refusal — the append-only invariant holds. Check order at function entry (ADR-0161 §3): 1. Capacity (Step 2) → RefusedAtCapacity 2. Duplicate → RefusedAsDuplicate 3. Dependent_on_pending → RefusedAsDependent 4. Replay gate → auto-reject on regression New frozen dataclasses: @dataclass(frozen=True, slots=True) class RefusedAsDuplicate: proposal_id: str existing_state: str # covers all states: pending/accepted/rejected/withdrawn reason: str = "duplicate" @dataclass(frozen=True, slots=True) class RefusedAsDependent: candidate_id: str dependent_on: tuple[str, ...] # pending proposal_ids that block overlapping_lemmas: tuple[str, ...] # normalised lemmas that triggered reason: str = "dependent_on_pending" Lemma-overlap rule: case-insensitive exact-match on strip().lower(). Conservative — over-reject rather than admit-with-hidden-dependency. False positives are recoverable (re-emit after blocker is ratified); false negatives silently couple ratification choices. CLI surfaces both outcomes in cmd_teaching_propose and cmd_teaching_propose_from_exemplars (exit code 1). Step 2 backpressure tests updated: made pre-populated candidates use unique objects to avoid triggering the new dependency check, and updated idempotency assertions to reflect the new RefusedAsDuplicate return for re-submitted content. Co-references: ADR-0161 §3, Step 1 PR #296, Step 2 PR #311, ADR-0057, ADR-0151. --- core/cli.py | 24 +- docs/decisions/ADR-0161-hitl-async-queue.md | 48 ++- docs/hitl-backpressure.md | 61 +++ teaching/proposals.py | 66 ++- tests/test_hitl_queue_backpressure.py | 51 ++- .../test_hitl_queue_submission_invariants.py | 383 ++++++++++++++++++ 6 files changed, 593 insertions(+), 40 deletions(-) create mode 100644 tests/test_hitl_queue_submission_invariants.py diff --git a/core/cli.py b/core/cli.py index 9349dbc4..6a17a251 100644 --- a/core/cli.py +++ b/core/cli.py @@ -1352,7 +1352,8 @@ def _load_candidate_jsonl(path: str) -> Any: def cmd_teaching_propose(args: argparse.Namespace) -> int: """ADR-0057 Phase C2 — build a proposal from an enriched candidate JSONL.""" from teaching.proposals import ( - ProposalError, ProposalLog, propose_from_candidate, + ProposalError, ProposalLog, RefusedAsDependent, RefusedAsDuplicate, + RefusedAtCapacity, propose_from_candidate, ) candidate = _load_candidate_jsonl(args.candidate_path) @@ -1365,7 +1366,6 @@ 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) @@ -1379,6 +1379,15 @@ def cmd_teaching_propose(args: argparse.Namespace) -> int: print(f"report_written: {rel_path}") return 1 + if isinstance(proposal, RefusedAsDuplicate): + print(f"duplicate: proposal_id={proposal.proposal_id} existing_state={proposal.existing_state}") + return 1 + + if isinstance(proposal, RefusedAsDependent): + print(f"dependent_on_pending: dependent_on={list(proposal.dependent_on)}") + print(f"overlapping_lemmas={list(proposal.overlapping_lemmas)}") + return 1 + rec = log.find(proposal.proposal_id) print(f"proposal_id : {proposal.proposal_id}") print(f"state : {rec['state']}") @@ -1469,7 +1478,7 @@ def cmd_teaching_propose_from_exemplars(args: argparse.Namespace) -> int: code=1, ) - from teaching.proposals import RefusedAtCapacity + from teaching.proposals import RefusedAsDependent, RefusedAsDuplicate, RefusedAtCapacity if isinstance(proposal, RefusedAtCapacity): try: rel_path = proposal.report_path.relative_to(_REPO_ROOT) @@ -1483,6 +1492,15 @@ def cmd_teaching_propose_from_exemplars(args: argparse.Namespace) -> int: print(f"report_written: {rel_path}") return 1 + if isinstance(proposal, RefusedAsDuplicate): + print(f"duplicate: proposal_id={proposal.proposal_id} existing_state={proposal.existing_state}") + return 1 + + if isinstance(proposal, RefusedAsDependent): + print(f"dependent_on_pending: dependent_on={list(proposal.dependent_on)}") + print(f"overlapping_lemmas={list(proposal.overlapping_lemmas)}") + 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 220d506b..dd261013 100644 --- a/docs/decisions/ADR-0161-hitl-async-queue.md +++ b/docs/decisions/ADR-0161-hitl-async-queue.md @@ -378,15 +378,45 @@ existing recorded queue history: ### Step 3 — Submission-time invariants -- `propose_from_candidate` rejects duplicate `proposal_id` with reason - `duplicate` (already enforced; this step adds the explicit recorded - reason). -- `propose_from_candidate` rejects `dependent_on_pending` proposals. - Heuristic: chain whose `subject` or `object` lemma is a substring - of any pending proposal's `proposed_chain` is considered dependent; - the conservative check fails loud and the proposal is re-emitted - after the dependency lands. -- Tests for both rejection paths. +**Landed in `feat(ADR-0161.3): submission-time invariants — duplicate + +dependent_on_pending auto-reject`.** + +Two pre-gate checks added to `propose_from_candidate` in +`teaching/proposals.py`, firing in this order (after the Step 2 cap check): + +1. **Duplicate check** — computes the deterministic `proposal_id` and scans + `derive_queue()` for any existing item with the same id. If found, + returns `RefusedAsDuplicate(proposal_id, existing_state)`. Covers all + states (pending, accepted, rejected, withdrawn). + +2. **Dependent_on_pending check** — walks all pending queue items; if any + pending item's `proposed_chain.subject` or `.object` lemma matches the + candidate's subject or object (case-insensitive exact-match), returns + `RefusedAsDependent(candidate_id, dependent_on, overlapping_lemmas)`. + Conservative: over-reject rather than admit-with-hidden-dependency. + +Neither refusal writes to `proposals.jsonl`. The append-only invariant holds. + +New frozen dataclasses exported from `teaching/proposals.py`: + +```python +@dataclass(frozen=True, slots=True) +class RefusedAsDuplicate: + proposal_id: str + existing_state: str + reason: str = "duplicate" + +@dataclass(frozen=True, slots=True) +class RefusedAsDependent: + candidate_id: str + dependent_on: tuple[str, ...] + overlapping_lemmas: tuple[str, ...] + reason: str = "dependent_on_pending" +``` + +CLI surfaces both in `cmd_teaching_propose` and +`cmd_teaching_propose_from_exemplars` with exit code 1. Tests in +`tests/test_hitl_queue_submission_invariants.py`. ### Step 4 — Extend ratification workflow to reject/withdraw diff --git a/docs/hitl-backpressure.md b/docs/hitl-backpressure.md index cf15b58c..af90b078 100644 --- a/docs/hitl-backpressure.md +++ b/docs/hitl-backpressure.md @@ -53,3 +53,64 @@ When the queue is full and proposals are skipped: 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`. + +## Submission invariants + +Before the replay gate runs, `propose_from_candidate` applies two additional +content-based checks (ADR-0161 §3, Step 3). They fire in this order, after +the capacity check: + +### Duplicate + +A candidate whose deterministic `proposal_id` (SHA-256 over `candidate_id + +proposed_chain`, per ADR-0151) already exists in the log is refused with +`RefusedAsDuplicate`. This covers all existing states — pending, accepted, +rejected, and withdrawn — because content-identical proposals carry the same +id regardless of their history. + +CLI output: + +``` +duplicate: proposal_id= existing_state= +``` + +**No log entry is written.** The refusal is operator-facing only. + +### Dependent on pending + +A candidate whose `proposed_chain.subject` or `.object` lemma (case-insensitive +exact-match) overlaps with any **pending** proposal's subject or object is +refused with `RefusedAsDependent`. This prevents ratification-ordering +constraints from being silently baked into the queue. + +CLI output: + +``` +dependent_on_pending: dependent_on=[] +overlapping_lemmas=[] +``` + +**No log entry is written.** The operator should re-emit the candidate after +the dependency proposal is ratified. + +**Conservatism trade-off**: the heuristic uses exact-match on the normalised +lemma string (`strip().lower()`). A genuinely independent chain that happens +to share a common lemma word (e.g. "truth") will be refused. This is +intentional: false positives are recoverable (re-emit after the blocking +proposal clears); false negatives silently couple ratification choices. If +over-rejection becomes frequent, the operator should ratify or withdraw the +blocking pending proposals rather than loosening the heuristic. + +### Check order summary + +``` +capacity check (Step 2) ← queue_full report, no log entry + ↓ (if under cap) +duplicate check (Step 3) ← RefusedAsDuplicate, no log entry + ↓ (if not duplicate) +dependent_on_pending (Step 3) ← RefusedAsDependent, no log entry + ↓ (if no dependency) +replay gate ← runs; regression auto-rejects via transition + ↓ (if replay-equivalent) +pending proposal created ← created event appended to proposals.jsonl +``` diff --git a/teaching/proposals.py b/teaching/proposals.py index a07fa60c..ac36e77e 100644 --- a/teaching/proposals.py +++ b/teaching/proposals.py @@ -129,6 +129,21 @@ class RefusedAtCapacity: report_path: Path +@dataclass(frozen=True, slots=True) +class RefusedAsDuplicate: + proposal_id: str + existing_state: str + reason: str = "duplicate" + + +@dataclass(frozen=True, slots=True) +class RefusedAsDependent: + candidate_id: str + dependent_on: tuple[str, ...] + overlapping_lemmas: tuple[str, ...] + reason: str = "dependent_on_pending" + + 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.""" @@ -467,7 +482,7 @@ def propose_from_candidate( allow_evaluative: bool = False, source: ProposalSource | None = None, cap: int | None = None, -) -> TeachingChainProposal | RefusedAtCapacity: +) -> TeachingChainProposal | RefusedAtCapacity | RefusedAsDuplicate | RefusedAsDependent: """End-to-end: build proposal, run replay-equivalence gate, auto-reject on regression, otherwise leave pending. @@ -476,17 +491,18 @@ def propose_from_candidate( keeps tests fast — they can pass a fake that returns a stub ``ReplayEvidence`` without booting the cognition lane. - Idempotent on (candidate_id, chain): re-proposing returns the - existing proposal record if any. + Submission-time checks fire in this order (ADR-0161 §3): + 1. Capacity (Step 2) — queue_full if pending_count >= cap + 2. Duplicate — RefusedAsDuplicate if proposal_id already in log + 3. Dependent_on_pending — RefusedAsDependent if any pending item + shares a subject or object lemma with this candidate + Then the replay gate runs. No log entry is written on refusal. """ proposal = build_proposal( candidate, allow_evaluative=allow_evaluative, source=source, ) - 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 @@ -550,6 +566,42 @@ def propose_from_candidate( cap=resolved_cap, report_path=report_path, ) + + # Step 3a: duplicate check — proposal_id already exists in the log + for item in queue_items: + if item.proposal_id == proposal.proposal_id: + return RefusedAsDuplicate( + proposal_id=proposal.proposal_id, + existing_state=item.state, + ) + + # Step 3b: dependent_on_pending check — conservative lemma-overlap + # Case-insensitive exact-match; over-reject rather than admit-with-dependency. + candidate_subject = (proposal.proposed_chain.get("subject") or "").strip().lower() + candidate_object = (proposal.proposed_chain.get("object") or "").strip().lower() + candidate_lemmas = {lem for lem in (candidate_subject, candidate_object) if lem} + + blocking_ids: list[str] = [] + blocking_lemmas: set[str] = set() + for item in queue_items: + if item.state != "pending": + continue + chain = item.proposed_chain or {} + item_subject = (chain.get("subject") or "").strip().lower() + item_object = (chain.get("object") or "").strip().lower() + item_lemmas = {lem for lem in (item_subject, item_object) if lem} + overlap = candidate_lemmas & item_lemmas + if overlap: + blocking_ids.append(item.proposal_id) + blocking_lemmas.update(overlap) + + if blocking_ids: + return RefusedAsDependent( + candidate_id=candidate.candidate_id, + dependent_on=tuple(blocking_ids), + overlapping_lemmas=tuple(sorted(blocking_lemmas)), + ) + log.record_created(proposal) if run_replay is None: @@ -638,6 +690,8 @@ __all__ = [ "DEFAULT_PROPOSAL_LOG_PATH", "ProposalError", "ProposalLog", + "RefusedAsDependent", + "RefusedAsDuplicate", "RefusedAtCapacity", "ReplayEvidence", "ReviewState", diff --git a/tests/test_hitl_queue_backpressure.py b/tests/test_hitl_queue_backpressure.py index d5abe0b6..b5463376 100644 --- a/tests/test_hitl_queue_backpressure.py +++ b/tests/test_hitl_queue_backpressure.py @@ -16,6 +16,7 @@ from teaching.proposals import ( ProposalLog, RefusedAtCapacity, ReplayEvidence, + TeachingChainProposal, build_proposal, propose_from_candidate, ) @@ -71,24 +72,25 @@ def test_capacity_boundaries(tmp_path: Path): runs_dir = tmp_path / "runs" runs_dir.mkdir() - # Pre-populate log with 2 pending proposals + # Pre-populate log with 2 pending proposals; unique objects to avoid + # the dependent_on_pending check introduced in Step 3. for i in range(2): - c = make_candidate(f"cand_{i}", f"subject_{i}") + c = make_candidate(f"cand_{i}", f"subject_{i}", obj=f"object_{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") + c3 = make_candidate("cand_2", "subject_2", obj="object_2") res3 = propose_from_candidate( c3, log=log, run_replay=_fake_replay_equivalent, cap=3 ) - assert not isinstance(res3, RefusedAtCapacity) + assert isinstance(res3, TeachingChainProposal) 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") + c4 = make_candidate("cand_3", "subject_3", obj="object_3") res4 = propose_from_candidate( c4, log=log, run_replay=_fake_replay_equivalent, cap=3 ) @@ -159,61 +161,66 @@ def test_explicit_cap_kwarg_overrides_env_var(tmp_path: Path): log_path = tmp_path / "proposals.jsonl" log = ProposalLog(log_path) + # Unique objects to avoid the dependent_on_pending check (Step 3). for i in range(2): - c = make_candidate(f"cand_{i}", f"subject_{i}") + c = make_candidate(f"cand_{i}", f"subject_{i}", obj=f"object_{i}") proposal = build_proposal(c) log.record_created(proposal) - c3 = make_candidate("cand_2", "subject_2") + c3 = make_candidate("cand_2", "subject_2", obj="object_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 isinstance(res, TeachingChainProposal) assert log.find(res.proposal_id)["state"] == "pending" def test_repropose_same_candidate_post_clearance(tmp_path: Path): + """Step 3 behavior: at-capacity duplicate → capacity refusal; under-cap duplicate + (even if accepted) → RefusedAsDuplicate. Unique objects used throughout to + avoid the dependent_on_pending check introduced in Step 3.""" 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") + c1 = make_candidate("cand_1", "subject_1", obj="object_1") res1 = propose_from_candidate( c1, log=log, run_replay=_fake_replay_equivalent, cap=2 ) - assert not isinstance(res1, RefusedAtCapacity) + assert isinstance(res1, TeachingChainProposal) proposal_id_original = res1.proposal_id - # 2. Add another pending to hit cap - c2 = make_candidate("cand_2", "subject_2") - propose_from_candidate( + # 2. Add another pending to hit cap (unique object avoids dependency check) + c2 = make_candidate("cand_2", "subject_2", obj="object_2") + res2 = propose_from_candidate( c2, log=log, run_replay=_fake_replay_equivalent, cap=2 ) + assert isinstance(res2, TeachingChainProposal) # 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. + # 3. Re-proposing candidate 1 at capacity: capacity check fires first (Step 3 + # order: cap → duplicate → dependency). Returns RefusedAtCapacity. + from teaching.proposals import RefusedAtCapacity as _RefusedAtCapacity 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 + assert isinstance(res1_again, _RefusedAtCapacity) # 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. + # Step 3: re-proposing the same content (same proposal_id) returns RefusedAsDuplicate + # regardless of the existing state. The duplicate check fires before the replay gate. + from teaching.proposals import RefusedAsDuplicate as _RefusedAsDuplicate res1_post_clear = propose_from_candidate( c1, log=log, run_replay=_fake_replay_equivalent, cap=2 ) - assert not isinstance(res1_post_clear, RefusedAtCapacity) + assert isinstance(res1_post_clear, _RefusedAsDuplicate) assert res1_post_clear.proposal_id == proposal_id_original + assert res1_post_clear.existing_state == "accepted" def test_queue_full_report_byte_stability(tmp_path: Path): diff --git a/tests/test_hitl_queue_submission_invariants.py b/tests/test_hitl_queue_submission_invariants.py new file mode 100644 index 00000000..efc2f64f --- /dev/null +++ b/tests/test_hitl_queue_submission_invariants.py @@ -0,0 +1,383 @@ +"""Tests for ADR-0161 Step 3: submission-time invariants. + +Covers duplicate + dependent_on_pending auto-rejection checks that fire +BEFORE the replay gate, in the order: capacity → duplicate → dependent. +""" + +from __future__ import annotations + +from pathlib import Path + +from teaching.discovery import DiscoveryCandidate, EvidencePointer +from teaching.proposals import ( + ProposalLog, + RefusedAsDependent, + RefusedAsDuplicate, + RefusedAtCapacity, + ReplayEvidence, + TeachingChainProposal, + build_proposal, + propose_from_candidate, +) + + +# --------------------------------------------------------------------------- +# Fixtures / helpers +# --------------------------------------------------------------------------- + + +def _make_candidate( + candidate_id: str, + subject: str, + obj: str = "truth", + intent: str = "cause", + connective: str = "reveals", +) -> DiscoveryCandidate: + return DiscoveryCandidate( + candidate_id=candidate_id, + proposed_chain={ + "subject": subject, + "intent": intent, + "connective": connective, + "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_ok(_chain): # noqa: ANN001 + return ReplayEvidence( + baseline={"intent_accuracy": 1.0}, + candidate={"intent_accuracy": 1.0}, + regressed_metrics=(), + replay_equivalent=True, + ) + + +def _write_pending_proposal( + log: ProposalLog, + candidate_id: str, + subject: str, + obj: str = "truth", +) -> str: + """Inject a pending proposal directly into the log; returns proposal_id.""" + candidate = _make_candidate(candidate_id, subject, obj) + proposal = build_proposal(candidate) + log.record_created(proposal) + return proposal.proposal_id + + +def _write_accepted_proposal( + log: ProposalLog, + candidate_id: str, + subject: str, + obj: str = "truth", +) -> str: + """Inject a proposal in accepted state; returns proposal_id.""" + pid = _write_pending_proposal(log, candidate_id, subject, obj) + log.record_transition(pid, "accepted", "accepted in test") + return pid + + +# --------------------------------------------------------------------------- +# Duplicate: pending state +# --------------------------------------------------------------------------- + + +def test_duplicate_pending(tmp_path: Path) -> None: + """Re-submitting a candidate whose proposal_id is already pending → RefusedAsDuplicate.""" + log = ProposalLog(tmp_path / "proposals.jsonl") + runs_dir = tmp_path / "runs" + runs_dir.mkdir() + + # First submission lands successfully. + c1 = _make_candidate("cand-1", "light") + result1 = propose_from_candidate(c1, log=log, run_replay=_fake_replay_ok, cap=100) + assert isinstance(result1, TeachingChainProposal) + + # Second submission of identical content → RefusedAsDuplicate. + c1_again = _make_candidate("cand-1", "light") + result2 = propose_from_candidate(c1_again, log=log, run_replay=_fake_replay_ok, cap=100) + assert isinstance(result2, RefusedAsDuplicate) + assert result2.proposal_id == result1.proposal_id + assert result2.existing_state == "pending" + assert result2.reason == "duplicate" + + +# --------------------------------------------------------------------------- +# Duplicate: accepted state +# --------------------------------------------------------------------------- + + +def test_duplicate_after_accept(tmp_path: Path) -> None: + """proposal_id in accepted state → RefusedAsDuplicate (idempotent for all states).""" + log = ProposalLog(tmp_path / "proposals.jsonl") + runs_dir = tmp_path / "runs" + runs_dir.mkdir() + + pid = _write_accepted_proposal(log, "cand-accept", "memory") + + candidate = _make_candidate("cand-accept", "memory") + result = propose_from_candidate(candidate, log=log, run_replay=_fake_replay_ok, cap=100) + assert isinstance(result, RefusedAsDuplicate) + assert result.proposal_id == pid + assert result.existing_state == "accepted" + assert result.reason == "duplicate" + + +# --------------------------------------------------------------------------- +# Dependent_on_pending: subject overlap +# --------------------------------------------------------------------------- + + +def test_dependent_on_pending_subject_overlap(tmp_path: Path) -> None: + """Candidate whose chain subject matches a pending proposal's subject → RefusedAsDependent.""" + log = ProposalLog(tmp_path / "proposals.jsonl") + runs_dir = tmp_path / "runs" + runs_dir.mkdir() + + # Proposal A is pending with subject="light". + pid_a = _write_pending_proposal(log, "cand-A", "light", "order") + + # Candidate B shares subject="light". + c_b = _make_candidate("cand-B", "light", "truth") + result = propose_from_candidate(c_b, log=log, run_replay=_fake_replay_ok, cap=100) + + assert isinstance(result, RefusedAsDependent) + assert pid_a in result.dependent_on + assert "light" in result.overlapping_lemmas + assert result.reason == "dependent_on_pending" + + +def test_dependent_on_pending_object_overlap(tmp_path: Path) -> None: + """Candidate whose chain object matches a pending proposal's object → RefusedAsDependent.""" + log = ProposalLog(tmp_path / "proposals.jsonl") + runs_dir = tmp_path / "runs" + runs_dir.mkdir() + + # Proposal A: object="knowledge" + pid_a = _write_pending_proposal(log, "cand-A", "thought", "knowledge") + + # Candidate B: object="knowledge" (from different subject) + c_b = _make_candidate("cand-B", "reason", "knowledge") + result = propose_from_candidate(c_b, log=log, run_replay=_fake_replay_ok, cap=100) + + assert isinstance(result, RefusedAsDependent) + assert pid_a in result.dependent_on + assert "knowledge" in result.overlapping_lemmas + + +def test_dependent_on_pending_cross_overlap(tmp_path: Path) -> None: + """Candidate whose subject matches pending item's object → RefusedAsDependent.""" + log = ProposalLog(tmp_path / "proposals.jsonl") + runs_dir = tmp_path / "runs" + runs_dir.mkdir() + + # Proposal A: subject="perception", object="light" + pid_a = _write_pending_proposal(log, "cand-A", "perception", "light") + + # Candidate B: subject="light" (same as A's object) + c_b = _make_candidate("cand-B", "light", "clarity") + result = propose_from_candidate(c_b, log=log, run_replay=_fake_replay_ok, cap=100) + + assert isinstance(result, RefusedAsDependent) + assert pid_a in result.dependent_on + assert "light" in result.overlapping_lemmas + + +# --------------------------------------------------------------------------- +# Dependent_on_pending: accepted state does NOT block +# --------------------------------------------------------------------------- + + +def test_no_block_when_dependency_accepted(tmp_path: Path) -> None: + """Accepted proposal does not trigger dependent_on_pending check.""" + log = ProposalLog(tmp_path / "proposals.jsonl") + runs_dir = tmp_path / "runs" + runs_dir.mkdir() + + # Proposal A is accepted (not pending). + _write_accepted_proposal(log, "cand-A", "light", "order") + + # Candidate B with overlapping lemma should land. + c_b = _make_candidate("cand-B", "light", "truth") + result = propose_from_candidate(c_b, log=log, run_replay=_fake_replay_ok, cap=100) + + assert isinstance(result, TeachingChainProposal) + + +# --------------------------------------------------------------------------- +# No false positives: disjoint lemmas +# --------------------------------------------------------------------------- + + +def test_no_false_positives_disjoint_lemmas(tmp_path: Path) -> None: + """Two candidates with completely disjoint lemmas both submit successfully.""" + log = ProposalLog(tmp_path / "proposals.jsonl") + runs_dir = tmp_path / "runs" + runs_dir.mkdir() + + c1 = _make_candidate("cand-1", "fire", "heat") + r1 = propose_from_candidate(c1, log=log, run_replay=_fake_replay_ok, cap=100) + assert isinstance(r1, TeachingChainProposal) + + c2 = _make_candidate("cand-2", "water", "flow") + r2 = propose_from_candidate(c2, log=log, run_replay=_fake_replay_ok, cap=100) + assert isinstance(r2, TeachingChainProposal) + + +# --------------------------------------------------------------------------- +# Case-insensitive overlap +# --------------------------------------------------------------------------- + + +def test_case_insensitive_overlap(tmp_path: Path) -> None: + """Lemma matching is case-insensitive.""" + log = ProposalLog(tmp_path / "proposals.jsonl") + runs_dir = tmp_path / "runs" + runs_dir.mkdir() + + # Proposal A with subject="Light" (mixed case). + pid_a = _write_pending_proposal(log, "cand-A", "Light", "order") + + # Candidate B with subject="LIGHT". + c_b = _make_candidate("cand-B", "LIGHT", "clarity") + result = propose_from_candidate(c_b, log=log, run_replay=_fake_replay_ok, cap=100) + + assert isinstance(result, RefusedAsDependent) + assert pid_a in result.dependent_on + + +# --------------------------------------------------------------------------- +# cap-then-duplicate ordering: capacity refusal wins +# --------------------------------------------------------------------------- + + +def test_cap_beats_duplicate_ordering(tmp_path: Path) -> None: + """When queue is full AND proposal is a duplicate, capacity refusal wins.""" + log = ProposalLog(tmp_path / "proposals.jsonl") + runs_dir = tmp_path / "runs" + runs_dir.mkdir() + + # Pre-populate log with 3 pending proposals to fill a cap=3 queue. + for i in range(3): + _write_pending_proposal(log, f"pre-{i}", f"subject_{i}", f"object_{i}") + + # First submission of "cand-dup" lands (cap=4, so it fits). + c_first = _make_candidate("cand-dup", "unique_subject_x", "unique_object_x") + r_first = propose_from_candidate(c_first, log=log, run_replay=_fake_replay_ok, cap=4) + assert isinstance(r_first, TeachingChainProposal) + + # Now queue has 4 pending items. Re-submit same candidate with cap=4 still full. + c_dup = _make_candidate("cand-dup", "unique_subject_x", "unique_object_x") + result = propose_from_candidate(c_dup, log=log, run_replay=_fake_replay_ok, cap=4) + # Cap fires first: 4 pending >= cap 4. + assert isinstance(result, RefusedAtCapacity) + + +# --------------------------------------------------------------------------- +# Empty log: both checks pass, first proposal lands +# --------------------------------------------------------------------------- + + +def test_empty_log_first_proposal_lands(tmp_path: Path) -> None: + """With an empty log, duplicate and dependency checks both pass cleanly.""" + log = ProposalLog(tmp_path / "proposals.jsonl") + runs_dir = tmp_path / "runs" + runs_dir.mkdir() + + candidate = _make_candidate("cand-first", "thought", "form") + result = propose_from_candidate(candidate, log=log, run_replay=_fake_replay_ok, cap=100) + assert isinstance(result, TeachingChainProposal) + assert result.proposal_id is not None + + +# --------------------------------------------------------------------------- +# Append-only invariant: no log entry written on refusal +# --------------------------------------------------------------------------- + + +def test_duplicate_refusal_writes_no_log_entry(tmp_path: Path) -> None: + """RefusedAsDuplicate must not append anything to proposals.jsonl.""" + log = ProposalLog(tmp_path / "proposals.jsonl") + runs_dir = tmp_path / "runs" + runs_dir.mkdir() + + c = _make_candidate("cand-x", "entropy") + propose_from_candidate(c, log=log, run_replay=_fake_replay_ok, cap=100) + count_before = len(log.events()) + + c_again = _make_candidate("cand-x", "entropy") + result = propose_from_candidate(c_again, log=log, run_replay=_fake_replay_ok, cap=100) + assert isinstance(result, RefusedAsDuplicate) + assert len(log.events()) == count_before + + +def test_dependent_refusal_writes_no_log_entry(tmp_path: Path) -> None: + """RefusedAsDependent must not append anything to proposals.jsonl.""" + log = ProposalLog(tmp_path / "proposals.jsonl") + runs_dir = tmp_path / "runs" + runs_dir.mkdir() + + _write_pending_proposal(log, "cand-A", "wave", "particle") + count_before = len(log.events()) + + c_b = _make_candidate("cand-B", "wave", "energy") + result = propose_from_candidate(c_b, log=log, run_replay=_fake_replay_ok, cap=100) + assert isinstance(result, RefusedAsDependent) + assert len(log.events()) == count_before + + +# --------------------------------------------------------------------------- +# Determinism: same log + same candidate → same outcome +# --------------------------------------------------------------------------- + + +def test_determinism_duplicate_check(tmp_path: Path) -> None: + """Duplicate check is deterministic: same inputs produce same outcome.""" + log = ProposalLog(tmp_path / "proposals.jsonl") + runs_dir = tmp_path / "runs" + runs_dir.mkdir() + + c = _make_candidate("cand-det", "determinism") + propose_from_candidate(c, log=log, run_replay=_fake_replay_ok, cap=100) + + results = [] + for _ in range(3): + c_again = _make_candidate("cand-det", "determinism") + results.append( + propose_from_candidate(c_again, log=log, run_replay=_fake_replay_ok, cap=100) + ) + assert all(isinstance(r, RefusedAsDuplicate) for r in results) + assert len({r.proposal_id for r in results}) == 1 + + +def test_determinism_dependency_check(tmp_path: Path) -> None: + """Dependency check is deterministic: same log + same candidate → same outcome.""" + log = ProposalLog(tmp_path / "proposals.jsonl") + runs_dir = tmp_path / "runs" + runs_dir.mkdir() + + pid_a = _write_pending_proposal(log, "cand-A", "chaos", "order") + + results = [] + for i in range(3): + c_b = _make_candidate(f"cand-B-{i}", "chaos", "clarity") + results.append( + propose_from_candidate(c_b, log=log, run_replay=_fake_replay_ok, cap=100) + ) + assert all(isinstance(r, RefusedAsDependent) for r in results) + assert all(pid_a in r.dependent_on for r in results)