From 05ae8cb1b4956595761285a920eae2b1765afe9e Mon Sep 17 00:00:00 2001 From: Shay Date: Fri, 17 Jul 2026 23:56:01 -0700 Subject: [PATCH] feat(ring2-ring3): multi-port residual protocol + integrity-coordinated handoff MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rings 2 and 3 of the ADR-0246 preflight §9, built exactly as the brief bounds them. ADR-0247 + ADR-0248 both **Proposed** — no self-Accept. Ring 2 (core/ports/residual_protocol.py — ADR-0247): the 7-stage shared control grammar: witness -> typed residual decomposition -> permitted operator selection -> bounded operation or abstention -> re-certification -> action decision -> append-only replay record. Port-agnostic (no registry, no unified scheduler — §7 non-goal #2 honored); v1 operators are ZERO-BOUND only (nonzero fails closed — no-silent-correction doctrine); unaccounted residual fails closed; re-certification RAISES on witness drift during a zero-bound pass; replay chain is append-only, full-SHA-256 content-addressed, tamper-evident (verify_replay_chain). NOTE: named core/ports/ because core/protocol/ is the existing CTP v0 wire format (collision checked before naming). Ring 2 adapters (core/ports/adapters.py): IdentityPort (ADR-0246 grade-1 geometry; typed channels travel IN the witness so decompose is a pure re-shaping, no hidden state; admit = evaluate_admission, single source of truth) + PrecisionPort (ADR-0244 §2.5 cast transport; ServingState subjects) — two genuinely non-identical native geometries + a synthetic third port proving grammar agnosticism. Ring 3 (core/ports/integrity_handoff.py — ADR-0248): the coordination seam ('Integrity coordinates handoffs; it does not replace content-bearing cognition'): fuses Ring-2 port decisions + existing EpistemicState/NormativeClearance into content-free proceed/hedge/abstain (conjunctive, strongest-restriction-wins; fail-closed on missing/invalid evidence; binds replay-chain tip digest; handoff itself content-addressed). Weak epistemic standing hedges (mirrors hedge doctrine) — only integrity violations silence a turn. OBSERVE-ONLY: no serve consumer yet; that is a future flag-gated unit. Remaining Ring-3 programme honestly listed open (world-model, governed-learning consumption, discourse widening). Both pure/deterministic/off-serving (A-04 pinned); brief §0a/§9 updated. [Verification]: uv run core test --suite smoke -q => 176 passed; Ring 2/3 + ADR-0246 suites + D4 gate surfaces => 125 passed (28 new Ring-2/3 pins); run log docs/audit/artifacts/ring2-ring3-runlog.txt --- core/ports/__init__.py | 16 + core/ports/adapters.py | 168 +++++++++ core/ports/integrity_handoff.py | 151 +++++++++ core/ports/residual_protocol.py | 319 ++++++++++++++++++ .../ADR-0247-multi-port-residual-protocol.md | 71 ++++ ...ADR-0248-integrity-coordinated-handoffs.md | 77 +++++ docs/audit/artifacts/ring2-ring3-runlog.txt | 11 + ...ity-action-and-path-integrity-preflight.md | 7 + tests/test_ring2_residual_protocol.py | 271 +++++++++++++++ tests/test_ring3_integrity_handoff.py | 170 ++++++++++ 10 files changed, 1261 insertions(+) create mode 100644 core/ports/__init__.py create mode 100644 core/ports/adapters.py create mode 100644 core/ports/integrity_handoff.py create mode 100644 core/ports/residual_protocol.py create mode 100644 docs/adr/ADR-0247-multi-port-residual-protocol.md create mode 100644 docs/adr/ADR-0248-integrity-coordinated-handoffs.md create mode 100644 docs/audit/artifacts/ring2-ring3-runlog.txt create mode 100644 tests/test_ring2_residual_protocol.py create mode 100644 tests/test_ring3_integrity_handoff.py diff --git a/core/ports/__init__.py b/core/ports/__init__.py new file mode 100644 index 00000000..a42ebe09 --- /dev/null +++ b/core/ports/__init__.py @@ -0,0 +1,16 @@ +"""core.ports — Ring 2/3 multi-port residual protocol + integrity handoff. + +The shared control grammar of the ADR-0246 preflight §9 (Ring 2) and the +integrity-handoff coordinator (Ring 3): an INTERCONNECT GRAMMAR between CORE's +organs — never a second physics, never a unified scheduler, never a content +generator. Ports keep their non-identical native geometry; the grammar only +standardizes witness → typed residual decomposition → permitted operator +selection → bounded operation or abstention → re-certification → action +decision → append-only replay record. + +Distinct from :mod:`core.protocol` (the CORE Trace Protocol v0 wire format) — +that package serializes trace events; this one governs residual decisions. + +Every module here is pure, deterministic, f64, content-addressed +(full SHA-256, canonical JSON, no ``default=str``), and off-serving. +""" diff --git a/core/ports/adapters.py b/core/ports/adapters.py new file mode 100644 index 00000000..59515fd6 --- /dev/null +++ b/core/ports/adapters.py @@ -0,0 +1,168 @@ +"""core.ports.adapters — real port adapters for the Ring-2 residual protocol. + +Two shipped ports with genuinely NON-identical native geometry (preflight §9: +"Ports retain non-identical native geometry"): + +* :class:`IdentityPort` — ADR-0246 grade-1 frame preservation. The witness + carries the §3.7 admit-surface scalars AND the §3.6 typed residual energies + (they are all measurements of the subject, so they belong in the witness — + this also keeps ``decompose`` a pure re-shaping of witness content, with no + hidden adapter state to corrupt a replay). The admit verdict is + ``evaluate_admission`` — the single source of truth; the adapter adds no + policy of its own. +* :class:`PrecisionPort` — ADR-0244 §2.5 / ADR-0245 §2.2 f64→f32 cast + transport. Witness = measured cast round-trip error + f32 unit-norm + deviation; admit = both within tolerance. Subjects come from a certified + ``ServingState`` (via :meth:`PrecisionSubject.from_serving_state`) or are + built directly in evals. + +Future adapters (Atlas, Evidence, Temporal/causal, Articulation, Action) plug +in the same way — the grammar has no registry and needs no change (proven by +the synthetic third port in the test suite). + +Pure, deterministic, off-serving. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +import numpy as np + +from core.physics.identity_action import AdmissionPolicy, evaluate_admission +from core.physics.identity_manifold import IdentityManifoldGeometry +from core.ports.residual_protocol import ( + PortWitness, + ResidualDecomposition, +) + + +class IdentityPort: + """ADR-0246 identity organ speaking the Ring-2 grammar. + + Subject: a live versor ``F`` (32-component Cl(4,1) array). + """ + + port_id = "identity" + schema_version = "identity_action_v1" + + def __init__( + self, geometry: IdentityManifoldGeometry, policy: AdmissionPolicy + ) -> None: + self._geometry = geometry + self._policy = policy + + def witness(self, subject: Any) -> PortWitness: + versor = np.asarray(subject, dtype=np.float64) + result = evaluate_admission(self._geometry, versor, self._policy) + channels = self._geometry.typed_residual_energy(versor) + return PortWitness( + port_id=self.port_id, + schema_version=self.schema_version, + measurements=( + ("d_orth", result.d_orth), + ("d_stab", result.d_stab), + ("leakage_rms", result.leakage_rms), + ("max_leakage", result.max_leakage), + ("min_self_alignment", result.min_self_alignment), + ("ch_null_or_conformal", channels["null_or_conformal"]), + ("ch_boost_like", channels["boost_like"]), + ("ch_spatial_foreign", channels["spatial_foreign"]), + ("ch_unclassified", channels["unclassified"]), + ), + ) + + def decompose(self, witness: PortWitness) -> ResidualDecomposition: + values = dict(witness.measurements) + return ResidualDecomposition( + port_id=self.port_id, + channels=( + ("null_or_conformal", values["ch_null_or_conformal"]), + ("boost_like", values["ch_boost_like"]), + ("spatial_foreign", values["ch_spatial_foreign"]), + ("unclassified", values["ch_unclassified"]), + ), + unaccounted=values["ch_unclassified"], + unaccounted_tol=self._policy.unclassified_tol, + ) + + def admit( + self, subject: Any, decomposition: ResidualDecomposition + ) -> tuple[bool, tuple[str, ...]]: + result = evaluate_admission( + self._geometry, np.asarray(subject, dtype=np.float64), self._policy + ) + return result.admitted, tuple(result.refusal_reasons) + + +@dataclass(frozen=True) +class PrecisionSubject: + """The Precision port's subject: a cast-transport measurement pair. + + Built from a certified :class:`~core.physics.cognitive_lifecycle.ServingState` + (duck-typed via :meth:`from_serving_state` so this module stays light) or + directly in evals/tests. + """ + + cast_error: float + unit_norm_f32: float + source_psi_digest: str = "" + certificate_id: str = "" + + @classmethod + def from_serving_state(cls, state: Any) -> "PrecisionSubject": + return cls( + cast_error=float(getattr(state, "cast_error")), + unit_norm_f32=float(getattr(state, "unit_norm_f32")), + source_psi_digest=str(getattr(state, "source_psi_digest", "")), + certificate_id=str(getattr(state, "certificate_id", "")), + ) + + +class PrecisionPort: + """ADR-0244 §2.5 cast-transport organ speaking the Ring-2 grammar. + + Native geometry: float-precision transport (round-trip error and unit-norm + preservation across the governed f64→f32 boundary) — nothing like the + identity port's grade-1 subspace geometry, which is exactly the point of a + multi-port grammar. + """ + + port_id = "precision" + schema_version = "serving_cast_v1" + + def __init__(self, tol: float) -> None: + self._tol = float(tol) + + def witness(self, subject: PrecisionSubject) -> PortWitness: + return PortWitness( + port_id=self.port_id, + schema_version=self.schema_version, + measurements=( + ("cast_error", float(subject.cast_error)), + ("unit_norm_deviation", abs(1.0 - float(subject.unit_norm_f32))), + ), + ) + + def decompose(self, witness: PortWitness) -> ResidualDecomposition: + values = dict(witness.measurements) + return ResidualDecomposition( + port_id=self.port_id, + channels=( + ("cast_error", values["cast_error"]), + ("unit_norm_deviation", values["unit_norm_deviation"]), + ), + unaccounted=0.0, # both transport channels are fully typed + unaccounted_tol=1e-12, + ) + + def admit( + self, subject: PrecisionSubject, decomposition: ResidualDecomposition + ) -> tuple[bool, tuple[str, ...]]: + reasons: list[str] = [] + if float(subject.cast_error) > self._tol: + reasons.append(f"cast_error>{self._tol:.3g}") + if abs(1.0 - float(subject.unit_norm_f32)) > self._tol: + reasons.append(f"unit_norm_deviation>{self._tol:.3g}") + return (not reasons), tuple(reasons) diff --git a/core/ports/integrity_handoff.py b/core/ports/integrity_handoff.py new file mode 100644 index 00000000..89e1f561 --- /dev/null +++ b/core/ports/integrity_handoff.py @@ -0,0 +1,151 @@ +"""core.ports.integrity_handoff — Ring 3 integrity-coordinated handoff (ADR-0248). + +The preflight §9 fixes Ring 3's boundary: "Integrity coordinates handoffs; it +does not replace content-bearing cognition." This module is exactly that +coordination seam — a pure function that fuses: + + * the Ring-2 per-port action decisions (from the append-only replay chain), + * the content organs' EXISTING epistemic standing (:class:`EpistemicState`), + * the turn's EXISTING normative clearance (:class:`NormativeClearance`), + +into one typed routing decision — ``proceed`` / ``hedge`` / ``abstain`` — for +discourse-planning and composition consumers. It carries NO content fields: +what gets said (if anything) remains the content organs' job; this only +decides whether integrity permits saying it plainly, qualified, or not at all. + +Fusion rules (deterministic, conservative — the integrity floor is +conjunctive; the strongest restriction wins): + + 1. no port evidence, or an unverifiable replay chain → ABSTAIN (fail-closed) + 2. any port abstained → ABSTAIN + 3. normative VIOLATED or SUPPRESSED → ABSTAIN + 4. normative UNASSESSABLE → at most HEDGE + 5. weak epistemic standing (undetermined / unverified / ambiguous / + contradicted / needs-state / scope- or compute-bounded) → at most HEDGE + 6. otherwise → PROCEED + +HEDGE (not ABSTAIN) for weak standing mirrors the existing hedge-injection +doctrine: qualified content may still surface; only integrity violations +silence a turn. The handoff binds the replay-chain tip digest so every routing +decision is auditable back to each port's recorded evidence. + +Pure, deterministic, off-serving. Observe-only: nothing at serve consumes this +yet; wiring it into discourse planning is a future, flag-gated unit. +""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass +from typing import Any, Sequence + +from core.epistemic_state import EpistemicState, NormativeClearance +from core.ports.residual_protocol import ( + ACTION_ABSTAIN, + ReplayRecord, + verify_replay_chain, +) + +HANDOFF_PROCEED: str = "proceed" +HANDOFF_HEDGE: str = "hedge" +HANDOFF_ABSTAIN: str = "abstain" + +# Epistemic states whose standing is too weak for an unqualified PROCEED — +# content may still surface, hedged (mirrors the ADR-0028/0031 hedge doctrine). +_WEAK_EPISTEMIC_STATES: frozenset[EpistemicState] = frozenset( + { + EpistemicState.UNDETERMINED, + EpistemicState.UNVERIFIED_POSSIBLE, + EpistemicState.UNVERIFIED_NOVEL, + EpistemicState.AMBIGUOUS, + EpistemicState.CONTRADICTED, + EpistemicState.EPISTEMIC_STATE_NEEDED, + EpistemicState.SCOPE_BOUNDARY, + EpistemicState.COMPUTATIONALLY_BOUNDED, + } +) + + +@dataclass(frozen=True) +class IntegrityHandoff: + """The typed routing decision integrity hands to content consumers. + + Deliberately content-free: routing, attribution, and digests only — + Ring 3's contract is coordination, never generation. + """ + + handoff: str # HANDOFF_PROCEED | HANDOFF_HEDGE | HANDOFF_ABSTAIN + reasons: tuple[str, ...] + port_actions: tuple[tuple[str, str], ...] # (port_id, action) in chain order + chain_tip_digest: str + epistemic_state: str + normative_clearance: str + + def as_dict(self) -> dict[str, Any]: + return { + "handoff": self.handoff, + "reasons": list(self.reasons), + "port_actions": [[p, a] for p, a in self.port_actions], + "chain_tip_digest": self.chain_tip_digest, + "epistemic_state": self.epistemic_state, + "normative_clearance": self.normative_clearance, + } + + def handoff_digest(self) -> str: + canonical = json.dumps( + self.as_dict(), ensure_ascii=False, sort_keys=True, separators=(",", ":") + ) + return hashlib.sha256(canonical.encode("utf-8")).hexdigest() + + +def coordinate_handoff( + chain: Sequence[ReplayRecord], + *, + epistemic_state: EpistemicState, + normative_clearance: NormativeClearance, +) -> IntegrityHandoff: + """Fuse port decisions + content-organ standing into one routing decision. + + See the module docstring for the rule order. Conservative by construction: + the strongest restriction wins, and missing/unverifiable evidence abstains. + """ + port_actions = tuple((r.port_id, r.action) for r in chain) + reasons: list[str] = [] + level = HANDOFF_PROCEED + + if not chain: + level = HANDOFF_ABSTAIN + reasons.append("no_port_evidence") + elif not verify_replay_chain(chain): + level = HANDOFF_ABSTAIN + reasons.append("replay_chain_invalid") + else: + for record in chain: + if record.action == ACTION_ABSTAIN: + level = HANDOFF_ABSTAIN + for reason in record.reasons or ("unspecified",): + reasons.append(f"port:{record.port_id}:{reason}") + if normative_clearance in ( + NormativeClearance.VIOLATED, + NormativeClearance.SUPPRESSED, + ): + level = HANDOFF_ABSTAIN + reasons.append(f"normative:{normative_clearance.value}") + elif normative_clearance is NormativeClearance.UNASSESSABLE: + if level == HANDOFF_PROCEED: + level = HANDOFF_HEDGE + reasons.append("normative:unassessable") + if epistemic_state in _WEAK_EPISTEMIC_STATES: + if level == HANDOFF_PROCEED: + level = HANDOFF_HEDGE + reasons.append(f"epistemic:{epistemic_state.value}") + + return IntegrityHandoff( + handoff=level, + reasons=tuple(reasons), + port_actions=port_actions, + chain_tip_digest=chain[-1].record_digest() if chain else "", + epistemic_state=epistemic_state.value, + normative_clearance=normative_clearance.value, + ) diff --git a/core/ports/residual_protocol.py b/core/ports/residual_protocol.py new file mode 100644 index 00000000..53a16ae5 --- /dev/null +++ b/core/ports/residual_protocol.py @@ -0,0 +1,319 @@ +"""core.ports.residual_protocol — Ring 2 shared control grammar (ADR-0247). + +The seven-stage residual protocol the ADR-0246 preflight §9 fixes for Ring 2: + + witness → typed residual decomposition → permitted operator selection + → bounded operation or abstention → re-certification + → articulation/action decision → append-only replay record + +Design commitments (the engineering pillars, applied): + +* **Port-agnostic grammar, port-native geometry.** The protocol never + interprets a port's measurements — Identity speaks grade-1 frame + preservation, Precision speaks f32 cast transport, a future Atlas port + speaks packing residuals. The grammar only sequences the stages and records + the evidence. There is NO port registry and NO unified scheduler (preflight + §7 non-goal #2): callers invoke the protocol per port, per subject. +* **Abstain-or-proceed v1.** The only operators permitted are zero-bound + (``proceed_unmodified`` / ``abstain``). A nonzero-bound operator — anything + that would *modify* the subject to green a metric — fails closed with a + typed error until a dedicated ADR ratifies a bounded corrector + (no-silent-correction doctrine; mirrors ADR-0244/0246 admit-or-abstain). +* **Fail-closed on unaccounted residual.** Energy the port's typed channels + cannot account for forces abstention; no correction policy ever attaches to + the unclassified remainder (ADR-0246 §3.6 doctrine, generalized). +* **Re-certification.** After the (zero-bound) operation the witness is + re-measured; any drift means the port mutated state during a supposedly + non-mutating pass — a protocol violation, not a policy question — and + raises rather than recording a corrupted replay. +* **Append-only, content-addressed replay.** Every run appends exactly one + ``ReplayRecord`` whose full-SHA-256 digest chains to its predecessor + (ADR-0245 §2.3 semantic rigor: canonical JSON, no ``default=str``, no + truncation). ``verify_replay_chain`` re-derives the chain from content. + +Pure, deterministic, off-serving. No wall-clock, no randomness. +""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass +from typing import Any, Protocol, Sequence + +GENESIS_DIGEST: str = "0" * 64 + +ACTION_PROCEED: str = "proceed" +ACTION_ABSTAIN: str = "abstain" + +OPERATOR_PROCEED_ID: str = "proceed_unmodified" +OPERATOR_ABSTAIN_ID: str = "abstain" + + +class ResidualProtocolError(ValueError): + """Typed protocol violation — always fail-closed, never a warning.""" + + +def _canonical(payload: Any) -> str: + return json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + + +def _digest(payload: Any) -> str: + return hashlib.sha256(_canonical(payload).encode("utf-8")).hexdigest() + + +@dataclass(frozen=True) +class PortWitness: + """Stage 1 — a port's raw, typed measurement of its subject. + + ``measurements`` is a tuple of ``(name, value)`` pairs, canonicalized to + lexicographic order at construction so two witnesses of the same physical + measurement are digest-identical regardless of emission order. + """ + + port_id: str + schema_version: str + measurements: tuple[tuple[str, float], ...] + + def __post_init__(self) -> None: + canonical = tuple( + sorted((str(k), float(v)) for k, v in self.measurements) + ) + object.__setattr__(self, "measurements", canonical) + if not self.port_id: + raise ResidualProtocolError("witness requires a port_id") + + def as_dict(self) -> dict[str, Any]: + return { + "port_id": self.port_id, + "schema_version": self.schema_version, + "measurements": [[k, v] for k, v in self.measurements], + } + + def digest(self) -> str: + return _digest(self.as_dict()) + + +@dataclass(frozen=True) +class ResidualDecomposition: + """Stage 2 — the witness re-expressed on the port's typed channels. + + ``unaccounted`` is the residual energy the typed channels cannot explain; + above ``unaccounted_tol`` the protocol abstains (fail-closed — the + generalized "no correction policy on the unclassified channel"). + """ + + port_id: str + channels: tuple[tuple[str, float], ...] + unaccounted: float + unaccounted_tol: float + + def __post_init__(self) -> None: + canonical = tuple(sorted((str(k), float(v)) for k, v in self.channels)) + object.__setattr__(self, "channels", canonical) + + @property + def fail_closed(self) -> bool: + return float(self.unaccounted) > float(self.unaccounted_tol) + + def as_dict(self) -> dict[str, Any]: + return { + "port_id": self.port_id, + "channels": [[k, v] for k, v in self.channels], + "unaccounted": float(self.unaccounted), + "unaccounted_tol": float(self.unaccounted_tol), + } + + +@dataclass(frozen=True) +class PermittedOperator: + """Stage 3 element — an operator a port's policy permits. + + ``bound`` is the maximum magnitude of the operation. v1 accepts ONLY + ``bound == 0.0`` (non-mutating): any nonzero bound requires a future + dedicated ADR (bounded-corrector), and the protocol fails closed on it. + """ + + operator_id: str + bound: float = 0.0 + + def as_dict(self) -> dict[str, Any]: + return {"operator_id": self.operator_id, "bound": float(self.bound)} + + +_DEFAULT_OPERATORS: tuple[PermittedOperator, ...] = ( + PermittedOperator(operator_id=OPERATOR_PROCEED_ID, bound=0.0), + PermittedOperator(operator_id=OPERATOR_ABSTAIN_ID, bound=0.0), +) + + +@dataclass(frozen=True) +class ActionDecision: + """Stage 6 — the articulation/action decision handed to the caller.""" + + action: str # ACTION_PROCEED | ACTION_ABSTAIN + reasons: tuple[str, ...] = () + + def as_dict(self) -> dict[str, Any]: + return {"action": self.action, "reasons": list(self.reasons)} + + +class ResidualPort(Protocol): + """The structural contract a port must satisfy (no registry — duck-typed). + + Ports keep their native geometry: the grammar calls these hooks and never + interprets the measurements beyond the fail-closed ``unaccounted`` check. + ``permitted_operators`` is optional; absent, the v1 default + (proceed_unmodified / abstain) applies. + """ + + port_id: str + schema_version: str + + def witness(self, subject: Any) -> PortWitness: ... + def decompose(self, witness: PortWitness) -> ResidualDecomposition: ... + def admit( + self, subject: Any, decomposition: ResidualDecomposition + ) -> tuple[bool, tuple[str, ...]]: ... + + +@dataclass(frozen=True) +class ReplayRecord: + """Stage 7 — one append-only, content-addressed record of a protocol run. + + ``record_digest()`` covers every stage payload plus the predecessor's + digest, so any tamper anywhere breaks the chain from that point forward. + """ + + sequence_index: int + port_id: str + schema_version: str + witness: dict[str, Any] + decomposition: dict[str, Any] + operator: dict[str, Any] + recertified: bool + action: str + reasons: tuple[str, ...] + prev_record_digest: str + + def _payload(self) -> dict[str, Any]: + return { + "sequence_index": int(self.sequence_index), + "port_id": self.port_id, + "schema_version": self.schema_version, + "witness": self.witness, + "decomposition": self.decomposition, + "operator": self.operator, + "recertified": bool(self.recertified), + "action": self.action, + "reasons": list(self.reasons), + "prev_record_digest": self.prev_record_digest, + } + + def record_digest(self) -> str: + return _digest(self._payload()) + + def as_dict(self) -> dict[str, Any]: + out = self._payload() + out["record_digest"] = self.record_digest() + return out + + +def run_residual_protocol( + port: ResidualPort, + subject: Any, + chain: Sequence[ReplayRecord], +) -> tuple[tuple[ReplayRecord, ...], ActionDecision]: + """Run the seven-stage grammar for one port over one subject. + + Returns ``(chain + one new record, decision)``. Deterministic; the chain is + never mutated (a new tuple is returned). Raises + :class:`ResidualProtocolError` on protocol violations (nonzero-bound + operator; witness drift across a zero-bound operation; port_id mismatch) — + violations are never recorded as if they were decisions. + """ + # Stage 1 — witness + witness = port.witness(subject) + if witness.port_id != port.port_id: + raise ResidualProtocolError( + f"witness port_id {witness.port_id!r} != port {port.port_id!r}" + ) + # Stage 2 — typed residual decomposition + decomposition = port.decompose(witness) + if decomposition.port_id != port.port_id: + raise ResidualProtocolError( + f"decomposition port_id {decomposition.port_id!r} != port {port.port_id!r}" + ) + # Stage 3 — permitted operator selection (v1: zero-bound only) + permitted_hook = getattr(port, "permitted_operators", None) + operators = ( + tuple(permitted_hook(decomposition)) if permitted_hook is not None + else _DEFAULT_OPERATORS + ) + for op in operators: + if float(op.bound) != 0.0: + raise ResidualProtocolError( + f"nonzero_bound_operator_not_ratified: {op.operator_id!r} " + f"(bound={op.bound}) — bounded correctors require a dedicated ADR" + ) + # Stage 4 — bounded operation OR abstention (v1 operators are non-mutating) + reasons: list[str] = [] + if decomposition.fail_closed: + admitted = False + reasons.append( + f"unaccounted_residual_exceeds_tol:{decomposition.unaccounted:.6g}" + f">{decomposition.unaccounted_tol:.6g}" + ) + else: + admitted, admit_reasons = port.admit(subject, decomposition) + reasons.extend(admit_reasons) + proceed_permitted = any(op.operator_id == OPERATOR_PROCEED_ID for op in operators) + if admitted and proceed_permitted: + selected = next(op for op in operators if op.operator_id == OPERATOR_PROCEED_ID) + action = ACTION_PROCEED + else: + selected = next( + (op for op in operators if op.operator_id == OPERATOR_ABSTAIN_ID), + PermittedOperator(operator_id=OPERATOR_ABSTAIN_ID, bound=0.0), + ) + action = ACTION_ABSTAIN + if admitted and not proceed_permitted: + reasons.append("proceed_not_in_permitted_operators") + # Stage 5 — re-certification: a zero-bound operation must not move the witness + recheck = port.witness(subject) + if recheck.digest() != witness.digest(): + raise ResidualProtocolError( + "recertification_witness_changed: port mutated state during a " + "zero-bound operation — protocol violation, refusing to record" + ) + # Stage 6 — action decision + decision = ActionDecision(action=action, reasons=tuple(reasons)) + # Stage 7 — append-only replay record + prev = chain[-1].record_digest() if chain else GENESIS_DIGEST + record = ReplayRecord( + sequence_index=len(chain), + port_id=port.port_id, + schema_version=port.schema_version, + witness=witness.as_dict(), + decomposition=decomposition.as_dict(), + operator=selected.as_dict(), + recertified=True, + action=action, + reasons=tuple(reasons), + prev_record_digest=prev, + ) + return tuple(chain) + (record,), decision + + +def verify_replay_chain(records: Sequence[ReplayRecord]) -> bool: + """True iff the chain is intact: indices contiguous from 0 and every + ``prev_record_digest`` equals the recomputed digest of its predecessor + (genesis for the first). Any payload tamper changes a recomputed digest and + breaks the link that stored the original.""" + for i, record in enumerate(records): + if record.sequence_index != i: + return False + expected_prev = GENESIS_DIGEST if i == 0 else records[i - 1].record_digest() + if record.prev_record_digest != expected_prev: + return False + return True diff --git a/docs/adr/ADR-0247-multi-port-residual-protocol.md b/docs/adr/ADR-0247-multi-port-residual-protocol.md new file mode 100644 index 00000000..e8fdf2fe --- /dev/null +++ b/docs/adr/ADR-0247-multi-port-residual-protocol.md @@ -0,0 +1,71 @@ +# ADR-0247: Multi-Port Residual Protocol — the Ring-2 Shared Control Grammar + +**Status**: **Proposed** — pending explicit human ratification (no self-Accept) +**Date**: 2026-07-17 +**Authors**: Joshua Shay + multi-model R&D (implemented Fable 5) +**Depends on**: ADR-0244/0245 (Accepted), ADR-0246 (Proposed — first port consumer) +**Preflight authority**: ADR-0246 preflight brief §9 (Ring 2), §7 non-goal #2 + +--- + +## 1. Decision + +Implement Ring 2 exactly as the preflight bounds it — a **shared control +grammar only** — as the pure package `core/ports/`: + +```text +witness → typed residual decomposition → permitted operator selection + → bounded operation or abstention → re-certification + → articulation/action decision → append-only replay record +``` + +(`core/ports/residual_protocol.py`; note `core/protocol/` was already taken by +the CORE Trace Protocol v0 wire format — checked before naming, no collision.) + +Key properties, each pinned by tests (`tests/test_ring2_residual_protocol.py`): + +1. **Port-agnostic grammar, port-native geometry.** The grammar never + interprets a port's measurements; there is **no port registry and no + unified scheduler** (preflight §7 non-goal #2 honored — callers invoke the + protocol per port, per subject). A synthetic third port runs with zero + grammar changes. +2. **Abstain-or-proceed v1.** Only zero-bound operators exist + (`proceed_unmodified` / `abstain`). A **nonzero-bound operator fails + closed** with a typed error — bounded correctors require their own future + ADR (no-silent-correction doctrine, inherited from ADR-0244/0246). +3. **Fail-closed on unaccounted residual** — energy the port's typed channels + cannot account for forces abstention (ADR-0246 §3.6 doctrine generalized); + no correction policy ever attaches to it. +4. **Re-certification** — the witness is re-measured after the zero-bound + operation; drift means the port mutated state mid-pass and **raises** + rather than recording a corrupted replay. +5. **Append-only, content-addressed replay** — full-SHA-256 record digests + chained from a genesis digest (ADR-0245 §2.3: canonical JSON, no + `default=str`, no truncation); `verify_replay_chain` re-derives everything + from content, and any tamper breaks verification (pinned). +6. **Deterministic and pure** — no wall-clock, no randomness, no serve import + (A-04 pinned). + +## 2. Shipped ports (two genuinely non-identical native geometries) + +| Port | Native geometry | Witness | Admit source of truth | +|---|---|---|---| +| `IdentityPort` (`core/ports/adapters.py`) | ADR-0246 grade-1 frame preservation | §3.7 scalars + §3.6 typed channel energies (channels live IN the witness so `decompose` is a pure re-shaping — no hidden adapter state to corrupt a replay) | `evaluate_admission` (ADR-0246; the adapter adds no policy) | +| `PrecisionPort` | ADR-0244 §2.5 / 0245 §2.2 f64→f32 cast transport | measured `cast_error` + f32 unit-norm deviation (subject from a certified `ServingState`) | both within caller tolerance | + +Future adapters (Atlas, Evidence, Temporal/causal, Articulation, Action) plug +in identically; the interconnect grammar — not a second physics — is the whole +Ring-2 deliverable, exactly as §9 words it ("Smith/conformal language is the +interconnect grammar"). + +## 3. Non-goals (unchanged from the preflight) + +No unified scheduler; no serve wiring (nothing imports this at serve — future +consumption is its own flag-gated unit); no bounded correctors; no new +algebra; no port semantics invented for organs that do not exist yet. + +## 4. Consequences + +Every organ that can measure a typed residual can now speak one auditable +decision grammar with abstain-or-proceed semantics and tamper-evident replay — +without surrendering its native geometry or gaining a central controller. diff --git a/docs/adr/ADR-0248-integrity-coordinated-handoffs.md b/docs/adr/ADR-0248-integrity-coordinated-handoffs.md new file mode 100644 index 00000000..9eb27b8c --- /dev/null +++ b/docs/adr/ADR-0248-integrity-coordinated-handoffs.md @@ -0,0 +1,77 @@ +# ADR-0248: Integrity-Coordinated Handoffs — the Ring-3 Coordination Seam + +**Status**: **Proposed** — pending explicit human ratification (no self-Accept) +**Date**: 2026-07-17 +**Authors**: Joshua Shay + multi-model R&D (implemented Fable 5) +**Depends on**: ADR-0247 (Proposed — supplies the port decisions), existing epistemic/normative organs (`core/epistemic_state.py`) +**Preflight authority**: ADR-0246 preflight brief §9 (Ring 3) + +--- + +## 1. What Ring 3 is — and what this ADR does and does not claim + +The preflight fixes Ring 3's boundary in one sentence: **"Integrity +coordinates handoffs; it does not replace content-bearing cognition."** The +content organs themselves largely EXIST in CORE already: + +| Ring-3 element (preflight §9) | Existing organ | +|---|---| +| Composition / bind–infer | proof-DAG substrate (`proof_chain`, PropositionGraph, EntailmentTrace), depth-1 composer, transitive-chain surface | +| Epistemic standing | `EpistemicState` (15 typed states) + `NormativeClearance`, live in every TurnEvent | +| Observe→learn (governed) | contemplation loop — proposal-only, replay-gated corpus extension | +| Discourse planning | response governance (`govern_response`, STRICT-only scaffold) + hedge doctrine | + +What was MISSING is the coordination seam itself: nothing fused the +integrity-side evidence (the Ring-2 port decisions) with the content-side +standing into one typed decision at the handoff boundary. **This ADR builds +exactly that seam and nothing more.** It does not claim to deliver the Ring-3 +*programme* (world-model, counterfactuals, full governed-learning loop) — that +remains open, honestly listed in §4. + +## 2. Decision + +`core/ports/integrity_handoff.py` — a pure coordinator +(`coordinate_handoff(chain, epistemic_state=…, normative_clearance=…) → +IntegrityHandoff`) producing `proceed | hedge | abstain`, with conjunctive, +strongest-restriction-wins fusion rules (pinned in +`tests/test_ring3_integrity_handoff.py`): + +1. no port evidence / unverifiable replay chain → **abstain** (fail-closed); +2. any Ring-2 port abstained → **abstain**; +3. normative `violated`/`suppressed` → **abstain**; +4. normative `unassessable` → at most **hedge**; +5. weak epistemic standing (undetermined / unverified_possible / + unverified_novel / ambiguous / contradicted / epistemic_state_needed / + scope_boundary / computationally_bounded) → at most **hedge** — mirroring + the existing hedge-injection doctrine: qualified content may still surface; + only integrity violations silence a turn; +6. otherwise → **proceed**. + +`IntegrityHandoff` is **content-free by type** (pinned: no surface/content/ +text fields) — routing, per-port attribution, and digests only. It binds the +replay-chain tip digest and is itself content-addressed (`handoff_digest()`, +full SHA-256), so every routing decision is auditable back to every port's +recorded evidence. + +## 3. Operational status + +Observe-only and off-serving: no serve code consumes the coordinator yet. +Wiring it into discourse planning / response governance is a future, +flag-gated, byte-identical-off unit — subject to the same doctrine as every +identity-adjacent serve change (default-off, calibrated evidence, human +ratification). + +## 4. Honestly open (the rest of the Ring-3 programme) + +World-model construction; counterfactual binding at depth; the full +observe→learn consumption loop (ratify-vs-consume gap); discourse planning +beyond STRICT; semantic axis grounding (blocked on a positive §11-style result +— the current evidence is a validated NULL). None of these are diminished by +this seam existing; they now have a typed integrity boundary to hand through. + +## 5. Consequences + +Content cognition and integrity measurement stay separate powers: integrity +can silence or qualify a turn but can never write one, and content can never +bypass the conjunctive integrity floor — with the whole negotiation recorded +in tamper-evident, content-addressed form. diff --git a/docs/audit/artifacts/ring2-ring3-runlog.txt b/docs/audit/artifacts/ring2-ring3-runlog.txt new file mode 100644 index 00000000..c88e201c --- /dev/null +++ b/docs/audit/artifacts/ring2-ring3-runlog.txt @@ -0,0 +1,11 @@ +Ring 2 + Ring 3 — verification run log +branch: feat/ring2-residual-protocol (off main @ f49b391d) + +=== pytest: Ring 2/3 + ADR-0246 suites + D4 gate surfaces === +........................................................................ [ 57%] +..................................................... [100%] +125 passed in 8.70s + +=== uv run core test --suite smoke -q === +................................ [100%] +176 passed in 129.93s (0:02:09) diff --git a/docs/briefs/ADR-0246-induced-identity-action-and-path-integrity-preflight.md b/docs/briefs/ADR-0246-induced-identity-action-and-path-integrity-preflight.md index d292db76..e0a19551 100644 --- a/docs/briefs/ADR-0246-induced-identity-action-and-path-integrity-preflight.md +++ b/docs/briefs/ADR-0246-induced-identity-action-and-path-integrity-preflight.md @@ -60,6 +60,8 @@ structure is stabilized would instrument lawfulness on a frame the dynamics igno | **§4.1/§4.2 telemetry + path serve integration** (completion pass) | `IdentityActionRecord` (full digests, `policy_version=version_id()`, multi-condition `refusal_reason`); `manifold_content_digest` + geometry/gate version ids; §3.4-step-2 `admitted` gate on the ledger; `advance_session_identity_path` observe-only serve wiring (same flags, instance lifetime = session boundary); telemetry emits `identity_action_*`/`identity_path_*` keys only when the paths ran | **committed** `feat/adr-0246-slice1-complete` — flag-off byte-identical; all suites green | | **§11 grounding-feasibility** (completion pass) | fixed TRAIN(13)/HELD-OUT(12)/ADVERSARIAL(8) splits; bivector generator proxy (numpy-only); **sample-size-calibrated null** (200 noise-pair trials at real n) + shared-basis positive recovery control; precision pairs; per-plane energy | **DONE — honest NULL, method validated**: positive control 0.9995 (100th pctile of null) but real cross-cohort cosine 0.52 = 87th pctile of chance; AUC 0.49; no stable generator subspace at this n. Artifact: `docs/audit/artifacts/adr-0246-grounding-feasibility-report.json` | | **ADR-0246 body + acceptance packet** | `docs/adr/ADR-0246-induced-identity-action-and-path-integrity.md` (**Proposed**; F1 semantics, ‖·‖_G convention, turn-ownership + refusal_reason rulings requested; binding claims language; honest §6.3 + §11 numbers; machine-readable operational-status block); packet `docs/audit/adr-0246-acceptance-packet-2026-07-17.md` §8 **RULING PENDING** | **committed** — no self-Accept; awaiting Shay ruling | +| **Ring 2 — multi-port residual protocol** | the §9 shared control grammar as `core/ports/residual_protocol.py` (7 stages; zero-bound operators only, nonzero fails closed; fail-closed unaccounted residual; re-certification raises on witness drift; append-only full-SHA-256 replay chain + `verify_replay_chain`); real adapters `IdentityPort` (ADR-0246 geometry) + `PrecisionPort` (ADR-0244 §2.5 cast transport) + synthetic port proving agnosticism; **no registry, no scheduler** (§7 non-goal #2 honored) | **built** — ADR-0247 **Proposed**; `tests/test_ring2_residual_protocol.py` green; pure/off-serving | +| **Ring 3 — integrity-coordinated handoff** | the §9 coordination seam as `core/ports/integrity_handoff.py`: fuses Ring-2 port decisions + existing `EpistemicState`/`NormativeClearance` into content-free `proceed/hedge/abstain` (conjunctive, strongest-restriction-wins; fail-closed on missing/invalid evidence; binds replay-chain digest); content organs mapped to existing substrate; remainder (world-model, governed-learning consumption, discourse widening) honestly listed open | **built (observe-only)** — ADR-0248 **Proposed**; `tests/test_ring3_integrity_handoff.py` green; no serve consumer yet | Nothing in the reordering relaxes a §7 non-goal: no `C_id` corrector, no `H_id` enlargement, no pack/axis redesign, no gate activation. The §11 grounding study @@ -458,6 +460,11 @@ PrecisionTransportCertificate: ## 9. Ring 2 / Ring 3 pointers (out of scope, preserved direction) +> **Status (2026-07-17):** both pointers are now first-slice BUILT exactly as +> bounded here — the shared control grammar (`core/ports/`, ADR-0247 Proposed) +> and the integrity-handoff coordination seam (ADR-0248 Proposed) — see §0a. +> The direction text below is preserved verbatim as the design authority. + ### Ring 2 — multi-port residual protocol (future) Shared control grammar only: diff --git a/tests/test_ring2_residual_protocol.py b/tests/test_ring2_residual_protocol.py new file mode 100644 index 00000000..0d60473f --- /dev/null +++ b/tests/test_ring2_residual_protocol.py @@ -0,0 +1,271 @@ +"""Ring 2 — multi-port residual protocol pins (shared control grammar). + +The preflight (ADR-0246 brief §9) fixes Ring 2 as a SHARED CONTROL GRAMMAR +only: witness → typed residual decomposition → permitted operator selection → +bounded operation or abstention → re-certification → articulation/action +decision → append-only replay record — with ports retaining NON-identical +native geometry. These tests pin: + + * the grammar runs all seven stages deterministically and purely; + * v1 ships ONLY zero-bound operators (proceed_unmodified / abstain) — a + nonzero-bound operator fails closed (no-silent-correction doctrine); + * re-certification requires the witness unchanged after a zero-bound + operation (a mutating "zero-bound" port fails closed); + * the replay chain is append-only, content-addressed (full SHA-256), and + verifiable; tampering any record breaks verification; + * two REAL ports with genuinely different native geometry — Identity + (ADR-0246 grade-1 frame preservation) and Precision (ADR-0244 §2.5 f32 + cast transport) — both speak the grammar without the grammar knowing + either geometry; + * a synthetic third port proves port-agnosticism (no port registry, no + unified scheduler — brief §7 non-goal honored). +""" + +from __future__ import annotations + +import numpy as np +import pytest + +from algebra.cl41 import N_COMPONENTS +from core.physics.identity_action import AdmissionPolicy +from core.physics.identity_manifold import IdentityManifoldGeometry +from core.ports.residual_protocol import ( + GENESIS_DIGEST, + ActionDecision, + PermittedOperator, + PortWitness, + ReplayRecord, + ResidualDecomposition, + ResidualProtocolError, + run_residual_protocol, + verify_replay_chain, +) +from core.ports.adapters import IdentityPort, PrecisionPort, PrecisionSubject + +_E12, _E14 = 6, 8 + + +def _rotor(biv, theta): + r = np.zeros(N_COMPONENTS, dtype=np.float64) + r[0] = np.cos(theta / 2.0) + r[biv] = np.sin(theta / 2.0) + return r + + +def _identity_versor(): + v = np.zeros(N_COMPONENTS, dtype=np.float64) + v[0] = 1.0 + return v + + +@pytest.fixture(scope="module") +def geometry(): + return IdentityManifoldGeometry.from_directions( + ((1.0, 0.0, 0.0), (0.0, 1.0, 0.0), (0.0, 0.0, 1.0)) + ) + + +@pytest.fixture(scope="module") +def identity_port(geometry): + return IdentityPort(geometry, AdmissionPolicy.placeholder_default()) + + +# --- a synthetic third port: proves the grammar is port-agnostic --------------- + + +class _ToyPort: + """Minimal conforming port with a made-up native geometry (scalar residual).""" + + port_id = "toy" + schema_version = "toy_v1" + + def witness(self, subject): + return PortWitness( + port_id=self.port_id, + schema_version=self.schema_version, + measurements=(("residual", float(subject)),), + ) + + def decompose(self, witness): + value = dict(witness.measurements)["residual"] + return ResidualDecomposition( + port_id=self.port_id, + channels=(("scalar", value),), + unaccounted=0.0, + unaccounted_tol=1e-9, + ) + + def admit(self, subject, decomposition): + value = dict(decomposition.channels)["scalar"] + if value <= 0.5: + return True, () + return False, ("scalar>0.5",) + + +# --- grammar mechanics ---------------------------------------------------------- + + +def test_toy_port_proceed_and_abstain_paths(): + port = _ToyPort() + chain, decision = run_residual_protocol(port, 0.1, ()) + assert decision.action == "proceed" + assert len(chain) == 1 + chain, decision = run_residual_protocol(port, 0.9, chain) + assert decision.action == "abstain" + assert decision.reasons == ("scalar>0.5",) + assert len(chain) == 2 + + +def test_replay_chain_is_append_only_content_addressed(): + port = _ToyPort() + chain, _ = run_residual_protocol(port, 0.1, ()) + chain, _ = run_residual_protocol(port, 0.9, chain) + chain, _ = run_residual_protocol(port, 0.2, chain) + assert [r.sequence_index for r in chain] == [0, 1, 2] + assert chain[0].prev_record_digest == GENESIS_DIGEST + assert chain[1].prev_record_digest == chain[0].record_digest() + assert chain[2].prev_record_digest == chain[1].record_digest() + assert verify_replay_chain(chain) is True + for record in chain: + assert len(record.record_digest()) == 64 + int(record.record_digest(), 16) + + +def test_tampered_chain_fails_verification(): + port = _ToyPort() + chain, _ = run_residual_protocol(port, 0.1, ()) + chain, _ = run_residual_protocol(port, 0.2, chain) + from dataclasses import replace + tampered = (chain[0], replace(chain[1], prev_record_digest="0" * 64)) + assert verify_replay_chain(tampered) is False + # and altering a payload breaks the digest linkage of any later record + tampered2 = (replace(chain[0], action="abstain"), chain[1]) + assert verify_replay_chain(tampered2) is False + + +def test_determinism_same_inputs_same_digests(): + port = _ToyPort() + a, _ = run_residual_protocol(port, 0.3, ()) + b, _ = run_residual_protocol(port, 0.3, ()) + assert a[0].record_digest() == b[0].record_digest() + + +def test_nonzero_bound_operator_fails_closed(): + class _CorrectorPort(_ToyPort): + port_id = "toy_corrector" + + def permitted_operators(self, decomposition): + return (PermittedOperator(operator_id="nudge", bound=0.1),) + + with pytest.raises(ResidualProtocolError, match="nonzero_bound"): + run_residual_protocol(_CorrectorPort(), 0.1, ()) + + +def test_mutating_zero_bound_port_fails_recertification(): + class _MutatingPort(_ToyPort): + port_id = "toy_mutator" + + def __init__(self): + self._calls = 0 + + def witness(self, subject): + # a port whose measurement drifts between witness and re-certify: + # a zero-bound operation must leave the witness unchanged, so this + # MUST fail closed rather than record a corrupted replay. + self._calls += 1 + return PortWitness( + port_id=self.port_id, + schema_version=self.schema_version, + measurements=(("residual", 0.1 * self._calls),), + ) + + with pytest.raises(ResidualProtocolError, match="recertification"): + run_residual_protocol(_MutatingPort(), 0.1, ()) + + +def test_unaccounted_residual_fails_closed(): + class _LeakyPort(_ToyPort): + port_id = "toy_leaky" + + def decompose(self, witness): + return ResidualDecomposition( + port_id=self.port_id, + channels=(("scalar", 0.1),), + unaccounted=0.5, # energy the typed channels cannot account for + unaccounted_tol=1e-6, # -> fail closed, no correction policy + ) + + chain, decision = run_residual_protocol(_LeakyPort(), 0.1, ()) + assert decision.action == "abstain" + assert any("unaccounted" in r for r in decision.reasons) + + +def test_cross_port_records_interleave_in_one_chain(): + toy = _ToyPort() + chain, _ = run_residual_protocol(toy, 0.1, ()) + other = _ToyPort() + other.port_id = "toy_b" + chain, _ = run_residual_protocol(other, 0.2, chain) + assert [r.port_id for r in chain] == ["toy", "toy_b"] + assert verify_replay_chain(chain) is True + + +# --- real port: Identity (ADR-0246 native geometry) ----------------------------- + + +def test_identity_port_proceeds_on_near_identity(identity_port): + chain, decision = run_residual_protocol(identity_port, _identity_versor(), ()) + assert decision.action == "proceed" + record = chain[-1] + assert record.port_id == "identity" + channels = dict(record.decomposition["channels"]) + assert set(channels) == { + "null_or_conformal", "boost_like", "spatial_foreign", "unclassified", + } + + +def test_identity_port_abstains_on_alien_tilt(identity_port): + chain, decision = run_residual_protocol(identity_port, _rotor(_E14, 1.5), ()) + assert decision.action == "abstain" + assert any("leakage" in r or "d_" in r for r in decision.reasons) + + +# --- real port: Precision (ADR-0244 §2.5 cast-transport native geometry) -------- + + +def test_precision_port_proceeds_on_clean_cast(): + port = PrecisionPort(tol=1e-6) + subject = PrecisionSubject(cast_error=1e-8, unit_norm_f32=1.0 + 1e-8) + chain, decision = run_residual_protocol(port, subject, ()) + assert decision.action == "proceed" + channels = dict(chain[-1].decomposition["channels"]) + assert set(channels) == {"cast_error", "unit_norm_deviation"} + + +def test_precision_port_abstains_on_precision_cliff(): + port = PrecisionPort(tol=1e-6) + subject = PrecisionSubject(cast_error=1e-3, unit_norm_f32=1.0) + chain, decision = run_residual_protocol(port, subject, ()) + assert decision.action == "abstain" + + +def test_identity_and_precision_share_one_replay_chain(identity_port): + chain, _ = run_residual_protocol(identity_port, _identity_versor(), ()) + chain, _ = run_residual_protocol( + PrecisionPort(tol=1e-6), + PrecisionSubject(cast_error=1e-8, unit_norm_f32=1.0), + chain, + ) + assert [r.port_id for r in chain] == ["identity", "precision"] + assert verify_replay_chain(chain) is True + + +def test_module_is_pure_offserving(): + import core.ports.residual_protocol as protocol + import core.ports.adapters as ports + + for mod in (protocol, ports): + assert mod.__file__ is not None + with open(mod.__file__, encoding="utf-8") as fh: + src = fh.read() + assert "import chat" not in src and "from chat" not in src diff --git a/tests/test_ring3_integrity_handoff.py b/tests/test_ring3_integrity_handoff.py new file mode 100644 index 00000000..9b9e929b --- /dev/null +++ b/tests/test_ring3_integrity_handoff.py @@ -0,0 +1,170 @@ +"""Ring 3 — integrity-coordinated handoff pins. + +The preflight (ADR-0246 brief §9) fixes Ring 3's coordination role: "Integrity +coordinates handoffs; it does not replace content-bearing cognition." These pin +the pure coordinator that fuses the Ring-2 per-port action decisions with the +EXISTING content-organ types (EpistemicState, NormativeClearance) into a single +typed handoff for discourse/composition consumers: + + * any port abstain → ABSTAIN (integrity floor is conjunctive); + * normative VIOLATED/SUPPRESSED → ABSTAIN; UNASSESSABLE → at most HEDGE; + * weak epistemic standing (undetermined/speculative-class) → at most HEDGE; + * strong standing + all ports proceed + cleared → PROCEED; + * the handoff binds the replay chain (content-addressed) so the decision is + auditable back to every port's evidence; + * the coordinator NEVER fabricates content — it returns a routing decision + only (pinned by its output type having no surface/content fields). +""" + +from __future__ import annotations + +import pytest + +from core.epistemic_state import EpistemicState, NormativeClearance +from core.ports.integrity_handoff import ( + HANDOFF_ABSTAIN, + HANDOFF_HEDGE, + HANDOFF_PROCEED, + IntegrityHandoff, + coordinate_handoff, +) +from core.ports.residual_protocol import PortWitness, ResidualDecomposition, run_residual_protocol + + +class _Port: + schema_version = "toy_v1" + + def __init__(self, port_id, value): + self.port_id = port_id + self._value = value + + def witness(self, subject): + return PortWitness( + port_id=self.port_id, schema_version=self.schema_version, + measurements=(("residual", self._value),), + ) + + def decompose(self, witness): + return ResidualDecomposition( + port_id=self.port_id, + channels=(("scalar", dict(witness.measurements)["residual"]),), + unaccounted=0.0, unaccounted_tol=1e-9, + ) + + def admit(self, subject, decomposition): + value = dict(decomposition.channels)["scalar"] + return (True, ()) if value <= 0.5 else (False, (f"{self.port_id}:refused",)) + + +def _chain(*port_values): + chain = () + for port_id, value in port_values: + chain, _ = run_residual_protocol(_Port(port_id, value), None, chain) + return chain + + +def test_all_proceed_cleared_verified_gives_proceed(): + chain = _chain(("identity", 0.1), ("precision", 0.2)) + handoff = coordinate_handoff( + chain, + epistemic_state=EpistemicState.VERIFIED, + normative_clearance=NormativeClearance.CLEARED, + ) + assert handoff.handoff == HANDOFF_PROCEED + assert handoff.port_actions == (("identity", "proceed"), ("precision", "proceed")) + + +def test_any_port_abstain_forces_abstain(): + chain = _chain(("identity", 0.1), ("precision", 0.9)) + handoff = coordinate_handoff( + chain, + epistemic_state=EpistemicState.VERIFIED, + normative_clearance=NormativeClearance.CLEARED, + ) + assert handoff.handoff == HANDOFF_ABSTAIN + assert any("precision" in r for r in handoff.reasons) + + +@pytest.mark.parametrize("clearance", [NormativeClearance.VIOLATED, NormativeClearance.SUPPRESSED]) +def test_normative_violation_forces_abstain(clearance): + chain = _chain(("identity", 0.1)) + handoff = coordinate_handoff( + chain, epistemic_state=EpistemicState.VERIFIED, normative_clearance=clearance, + ) + assert handoff.handoff == HANDOFF_ABSTAIN + + +def test_unassessable_clearance_caps_at_hedge(): + chain = _chain(("identity", 0.1)) + handoff = coordinate_handoff( + chain, + epistemic_state=EpistemicState.VERIFIED, + normative_clearance=NormativeClearance.UNASSESSABLE, + ) + assert handoff.handoff == HANDOFF_HEDGE + + +@pytest.mark.parametrize( + "state", + [ + EpistemicState.UNDETERMINED, + EpistemicState.UNVERIFIED_POSSIBLE, + EpistemicState.UNVERIFIED_NOVEL, + EpistemicState.AMBIGUOUS, + EpistemicState.CONTRADICTED, + ], +) +def test_weak_epistemic_standing_caps_at_hedge(state): + chain = _chain(("identity", 0.1)) + handoff = coordinate_handoff( + chain, epistemic_state=state, normative_clearance=NormativeClearance.CLEARED, + ) + assert handoff.handoff in (HANDOFF_HEDGE, HANDOFF_ABSTAIN) + assert handoff.handoff == HANDOFF_HEDGE # hedge, not abstain: content may still surface qualified + + +def test_handoff_binds_replay_chain_digest(): + chain = _chain(("identity", 0.1)) + handoff = coordinate_handoff( + chain, + epistemic_state=EpistemicState.VERIFIED, + normative_clearance=NormativeClearance.CLEARED, + ) + assert handoff.chain_tip_digest == chain[-1].record_digest() + assert len(handoff.handoff_digest()) == 64 + int(handoff.handoff_digest(), 16) + # deterministic + again = coordinate_handoff( + chain, + epistemic_state=EpistemicState.VERIFIED, + normative_clearance=NormativeClearance.CLEARED, + ) + assert again.handoff_digest() == handoff.handoff_digest() + + +def test_empty_chain_fails_closed(): + handoff = coordinate_handoff( + (), + epistemic_state=EpistemicState.VERIFIED, + normative_clearance=NormativeClearance.CLEARED, + ) + assert handoff.handoff == HANDOFF_ABSTAIN + assert any("no_port_evidence" in r for r in handoff.reasons) + + +def test_coordinator_never_carries_content(): + # Ring 3 contract: integrity coordinates handoffs, it does NOT replace + # content-bearing cognition. The handoff type must carry no content/surface + # fields — only routing, attribution, and digests. + fields = set(IntegrityHandoff.__dataclass_fields__) + assert "surface" not in fields and "content" not in fields and "text" not in fields + assert {"handoff", "reasons", "port_actions", "chain_tip_digest"} <= fields + + +def test_module_is_pure_offserving(): + import core.ports.integrity_handoff as mod + + assert mod.__file__ is not None + with open(mod.__file__, encoding="utf-8") as fh: + src = fh.read() + assert "import chat" not in src and "from chat" not in src