SAFETY-91 review: a held-back battery is released 1 point above its reserve

A SoC reading flipping across the reserve swapped ~1000 W between the
batteries on each flip (grid +-500 W with unequal lags). Also grades
next_quarter()'s staleness boundary (29.5 / 30.5 s).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

goodwe_controller/test_safety91.py

@@ -14,7 +14,8 @@ on the rig). DECISIONS/SAFETY-91-held-battery-last.md.
 1. B draining, A held, house over B's rating: B rises to full before A gives.
 2. A held reaching its cut-out: nothing is placed on it, B gives its rating.
 3. HOLD across a quarter edge: the held battery gives nothing at the edge.
-4. Idle, A held: B carries the house first; both held or B refusing: as before.
+4. Idle, A held: B carries the house first; both held or B refusing: as before;
+   a SoC reading flipping across the reserve does not swap the split.
 5. The rules at their boundaries (claim lists, N=3, a site stop).
 6. next_quarter() at its boundaries.
 7. N=1 is untouched.
@@ -58,9 +59,10 @@ def check(name, cond):
         fails.append(name)
 
 
-def lagging(fn, lag=10.0, tick=1.0):
-    """`fn` on plantsim's battery (800 W/s ramp, `lag` s first-order lag), a
-    1 s meter tick, and a BMS that stops discharge at its cut-out."""
+def lagging(fn, lag=10.0, tick=1.0, lags=None):
+    """`fn` on plantsim's battery (800 W/s ramp, `lag` s first-order lag, or
+    `lags[u]`), a 1 s meter tick, and a BMS that stops discharge at its
+    cut-out."""
     real_p, real_t = rig._Hass.power, rig.TICK
 
     def power(self, u):
@@ -69,7 +71,7 @@ def lagging(fn, lag=10.0, tick=1.0):
         for _ in range(int(round((self.t - tl) / 0.1))):
             d = self.cmd[u] - tgt
             tgt += math.copysign(min(abs(d), 80.0), d)
-            a += (tgt - a) * (1 - math.exp(-0.1 / lag))
+            a += (tgt - a) * (1 - math.exp(-0.1 / (lags or {}).get(u, lag)))
         st[u] = (a, tgt, self.t)
         return 0.0 if self.soc(u, self.t) <= CUT and a > 0 else float(round(a))
     rig._Hass.power, rig.TICK = power, tick
@@ -92,12 +94,16 @@ def store(phase_of, n=2):
     return s
 
 
