import json import math from pathlib import Path import numpy as np SEED = 2734 rng = np.random.default_rng(SEED) # 1-D specialization of a 2-D obstacle margin: Phi(x)=distance to obstacle. # The plant moves toward the obstacle with constant speed v, so dPhi/dt=-v exactly. def certificate_sweep(): gamma_true = 1.0 gamma_est = 1.25 # conservative bound delta = 0.05 distances = np.linspace(0.15, 2.0, 80) latencies = np.linspace(0.0, 2.0, 81) permitted = [] unsafe_permitted = [] for d in distances: for L in latencies: contract_permits = L <= d / gamma_est - delta physically_safe = L <= d / gamma_true + 1e-12 permitted.append(contract_permits) unsafe_permitted.append(contract_permits and not physically_safe) # The implication is checked over a dense grid, including the boundary. conservative_violations = int(np.sum(unsafe_permitted)) # With an intentionally underestimated gamma, the observed permit/fail # transition is compared to the exact physical boundary d/gamma_true-delta. under_gamma = 0.75 transitions = [] for d in distances: ls = np.linspace(0, 2.5, 10001) permits = ls <= d / under_gamma - delta unsafe = ls > d / gamma_true bad = np.where(permits & unsafe)[0] transitions.append(float(ls[bad[0]]) if len(bad) else float("nan")) valid = np.isfinite(transitions) observed_underestimate_boundary = float(np.nanmedian(np.array(transitions)[valid])) # For this sweep the onset is expected near the smallest boundary where # under-estimation starts to allow an unsafe handoff: d/gamma_true. expected_underestimate_boundary = float(np.median(distances[distances / under_gamma - delta > distances / gamma_true])) return { "grid_points": len(distances) * len(latencies), "conservative_contract_violations": conservative_violations, "conservative_violation_rate": conservative_violations / len(permitted), "underestimated_gamma": under_gamma, "underestimate_unsafe_onset_median_L": observed_underestimate_boundary, "expected_physical_boundary_median_L": expected_underestimate_boundary, } def derivative_verification(): # Phi(t)=phi0-v*t and gamma_true=v: finite differences satisfy # dPhi/dt=-gamma_true, hence Phi(t)-Phi(0)+gamma_true*t == 0. phi0, gamma_true = 1.7, 1.3 ts = np.linspace(0.0, 1.0, 1001) phi = phi0 - gamma_true * ts finite_slopes = np.diff(phi) / np.diff(ts) residual = phi - (phi0 - gamma_true * ts) return {"max_abs_slope_error": float(np.max(np.abs(finite_slopes + gamma_true))), "max_contract_residual": float(np.max(np.abs(residual))), "predicted_unit_rate": 1.0, "observed_normalized_slope": float(np.max(-finite_slopes / gamma_true))} def gamma_boundary_sweep(): # Prediction: for a fixed latency L and margin delta, permission changes at # gamma_est = d/(L+delta). Conservative gamma has no unsafe permissions; # underestimated gamma creates unsafe permissions. d = 1.0 L = 1.05 delta = 0.05 gamma_true = 1.0 gammas = np.linspace(0.4, 1.8, 141) unsafe = [] permitted = [] for g in gammas: p = L <= d / g - delta permitted.append(p) unsafe.append(p and L > d / gamma_true) transition = gammas[np.where(np.array(permitted) == False)[0][0]] predicted = d / (L + delta) return { "predicted_gamma_transition": predicted, "observed_gamma_transition_grid": float(transition), "transition_abs_error": abs(float(transition) - predicted), "unsafe_permissions_at_conservative_gamma": int(sum(unsafe[i] for i,g in enumerate(gammas) if g >= gamma_true)), "unsafe_permissions_at_underestimated_gamma": int(sum(unsafe[i] for i,g in enumerate(gammas) if g < gamma_true)), "predicted_unsafe_onset_gamma": predicted, "observed_first_unsafe_gamma": float(gammas[np.where(np.array(unsafe))[0][-1]]) if np.any(unsafe) else None, } def evaluation_scaling(): # Prediction: a constant-distance contract with usable interval # B=d/gamma-delta-L has evaluation rate 1/B, and rate ratios scale as # the inverse certified budget. d, delta, L, horizon = 1.0, 0.05, 0.05, 100.0 gammas = np.array([0.5, 1.0, 1.5, 2.0]) rows = [] for g in gammas: B = d / g - delta - L observed = math.ceil(horizon / B) predicted = horizon / B rows.append({"gamma": float(g), "budget": B, "observed_evals": observed, "predicted_evals": predicted, "rate_error": abs(observed-predicted)/predicted}) # Relative scaling is the important quantity and is not affected by ceil. return {"rows": rows, "rate_ratio_observed_gamma2_over_gamma1": rows[3]["observed_evals"] / rows[1]["observed_evals"], "rate_ratio_predicted_gamma2_over_gamma1": rows[3]["predicted_evals"] / rows[1]["predicted_evals"]} def scheduler_comparison(): # Same finite approach episodes, random initial distance and speed. A # handoff is unsafe if latency exceeds the distance remaining at launch. n = 3000 d0 = rng.uniform(0.5, 2.0, n) v = rng.uniform(0.5, 1.5, n) L = rng.uniform(0.0, 0.8, n) delta = 0.05 gamma = 1.6 # safely above the sampled vmax # ordinary periodic inference: fixed interval, no certificate fallback periodic_h = 0.50 periodic_unsafe = L > np.maximum(d0 - v * periodic_h, 0) / v # contract: only launch when the certified remaining interval covers latency; # otherwise fallback/urgent refresh is used, so unsafe handoffs are rejected. budget = d0 / gamma - delta contract_launch = L <= budget contract_unsafe = contract_launch & (L > d0 / v) # Naive event trigger uses a fixed distance threshold, not a rate certificate. naive_launch = d0 > 0.25 naive_unsafe = naive_launch & (L > d0 / v) return { "episodes": n, "periodic_unsafe_handoff_rate": float(periodic_unsafe.mean()), "naive_event_unsafe_handoff_rate": float(naive_unsafe.mean()), "contract_unsafe_handoff_rate": float(contract_unsafe.mean()), "contract_launch_fraction": float(contract_launch.mean()), "contract_fallback_fraction": float((~contract_launch).mean()), "periodic_evaluations_per_episode": 1.0, "contract_evaluations_per_episode": float(contract_launch.mean()), } def main(): out = {"seed": SEED, "derivative_verification": derivative_verification(), "math_verification": certificate_sweep(), "gamma_boundary_sweep": gamma_boundary_sweep(), "evaluation_scaling": evaluation_scaling(), "scheduler_comparison": scheduler_comparison()} Path("results.json").write_text(json.dumps(out, indent=2)) print(json.dumps(out, indent=2)) if __name__ == "__main__": main()