1"""Small certified temporal-budget runtime for a learned controller."""
 2from dataclasses import dataclass
 3from typing import Callable, Optional
 4
 5
 6@dataclass
 7class Contract:
 8    phi0: float
 9    gamma: float
10    delta: float
11    t0: float
12
13    @property
14    def budget(self) -> float:
15        if self.gamma <= 0:
16            return float("inf") if self.phi0 >= 0 else -float("inf")
17        return self.phi0 / self.gamma - self.delta
18
19    def remaining(self, t: float) -> float:
20        """Conservative usable latency budget at time t."""
21        return self.budget - (t - self.t0)
22
23    def permits(self, t: float, latency: float) -> bool:
24        return self.phi0 >= 0 and latency <= self.remaining(t)
25
26
27class CertifiedScheduler:
28    """Certificate gate between sensing and a learned policy."""
29    def __init__(self, gamma: float, delta: float, fallback: Callable):
30        if gamma < 0 or delta < 0:
31            raise ValueError("gamma and delta must be nonnegative")
32        self.gamma, self.delta, self.fallback = gamma, delta, fallback
33        self.contract: Optional[Contract] = None
34        self.evaluations = 0
35
36    def refresh(self, phi: float, t: float) -> Contract:
37        self.contract = Contract(phi, self.gamma, self.delta, t)
38        return self.contract
39
40    def choose(self, state, t: float, measured_latency: float,
41               policy: Callable):
42        if self.contract is None:
43            return self.fallback(state), False
44        if self.contract.permits(t, measured_latency):
45            self.evaluations += 1
46            return policy(state), True
47        return self.fallback(state), False