fix: drift watch — gate truncated fits, buzz the phone only for confirmed drift

docs/notifications.md

@@ -38,7 +38,11 @@ sudo ./install-service.sh --host <wall-connector-ip> [your other flags] \
 On the phone, install the ntfy app, point it at `http://<box-lan-ip>:8481`
 as the default server, and subscribe to the `wallmonitor` topic. Warnings
 arrive prioritized (a predicted derate is *urgent* — it's the one you can
-act on in the moment by lowering the vehicle's charge current).
+act on in the moment by lowering the vehicle's charge current). The
+degradation watch sends at *high* priority only when its confidence
+interval clears zero; a verdict that merely crosses the tripwire is sent as
+a lead at *default* priority — the same alert on the dashboard, without the
+phone buzz.
 
 iOS caveat, stated plainly: Apple only delivers instant background pushes
 through its own push service, so a purely self-hosted server means the iOS

docs/thermal-model.md

@@ -36,6 +36,16 @@ So the fitter works per **charging segment**: it finds each segment's opening
 ramp wherever it occurs in the session and lets the quality gates decide
 what teaches the model. No configuration or "monitoring mode" is needed.
 
+One gate deserves spelling out: the steady-current window must span at
+least 1.8× the install's median τ (≈ 20 min at a typical 11 min τ). Shorter
+than that the plateau is never observed, and the exponential can explain
+the same samples with a lower rise and a faster τ — passing every other
+gate with a fine RMSE while under-reading the rise by several degrees. Such
+fits are not "noisy"; they are biased low, and a few of them in a drift
+baseline manufacture a degradation verdict. The span is judged against the
+install's τ, not the fit's own, so a truncated segment cannot vouch for
+itself.
+
 The unit of thermal analysis is the **load window** — the stretch where
 current actually flows.
 

tests/test_wallmonitor.py

@@ -560,11 +560,11 @@ async def test_thermal_fit_survives_ramp_and_midsession_derate(db):
         into = ts - start
         if into < 60:
             current = amps * into / 60.0  # ramp-up
-        elif into < 1200:
-            current = amps  # full rate for 20 min...
+        elif into < 1500:
+            current = amps  # full rate for 25 min...
         else:
             current = amps / 2  # ...then derated for hours (most samples)
-        temp = t_inf - (t_inf - temp0) * math.exp(-into / tau_s) if into < 1200 else 60.0
+        temp = t_inf - (t_inf - temp0) * math.exp(-into / tau_s) if into < 1500 else 60.0
         db.insert_vitals(ts, {
             "vehicle_connected": 1, "contactor_closed": 1, "vehicle_current_a": round(current, 2),
             "handle_temp_c": round(temp, 3), "pcba_temp_c": 55.0, "mcu_temp_c": 50.0,
@@ -579,6 +579,61 @@ async def test_thermal_fit_survives_ramp_and_midsession_derate(db):
     assert fits[0]["rise_ref_c"] is not None and abs(fits[0]["rise_ref_c"] - rise) < 3.0
 
 
+async def test_thermal_fit_rejects_window_short_against_tau(db):
+    # A steady window that ends before the plateau shows can't separate rise
+    # from tau: the fitter trades a lower rise for a faster tau and passes
+    # every other gate with a fine RMSE. Seen on a real install as 21 min
+    # charges fitting 8 C under the rest and dragging the drift baseline
+    # down. The gate judges span against the install's median tau, not the
+    # fit's own (the biased quantity), so a truncated segment can't vouch
+    # for itself.
+    now = time.time()
+    tau_s = 720.0
+    for i in range(4):  # establish the install's tau from full-length ramps
+        _seed_thermal_session(db, now - (6 - i) * 7200, ambient_c=25.0, tau_s=tau_s, charge_s=1500.0)
+    short_start = now - 7200
+    _seed_thermal_session(db, short_start, ambient_c=25.0, tau_s=tau_s, charge_s=720.0)  # 1.0 tau
+    fits = thermal.fit_sessions(db, now)
+    assert len(fits) == 4, "the four plateau-observing ramps fit"
+    assert all(abs(fit["start_ts"] - short_start) > 120 for fit in fits), "the 1 tau window does not"
+    # The boundary itself: a 1.8 tau window is the shortest that passes.
+    _seed_thermal_session(db, now - 3600, ambient_c=25.0, tau_s=tau_s,
+                          charge_s=thermal.MIN_SPAN_TAU * tau_s + 60.0)
+    assert len(thermal.fit_sessions(db, now)) == 5
+
+
+async def test_thermal_fit_first_fit_judged_against_default_tau(db):
+    # Fresh install, no history: the gate has no earlier fits to judge a
+    # window against, and the fit's own tau is exactly the quantity a
+    # truncated window biases low. Floored at DEFAULT_TAU_MIN, a 12 min
+    # first charge is rejected even though its own tau (6 min) would have
+    # let it vouch for itself at 1.8 tau = 10.8 min.
+    now = time.time()
+    _seed_thermal_session(db, now - 7200, ambient_c=25.0, tau_s=360.0, charge_s=720.0)
+    assert thermal.fit_sessions(db, now) == []
+    # A charge that clears 1.8 x DEFAULT_TAU_MIN is the first to teach the model.
+    _seed_thermal_session(db, now - 3600, ambient_c=25.0, tau_s=360.0,
+                          charge_s=thermal.MIN_SPAN_TAU * thermal.DEFAULT_TAU_MIN * 60.0 + 60.0)
+    fits = thermal.fit_sessions(db, now)
+    assert len(fits) == 1 and abs(fits[0]["tau_min"] - 6.0) < 1.0
+
+
+async def test_thermal_fit_slow_tau_install_still_fits(db):
+    # A heavier cable or enclosed handle can have a tau near 20 min. The
+    # steady-prefix window must scale with the install's tau: a fixed 30 min
+    # cap would leave every window under 1.8 tau and the install blind.
+    now = time.time()
+    tau_s, rise = 1200.0, 30.0
+    for i in range(4):
+        _seed_thermal_session(db, now - (5 - i) * 4 * 3600, ambient_c=22.0, tau_s=tau_s,
+                              rise_ref_c=rise, charge_s=3900.0)
+    fits = thermal.fit_sessions(db, now)
+    assert len(fits) == 4
+    for fit in fits:
+        assert abs(fit["tau_min"] - 20.0) < 2.0
+        assert fit["rise_ref_c"] is not None and abs(fit["rise_ref_c"] - rise) < 3.0
+
+
 async def test_thermal_fit_covers_late_charging_segments(db):
     # A session shaped like real overnight use: a plug-in burst too short to
     # fit, hours of connected idle, then distinct charging segments (vehicle
@@ -1009,6 +1064,46 @@ async def test_thermal_drift_poller_alert(db):
     assert any(event["kind"] == "thermal_drift" for event in events)
 
 
+async def _drift_notification(db, rises):
+    """Seed sessions with the given fitted rises, run the drift recheck, and
+    return the single (kind, body, detail) the poller tried to send."""
+    now = time.time()
+    for i, rise in enumerate(rises):
+        _seed_thermal_session(db, now - (len(rises) - i) * 7200, ambient_c=25.0, rise_ref_c=rise)
+    sent = []
+    async with aiohttp.ClientSession() as client:
+        poller = Poller(Config(host="127.0.0.1:1"), db, EventBus(), client)
+
+        async def capture(kind, title, body, detail):
+            sent.append((kind, body, detail))
+
+        poller._notify = capture
+        await poller.recheck_thermal_drift(now)
+    assert any(a["alert"] == thermal.DRIFT_ALERT for a in db.active_alerts())
+    assert len(sent) == 1
+    return sent[0]
+
+
+async def test_thermal_drift_confirmed_notifies_high_priority(db):
+    # Tight baseline, tight recent, big step: the interval clears zero, and
+    # that is the verdict worth interrupting a phone for.
+    kind, body, detail = await _drift_notification(db, [36.0, 36.5, 35.8, 36.2, 42.0, 41.5, 42.3])
+    assert detail["confident"] is True and kind == "thermal_drift"
+    assert "statistically confirmed" in body
+    assert Poller.NTFY_PRIORITY[kind] == "high"
+
+
+async def test_thermal_drift_lead_notifies_default_priority(db):
+    # Scattered baseline, modest step past the tripwire: drifting, but the
+    # interval straddles zero. Same alert and event, but with ~3.4 C
+    # session-to-session scatter the 2.5 C tripwire sits near one sigma, so
+    # an unconfirmed verdict is a lead for the dashboard, not a buzz.
+    kind, body, detail = await _drift_notification(db, [31.0, 38.0, 33.0, 38.0, 32.0, 37.0, 39.5, 38.0, 39.0])
+    assert detail["drifting"] is True and detail["confident"] is False
+    assert kind == "thermal_drift_lead" and "treat as a lead" in body
+    assert Poller.NTFY_PRIORITY[kind] == "default"
+
+
 async def test_baseline_anchor_reevaluates_drift_alert(db):
     # Moving the anchor must re-judge the active alert immediately — not at
     # the next session end, which can be a full plugged-in day away. Setting

wallmonitor/poller.py

@@ -528,8 +528,11 @@ async def recheck_thermal_drift(self, ts: float) -> None:
                     if drift["confident"]
                     else f"not yet confirmed at n={drift['baseline_n']}+{drift['recent_n']} — treat as a lead"
                 )
+                # Same alert and event either way; only the phone's interrupt
+                # level follows the statistics. A lead is a dashboard note,
+                # not a buzz — see the false-positive rate in detect_drift.
                 await self._notify(
-                    "thermal_drift",
+                    "thermal_drift" if drift["confident"] else "thermal_drift_lead",
                     "Heat rise climbing vs baseline",
                     f"Recent sessions run +{drift['recent_rise_c']:.1f} °C vs a +{drift['baseline_rise_c']:.1f} °C "
                     f"baseline at the same current (Δ {drift['delta_c']:.1f} °C, 95% CI {ci_lo:.1f}..{ci_hi:.1f}, "
@@ -627,13 +630,15 @@ def _alert_label(self, code: str) -> str:
     NTFY_PRIORITY = {
         "derate_warning": "urgent",  # actionable *now* — lower the amps
         "alert_raised": "high",
-        "thermal_drift": "high",
+        "thermal_drift": "high",  # confidence interval clears zero
+        "thermal_drift_lead": "default",  # tripwire only — a lead, not a conviction
         "poll_error": "default",
     }
     NTFY_TAGS = {
         "derate_warning": "zap,warning",
         "alert_raised": "rotating_light",
         "thermal_drift": "wrench",
+        "thermal_drift_lead": "wrench,mag",
         "poll_error": "electric_plug,x",
     }
 

wallmonitor/thermal.py

@@ -105,6 +105,20 @@ def ambient_from_idle_handle(handle_c: float) -> float:
 MAX_FIT_RMSE_C = 0.6
 TAU_RANGE_MIN = (3.0, 40.0)
 RISE_RANGE_C = (10.0, 80.0)
+# The steady-current window must be long against the install's time
+# constant, or rise and tau are not separately identifiable: a window that
+# ends before the plateau shows lets the fitter trade a lower rise for a
+# faster tau and pass every other gate with a fine RMSE. Judged against the
+# install's median tau (not this fit's own, which is exactly the biased
+# quantity), so a truncated segment can't vouch for itself — and floored at
+# DEFAULT_TAU_MIN, so on a fresh install the very first fit is judged
+# against a sane prior rather than its own possibly-truncated tau. At 1.8
+# tau the handle has covered ~83% of its rise. The steady-prefix window
+# scales with the same tau estimate (PREFIX_SPAN_TAU) so a slow-tau install
+# is not starved of fits by a fixed cap it can never clear.
+MIN_SPAN_TAU = 1.8
+PREFIX_SPAN_TAU = 2.5
+PREFIX_SPAN_MIN_S = 1800.0
 
 # Live-forecast gate: a steady-current window must hold this many samples
 # over this much time before its trajectory is projected.
@@ -185,8 +199,8 @@ def _fit_exponential(points: list[tuple[float, float]]) -> tuple[float, float, f
     return best
 
 
-def _steady_current_prefix(samples: list[dict]) -> list[dict]:
-    """The session's first steady-current run.
+def _steady_current_prefix(samples: list[dict], max_span_s: float = PREFIX_SPAN_MIN_S) -> list[dict]:
+    """The session's first steady-current run, capped at max_span_s.
 
     The reference current is the median of the first 10 minutes of charging,
     not of the whole session: a session that derates midway spends most of
@@ -218,7 +232,7 @@ def _steady_current_prefix(samples: list[dict]) -> list[dict]:
                 break  # the steady run ended (derate or charge stop)
             continue  # still ramping up to the plateau
         prefix.append(sample)
-        if sample["ts"] - prefix[0]["ts"] > 1800:  # first 30 min is where the ramp lives
+        if sample["ts"] - prefix[0]["ts"] > max_span_s:  # the ramp lives in the first few tau
             break
     return prefix
 
@@ -454,26 +468,47 @@ def fit_sessions(db: Database, now: float, lookback_days: float = 120.0) -> list
         segments = _segments(coarse)[:MAX_SEGMENTS_PER_SESSION]
         for idx, (seg_start, seg_end) in enumerate(segments):
             next_start = segments[idx + 1][0] if idx + 1 < len(segments) else sess["end_ts"] + 1
-            t_hi = min(sess["end_ts"], seg_start + 2700, next_start - 1)
-            samples = db.vitals_range(seg_start - 1, t_hi + 1, 5000)
-            prefix = _steady_current_prefix(samples)
-            seg = [
-                (sample["ts"], sample["handle_temp_c"])
-                for sample in prefix
-                if sample.get("handle_temp_c") is not None
-            ]
-            if len(seg) < MIN_SEGMENT_SAMPLES or seg[-1][0] - seg[0][0] < MIN_SEGMENT_S:
-                continue
-            if max(temp for _, temp in seg) - seg[0][1] < MIN_RISE_SEEN_C:
-                continue
-            fit = _fit_exponential(seg)
-            if fit is None:
+            # The install's tau so far (the default until fits exist) sizes
+            # the window: a fixed cap that suits an 11 min tau would leave
+            # a 20 min tau install unable to clear the identifiability gate.
+            tau_prior = median(fit["tau_min"] for fit in fits) if fits else DEFAULT_TAU_MIN
+            fit = None
+            for _pass in range(2):
+                fit = None
+                span_cap_s = max(PREFIX_SPAN_MIN_S, PREFIX_SPAN_TAU * tau_prior * 60.0)
+                t_hi = min(sess["end_ts"], seg_start + span_cap_s + 900, next_start - 1)
+                samples = db.vitals_range(seg_start - 1, t_hi + 1, 5000)
+                prefix = _steady_current_prefix(samples, span_cap_s)
+                seg = [
+                    (sample["ts"], sample["handle_temp_c"])
+                    for sample in prefix
+                    if sample.get("handle_temp_c") is not None
+                ]
+                if len(seg) < MIN_SEGMENT_SAMPLES or seg[-1][0] - seg[0][0] < MIN_SEGMENT_S:
+                    break
+                if max(temp for _, temp in seg) - seg[0][1] < MIN_RISE_SEEN_C:
+                    break
+                fit = _fit_exponential(seg)
+                if fit is None:
+                    break
+                # The window was sized from the tau prior. If this segment
+                # fits slower than that, the prior under-sized it: widen to
+                # the fitted tau and refit once, so a slow-tau install (or
+                # the first fit on a fresh one) isn't stuck behind a window
+                # it can never clear.
+                slower = fit[0] / 60.0 > tau_prior * 1.1
+                if not slower or prefix[-1]["ts"] - prefix[0]["ts"] < span_cap_s - 60:
+                    break  # window is tau-sized, or the run ended on its own
+                tau_prior = fit[0] / 60.0
+            if fit is None or len(seg) < MIN_SEGMENT_SAMPLES:
                 continue
             tau_s, t_inf, rmse = fit
             if rmse > MAX_FIT_RMSE_C or t_inf <= seg[0][1] + 3.0:
                 continue
             i_med = median(sample["vehicle_current_a"] for sample in prefix)
             tau_est = median([tau_s / 60.0] + [fit["tau_min"] for fit in fits])
+            if seg[-1][0] - seg[0][0] < MIN_SPAN_TAU * max(tau_est, DEFAULT_TAU_MIN) * 60.0:
+                continue  # plateau never observed: rise/tau not identifiable
             measured = _measured_ambient(db, seg_start - MEASURED_AMBIENT_WINDOW_S, seg_start + 60)
             ambient, ambient_source = measured if measured is not None else (None, None)
             if ambient is None: