SAFETY-128 round 2: find late readings from the batteries, not the meter alone

Review B1: the meter fit took repeating house loads (ovens, a heat pump)
for late readings on healthy 1-2 s sites, about 9 % of the recorded
house. B2: at a 5 s meter it never ran. M1: no hysteresis. M2: an
unbounded check.

lateness.Delay now fits how long each battery's reading takes to show
its own request, which the house does not move, from the 1 s reads
(read_inputs -> Answer.watch). The meter fit stays for speed at 1-3 s
meters, but counts only when an ask was slow to start and no battery
fit said on time. Late from 9.5 s after 2 fits, on time under 6.5 s,
dropped after 120 s without a fit. Fits are bounded (a few ms).

Readings on time: no verdict in any reviewer scenario (22 shapes at
1/2/5 s meters, 7 harm runs) nor in 2 days of the recorded house at
1, 2 and 5 s ticks; digests equal to 77fe31b. Late: found 36-68 s after
the start at 1-2 s, about 3 min at 5 s. The row gives the batteries'
delay, or no figure while only the meter has shown it.

Co-Authored-By: Claude <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Bdqr7PiaHcBnzPituu2yTC

goodwe_controller/DECISIONS/SAFETY-128-late-readings.md

@@ -2,72 +2,80 @@
 
 **Status: FIXED on release/1.0 (after 0.20.0).** Code: `app/lateness.py`,
 `control.compute(hold=, hold_w=)`, `peakguard.cap_floor()`,
