SAFETY-52: a battery that refuses is asked for 0 W, with a 200 W probe every 15 min

At the BMS cut-out the saturation freeze held the last request for hours
(+5000 W for 11.87 h on 09-06, +1940 W for 3 h 55 m on 09-19) until the grid
error changed sign. compute() now carries a refused direction: after
SETTLE_CYCLES of a ~0 W reading against a larger ask it holds that direction
at 0 W, probes 200 W (below the gate's 250 W) every 15 minutes, and lets go
on charge, an answered probe, stop/start or another writer. A stale-reading
trip on a frozen ~0 W reading seeds the same hold, which ends SAFETY-59's
~70 s cycle. One inverter only; no SoC in the loop (SAFETY-25 Option A).

test_refusal.py drives the real Controller and gate; tools/replay_refusal.py
replays the 16-day capture. test_clamp.py and test_delivery.py re-baselined
where they asserted the old hold. DECISIONS/SAFETY-52 has the record and what
SAFETY-58/59 must take into account.

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

goodwe_controller/tools/replay_refusal.py

@@ -0,0 +1,97 @@
+#!/usr/bin/env python3
+"""SAFETY-52's evidence: the 16-day capture replayed with and without the
+battery-refuses hold.  `python3 tools/replay_refusal.py`
+
+Same disturbance, fitted plant and pack floor as `replay_recorded_loop.py`,
+plus main.py's reseed, 15 s input hold and 0 W rules and the REAL SAFETY-37/57
+gate on a register published every 10 s. `old` is `compute()` with
+`refused=None`, which is release/1.0's law unchanged; `new` carries the refusal,
+runs the probe on main.py's clock and seeds a refusal from a gate trip.
+Modelled watt-hours: read the deltas, not the absolutes (see that tool).
+"""
+import math
+import pathlib
+import sys
+
+ADDON = pathlib.Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(ADDON))
+sys.path.insert(0, str(ADDON / "tools"))
+import replay_recorded_loop as R                            # noqa: E402
+from app import freshness, control as NEW                   # noqa: E402
+
+def go(which, rows, fit, publish_s=10.0, stale_input_s=15.0):
+    law = NEW
+    tun = law.Tuning(**R.LIVE)
+    t0, g0, b0, sp0, soc0 = rows[0]
+    pl = R.Plant(*fit[:4], soc0); pl.seed(t0, b0, sp0)
+    st = dict(target=sp0, i_w=sp0, sat=0, loop_w=sp0, ref=0, anchor=None)
+    fr = freshness.Freshness(freshness.BOUNDS["batt"]); reg, reg_t = None, -1e18
+    bad, trips, was = None, 0, False
+    M = dict(und=0.0, imp=0.0, exp=0.0, q={}, run=0.0, mx=0.0, last=None, refh=0.0, probes=0)
+    for i in range(1, len(rows)):
+        t, gr, ba, sp, soc = rows[i]
+        dt = t - rows[i - 1][0]
+        if dt > R.TICK_CAP_S:
+            pl = R.Plant(*fit[:4], soc); pl.seed(t, ba, sp)
+            st.update(target=sp, i_w=sp, sat=0, loop_w=sp, ref=0, anchor=None)
+            fr = freshness.Freshness(freshness.BOUNDS["batt"]); reg_t = -1e18; bad = None
+            continue
+        b = pl.step(t, dt, st["target"]); grid = gr + ba - b
+        if t - reg_t >= publish_s:
+            reg, reg_t = float(round(b)), t
+        fr.submit(reg, st["target"], t); batt = fr.reading()
+        if fr.stale and not was:
+            trips += 1
+            if (which == "new" and not st["ref"] and st["sat"] >= tun.saturation_cycles
+                    and abs(reg) < tun.saturation_w and abs(st["target"]) > tun.saturation_w):
+                st["ref"], st["anchor"] = (1 if st["target"] > 0 else -1), t
+        was = fr.stale
+        if batt is None:
+            bad = t if bad is None else bad
+            if t - bad >= stale_input_s:
+                st.update(target=0.0, i_w=0.0, loop_w=None)
+        else:
+            bad = None
+            if st["loop_w"] is None or st["target"] != st["loop_w"]:
+                st["i_w"] = st["target"]
+            if which == "old":
+                d = law.compute(st["target"], grid, batt, tun, st["sat"], st["i_w"],
+                                refused=None)
+            else:
+                probe = False
+                if st["ref"] and st["anchor"] is not None:
+                    h = t - st["anchor"]
+                    if h >= NEW.PROBE_EVERY_S + NEW.PROBE_S: st["anchor"] = t
+                    else: probe = h >= NEW.PROBE_EVERY_S
+                d = law.compute(st["target"], grid, batt, tun, st["sat"], st["i_w"], refused=st["ref"], probe=probe)
+                if d.refused and not st["ref"]: st["anchor"] = t
+                if not d.refused: st["anchor"] = None
+                if probe and d.target_w == NEW.PROBE_W and M["last"] != NEW.PROBE_W: M["probes"] += 1
+                st["ref"] = d.refused
+            st.update(sat=d.sat_count, i_w=d.i_w, loop_w=d.target_w, target=d.target_w)
+        tg = st["target"]
+        if tg - b > 1000: M["und"] += dt / 3600
+        if st["ref"]: M["refh"] += dt / 3600
+        M["imp"] += max(0, grid) * dt / 3.6e6; M["exp"] += max(0, -grid) * dt / 3.6e6
+        k = math.floor(t / 900); a = M["q"].setdefault(k, [0.0, 0.0]); a[0] += grid * dt; a[1] += dt
+        M["run"] = M["run"] + dt if (M["last"] == tg and abs(tg) >= 250) else 0.0
+        M["mx"] = max(M["mx"], M["run"]); M["last"] = tg
+    worst = max(max(0.0, s / d) for s, d in M["q"].values() if d > 600)
+    return trips, M, worst
+
+def main():
+    rows = R.stream()
+    fit = R.fit_plant(rows, stride=7)
+    print(f"fit {[round(x, 2) for x in fit]}, {len(rows)} meter instants")
+    for label, f in (("fitted plant", fit),
+                     ("plant offset 0 W", (fit[0], fit[1], fit[2], 0.0, fit[4]))):
+        for which in ("old", "new"):
+            tr, M, w = go(which, rows, f)
+            print(f"{label:17s} {which}: gate trips {tr:4d} | request >1 kW above "
+                  f"delivery {M['und']:6.1f} h | longest held >=250 W "
+                  f"{M['mx'] / 3600:5.2f} h | import {M['imp']:6.1f} kWh export "
+                  f"{M['exp']:5.1f} kWh | worst 15-min {w:5.0f} W | refused "
+                  f"{M['refh']:5.1f} h, probes {M['probes']}", flush=True)
+
+if __name__ == "__main__":
+    main()