core-labs/core
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

feat(adr-0057): Phase C2 — TeachingChainProposal + replay gate + review CLI

Browse source 
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 <candidate-jsonl-path> [--allow-evaluative]
    core teaching proposals [--state pending|accepted|rejected|withdrawn] [--json]
    core teaching review <proposal_id> --accept --review-date YYYY-MM-DD
    core teaching review <proposal_id> --reject [--note ...]
    core teaching review <proposal_id> --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 <noreply@anthropic.com>
This commit is contained in:
Shay 2026-05-18 10:23:14 -07:00
parent db6ce08589
commit e03ab4b609
7 changed files with 1397 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

187
core/cli.py
View file

@ -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 <candidate-jsonl-path>\n core teaching proposals --state pending\n core teaching review <proposal_id> --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",

243
docs/decisions/ADR-0057-teaching-chain-proposal-review.md Normal file
View file

@ -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: <metric list>"`.
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 <candidate_id> [--from-sink <path>]
Convert an eligible enriched DiscoveryCandidate into a
TeachingChainProposal. Runs the replay-equivalence gate
immediately. Idempotent on (candidate_id, chain payload).
core teaching proposals [--state <pending|accepted|rejected|withdrawn>] [--json]
List proposals; default lists pending.
core teaching review <proposal_id> --accept [--note "..."]
core teaching review <proposal_id> --reject [--note "..."]
core teaching review <proposal_id> --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.

1
docs/decisions/README.md
View file

@ -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) |
---

11
teaching/discovery.py
View file

@ -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 ()

484
teaching/proposals.py Normal file
View file

@ -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",
]

153
teaching/replay.py Normal file
View file

@ -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"]

321
tests/test_teaching_proposals.py Normal file
View file

@ -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