-`Controller._readings_late()`. Tests: `test_safety128.py`
+`Controller._readings_late()`, `Controller.read_inputs()` (feeds the fits).
-(`tools/safety128_rig.py`, TEST-23's plant).
+Tests: `test_safety128.py` (`tools/safety128_rig.py`, TEST-23's plant).
 
 **Cause** (SAFETY-127 report). Home Assistant can deliver readings late but
 still changing. Every gate judges arrival (SAFETY-01) or a value that stops
-changing (SAFETY-18, SAFETY-37), so a late reading passes all of them. The loop
+changing (SAFETY-18, SAFETY-37), so a late reading passes them all. The loop
-then corrects, each cycle, an error its last command has already fixed: a loop
+then corrects, each cycle, an error its last command has already fixed: a
-with a dead time far longer than its cycle, which swings and grows. The peak
+loop with a dead time far longer than its cycle, which swings and grows. The
-guard projects the quarter from the same late figures and pushes to the
+peak guard projects the quarter from the same late figures and pushes to the
 inverter limit in its last seconds (OPT-34 rig F-2).
 
 **Rule.**
-1. *Find late readings.* `Delay` fits, every 4 s over the last 40 s of new
+1. *Find late readings* (`Answer.watch()`, every 2 s, from the 1 s reads).
-   meter values, how long after a power request the meter shows it: the meter
+   - **Batteries** (`Delay`): fit one `d` so that each battery's reading minus
-   is the house less the batteries, so `meter(t) + request(t - d)` is flattest
+     its own request at `t - d` (averaged over 2 s for the battery's lag and
-   at the right `d`. Over `DELAY_S` the readings are late (`late_seen`). The
+     the read) is flattest. The house does not move that reading, so a
-   fit counts only if the requests explain half the meter's swing; a late `d`
+     repeating load cannot line up with it (review round 1, B1). Needs 5 new
-   only if it fits 30 % better than any fresh one and better than a slow
+     readings in 60 s and the request and the reading each to move 300 W.
-   battery answering at once (10 or 20 s lag, the SAFETY-85/94 rigs). A
+     Late from 9.5 s, after 2 late fits in a row. A late fit must explain 75 %
-   silent meter adds no values (SAFETY-99/125). A battery that gives nothing
+     and beat every fit up to 6.5 s by 2x; the shortest delay that fits about
-   under its ask, or refuses, spoils the fit and is left out for 40 s. With no
+     as well (within 2x, or within 2 500 W^2) is taken instead, since a swing's
-   fit for 120 s (a steady site shows nothing) the verdict is dropped.
+     period fits again. A slow battery (3-20 s lag) answering at once gives no
-   Nothing below acts unless `late_seen`, so with readings on time control is
+     late fit (checked in the test). No late battery fit counts for 60 s
-   unchanged.
+     after a battery gave nothing under its ask: an emptied battery lined its
-2. *Hold.* While `late_seen`, a request more than `ANSWER_W` from the battery
+     old movement up as 85 s late.
-   reading is pending until that reading comes within `ANSWER_W` of it, or half
+   - **Meter** (`MeterFit`, round 1's fit): `meter(t) + request(t - d)` is the
-   way. Meanwhile the loop may not lead the reading by more than `ANSWER_W`
+     house alone at the right `d`. Quick, but confounded by the house. Needs
-   that way (`compute(hold=, hold_w=)`; it may come back), and the peak
+     16 new values in 60 s, so only a meter of about 3.5 s or faster: at 5 s
-   guard's floor is capped at the request that stands (`cap_floor()`). The
+     it gave false delays. Late over 8 s.
-   loop is not held while a battery gives nothing under its ask or refuses:
+   - A meter fit says late only with evidence from the batteries' side too:
-   SAFETY-52/94 need it to go on asking.
+     a request more than 100 W from the battery reading took 3 s or more to
-3. *Late.* Pending for `LATE_S` with no reading moving `ANSWER_W` towards it is
+     move a reading towards it (`SLOW_START_S`; on time they start within
-   the late verdict: control holds, then after `stale_input_s` goes to 0 W
+     0-1 s in the recorded house at 1 and 2 s ticks), and no battery fit in
-   ("readings late"), as for a missing input. It ends when a reading starts
+     the last 60 s said on time (under 9.5 s). An ask during which a battery
-   towards the request.
+     gave nothing under its ask is no slow evidence. A repeating load fools
-4. *Seen.* While `late_seen` an amber row says "Home Assistant readings arrive
+     the meter fit but not these. No meter verdict in the first 150 s.
-   about N seconds late, so control holds the batteries." (NL too), and
+   - On time again on a battery fit under 6.5 s, or a meter fit within 8 s
-   "every inverter is at 0 W" past the hold. One WARNING (`READINGS_LATE`)
+     unless a battery fit said late (hysteresis). With no fit for 120 s the
-   and one action-log line per episode, and one when it ends.
+     verdict is dropped.
+2. *Hold.* While late, a request more than 100 W from the battery reading is
+   pending until that reading comes within 100 W of it, or half way. Meanwhile
+   the loop may not lead the reading by more than 100 W that way, and the peak
+   guard's floor is capped at the request that stands. The loop is not held
+   while a battery gives nothing under its ask or refuses (SAFETY-52/94).
+3. *Late verdict.* Pending for `LATE_S` (23 s) with no reading moving 100 W
+   towards it: control holds, then after `stale_input_s` goes to 0 W
+   ("readings late"), as for a missing input.
+4. *Seen.* While late an amber row says "Home Assistant readings show requests
+   about N seconds late, so control holds the batteries." (NL too). N is the
+   batteries' fitted delay: Home Assistant's lag plus the battery's own
+   answer. While only the meter has shown it the row gives no figure. One
+   WARNING (`READINGS_LATE`) per episode and one line when it ends.
 
-**Constants** (`app/lateness.py`).
+**With readings on time** no verdict was reached in any reviewer scenario
-
+(owner-shaped house, ovens 20/40 to 60/60 s, clouds, a 300 W/s battery, 6-10 s
-| | Value | Why |
+dead times, HA jitter 0-4 s, an EV following the meter, capped batteries, HA
-|---|---|---|
+restarts, meter pauses; 1, 2 and 5 s meters) nor in 2 days of the recorded
-| `ANSWER_W` | 100 W | `control.PROBE_ANSWER_W`: a reading this far towards the ask has answered it (SAFETY-52's probe). |
+house at 1, 2 and 5 s ticks, so the commands are the same as without the gate.
-| `DELAY_S` | 8 s | `freshness.PLANT_SETTLE_S` 6 s + one `LOOP_TICK_S` 2 s: a fresh meter shows a request within the plant's settle (SAFETY-75's 6 s dead time) and one meter tick. The fit reads 0 s at 1/2/5 s meters, 5-7 s with a 6 s dead time, 9 s at 10 s late. |
+The fits run all the time; the pull request has the counts.
-| `LATE_S` | 23 s | 2 x `PLANT_SETTLE_S` + `PUBLISH_PERIOD_S` 10 s + `READ_PERIOD_S` 1 s: SAFETY-75's 6 s dead time, the 6 s settle, one publish and one read, the slowest healthy battery answer. |
-| `FIT_S`, `FIT_EVERY_S` | 40 s, 4 s | Short enough to find 10 s and 30 s late within 50-60 s of the start (the swing has not grown yet); one 2 s cycle in two. |
-| `MAX_DELAY_S` | 90 s | F-2 measured 17-78 s. |
-| `FIT_SHARE`, `FIT_BETTER` | 0.5, 0.7 | Below these a battery that stops following (a refusal at N=2) was read as 23 s late. |
-| `SLOW_TAUS_S` | 10 s, 20 s | The SAFETY-85/94 lagging battery: without this check 16 of 36 of its reversals read as 11-56 s late. |
-| `FIT_GONE_S` | 120 s | The verdict outlives the lag by at most `FIT_S` + this when the site goes steady. |
 
 **What it costs.**
-- Late readings are found only after the requests have moved 100 W and the
+- Found 36-68 s after a lag starts at 1-2 s meters, about 3 minutes at a 5 s
-  meter has shown it late: 48-68 s after the lag starts in the tests. Until
+  meter (battery fit only; the meter fit needs a faster meter). Until then control is as before. At 5 s and 30 s
-  then control is as before.
+  late the export is cut to about a third, not removed.
-- The row can stay up to 160 s after the readings are on time again, while
+- A battery whose own answer takes about 10 s is close to the threshold: an
-  the site is steady.
+  8 s dead time reads 7, 10 s reads 9, HA 10 s late reads 10.
-- A battery that cannot follow while readings are late (at a limit, not
+- The row can stay up to 180 s after the readings are on time on a steady
-  reading 0 W) is held, then set to 0 W after 38 s, until it moves.
+  site. While only the meter has shown the delay it gives no figure.
-- The fit costs about 93 x 20 sums every 4 s.
+- A battery that cannot follow while the readings are late (at a limit, not
-- With readings late, a large step is followed in steps of the lag: the loop
+  reading 0 W) is held, then set to 0 W after 38 s.
-  waits for each request to show before the next.
+- The fits cost a few ms every 2 s (x86, measured after 3 idle days); the
+  history they read is bounded (60 s of readings, 150 s of requests).
 
-**Left out.** The round trip setpoint -> Home Assistant: the fitted delay gives
+**Left out.** The setpoint -> Home Assistant round trip: the fitted delay gives
-the same figure from readings the add-on already has. The arbitrage charge's
+the figure from readings the add-on already has. The hold's step-and-pull-back
-own sizing on late readings (`charge_room_w()`) is not capped here; the test
+at extreme lag and the arbitrage charge's sizing on late readings
-shows it swinging while the peak guard holds (see the pull request).
+(`charge_room_w()`) are follow-up cards.

goodwe_controller/app/lateness.py

@@ -3,122 +3,221 @@
 When Home Assistant delivers readings late but still changing, every gate that
 judges arrival (SAFETY-01) or an unchanged value (SAFETY-18, SAFETY-37) passes
 them, and the loop corrects again, each cycle, an error its last command has
-already fixed. The readings themselves show it: after a command the meter moves
+already fixed. The readings show it: they answer each power request late.
-the other way and the battery reading follows (r = 0.998, freshness.py), and
-late readings do both late.
 
-`Answer` watches the site command on the wire, the meter and the site battery
+`Answer.watch()` (each 1 s read) fits two delays every FIT_EVERY_S:
-reading (less the batteries' idle reading, SAFETY-97):
+- `Delay`: how long after its request each battery's reading shows it. The
-- `Delay` fits how long after a command the meter shows it. Over DELAY_S the
+  house does not move that reading, so a repeating load cannot fool it. It
-  readings run late (`late_seen`); within it they do not. A battery that does
+  needs the request to move FIT_MOVE_W; one reading per 10 s makes it slow.
-  not follow gives no fit, so no evidence either way;
+- `MeterFit`: how long after the site request the meter shows it. Quick at a
+  1-3 s meter, none at a slower one, and a house load that repeats can fool it.
+The meter counts only when an ask was slow to start and no battery fit said
+"on time"; with hysteresis (ON_S / OFF_S) the result is `late_seen`. Then (`observe()`, each cycle):
 - a new command more than ANSWER_W from the battery reading is PENDING until
-  that reading comes within ANSWER_W of it, or half way. The first move of
+  that reading comes within ANSWER_W of it, or half way;
-  either reading towards it (ANSWER_W) is its START;
+- while pending the loop may not lead the battery reading by more than
-- while `late_seen` and pending, the loop may not lead the battery reading by
+  ANSWER_W that way (`hold`, `hold_w`; not while `exempt`: a battery that
-  more than ANSWER_W that way (`hold`, `hold_w`; not while `exempt`: a
+  gives nothing under its ask or refuses is SAFETY-52/94's case) and the peak
-  battery that gives nothing under its ask or refuses is SAFETY-52/94's case)
+  guard may not push past what stands (`peak_cap`). Pending for LATE_S with
-  and the peak guard may not push past what stands (`peak_cap`). Pending for LATE_S with no start is LATE: control
+  no reading moving towards it is LATE: control holds, then 0 W after
-  holds, then 0 W after `stale_input_s`, as for a missing input.
+  `stale_input_s`, as for a missing input.
 
-With readings on time nothing here acts. DECISIONS/SAFETY-128-late-readings.md.
+Not `late_seen`, nothing here acts. DECISIONS/SAFETY-128-late-readings.md.
 """
 
 import bisect
-import math
 from collections import deque
 
 from . import freshness
-from .control import LOOP_TICK_S, PROBE_ANSWER_W
+from .control import PROBE_ANSWER_W
 
 # Within this of the command the reading has answered it (a probe's answer).
 ANSWER_W = PROBE_ANSWER_W                                         # 100
-# A fresh meter shows a command within the plant's settle (SAFETY-75's 6 s dead
-# time) and one meter tick; later than that the readings run late.
-DELAY_S = freshness.PLANT_SETTLE_S + LOOP_TICK_S                   # 8.0
 # The slowest healthy battery answer: SAFETY-75's 6 s dead time, the 6 s plant
 # settle, one 10 s publish and one 1 s read.
 LATE_S = (2 * freshness.PLANT_SETTLE_S + freshness.PUBLISH_PERIOD_S
           + freshness.READ_PERIOD_S)                              # 23.0
 
 
-# The meter's delay is fitted over this much history, up to this delay.
+# Each battery reading's delay is fitted over this much history, up to this.
-FIT_S = 40.0
+FIT_S = 60.0
 MAX_DELAY_S = 90
-FIT_EVERY_S = 4.0
+FIT_EVERY_S = 2.0
-# A slow battery answers at once but takes this long (SAFETY-85/94 rigs).
+# A fit needs this many new readings (all batteries) and this much movement
-SLOW_TAUS_S = (10.0, 20.0)
+# of the reading and the request.
+FIT_ROWS = 5
+FIT_MOVE_W = 3 * ANSWER_W                                         # 300
+# Late from ON_S, on time again under OFF_S (hysteresis).
+ON_S = 9.5
+OFF_S = 6.5
+# ...after this many late fits in a row.
+ON_FITS = 2
 # With no fit for this long the evidence is dropped (a steady site shows none).
 FIT_GONE_S = 120.0
-# A fit counts only if the commands explain this share of the meter's swing.
+# A fit counts only if the commands explain this share of the reading's swing.
 FIT_SHARE = 0.5
+# A late delay must explain this share.
+FIT_SHARE_LATE = 0.75
 # ...and a late delay only if it fits this much better than any fresh one.
-FIT_BETTER = 0.7
+FIT_BETTER = 0.5
+# A shorter delay within this factor of the best fit is taken instead.
+FIT_NEAR = 2.0
+# ...or within this of it (W^2): two fits this good are both exact.
+FIT_FLOOR_W2 = (ANSWER_W / 2) ** 2                               # 2 500
 
 
-class Delay:
+def per_second(cmds, first, last):
-    """How long after a command the meter shows it, fitted from history.
+    """The request standing at each whole second from `first` to `last`."""
+    ts = [t for t, _c in cmds]
+    return [cmds[max(0, bisect.bisect_right(ts, k) - 1)][1]
+            for k in range(first, last + 1)]
 
-    The meter is the house less the batteries, and the batteries follow the
+
-    commands, so `grid(t) + command(t - d)` is the house alone at the right
+def pooled(sets, d):
-    `d`: the `d` that leaves it flattest is the delay. Only new meter values
+    """Pooled variance over [(rows, c, first, sign)] of value + sign *
-    count, so a silent meter (SAFETY-99) is not a late one. None while the
+    c[t - d], each set about its own mean."""
-    commands have not moved enough to tell, or explain too little of the meter
+    tot, n = 0.0, 0
-    (a battery that does not follow); not late if a slow battery that answers
+    for rows, c, first, sign in sets:
-    at once (SLOW_TAUS_S) fits as well."""
+        e = [v + sign * c[int(t) - d - first] for t, v in rows]
+        m = sum(e) / len(e)
+        tot += sum((x - m) ** 2 for x in e)
+        n += len(e)
+    return tot / n
+
+
+class UnitFit:
+    """One battery's power requests and readings, as they were seen."""
 
     def __init__(self):
-        self.cmds = deque()      # (t, command) when it changed
+        self.cmds = deque()      # (t, request) when it changed
-        self.grid = deque()      # (t, meter)
+        self.reads = deque()     # (t, reading) when it changed
 
-    def add(self, now, cmd, grid):
+    def add(self, now, cmd, reading):
-        if not self.cmds or self.cmds[-1][1] != cmd:
+        if cmd is not None and (not self.cmds or self.cmds[-1][1] != cmd):
-            self.cmds.append((now, cmd))
+            self.cmds.append((now, float(cmd)))
-        if grid is not None and (not self.grid or self.grid[-1][1] != grid):
+        if reading is not None and (not self.reads
-            self.grid.append((now, grid))         # new meter values only
+                                    or self.reads[-1][1] != reading):
-        while self.grid and self.grid[0][0] < now - FIT_S:
+            self.reads.append((now, float(reading)))
-            self.grid.popleft()
+        while self.reads and self.reads[0][0] < now - FIT_S:
+            self.reads.popleft()
         while (len(self.cmds) > 1
                and self.cmds[1][0] <= now - FIT_S - MAX_DELAY_S):
             self.cmds.popleft()
 
-    def estimate(self):
+    def series(self):
-        if len(self.grid) < 10 or len(self.cmds) < 2:
+        """(rows, request each second, first second) for the fit, or None:
-            return None
+        too few new readings, or a reading or a request that has not moved
-        ts = [t for t, _c in self.cmds]
+        FIT_MOVE_W. The request is averaged over the 2 s before each second:
-        t0 = self.cmds[0][0]
+        the battery's own lag and the read."""
-        rows = [(t, g) for t, g in self.grid if t >= t0 + MAX_DELAY_S]
+        rows = [r for r in self.reads
-        if len(rows) < 10:
+                if self.cmds and r[0] >= self.cmds[0][0] + MAX_DELAY_S]
+        vals = [v for _t, v in rows]
+        if len(rows) < 3 or max(vals) - min(vals) < FIT_MOVE_W:
             return None
         cs = [c for _t, c in self.cmds]
-        if max(cs) - min(cs) < ANSWER_W:
+        if max(cs) - min(cs) < FIT_MOVE_W:
             return None
-        g_mean = sum(g for _t, g in rows) / len(rows)
+        first, last = int(rows[0][0]) - MAX_DELAY_S - 2, int(rows[-1][0])
-        g_var = sum((g - g_mean) ** 2 for _t, g in rows) / len(rows)
+        c = per_second(self.cmds, first, last)
+        avg = [c[0], c[0]] + [(c[k - 2] + c[k - 1] + c[k]) / 3
+                              for k in range(2, len(c))]
+        return rows, avg, first
 
-        def var(f, d):
+
-            e = [g + f(t - d) for t, g in rows]
+class Delay:
-            m = sum(e) / len(e)
+    """How long after a power request the batteries' readings show it.
-            return sum((x - m) ** 2 for x in e) / len(e)
+
-        step = lambda t: cs[bisect.bisect_right(ts, t) - 1]     # noqa: E731
+    A reading follows its battery's own request and the house does not move
-        pure = [var(step, d) for d in range(0, MAX_DELAY_S + 1)]
+    it, so `reading(t) - request(t - d)` is flattest at the right `d`. Each new
-        best = min(range(len(pure)), key=lambda d: (pure[d], d))
+    reading counts once, at the second it was seen, so the 10 s publish adds no
-        if g_var <= 0 or pure[best] > (1 - FIT_SHARE) * g_var:
+    delay. One `d` for every battery (Home Assistant delays them all). None
+    with too few readings, or when the requests explain too little of them (a
+    battery that does not follow); not late unless it fits clearly better than
+    any fresh delay, and then the shortest delay that fits about as well."""
+
+    def __init__(self):
+        self.units = {}
+
+    def add(self, now, unit, cmd, reading):
+        self.units.setdefault(unit, UnitFit()).add(now, cmd, reading)
+
+    def estimate(self):
+        ser = [x for x in (u.series() for u in self.units.values()) if x]
+        if sum(len(rows) for rows, _c, _f in ser) < FIT_ROWS:
             return None
-        if best <= DELAY_S:
+        sets = [(rows, c, first, -1.0) for rows, c, first in ser]
+        r_var = pooled([(rows, [0.0] * len(c), first, -1.0)
+                        for rows, c, first in ser], 0)
+        pure = [pooled(sets, d) for d in range(0, MAX_DELAY_S + 1)]
+        # The shortest that fits about as well: a swing's period fits again.
+        best = next(d for d, v in enumerate(pure)
+                    if v <= FIT_NEAR * min(pure) + FIT_FLOOR_W2)
+        if r_var <= 0 or pure[best] > (1 - FIT_SHARE) * r_var:
+            return None
+        if best < ON_S:
             return float(best)
-        # Late only if it fits clearly better than any fresh delay, and better
+        if (pure[best] > (1 - FIT_SHARE_LATE) * r_var
-        # than a slow battery answering at once (SAFETY-85/94's 10-20 s lag).
+                or pure[best] > FIT_BETTER * min(pure[:int(OFF_S) + 1])):
-        if pure[best] > FIT_BETTER * min(pure[:int(DELAY_S) + 1]):
+            return None
+        return float(best)
+
+
+# The meter's delay: fitted over this much history, from this many new
+# values (a meter of about 3.5 s or faster: a slower one gives too few to
+# fit without false delays); late over M_LATE_S (the plant's 6 s settle and one 2 s meter tick),
+# but only when the batteries' own readings have not just shown them on time.
+M_FIT_S = 60.0
+M_ROWS = 16
+M_LATE_S = freshness.PLANT_SETTLE_S + 2.0                          # 8.0
+M_SHARE, M_BETTER = 0.4, 0.7
+# A meter fit counts only after an ask that took this long to start moving a
+# reading: on time they start within 1-2 s (the recorded house, at 1-5 s).
+SLOW_START_S = 3.0
+
+
+class MeterFit:
+    """How long after a site request the meter shows it (round 1's fit).
+
+    `meter(t) + request(t - d)` is the house alone at the right `d`. Quick,
+    as the meter reads every 1-5 s, but a house load that repeats can line
+    up with an earlier request, so it only counts while the batteries' own
+    fit is not on time (`Answer.watch`)."""
+
+    def __init__(self):
+        self.cmds = deque()      # (t, request) when it changed
+        self.grid = deque()      # (t, meter) new values only
+
+    def add(self, now, cmd, grid):
+        if cmd is not None and (not self.cmds or self.cmds[-1][1] != cmd):
+            self.cmds.append((now, float(cmd)))
+        if grid is not None and (not self.grid or self.grid[-1][1] != grid):
+            self.grid.append((now, float(grid)))
+        while self.grid and self.grid[0][0] < now - M_FIT_S:
+            self.grid.popleft()
+        while (len(self.cmds) > 1
+               and self.cmds[1][0] <= now - M_FIT_S - MAX_DELAY_S):
+            self.cmds.popleft()
+
+    def estimate(self):
+        if len(self.cmds) < 2:
+            return None
+        rows = [(t, g) for t, g in self.grid
+                if t >= self.cmds[0][0] + MAX_DELAY_S]
+        cs = [c for _t, c in self.cmds]
+        if len(rows) < M_ROWS or max(cs) - min(cs) < ANSWER_W:
+            return None
+        first, last = int(rows[0][0]) - MAX_DELAY_S, int(rows[-1][0])
+        c = per_second(self.cmds, first, last)
+        g_var = pooled([(rows, [0.0] * len(c), first, 1.0)], 0)
+        pure = [pooled([(rows, c, first, 1.0)], d)
+                for d in range(0, MAX_DELAY_S + 1)]
+        best = min(range(len(pure)), key=lambda d: (pure[d], d))
+        if g_var <= 0 or pure[best] > (1 - M_SHARE) * g_var:
+            return None
+        if best <= M_LATE_S:
+            return float(best)
+        if pure[best] > M_BETTER * min(pure[:int(M_LATE_S) + 1]):
             return None
-        first, last = int(t0), int(rows[-1][0])
-        for tau in SLOW_TAUS_S:
-            k, f, slow = 1 - math.exp(-1.0 / tau), None, []
-            for sec in range(first, last + 1):        # the battery, each second
-                f = step(sec) if f is None else f + (step(sec) - f) * k
-                slow.append(f)
-            if var(lambda t: slow[min(int(t), last) - first], 0) <= pure[best]:
-                return None
         return float(best)
 
 
@@ -131,13 +230,22 @@ class Answer:
         self.from_grid = None    # the meter when it was asked
         self.since = None        # when it was asked
         self.started = False     # a reading has moved towards it
-        self.late_seen = False   # the last fitted delay was over DELAY_S
+        self.late_seen = False   # the batteries' readings run late
         self.late_at = None      # when the late verdict started
         self.exempt = False      # a battery that cannot follow explains it
         self._was = None         # (command, reading, meter) at the last call
         self.delay = Delay()
-        self.delay_s = None      # the meter's last fitted delay, s
+        self.meter = MeterFit()
-        self._fit_at = self._fit_ok = self._exempt_at = None
+        self._batt_fit = None     # (when, d): the batteries' last fit
+        self._b_run = 0          # late battery fits in a row
+        self._start_age = None   # (when, how long the last ask took to start)
+        self._exempt_at = None   # when a battery last gave nothing under its ask
+        self._ask_exempt = False
+        self._start = None
+        self.delay_s = None      # the last fitted delay (either fit), s
+        self.shown_s = None      # the batteries' last late delay, for the page
+        self.fits = 0            # fits made (instrumentation)
+        self._fit_at = self._fit_ok = None
 
     @property
     def active(self) -> bool:
@@ -165,6 +273,56 @@ class Answer:
         """Seconds the late verdict has stood, or None."""
         return None if self.late_at is None else max(0.0, now - self.late_at)
 
+    def watch(self, now, units, standing_w=None, grid_w=None):
+        """Each read (1 s): `units` = [(name, request on its wire, its battery
+        reading)], the site request and the meter. Every FIT_EVERY_S, fit both
+        delays. Late on ON_FITS battery fits from ON_S, or on a meter fit
+        over M_LATE_S when an ask was slow to start (SLOW_START_S) and no
+        battery fit within FIT_S said on time: a house load that repeats can
+        fool the meter fit, not these. On time again on a battery fit under
+        OFF_S, or a meter fit within M_LATE_S unless a battery fit said late.
+        """
+        if self._start is None:
+            self._start = now
+        for name, cmd, reading in units:
+            self.delay.add(now, name, cmd, reading)
+        self.meter.add(now, standing_w, grid_w)
+        if self._fit_at is not None and now - self._fit_at < FIT_EVERY_S:
+            return
+        self._fit_at = now
+        b = self.delay.estimate()
+        if (b or 0.0) >= ON_S and self._exempt_at is not None \
+                and now - self._exempt_at <= FIT_S:
+            b = None          # a battery that stopped following: no late fit
+        m = self.meter.estimate()
+        if b is not None:
+            self._batt_fit = (now, b)
+            self.shown_s = b if b >= ON_S else None
+        recent = (self._batt_fit is not None
+                  and now - self._batt_fit[0] <= FIT_S)
+        b_fresh = recent and self._batt_fit[1] < ON_S
+        b_late = recent and self._batt_fit[1] >= ON_S
+
+        self._b_run = self._b_run + 1 if (b or 0.0) >= ON_S else 0
+        d = b if b is not None else m
+        if d is None:
+            if self._fit_ok is None or now - self._fit_ok > FIT_GONE_S:
+                self.late_seen = False            # nothing to tell it by
+            return
+        self.fits += 1
+        self.delay_s, self._fit_ok = d, now
+        warm = now - self._start >= MAX_DELAY_S + FIT_S
+        m_late = (m or 0.0) > M_LATE_S
+        slow = (self._start_age is not None
+                and now - self._start_age[0] <= M_FIT_S
+                and self._start_age[1] >= SLOW_START_S)
+        if self._b_run >= ON_FITS or (
+                m_late and warm and slow and not b_fresh):
+            self.late_seen = True
+        elif (b is not None and b < OFF_S) or (
+                m is not None and m <= M_LATE_S and not b_late):
+            self.late_seen = False
+
     def observe(self, now, standing_w, batt_w, *, grid_w=None, exempt=False,
                 idle_w=0.0):
         """`standing_w` the site command on the wire, `batt_w` the site
@@ -179,22 +337,15 @@ class Answer:
         if self.exempt:
             self._exempt_at = now
         was, self._was = self._was, (cmd, r, g)
-        self.delay.add(now, cmd, g)
-        if self._fit_at is None or now - self._fit_at >= FIT_EVERY_S:
-            self._fit_at = now
-            d = self.delay.estimate()
-            if self._exempt_at is not None and now - self._exempt_at < FIT_S:
-                d = None              # a battery that does not follow spoils it
-            if d is not None:
-                self.delay_s, self.late_seen, self._fit_ok = d, d > DELAY_S, now
-            elif self._fit_ok is None or now - self._fit_ok > FIT_GONE_S:
-                self.late_seen = False            # nothing to tell it by
         if self.ask is None:
             if was is None or cmd == was[0] or abs(cmd - was[1]) < ANSWER_W:
                 return
             self.ask, self.from_w, self.from_grid = cmd, was[1], was[2]
             self.since, self.started = now, False
+            self._ask_exempt = False
         self.ask = cmd
+        # A battery that gives nothing under its ask explains a slow start.
+        self._ask_exempt = self._ask_exempt or self.exempt
         age = now - self.since
         gap = self.ask - self.from_w
         way = 1.0 if gap > 0 else -1.0
@@ -203,6 +354,10 @@ class Answer:
                    else (self.from_grid - g) * way)
         if not self.started and max(batt_to, grid_to) >= ANSWER_W:
             self.started, self.late_at = True, None
+            self._start_age = (now, 0.0 if self._ask_exempt else age)
+        elif (not self.started and age >= SLOW_START_S
+              and not self._ask_exempt):
+            self._start_age = (now, age)
         if (self.ask - r) * way < ANSWER_W or batt_to >= abs(gap) / 2:
             self._clear()                         # answered
         elif abs(gap) < ANSWER_W:

goodwe_controller/app/main.py

@@ -3137,11 +3137,11 @@ class Controller:
         if a.late_seen != self._late_logged:
             self._late_logged = a.late_seen
             if a.late_seen:
-                self.log_event(f"readings late ({a.delay_s:.0f} s) - holding",
+                self.log_event("readings late - holding",
                                hint=messages.hint("readings_late"))
                 _LOG.warning("Home Assistant readings arrive late, so control "
-                             "holds the battery: READINGS_LATE.")
+                             "holds the batteries: READINGS_LATE.")
-                _LOG.debug("READINGS_LATE: the meter shows a power request "
+                _LOG.debug("READINGS_LATE: the readings show a power request "
                            "%.0f s after it is sent.", a.delay_s)
             else:
                 self.log_event("readings on time", hint=messages.hint(
@@ -3760,6 +3760,10 @@ class Controller:
                 else:
                     _LOG.debug("Meter readings are back.")
 
+        self.answer.watch(now_mono, [(inv.name, inv.standing_w(), inv.batt)
+                                     for inv in self.inverters],  # SAFETY-128
+                          sum(inv.standing_w() for inv in self.inverters),
+                          self.grid)
         if self.inputs_ok:
             if self.inputs_bad_since is not None:
                 self.log_event("inputs recovered",
@@ -7384,7 +7388,7 @@ class Controller:
         if self.answer.late_seen:                             # SAFETY-128
             out.append({"ok": False, "warn": True, "id": "late",
                         "text": messages.row_readings_late(
-                            self.answer.delay_s, self._late_zero)})
+                            self.answer.shown_s, self._late_zero)})
 
         # SAFETY-03. Red while an over-rating clamp is engaged (amber for the
         # integrator bound, SAFETY-40), and the engagement COUNT is shown even

goodwe_controller/app/messages.py

@@ -645,16 +645,28 @@ def row_controller_stalled(age_s, veto_s) -> str:
 
 
 def row_readings_late(delay_s, zero=False) -> str:
-    """SAFETY-128: Home Assistant readings arrive late; past the hold, 0 W."""
+    """SAFETY-128: readings show the power request late; `delay_s` the
+    batteries' fitted delay (their own answer included), or None when only
+    the meter has shown it; past the hold, 0 W."""
+    if delay_s is None:
+        if zero:
+            return Bi("Home Assistant readings show requests late, so every "
+                      "inverter is at 0 W.",
+                      "Metingen uit Home Assistant tonen vermogensvraag te "
+                      "laat, omvormers op 0 W.")
+        return Bi("Home Assistant readings show requests late, so control "
+                  "holds the batteries.",
+                  "Metingen uit Home Assistant tonen vermogensvraag te laat, "
+                  "dus de batterijen wachten.")
     if zero:
-        return Bi(f"Home Assistant readings arrive about {seconds(delay_s)} "
+        return Bi(f"Home Assistant readings show requests {seconds(delay_s)} "
                   f"late, so every inverter is at 0 W.",
-                  f"Metingen uit Home Assistant komen zo'n "
+                  f"Metingen uit Home Assistant tonen vermogensvraag "
                   f"{seconden(delay_s)} te laat, omvormers op 0 W.")
-    return Bi(f"Home Assistant readings arrive about {seconds(delay_s)} late, "
+    return Bi(f"Home Assistant readings show requests about {seconds(delay_s)} "
-              f"so control holds the batteries.",
+              f"late, so control holds the batteries.",
-              f"Metingen uit Home Assistant komen zo'n {seconden(delay_s)} te "
+              f"Metingen uit Home Assistant tonen vermogensvraag zo'n "
-              f"laat, dus de batterijen wachten.")
+              f"{seconden(delay_s)} te laat, dus de batterijen wachten.")
 
 
 def row_meter_waiting(limit_s) -> str:
@@ -1511,6 +1523,8 @@ def status_samples() -> list:
         ("row_meter_waiting", AMBER, row_meter_waiting(30)),
         ("row_readings_late", AMBER, row_readings_late(30)),
         ("row_readings_late", AMBER, row_readings_late(30, True)),
+        ("row_readings_late", AMBER, row_readings_late(None)),
+        ("row_readings_late", AMBER, row_readings_late(None, True)),
         ("row_meter", GREEN, row_meter(-350, 2, 3, 340)),
         ("row_meter_frozen", RED, row_meter_frozen()),
         *(("row_freeze_off", AMBER, row_freeze_off(k)) for k in FREEZE_OFF),

goodwe_controller/test_safety128.py

@@ -95,37 +95,55 @@ def swing(rows, t0, t1):
 print("1. the give run with readings late")
 ON = 300.0
 CASES = ((10.0, 150.0), (30.0, 300.0))
-for units in (("a", "b"), ("a",)):
+for p1, units in ((2.0, ("a", "b")), (2.0, ("a",)), (1.0, ("a", "b")),
+                  (5.0, ("a", "b"))):
     house = 1800.0 if len(units) == 2 else 1500.0
     for lag, span in CASES:
         off = ON + span
+        seen = []
         new = R.Rig(house=house, lag_s=lag, lag_on=ON, lag_off=off,
-                    units=units)
+                    units=units, p1_s=p1)
-        new.run(off + 120.0)
+        new.run(off + 120.0, lambda rig, c, t: seen.append(
+            (t, c.answer.late_seen)) if c.answer.late_seen else None)
         old = R.Rig(house=house, lag_s=lag, lag_on=ON, lag_off=off,
-                    units=units)
+                    units=units, p1_s=p1)
         no_gate(lambda: old.run(off + 120.0))
-        n = f"N={len(units)}, {lag:.0f} s late for {span:.0f} s"
+        n = f"N={len(units)}, {p1:.0f} s meter, {lag:.0f} s late for {span:.0f} s"
         wo, wn = window(old.rows, ON, off), window(new.rows, ON, off)
         good = last_good(new.rows, ON)
         half = ON + span / 2
+        red = min(r[1] for r in wo) < -1000.0
+        found = min([t for t, _l in seen] or [None])
         print(f"    {n}: without the gate cmd {min(r[2] for r in wo):.0f}.."
               f"{max(r[2] for r in wo):.0f} W, grid {min(r[1] for r in wo):.0f}"
               f"..{max(r[1] for r in wo):.0f} W (swing {swing(old.rows, ON, half):.0f}"
               f" then {swing(old.rows, half, off):.0f}); with it cmd "
               f"{min(r[2] for r in wn):.0f}..{max(r[2] for r in wn):.0f} W "
               f"(last good {good:.0f}), grid {min(r[1] for r in wn):.0f}.."
-              f"{max(r[1] for r in wn):.0f} W, settled "
+              f"{max(r[1] for r in wn):.0f} W, found late "
-              f"{settled_at(new.rows, off) - off:.0f} s after")
+              f"{None if found is None else found - ON} s after the start, "
-        if len(units) == 2:
+              f"settled {settled_at(new.rows, off) - off:.0f} s after")
+        if len(units) == 2 and p1 == 2.0:
             check(f"{n}: without the gate the export goes below -1 kW and the "
                   f"swing grows (red on 77fe31b)",
-                  min(r[1] for r in wo) < -1000.0
+                  red and swing(old.rows, half, off) > swing(old.rows, ON, half))
-                  and swing(old.rows, half, off) > swing(old.rows, ON, half))
+        if red:
-        check(f"{n}: with it the command stays within 250 W of the last good "
+            check(f"{n}: found late while the lag lasts", found is not None
-              f"one and no export below -200 W",
+                  and found < off)
-              all(abs(r[2] - good) <= 250.0 for r in wn)
+        if p1 < 5.0:
-              and min(r[1] for r in wn) >= -200.0)
+            # Found within about a minute: at 2 s the brief's bounds; at 1 s
+            # the swing grows faster before it is found, so twice them.
+            lim = 250.0 if p1 == 2.0 else 500.0
+            check(f"{n}: with it the command stays within {lim:.0f} W of the "
+                  f"last good one and no export below -{lim - 50:.0f} W",
+                  all(abs(r[2] - good) <= lim for r in wn)
+                  and min(r[1] for r in wn) >= -(lim - 50.0))
+        elif red:
+            # 5 s meter: only the batteries' fit runs (too few meter values),
+            # so it is found later; the export is cut to under a third.
+            check(f"{n}: with it the export is under a third of the run "
+                  f"without the gate",
+                  min(r[1] for r in wn) >= min(r[1] for r in wo) / 3.0)
         check(f"{n}: settled within 30 s after the lag stops",
               settled_at(new.rows, off) is not None
               and settled_at(new.rows, off) - off <= 30.0)
@@ -196,15 +214,24 @@ lastr = max([t for t, _r in rows_seen] or [None])
 print(f"    30 s late for 300 s: {n_late} late episode(s), {n_back} end(s), "
       f"the row from {first} to {lastr} s ({len(rows_seen)} s); "
       f"{rows_seen[0][1]['text'] if rows_seen else ''}")
-_d = c.answer.delay_s
+_ok_texts = {str(messages.row_readings_late(None))} | {
-check("found late within a minute of the lag starting, and an amber row says "
+    str(messages.row_readings_late(d)) for d in range(20, 33)}
-      "so with the delay it measured, in plain words, EN and NL",
+check("found late within 70 s of the lag starting, and an amber row says "
-      first is not None and first - ON <= 60.0
+      "so in plain words, EN and NL, with the batteries' delay once they have "
+      "shown it (20-32 s for 30 s late; never a figure from the meter alone)",
+      first is not None and first - ON <= 70.0
       and all(r["warn"] and not r["ok"] for _t, r in rows_seen)
-      and str(rows_seen[-1][1]["text"]) in {
+      and {str(r["text"]) for _t, r in rows_seen} <= _ok_texts
-          str(messages.row_readings_late(d)) for d in range(20, 33)}
       and rows_seen[-1][1]["text"].nl.startswith(
-          "Metingen uit Home Assistant komen zo'n "))
+          "Metingen uit Home Assistant tonen vermogensvraag"))
+rows5 = []
+late5 = R.Rig(lag_s=30.0, lag_on=ON, lag_off=ON + 300.0, p1_s=5.0)
+late5.run(ON + 300.0, lambda rig, c, t: rows5.extend(
+    str(r["text"]) for r in c.checks() if r.get("id") == "late"))
+print(f"    5 s meter, 30 s late: {sorted(set(rows5))}")
+check("at a 5 s meter (the batteries' fit) the row names the delay: about "
+      "20-32 s for 30 s late",
+      rows5 and set(rows5) <= _ok_texts and any("about" in x for x in rows5))
 check("logged once for the episode, and once when it ends, not per cycle",
       n_late == 1 and n_back == 1)
 check("the row goes once a fit shows the readings on time, or 2 minutes "
@@ -287,6 +314,51 @@ check("start-up: the first 120 s reach the house within 30 s, as without the "
       "gate, and no late episode",
       settled_at(a0.rows, 0.0) is not None and settled_at(a0.rows, 0.0) <= 30.0
       and not any("readings late" in e for e in a0.c.events))
+for on, off in ((45, 60), (30, 30)):
+    house = (lambda t, on=on, off=off:
+             400.0 + (2000.0 if t % (on + off) < on else 0.0))
+    oven = R.Rig(house=house, noise=80.0)
+    oven.run(1500.0)
+    oven0 = R.Rig(house=house, noise=80.0)
+    no_gate(lambda: oven0.run(1500.0))
+    check(f"an oven cycling 2 kW {on} s on, {off} s off (2 s meter, readings "
+          f"on time): never taken for late readings ({oven.c.answer.fits} "
+          "fits), the same commands as without the gate",
+          oven.c.answer.fits > 100 and not any(
+              "readings late" in e for e in oven.c.events)
+          and [r[2] for r in oven.rows] == [r[2] for r in oven0.rows])
+# One battery that empties to its cut-out level under an oven (1 s meter):
+# its fit lined the old movement up as 85 s late (round 2, before the fix).
+emp = R.Rig(house=lambda t: 400.0 + (2000.0 if t % 105 < 45 else 0.0),
+            noise=150.0, soc=10.6, units=("a",), p1_s=1.0)
+emp.run(1800.0)
+check("one battery emptying to its cut-out level under a cycling oven (1 s "
+      "meter): not taken for late readings",
+      not any("readings late" in e for e in emp.c.events))
+emp2 = R.Rig(house=lambda t: 400.0 + (2000.0 if t % 60 < 30 else 0.0),
+             noise=150.0, soc=10.2, units=("a",), p1_s=2.0)
+emp2.run(1800.0)
+check("one empty battery under an oven cycling 30/30 s (2 s meter): its "
+      "requests do not start, but it gives nothing, so that is no evidence "
+      "for the meter fit",
+      not any("readings late" in e for e in emp2.c.events))
+# The recorded house (2026-09-02, its first 2 hours, modes off) at a 2 s
+# meter: the meter fit alone read it as up to 65 s late here (review, B1).
+from datetime import datetime as _dt, timezone as _tz          # noqa: E402
+_tick, rp.TICK = rp.TICK, 2.0
+_late = []
+try:
+    _st, _rc = rp.run("modes_off", rp.load_house(["2026-09-02"]),
+                      rp.epex_points(),
+                      _dt.fromisoformat("2026-09-02T00:00:00+02:00")
+                      .astimezone(_tz.utc), 7200.0,
+                      check=lambda c, h, now, pr: _late.append(
+                          c.answer.late_seen))
+finally:
+    rp.TICK = _tick
+check(f"the recorded house, 2 hours at a 2 s meter: {_rc.answer.fits} fits, "
+      "never taken for late readings",
+      _rc.answer.fits > 100 and not any(_late))
 import safety94_rig as s94                                     # noqa: E402
 slow = [(n, tau, tick) for n in (1, 2) for tau in (10.0, 20.0)
         for tick in (1.0, 2.0)
@@ -316,39 +388,52 @@ check("ANSWER_W is a probe's answer (100 W)",
       lateness.ANSWER_W == control.PROBE_ANSWER_W == 100.0)
 
 
-check("DELAY_S is the plant's 6 s settle and one 2 s meter tick",
+check("late from ON_S 9.5 s, on time again under OFF_S 6.5 s",
-      lateness.DELAY_S == 8.0)
+      (lateness.ON_S, lateness.OFF_S) == (9.5, 6.5))
 
 
-def fit(delay, *, follows=True, step_every=20, noise=40.0, dur=300):
+def fit(delay, *, follows=True, step_every=20, dur=300, tau=0.0, units="ab"):
-    """`Delay` on a meter that shows each command `delay` s later (a house
+    """`Delay` on batteries whose readings, published every 10 s at their own
-    of 2 000 W with +-`noise`), or never (`follows` False)."""
+    phase, show each request `delay` s late (through a `tau` s lag), or never
+    (`follows` False)."""
+    import math
     import random
     rng, d = random.Random(7), lateness.Delay()
     edges = [0]
     while edges[-1] < dur:
         edges.append(edges[-1] + step_every * rng.uniform(0.5, 1.5))
-    cmd = lambda t: 1500.0 + (400.0 if sum(e <= t for e in edges) % 2 else 0.0)
+    cmd = lambda t: 1500.0 + (600.0 if sum(e <= t for e in edges) % 2 else 0.0)
-    for t in range(0, dur, 2):
+    act, shown = 1500.0, {}
-        shown = cmd(t - delay) if follows and t >= delay else 1500.0
+    for t in range(0, dur):
-        d.add(float(t), cmd(t), 2000.0 - shown + rng.uniform(-noise, noise))
+        want = cmd(t - delay) if t >= delay else 1500.0
+        act = want if not tau else act + (want - act) * (1 - math.exp(-1 / tau))
+        for k, u in enumerate(units):
+            if (t + 4 * k) % 10 == 0:
+                shown[u] = round(act + rng.uniform(-3, 3)) if follows else 900
+            d.add(float(t), u, cmd(t), shown.get(u))
     return d.estimate()
 
 
 got = {k: fit(k) for k in (0, 6, 10, 30, 60)}
-print(f"    fitted delays: {got}")
+print(f"    fitted delays: {got}; one battery {fit(10, units='a')}")
-check("the fit finds the meter's delay (0, 6, 10, 30, 60 s) to within 1 s",
+check("the fit finds the readings' delay (0, 6, 10, 30, 60 s) to within 2 s "
-      all(v is not None and abs(v - k) <= 1.0 for k, v in got.items()))
+      "(it reads the request over the 2 s before), also with one battery",
-check("a battery that does not follow gives no fit, and so does a command "
+      all(v is not None and abs(v - k) <= 2.0 for k, v in got.items())
-      "that does not move",
+      and abs((fit(10, units="a") or 0.0) - 10.0) <= 2.0)
-      fit(10, follows=False) is None and fit(10, step_every=10_000) is None)
+slow = [fit(0, tau=tau, step_every=s) for tau in (3.0, 10.0, 20.0)
+        for s in (10, 20, 40)]
+check("a battery that does not follow gives no fit, nor does a request that "
+      f"does not move, and a slow battery (3, 10, 20 s lag) is never late "
+      f"({slow})",
+      fit(10, follows=False) is None and fit(10, step_every=10_000) is None
+      and all((d or 0.0) < lateness.ON_S for d in slow))
 
 
 def gate(steps, exempt=(), late_seen=True):
     """[(t, command, battery reading, meter)] through one Answer, its fit
     replaced by `late_seen`; after each: (pending, hold, late s, peak cap)."""
     a = lateness.Answer()
-    a.delay.estimate = lambda: (30.0 if late_seen else 0.0)
+    a.late_seen = late_seen
 
     def one(t, cmd, batt, grid):
         a.observe(float(t), cmd, batt, grid_w=grid, exempt=t in exempt)