"""Sovereign Circuit bench test station — SCST1/1.0 Enforces the two-phase ordering that SPEC.md section 4.4 specifies and that `attestation.py` alone does not enforce: a **price** commitment window cannot open until a **grade** round has completed and published a `grade_token`. PHASE 1 — GRADE PHASE 2 — PRICE observe -> commit -> reveal -> reconcile -> open -> commit -> reveal -> settle -> seal | ^ +- token -+ Why this module exists rather than a patch to `attestation.py` -------------------------------------------------------------- The obvious change is to add `rubric_hash` and `grade_token` as required fields on `Policy`. That was rejected. `Policy` is hashed into `policy_hash`, and `policy_hash` is pinned by the normative test vectors in SPEC.md section 10. Adding required fields silently changes that hash, which would mean every published SCDA1 vector stopped reproducing — the exact failure the vectors exist to catch. A protocol that breaks its own published constants to add a feature has not added a feature, it has forked. So the binding lives here, at the station: the component that actually holds a unit on a bench and walks an operator through a round. `attestation.py` stays byte-compatible with its published vectors and remains usable on its own for a bare price round; the station is what refuses to run one out of order. `bound_policy_hash()` extends `policy_hash()` with the grade binding without altering the legacy value, and `test_station.py` asserts the legacy hash is unchanged. What is actually enforced ------------------------- 1. The price window cannot open before a `grade_token` is published. 2. A void grade round (Class M instrument conflict) blocks the price window permanently. There is no override path in this code, deliberately. 3. A material divergence blocks the price window until it is resolved, either by re-inspection or by an explicitly recorded conservative merge. 4. The `grade` field inside each SCDA1 price commitment MUST equal the grade the two parties jointly derived in phase 1. This is the load-bearing check: without it, phase 1 becomes advisory and a party can commit to a price justified by a grade nobody agreed to. 5. Observation reveals must reproduce their commitments, and neither party's reveal is accepted until both parties have committed. 6. A sheet cannot be re-committed, and a phase cannot be re-run. What is deliberately NOT enforced --------------------------------- Identity. There are no signatures here. `party_id` is a label, not an authenticated claim, so this module assumes an operator who is not trying to impersonate the counterparty. Ed25519 signing is SPEC.md section 11 item 7 and remains open. The station closes an ordering hole, not an identity hole, and it should not be described as doing more than that. """ from __future__ import annotations import json from dataclasses import dataclass, field from enum import Enum from typing import Any, Dict, List, Optional import attestation as att import grading as gr STATION_ID = "sovereign-circuit/test-station/1" STATION_VERSION = "SCST1/1.0" HOLDER = "holder" INSPECTOR = "inspector" ROLES = (HOLDER, INSPECTOR) # --------------------------------------------------------------------------- # Errors # # Each failure mode gets its own type. A single StationError with a message # would let a caller accidentally treat "the grade round voided" the same as # "you called this method too early", and those need different handling: the # first ends the round, the second is a retry. # --------------------------------------------------------------------------- class StationError(Exception): """Base for every refusal this station issues.""" class PhaseOrderError(StationError): """An operation was attempted from the wrong phase.""" class GradeNotPublishedError(StationError): """The price window was asked to open with no agreed grade.""" class GradeRoundVoidError(StationError): """The grade round voided. The price window will not open for this unit.""" class GradeUnresolvedError(StationError): """A material divergence is outstanding and has not been resolved.""" class GradeTokenMismatchError(StationError): """A price commitment carries a grade the parties did not jointly derive.""" class RubricMismatchError(StationError): """The station's rubric is not the one the round was opened against.""" class AlreadySubmittedError(StationError): """A party tried to commit or reveal twice.""" class RevealMismatchError(StationError): """A revealed sheet does not reproduce its commitment.""" # --------------------------------------------------------------------------- # Phases # --------------------------------------------------------------------------- class Phase(str, Enum): GRADE_OPEN = "grade_open" GRADE_COMMITTED = "grade_committed" GRADE_REVEALED = "grade_revealed" GRADE_PUBLISHED = "grade_published" PRICE_OPEN = "price_open" PRICE_COMMITTED = "price_committed" PRICE_REVEALED = "price_revealed" SETTLED = "settled" SEALED = "sealed" VOID = "void" UNRESOLVED = "unresolved" # Phases from which no further progress is possible without operator action # outside this module. TERMINAL_BLOCKED = (Phase.VOID, Phase.UNRESOLVED) @dataclass class GradeBinding: """What phase 1 hands to phase 2. `grade_token` is computed over the canonical agreed sheet, not over either party's sheet, because on an immaterial divergence the two sheets differ while deriving to the same grade — so the two parties would otherwise produce two different tokens for one agreed grade. """ rubric_version: str rubric_hash: str grade_code: str tier: str grade_token: str reconciliation_status: str agreed_sheet: Dict[str, Any] def to_dict(self) -> Dict[str, Any]: return { "rubric_version": self.rubric_version, "rubric_hash": self.rubric_hash, "grade_code": self.grade_code, "tier": self.tier, "grade_token": self.grade_token, "reconciliation_status": self.reconciliation_status, } @dataclass class Station: """One unit, one bench, one round. Construct, run phase 1, run phase 2. Every method that could be called early raises rather than warning, because a station that warns and then proceeds is a station that produces an unordered round with a reassuring log line next to it. """ round_id: str asset_id: str policy: att.Policy = field(default_factory=att.Policy) rubric_version: str = gr.RUBRIC_VERSION expected_rubric_hash: Optional[str] = None phase: Phase = Phase.GRADE_OPEN obs_commits: Dict[str, str] = field(default_factory=dict) obs_nonces: Dict[str, str] = field(default_factory=dict) obs_sheets: Dict[str, Dict[str, Any]] = field(default_factory=dict) reconciliation: Optional[gr.Reconciliation] = None binding: Optional[GradeBinding] = None price_commits: Dict[str, str] = field(default_factory=dict) price_reveals: Dict[str, dict] = field(default_factory=dict) settlement: Optional[att.Settlement] = None record: Optional[dict] = None log: List[str] = field(default_factory=list) def __post_init__(self) -> None: live = gr.rubric_hash() if self.expected_rubric_hash is None: self.expected_rubric_hash = live elif not att._ct_eq(self.expected_rubric_hash, live): # A station running a different rubric than the round was opened # against would grade the unit by rules nobody published. raise RubricMismatchError( f"station rubric {live[:12]}... does not match round rubric " f"{self.expected_rubric_hash[:12]}..." ) self._note(f"station {STATION_VERSION} open on {self.asset_id}; " f"rubric {self.rubric_version} {live[:12]}...") # -- internals --------------------------------------------------------- def _note(self, msg: str) -> None: self.log.append(msg) def _require(self, *allowed: Phase) -> None: if self.phase in TERMINAL_BLOCKED: if self.phase is Phase.VOID: raise GradeRoundVoidError( "grade round voided on a machine-read conflict; this unit " "must be re-read before any price round can open" ) raise GradeUnresolvedError( "a material grade divergence is outstanding; resolve it " "before continuing" ) if self.phase not in allowed: raise PhaseOrderError( f"cannot do this from phase '{self.phase.value}'; " f"expected one of {[p.value for p in allowed]}" ) @staticmethod def _check_role(role: str) -> None: if role not in ROLES: raise StationError(f"unknown role {role!r}; expected one of {ROLES}") # -- phase 1: grade ---------------------------------------------------- def commit_sheet(self, role: str, sheet: Dict[str, Any], nonce: Optional[str] = None) -> str: """Seal one party's observation sheet. Returns the observation hash.""" self._check_role(role) self._require(Phase.GRADE_OPEN) if role in self.obs_commits: raise AlreadySubmittedError( f"{role} has already committed a sheet; a sheet cannot be " f"amended once sealed" ) gr.validate(sheet) n = nonce or att.new_nonce() c = gr.observation_hash(sheet, n) self.obs_commits[role] = c self.obs_nonces[role] = n self._note(f"{role} committed sheet {c[:12]}...") if len(self.obs_commits) == len(ROLES): self.phase = Phase.GRADE_COMMITTED self._note("both sheets sealed; reveals may open") return c def reveal_sheet(self, role: str, sheet: Dict[str, Any], nonce: Optional[str] = None) -> None: """Open one party's sheet and check it against its commitment. Reveals are refused until BOTH parties have committed. If the first party could reveal while the second was still writing, the second could copy it, and the whole point of sealing the sheets would be lost. """ self._check_role(role) self._require(Phase.GRADE_COMMITTED) if role in self.obs_sheets: raise AlreadySubmittedError(f"{role} has already revealed") n = nonce if nonce is not None else self.obs_nonces[role] recomputed = gr.observation_hash(sheet, n) if not att._ct_eq(recomputed, self.obs_commits[role]): raise RevealMismatchError( f"{role}'s revealed sheet does not reproduce its commitment" ) self.obs_sheets[role] = dict(sheet) self._note(f"{role} revealed a sheet matching its commitment") if len(self.obs_sheets) == len(ROLES): self.phase = Phase.GRADE_REVEALED def reconcile(self) -> gr.Reconciliation: """Compare the two sheets. Sets the phase according to the outcome.""" self._require(Phase.GRADE_REVEALED) rec = gr.reconcile(self.obs_sheets[HOLDER], self.obs_sheets[INSPECTOR]) self.reconciliation = rec if rec.status == "void": self.phase = Phase.VOID self._note(f"VOID — machine conflict on " f"{', '.join(rec.machine_conflicts)}") elif rec.status == "material": self.phase = Phase.UNRESOLVED self._note(f"MATERIAL divergence — re-inspect " f"{', '.join(rec.pivotal_fields)}") else: self._note(f"{rec.status} — agreed grade " f"{rec.agreed_grade.code} ({rec.agreed_grade.tier})") return rec def publish_grade(self) -> GradeBinding: """Publish the grade token. The only legitimate exit from phase 1. On an immaterial divergence the agreed sheet is the conservative merge, and the merge is checked to derive to the agreed grade before anything is published. That check is not ceremonial: taking the worse value on every differing field can in principle cross a band boundary that neither party's own sheet crossed, and publishing a token for a grade neither party derived would be a quiet substitution of the station's judgement for theirs. If it happens, the round escalates instead. """ self._require(Phase.GRADE_REVEALED) rec = self.reconciliation if rec is None: raise PhaseOrderError("reconcile() before publish_grade()") if rec.agreed_grade is None: raise GradeUnresolvedError("no agreed grade to publish") if rec.status == "agreed": agreed_sheet = dict(self.obs_sheets[HOLDER]) else: agreed_sheet = gr.conservative_merge( self.obs_sheets[HOLDER], self.obs_sheets[INSPECTOR] ) merged = gr.grade(agreed_sheet) if merged.code != rec.agreed_grade.code: self.phase = Phase.UNRESOLVED self._note( f"escalated — conservative merge derives {merged.code} but " f"both sheets derive {rec.agreed_grade.code}" ) raise GradeUnresolvedError( f"conservative merge of two sheets that both derive " f"{rec.agreed_grade.code} itself derives {merged.code}; " f"the divergence is material after all and must be " f"re-inspected" ) g = gr.grade(agreed_sheet) self.binding = GradeBinding( rubric_version=self.rubric_version, rubric_hash=gr.rubric_hash(), grade_code=g.code, tier=g.tier, grade_token=gr.grade_token(agreed_sheet), reconciliation_status=rec.status, agreed_sheet=agreed_sheet, ) self.phase = Phase.GRADE_PUBLISHED self._note(f"grade published {g.code} tier {g.tier}, token " f"{self.binding.grade_token[:12]}...") return self.binding # -- the gate ---------------------------------------------------------- def open_price_window(self) -> str: """Open phase 2. Returns the bound policy hash. This is the enforcement point named in SPEC.md section 4.4. Every refusal below is a path that `attestation.py` on its own would have allowed. """ if self.phase is Phase.VOID: raise GradeRoundVoidError( "grade round voided; no price round opens for this unit" ) if self.phase is Phase.UNRESOLVED: raise GradeUnresolvedError( "material grade divergence outstanding; the price window " "stays shut until the pivotal fields are re-inspected" ) if self.binding is None: raise GradeNotPublishedError( "no grade_token published; the price window cannot open " "before the grade is fixed (SPEC.md 4.4)" ) self._require(Phase.GRADE_PUBLISHED) self.phase = Phase.PRICE_OPEN bph = self.bound_policy_hash() self._note(f"price window open; bound policy {bph[:12]}...") return bph def bound_policy_hash(self) -> str: """Policy hash extended with the grade binding. Computed over the legacy policy dict PLUS the binding, under a distinct domain tag. `att.policy_hash(policy)` is untouched and still reproduces the value published in SPEC.md section 10. """ if self.binding is None: raise GradeNotPublishedError("no binding to hash") return att.h( b"SCST1|policy|", att.canon({ "policy": att._as_dict(self.policy), "binding": self.binding.to_dict(), "station": STATION_VERSION, }), ) # -- phase 2: price ---------------------------------------------------- def commit_price(self, role: str, amount_cents: int, grade_code: Optional[str] = None, nonce: Optional[str] = None) -> str: """Seal one party's price. The grade is not a free parameter. `grade_code` defaults to the agreed grade and, if passed explicitly, must equal it. This is what makes phase 1 binding rather than advisory. """ self._check_role(role) self._require(Phase.PRICE_OPEN, Phase.PRICE_COMMITTED) if role in self.price_commits: raise AlreadySubmittedError(f"{role} has already committed a price") agreed = self.binding.grade_code if grade_code is None: grade_code = agreed elif grade_code != agreed: raise GradeTokenMismatchError( f"{role} committed grade {grade_code!r} but the parties " f"jointly derived {agreed!r} in phase 1" ) n = nonce or att.new_nonce() c = att.commitment( round_id=self.round_id, asset_id=self.asset_id, party_id=role, role=role, amount_cents=amount_cents, grade=grade_code, nonce_hex=n, ) self.price_commits[role] = c self.price_reveals[role] = { "round_id": self.round_id, "asset_id": self.asset_id, "party_id": role, "role": role, "amount_cents": amount_cents, "grade": grade_code, "nonce": n, } self._note(f"{role} committed price {c[:12]}...") if len(self.price_commits) == len(ROLES): self.phase = Phase.PRICE_COMMITTED return c def reveal_prices(self) -> None: """Open both price commitments and verify each reproduces its hash.""" self._require(Phase.PRICE_COMMITTED) for role in ROLES: if not att.verify_reveal(self.price_commits[role], self.price_reveals[role]): raise RevealMismatchError(f"{role}'s price reveal does not verify") self.phase = Phase.PRICE_REVEALED self._note("both prices revealed and verified") def settle(self, index_cents: Optional[int] = None) -> att.Settlement: """Run the settlement ladder. `index_cents` is required only when the revealed spread exceeds the tolerance band, because escalation resolves against the published index rather than against either party. Passing it unconditionally is harmless; omitting it on an escalating round raises, which is correct — a station with no index value cannot settle an escalated round and should say so rather than inventing a midpoint. """ self._require(Phase.PRICE_REVEALED) s = att.settle( holder_cents=self.price_reveals[HOLDER]["amount_cents"], inspector_cents=self.price_reveals[INSPECTOR]["amount_cents"], policy=self.policy, index_cents=index_cents, ) self.settlement = s self.phase = Phase.SETTLED self._note(f"settled {s.outcome} at {s.amount_cents}c") return s def seal(self, chain: att.Chain, timestamps: Optional[dict] = None) -> dict: """Append the round to a hash chain. Carries the binding in the body.""" self._require(Phase.SETTLED) rec = att.round_record( round_id=self.round_id, asset_id=self.asset_id, policy=self.policy, holder_commit=self.price_commits[HOLDER], inspector_commit=self.price_commits[INSPECTOR], holder_reveal=self.price_reveals[HOLDER], inspector_reveal=self.price_reveals[INSPECTOR], settlement=self.settlement, timestamps=timestamps or {"opened": "unset", "committed": "unset", "revealed": "unset", "settled": "unset"}, ) rec["grade_binding"] = self.binding.to_dict() rec["station"] = STATION_VERSION rec["bound_policy_hash"] = self.bound_policy_hash() entry = chain.append(rec) self.record = rec self.phase = Phase.SEALED self._note(f"sealed as chain entry {entry['entry_hash'][:12]}...") return entry # -- reporting --------------------------------------------------------- def transcript(self) -> str: return "\n".join(f" {i+1:2d}. {m}" for i, m in enumerate(self.log)) def state(self) -> Dict[str, Any]: return { "station": STATION_VERSION, "round_id": self.round_id, "asset_id": self.asset_id, "phase": self.phase.value, "grade_binding": self.binding.to_dict() if self.binding else None, "reconciliation": (self.reconciliation.to_dict() if self.reconciliation else None), "settlement": (self.settlement.as_dict() if self.settlement else None), } # --------------------------------------------------------------------------- # Convenience: a whole round in one call, for scripted use # --------------------------------------------------------------------------- def run_round(round_id: str, asset_id: str, holder_sheet: Dict[str, Any], inspector_sheet: Dict[str, Any], holder_cents: int, inspector_cents: int, policy: Optional[att.Policy] = None, chain: Optional[att.Chain] = None) -> Station: """Run both phases in the only order the station permits. Raises on any blocked outcome rather than returning a partial result, so a caller cannot mistake a voided unit for a settled one. """ st = Station(round_id=round_id, asset_id=asset_id, policy=policy or att.Policy()) st.commit_sheet(HOLDER, holder_sheet) st.commit_sheet(INSPECTOR, inspector_sheet) st.reveal_sheet(HOLDER, holder_sheet) st.reveal_sheet(INSPECTOR, inspector_sheet) st.reconcile() st.publish_grade() st.open_price_window() st.commit_price(HOLDER, holder_cents) st.commit_price(INSPECTOR, inspector_cents) st.reveal_prices() st.settle() st.seal(chain or att.Chain()) return st if __name__ == "__main__": sheet = gr.blank_sheet() print(json.dumps({"station": STATION_VERSION, "rubric": gr.rubric_hash(), "fields": len(gr.FIELDS)}, indent=2))