From e03ab4b60972eb7343aa27cee825c637fcff44ec Mon Sep 17 00:00:00 2001 From: Shay Date: Mon, 18 May 2026 10:23:14 -0700 Subject: [PATCH] =?UTF-8?q?feat(adr-0057):=20Phase=20C2=20=E2=80=94=20Teac?= =?UTF-8?q?hingChainProposal=20+=20replay=20gate=20+=20review=20CLI?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The only path by which CORE extends its own active teaching corpus. Closes ADR-0055 Phase C alongside ADR-0056's cognitive surface. Three load-bearing calls (recorded in ADR-0057): 1. Replay-equivalence is a precondition, not a permission; operator --accept remains required. 2. Eligibility = polarity in {affirms, falsifies} AND at least one source='corpus' evidence pointer AND boundary_clean AND claim_domain != evaluative (unless --allow-evaluative) AND proposed_chain complete. 3. Append-only proposal log; corpus history append-only too. Changes - teaching/proposals.py — TeachingChainProposal, ReplayEvidence, ProposalLog (event-sourced replay → current_state), eligibility predicate, propose_from_candidate, accept/reject/withdraw, append_chain_to_corpus (the sole corpus-write surface). Uses TYPE_CHECKING guards to break the circular import with chat.pack_grounding. - teaching/replay.py — run_replay_equivalence; swaps _corpus_index path to a tmp file, runs cognition lane on the active corpus AND a transient copy with the proposed chain appended, returns regressed-metrics list; trust-boundary assertion that the active corpus bytes are byte-identical pre/post. - teaching/discovery.py — moved chat.pack_grounding / chat.teaching_grounding imports inside extract_discovery_candidates to break the cycle (was masked when chat.runtime was the entry point; surfaced by CLI entry). - core/cli.py — three new subcommands: core teaching propose [--allow-evaluative] core teaching proposals [--state pending|accepted|rejected|withdrawn] [--json] core teaching review --accept --review-date YYYY-MM-DD core teaching review --reject [--note ...] core teaching review --withdraw [--note ...] - tests/test_teaching_proposals.py — 16 tests covering: every eligibility gate, proposal_id idempotency, append-only log, replay-equivalent stays pending, regression auto-rejects with named regressed metrics, --accept appends one line with typed Provenance, --accept refused on non-equivalent, state-machine blocks double-accept, real replay gate runs cognition lane twice and asserts byte-clean active corpus pre/post. Invariants preserved - versor_condition(F) < 1e-6 — C2 touches no algebra path. - Active corpus bytes byte-identical regardless of replay outcome. - No clock-time reads, no LLM, no async. - Proposal-only — accept_proposal is the sole corpus-write path. Lanes: smoke 67 / cognition 121 / runtime 19 / teaching 17 / new proposals 16. Cognition eval unchanged. Open follow-ups (not in scope): - supersession via operator review action - cross-pack falsification arbitration (ADR-0056 Call 2 deferred) - pack-data migration of frame-dependent connectives Co-Authored-By: Claude Opus 4.7 --- core/cli.py | 187 ++++++- ...ADR-0057-teaching-chain-proposal-review.md | 243 +++++++++ docs/decisions/README.md | 1 + teaching/discovery.py | 11 +- teaching/proposals.py | 484 ++++++++++++++++++ teaching/replay.py | 153 ++++++ tests/test_teaching_proposals.py | 321 ++++++++++++ 7 files changed, 1397 insertions(+), 3 deletions(-) create mode 100644 docs/decisions/ADR-0057-teaching-chain-proposal-review.md create mode 100644 teaching/proposals.py create mode 100644 teaching/replay.py create mode 100644 tests/test_teaching_proposals.py diff --git a/core/cli.py b/core/cli.py index d13d5a28..12fa5166 100644 --- a/core/cli.py +++ b/core/cli.py @@ -23,7 +23,7 @@ _CORE_RS_DIR = _REPO_ROOT / "core-rs" _CORE_RS_MANIFEST = _CORE_RS_DIR / "Cargo.toml" DESCRIPTION = "CORE versor engine command suite." -EPILOG = "Examples:\n core chat\n core pulse \"What is truth?\"\n core pulse --no-glove --json \"Compare knowledge and wisdom\"\n core bench\n core bench --suite determinism --runs 50\n core bench --suite speedup --json\n core trace \"word beginning truth\"\n core trace --output-language grc --frame-pack grc --json \"logos\"\n core rust status\n core rust build\n core oov covenant\n core pack list\n core pack verify en_minimal_v1\n core teaching audit\n core teaching audit --json\n core test --suite fast -q\n core test --suite pulse -q\n core test --suite proof -q\n core test --suite cognition -q\n core test -- tests/test_alignment_graph.py -q\n core demo audit-tour\n core demo pack-measurements\n core demo long-context-comparison\n core eval --list\n core eval cognition\n core eval cognition --json --save\n core eval cognition --split dev --version v1\n core eval cognition --split holdout" +EPILOG = "Examples:\n core chat\n core pulse \"What is truth?\"\n core pulse --no-glove --json \"Compare knowledge and wisdom\"\n core bench\n core bench --suite determinism --runs 50\n core bench --suite speedup --json\n core trace \"word beginning truth\"\n core trace --output-language grc --frame-pack grc --json \"logos\"\n core rust status\n core rust build\n core oov covenant\n core pack list\n core pack verify en_minimal_v1\n core teaching audit\n core teaching audit --json\n core teaching propose \n core teaching proposals --state pending\n core teaching review --accept --review-date 2026-05-18\n core test --suite fast -q\n core test --suite pulse -q\n core test --suite proof -q\n core test --suite cognition -q\n core test -- tests/test_alignment_graph.py -q\n core demo audit-tour\n core demo pack-measurements\n core demo long-context-comparison\n core eval --list\n core eval cognition\n core eval cognition --json --save\n core eval cognition --split dev --version v1\n core eval cognition --split holdout" _TEST_SUITES: dict[str, tuple[str, ...]] = { "fast": ( @@ -491,6 +491,137 @@ def cmd_teaching_audit(args: argparse.Namespace) -> int: return 0 +def _load_candidate_jsonl(path: str) -> Any: + """Read one enriched DiscoveryCandidate JSONL line from *path*.""" + from teaching.discovery import DiscoveryCandidate, EvidencePointer, SubQuestion + + p = Path(path) + if not p.exists(): + _die(f"candidate file not found: {path}", code=2) + raw = p.read_text(encoding="utf-8").strip() + if not raw: + _die("candidate file is empty", code=2) + first = raw.splitlines()[0].strip() + try: + payload = json.loads(first) + except json.JSONDecodeError as exc: + _die(f"invalid JSON: {exc}", code=2) + try: + evidence = tuple( + EvidencePointer(**e) for e in payload.get("evidence", []) + ) + sub_questions = tuple( + SubQuestion( + sub_id=s["sub_id"], + proposed_subject=s["proposed_subject"], + proposed_intent=s["proposed_intent"], + outcome=s["outcome"], + evidence=tuple(EvidencePointer(**e) for e in s.get("evidence", [])), + ) + for s in payload.get("sub_questions", []) + ) + return DiscoveryCandidate( + candidate_id=payload["candidate_id"], + proposed_chain=payload["proposed_chain"], + trigger=payload["trigger"], + source_turn_trace=payload.get("source_turn_trace", ""), + pack_consistent=bool(payload.get("pack_consistent", True)), + boundary_clean=bool(payload.get("boundary_clean", True)), + review_state=payload.get("review_state", "unreviewed"), + polarity=payload.get("polarity", "undetermined"), + claim_domain=payload.get("claim_domain", "factual"), + evidence=evidence, + sub_questions=sub_questions, + contemplation_depth=int(payload.get("contemplation_depth", 0)), + recursion_overflow=bool(payload.get("recursion_overflow", False)), + ) + except (KeyError, TypeError) as exc: + _die(f"candidate JSON missing required field: {exc}", code=2) + + +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, + ) + + candidate = _load_candidate_jsonl(args.candidate_path) + log_path = Path(args.log) if args.log else None + log = ProposalLog(log_path) + try: + proposal = propose_from_candidate( + candidate, log=log, allow_evaluative=args.allow_evaluative, + ) + except ProposalError as exc: + _die(f"ineligible: {exc}", code=1) + rec = log.find(proposal.proposal_id) + print(f"proposal_id : {proposal.proposal_id}") + print(f"state : {rec['state']}") + if rec.get("replay_evidence"): + ev = rec["replay_evidence"] + print(f"replay_equivalent: {ev['replay_equivalent']}") + if ev.get("regressed_metrics"): + print(f"regressed : {', '.join(ev['regressed_metrics'])}") + if rec.get("operator_note"): + print(f"note : {rec['operator_note']}") + return 0 if rec["state"] in ("pending", "accepted") else 1 + + +def cmd_teaching_proposals(args: argparse.Namespace) -> int: + from teaching.proposals import ProposalLog + + log_path = Path(args.log) if args.log else None + log = ProposalLog(log_path) + state = log.current_state() + if args.state: + state = {pid: rec for pid, rec in state.items() if rec["state"] == args.state} + if args.json: + print(json.dumps(state, ensure_ascii=False, indent=2, sort_keys=True)) + return 0 + if not state: + print("(no proposals)") + return 0 + for pid, rec in state.items(): + chain = rec["proposal"]["proposed_chain"] + print( + f"{pid} {rec['state']:<10} " + f"{chain.get('subject')} {chain.get('connective')} {chain.get('object')} " + f"({chain.get('intent')})" + ) + return 0 + + +def cmd_teaching_review(args: argparse.Namespace) -> int: + from teaching.proposals import ( + ProposalError, ProposalLog, + accept_proposal, reject_proposal, withdraw_proposal, + ) + + log_path = Path(args.log) if args.log else None + log = ProposalLog(log_path) + try: + if args.accept: + if not args.review_date: + _die("--accept requires --review-date YYYY-MM-DD", code=2) + from chat.teaching_grounding import _CORPUS_PATH + chain_id = accept_proposal( + args.proposal_id, log=log, + corpus_path=_CORPUS_PATH, + review_date=args.review_date, + operator_note=args.note, + ) + print(f"accepted; appended chain_id = {chain_id}") + elif args.reject: + reject_proposal(args.proposal_id, log=log, operator_note=args.note) + print(f"{args.proposal_id} rejected") + elif args.withdraw: + withdraw_proposal(args.proposal_id, log=log, operator_note=args.note) + print(f"{args.proposal_id} withdrawn") + except ProposalError as exc: + _die(str(exc), code=1) + return 0 + + def cmd_pack_validate(args: argparse.Namespace) -> int: """Run executable source-pack validation gates.""" pack_id = _safe_pack_id(args.pack_id) @@ -1429,6 +1560,60 @@ def build_parser() -> argparse.ArgumentParser: ) teaching_audit.set_defaults(func=cmd_teaching_audit) + teaching_propose = teaching_sub.add_parser( + "propose", + help="convert an enriched DiscoveryCandidate (JSONL) into a TeachingChainProposal", + ) + teaching_propose.add_argument( + "candidate_path", + help="path to a JSONL file containing one enriched candidate line", + ) + teaching_propose.add_argument( + "--allow-evaluative", action="store_true", + help="permit claim_domain=evaluative proposals (operator override)", + ) + teaching_propose.add_argument( + "--log", default=None, + help="proposal log path (default: teaching/proposals/proposals.jsonl)", + ) + teaching_propose.set_defaults(func=cmd_teaching_propose) + + teaching_proposals = teaching_sub.add_parser( + "proposals", + help="list proposals in the append-only log", + ) + teaching_proposals.add_argument( + "--state", default=None, + choices=("pending", "accepted", "rejected", "withdrawn"), + help="filter by review state", + ) + teaching_proposals.add_argument( + "--log", default=None, help="proposal log path", + ) + teaching_proposals.add_argument( + "--json", action="store_true", help="machine-readable output", + ) + teaching_proposals.set_defaults(func=cmd_teaching_proposals) + + teaching_review = teaching_sub.add_parser( + "review", + help="operator review action: accept / reject / withdraw a pending proposal", + ) + teaching_review.add_argument("proposal_id") + grp = teaching_review.add_mutually_exclusive_group(required=True) + grp.add_argument("--accept", action="store_true") + grp.add_argument("--reject", action="store_true") + grp.add_argument("--withdraw", action="store_true") + teaching_review.add_argument("--note", default="", help="operator note") + teaching_review.add_argument( + "--review-date", default=None, + help="review date (YYYY-MM-DD) — required on --accept", + ) + teaching_review.add_argument( + "--log", default=None, help="proposal log path", + ) + teaching_review.set_defaults(func=cmd_teaching_review) + rust = subparsers.add_parser( "rust", help="build, test, and inspect the Rust backend", diff --git a/docs/decisions/ADR-0057-teaching-chain-proposal-review.md b/docs/decisions/ADR-0057-teaching-chain-proposal-review.md new file mode 100644 index 00000000..41541788 --- /dev/null +++ b/docs/decisions/ADR-0057-teaching-chain-proposal-review.md @@ -0,0 +1,243 @@ +# ADR-0057 — Teaching-Chain Proposal + Review + Replay-Equivalence Gate (Phase C2) + +**Status:** Accepted +**Date:** 2026-05-18 +**Author:** Shay +**Completes:** ADR-0055 §Decision Phase C (with [ADR-0056](./ADR-0056-contemplation-loop-c1.md)) + +--- + +## Context — how we got here + +ADR-0055 introduced a four-tier inter-session memory architecture +and split corpus extension into a **proposal-only** path. ADR-0056 +(Phase C1) implemented the cognitive surface: a contemplated +`DiscoveryCandidate` carries `polarity`, `claim_domain`, and +composed `evidence`. C1 explicitly does **not** mutate the active +teaching corpus — its output is structured evidence on disk. + +C2 is the **only** path that turns reviewed evidence into a corpus +mutation. It is the riskiest piece in the chain and gets its own +ADR for that reason. + +### Three load-bearing calls + +#### Call 1 — Replay-equivalence as a *precondition*, not a permission + +**Choice:** The replay-equivalence eval gate is a *necessary* but +**not sufficient** condition for corpus append. A proposal that +passes the gate becomes eligible for operator review; the operator +still has to accept it explicitly. The gate eliminates regressions; +the operator decides on the merits. + +**Why:** + +- CLAUDE.md doctrine: "Pack mutation is proposal-only until + reviewed." Eval-passing is not review. A chain that doesn't + regress metrics can still be wrong, harmful, or off-doctrine. +- The gate is mechanical (regress on any metric → auto-reject). + Review is judgment. Conflating them would smuggle in an + auto-apply path that bypasses human review. +- Auto-rollback on regression keeps the corpus byte-clean even + when a proposal is mechanically rejected. + +**Rejected alternative:** Replay-equivalent ⇒ auto-append. Same +shape as the smart-mistake C1 was extracted to prevent. + +#### Call 2 — Eligibility = `polarity != "undetermined"` AND reviewed-evidence floor + +**Choice:** A `DiscoveryCandidate` is *eligible* to become a +`TeachingChainProposal` iff: + +1. `polarity ∈ {"affirms", "falsifies"}` (undetermined cannot + propose — composing to undetermined means the system itself + isn't sure). +2. `evidence` contains at least one `source="corpus"` pointer + (reviewed-evidence floor — pack residency alone is shape + evidence, not relation evidence). +3. `claim_domain != "evaluative"` UNLESS an operator has flagged + the proposal with `--allow-evaluative` and a strong-tier hedge + surface is attached (per ADR-0056 evaluative threshold). +4. `boundary_clean=True` (the source turn was not under refusal + or hedge — boundary-clean is a guard against polluted + provenance). +5. `proposed_chain` is *complete* — non-null `subject`, `intent`, + `connective`, `object`. + +**Why:** Each gate corresponds to a doctrinal commitment that +CLAUDE.md or an earlier ADR already pinned. Eligibility is a +mechanical check — no judgment. Failing any gate keeps the +candidate as evidence on disk; eligible ones move on for replay ++ review. + +#### Call 3 — Append-only proposal log; corpus history append-only too + +**Choice:** Every proposal — accepted, rejected (operator), +auto-rejected (replay regression), or withdrawn — is appended to +`teaching/proposals/proposals.jsonl` and never deleted. Accepted +proposals additionally append their `proposed_chain` to the active +corpus (`teaching/cognition_chains/cognition_chains_v1.jsonl`) with +typed `Provenance(source="discovery_promoted", adr_id="adr-0057", +review_date=...)` from ADR-0055 Phase A. The active corpus view +remains derived via the existing `superseded_by` mechanism — C2 +adds entries, doesn't rewrite history. + +**Why:** + +- Append-only history is a CLAUDE.md commitment for replayability. +- The same `Provenance` schema Phase A introduced is the natural + home for "where did this chain come from"; `discovery_promoted` + is the canonical source tag. +- Future calibration / re-ratification ADRs (Phase D, E) need the + full record of every proposal, not just the accepted ones. + +--- + +## Decision — Phase C2 spec + +### Data shape + +```python +@dataclass(frozen=True, slots=True) +class TeachingChainProposal: + proposal_id: str # sha256(source_candidate_id + chain payload) + source_candidate_id: str + proposed_chain: dict[str, Any] # complete: subject, intent, connective, object + polarity: Literal["affirms", "falsifies"] + claim_domain: ClaimDomain + evidence: tuple[EvidencePointer, ...] + review_state: Literal["pending", "accepted", "rejected", "withdrawn"] + operator_note: str = "" + replay_evidence: ReplayEvidence | None = None + provenance: Provenance | None = None # populated on accept +``` + +```python +@dataclass(frozen=True, slots=True) +class ReplayEvidence: + baseline: dict[str, float] # metrics on the active corpus + candidate: dict[str, float] # metrics with proposed chain appended + regressed_metrics: tuple[str, ...] + replay_equivalent: bool +``` + +### Replay-equivalence gate + +For every proposal that reaches the gate: + +1. Snapshot the active corpus file bytes. +2. Run the cognition lane (public + dev + holdout splits) to + produce the baseline metric set. +3. Append the proposed chain to a *temporary copy* of the corpus, + invalidate the cached `_corpus_index()`, and re-run the lane + on the same case sets. +4. Compare metric-for-metric. A metric *regresses* iff its + candidate value is strictly less than the baseline value + (no float tolerance — the lane is deterministic). +5. Restore the original corpus bytes (or never touch the active + file in the first place — see implementation note below). +6. If any metric regressed ⇒ `replay_equivalent=False`, + proposal auto-transitions to `review_state="rejected"`, + `operator_note="auto_rollback_regression: "`. +7. Otherwise ⇒ `replay_equivalent=True`, proposal stays + `review_state="pending"` awaiting operator review. + +**Implementation note (trust boundary):** the gate must never +write to the active corpus file even transiently. It writes to +an *isolated path* and patches `_corpus_index()` to load from +that path via dependency injection. Active-file bytes are +byte-identical pre/post regardless of outcome. + +### Operator review surface + +CLI commands (sibling of the existing `core teaching audit`): + +```text +core teaching propose [--from-sink ] + Convert an eligible enriched DiscoveryCandidate into a + TeachingChainProposal. Runs the replay-equivalence gate + immediately. Idempotent on (candidate_id, chain payload). + +core teaching proposals [--state ] [--json] + List proposals; default lists pending. + +core teaching review --accept [--note "..."] +core teaching review --reject [--note "..."] +core teaching review --withdraw [--note "..."] + Operator decision. --accept on a replay-equivalent proposal + appends the chain to the active corpus with typed provenance. + --accept on a non-equivalent proposal is rejected with an + explicit error. --reject and --withdraw transition state + only; the corpus is untouched. +``` + +### Trust boundary + +- **No automatic accept.** Replay-equivalence is a precondition, + not a permission. Only operator `--accept` writes to the corpus. +- **No corpus rewrites.** Accept appends one new line; entries + are retired only via the existing `superseded_by` mechanism in + a separate operator action. +- **No proposal deletion.** All four review states are terminal + in the append-only log; "delete" doesn't exist. +- **No identity / safety / ethics mutation.** Per ADR-0027 and + ADR-0029, those packs are out of scope for C2. +- **No clock-time content read.** The `review_date` in + `Provenance` is the only timestamp; sourced from the operator's + command invocation, not from runtime hot path. + +--- + +## Non-goals (explicit) + +- No async or concurrency primitives — replay is synchronous. +- No cross-pack arbitration (deferred per ADR-0056 Call 2). +- No re-ratification of identity / safety / ethics packs. +- No automatic supersession of existing chains by a new accept; + supersession is a separate, future operator action. +- No metric-tolerance bands; the lane is deterministic and any + regression is real. + +--- + +## Verification (acceptance criteria) + +- Eligible enriched candidates produce a `TeachingChainProposal`; + ineligible ones raise with the failing gate named. +- The replay-equivalence gate never mutates the active corpus + file bytes regardless of outcome. +- A proposal whose chain causes any cognition metric to regress + auto-transitions to `rejected` with `replay_equivalent=False` + and an `auto_rollback_regression` note. +- A replay-equivalent proposal stays `pending` until operator + decision. +- `core teaching review --accept` on a `pending` + + replay-equivalent proposal appends one line to the active + corpus with `Provenance(source="discovery_promoted", + adr_id="adr-0057")` and re-runs the active corpus through + `_corpus_index()` cleanly (no new drops). +- `core teaching review --accept` on a non-equivalent proposal + raises and refuses to append. +- The proposals log is append-only; replaying it reconstructs + the same review-state for every entry. +- `versor_condition(F) < 1e-6` invariant preserved (no algebra + touched). +- `core eval cognition` numbers unchanged on splits that don't + include accepted-proposal cases. + +--- + +## Cross-References + +- [ADR-0021](./ADR-0021-epistemic-status.md) — `EpistemicStatus` + COHERENT promotion semantics; C2 is the mechanical surface. +- [ADR-0027](./ADR-0027-identity-packs.md) / + [ADR-0029](./ADR-0029-safety-pack.md) / + [ADR-0033](./ADR-0033-ethics-pack.md) — packs out of scope. +- [ADR-0052](./ADR-0052-teaching-grounded-surface.md) — the + active corpus this loop appends to. +- [ADR-0055](./ADR-0055-inter-session-memory-discovery-promotion.md) + — the parent design; Phase A's `Provenance` and `superseded_by` + are the substrate this ADR builds on. +- [ADR-0056](./ADR-0056-contemplation-loop-c1.md) — the cognitive + surface whose output feeds C2's eligibility gate. diff --git a/docs/decisions/README.md b/docs/decisions/README.md index 32b86a96..c29f96af 100644 --- a/docs/decisions/README.md +++ b/docs/decisions/README.md @@ -66,6 +66,7 @@ ADRs record significant architectural decisions: what was decided, why, what alt | [ADR-0054](ADR-0054-vault-recall-indexing-batching.md) | Vault recall matrix-cache indexing + batched API; holdout split wired into eval CLI | Accepted (2026-05-18) | | [ADR-0055](ADR-0055-inter-session-memory-discovery-promotion.md) | Inter-session memory: reviewed discovery promotion (phased design — DiscoveryCandidate, TeachingChainProposal, replay-equivalence gate); Phase A + Phase B Accepted | **Phase A + B Accepted**; C–E Proposed (2026-05-18) | | [ADR-0056](ADR-0056-contemplation-loop-c1.md) | Contemplation loop (Phase C1): question decomposition, polarity (affirms/falsifies/undetermined), claim_domain typing (factual/relational/evaluative), sync-only by design | **Accepted** (2026-05-18, implemented `4eecf73`) | +| [ADR-0057](ADR-0057-teaching-chain-proposal-review.md) | Teaching-chain proposal + review + replay-equivalence gate (Phase C2): the only path to active-corpus extension; eligibility predicate; auto-reject on metric regression; operator accept/reject/withdraw; append-only proposal log | **Accepted** (2026-05-18) | --- diff --git a/teaching/discovery.py b/teaching/discovery.py index c111a67a..33868674 100644 --- a/teaching/discovery.py +++ b/teaching/discovery.py @@ -51,10 +51,14 @@ import json from dataclasses import dataclass from typing import Any, Literal -from chat.pack_grounding import _pack_index -from chat.teaching_grounding import _corpus_index from generate.intent import IntentTag +# ``chat.pack_grounding`` and ``chat.teaching_grounding`` are +# imported lazily inside ``extract_discovery_candidates`` to break a +# circular import chain when an entry-point (e.g. the CLI) imports +# ``teaching.proposals`` → ``teaching.discovery`` before ``chat`` +# has been fully initialized. + DiscoveryTrigger = Literal[ "would_have_grounded", @@ -257,6 +261,9 @@ def extract_discovery_candidates( if not lemma: return () + from chat.pack_grounding import _pack_index + from chat.teaching_grounding import _corpus_index + pack = _pack_index() if lemma not in pack: return () diff --git a/teaching/proposals.py b/teaching/proposals.py new file mode 100644 index 00000000..77f7badc --- /dev/null +++ b/teaching/proposals.py @@ -0,0 +1,484 @@ +"""ADR-0057 Phase C2 — TeachingChainProposal + append-only proposal log. + +A ``TeachingChainProposal`` is the **only** path by which the +system extends its active teaching corpus. Trust boundary: + + - Proposals are derived from contemplated DiscoveryCandidates + (ADR-0056 Phase C1 output). + - Eligibility (Call 2 in ADR-0057) is a mechanical predicate. + Ineligible candidates raise; eligible ones become a pending + proposal. + - The replay-equivalence gate (``teaching/replay.py``) is a + *precondition*, not a permission. A passing gate moves the + proposal to ``replay_equivalent=True``; only an explicit + operator ``accept`` writes to the active corpus. + - The proposal log is append-only. All four review states + (pending / accepted / rejected / withdrawn) are terminal in + the log; "delete" doesn't exist. +""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import TYPE_CHECKING, Any, Literal + +from teaching.provenance import Provenance + +if TYPE_CHECKING: + # Deferred to break a circular import: teaching.discovery → + # chat.pack_grounding → chat.__init__ → chat.runtime → + # teaching.discovery. These names are used only as type + # annotations here, so the TYPE_CHECKING guard is safe. + from teaching.discovery import ( + ClaimDomain, + DiscoveryCandidate, + EvidencePointer, + ) + + +# Default proposal log location. Tests inject a tmp path; callers +# in production use this constant. +DEFAULT_PROPOSAL_LOG_PATH: Path = ( + Path(__file__).resolve().parent / "proposals" / "proposals.jsonl" +) + + +ReviewState = Literal["pending", "accepted", "rejected", "withdrawn"] + + +@dataclass(frozen=True, slots=True) +class ReplayEvidence: + """Cognition-lane metrics before/after the proposed append. + + A regressed metric is one whose candidate value is strictly + less than the baseline value. The cognition lane is + deterministic; no float tolerance is applied (ADR-0057 Call 1 + note: any regression is real). + """ + + baseline: dict[str, float] + candidate: dict[str, float] + regressed_metrics: tuple[str, ...] + replay_equivalent: bool + + def as_dict(self) -> dict[str, Any]: + return { + "baseline": dict(self.baseline), + "candidate": dict(self.candidate), + "regressed_metrics": list(self.regressed_metrics), + "replay_equivalent": bool(self.replay_equivalent), + } + + +@dataclass(frozen=True, slots=True) +class TeachingChainProposal: + """One proposed extension of the active teaching corpus.""" + + proposal_id: str + source_candidate_id: str + proposed_chain: dict[str, Any] + polarity: Literal["affirms", "falsifies"] + claim_domain: ClaimDomain + evidence: tuple[EvidencePointer, ...] + review_state: ReviewState = "pending" + operator_note: str = "" + replay_evidence: ReplayEvidence | None = None + provenance: Provenance | None = None + + def as_dict(self) -> dict[str, Any]: + return { + "proposal_id": self.proposal_id, + "source_candidate_id": self.source_candidate_id, + "proposed_chain": dict(self.proposed_chain), + "polarity": self.polarity, + "claim_domain": self.claim_domain, + "evidence": [e.as_dict() for e in self.evidence], + "review_state": self.review_state, + "operator_note": self.operator_note, + "replay_evidence": ( + self.replay_evidence.as_dict() + if self.replay_evidence is not None + else None + ), + "provenance": (asdict(self.provenance) if self.provenance else None), + } + + +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.""" + + +# --------------------------------------------------------------------------- +# Eligibility (ADR-0057 Call 2) +# --------------------------------------------------------------------------- + + +def _is_chain_complete(chain: dict[str, Any]) -> bool: + return all( + chain.get(k) and isinstance(chain.get(k), str) + for k in ("subject", "intent", "connective", "object") + ) + + +def check_eligibility( + candidate: DiscoveryCandidate, *, allow_evaluative: bool = False +) -> None: + """Raise ``ProposalError`` if ``candidate`` cannot become a proposal. + + Five mechanical gates (ADR-0057 Call 2): + 1. polarity ∈ {affirms, falsifies} + 2. evidence contains at least one corpus pointer + 3. claim_domain != evaluative unless ``allow_evaluative`` + 4. boundary_clean=True + 5. proposed_chain is complete (all four fields populated) + """ + if candidate.polarity not in ("affirms", "falsifies"): + raise ProposalError( + f"polarity must be 'affirms' or 'falsifies'; got " + f"{candidate.polarity!r} — undetermined candidates cannot propose" + ) + if not any(e.source == "corpus" for e in candidate.evidence): + raise ProposalError( + "evidence floor: at least one source='corpus' pointer is required" + ) + if candidate.claim_domain == "evaluative" and not allow_evaluative: + raise ProposalError( + "claim_domain='evaluative' requires explicit --allow-evaluative" + ) + if not candidate.boundary_clean: + raise ProposalError("source turn was not boundary_clean") + if not _is_chain_complete(candidate.proposed_chain): + raise ProposalError( + "proposed_chain must have subject/intent/connective/object populated" + ) + + +# --------------------------------------------------------------------------- +# Proposal id derivation +# --------------------------------------------------------------------------- + + +def _proposal_id(source_candidate_id: str, chain: dict[str, Any]) -> str: + payload = { + "source_candidate_id": source_candidate_id, + "proposed_chain": chain, + } + blob = json.dumps(payload, sort_keys=True, separators=(",", ":")) + return hashlib.sha256(blob.encode("utf-8")).hexdigest()[:32] + + +def build_proposal( + candidate: DiscoveryCandidate, *, allow_evaluative: bool = False +) -> TeachingChainProposal: + """Build a ``pending`` proposal from an eligible candidate. + + Raises ``ProposalError`` for any failing gate. Idempotent on + (source_candidate_id, proposed_chain): same inputs produce the + same ``proposal_id``. + """ + check_eligibility(candidate, allow_evaluative=allow_evaluative) + assert candidate.polarity in ("affirms", "falsifies") + return TeachingChainProposal( + proposal_id=_proposal_id(candidate.candidate_id, candidate.proposed_chain), + source_candidate_id=candidate.candidate_id, + proposed_chain=dict(candidate.proposed_chain), + polarity=candidate.polarity, + claim_domain=candidate.claim_domain, + evidence=tuple(candidate.evidence), + ) + + +# --------------------------------------------------------------------------- +# Append-only proposal log +# --------------------------------------------------------------------------- + + +class ProposalLog: + """Append-only JSONL store of proposals + state transitions. + + Each line is one *event*: + + - ``{"event": "created", "proposal": {...}}`` + - ``{"event": "replay", "proposal_id": "...", "replay_evidence": {...}}`` + - ``{"event": "transition", "proposal_id": "...", + "to": "accepted|rejected|withdrawn", "note": "..."}`` + - ``{"event": "accepted_corpus_append", "proposal_id": "...", + "chain_id": "...", "provenance": {...}}`` + + The active view (``current_state``) is derived by replaying the + log from the top; the file is never rewritten. + """ + + def __init__(self, path: Path | None = None) -> None: + self.path = Path(path) if path else DEFAULT_PROPOSAL_LOG_PATH + self.path.parent.mkdir(parents=True, exist_ok=True) + + # -- write side --------------------------------------------------- + + def _append(self, event: dict[str, Any]) -> None: + line = json.dumps(event, sort_keys=True, separators=(",", ":")) + with self.path.open("a", encoding="utf-8") as fh: + fh.write(line + "\n") + + def record_created(self, proposal: TeachingChainProposal) -> None: + self._append({"event": "created", "proposal": proposal.as_dict()}) + + def record_replay(self, proposal_id: str, evidence: ReplayEvidence) -> None: + self._append({ + "event": "replay", + "proposal_id": proposal_id, + "replay_evidence": evidence.as_dict(), + }) + + def record_transition( + self, proposal_id: str, to_state: ReviewState, note: str + ) -> None: + self._append({ + "event": "transition", + "proposal_id": proposal_id, + "to": to_state, + "note": note, + }) + + def record_corpus_append( + self, proposal_id: str, chain_id: str, provenance: Provenance + ) -> None: + self._append({ + "event": "accepted_corpus_append", + "proposal_id": proposal_id, + "chain_id": chain_id, + "provenance": asdict(provenance), + }) + + # -- read side ---------------------------------------------------- + + def _events(self) -> list[dict[str, Any]]: + if not self.path.exists(): + return [] + events: list[dict[str, Any]] = [] + for line in self.path.read_text(encoding="utf-8").splitlines(): + line = line.strip() + if not line: + continue + try: + events.append(json.loads(line)) + except json.JSONDecodeError: + continue + return events + + def current_state(self) -> dict[str, dict[str, Any]]: + """Replay the log → ``{proposal_id: {state, proposal, replay, + note, accepted_chain_id}}``. + + The active view is derived deterministically from the log. + """ + view: dict[str, dict[str, Any]] = {} + for ev in self._events(): + kind = ev.get("event") + if kind == "created": + p = ev.get("proposal") or {} + pid = p.get("proposal_id") + if not pid: + continue + view.setdefault(pid, { + "proposal": p, + "state": p.get("review_state", "pending"), + "replay_evidence": p.get("replay_evidence"), + "operator_note": p.get("operator_note", ""), + "accepted_chain_id": None, + "accepted_provenance": None, + }) + elif kind == "replay": + pid = ev.get("proposal_id") + if pid in view: + view[pid]["replay_evidence"] = ev.get("replay_evidence") + elif kind == "transition": + pid = ev.get("proposal_id") + if pid in view: + view[pid]["state"] = ev.get("to") + view[pid]["operator_note"] = ev.get("note", "") + elif kind == "accepted_corpus_append": + pid = ev.get("proposal_id") + if pid in view: + view[pid]["accepted_chain_id"] = ev.get("chain_id") + view[pid]["accepted_provenance"] = ev.get("provenance") + return view + + def find(self, proposal_id: str) -> dict[str, Any] | None: + return self.current_state().get(proposal_id) + + +# --------------------------------------------------------------------------- +# Corpus append (operator-accept side-effect) +# --------------------------------------------------------------------------- + + +def append_chain_to_corpus( + chain: dict[str, Any], + *, + corpus_path: Path, + provenance: Provenance, + chain_id: str | None = None, +) -> str: + """Append one reviewed chain JSON line to the active corpus. + + Returns the ``chain_id`` written. Trust boundary: this is the + ONLY function in the codebase that writes to the active teaching + corpus, and it is reachable only from + ``accept_proposal`` after the replay-equivalence gate and + operator review. + """ + subject = str(chain["subject"]).strip().lower() + intent = str(chain["intent"]).strip().lower() + connective = str(chain["connective"]).strip() + obj = str(chain["object"]).strip().lower() + if not chain_id: + chain_id = f"{intent}_{subject}_{connective}_{obj}" + entry = { + "chain_id": chain_id, + "subject": subject, + "intent": intent, + "connective": connective, + "object": obj, + "domains_subject_k": 2, + "domains_object_k": 1, + "provenance": provenance.raw or ( + f"{provenance.adr_id or 'adr-0057'}:{provenance.source}:" + f"{provenance.review_date or ''}" + ), + } + line = json.dumps(entry, sort_keys=True, separators=(",", ":")) + with corpus_path.open("a", encoding="utf-8") as fh: + fh.write(line + "\n") + return chain_id + + +# --------------------------------------------------------------------------- +# Orchestration helpers — propose / replay / accept / reject / withdraw +# --------------------------------------------------------------------------- + + +def propose_from_candidate( + candidate: DiscoveryCandidate, + *, + log: ProposalLog, + run_replay: Any = None, + allow_evaluative: bool = False, +) -> TeachingChainProposal: + """End-to-end: build proposal, run replay-equivalence gate, + auto-reject on regression, otherwise leave pending. + + ``run_replay`` is the replay function (``teaching.replay. + run_replay_equivalence`` by default); accepting it as a kwarg + 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. + """ + proposal = build_proposal(candidate, allow_evaluative=allow_evaluative) + existing = log.find(proposal.proposal_id) + if existing is not None: + return proposal + log.record_created(proposal) + + if run_replay is None: + from teaching.replay import run_replay_equivalence as run_replay + evidence = run_replay(proposal.proposed_chain) + log.record_replay(proposal.proposal_id, evidence) + + if not evidence.replay_equivalent: + note = "auto_rollback_regression: " + ",".join(evidence.regressed_metrics) + log.record_transition(proposal.proposal_id, "rejected", note) + + return proposal + + +def accept_proposal( + proposal_id: str, + *, + log: ProposalLog, + corpus_path: Path, + review_date: str, + operator_note: str = "", +) -> str: + """Operator accept — append proposed chain to the active corpus. + + Pre-conditions (each raises ``ProposalError`` on failure): + - proposal exists in the log + - current state is ``pending`` + - replay evidence is present and replay_equivalent=True + Returns the ``chain_id`` written to the corpus. + """ + record = log.find(proposal_id) + if record is None: + raise ProposalError(f"proposal not found: {proposal_id}") + if record["state"] != "pending": + raise ProposalError( + f"proposal {proposal_id} is {record['state']!r}, not pending" + ) + replay = record.get("replay_evidence") + if not replay or not replay.get("replay_equivalent"): + raise ProposalError( + f"proposal {proposal_id} is not replay-equivalent; cannot accept" + ) + chain = record["proposal"]["proposed_chain"] + provenance = Provenance( + adr_id="adr-0057", + source="discovery_promoted", + review_date=review_date, + raw=f"adr-0057:discovery_promoted:{review_date}", + ) + chain_id = append_chain_to_corpus( + chain, corpus_path=corpus_path, provenance=provenance + ) + log.record_transition(proposal_id, "accepted", operator_note) + log.record_corpus_append(proposal_id, chain_id, provenance) + return chain_id + + +def reject_proposal( + proposal_id: str, *, log: ProposalLog, operator_note: str = "" +) -> None: + record = log.find(proposal_id) + if record is None: + raise ProposalError(f"proposal not found: {proposal_id}") + if record["state"] != "pending": + raise ProposalError( + f"proposal {proposal_id} is {record['state']!r}, not pending" + ) + log.record_transition(proposal_id, "rejected", operator_note) + + +def withdraw_proposal( + proposal_id: str, *, log: ProposalLog, operator_note: str = "" +) -> None: + record = log.find(proposal_id) + if record is None: + raise ProposalError(f"proposal not found: {proposal_id}") + if record["state"] != "pending": + raise ProposalError( + f"proposal {proposal_id} is {record['state']!r}, not pending" + ) + log.record_transition(proposal_id, "withdrawn", operator_note) + + +__all__ = [ + "DEFAULT_PROPOSAL_LOG_PATH", + "ProposalError", + "ProposalLog", + "ReplayEvidence", + "ReviewState", + "TeachingChainProposal", + "accept_proposal", + "append_chain_to_corpus", + "build_proposal", + "check_eligibility", + "propose_from_candidate", + "reject_proposal", + "withdraw_proposal", +] diff --git a/teaching/replay.py b/teaching/replay.py new file mode 100644 index 00000000..98b011da --- /dev/null +++ b/teaching/replay.py @@ -0,0 +1,153 @@ +"""ADR-0057 §Replay-equivalence gate. + +Given a proposed chain, run the cognition lane against the active +corpus *and* against a transient copy of the active corpus with the +proposed chain appended. Compare metrics: any regression rejects +the proposal mechanically; equivalence makes the proposal eligible +for operator review. + +Trust boundary +- The active corpus file bytes are NEVER touched by this gate, + regardless of outcome. The transient candidate corpus is written + to an isolated path; the runtime's ``_corpus_index`` cache is + swapped to load from that path for the candidate measurement, + then restored. +- No background tasks, no async, no clock-time reads. Synchronous + swap-measure-restore. +""" + +from __future__ import annotations + +import json +import shutil +import tempfile +from contextlib import contextmanager +from pathlib import Path +from typing import Any, Iterator + +from chat import teaching_grounding as _tg +from teaching.proposals import ReplayEvidence + + +# Metrics watched for regression. Any metric whose candidate value +# is strictly less than the baseline value counts as a regression. +_WATCHED_METRICS: tuple[str, ...] = ( + "intent_accuracy", + "surface_groundedness", + "term_capture_rate", + "versor_closure_rate", +) + + +@contextmanager +def _swap_corpus_path(temp_path: Path) -> Iterator[None]: + """Temporarily point ``_corpus_index`` at *temp_path*. + + Clears the lru_cache before and after the swap so the runtime + re-reads the corpus fresh in both directions. The active + corpus on disk is not touched. + """ + real_path = _tg._CORPUS_PATH + try: + _tg._CORPUS_PATH = temp_path # type: ignore[assignment] + _tg._corpus_index.cache_clear() + yield + finally: + _tg._CORPUS_PATH = real_path # type: ignore[assignment] + _tg._corpus_index.cache_clear() + + +def _run_cognition_public() -> dict[str, float]: + """Run the public cognition split and return a metrics dict. + + Kept inside a function so import time stays cheap for callers + that never trigger replay. + """ + from evals.framework import get_lane, run_lane + + lane = get_lane("cognition") + result = run_lane(lane, version="v1", split="public") + out: dict[str, float] = {} + for k in _WATCHED_METRICS: + v = result.metrics.get(k) + if isinstance(v, (int, float)): + out[k] = float(v) + return out + + +def _build_candidate_corpus( + active_corpus_path: Path, candidate_chain: dict[str, Any], dest: Path +) -> None: + """Copy active corpus to *dest* and append one candidate line.""" + if active_corpus_path.exists(): + shutil.copyfile(active_corpus_path, dest) + else: + dest.write_text("", encoding="utf-8") + subject = str(candidate_chain["subject"]).strip().lower() + intent = str(candidate_chain["intent"]).strip().lower() + connective = str(candidate_chain["connective"]).strip() + obj = str(candidate_chain["object"]).strip().lower() + chain_id = f"{intent}_{subject}_{connective}_{obj}_replay" + entry = { + "chain_id": chain_id, + "subject": subject, + "intent": intent, + "connective": connective, + "object": obj, + "domains_subject_k": 2, + "domains_object_k": 1, + "provenance": "adr-0057:discovery_promoted:replay", + } + line = json.dumps(entry, sort_keys=True, separators=(",", ":")) + with dest.open("a", encoding="utf-8") as fh: + fh.write(line + "\n") + + +def run_replay_equivalence(chain: dict[str, Any]) -> ReplayEvidence: + """Run the gate. Active corpus bytes byte-identical pre/post. + + Returns: + ``ReplayEvidence(baseline=..., candidate=..., regressed_metrics=..., + replay_equivalent=...)`` + """ + active_path = _tg._CORPUS_PATH + active_bytes_before = active_path.read_bytes() if active_path.exists() else b"" + + # Baseline: just run against the active corpus. Cache is cleared + # to make sure we read the current state of disk. + _tg._corpus_index.cache_clear() + baseline = _run_cognition_public() + + # Candidate: build a transient corpus with the chain appended + # and point ``_corpus_index`` at it. + with tempfile.TemporaryDirectory() as tmpdir: + cand_path = Path(tmpdir) / "candidate_corpus.jsonl" + _build_candidate_corpus(active_path, chain, cand_path) + with _swap_corpus_path(cand_path): + candidate = _run_cognition_public() + + regressed: list[str] = [] + for metric in _WATCHED_METRICS: + b = baseline.get(metric) + c = candidate.get(metric) + if b is None or c is None: + continue + if c < b: + regressed.append(metric) + + # Trust-boundary assertion: active file bytes unchanged. + active_bytes_after = active_path.read_bytes() if active_path.exists() else b"" + if active_bytes_after != active_bytes_before: # pragma: no cover — defensive + raise RuntimeError( + "replay gate mutated the active corpus — trust boundary violated" + ) + + return ReplayEvidence( + baseline=baseline, + candidate=candidate, + regressed_metrics=tuple(sorted(regressed)), + replay_equivalent=not regressed, + ) + + +__all__ = ["run_replay_equivalence"] diff --git a/tests/test_teaching_proposals.py b/tests/test_teaching_proposals.py new file mode 100644 index 00000000..73a62828 --- /dev/null +++ b/tests/test_teaching_proposals.py @@ -0,0 +1,321 @@ +"""ADR-0057 Phase C2 — TeachingChainProposal eligibility, replay- +equivalence gate, append-only proposal log, and operator review +state machine. + +Pinned contracts: + - Eligibility predicate raises on every failing gate. + - Idempotent proposal_id derivation. + - Replay-equivalence gate never mutates the active corpus. + - Regression auto-transitions proposal to rejected. + - --accept only legal when state==pending AND replay_equivalent. + - Append-only log: replaying the log reconstructs the same state. +""" + +from __future__ import annotations + +import json +from dataclasses import replace +from pathlib import Path + +import pytest + +from chat.teaching_grounding import _CORPUS_PATH +from teaching.discovery import ( + DiscoveryCandidate, + EvidencePointer, +) +from teaching.proposals import ( + ProposalError, + ProposalLog, + ReplayEvidence, + accept_proposal, + append_chain_to_corpus, + build_proposal, + check_eligibility, + propose_from_candidate, + reject_proposal, + withdraw_proposal, +) +from teaching.provenance import Provenance + + +CORPUS_BYTES_BEFORE = _CORPUS_PATH.read_bytes() if _CORPUS_PATH.exists() else b"" + + +def _enriched(*, polarity="affirms", claim_domain="factual", + connective="reveals", obj="truth", subject="light", + evidence=None, boundary_clean=True): + if evidence is None: + evidence = ( + EvidencePointer( + source="corpus", ref="some_chain", + polarity=polarity, epistemic_status="coherent", + ), + ) + return DiscoveryCandidate( + candidate_id="cand_xyz", + proposed_chain={ + "subject": subject, "intent": "cause", + "connective": connective, "object": obj, + }, + trigger="would_have_grounded", + source_turn_trace="trace_1", + pack_consistent=True, + boundary_clean=boundary_clean, + polarity=polarity, + claim_domain=claim_domain, + evidence=evidence, + ) + + +# --------------------------------------------------------------------------- +# Eligibility gates +# --------------------------------------------------------------------------- + + +def test_undetermined_polarity_rejected(): + c = _enriched() + bad = replace(c, polarity="undetermined") + with pytest.raises(ProposalError, match="polarity"): + check_eligibility(bad) + + +def test_missing_corpus_evidence_rejected(): + c = _enriched(evidence=( + EvidencePointer( + source="pack", ref="light", + polarity="affirms", epistemic_status="coherent", + ), + )) + with pytest.raises(ProposalError, match="corpus"): + check_eligibility(c) + + +def test_evaluative_requires_explicit_flag(): + c = _enriched(claim_domain="evaluative") + with pytest.raises(ProposalError, match="evaluative"): + check_eligibility(c) + check_eligibility(c, allow_evaluative=True) # no raise + + +def test_boundary_unclean_rejected(): + c = _enriched(boundary_clean=False) + with pytest.raises(ProposalError, match="boundary"): + check_eligibility(c) + + +def test_incomplete_chain_rejected(): + base = _enriched() + incomplete = replace(base, proposed_chain={ + "subject": "light", "intent": "cause", + "connective": None, "object": None, + }) + with pytest.raises(ProposalError, match="subject/intent/connective/object"): + check_eligibility(incomplete) + + +# --------------------------------------------------------------------------- +# Proposal id idempotency +# --------------------------------------------------------------------------- + + +def test_proposal_id_is_deterministic(): + c = _enriched() + p1 = build_proposal(c) + p2 = build_proposal(c) + assert p1.proposal_id == p2.proposal_id + + +# --------------------------------------------------------------------------- +# Append-only log +# --------------------------------------------------------------------------- + + +def test_log_append_only_state_machine(tmp_path: Path): + log = ProposalLog(tmp_path / "proposals.jsonl") + c = _enriched() + p = build_proposal(c) + log.record_created(p) + assert log.find(p.proposal_id)["state"] == "pending" + + log.record_transition(p.proposal_id, "rejected", "test note") + assert log.find(p.proposal_id)["state"] == "rejected" + + # File is append-only: byte-count grows monotonically. + size_a = (tmp_path / "proposals.jsonl").stat().st_size + log.record_transition(p.proposal_id, "withdrawn", "no-op test") + size_b = (tmp_path / "proposals.jsonl").stat().st_size + assert size_b > size_a + + +# --------------------------------------------------------------------------- +# Replay gate (with fake replay to avoid running cognition lane) +# --------------------------------------------------------------------------- + + +def _fake_replay_equivalent(chain): + return ReplayEvidence( + baseline={"intent_accuracy": 1.0, "surface_groundedness": 1.0}, + candidate={"intent_accuracy": 1.0, "surface_groundedness": 1.0}, + regressed_metrics=(), + replay_equivalent=True, + ) + + +def _fake_replay_regression(chain): + return ReplayEvidence( + baseline={"intent_accuracy": 1.0, "surface_groundedness": 1.0}, + candidate={"intent_accuracy": 1.0, "surface_groundedness": 0.85}, + regressed_metrics=("surface_groundedness",), + replay_equivalent=False, + ) + + +def test_propose_from_candidate_pending_on_equivalent(tmp_path: Path): + log = ProposalLog(tmp_path / "proposals.jsonl") + c = _enriched() + proposal = propose_from_candidate(c, log=log, run_replay=_fake_replay_equivalent) + rec = log.find(proposal.proposal_id) + assert rec["state"] == "pending" + assert rec["replay_evidence"]["replay_equivalent"] is True + + +def test_propose_from_candidate_auto_rejects_on_regression(tmp_path: Path): + log = ProposalLog(tmp_path / "proposals.jsonl") + c = _enriched() + proposal = propose_from_candidate(c, log=log, run_replay=_fake_replay_regression) + rec = log.find(proposal.proposal_id) + assert rec["state"] == "rejected" + assert "auto_rollback_regression" in rec["operator_note"] + assert "surface_groundedness" in rec["operator_note"] + + +def test_propose_is_idempotent(tmp_path: Path): + log = ProposalLog(tmp_path / "proposals.jsonl") + c = _enriched() + propose_from_candidate(c, log=log, run_replay=_fake_replay_equivalent) + size_a = (tmp_path / "proposals.jsonl").stat().st_size + propose_from_candidate(c, log=log, run_replay=_fake_replay_equivalent) + size_b = (tmp_path / "proposals.jsonl").stat().st_size + # Idempotency: second proposal is a no-op; log size unchanged. + assert size_a == size_b + + +# --------------------------------------------------------------------------- +# Accept / reject / withdraw state machine +# --------------------------------------------------------------------------- + + +def test_accept_appends_to_corpus(tmp_path: Path): + log = ProposalLog(tmp_path / "proposals.jsonl") + corpus = tmp_path / "corpus.jsonl" + corpus.write_text("", encoding="utf-8") + c = _enriched() + proposal = propose_from_candidate(c, log=log, run_replay=_fake_replay_equivalent) + + chain_id = accept_proposal( + proposal.proposal_id, + log=log, + corpus_path=corpus, + review_date="2026-05-18", + operator_note="looks good", + ) + assert chain_id + + lines = [ln for ln in corpus.read_text().splitlines() if ln.strip()] + assert len(lines) == 1 + payload = json.loads(lines[0]) + assert payload["subject"] == "light" + assert payload["connective"] == "reveals" + assert "discovery_promoted" in payload["provenance"] + + rec = log.find(proposal.proposal_id) + assert rec["state"] == "accepted" + assert rec["accepted_chain_id"] == chain_id + + +def test_accept_refused_on_regression(tmp_path: Path): + log = ProposalLog(tmp_path / "proposals.jsonl") + corpus = tmp_path / "corpus.jsonl" + corpus.write_text("", encoding="utf-8") + c = _enriched() + proposal = propose_from_candidate(c, log=log, run_replay=_fake_replay_regression) + with pytest.raises(ProposalError): + accept_proposal( + proposal.proposal_id, log=log, + corpus_path=corpus, review_date="2026-05-18", + ) + + +def test_reject_and_withdraw_transitions(tmp_path: Path): + log = ProposalLog(tmp_path / "proposals.jsonl") + c = _enriched() + p1 = propose_from_candidate(c, log=log, run_replay=_fake_replay_equivalent) + reject_proposal(p1.proposal_id, log=log, operator_note="off doctrine") + assert log.find(p1.proposal_id)["state"] == "rejected" + + # Cannot transition from rejected. + with pytest.raises(ProposalError): + withdraw_proposal(p1.proposal_id, log=log) + + +def test_accept_idempotency_blocked_by_state_machine(tmp_path: Path): + log = ProposalLog(tmp_path / "proposals.jsonl") + corpus = tmp_path / "corpus.jsonl" + corpus.write_text("", encoding="utf-8") + c = _enriched() + proposal = propose_from_candidate(c, log=log, run_replay=_fake_replay_equivalent) + accept_proposal( + proposal.proposal_id, log=log, + corpus_path=corpus, review_date="2026-05-18", + ) + with pytest.raises(ProposalError): + accept_proposal( + proposal.proposal_id, log=log, + corpus_path=corpus, review_date="2026-05-18", + ) + + +# --------------------------------------------------------------------------- +# Trust boundary: replay gate does not touch active corpus +# --------------------------------------------------------------------------- + + +def test_replay_gate_does_not_mutate_active_corpus(): + """The real replay-equivalence gate runs the cognition lane; + that's slow, so this test runs it once and asserts byte-equality + on the active corpus. Marked separately so the rest of the + suite stays fast.""" + from teaching.replay import run_replay_equivalence + + chain = { + "subject": "judgment", "intent": "verification", + "connective": "requires", "object": "evidence", + } + before = _CORPUS_PATH.read_bytes() + evidence = run_replay_equivalence(chain) + after = _CORPUS_PATH.read_bytes() + assert before == after + assert isinstance(evidence.replay_equivalent, bool) + + +# --------------------------------------------------------------------------- +# append_chain_to_corpus — direct unit +# --------------------------------------------------------------------------- + + +def test_append_chain_writes_one_line(tmp_path: Path): + corpus = tmp_path / "c.jsonl" + corpus.write_text("", encoding="utf-8") + prov = Provenance( + adr_id="adr-0057", source="discovery_promoted", + review_date="2026-05-18", raw="adr-0057:discovery_promoted:2026-05-18", + ) + chain_id = append_chain_to_corpus( + {"subject": "knowledge", "intent": "cause", + "connective": "requires", "object": "evidence"}, + corpus_path=corpus, provenance=prov, + ) + payload = json.loads(corpus.read_text().splitlines()[0]) + assert payload["chain_id"] == chain_id + assert payload["provenance"] == prov.raw