SAFETY-78 review: card truth while the drain waits, more checks

- homeview: no "This one does not charge" while the draining battery waits
  (no floor then, so the other battery may charge from solar).
- test_safety78: one wait line across a probe, a second episode logs again,
  the card while waiting, and a seeded sweep of the floor against every
  other claim (part 6).
- DECISIONS: the wait case, and the measured N=1 peak and N=2 charge
  follow-ups.

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

goodwe_controller/DECISIONS/SAFETY-78-drain-at-two.md

@@ -29,10 +29,16 @@ direction. Opposite splits stay forbidden.
   floor, and the wait is logged once per episode.
 - With battery modes on, no pack is a grid charger while another drains
   (`modetrade.decide(draining=)`, why `cycle`), and the other battery's card
-  says why (`other_cycle`).
+  says why (`other_cycle`). Not while the drain waits: no floor is filed
+  then, so the other battery may charge from solar.
 - Not filed at N=1, where the drain is the site SET itself: byte-identical
   (`tools/safety78_identity.py`, which hashes the claims too).
 
-**Not changed.** At N=1 a drain (60) still outranks the peak floor (50), so
-the peak guard does not add to a drain there. That is pre-existing and kept by
-the N=1 identity.
+**Not changed (measured in the PR #154 review, follow-ups).**
+- At N=1 a drain (60) outranks the peak floor (50), so the guard does not add
+  to it: a 5000 W house imports 3000 W against a 2500 W limit, a 6000 W house
+  4000 W. Pre-existing, kept by the N=1 identity.
+- At N=2 a maintenance CHARGE in a cheap window has the same stall: the
+  charging pack's `hi = -floor` against the loop asking the other to give is
+  refused on 141 of 261 cycles. The mirror fix is a site `hi`, which can meet
+  the peak floor's `lo` (0 W), so it is not this card.

goodwe_controller/app/homeview.py

@@ -641,10 +641,13 @@ def battery_rows(c, shadow, running, held=False):
         other = next((o.name for o in c.inverters if o is not inv
                       and refusing(c, o) > 0 and not _off(c, o)), None)
         # SAFETY-78: another battery drains for its monthly cycle, so this
-        # one never charges now.
+        # one never charges now. Not while that drain waits: it files no
+        # floor then, so this battery may charge from solar.
+        waits = getattr(c, "_drain_waits", ())
         cycling = next((o.name for j, o in enumerate(c.inverters)
                         if o is not inv and j < len(c.maints)
-                        and c.maints[j].phase == M_DRAIN), None)
+                        and c.maints[j].phase == M_DRAIN
+                        and o.name not in waits), None)
         role = None if tick is None else tick.roles.get(inv.name)
         # Only once a cycle has run with modes on (`tick`): before that
         # nothing is known, so nothing is claimed about the prices.

goodwe_controller/test_safety78.py

@@ -16,6 +16,8 @@ DECISIONS/SAFETY-78-drain-at-two.md has the record.
    no grid charge, and its card says why.
 4. The peak guard binds during a drain, and the arbiter's order holds.
 5. N=1 is byte-identical to release/1.0 5a4cdc2, and the harness bites.
+6. A seeded sweep of the floor against the loop, trades, the peak floor, a
+   safety stop, refusals and an Off battery.
 
 Mutations (MUTATION.md, purged copies, graded over this file and the named
 tests) are listed in the pull request.
@@ -24,6 +26,7 @@ tests) are listed in the pull request.
 import functools
 import logging
 import os
+import random
 import sys
 import tempfile
 
@@ -36,9 +39,9 @@ import opt11_identity as rig                                   # noqa: E402
 import opt11p3_identity as p3                                  # noqa: E402
 import safety78_identity as s78                                # noqa: E402
 import safety79_identity as s79                                # noqa: E402
-from app import arbitrage, homeview, packmode                  # noqa: E402
+from app import allocator, arbitrage, homeview, packmode       # noqa: E402
 from app.arbiter import (Claim, P_ARBITRAGE, P_LOOP, P_MAINTENANCE,  # noqa
-                         P_PEAK, P_SAFETY, resolve)
+                         P_MODE, P_PEAK, P_SAFETY, pre_resolve, resolve)
 from app.main import Controller                                # noqa: E402
 from app.maintenance import CHARGE, DRAIN                      # noqa: E402
 from app.store import unit_key                                 # noqa: E402
