"""Bellman stopping controller MVP and mechanism checks. Run with: /home/maxwelhelp/main/bin/python3 bellman_stopping_experiment.py """ import json import math import random from dataclasses import dataclass from pathlib import Path import numpy as np SEED = 2092 rng = np.random.default_rng(SEED) @dataclass class ToyEnv: n: int = 21 # score states 0,.05,...,1 max_steps: int = 5 def p_improve(self, s): # Refinement has diminishing returns at high verifier score. return 0.10 + 0.75 * (1.0 - s) def transition(self, i): s = i / (self.n - 1) p = self.p_improve(s) return [(i, 1-p), (min(i+1, self.n-1), p)] def payoff(self, i): # Normalized terminal task utility / calibrated pass probability. return i / (self.n - 1) def bellman(env, cost): """Exact finite-horizon DP. V[k,i] is value with k refinements left.""" H, n = env.max_steps, env.n V = np.zeros((H + 1, n), dtype=float) policy = np.zeros((H, n), dtype=bool) # True = continue V[0] = np.arange(n) / (n - 1) q_all = np.zeros_like(policy, dtype=float) for left in range(1, H + 1): for i in range(n): expected = sum(prob * V[left-1][j] for j, prob in env.transition(i)) q = -cost + expected q_all[left-1, i] = q policy[left-1, i] = q > env.payoff(i) + 1e-12 V[left, i] = max(env.payoff(i), q) return V, policy, q_all def simulate(env, cost, adaptive=True, fixed_steps=None, trials=100000): V, policy, q = bellman(env, cost) n_calls, qualities, objectives, stops = [], [], [], [] for _ in range(trials): i = 0 calls = 0 while calls < env.max_steps: if adaptive: # policy indexed by remaining refinements. if not policy[env.max_steps - calls - 1, i]: break elif calls >= fixed_steps: break s = i / (env.n - 1) if rng.random() < env.p_improve(s): i = min(i + 1, env.n - 1) calls += 1 quality = env.payoff(i) n_calls.append(calls) qualities.append(quality) objectives.append(quality - cost * calls) stops.append(i) return { "calls": float(np.mean(n_calls)), "quality": float(np.mean(qualities)), "objective": float(np.mean(objectives)), "pass_at_0.75": float(np.mean(np.asarray(qualities) >= .75)), "stop_state_mean": float(np.mean(stops)), } def threshold(policy_row, env): # Return first score at which policy stops; score monotonicity can be checked. stop = np.flatnonzero(~policy_row) return float(stop[0] / (env.n - 1)) if len(stop) else None def main(): env = ToyEnv() # Prediction 1: continuation at the initial state decreases monotonically with c. costs = [0.00, 0.02, 0.03, 0.035, 0.038, 0.040, 0.045, 0.10] initial_continue = [] initial_thresholds = [] boundary_monotone = [] for c in costs: V, pol, q = bellman(env, c) initial_continue.append(bool(pol[env.max_steps-1, 0])) # At the first decision, policy row has max remaining horizon. row = pol[env.max_steps-1] initial_thresholds.append(threshold(row, env)) # Stop region should be an upper score interval. stop_indices = np.flatnonzero(~row) boundary_monotone.append(bool(len(stop_indices) == 0 or np.all(np.diff(stop_indices) == 1))) # Prediction 2: the stopping boundary moves to lower score as cost rises. # Ignore c=0's no-stop case when calculating finite boundaries. finite_thresholds = [x for x in initial_thresholds if x is not None] threshold_nonincreasing = all(a >= b for a, b in zip(finite_thresholds, finite_thresholds[1:])) # Prediction 3: Bellman value is non-increasing in cost and equals max(stop, continue). bellman_residuals, value_by_cost = [], [] for c in costs: V, pol, q = bellman(env, c) residual = np.max(np.abs(V[env.max_steps] - np.maximum(env.payoff(np.arange(env.n)), q[env.max_steps-1]))) bellman_residuals.append(float(residual)) value_by_cost.append(float(V[env.max_steps, 0])) value_nonincreasing = all(a >= b - 1e-12 for a, b in zip(value_by_cost, value_by_cost[1:])) # Monte Carlo comparison at moderate cost, with identical starting state. c_eval = 0.038 adaptive = simulate(env, c_eval, adaptive=True) fixed = {str(k): simulate(env, c_eval, adaptive=False, fixed_steps=k) for k in [1, 3, 5]} # A direct numerical check of Delta > c versus policy on every state/horizon. V, pol, q = bellman(env, c_eval) decisions_agree = bool(np.all(pol == (q > np.tile(env.payoff(np.arange(env.n)), (env.max_steps, 1))))) report = { "seed": SEED, "environment": {"score_states": env.n, "max_refinements": env.max_steps, "transition": "score increases one grid step with p=0.10+0.75*(1-score)", "payoff": "score"}, "math_check": { "max_bellman_residual": max(bellman_residuals), "delta_gt_cost_matches_policy": decisions_agree, "value_at_initial_state_by_cost": dict(zip(map(str, costs), value_by_cost)), }, "predictions": { "P1_initial_continue_decreases_with_cost": { "predicted": "monotone non-increasing", "costs": costs, "observed_continue": initial_continue, "observed_threshold_score": initial_thresholds, "confirmed": bool(all(a >= b for a, b in zip([int(x) for x in initial_continue], [int(x) for x in initial_continue][1:]))) }, "P2_stopping_boundary_is_upper_score_region_and_moves_lower_with_cost": { "predicted": "stop for sufficiently high score; boundary non-increasing in cost", "boundary_monotone_each_cost": boundary_monotone, "finite_boundaries": finite_thresholds, "confirmed": bool(all(boundary_monotone) and threshold_nonincreasing), }, "P3_bellman_value_decreases_with_cost": { "predicted": "non-increasing value and exact max recursion", "values": value_by_cost, "residuals": bellman_residuals, "confirmed": bool(value_nonincreasing and max(bellman_residuals) < 1e-12), }, }, "simulation_cost_0.038": {"adaptive": adaptive, "fixed": fixed}, "simulation_cost_sweep": { str(c): {"adaptive": simulate(env, c, adaptive=True, trials=30000)} for c in costs }, } report["worked"] = all(x["confirmed"] for x in report["predictions"].values()) and decisions_agree Path("results.json").write_text(json.dumps(report, indent=2)) print(json.dumps(report, indent=2)) if __name__ == "__main__": main()