Confidence guard: per-projection standard error, not historical RMSE

contrib/derate_amp_control.py

@@ -67,11 +67,13 @@
 gets a clean slate.
 
 Finally, ``will_trip: false`` is a point estimate, not a certainty — it means
-the *projected* plateau landed under the trip point, and that projection
-carries the model's own fit error. When the two are within
-``--forecast-confidence-k`` times ``fit_rmse_c`` of each other, that verdict
-is a coin flip dressed up as a decision, so the daemon steps down instead of
-trusting it. Observed live 2026-08-04: a projected 64.6 C plateau against a
+the *projected* plateau landed under the trip point, and every projection
+carries uncertainty. When plateau and trip point are within
+``--forecast-confidence-k`` times the projection's own standard error
+(``steady_state_se_c``, computed by the server per 30 s tick — wide early in
+a trajectory window, tight near the plateau; ``fit_rmse_c`` is the fallback
+for servers that don't report it), that verdict is a coin flip dressed up as
+a decision, so the daemon steps down instead of trusting it. Observed live 2026-08-04: a projected 64.6 C plateau against a
 65.0 C trip with ~0.31 C fit RMSE — a 1.3-sigma call that nothing in the
 logic had authority to act on, since only ``will_trip: true`` could trigger a
 cap. It held that time, but by luck rather than by design. Note this is
@@ -158,6 +160,7 @@ def decide(thermal: dict, state: State, cfg: Config) -> tuple[Action, State, str
     now_ts = thermal.get("ts")
     current_a = thermal.get("current_a")
     steady_state_c = forecast.get("steady_state_c")
+    steady_state_se_c = forecast.get("steady_state_se_c")
     model = thermal.get("model") or {}
     fit_rmse_c = model.get("fit_rmse_c")
     trip_c = model.get("trip_c", TRIP_HANDLE_C)
@@ -261,18 +264,30 @@ def decide(thermal: dict, state: State, cfg: Config) -> tuple[Action, State, str
     # the trip is not the danger; proximity plus an untrustworthy forecast
     # is. As fits improve and fit_rmse_c shrinks, this guard narrows on its
     # own and permits more aggressive operation.
+    # The denominator is this projection's own standard error when the
+    # server reports one (issue #4): early in a trajectory window the
+    # extrapolation is wild and the SE says so; near the plateau it
+    # tightens, and the guard relaxes with it. fit_rmse_c — a single
+    # model-adequacy constant across historical sessions — remains the
+    # fallback for older servers or windows too short for a meaningful SE.
     plateau_c = gap_c = sigma = None
+    sigma_src = None
     if (
         basis == "trajectory"
         and will_trip is False
         and isinstance(steady_state_c, (int, float))
-        and isinstance(fit_rmse_c, (int, float))
         and isinstance(trip_c, (int, float))
-        and fit_rmse_c > 0
     ):
-        plateau_c = float(steady_state_c)
-        gap_c = float(trip_c) - plateau_c
-        sigma = gap_c / float(fit_rmse_c)
+        if isinstance(steady_state_se_c, (int, float)) and steady_state_se_c > 0:
+            denom, sigma_src = float(steady_state_se_c), "proj se"
+        elif isinstance(fit_rmse_c, (int, float)) and fit_rmse_c > 0:
+            denom, sigma_src = float(fit_rmse_c), "fit rmse"
+        else:
+            denom = None
+        if denom is not None:
+            plateau_c = float(steady_state_c)
+            gap_c = float(trip_c) - plateau_c
+            sigma = gap_c / denom
 
     if sigma is not None and gap_c is not None and plateau_c is not None and sigma < cfg.forecast_confidence_k:
         streak = state.trip_streak + 1
@@ -283,7 +298,7 @@ def decide(thermal: dict, state: State, cfg: Config) -> tuple[Action, State, str
                 next_state,
                 (
                     f"plateau {plateau_c:.1f}C is only {gap_c:.1f}C under trip "
-                    f"({sigma:.1f} sigma, need {cfg.forecast_confidence_k:g}): "
+                    f"({sigma:.1f} sigma vs {sigma_src}, need {cfg.forecast_confidence_k:g}): "
                     f"{streak}/{cfg.confirm_ticks} confirming polls"
                 ),
             )
@@ -304,7 +319,7 @@ def decide(thermal: dict, state: State, cfg: Config) -> tuple[Action, State, str
             Action("cap", target),
             final_state,
             (
-                f"plateau {plateau_c:.1f}C only {gap_c:.1f}C under trip ({sigma:.1f} sigma): "
+                f"plateau {plateau_c:.1f}C only {gap_c:.1f}C under trip ({sigma:.1f} sigma vs {sigma_src}): "
                 f"forecast too uncertain to trust, stepping down to {target:g}A"
             ),
         )

docs/amp-control.md

@@ -59,18 +59,26 @@ session always starts fresh.
 
 One more guard covers a subtler failure: **`will_trip: false` is a point
 estimate, not a certainty.** It means the *projected* plateau landed under
-the trip point, and that projection carries the model's own fit error. When
-the two are within `--forecast-confidence-k` times `fit_rmse_c` of each
-other (default 2), that verdict is a coin flip dressed up as a decision, so
-the daemon steps down instead of trusting it. Live testing produced exactly
-this: a projected 64.6 °C plateau against a 65.0 °C trip with ~0.31 °C fit
-RMSE — a 1.3-sigma call that nothing in the logic had authority to act on,
-since only `will_trip: true` could trigger a cap. Note this is deliberately
+the trip point, and every projection carries uncertainty. The forecast
+reports its own: `steady_state_se_c`, the standard error of the projected
+plateau, computed per 30 s tick from the same regression that produces the
+plateau (floored at the handle sensor's 0.1 °C resolution). Early in a
+trajectory window, before much curvature is visible, it is honestly wide —
+measured on a real 43 A stretch: 0.8 °C at 24 seconds in, 0.11 °C by three
+minutes, ~0.04 °C near the plateau. When plateau and trip point are within
+`--forecast-confidence-k` times that error (default 2), the "no trip"
+verdict is a coin flip dressed up as a decision, so the daemon steps down
+instead of trusting it; when the projection is tight, the guard trusts it
+even close to the limit. (`fit_rmse_c`, the historical model-adequacy
+constant the guard originally used, remains the fallback for servers that
+don't report a per-projection error.) Live testing motivated this: a
+projected 64.6 °C plateau against a 65.0 °C trip — a coin-flip call that
+nothing in the logic had authority to act on. Note this is deliberately
 *not* a "handle is within X degrees of the trip" rule: the same session
 settled into a genuinely stable 63.8 °C plateau that such a rule would have
 banned outright. Proximity to the trip is not the danger; proximity plus an
-untrustworthy forecast is. As fits improve and `fit_rmse_c` shrinks, the
-guard narrows on its own and permits more aggressive operation.
+untrustworthy forecast is. As a window matures its SE shrinks, and the
+guard relaxes tick by tick on its own.
 
 A cap fully lifts three ways: the trajectory forecast reports the risk has
 passed *and* the handle has real thermal margin (stepped up gradually, see

docs/thermal-model.md

@@ -81,8 +81,11 @@ sessions run back-to-back with no idle gap to read ambient from, the
 forecast bridges with ambient inferred from the newest steady run still in
 the buffer instead of going dark.
 
-Every 30 s tick is recorded, so the session page can show in hindsight what
-was predicted against what the handle did. The line is labelled *predicted
+Each trajectory projection also reports its own standard error
+(`steady_state_se_c`) — wide early in a window, tight near the plateau —
+which is what the [amp controller](amp-control.md)'s confidence guard
+weighs margins against. Every 30 s tick is recorded, so the session page
+can show in hindsight what was predicted against what the handle did. The line is labelled *predicted
 plateau (if this current holds)* for a reason: it is the asymptote at the
 present current, not where a six-minute top-off will stop — see the faint
 model-only ticks before trajectory data exists.

tests/test_derate_amp_control.py

@@ -55,6 +55,7 @@ def _thermal(
     # Default plateau sits far below the trip so the confidence guard stays
     # dormant unless a test deliberately puts it in play.
     steady_state_c: float | None = 55.0,
+    steady_state_se_c: float | None = None,
     fit_rmse_c: float | None = 0.3,
     trip_c: float = 65.0,
 ):
@@ -70,6 +71,7 @@ def _thermal(
             "minutes_to_trip": mtt,
             "suggested_max_a": suggested,
             "steady_state_c": steady_state_c,
+            "steady_state_se_c": steady_state_se_c,
         },
     }
 
@@ -430,3 +432,42 @@ def test_event_for_restore_steps_from_the_active_cap_not_live_current():
     kind, detail = dac.event_for(dac.Action("restore", 34.0), "clear streak met", thermal, prev)
     assert kind == "amp_restored"
     assert detail["from_a"] == 32.0 and detail["to_a"] == 34.0
+
+
+def test_confidence_guard_prefers_projection_se_over_fit_rmse():
+    # Same 0.4C gap, but the projection itself is wide (SE 0.8): 0.5 sigma.
+    # Under the old fit_rmse denominator (0.1) this would read 4 sigma and
+    # the guard would sleep through exactly the case it exists for.
+    cfg = _cfg(confirm_ticks=1, forecast_confidence_k=2.0, restore_step_a=2.0)
+    state = dac.State(last_session_state="charging")
+    wide = _thermal(will_trip=False, mtt=None, suggested=None, steady_state_c=64.6,
+                    steady_state_se_c=0.8, fit_rmse_c=0.1, current_a=45.0)
+    action, state, reason = dac.decide(wide, state, cfg)
+    assert action.kind == "cap" and action.value == 43.0
+    assert "proj se" in reason and "too uncertain" in reason
+
+
+def test_confidence_guard_trusts_a_tight_projection_near_the_trip():
+    # 0.5C gap over a tight per-projection SE (0.12) is >4 sigma: the
+    # forecast has earned trust even this close to the limit. The old
+    # constant denominator (0.31) would have called this 1.6 sigma and
+    # stepped down, costing charge rate for no reason.
+    cfg = _cfg(confirm_ticks=1, forecast_confidence_k=2.0)
+    state = dac.State(capped=True, cap_value=45.0, last_session_state="charging")
+    tight = _thermal(will_trip=False, mtt=None, suggested=None, steady_state_c=64.5,
+                     steady_state_se_c=0.12, fit_rmse_c=0.31, handle_c=50.0)
+    action, state, reason = dac.decide(tight, state, cfg)
+    assert action.kind == "cap" and action.value == 47.0  # normal restore step-up
+    assert "stepping up" in reason
+
+
+def test_confidence_guard_falls_back_to_fit_rmse_without_se():
+    # Older server: no steady_state_se_c in the payload. The guard keeps
+    # its previous behavior against fit_rmse_c.
+    cfg = _cfg(confirm_ticks=1, forecast_confidence_k=2.0, restore_step_a=2.0)
+    state = dac.State(last_session_state="charging")
+    legacy = _thermal(will_trip=False, mtt=None, suggested=None, steady_state_c=64.6,
+                      steady_state_se_c=None, fit_rmse_c=0.31, current_a=45.0)
+    action, state, reason = dac.decide(legacy, state, cfg)
+    assert action.kind == "cap" and action.value == 43.0
+    assert "fit rmse" in reason

tests/test_wallmonitor.py

@@ -1341,3 +1341,26 @@ async def test_session_detail_includes_forecast_history(db):
     plateaus = [row["steady_state_c"] for row in detail["forecasts"]]
     assert plateaus == [70.0, 58.0, 54.0]
     assert all(row["session_id"] == sid for row in detail["forecasts"])
+
+
+def test_project_t_inf_se_wide_early_tight_late():
+    # Synthetic exponential toward 62C from 30C, tau 11 min, 0.05C noise
+    # (deterministic): the projection's standard error must be far larger
+    # from the first quarter of the ramp than from a window that has seen
+    # the bend, and the flat-window fallback reports no SE at all.
+    import math as m
+    tau_s = 11 * 60.0
+    curve = [(t, 62.0 - 32.0 * m.exp(-t / tau_s) + 0.05 * m.sin(t)) for t in range(0, 2400, 10)]
+    early = curve[:18]      # first 3 minutes
+    late = curve[:180]      # 30 minutes, bend well captured
+    t_early, se_early = thermal._project_t_inf(early, 11.0)
+    t_late, se_late = thermal._project_t_inf(late, 11.0)
+    assert se_early is not None and se_late is not None
+    assert se_early > 3 * se_late
+    assert abs(t_late - 62.0) < 0.5
+    # A noiseless flat window regresses to SSE = 0, so the raw SE is 0.0 —
+    # numerically true and physically overconfident, which is why predict()
+    # floors the published steady_state_se_c at the sensor's 0.1C step.
+    flat = [(t, 62.0) for t in range(0, 300, 10)]
+    t_flat, se_flat = thermal._project_t_inf(flat, 11.0)
+    assert t_flat == 62.0 and se_flat == 0.0

wallmonitor/thermal.py

@@ -730,14 +730,25 @@ def suggest_max_current(ambient_c: float, params: ThermalParams) -> float | None
     return float(amps)
 
 
-def _project_t_inf(window: list[tuple[float, float]], tau_min: float) -> float:
-    """Steady state projected from a steady-current window's trajectory.
+def _project_t_inf(window: list[tuple[float, float]], tau_min: float) -> tuple[float, float | None]:
+    """Steady state projected from a steady-current window's trajectory,
+    with the standard error of that projection.
 
     With tau known, T(t) = T_inf - C*exp(-t/tau) is linear in (T_inf, C), so
     an ordinary least-squares line on x = exp(-t/tau) gives an unbiased T_inf
     (a straight-line slope would read the window's average rate and overshoot
     during a fast ramp). No ambient input needed. A flat window (variance ~0,
     i.e. already converged) reads as the latest temperature.
+
+    T_inf is the fitted line's intercept at decay = 0 (t -> infinity), so its
+    standard error comes from the OLS intercept formula:
+    SE = s * sqrt(1/n + mean_x^2 / Sxx) with s^2 = SSE / (n - 2). Early in a
+    window, before much curvature is visible, mean_x is near 1 and Sxx is
+    tiny — the extrapolation is honest about being wild; near the plateau
+    it tightens. That is the per-projection uncertainty the amp controller's
+    confidence guard wants (issue #4) — fit_rmse_c is a model-adequacy score
+    across historical sessions and carries none of this variation. SE is
+    None when it cannot be computed (flat window, n <= 2).
     """
     count = len(window)
     decays = [math.exp(-(ts - window[0][0]) / (tau_min * 60.0)) for ts, _ in window]
@@ -748,7 +759,18 @@ def _project_t_inf(window: list[tuple[float, float]], tau_min: float) -> float:
         (decay - mean_decay) * (temp - mean_temp)
         for decay, (_, temp) in zip(decays, window)
     )
-    return mean_temp - (cov / var) * mean_decay if var > 1e-9 else window[-1][1]
+    if var <= 1e-9:
+        return window[-1][1], None
+    slope = cov / var
+    t_inf = mean_temp - slope * mean_decay
+    if count <= 2:
+        return t_inf, None
+    sse = sum(
+        (temp - (t_inf + slope * decay)) ** 2
+        for decay, (_, temp) in zip(decays, window)
+    )
+    se = math.sqrt(sse / (count - 2)) * math.sqrt(1.0 / count + mean_decay**2 / var)
+    return t_inf, se
 
 
 def _recent_steady_ambient(recent: list[dict], params: ThermalParams) -> float | None:
@@ -778,7 +800,7 @@ def _recent_steady_ambient(recent: list[dict], params: ThermalParams) -> float |
         if len(run) < TRAJECTORY_MIN_SAMPLES or run[-1]["ts"] - run[0]["ts"] < TRAJECTORY_MIN_SPAN_S:
             continue
         window = [(sample["ts"], sample["handle_temp_c"]) for sample in run]
-        t_inf = _project_t_inf(window, params.tau_min)
+        t_inf, _se = _project_t_inf(window, params.tau_min)
         run_current = median(sample["vehicle_current_a"] for sample in run)
         ambient = t_inf - params.rise_ref_c * (run_current / REF_CURRENT_A) ** 2
         if -30.0 <= ambient <= TRIP_HANDLE_C:
@@ -828,8 +850,14 @@ def predict(db: Database, now: float, params: ThermalParams) -> dict:
         window.reverse()
         forecast: dict = {}
         if len(window) >= TRAJECTORY_MIN_SAMPLES and window[-1][0] - window[0][0] >= TRAJECTORY_MIN_SPAN_S:
-            t_inf = _project_t_inf(window, tau_min)
+            t_inf, t_inf_se = _project_t_inf(window, tau_min)
             forecast["basis"] = "trajectory"
+            # This projection's own uncertainty — what the amp controller's
+            # confidence guard compares the margin against. Floored at the
+            # handle sensor's 0.1 C quantization: a perfectly smooth window
+            # can drive the raw SE below the sensor's resolution, and a
+            # guard fed that would claim more confidence than the data has.
+            forecast["steady_state_se_c"] = round(max(t_inf_se, 0.1), 2) if t_inf_se is not None else None
         else:
             # Too early at this current for a slope: model from ambient and
             # the present current scaled by I^2. Ambient comes from the LAN