ab4c7cb0c3
* feat(epistemic): add first-class state enums * feat(epistemic): tag TurnEvent with state axes * feat(epistemic): serialize turn state axes * feat(packs): tag curated and inferred unit entries * feat(epistemic): expose word-level state on manifold * feat(epistemic): expose vault status mapping * feat(epistemic): preserve pack entry states through compiler * test(epistemic): cover phase 3 state tagging spine * feat(runtime): wire epistemic_state + normative_clearance into ChatResponse Add first-class epistemic_state and normative_clearance fields to ChatResponse (defaulting to "undetermined"/"unassessable" for backward compat). Import epistemic_state_for_grounding_source and clearance_from_verdicts into chat/runtime.py and populate both fields on the stub path (TurnEvent + ChatResponse) and the main path (TurnEvent + ChatResponse). Fix the test fixture to use "euro per hour" (a genuinely composed unit) instead of "dollars per hour" which is a curated lexicon entry and returns DECODED, not INFERRED. * test(cognition): update term_capture_rate baseline from 0.9167 to 1.0 unknown_logos_019 now correctly surfaces "light" as a pack-resident token near the logos versor — producing term_capture_rate 1.0 on both main and Phase 3. The 0.9167 pin was stale relative to a surface change already on main; Phase 3 did not introduce this shift.
437 lines
15 KiB
Python
437 lines
15 KiB
Python
"""Anti-regression demo — three scenes showing how CORE refuses to learn
|
|
something that would make it worse.
|
|
|
|
The thesis: when a system extends its own knowledge, **the gate that
|
|
decides what to admit is the load-bearing part** — not the proposer.
|
|
CORE's reviewed-corpus extension path (ADR-0057) has three independent
|
|
gates that must each pass before any byte is written:
|
|
|
|
S1. Eligibility predicate (mechanical, pre-replay).
|
|
Five mechanical checks on the candidate's shape (polarity,
|
|
evidence-floor, claim-domain, boundary-clean, chain-complete).
|
|
Ineligible candidates raise ``ProposalError`` and never enter
|
|
the proposal log.
|
|
|
|
S2. Replay-equivalence gate (mechanical, post-eligibility).
|
|
The full cognition lane runs against the active corpus AND
|
|
against a transient copy with the proposed chain appended.
|
|
Any strict-decrease in a watched metric auto-rejects the
|
|
proposal with the metrics named in the operator note.
|
|
Active corpus file bytes are byte-identical pre/post.
|
|
|
|
S3. Operator review (manual, post-replay).
|
|
Even a replay-equivalent proposal only reaches the *pending*
|
|
state — explicit ``core teaching review <id> --accept`` is
|
|
required to write to the active corpus.
|
|
|
|
This demo runs each scene end-to-end against the real ``ProposalLog``
|
|
in an isolated temp directory. No active corpus or production log is
|
|
touched.
|
|
|
|
Scenes 1 and 3 use the **real** ``teaching.replay.run_replay_equivalence``
|
|
function. Scene 2 injects a controlled replay function (via the
|
|
documented ``run_replay=`` kwarg of ``propose_from_candidate``) that
|
|
returns a regressed ``ReplayEvidence`` of the same shape the real gate
|
|
produces — demonstrating the auto-rejection lifecycle on a synthetic
|
|
regression deterministically. In production the real gate produces
|
|
this same shape when a real regression is detected.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import tempfile
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from teaching.discovery import DiscoveryCandidate, EvidencePointer
|
|
from teaching.proposals import (
|
|
ProposalError,
|
|
ProposalLog,
|
|
ReplayEvidence,
|
|
propose_from_candidate,
|
|
)
|
|
|
|
|
|
_VERBOSE = True
|
|
|
|
|
|
def _say(*args: Any, **kwargs: Any) -> None:
|
|
if _VERBOSE:
|
|
print(*args, **kwargs)
|
|
|
|
|
|
def _print_header(title: str, claim: str) -> None:
|
|
_say()
|
|
_say("─" * 72)
|
|
_say(f" {title}")
|
|
_say("─" * 72)
|
|
_say(f" CLAIM: {claim}")
|
|
_say()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Synthetic ReplayEvidence builder for Scene 2
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_regressed_replay(
|
|
*, regressed_metrics: tuple[str, ...]
|
|
) -> Any:
|
|
"""Return a ``run_replay`` function that emits a regressed
|
|
``ReplayEvidence`` with the same shape the real gate produces.
|
|
"""
|
|
baseline = {
|
|
"intent_accuracy": 1.0,
|
|
"surface_groundedness": 1.0,
|
|
"term_capture_rate": 1.0,
|
|
"versor_closure_rate": 1.0,
|
|
}
|
|
candidate = dict(baseline)
|
|
for m in regressed_metrics:
|
|
candidate[m] = round(candidate[m] - 0.0833, 4)
|
|
|
|
def _fn(chain: dict[str, Any]) -> ReplayEvidence: # noqa: ARG001
|
|
return ReplayEvidence(
|
|
baseline=baseline,
|
|
candidate=candidate,
|
|
regressed_metrics=tuple(sorted(regressed_metrics)),
|
|
replay_equivalent=False,
|
|
)
|
|
|
|
return _fn
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Candidate builders
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _candidate_undetermined() -> DiscoveryCandidate:
|
|
"""A candidate that fails the eligibility predicate at the polarity
|
|
gate. Used for Scene 1."""
|
|
return DiscoveryCandidate(
|
|
candidate_id="demo_undetermined_001",
|
|
proposed_chain={
|
|
"subject": "wisdom", "intent": "cause",
|
|
"connective": "informs", "object": "judgment",
|
|
},
|
|
trigger="would_have_grounded",
|
|
source_turn_trace="demo_trace_001",
|
|
pack_consistent=True,
|
|
boundary_clean=True,
|
|
polarity="undetermined",
|
|
claim_domain="factual",
|
|
evidence=(
|
|
EvidencePointer(
|
|
source="corpus",
|
|
ref="cause_wisdom_orders_judgment",
|
|
polarity="affirms",
|
|
epistemic_status="reviewed",
|
|
),
|
|
),
|
|
)
|
|
|
|
|
|
def _candidate_for_regression() -> DiscoveryCandidate:
|
|
"""A candidate that passes eligibility but (under the injected
|
|
regression replay) is auto-rejected for regressing
|
|
``surface_groundedness`` and ``term_capture_rate``."""
|
|
return DiscoveryCandidate(
|
|
candidate_id="demo_regression_002",
|
|
proposed_chain={
|
|
"subject": "knowledge", "intent": "cause",
|
|
"connective": "obscures", "object": "wisdom",
|
|
},
|
|
trigger="would_have_grounded",
|
|
source_turn_trace="demo_trace_002",
|
|
pack_consistent=True,
|
|
boundary_clean=True,
|
|
polarity="affirms",
|
|
claim_domain="factual",
|
|
evidence=(
|
|
EvidencePointer(
|
|
source="corpus",
|
|
ref="cause_knowledge_requires_evidence",
|
|
polarity="affirms",
|
|
epistemic_status="reviewed",
|
|
),
|
|
),
|
|
)
|
|
|
|
|
|
def _candidate_pass_through() -> DiscoveryCandidate:
|
|
"""A candidate that passes both eligibility and the real
|
|
replay-equivalence gate. Lands in ``pending`` awaiting
|
|
operator review."""
|
|
return DiscoveryCandidate(
|
|
candidate_id="demo_pass_003",
|
|
proposed_chain={
|
|
"subject": "judgment", "intent": "verification",
|
|
"connective": "requires", "object": "evidence",
|
|
},
|
|
trigger="would_have_grounded",
|
|
source_turn_trace="demo_trace_003",
|
|
pack_consistent=True,
|
|
boundary_clean=True,
|
|
polarity="affirms",
|
|
claim_domain="factual",
|
|
evidence=(
|
|
EvidencePointer(
|
|
source="corpus",
|
|
ref="verification_truth_requires_evidence",
|
|
polarity="affirms",
|
|
epistemic_status="reviewed",
|
|
),
|
|
),
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Scene results
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class SceneResult:
|
|
scene: str
|
|
claim: str
|
|
outcome: str
|
|
candidate_id: str
|
|
proposed_chain: dict[str, Any]
|
|
proposal_id: str | None
|
|
review_state: str
|
|
replay_evidence: dict[str, Any] | None
|
|
operator_note: str
|
|
error: str | None
|
|
corpus_byte_identical: bool
|
|
|
|
def as_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"scene": self.scene,
|
|
"claim": self.claim,
|
|
"outcome": self.outcome,
|
|
"candidate_id": self.candidate_id,
|
|
"proposed_chain": self.proposed_chain,
|
|
"proposal_id": self.proposal_id,
|
|
"review_state": self.review_state,
|
|
"replay_evidence": self.replay_evidence,
|
|
"operator_note": self.operator_note,
|
|
"error": self.error,
|
|
"corpus_byte_identical": self.corpus_byte_identical,
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class DemoReport:
|
|
scenes: tuple[SceneResult, ...]
|
|
all_gates_held: bool
|
|
active_corpus_byte_identical: bool
|
|
|
|
def as_dict(self) -> dict[str, Any]:
|
|
# ``all_claims_supported`` is the canonical cross-demo success
|
|
# field — added as an alias so operator tooling (and the CI gate)
|
|
# can rely on one uniform boolean key across every ``core demo``
|
|
# target. Existing fields are preserved for backwards compat.
|
|
return {
|
|
"scenes": [s.as_dict() for s in self.scenes],
|
|
"all_gates_held": self.all_gates_held,
|
|
"active_corpus_byte_identical": self.active_corpus_byte_identical,
|
|
"all_claims_supported": (
|
|
self.all_gates_held and self.active_corpus_byte_identical
|
|
),
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Scenes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _read_active_corpus_bytes() -> bytes:
|
|
from chat.teaching_grounding import _CORPUS_PATH
|
|
return _CORPUS_PATH.read_bytes() if _CORPUS_PATH.exists() else b""
|
|
|
|
|
|
def _scene1_eligibility_gate(log_path: Path) -> SceneResult:
|
|
_print_header(
|
|
"S1. Eligibility predicate refuses ineligible candidates",
|
|
"An undetermined-polarity candidate never enters the proposal "
|
|
"log. ProposalError raised; no log row; no replay invocation.",
|
|
)
|
|
log = ProposalLog(log_path)
|
|
candidate = _candidate_undetermined()
|
|
bytes_before = _read_active_corpus_bytes()
|
|
error: str | None = None
|
|
try:
|
|
propose_from_candidate(candidate, log=log)
|
|
except ProposalError as exc:
|
|
error = str(exc)
|
|
bytes_after = _read_active_corpus_bytes()
|
|
|
|
_say(f" candidate.polarity : {candidate.polarity}")
|
|
_say(f" outcome : ProposalError raised")
|
|
_say(f" error : {error}")
|
|
_say(f" proposal log rows : {len(log.current_state())}")
|
|
_say(f" active corpus byte-eq : {bytes_before == bytes_after}")
|
|
return SceneResult(
|
|
scene="S1_eligibility_gate",
|
|
claim=(
|
|
"Five mechanical eligibility gates fire before any replay "
|
|
"is invoked. Undetermined-polarity candidates never enter "
|
|
"the proposal log."
|
|
),
|
|
outcome="rejected_pre_replay",
|
|
candidate_id=candidate.candidate_id,
|
|
proposed_chain=candidate.proposed_chain,
|
|
proposal_id=None,
|
|
review_state="(not in log)",
|
|
replay_evidence=None,
|
|
operator_note="",
|
|
error=error,
|
|
corpus_byte_identical=(bytes_before == bytes_after),
|
|
)
|
|
|
|
|
|
def _scene2_replay_auto_reject(log_path: Path) -> SceneResult:
|
|
_print_header(
|
|
"S2. Replay-equivalence gate auto-rejects a regressing chain",
|
|
"An eligible candidate whose append would regress the cognition "
|
|
"lane is auto-rejected with the named regressed metrics in the "
|
|
"operator note. Active corpus byte-identical pre/post.",
|
|
)
|
|
log = ProposalLog(log_path)
|
|
candidate = _candidate_for_regression()
|
|
bytes_before = _read_active_corpus_bytes()
|
|
proposal = propose_from_candidate(
|
|
candidate,
|
|
log=log,
|
|
run_replay=_make_regressed_replay(
|
|
regressed_metrics=("surface_groundedness", "term_capture_rate"),
|
|
),
|
|
)
|
|
bytes_after = _read_active_corpus_bytes()
|
|
rec = log.find(proposal.proposal_id) or {}
|
|
ev = rec.get("replay_evidence") or {}
|
|
|
|
_say(f" proposal_id : {proposal.proposal_id}")
|
|
_say(f" baseline metrics : {ev.get('baseline')}")
|
|
_say(f" candidate metrics : {ev.get('candidate')}")
|
|
_say(f" regressed_metrics : {ev.get('regressed_metrics')}")
|
|
_say(f" replay_equivalent : {ev.get('replay_equivalent')}")
|
|
_say(f" state : {rec.get('state')}")
|
|
_say(f" operator_note : {rec.get('operator_note')}")
|
|
_say(f" active corpus byte-eq : {bytes_before == bytes_after}")
|
|
return SceneResult(
|
|
scene="S2_replay_auto_reject",
|
|
claim=(
|
|
"Replay-equivalence gate compares the full cognition lane "
|
|
"metrics; any strict-decrease auto-rejects with the regressed "
|
|
"metric names in the operator note. Active corpus untouched."
|
|
),
|
|
outcome="auto_rejected_on_regression",
|
|
candidate_id=candidate.candidate_id,
|
|
proposed_chain=candidate.proposed_chain,
|
|
proposal_id=proposal.proposal_id,
|
|
review_state=str(rec.get("state")),
|
|
replay_evidence=ev,
|
|
operator_note=str(rec.get("operator_note") or ""),
|
|
error=None,
|
|
corpus_byte_identical=(bytes_before == bytes_after),
|
|
)
|
|
|
|
|
|
def _scene3_real_gate_pass_through(log_path: Path) -> SceneResult:
|
|
_print_header(
|
|
"S3. Real replay gate runs cognition lane; pass → pending",
|
|
"An eligible candidate whose append does not regress reaches "
|
|
"'pending' state. Operator --accept is still required to write "
|
|
"to the active corpus; the gate is a precondition, not a "
|
|
"permission.",
|
|
)
|
|
log = ProposalLog(log_path)
|
|
candidate = _candidate_pass_through()
|
|
bytes_before = _read_active_corpus_bytes()
|
|
proposal = propose_from_candidate(candidate, log=log)
|
|
bytes_after = _read_active_corpus_bytes()
|
|
rec = log.find(proposal.proposal_id) or {}
|
|
ev = rec.get("replay_evidence") or {}
|
|
|
|
_say(f" proposal_id : {proposal.proposal_id}")
|
|
_say(f" baseline metrics : {ev.get('baseline')}")
|
|
_say(f" candidate metrics : {ev.get('candidate')}")
|
|
_say(f" regressed_metrics : {ev.get('regressed_metrics')}")
|
|
_say(f" replay_equivalent : {ev.get('replay_equivalent')}")
|
|
_say(f" state : {rec.get('state')}")
|
|
_say(f" next step : core teaching review {proposal.proposal_id} "
|
|
"--accept --review-date YYYY-MM-DD")
|
|
_say(f" active corpus byte-eq : {bytes_before == bytes_after}")
|
|
return SceneResult(
|
|
scene="S3_real_gate_pass_through",
|
|
claim=(
|
|
"A replay-equivalent candidate reaches 'pending' but is "
|
|
"not auto-applied. Operator --accept is the third gate."
|
|
),
|
|
outcome="pending_awaiting_operator",
|
|
candidate_id=candidate.candidate_id,
|
|
proposed_chain=candidate.proposed_chain,
|
|
proposal_id=proposal.proposal_id,
|
|
review_state=str(rec.get("state")),
|
|
replay_evidence=ev,
|
|
operator_note="",
|
|
error=None,
|
|
corpus_byte_identical=(bytes_before == bytes_after),
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Public entry point
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def run_demo(*, emit_json: bool = False) -> dict[str, Any]:
|
|
"""Run all three scenes and return a structured report."""
|
|
global _VERBOSE
|
|
_VERBOSE = not emit_json
|
|
|
|
active_bytes_before = _read_active_corpus_bytes()
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
log_path = Path(tmpdir) / "demo_proposals.jsonl"
|
|
s1 = _scene1_eligibility_gate(log_path)
|
|
s2 = _scene2_replay_auto_reject(log_path)
|
|
s3 = _scene3_real_gate_pass_through(log_path)
|
|
|
|
active_bytes_after = _read_active_corpus_bytes()
|
|
|
|
scenes = (s1, s2, s3)
|
|
all_gates_held = (
|
|
s1.outcome == "rejected_pre_replay"
|
|
and s2.outcome == "auto_rejected_on_regression"
|
|
and s3.outcome == "pending_awaiting_operator"
|
|
)
|
|
report = DemoReport(
|
|
scenes=scenes,
|
|
all_gates_held=all_gates_held,
|
|
active_corpus_byte_identical=(active_bytes_before == active_bytes_after),
|
|
)
|
|
|
|
if _VERBOSE:
|
|
_say()
|
|
_say("═" * 72)
|
|
_say(" RESULT")
|
|
_say("═" * 72)
|
|
_say(f" all three gates held : {report.all_gates_held}")
|
|
_say(f" active corpus byte-eq : {report.active_corpus_byte_identical}")
|
|
_say()
|
|
_say(
|
|
" Each gate is independent and fails closed. Bad proposals "
|
|
"stop at the cheapest applicable gate. The active corpus is "
|
|
"never written to anywhere in this demo."
|
|
)
|
|
_say()
|
|
|
|
return report.as_dict()
|
|
|
|
|
|
__all__ = ["run_demo"]
|