@@ -151,6 +154,22 @@ check("⭐ A refuses to give power (SAFETY-73): the cycle waits, the line is "
 check("...and B covers the house on its own meanwhile",
       1400.0 <= cmd(_w_rows[-1])["b"] <= 1600.0)
 
+# PR #154 review: one line per episode, across a probe and a second episode.
+_pr_rows, _pr = drain_run(1500.0, dur=1300.0, cut=lambda u, t: u == "a")
+check("the SAFETY-73 probe (A pinned at 200 W, not by maintenance) does not "
+      "end the wait: still ONE line over 1300 s with a probe in it",
+      any(cmd(r)["a"] == 200.0 for r in _pr_rows)
+      and sum("cannot give power now" in e for e in _pr.events) == 1)
+_e2_rows, _e2 = drain_run(1500.0, dur=2600.0, cut=lambda u, t: u == "a" and (
+    t < 600.0 or t >= 1400.0))
+check("...and a second episode, after the probe ends the first, logs again",
+      sum("cannot give power now" in e for e in _e2.events) == 2)
+_cw_rows, _cw = drain_run(-1500.0, dur=300.0, cut=lambda u, t: u == "a")
+_cw_cards = homeview.battery_rows(_cw, False, True)
+check("while A waits, B may take the solar surplus, so its card does not "
+      "say 'This one does not charge'",
+      cmd(_cw_rows[-1])["b"] < 0.0 and _cw_cards[1]["why_key"] != "other_cycle")
+
 
 def reject_a(c, t):
     """Every write to A is rejected, so A's link goes stale (not live)."""
@@ -339,6 +358,65 @@ check("non-vacuous: filing the drain floor at N=1 (which moves no watt "
       "there) moves the digest, because the harness hashes the claims",
       _leak != _ref and _leak == _same)
 
+# ============================================================================ #
+print("\n6. the floor against every other claim (PR #154 review, seeded sweep)")
+_rnd = random.Random(78)
+_bad, _peak_above, _off_gave = [], 0, 0
+for _i in range(20000):
+    ma, mb = _rnd.choice((2500., 3000., 5000.)), _rnd.choice((1500., 2500.))
+    ua = [Claim.limit("supervised", P_SAFETY, -ma, ma, actuator=A),
+          Claim.set("maintenance", P_MAINTENANCE,
+                    _rnd.choice((500., 2000., 6000.)), "drain", actuator=A)]
+    if _rnd.random() < .2:
+        ua.append(Claim.limit("pack", P_SAFETY, -ma,
+                              _rnd.choice((0., 800.)), actuator=A))
+    ub = [Claim.limit("supervised", P_SAFETY, -mb, mb, actuator=B)]
+    _offb = _rnd.random() < .4
+    if _offb:
+        ub.append(Claim.limit("mode", P_MODE, 0., 0., "off", actuator=B))
+    if _rnd.random() < .15:
+        ub.append(Claim.limit("refused", P_SAFETY, -mb, 0., actuator=B))
+    pin = allocator.pin_w(pre_resolve(ua))
+    site = [Claim.set("loop", P_LOOP, _rnd.uniform(-6000., 8000.))]
+    if pin is not None and pin > 0.0:
+        site.append(Claim.limit("drain", P_MAINTENANCE, lo=pin))
+    if _rnd.random() < .3:
+        site.append(Claim.set("arbitrage", P_ARBITRAGE,
+                              _rnd.choice((-3000., 1000., 4000.))))
+    pk = _rnd.uniform(0., 8000.) if _rnd.random() < .4 else None
+    if pk is not None:
+        site.append(Claim.limit("peak", P_PEAK, lo=pk))
+    stop = _rnd.random() < .05
+    if stop:
+        site.append(Claim.set("safety", P_SAFETY, 0.0, "stopped"))
+    r = resolve(ua + ub + site, units=UNITS)
+    s = dict(r.shares)
+    why = []
+    if min(s.values()) < 0 < max(s.values()) or r.contradiction:
+        why.append("opposite or contradiction")
+    if stop and (r.target_w != 0.0 or any(s.values())):
+        why.append("stop not parked")
+    if not stop and pin is not None and pin > 0.0:
+        if r.target_w < pin or s[B] < 0:
+            why.append("below the floor, or B charges")
+        if pk is not None and pk > pin:
+            _peak_above += 1
+            if r.allocated and r.target_w < min(pk, pin + mb):
+                why.append("peak floor not honoured")
+        if _offb and s[B] > 0:
+            _off_gave += 1
+            if not (pk is not None and pk > pin):
+                why.append("Off B used for the drain")
+    if why:
+        _bad.append((_i, why, r.target_w, s, pin, pk))
+print(f"    {_bad[:3]}; peak above the drain {_peak_above}, Off B gave for "
+      f"the peak {_off_gave}")
+check("⭐ over 20000 seeded claim lists: never an opposite split or a "
+      "contradiction; a safety stop parks both; the site never goes below "
+      "the pin and B never charges; a peak floor above the drain is met; an "
+      "Off battery gives only for the peak floor, never for the drain",
+      not _bad and _peak_above > 1000 and _off_gave > 100)
+
 print()
 if fails:
     print(f"{len(fails)} FAILED: {', '.join(fails)}")