-def run(house, phase_of, socs, dur, late=0.0, kwh=15.36, n=2):
+def run(house, phase_of, socs, dur, late=0.0, kwh=15.36, n=2, read=None,
+        lags=None):
     """Rows and controller. Levels follow each battery's measured power; the
-    P1 telegram read at each cycle is `late` s old."""
+    P1 telegram read at each cycle is `late` s old. `read(u, t, level)` is
+    the SoC the controller reads (default the level)."""
     soc, last, levels = dict(socs), [0.0], {}
+    grid = []
 
     def before(c, t):
+        grid.append((t, c.hass.grid()))
         for u in soc:
             soc[u] = max(CUT, soc[u] - c.hass.power(u) * (t - last[0])
                          / 3600.0 / (kwh * 10.0))
@@ -108,10 +114,12 @@ def run(house, phase_of, socs, dur, late=0.0, kwh=15.36, n=2):
                                 ingest_mono=time.monotonic()))
     rows, c = lagging(lambda: s84.run(
         n, load=house if callable(house) else (lambda t: house), cheap=False,
-        store=store(phase_of, n), soc=lambda u, t: soc[u], before=before,
-        dur=dur, extra={"maintenance_discharge_w": 1000,
-                        "maintenance_hold_min": 600}))
-    c.levels = levels
+        store=store(phase_of, n), before=before, dur=dur,
+        soc=(lambda u, t: soc[u]) if read is None
+        else (lambda u, t: read(u, t, soc[u])),
+        extra={"maintenance_discharge_w": 1000,
+               "maintenance_hold_min": 600}), lags=lags)
+    c.levels, c.grid = levels, grid
     return rows, c
 
 
@@ -273,6 +281,61 @@ check("...held back only while another can give: A held beside B able; "
       and held_back((15.0, 14.0)) == frozenset()
       and held_back((15.0, 60.0), (0, -1)) == {A})
 
+# A's SoC READING flips 20/21 % across its 20 % reserve every 10 s, and B
+# follows its command faster (3 s) than A (10 s), as two different inverters.
+def _flip(u, t, level):
+    return (20.0 + int(t // 10) % 2) if u == "a" else level
+
+
+def _a_jumps(rows, t0):
+    rs = [r for r in rows if r[0] >= t0]
+    return sum(1 for p, q in zip(rs, rs[1:])
+               if abs(cmd(p)["a"] - cmd(q)["a"]) > 300.0)
+
+
+def _grid_rms(c, t0):
+    g = [w + 10.0 for t, w in c.grid if t >= t0]
+    return math.sqrt(sum(w * w for w in g) / len(g))
+
+
+_lags = {"a": 10.0, "b": 3.0}
+FL = run(2000.0, {}, {"a": 20.0, "b": 60.0}, 600.0, read=_flip, lags=_lags)
+_real_was = Controller.__dict__["_held_back"]
+
+
+def _no_hyst(self):
+    self._held_was = frozenset()
+    return _real_was(self)
+
+
+Controller._held_back = _no_hyst
+try:
+    FL_OLD = run(2000.0, {}, {"a": 20.0, "b": 60.0}, 600.0, read=_flip,
+                 lags=_lags)
+finally:
+    Controller._held_back = _real_was
+print(f"    A's reading flipping 20/21 %: A jumps {_a_jumps(FL[0], 120)}, "
+      f"grid rms {_grid_rms(FL[1], 120):.0f} W (no hysteresis: "
+      f"{_a_jumps(FL_OLD[0], 120)}, {_grid_rms(FL_OLD[1], 120):.0f} W)")
+check("⭐ a SoC reading flipping across the reserve does not swap the split: "
+      "A stays held back (0 W) and the grid stays near its target",
+      _a_jumps(FL[0], 120) == 0 and seg(FL[0], 120, 600)[0] == 0
+      and _grid_rms(FL[1], 120) < 30.0)
+check("non-vacuous: without the 1-point hysteresis the split swaps on "
+      "every flip and the grid swings",
+      _a_jumps(FL_OLD[0], 120) >= 20 and _grid_rms(FL_OLD[1], 120) > 100.0)
+REL = run(2000.0, {}, {"a": 20.0, "b": 60.0}, 300.0,
+          read=lambda u, t, lv: (20.0 if t < 100 else 21.0 if t < 200
+                                 else 22.0) if u == "a" else lv)
+check("...held back at 20 %, still at 21 %, released at 22 % (more than 1 "
+      "point above its reserve)",
+      seg(REL[0], 40, 100)[0] == 0 and seg(REL[0], 140, 200)[0] == 0
+      and seg(REL[0], 240, 300)[0] > 500)
+ID[1]._held_was = frozenset({A, B})
+check("...a battery kept held back by the hysteresis is not one that can "
+      "give: A at 20 % and B kept at 21 % hold nobody back",
+      held_back((20.0, 21.0)) == frozenset())
+
 # ============================================================================ #
 print("\n5. the rules at their boundaries")
 
@@ -346,6 +409,12 @@ check("⭐ past its end with a fresh held reading: the next quarter, "
       and _past.covered_frac == 1.0
       and peakguard.blindness(_past, 30.0, False) is None)
 _old = proj(960.0, covered=880.0 / 960.0)
+_edge29 = proj(901.0, covered=871.5 / 901.0)     # 29.5 s uncovered
+_edge31 = proj(901.0, covered=870.5 / 901.0)     # 30.5 s uncovered
+check("...the held reading is credited up to max_hold_s and no further "
+      "(29.5 s uncovered: rolled; 30.5 s: unchanged)",
+      next_quarter(_edge29, 30.0) is not _edge29
+      and next_quarter(_edge31, 30.0) is _edge31)
 check("...a held reading too old to be credited, no block, or an export: "
       "unchanged, or projected at 0 W offtake",
       next_quarter(_old, 30.0) is _old