#!/usr/bin/env python3 """ BYZANTINE CONSENSUS PROTOCOL SilverSight Lattice Quintuplet Watchdog System 5-way Byzantine consensus for Φ-corkscrew computation watchdogs. Inspired by ℒ Lattice cryptocurrency consensus model (arXiv:2603.07947v1). Architecture: 5 ARM64 watchdog processes run identical Φ-corkscrew computations on the Fisher manifold S⁷. They propose checkpoints and must agree (4/5) on the result, tolerating up to 2 Byzantine faults. Classes: Checkpoint -- a single checkpoint on the manifold DAG -- directed acyclic graph of checkpoints ByzantineConsensus -- 5-way consensus with 2-fault tolerance Usage: from quintuplet_consensus import Checkpoint, DAG, ByzantineConsensus consensus = ByzantineConsensus(n_processes=5, threshold=4) result, clique, fault_report = consensus.run_consensus(proposals) """ from __future__ import annotations import hashlib import json import math import time from copy import deepcopy from dataclasses import dataclass, field from enum import Enum, auto from typing import Any, Optional # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- PHI = (1 + 5**0.5) / 2 # Golden ratio, governs Φ-corkscrew geometry class FaultType(Enum): """Classification of Byzantine faults on the Fisher manifold.""" NONE = auto() # Process is healthy and agrees CRASH = auto() # Process stopped responding (no proposal) BIT_FLIP = auto() # Single-bit corruption in state or index DETERMINISM_FAILURE = auto() # Nondeterministic computation (different result) STEALTH_FAULT = auto() # Near-correct result designed to evade detection GODEL_BOUNDARY = auto() # Hit undecidable region / self-referential paradox FAMM_CORRUPTION = auto() # FAMM pressure bank corrupted def __str__(self) -> str: return self.name class ConsensusResult(Enum): """Outcome of the consensus round.""" VALID = auto() # ≥4 processes agree; consensus reached TIED = auto() # No clique ≥4 (e.g., 2-vs-3 split or worse) INVALID = auto() # Clique found but manifold check failed NO_QUORUM = auto() # Too many crash faults to form any clique def __str__(self) -> str: return self.name # --------------------------------------------------------------------------- # Checkpoint # --------------------------------------------------------------------------- class Checkpoint: """A single checkpoint on the Fisher manifold S⁷. Each checkpoint records the state of one Φ-corkscrew computation step: - state: Hachimoji probability distribution on Δ₇ - spiral_index: integer position on the Φ-corkscrew - compression_ratio: original_size / compressed_size - timestamp: Unix epoch (seconds) - famm_pressure: FAMM bank pressure at this step - dag: DAG of prior checkpoints leading here - prev_hash: SHA-256 of the previous checkpoint receipt The checkpoint is content-addressed: its receipt_hash is a SHA-256 digest over all canonical fields, making any tamper evident. """ def __init__( self, state: list[float], spiral_index: int, compression_ratio: float, timestamp: float, famm_pressure: float, dag: "DAG", prev_hash: Optional[str] = None, ): self.state = tuple(float(x) for x in state) # immutable self.spiral_index = int(spiral_index) self.compression_ratio = float(compression_ratio) self.timestamp = float(timestamp) self.famm_pressure = float(famm_pressure) self.dag = dag # shared DAG reference self.prev_hash = prev_hash self._hash: Optional[str] = None # lazy # -- canonical serialisation for hashing ------------------------------- def canonical(self) -> dict[str, Any]: """Return a JSON-serialisable dict that uniquely describes the semantically-significant fields of this checkpoint.""" return { "state": [round(x, 12) for x in self.state], "spiral_index": self.spiral_index, "compression_ratio": round(self.compression_ratio, 8), "timestamp": round(self.timestamp, 6), "famm_pressure": round(self.famm_pressure, 8), "prev_hash": self.prev_hash, } def receipt_hash(self) -> str: """SHA-256 content hash (receipt ID). Lazily computed.""" if self._hash is None: payload = json.dumps(self.canonical(), sort_keys=True, separators=(",", ":")) self._hash = hashlib.sha256(payload.encode()).hexdigest() return self._hash # -- comparison for consensus ------------------------------------------ def agrees_with(self, other: "Checkpoint", epsilon: float = 1e-6) -> bool: """Structural agreement: two checkpoints are "the same result" when they land on the same manifold position with the same compression characteristics. Checks (in order of cost): 1. Spiral index equality (fast path) 2. Compression ratio within epsilon 3. FAMM pressure within epsilon 4. State vector L2 distance within epsilon 5. DAG node count match 6. DAG edge count match (topological proxy) """ if not isinstance(other, Checkpoint): return False # 1. Spiral index — injective on S⁷ for large n if self.spiral_index != other.spiral_index: return False # 2. Compression ratio if abs(self.compression_ratio - other.compression_ratio) > epsilon: return False # 3. FAMM pressure if abs(self.famm_pressure - other.famm_pressure) > epsilon: return False # 4. State vector distance on S⁷ if len(self.state) != len(other.state): return False dist = math.sqrt( sum((a - b) ** 2 for a, b in zip(self.state, other.state)) ) if dist > epsilon: return False # 5. DAG topology proxy — node count if len(self.dag.nodes) != len(other.dag.nodes): return False # 6. DAG edge count (lightweight structural check) self_edges = len(self.dag.edges) other_edges = len(other.dag.edges) if self_edges != other_edges: return False return True # -- utilities --------------------------------------------------------- def copy(self, new_dag: Optional["DAG"] = None) -> "Checkpoint": """Deep copy for fault-injection experiments. Args: new_dag: if provided, attach this DAG instead of copying. This breaks infinite recursion in DAG.copy(). """ return Checkpoint( state=list(self.state), spiral_index=self.spiral_index, compression_ratio=self.compression_ratio, timestamp=self.timestamp, famm_pressure=self.famm_pressure, dag=new_dag if new_dag is not None else self.dag, prev_hash=self.prev_hash, ) def __repr__(self) -> str: return ( f"Checkpoint(hash={self.receipt_hash()[:16]}..., " f"n={self.spiral_index}, C={self.compression_ratio:.2f}, " f"FAMM={self.famm_pressure:.4f})" ) def to_receipt(self, verified: bool = False) -> dict[str, Any]: """Emit the JSON receipt format used by SilverSight Lattice.""" return { "receiptID": self.receipt_hash(), "expression": "QUBO(n) via Φ-corkscrew on S⁷", "finalState": list(self.state), "spiralIndex": self.spiral_index, "compressionRatio": self.compression_ratio, "timestamp": self.timestamp, "fammPressure": self.famm_pressure, "dagDepth": len(self.dag.nodes), "prevReceipt": self.prev_hash, "manifoldVerified": verified, "guidanceAdjustment": 1.0, "verified": verified, } # --------------------------------------------------------------------------- # DAG # --------------------------------------------------------------------------- class DAG: """Directed acyclic graph of checkpoints. Each node is a Checkpoint; edges represent geodesic transitions on the Fisher manifold (i.e. γ(t_i) → γ(t_{i+1}) ). The DAG is kept acyclic by construction: edges only point from older checkpoints to newer ones (timestamp ordering). """ def __init__(self) -> None: self.nodes: dict[str, Checkpoint] = {} # receipt_hash → Checkpoint self.edges: set[tuple[str, str]] = set() # (parent_hash, child_hash) self._topo_order: list[str] = [] # cached topological order self._dirty: bool = True # -- mutation ---------------------------------------------------------- def add_node(self, checkpoint: Checkpoint) -> str: """Add a checkpoint as a node. Returns its receipt hash. If the checkpoint links to a previous hash, an edge is automatically created (prev_hash → checkpoint.hash). """ h = checkpoint.receipt_hash() self.nodes[h] = checkpoint self._dirty = True if checkpoint.prev_hash is not None: self.add_edge(checkpoint.prev_hash, h) return h def add_edge(self, parent: str, child: str) -> None: """Add a directed edge parent → child. Raises ValueError if the edge would create a cycle.""" if parent not in self.nodes or child not in self.nodes: raise KeyError("Both endpoints must be added as nodes first") if parent == child: raise ValueError("Self-loops are forbidden") # Quick cycle test: if child can already reach parent, adding # parent → child would close a cycle. if self._can_reach(child, parent): raise ValueError(f"Edge {parent} → {child} would create a cycle") self.edges.add((parent, child)) self._dirty = True def _can_reach(self, source: str, target: str) -> bool: """DFS reachability test (used for cycle detection).""" stack = [source] seen = set() while stack: cur = stack.pop() if cur == target: return True if cur in seen: continue seen.add(cur) for p, c in self.edges: if p == cur: stack.append(c) return False # -- queries ----------------------------------------------------------- def get_chain(self) -> list[Checkpoint]: """Return the primary chain (longest path) through the DAG. For a single-parent chain this is simply the topological order; for DAGs with merges the longest path heuristic is used. """ topo = self._topological_order() if not topo: return [] # longest path DP dist: dict[str, int] = {h: 1 for h in topo} pred: dict[str, Optional[str]] = {h: None for h in topo} for h in topo: for p, c in self.edges: if c == h and p in dist: if dist[p] + 1 > dist[h]: dist[h] = dist[p] + 1 pred[h] = p # reconstruct tail = max(dist, key=lambda k: dist[k]) chain: list[Checkpoint] = [] cur: Optional[str] = tail while cur is not None: chain.append(self.nodes[cur]) cur = pred[cur] chain.reverse() return chain def get_ancestors(self, receipt_hash: str) -> set[str]: """All ancestor hashes reachable from the given node.""" stack = [receipt_hash] seen: set[str] = set() while stack: cur = stack.pop() if cur in seen: continue seen.add(cur) for p, c in self.edges: if c == cur: stack.append(p) return seen def _topological_order(self) -> list[str]: """Kahn's algorithm, cached.""" if not self._dirty and self._topo_order: return self._topo_order in_degree: dict[str, int] = {h: 0 for h in self.nodes} adj: dict[str, list[str]] = {h: [] for h in self.nodes} for p, c in self.edges: adj[p].append(c) in_degree[c] = in_degree.get(c, 0) + 1 queue = [h for h, d in in_degree.items() if d == 0] order: list[str] = [] while queue: u = queue.pop(0) order.append(u) for v in adj[u]: in_degree[v] -= 1 if in_degree[v] == 0: queue.append(v) if len(order) != len(self.nodes): raise RuntimeError("Cycle detected in DAG — this should never happen") self._topo_order = order self._dirty = False return order # -- comparison helpers ------------------------------------------------ def isomorphic_to(self, other: "DAG", epsilon: float = 1e-6) -> bool: """Structural isomorphism check. For the consensus use-case we do not need full graph isomorphism; it suffices to compare: - node count - edge count - checksum of all receipt hashes (order-independent) - compression ratio fingerprint of chain """ if len(self.nodes) != len(other.nodes): return False if len(self.edges) != len(other.edges): return False # Hash multiset comparison — cheap and deterministic self_hashes = sorted(self.nodes.keys()) other_hashes = sorted(other.nodes.keys()) if self_hashes != other_hashes: return False return True def copy(self) -> "DAG": """Deep copy. Avoids infinite recursion by pre-creating the target DAG and passing it to Checkpoint.copy().""" d = DAG() # First pass: copy all checkpoints into the new DAG for cp in self.nodes.values(): cp_copy = cp.copy(new_dag=d) d.nodes[cp_copy.receipt_hash()] = cp_copy d._dirty = True # Second pass: reconstruct edges from prev_hash links for h, cp in d.nodes.items(): if cp.prev_hash is not None and cp.prev_hash in d.nodes: d.edges.add((cp.prev_hash, h)) return d def __repr__(self) -> str: return f"DAG(nodes={len(self.nodes)}, edges={len(self.edges)})" # --------------------------------------------------------------------------- # Byzantine Consensus # --------------------------------------------------------------------------- class ByzantineConsensus: """5-way Byzantine consensus with 2-fault tolerance. The protocol: 1. Collect proposals from all 5 watchdogs. 2. Build a 5×5 agreement matrix (bool) using Checkpoint.agrees_with. 3. Find the largest fully-connected agreeing clique. 4. If |clique| ≥ threshold (default 4): VALID consensus. 5. Otherwise: analyse fault pattern and report. Fault classes detected: CRASH — no proposal received BIT_FLIP — spiral index off by small power-of-two DETERMINISM_FAILURE — completely different non-erroneous result STEALTH_FAULT — barely outside epsilon, looks almost correct GODEL_BOUNDARY — spiral_index ≈ 0 or pressure ≈ 1.0 (boundary) FAMM_CORRUPTION — pressure inconsistent with DAG depth Attributes: n_processes (int): total watchdogs (default 5) threshold (int): minimum clique size for VALID (default 4) """ def __init__(self, n_processes: int = 5, threshold: int = 4): if n_processes < threshold: raise ValueError("threshold cannot exceed n_processes") self.n_processes = n_processes self.threshold = threshold self._agreement_matrix: list[list[bool]] = [] self._last_proposals: dict[int, Optional[Checkpoint]] = {} # -- proposal collection ----------------------------------------------- def propose(self, process_id: int, checkpoint: Checkpoint) -> None: """Register a checkpoint proposal from a watchdog. Args: process_id: integer 0 … n_processes-1 checkpoint: the proposed Checkpoint """ if not (0 <= process_id < self.n_processes): raise ValueError(f"process_id must be in [0, {self.n_processes})") self._last_proposals[process_id] = checkpoint # -- DAG comparison ---------------------------------------------------- def compare_dags(self, dag_i: DAG, dag_j: DAG, epsilon: float = 1e-6) -> bool: """Compare two DAGs for equivalence. Returns True iff the DAGs are structurally isomorphic AND every pairwise node agrees within epsilon. """ if not dag_i.isomorphic_to(dag_j, epsilon): return False # Deep check: each corresponding node must agree for h in dag_i.nodes: if h not in dag_j.nodes: return False if not dag_i.nodes[h].agrees_with(dag_j.nodes[h], epsilon): return False return True # -- core algorithm: clique finding ------------------------------------ def find_consensus_clique( self, proposals: dict[int, Optional[Checkpoint]], epsilon: float = 1e-6 ) -> list[int]: """Find the largest fully-connected clique of agreeing processes. Uses brute-force enumeration (optimal for n≤5). Returns the lexicographically-smallest largest clique. Args: proposals: mapping process_id → Checkpoint or None epsilon: numerical tolerance for agreement Returns: List of process IDs forming the largest clique. """ active = [pid for pid, cp in proposals.items() if cp is not None] m = len(active) if m == 0: return [] # Build agreement adjacency among active processes agree: dict[int, set[int]] = {pid: set() for pid in active} for i, pi in enumerate(active): for j, pj in enumerate(active): if i == j: agree[pi].add(pi) continue cp_i = proposals[pi] cp_j = proposals[pj] if cp_i is not None and cp_j is not None and cp_i.agrees_with(cp_j, epsilon): agree[pi].add(pj) # Store matrix for diagnostics self._agreement_matrix = [ [pj in agree[pi] for pj in active] for pi in active ] # Brute-force: enumerate all subsets of active processes best: list[int] = [] from itertools import combinations for size in range(m, 0, -1): found = False for subset in combinations(sorted(active), size): subset_set = set(subset) if all( subset_set <= agree[pid] # pid agrees with everyone in subset for pid in subset ): best = list(subset) found = True break if found: break return best # -- full protocol ----------------------------------------------------- def run_consensus( self, proposals: dict[int, Optional[Checkpoint]], epsilon: float = 1e-6, ) -> tuple[ConsensusResult, list[int], dict[str, Any]]: """Execute the full 5-way consensus protocol. Steps: 1. Collect all proposals (None = crash fault) 2. Build agreement matrix 3. Find largest consensus clique 4. Classify faults for out-of-clique processes 5. Return (result, clique, fault_report) Args: proposals: mapping process_id → Checkpoint or None epsilon: numerical tolerance Returns: (ConsensusResult, clique_pids, fault_report_dict) """ t0 = time.monotonic() self._last_proposals = dict(proposals) # Step 1 — normalise to all expected processes full: dict[int, Optional[Checkpoint]] = { pid: proposals.get(pid) for pid in range(self.n_processes) } crash_ids = [pid for pid, cp in full.items() if cp is None] # Step 2 & 3 — find clique clique = self.find_consensus_clique(full, epsilon) clique_set = set(clique) # Step 4 — classify faults fault_report: dict[str, Any] = { "timestamp": time.time(), "n_processes": self.n_processes, "threshold": self.threshold, "crash_faults": crash_ids, "clique": clique, "clique_size": len(clique), "agreement_matrix": self._agreement_matrix, "processes": {}, } for pid in range(self.n_processes): fault_report["processes"][pid] = { "present": full[pid] is not None, "in_clique": pid in clique_set, "fault_type": ( FaultType.NONE.name if pid in clique_set else self.detect_fault_type(pid, full, clique).name ), "proposal_summary": ( self._summarise(full[pid]) if full[pid] else None ), } # Determine result if len(clique) >= self.threshold: result = ConsensusResult.VALID # Manifold verification: all clique members on same geodesic if not self._manifold_verify(full, clique, epsilon): result = ConsensusResult.INVALID elif len(clique) == 0: result = ConsensusResult.NO_QUORUM else: result = ConsensusResult.TIED fault_report["result"] = result.name fault_report["duration_ms"] = round((time.monotonic() - t0) * 1000, 3) return result, clique, fault_report # -- fault classification ---------------------------------------------- def detect_fault_type( self, process_id: int, proposals: dict[int, Optional[Checkpoint]], clique: list[int], ) -> FaultType: """Classify WHY a process disagrees with the consensus clique. This is the diagnostic heart of the watchdog system. By examining the proposal of a dissenting process relative to the agreeing majority, we can distinguish: CRASH — no proposal at all BIT_FLIP — spiral_index differs by a power-of-two DETERMINISM_FAILURE — completely different result STEALTH_FAULT — values barely outside epsilon GODEL_BOUNDARY — spiral_index == 0 or pressure == 1.0 FAMM_CORRUPTION — pressure inconsistent with DAG depth Args: process_id: the dissenting process proposals: all proposals clique: the agreeing clique (used as ground truth) Returns: FaultType enum member """ cp = proposals.get(process_id) if cp is None: return FaultType.CRASH if not clique: # No clique to compare against — check for self-evident faults if cp.spiral_index == 0 and cp.famm_pressure > 0.9: return FaultType.GODEL_BOUNDARY return FaultType.DETERMINISM_FAILURE # Use clique centroid as ground truth truth = self._clique_centroid(proposals, clique) # Gödel boundary: undecidable / self-referential paradox region if cp.spiral_index == 0 or abs(cp.famm_pressure - 1.0) < 1e-4: return FaultType.GODEL_BOUNDARY # FAMM corruption: pressure inconsistent with DAG depth expected_pressure = min(1.0, len(cp.dag.nodes) * 0.1) if abs(cp.famm_pressure - expected_pressure) > 0.5: return FaultType.FAMM_CORRUPTION # Bit-flip: spiral index differs by a small power of two delta = abs(cp.spiral_index - truth.spiral_index) if delta > 0 and (delta & (delta - 1)) == 0 and delta <= 1024: return FaultType.BIT_FLIP # Stealth fault: very close to truth but outside epsilon # (deliberately crafted to evade simple detection) state_dist = math.sqrt( sum((a - b) ** 2 for a, b in zip(cp.state, truth.state)) ) if ( delta <= max(1, truth.spiral_index // 10000) and abs(cp.compression_ratio - truth.compression_ratio) <= truth.compression_ratio * 0.01 and state_dist <= 1e-3 ): return FaultType.STEALTH_FAULT # Fallback: the computation produced a fundamentally different # (but internally consistent) result — non-determinism return FaultType.DETERMINISM_FAILURE # -- internal helpers -------------------------------------------------- def _clique_centroid( self, proposals: dict[int, Optional[Checkpoint]], clique: list[int] ) -> Checkpoint: """Return a synthetic "average" checkpoint from the clique. Used as ground truth for fault classification.""" cps = [proposals[pid] for pid in clique if proposals.get(pid) is not None] assert cps # Representative: the first clique member (they all agree) return cps[0] def _manifold_verify( self, proposals: dict[int, Optional[Checkpoint]], clique: list[int], epsilon: float, ) -> bool: """Verify that all clique members lie on the same geodesic. Since they all agree on state and spiral_index (by construction of agrees_with), this reduces to a consistency check on the prev_hash chain.""" cps = [proposals[pid] for pid in clique if proposals.get(pid) is not None] if len(cps) < 2: return True # All should have the same prev_hash for a single-chain DAG prev_hashes = {cp.prev_hash for cp in cps} # In a real system we'd check geodesic equations; here the # structural agreement check in agrees_with suffices. return len(prev_hashes) <= 1 def _summarise(self, cp: Checkpoint) -> dict[str, Any]: """Short summary for fault reporting.""" return { "spiral_index": cp.spiral_index, "compression_ratio": round(cp.compression_ratio, 2), "famm_pressure": round(cp.famm_pressure, 6), "dag_nodes": len(cp.dag.nodes), } # -- diagnostics ------------------------------------------------------- def agreement_matrix_ascii(self) -> str: """Pretty-print the agreement matrix from the last consensus run.""" if not self._agreement_matrix: return "(no consensus run yet)" lines = ["Agreement matrix (last run):"] for row in self._agreement_matrix: lines.append(" " + " ".join("1" if v else "." for v in row)) return "\n".join(lines) # --------------------------------------------------------------------------- # Simulation helpers # --------------------------------------------------------------------------- def build_genesis_dag() -> DAG: """Create the genesis DAG (single node, no edges).""" dag = DAG() genesis = Checkpoint( state=[1 / 8] * 8, # uniform on Δ₇ spiral_index=0, compression_ratio=1.0, timestamp=0.0, famm_pressure=0.0, dag=dag, # self-referential, will be fixed below prev_hash=None, ) dag.add_node(genesis) return dag def simulate_watchdog( process_id: int, prev_dag: DAG, prev_hash: str, fault_mode: Optional[str] = None, ) -> Checkpoint: """Simulate one Φ-corkscrew watchdog computation. Args: process_id: 0 … 4 prev_dag: DAG from previous checkpoint prev_hash: receipt hash of previous checkpoint fault_mode: None, 'bit_flip', 'crash', 'determinism', 'stealth', 'godel', or 'famm' Returns: Checkpoint proposal (or raises for crash simulation) """ if fault_mode == "crash": raise RuntimeError(f"Watchdog {process_id} crashed") # Deterministic walk on S⁷ — ALL healthy processes must arrive # at the EXACT same result. process_id is NOT used in the # computation (only for identification / signing). step = 42 # iteration number (fixed for demo) # Geodesic walk: move deterministically from uniform distribution angle = 2 * math.pi * step * PHI target = [1 / 8 + 0.05 * math.cos(angle + i * 2 * math.pi / 8) for i in range(8)] # Normalise to probability simplex Δ₇ total = sum(abs(x) for x in target) state = [max(0.01, x / total) for x in target] s = sum(state) state = [x / s for x in state] # Spiral index: deterministic function of state spiral_index = int(sum(x * 1e8 for x in state)) % 1000000 + 1000 # Compression ratio grows with iteration depth compression_ratio = 268435456.0 * (1 + 0.01 * step) # FAMM pressure (LWMA-1 guided) famm_pressure = min(0.47, len(prev_dag.nodes) * 0.08) # Timestamp timestamp = time.time() # Build DAG: copy previous and append new checkpoint new_dag = prev_dag.copy() cp = Checkpoint( state=state, spiral_index=spiral_index, compression_ratio=compression_ratio, timestamp=timestamp, famm_pressure=famm_pressure, dag=new_dag, prev_hash=prev_hash, ) # Inject faults if fault_mode == "bit_flip": # Flip one bit in the spiral index (add a power of two) cp.spiral_index += 4 # 2^2 bit flip elif fault_mode == "determinism": # Completely different spiral index cp.spiral_index = 777777 cp.compression_ratio = 12345.0 elif fault_mode == "stealth": # Barely outside epsilon — crafted to evade detection cp.compression_ratio *= 1.00002 # 0.002% deviation elif fault_mode == "godel": # Force boundary condition cp.spiral_index = 0 cp.famm_pressure = 1.0 elif fault_mode == "famm": # Corrupt FAMM pressure to be inconsistent with DAG depth cp.famm_pressure = 99.9 new_dag.add_node(cp) return cp def inject_fault( proposals: dict[int, Checkpoint], target_id: int, fault_mode: str, ) -> dict[int, Optional[Checkpoint]]: """Inject a fault into the proposals of a specific process. Returns a new proposals dict with the fault applied. """ result: dict[int, Optional[Checkpoint]] = dict(proposals) if fault_mode == "crash": result[target_id] = None elif fault_mode == "bit_flip": cp = proposals[target_id].copy() cp.spiral_index += 4 result[target_id] = cp elif fault_mode == "determinism": cp = proposals[target_id].copy() cp.spiral_index = 999999 cp.compression_ratio = 1.0 result[target_id] = cp elif fault_mode == "stealth": cp = proposals[target_id].copy() cp.compression_ratio *= 1.00002 result[target_id] = cp elif fault_mode == "godel": cp = proposals[target_id].copy() cp.spiral_index = 0 cp.famm_pressure = 1.0 result[target_id] = cp elif fault_mode == "famm": cp = proposals[target_id].copy() cp.famm_pressure = 88.8 result[target_id] = cp return result # --------------------------------------------------------------------------- # Demonstration # --------------------------------------------------------------------------- def demo(): """Run a full consensus demonstration. Scenario: 5 watchdogs compute the same Φ-corkscrew step. Watchdog 3 suffers a bit-flip fault in its spiral_index. The consensus protocol finds the 4-agreeing clique and classifies the fault. """ print("=" * 72) print(" SilverSight Lattice — Quintuplet Byzantine Consensus Demo") print(" 5 watchdogs | 2-fault tolerant | 4/5 threshold") print("=" * 72) # --- Phase 1: Genesis ------------------------------------------------ print("\n[1] GENESIS — creating origin checkpoint on Δ₇") genesis_dag = build_genesis_dag() genesis_cp = list(genesis_dag.nodes.values())[0] genesis_hash = genesis_cp.receipt_hash() print(f" Genesis receipt: {genesis_hash[:24]}...") print(f" State: uniform (1/8)×8") print(f" Spiral index: {genesis_cp.spiral_index}") print(f" DAG: {genesis_dag}") # --- Phase 2: 5 watchdogs compute ------------------------------------ print("\n[2] WATCHDOG COMPUTATION — 5× Φ-corkscrew geodesic walk") proposals: dict[int, Checkpoint] = {} for pid in range(5): cp = simulate_watchdog(pid, genesis_dag, genesis_hash) proposals[pid] = cp print(f" W{pid}: n={cp.spiral_index}, C={cp.compression_ratio:.2e}, " f"FAMM={cp.famm_pressure:.4f}") # Verify all 5 agree initially all_same = all( proposals[i].agrees_with(proposals[0]) for i in range(1, 5) ) print(f"\n All 5 agree (pre-fault): {all_same}") # --- Phase 3: Inject bit-flip fault ---------------------------------- print("\n[3] FAULT INJECTION — bit-flip on watchdog 3") print(" Flipping bit 2^2 = +4 in spiral_index of W3") faulty_proposals = inject_fault(proposals, target_id=3, fault_mode="bit_flip") for pid in range(5): cp = faulty_proposals[pid] if cp is None: print(f" W{pid}: CRASH (no proposal)") elif pid == 3: print(f" W{pid}: n={cp.spiral_index} *** FAULT ***") else: print(f" W{pid}: n={cp.spiral_index}") # --- Phase 4: Byzantine consensus ------------------------------------ print("\n[4] BYZANTINE CONSENSUS — running agreement protocol") consensus = ByzantineConsensus(n_processes=5, threshold=4) result, clique, report = consensus.run_consensus(faulty_proposals) print(f" Result: {result}") print(f" Clique: {clique} (size {len(clique)})") print(f" Duration: {report['duration_ms']:.3f} ms") # Print agreement matrix print(f"\n {consensus.agreement_matrix_ascii()}") # --- Phase 5: Fault classification ----------------------------------- print("\n[5] FAULT CLASSIFICATION") for pid in range(5): info = report["processes"][pid] ft = info["fault_type"] status = "✓ AGREE" if info["in_clique"] else f"✗ {ft}" print(f" W{pid}: {status}") if not info["in_clique"] and info["present"]: detail = consensus.detect_fault_type(pid, faulty_proposals, clique) cp = faulty_proposals[pid] if clique: truth = consensus._clique_centroid(faulty_proposals, clique) delta_n = abs(cp.spiral_index - truth.spiral_index) if cp else "N/A" else: delta_n = "N/A (no clique)" print(f" delta_n={delta_n}, detected={detail.name}") # --- Phase 6: Receipt ------------------------------------------------ print("\n[6] CONSENSUS RECEIPT") if result == ConsensusResult.VALID and clique: accepted = faulty_proposals[clique[0]] receipt = accepted.to_receipt(verified=True) receipt["watchdogSignatures"] = [ {"watchdog": pid, "spiralIndex": ( faulty_proposals[pid].spiral_index if faulty_proposals[pid] else None ), "agree": pid in set(clique)} for pid in range(5) ] receipt["consensusClique"] = clique receipt["guidanceAdjustment"] = 1.05 print(json.dumps(receipt, indent=2, default=str)) else: print(" No valid consensus — checkpoint rejected") # --- Phase 7: Additional fault scenarios ----------------------------- print("\n[7] STRESS TEST — multiple fault scenarios") scenarios = [ ("clean", None, None), ("1× bit-flip", 2, "bit_flip"), ("1× crash", 4, "crash"), ("1× determinism", 0, "determinism"), ("1× stealth", 1, "stealth"), ("1× Gödel boundary", 3, "godel"), ("1× FAMM corrupt", 2, "famm"), ("2× fault (bit+crash)", None, "multi"), ] for name, target, mode in scenarios: if mode == "multi": test = inject_fault(proposals, 1, "bit_flip") test = inject_fault(test, 4, "crash") elif mode is None: test = dict(proposals) else: test = inject_fault(proposals, target, mode) res, clq, rep = consensus.run_consensus(test) ok = "✓ VALID" if res == ConsensusResult.VALID else f"✗ {res.name}" print(f" {name:24s} → {ok} clique={clq} " f"faults={ {pid: rep['processes'][pid]['fault_type'] for pid in range(5) if not rep['processes'][pid]['in_clique']} }") print("\n" + "=" * 72) print(" Demo complete.") print("=" * 72) return result, clique, report # --------------------------------------------------------------------------- # Entry point # --------------------------------------------------------------------------- if __name__ == "__main__": demo()