"""Deprecated wormhole simulator stub.""" from __future__ import annotations __all__ = ["run_experiment", "SimulationResult"] def run_experiment(*_: object, **__: object) -> None: # pragma: no cover raise RuntimeError( "wormhole_proof.core.wormhole was retired. Use SmoothConfig and " "run_smooth_experiment from wormhole_smooth.py." ) class SimulationResult: # pragma: no cover - compatibility shim def __init__(self, *args: object, **kwargs: object) -> None: raise RuntimeError( "SimulationResult no longer exists. Switch to the smooth pipeline " "(wormhole_smooth)." ) def _stack_history(history: Iterable[np.ndarray], dim: int) -> np.ndarray: rows = list(history) if not rows: return np.zeros((1, dim), dtype=float) return np.vstack(rows) def _compute_soft_modes( donor_hist: Iterable[np.ndarray], receiver_hist: Iterable[np.ndarray], dim: int, rank: int, rng: np.random.Generator, random_subspace: bool, ) -> Tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]: if random_subspace: mat = rng.standard_normal((dim, rank)) u, _ = np.linalg.qr(mat) eigs = np.ones(rank, dtype=float) return u, u.copy(), eigs, eigs.copy() donor_stack = _stack_history(donor_hist, dim) receiver_stack = _stack_history(receiver_hist, dim) cov_d = donor_stack.T @ donor_stack + 1e-9 * np.eye(dim) eigvals_d, eigvecs_d = np.linalg.eigh(cov_d) order_d = np.argsort(eigvals_d)[::-1] donor_basis = eigvecs_d[:, order_d[:rank]] cov_r = receiver_stack.T @ receiver_stack + 1e-9 * np.eye(dim) eigvals_r, eigvecs_r = np.linalg.eigh(cov_r) order_r = np.argsort(eigvals_r)[::-1] receiver_basis = eigvecs_r[:, order_r[:rank]] donor_eigs = np.clip(eigvals_d[order_d[:rank]], a_min=0.0, a_max=None) receiver_eigs = np.clip(eigvals_r[order_r[:rank]], a_min=0.0, a_max=None) return donor_basis, receiver_basis, donor_eigs, receiver_eigs def _update_state( state: np.ndarray, proposal: np.ndarray, lambda_h: float, epsilon: float, ) -> Tuple[np.ndarray, float]: eta = 1.0 / (1.0 + lambda_h) update = eta * proposal norm = np.linalg.norm(update) if norm < epsilon: update = np.zeros_like(update) norm = 0.0 new_state = state + update return new_state, float(norm) def simulate_regime( config: WormholeConfig, rng: np.random.Generator, regime: str, baseline_throughput: float | None = None, warmup_schedule: bool = True, ) -> RegimeResult: state_a = np.zeros(config.dim, dtype=float) state_b = np.zeros(config.dim, dtype=float) debt_balance = 0.0 budgets = np.full(2, config.budget_total / 2.0, dtype=float) history_a: Deque[np.ndarray] = deque(maxlen=config.horizon_steps) history_b: Deque[np.ndarray] = deque(maxlen=config.horizon_steps) horizon_counters = np.zeros(2, dtype=int) throughput = 0.0 commits = 0 defects = 0 packets_sent = 0 first_lock_step: int | None = None base_lambda = config.lambda_h final_lambda = config.lambda_h_final if config.lambda_h_final is not None else base_lambda base_epsilon = config.epsilon final_epsilon = config.epsilon_final if config.epsilon_final is not None else base_epsilon base_step_scale = config.base_step_scale final_step_scale = ( config.base_step_scale_final if config.base_step_scale_final is not None else base_step_scale ) base_wormhole_scale = config.wormhole_step_scale final_wormhole_scale = ( config.wormhole_step_scale_final if config.wormhole_step_scale_final is not None else base_wormhole_scale ) base_verify_tol = config.verify_tolerance final_verify_tol = ( config.verify_tolerance_final if config.verify_tolerance_final is not None else base_verify_tol ) base_traversable_mu = config.traversable_mu final_traversable_mu = ( config.traversable_mu_final if config.traversable_mu_final is not None else base_traversable_mu ) base_shock_energy = config.shock_energy final_shock_energy = ( config.shock_energy_final if config.shock_energy_final is not None else base_shock_energy ) warmup_steps = max(config.warmup_steps, 0) log = logging.getLogger("wormhole") power_iters = 3 candidate_norms: List[float] = [] soft_top_eigs: List[float] = [] soft_active_ranks: List[float] = [] percolation_spans: List[float] = [] committed_deltas: List[float] = [] update_ratios: List[float] = [] prelock_update_norms: List[float] = [] lambda_max_values: List[float] = [] cfl_margins: List[float] = [] packet_energies: List[float] = [] traversable_kicks: List[float] = [] arrival_steps: List[float] = [] state_corrs: List[float] = [] shock_delay_records: List[float] = [] hp_enabled = ( regime == "wormhole" and config.hp_message_bits > 0 and config.hp_decode_packets > 0 ) hp_message_indices: np.ndarray | None = None hp_message_signs: np.ndarray | None = None hp_message_vector: np.ndarray | None = None hp_message_injected = False hp_message_pending = False hp_scramble_count = 0 hp_packets_used = 0 hp_decode_accum = np.zeros(config.hp_message_bits if hp_enabled else 1, dtype=float) hp_partial_match = [] hp_injection_step: int | None = None hp_trace: List[HPTraceEntry] = [] geometry: RegimeGeometry | None = RegimeGeometry() if regime == "wormhole" else None shock_enabled = regime == "wormhole" and ( config.shock_energy > 0.0 or (config.shock_energy_final or 0.0) > 0.0 ) shock_sent = False shock_gap_remaining = 0 shock_arrival_step: float | None = None shock_measured = False require_escrow = regime != "no_escrow" require_verify = regime != "no_verify" random_subspace = regime == "random_subspace" traversable_window = 0 for step in range(config.steps): if warmup_schedule and warmup_steps > 0: weight = min(step / warmup_steps, 1.0) else: weight = 1.0 lambda_step = base_lambda + (final_lambda - base_lambda) * weight epsilon_step = base_epsilon + (final_epsilon - base_epsilon) * weight base_scale_step = base_step_scale + (final_step_scale - base_step_scale) * weight wormhole_scale_step = ( base_wormhole_scale + (final_wormhole_scale - base_wormhole_scale) * weight ) verify_tol_step = base_verify_tol + (final_verify_tol - base_verify_tol) * weight traversable_mu_step = ( base_traversable_mu + (final_traversable_mu - base_traversable_mu) * weight ) shock_energy_step = base_shock_energy + (final_shock_energy - base_shock_energy) * weight proposal_a = rng.normal(scale=base_scale_step, size=config.dim) proposal_b = rng.normal(scale=base_scale_step, size=config.dim) state_a, delta_a = _update_state(state_a, proposal_a, lambda_step, epsilon_step) state_b, delta_b = _update_state(state_b, proposal_b, lambda_step, epsilon_step) prelock_update_norms.append(0.5 * (delta_a + delta_b)) update_ratios.append(float(delta_a / (np.linalg.norm(proposal_a) + 1e-12))) history_a.append(proposal_a) history_b.append(proposal_b) horizon_counters[0] = horizon_counters[0] + 1 if delta_a < epsilon_step else 0 horizon_counters[1] = horizon_counters[1] + 1 if delta_b < epsilon_step else 0 if debt_balance > 0.0: repayment = min(config.debt_repay_rate, debt_balance) debt_balance -= repayment budgets += repayment / 2.0 horizon_lock = ( horizon_counters[0] >= config.horizon_steps and horizon_counters[1] >= config.horizon_steps ) if not horizon_lock: continue if first_lock_step is None: first_lock_step = step log.debug( "regime=%s lock step=%d delta_a=%.4e delta_b=%.4e", regime, step, delta_a, delta_b, ) if ( regime == "wormhole" and config.traversable_mu > 0.0 and config.traversable_window > 0 ): traversable_window = config.traversable_window if hp_enabled and not hp_message_injected: hp_message_indices = rng.choice(config.dim, size=config.hp_message_bits, replace=False) hp_message_signs = rng.choice([-1.0, 1.0], size=config.hp_message_bits) hp_message_vector = np.zeros(config.dim, dtype=float) hp_message_vector[hp_message_indices] = ( hp_message_signs * config.hp_message_magnitude ) state_a += hp_message_vector history_a.append(hp_message_vector) hp_message_injected = True hp_message_pending = True hp_scramble_count = max(config.hp_scramble_wait, 0) hp_injection_step = step if regime == "baseline" or packets_sent >= config.max_packets: continue if hp_enabled and hp_scramble_count > 0: hp_scramble_count -= 1 continue donor_basis, receiver_basis, donor_eigs, receiver_eigs = _compute_soft_modes( history_a, history_b, config.dim, config.soft_rank, rng, random_subspace ) if donor_basis.shape[1] > 0: m = receiver_basis.T @ donor_basis v = rng.normal(size=m.shape[0]) candidate = receiver_basis @ (m @ v) candidate_norm = np.linalg.norm(candidate) candidate_norms.append(candidate_norm) soft_top_eigs.append(donor_eigs[0]) soft_active_ranks.append(donor_basis.shape[1]) if require_escrow: escrow = candidate_norm * config.escrow_factor budgets -= escrow / 2.0 debt_balance += escrow if require_verify: verify_error = np.linalg.norm(state_a - state_b) if verify_error > verify_tol_step: defects += 1 log.warning( "regime=%s step=%d verify error=%.4e", regime, step, verify_error, ) continue packet_energy = np.linalg.norm(candidate) packet_energies.append(packet_energy) if shock_enabled and not shock_sent: shock_sent = True shock_gap_remaining = config.shock_gap if shock_gap_remaining > 0: shock_gap_remaining -= 1 continue if shock_enabled and not shock_measured: shock_arrival_step = step shock_measured = True if regime == "wormhole" and traversable_window > 0: traversable_kicks.append(traversable_window) traversable_window -= 1 if hp_enabled: hp_packets_used += 1 hp_decode_accum += candidate hp_partial_match.append(np.linalg.norm(hp_decode_accum)) committed_deltas.append(candidate_norm) cfl_margins.append(config.cfl_margin) state_a += candidate state_b += candidate history_a.append(candidate) history_b.append(candidate) packets_sent += 1 throughput += packet_energy if geometry is not None: geometry.state_a.append(state_a.tolist()) geometry.state_b.append(state_b.tolist()) geometry.candidate.append(candidate.tolist()) invariant_ok = defects == 0 budget_error = np.abs(budgets[0] - budgets[1]) defect_rate = defects / packets_sent if packets_sent > 0 else 0.0 if hp_enabled: hp_decode_norm = float(np.linalg.norm(hp_decode_accum)) if hp_message_vector is not None: message_norm = float(np.linalg.norm(hp_message_vector)) else: message_norm = 0.0 if hp_message_vector is not None and hp_decode_norm > 0.0 and message_norm > 0.0: hp_cosine_metric = float(np.dot(hp_decode_accum, hp_message_vector) / (hp_decode_norm * message_norm)) else: hp_cosine_metric = 0.0 else: hp_decode_norm = 0.0 hp_cosine_metric = 0.0 metrics = RegimeMetrics( relaxation_steps=first_lock_step or 0.0, candidate_norm_mean=np.mean(candidate_norms) if candidate_norms else 0.0, candidate_norm_std=np.std(candidate_norms) if candidate_norms else 0.0, soft_top_eig_mean=np.mean(soft_top_eigs) if soft_top_eigs else 0.0, soft_active_rank_mean=np.mean(soft_active_ranks) if soft_active_ranks else 0.0, percolation_span_mean=np.mean(percolation_spans) if percolation_spans else 0.0, percolation_span_max=np.max(percolation_spans) if percolation_spans else 0.0, committed_delta_mean=np.mean(committed_deltas) if committed_deltas else 0.0, packets_sent=packets_sent, update_ratio_mean=np.mean(update_ratios) if update_ratios else 0.0, prelock_update_mean=np.mean(prelock_update_norms) if prelock_update_norms else 0.0, traversable_kick_total=np.sum(traversable_kicks) if traversable_kicks else 0.0, traversable_kick_mean=np.mean(traversable_kicks) if traversable_kicks else 0.0, arrival_delay_mean=np.mean(arrival_steps) if arrival_steps else 0.0, arrival_delay_min=np.min(arrival_steps) if arrival_steps else 0.0, state_corr_mean=np.mean(state_corrs) if state_corrs else 0.0, shock_delay_mean=np.mean(shock_delay_records) if shock_delay_records else 0.0, shock_delay_min=np.min(shock_delay_records) if shock_delay_records else 0.0, shock_delay_count=len(shock_delay_records), hp_message_bits=config.hp_message_bits if hp_enabled else 0.0, hp_packets_used=hp_packets_used, hp_match_fraction=np.mean(hp_partial_match) if hp_partial_match else 0.0, hp_success=hp_packets_used >= config.hp_decode_packets, hp_l2_error=hp_decode_norm, hp_cosine=hp_cosine_metric, hp_injection_step=hp_injection_step or 0, hp_scramble_wait=config.hp_scramble_wait, hp_partial_match_final=np.mean(hp_partial_match) if hp_partial_match else 0.0, lambda_max_mean=np.mean(lambda_max_values) if lambda_max_values else 0.0, cfl_margin_min=np.min(cfl_margins) if cfl_margins else 0.0, packet_energy_mean=np.mean(packet_energies) if packet_energies else 0.0, ) notes: str if regime == "baseline": notes = "horizons locked; throughput ~0" elif regime == "wormhole": if baseline_throughput is not None and baseline_throughput > 0: gain = throughput / baseline_throughput traversable_tag = " with traversable window" if traversable_kicks else "" notes = f"throughput gain ×{gain:.1f}{traversable_tag}" else: base = "wormhole opened; baseline ~0" traversable_tag = " with traversable window" if traversable_kicks else "" notes = f"{base}{traversable_tag}" elif regime == "no_escrow": notes = "escrow removed → budget drift" elif regime == "no_verify": notes = "verify disabled → expect defects" else: notes = "random subspace → throughput collapse" trace = RegimeTrace(hp_trace=hp_trace) if hp_trace else None if geometry is not None and not geometry.state_a: geometry = None return RegimeResult( name=regime, throughput=throughput, commits=commits, invariant_ok=invariant_ok, budget_error=budget_error, debt_balance=debt_balance, defect_rate=defect_rate, notes=notes, metrics=metrics, trace=trace, geometry=geometry, ) def run_experiment(config: WormholeConfig, seed: int) -> SimulationResult: """Execute the full set of regimes for a given seed.""" rng = np.random.default_rng(seed) baseline = simulate_regime(config, rng, regime="baseline") rng = np.random.default_rng(seed) wormhole = simulate_regime( config, rng, regime="wormhole", baseline_throughput=baseline.throughput, ) regimes = [baseline, wormhole] for regime in ("no_escrow", "no_verify", "random_subspace"): rng = np.random.default_rng(seed) regimes.append(simulate_regime(config, rng, regime=regime)) return SimulationResult(seed=seed, config=config, regimes=regimes)