""" MANIFOLD VERIFICATION — SilverSight Lattice Quintuplet Watchdog ================================================================ Verifies that 5 watchdog processes running Φ-corkscrew computation on the Fisher manifold S⁷ are actually on the SAME GEODESIC after reaching Byzantine consensus on a checkpoint. Mathematical Framework ---------------------- - State space: Δ₇ (7-simplex of 8 Hachimoji probabilities) p ∈ Δ₇ ⟺ p_i ≥ 0, Σᵢ pᵢ = 1 for i ∈ {0, ..., 7} - Fisher-Rao metric on Δ₇: gᵢⱼ(p) = δᵢⱼ/pᵢ + 1/p₇ for i,j ∈ {0, ..., 6} (using indices 0..6 with p₇ = 1 − Σᵢ₌₀⁶ pᵢ as the dependent coordinate) - √p-coordinate embedding (Chentsov): xᵢ = √pᵢ for i ∈ {0, ..., 7} This maps Δ₇ → S⁷₊ (positive octant of the 7-sphere) The Fisher metric becomes the ROUND metric on S⁷. - Geodesics: great circles on S⁷ (intersections of S⁷ with 2-planes through the origin). - Bhattacharyya / Fisher distance: d(p, q) = arccos( Σᵢ₌₀⁷ √(pᵢ qᵢ) ) This is the great-circle arc length on S⁷ in √p coordinates. Core Insight for Verification ----------------------------- Processes can agree on a checkpoint (similar DAGs) while being on different geodesics. This is the STEALTH FAULT — the most dangerous failure mode. To detect it, we verify that all agreeing processes lie on the same great circle of S⁷. A set of points {x⁽ᵏ⁾} on S⁷ lies on a single great circle iff all x⁽ᵏ⁾ are contained in a common 2-plane through the origin, i.e. rank[x⁽⁰⁾ | x⁽¹⁾ | ... ] ≤ 2. Equivalently: every x⁽ᵏ⁾ is orthogonal to the (ℓ-2)-dimensional normal space of the plane spanned by the first two independent vectors. Author: Lattice Watchdog Geometric Verification Module License: Internal — SilverSight Consensus Stack """ from __future__ import annotations __all__ = [ "fisher_rao_metric", "fisher_distance", "geodesic_point", "verify_same_geodesic", "verify_geodesic_consistency", "detect_stealth_fault", "is_valid_probability", "to_sqrt_sphere", "from_sqrt_sphere", "great_circle_plane", "angular_deviation_from_plane", ] import math from typing import List, Tuple import numpy as np # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- DIM_SIMPLEX: int = 7 # 7-simplex Δ₇ N_PROBS: int = 8 # 8 probabilities p₀ … p₇ PHI: float = (1.0 + math.sqrt(5.0)) / 2.0 GOLDEN_ANGLE_RAD: float = 2.0 * math.pi / (PHI ** 2) EPS: float = 1e-12 # numerical zero # =========================================================================== # 1. Probability / Manifold utilities # =========================================================================== def is_valid_probability(p: np.ndarray) -> bool: """Check whether *p* is a point on Δ₇ (non-negative, sums to 1).""" p = np.asarray(p, dtype=float) if p.shape != (N_PROBS,): return False if np.any(p < -EPS): return False if abs(p.sum() - 1.0) > 1e-9: return False return True def to_sqrt_sphere(p: np.ndarray) -> np.ndarray: """Embed Δ₇ → S⁷ via the Chentsov map xᵢ = √pᵢ. Returns a unit vector in ℝ⁸ because Σᵢ (√pᵢ)² = Σᵢ pᵢ = 1. """ p = np.asarray(p, dtype=float) assert is_valid_probability(p), "Input must lie on Δ₇" x = np.sqrt(np.maximum(p, 0.0)) # Renormalise to protect against drift norm = np.linalg.norm(x) if norm > EPS: x = x / norm return x def from_sqrt_sphere(x: np.ndarray) -> np.ndarray: """Inverse Chentsov map: S⁷ ∩ ℝ₊⁸ → Δ₇ via pᵢ = xᵢ².""" x = np.asarray(x, dtype=float) assert x.shape == (N_PROBS,), f"Expected shape {(N_PROBS,)}, got {x.shape}" p = x * x s = p.sum() if s > EPS: p = p / s return p # =========================================================================== # 2. Fisher-Rao geometry # =========================================================================== def fisher_rao_metric(p: np.ndarray) -> np.ndarray: """Compute the Fisher-Rao metric matrix at point *p* on Δ₇. Using the independent coordinates {p₀, …, p₆} with p₇ = 1 − Σᵢ₌₀⁶ pᵢ, the metric components are gᵢⱼ = δᵢⱼ / pᵢ + 1 / p₇ for i, j ∈ {0, …, 6} Parameters ---------- p : np.ndarray, shape (8,) A point on the 7-simplex (non-negative, sum = 1). Returns ------- g : np.ndarray, shape (7, 7) The Fisher information matrix in the {p₀,…,p₆} coordinate basis. """ p = np.asarray(p, dtype=float) assert is_valid_probability(p), "fisher_rao_metric: p must be on Δ₇" p7 = p[-1] # dependent coordinate if p7 < EPS: raise ValueError("fisher_rao_metric: p₇ is too close to zero — metric singular") if np.any(p[:DIM_SIMPLEX] < EPS): raise ValueError("fisher_rao_metric: some pᵢ too close to zero — metric singular") # g_ij = δ_ij / p_i + 1 / p_7 g = np.eye(DIM_SIMPLEX, dtype=float) / p[:DIM_SIMPLEX][:, None] g += 1.0 / p7 return g def fisher_distance(p: np.ndarray, q: np.ndarray) -> float: """Fisher-Rao distance between two points on Δ₇. In √p coordinates this is the great-circle arc length on S⁷: d(p, q) = arccos( Σᵢ₌₀⁷ √(pᵢ qᵢ) ) The result lies in [0, π]. For nearby points this is ≈ the chord length; for antipodal points on the sphere it equals π. Parameters ---------- p, q : np.ndarray, shape (8,) Two points on Δ₇. Returns ------- float The Fisher-Rao / Bhattacharyya angle between *p* and *q*. """ p = np.asarray(p, dtype=float) q = np.asarray(q, dtype=float) assert is_valid_probability(p), "fisher_distance: p must be on Δ₇" assert is_valid_probability(q), "fisher_distance: q must be on Δ₇" # Bhattacharyya coefficient: BC = Σ √(pᵢ qᵢ) # This is exactly the Euclidean inner product of √p and √q on S⁷. bc = np.sum(np.sqrt(np.maximum(p * q, 0.0))) # Numerical guard: clamp to [-1, 1] before arccos bc = float(np.clip(bc, -1.0, 1.0)) return math.acos(bc) # =========================================================================== # 3. Geodesic operations on S⁷ # =========================================================================== def geodesic_point(p: np.ndarray, d: np.ndarray, t: float) -> np.ndarray: """Point at Fisher distance *t* along the geodesic from *p* in direction *d*. On S⁷ in √p coordinates geodesics are great circles, so we use the standard spherical exponential map: γ(t) = cos(t) · x + sin(t) · v̂ where x = √p (unit vector on S⁷) and v̂ = projₓ(d) / |projₓ(d)| (tangent direction, normalised). The result is projected back from √p coordinates to Δ₇ via pᵢ = xᵢ². Parameters ---------- p : np.ndarray, shape (8,) Starting point on Δ₇. d : np.ndarray, shape (8,) Direction vector in the tangent space at *p* (in √p coordinates). This need NOT be normalised or exactly tangent — it is projected. t : float Distance along the geodesic (same units as fisher_distance). Use small *t* for local steps. Returns ------- q : np.ndarray, shape (8,) The point on Δ₇ reached after walking distance *t*. """ p = np.asarray(p, dtype=float) d = np.asarray(d, dtype=float) assert is_valid_probability(p), "geodesic_point: p must be on Δ₇" assert d.shape == (N_PROBS,), f"geodesic_point: d must have shape {(N_PROBS,)}" x = to_sqrt_sphere(p) # unit vector on S⁷ # Project *d* onto the tangent space TₓS⁷: remove radial component d_dot_x = float(d @ x) v = d - d_dot_x * x # now orthogonal to x v_norm = np.linalg.norm(v) if v_norm < EPS: # Direction is radial — stay at the same point return p.copy() v_hat = v / v_norm # unit tangent vector # Great circle formula on S⁷ gamma_sqrt = math.cos(t) * x + math.sin(t) * v_hat # Renormalise (protect against drift) gamma_sqrt_norm = np.linalg.norm(gamma_sqrt) if gamma_sqrt_norm > EPS: gamma_sqrt = gamma_sqrt / gamma_sqrt_norm # Map back to Δ₇ q = from_sqrt_sphere(gamma_sqrt) return q def great_circle_plane(x0: np.ndarray, x1: np.ndarray) -> Tuple[np.ndarray, np.ndarray, np.ndarray]: """Compute the 2-plane in ℝ⁸ that contains the great-circle geodesic through two points *x0*, *x1* on S⁷. Returns ------- u, v : np.ndarray, shape (8,) Orthonormal basis vectors spanning the geodesic plane. normal : np.ndarray, shape (8, 6) Matrix whose columns form an orthonormal basis of the normal space to the geodesic plane (so that a point *y* lies in the plane iff y @ normal ≈ 0). Raises ------ ValueError If *x0* and *x1* are antipodal (lie on opposite poles) or collinear, in which case the geodesic is not unique. """ x0 = np.asarray(x0, dtype=float) x1 = np.asarray(x1, dtype=float) # Normalise x0 = x0 / (np.linalg.norm(x0) + EPS) x1 = x1 / (np.linalg.norm(x1) + EPS) # Check for antipodal / collinear inner = float(x0 @ x1) if abs(abs(inner) - 1.0) < 1e-6: raise ValueError( "great_circle_plane: points are (anti)collinear — " "geodesic plane is not unique" ) # First basis vector: x0 u = x0.copy() # Second basis vector: Gram-Schmidt orthonormalise (x1 - proj_u(x1)) v_raw = x1 - inner * u v_norm = np.linalg.norm(v_raw) if v_norm < EPS: raise ValueError("great_circle_plane: degenerate points") v = v_raw / v_norm # Normal space: orthonormal complement of span{u, v} in ℝ⁸ basis = np.eye(N_PROBS, dtype=float) # Use QR decomposition on [u, v, e₀, e₁, …] and extract Q[:, 2:] M = np.column_stack([u, v, basis]) Q, _ = np.linalg.qr(M) normal = Q[:, 2:] # shape (8, 6) return u, v, normal def angular_deviation_from_plane( x: np.ndarray, normal: np.ndarray, ) -> float: """Compute the angular deviation (in radians) of a point *x* on S⁷ from the 2-plane defined by *normal* (columns = orthonormal basis of the normal space). A point lies exactly in the plane iff the deviation is 0. """ x = np.asarray(x, dtype=float) x = x / (np.linalg.norm(x) + EPS) # Components in the normal directions normal_comps = x @ normal # shape (6,) sin_theta = np.linalg.norm(normal_comps) sin_theta = float(np.clip(sin_theta, 0.0, 1.0)) return math.asin(sin_theta) # =========================================================================== # 4. Core verification routines # =========================================================================== def verify_same_geodesic( positions: List[np.ndarray], threshold: float = 1e-4, check_coincidence: bool = True, ) -> Tuple[bool, dict]: """Verify that all *positions* lie on the same geodesic of S⁷. Algorithm --------- 1. Convert every position to √p coordinates (points on S⁷). 2. Find the pair of points with maximal angular separation and use them to build the geodesic plane (numerically stable). 3. Measure the angular deviation of every point from that plane. 4. (Optional) Compute pairwise Fisher distances — this checks that processes are not just on the same geodesic but at the same checkpoint position. 5. PASS iff (a) all angular deviations < *threshold*, AND (b) if *check_coincidence*: all pairwise Fisher distances < *threshold*. Parameters ---------- positions : list of np.ndarray, each shape (8,) Points on Δ₇ from the consensus clique (typically 4–5 points). threshold : float Maximum allowed deviation (radians for angle; also used for Fisher distance when *check_coincidence* is True). Default 1e-4 is ~0.006°. check_coincidence : bool If True (default), also require that all points are close to each other in Fisher distance — this is the right mode when the consensus clique should be at the SAME checkpoint. Set to False when checking coplanarity of points that may be legitimately spread along the same geodesic. Returns ------- ok : bool True if all processes are on the same geodesic (and optionally at the same position), False otherwise. report : dict Detailed measurements: angular deviations, pairwise distances, and human-readable diagnostics. """ if len(positions) < 2: return False, { "error": "Need at least 2 points to define a geodesic", "n_points": len(positions), } # Validate and convert to √p coordinates sqrt_pts = [] for idx, p in enumerate(positions): if not is_valid_probability(np.asarray(p)): return False, {"error": f"positions[{idx}] is not on Δ₇", "point": p} sqrt_pts.append(to_sqrt_sphere(p)) # ------------------------------------------------------------------ # 4a. Angular-deviation test (great-circle coplanarity) # ------------------------------------------------------------------ angular_devs = [] plane_info = {} angular_ok: bool | None = None max_angular_dev = math.inf # Find the pair of points with maximal angular separation to define # the plane — this is numerically more stable than using adjacent # points that may be nearly collinear. n_pts = len(sqrt_pts) best_pair = (0, min(1, n_pts - 1)) best_sep = -1.0 for i in range(n_pts): for j in range(i + 1, n_pts): sep = float(abs(sqrt_pts[i] @ sqrt_pts[j])) # We want points that are well-separated (inner product small) # but not antipodal (inner product ≈ -1). score = 1.0 - abs(sep) # 0 = collinear, 1 = orthogonal if score > best_sep and score > 1e-3: best_sep = score best_pair = (i, j) try: u, v, normal = great_circle_plane(sqrt_pts[best_pair[0]], sqrt_pts[best_pair[1]]) plane_info = { "basis_u": u.tolist(), "basis_v": v.tolist(), "plane_from_pair": best_pair, } for idx, x in enumerate(sqrt_pts): dev = angular_deviation_from_plane(x, normal) angular_devs.append({ "index": idx, "deviation_rad": float(dev), "deviation_deg": float(math.degrees(dev)), }) max_angular_dev = max(d["deviation_rad"] for d in angular_devs) angular_ok = max_angular_dev < threshold except ValueError as exc: # Points are collinear / antipodal — fall back to distance-only angular_ok = None max_angular_dev = math.inf plane_info = {"error": str(exc)} # ------------------------------------------------------------------ # 4b. Pairwise Fisher-distance test (redundant safety net) # ------------------------------------------------------------------ n = len(positions) pairwise = [] max_fisher_dist = 0.0 max_pair = (0, 0) for i in range(n): for j in range(i + 1, n): d = fisher_distance(positions[i], positions[j]) pairwise.append({ "i": i, "j": j, "fisher_distance": float(d), }) if d > max_fisher_dist: max_fisher_dist = d max_pair = (i, j) distance_ok = max_fisher_dist < threshold # ------------------------------------------------------------------ # 4c. Decide # ------------------------------------------------------------------ if angular_ok is not None: ok = angular_ok if check_coincidence: ok = ok and distance_ok else: # Degraded mode: angular test failed, fall back to distance ok = distance_ok if check_coincidence else False report = { "verdict": "PASS" if ok else "FAIL", "n_points": n, "threshold": threshold, "angular_test": { "performed": angular_ok is not None, "max_deviation_rad": float(max_angular_dev), "max_deviation_deg": float(math.degrees(max_angular_dev)), "details": angular_devs, }, "distance_test": { "max_fisher_distance": float(max_fisher_dist), "max_pair": max_pair, "pairwise": pairwise, }, "plane": plane_info, "same_geodesic": ok, } return ok, report def verify_geodesic_consistency( chain_0: List[np.ndarray], chain_i: List[np.ndarray], tolerance: float = 0.05, ) -> Tuple[bool, float]: """Verify that process *i* followed the same geodesic as process 0. We compare the **step sizes** (Fisher distances between consecutive checkpoints). If both processes traverse the same geodesic with the same speed profile, the step-size sequences must match up to a common scaling factor (clock drift). Algorithm --------- 1. Compute step-size arrays: s_0[k] = d(chain_0[k], chain_0[k+1]) s_i[k] = d(chain_i[k], chain_i[k+1]) 2. Find the optimal ratio α that minimises |s_i − α·s_0|₂. (This accounts for a constant clock-speed difference.) 3. Compute the maximum relative deviation after scaling. 4. PASS if max deviation < *tolerance* (default 5 %). Parameters ---------- chain_0 : list of np.ndarray Reference checkpoints from the leader process. chain_i : list of np.ndarray Checkpoints from process *i*. tolerance : float Maximum allowed relative deviation (default 5 % to account for clock jitter and floating-point noise). Returns ------- ok : bool True if the step-size profiles are consistent. max_deviation : float Maximum relative deviation between scaled step sequences. """ if len(chain_0) < 2 or len(chain_i) < 2: return False, math.inf # Compute step sizes s_0 = np.array([fisher_distance(chain_0[k], chain_0[k + 1]) for k in range(len(chain_0) - 1)]) s_i = np.array([fisher_distance(chain_i[k], chain_i[k + 1]) for k in range(len(chain_i) - 1)]) min_len = min(len(s_0), len(s_i)) s_0 = s_0[:min_len] s_i = s_i[:min_len] # Filter out zero steps (shouldn't happen in normal operation) mask = (s_0 > EPS) & (s_i > EPS) if not np.any(mask): return False, math.inf s_0_f = s_0[mask] s_i_f = s_i[mask] # Optimal scale factor: minimise |s_i − α·s_0|² → α = (s_i·s_0)/(s_0·s_0) alpha = float(s_i_f @ s_0_f) / float(s_0_f @ s_0_f) if alpha < EPS or not np.isfinite(alpha): return False, math.inf # Relative deviation after scaling rel_dev = np.abs(s_i_f - alpha * s_0_f) / (alpha * s_0_f + EPS) max_deviation = float(np.max(rel_dev)) ok = max_deviation < tolerance return ok, max_deviation # =========================================================================== # 5. Stealth-fault detection # =========================================================================== def detect_stealth_fault( proposals: List[dict], clique: List[int], ) -> Tuple[bool, dict]: """Detect the stealth fault: DAGs agree but manifold positions diverge. After Byzantine consensus, the processes in *clique* have all committed to the same checkpoint DAG. This routine extracts their actual manifold positions from the *proposals* and verifies that they are on the same geodesic. Parameters ---------- proposals : list of dict One dict per process. Each dict must contain a key ``'manifold_position'`` whose value is an np.ndarray of shape (8,) on Δ₇, and optionally a key ``'dag_hash'`` for cross-checking. clique : list of int Process indices that formed the consensus clique. Returns ------- stealth_detected : bool True if the DAGs agree BUT the positions are NOT all on the same geodesic (this is the attack / fault). details : dict Full diagnostic report. """ if len(clique) < 2: return False, {"error": "Clique too small for geodesic check", "clique": clique} # Extract positions of clique members positions = [] missing = [] for pid in clique: if pid < 0 or pid >= len(proposals): missing.append(pid) continue prop = proposals[pid] if "manifold_position" not in prop: missing.append(pid) continue positions.append(np.asarray(prop["manifold_position"], dtype=float)) if missing: return True, { "stealth_detected": True, "reason": "missing_positions", "missing_processes": missing, "clique": clique, } # Check DAG agreement (basic sanity) dag_hashes = [] for pid in clique: h = proposals[pid].get("dag_hash") if h is not None: dag_hashes.append(h) dags_agree = len(set(dag_hashes)) <= 1 if dag_hashes else True # Now the crucial test: are they on the same geodesic AND at the # same checkpoint position? Use a slightly relaxed threshold for # stealth detection because real processes have small clock jitter. geodesic_ok, geo_report = verify_same_geodesic( positions, threshold=1e-3, check_coincidence=True ) stealth_detected = dags_agree and (not geodesic_ok) details = { "stealth_detected": stealth_detected, "dags_agree": dags_agree, "n_clique": len(clique), "n_positions": len(positions), "geodesic_report": geo_report, } if stealth_detected: details["severity"] = "CRITICAL" details["description"] = ( "STEALTH FAULT: Byzantine clique agreed on DAG but processes " "are on DIFFERENT GEODESICS of S⁷. This is a manifold-level " "equivocation attack — the processes appear consistent from " "the consensus layer but diverge in the physical state space." ) return stealth_detected, details # =========================================================================== # 6. Helpers for unit tests / demos # =========================================================================== def random_point_on_simplex(rng: np.random.Generator | None = None) -> np.ndarray: """Generate a uniformly random point on Δ₇ (Dirichlet(1,…,1)).""" rng = rng or np.random.default_rng() p = rng.random(N_PROBS) p = p / p.sum() return p def random_tangent_direction(x: np.ndarray, rng: np.random.Generator | None = None) -> np.ndarray: """Generate a random unit vector in the tangent space TₓS⁷.""" rng = rng or np.random.default_rng() v = rng.standard_normal(N_PROBS) v = v - (v @ x) * x # project onto tangent space v_norm = np.linalg.norm(v) if v_norm < EPS: return random_tangent_direction(x, rng) return v / v_norm # =========================================================================== # 7. Demonstration # =========================================================================== if __name__ == "__main__": print("=" * 70) print("MANIFOLD VERIFICATION — SilverSight Lattice Quintuplet Watchdog") print("Fisher manifold S⁷ | Φ-corkscrew consensus verification") print("=" * 70) rng = np.random.default_rng(42) # ------------------------------------------------------------------ # 7.1 Create a common geodesic (shared by 4 honest processes) # ------------------------------------------------------------------ print("\n--- 1. Creating 4 honest processes on the SAME GEODESIC ---") # Random starting point on Δ₇ p0 = random_point_on_simplex(rng) x0 = to_sqrt_sphere(p0) # Random tangent direction → defines a unique great circle d_hat = random_tangent_direction(x0, rng) # Walk 5 checkpoints along the great circle N_CHECKPOINTS = 5 STEP_SIZE = 0.1 # Fisher distance per step JITTER = 1e-5 # tiny clock jitter (small enough for clean tests) honest_chains: List[List[np.ndarray]] = [[] for _ in range(4)] for proc in range(4): for k in range(N_CHECKPOINTS): t = k * STEP_SIZE + rng.normal(0, JITTER) # Great circle: γ(t) = cos(t)·x0 + sin(t)·d̂ x_t = math.cos(t) * x0 + math.sin(t) * d_hat x_t = x_t / (np.linalg.norm(x_t) + EPS) p_t = from_sqrt_sphere(x_t) honest_chains[proc].append(p_t) # Print first and last checkpoint for this process p_first = honest_chains[proc][0] p_last = honest_chains[proc][-1] print(f" Process {proc}: d(p₀, p₄) = {fisher_distance(p_first, p_last):.6f}") # ------------------------------------------------------------------ # 7.2 Verify all 4 honest processes are on the same geodesic # ------------------------------------------------------------------ print("\n--- 2. Verifying SAME GEODESIC for honest processes ---") latest_positions = [chain[-1] for chain in honest_chains] ok, report = verify_same_geodesic( latest_positions, threshold=1e-3, check_coincidence=True ) print(f" Result: {'PASS ✓' if ok else 'FAIL ✗'}") print(f" Max angular deviation: {report['angular_test']['max_deviation_deg']:.6f}°") print(f" Max Fisher distance: {report['distance_test']['max_fisher_distance']:.8f}") # ------------------------------------------------------------------ # 7.3 Verify geodesic consistency (step-size profiles match) # ------------------------------------------------------------------ print("\n--- 3. Verifying GEODESIC CONSISTENCY (step-size profiles) ---") for proc in range(1, 4): ok_dev, max_dev = verify_geodesic_consistency(honest_chains[0], honest_chains[proc]) status = "PASS ✓" if ok_dev else "FAIL ✗" print(f" Process {proc} vs 0: max_dev = {max_dev:.6f} ({status})") # ------------------------------------------------------------------ # 7.4 Introduce a BYZANTINE process on a DIFFERENT GEODESIC # ------------------------------------------------------------------ print("\n--- 4. Creating 1 BYZANTINE process on a DIFFERENT GEODESIC ---") # Same starting point, but DIFFERENT tangent direction d_byz = random_tangent_direction(x0, rng) # Ensure it's actually different (not close to d_hat) dot = abs(float(d_byz @ d_hat)) while dot > 0.3: d_byz = random_tangent_direction(x0, rng) dot = abs(float(d_byz @ d_hat)) byz_chain: List[np.ndarray] = [] for k in range(N_CHECKPOINTS): t = k * STEP_SIZE + rng.normal(0, JITTER) x_t = math.cos(t) * x0 + math.sin(t) * d_byz x_t = x_t / (np.linalg.norm(x_t) + EPS) byz_chain.append(from_sqrt_sphere(x_t)) p_byz_first = byz_chain[0] p_byz_last = byz_chain[-1] print(f" Byzantine: d(p₀, p₄) = {fisher_distance(p_byz_first, p_byz_last):.6f}") print(f" Direction dot-product |honest · byz|: {dot:.4f} (low = different geodesic)") # ------------------------------------------------------------------ # 7.5 Verify the mixed group (4 honest + 1 byzantine) FAILS # ------------------------------------------------------------------ print("\n--- 5. Verifying mixed group (4 honest + 1 byzantine) ---") mixed_positions = latest_positions + [byz_chain[-1]] ok_mix, report_mix = verify_same_geodesic( mixed_positions, threshold=1e-3, check_coincidence=True ) print(f" Result: {'PASS ✓' if ok_mix else 'FAIL ✗'}") print(f" Max angular deviation: {report_mix['angular_test']['max_deviation_deg']:.6f}°") print(f" Max Fisher distance: {report_mix['distance_test']['max_fisher_distance']:.8f}") if not ok_mix: print(" → Byzantine process detected via manifold divergence!") # ------------------------------------------------------------------ # 7.6 Detect STEALTH FAULT: DAGs agree but positions diverge # ------------------------------------------------------------------ print("\n--- 6. STEALTH FAULT detection ---") # Build mock proposals: all processes claim the same DAG hash dag_hash = "checkpoint_42_deadbeef" proposals = [] for proc in range(4): proposals.append({ "dag_hash": dag_hash, "manifold_position": honest_chains[proc][-1], }) # Byzantine process claims the SAME DAG but is on a different geodesic proposals.append({ "dag_hash": dag_hash, "manifold_position": byz_chain[-1], }) clique_all = [0, 1, 2, 3, 4] # all 5 appeared to agree stealth_detected, details = detect_stealth_fault(proposals, clique_all) print(f" DAGs agree: {details['dags_agree']}") print(f" Same geodesic: {details['geodesic_report']['same_geodesic']}") print(f" STEALTH DETECTED: {stealth_detected}") if stealth_detected: print(f" SEVERITY: {details['severity']}") print(f" Description: {details['description']}") # Now test the honest-only clique (should NOT detect stealth) print("\n --- Honest-only clique (control) ---") clique_honest = [0, 1, 2, 3] stealth_h, details_h = detect_stealth_fault(proposals, clique_honest) print(f" STEALTH DETECTED: {stealth_h} (expected False)") if not stealth_h: print(" → Control passed: honest clique correctly cleared.") # ------------------------------------------------------------------ # 7.7 Summary # ------------------------------------------------------------------ print("\n" + "=" * 70) print("SUMMARY") print("=" * 70) honest_geo, _ = verify_same_geodesic( latest_positions, threshold=1e-3, check_coincidence=True ) print(f" Honest clique same-geodesic test: {'PASS ✓' if honest_geo else 'FAIL ✗'}") print(f" Mixed clique same-geodesic test: {'FAIL ✓ (Byzantine caught)' if not ok_mix else 'PASS ✗'}") print(f" Stealth fault on 5-process clique: DETECTED = {stealth_detected}") print(f" Stealth fault on honest clique: DETECTED = {stealth_h} (should be False)") print("=" * 70)