File size: 4,215 Bytes
2ca80ad | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 | #!/usr/bin/env python3
"""
Verify bit-identical deterministic execution across multiple runs.
"""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent))
from material_field_engine import (
MaterialFieldEngine, VerifiedSubstrate, Vector2D, load_config
)
import hashlib
import json
def run_determinism_test(num_runs=5):
"""Run inference multiple times and verify identical outputs"""
print("Determinism Replay (5 Runs)" if num_runs == 5 else f"Determinism Replay ({num_runs} Runs)")
print()
config = load_config("balanced")
# Setup substrate and candidates (fixed seed)
substrate = VerifiedSubstrate()
substrate.add_verified_state(Vector2D(x=0.95, y=0.92, properties=None))
candidates = [
(0.95, 0.92), # Paris - near verified state
(0.35, 0.30), # Lyon
(0.30, 0.25), # Marseille
]
results = []
for run in range(num_runs):
# Create fresh engine each time
engine = MaterialFieldEngine(
substrate,
lambda_min=config['lambda_min'],
lambda_max=config['lambda_max'],
inference_steps=config['total_steps']
)
engine.phase_controller.nucleation_threshold = config['nucleation_threshold']
engine.phase_controller.quenching_threshold = config['quenching_threshold']
engine.initialize_candidates(candidates)
result = engine.run_inference()
# Create hash of the result
result_str = json.dumps({
'output': (result['final_output'].x, result['final_output'].y) if result['final_output'] else None,
'excluded_count': result['total_excluded'],
'phase_log': [
{
'step': e['step'],
'phase': e['phase'],
'survivors': e['survivors'],
'pressure': round(e['pressure'], 6) # Round to avoid floating point noise
}
for e in result['phase_log']
]
}, sort_keys=True)
result_hash = hashlib.sha256(result_str.encode()).hexdigest()
results.append(result_hash)
print(f"Run {run + 1}: {result_hash}")
print()
# Check if all hashes are identical
if len(set(results)) == 1:
print("Determinism: SHA-256 stable across runs")
print(f"SHA-256: {results[0]}")
return True
else:
print("Determinism: SHA-256 not stable across runs")
print("Unique hashes:")
for h in sorted(set(results)):
count = results.count(h)
print(f"{h}: {count}")
return False
def run_cross_run_stability_test():
"""Test that same input produces same output even across separate processes"""
print("\nCross-Run Stability")
print()
substrate = VerifiedSubstrate()
substrate.add_verified_state(Vector2D(x=0.95, y=0.92, properties=None))
config = load_config("balanced")
engine = MaterialFieldEngine(
substrate,
lambda_min=config['lambda_min'],
lambda_max=config['lambda_max'],
inference_steps=config['total_steps']
)
# Test specific vector coordinates
test_vectors = [
(0.95, 0.92),
(0.35, 0.30),
(0.50, 0.50),
(0.10, 0.10),
]
print("Vector | E | σ_y | ε | Hash")
print("-" * 80)
for x, y in test_vectors:
v = Vector2D(x=x, y=y, properties=None)
props = engine._compute_material_properties(v)
# Hash of all properties
props_str = f"{props.elastic_modulus:.10f},{props.yield_strength:.10f},{props.strain:.10f}"
props_hash = hashlib.sha256(props_str.encode()).hexdigest()[:8]
print(f"({x:.2f}, {y:.2f}) | {props.elastic_modulus:.4f} | {props.yield_strength:.4f} | "
f"{props.strain:.4f} | {props_hash}")
print("\nDerived Properties: deterministic")
if __name__ == "__main__":
success = run_determinism_test(num_runs=5)
run_cross_run_stability_test()
print()
raise SystemExit(0 if success else 1)
|