SAFETY-87 review: maintenance pins beside a held battery

A draining battery that may still give is raised from its drain rate
towards its rating for the peak floor while another battery is held
(peak > drain); before, the floor's rest landed on the held battery. In a
HOLD, a held battery's own pin counts among the others: a held drain no
longer over-places the site, and a held charge probe no longer parks it.

Tests: test_safety87.py part 10 (drain beside a held battery, N=3 HOLD,
unknown cut-out and unread level at N=3, a stopping row outranks the
running peak alarm). DECISIONS entry updated, incl. the modes-on exception.

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

goodwe_controller/test_safety87.py

@@ -44,8 +44,8 @@ import opt11_identity as rig                                   # noqa: E402
 import safety84_identity as s84                                # noqa: E402
 import safety87_identity as s87                                # noqa: E402
 from app import messages, packmode, peakguard                  # noqa: E402
-from app.arbiter import (Claim, P_MODE, P_PEAK, P_SAFETY,       # noqa: E402
-                         UnitBounds, resolve)
+from app.arbiter import (Claim, P_MAINTENANCE, P_MODE, P_PEAK,  # noqa: E402
+                         P_SAFETY, UnitBounds, resolve)
 from app.main import Controller                                # noqa: E402
 from app.maintenance import DRAIN, HOLD, IDLE, MaintResult     # noqa: E402
 from app.store import unit_key                                 # noqa: E402
@@ -540,6 +540,122 @@ check("⭐ every N=1 digest (claims, commands, the guard's decision and the "
       "charge and a HOLD, and under a banked peak",
       _now == PINNED)
 
+# ============================================================================ #
+print("\n10. review: maintenance pins beside a held battery, N=3, the page")
+
+
+def _drain_store():
+    s = rig._Store()
+    s.data[unit_key(B, "phase")] = DRAIN
+    s.data[unit_key(B, "phase_started")] = rig.BASE.isoformat()
+    s.data[unit_key(A, "last_completed")] = rig.BASE.isoformat()
+    return s
+
+
+def _drain_run():
+    return run(8000.0, IDLE, {"a": 15.0, "b": 60.0}, store=_drain_store(),
+               dur=900.0)
+
+
+DR = _drain_run()
+_real_dor = Controller._drain_over_reserve
+Controller._drain_over_reserve = lambda self, claims, results: None
+try:
+    DR_OLD = _drain_run()
+finally:
+    Controller._drain_over_reserve = _real_dor
+
+
+def _a_for_peak(rows):
+    """A's command on the cycles where the site target is the peak floor
+    and B alone (2500 W) could place it."""
+    return [cmd(r)["a"] for r in rows
+            if r[-1][1] and r[-1][1][1] and r[1] <= 2500.0
+            and abs(r[1] - r[-1][1][1]) < 1.0]
+
+
+print(f"    B draining, A held: A for the peak {_a_for_peak(DR[0])[:4]}, "
+      f"without the raise {_a_for_peak(DR_OLD[0])[:4]}")
+check("⭐ B draining (2000 W pin) while A is held: B is raised to its rating "
+      "for the peak floor, so A gives nothing for it (peak > drain)",
+      DR[1].maints[1].phase == DRAIN and _a_for_peak(DR[0])
+      and all(a <= 0.0 for a in _a_for_peak(DR[0])))
+check("non-vacuous: without the raise A gave the floor's rest above B's pin",
+      any(a > 0.0 for a in _a_for_peak(DR_OLD[0])))
+
+
+def _stub(names, reserve, live=None):
+    """A Controller stand-in for the pin rules, on SIMULATED units."""
+    s = types.SimpleNamespace(
+        inverters=[types.SimpleNamespace(name=n) for n in names],
+        actuator_of=lambda src, inv=None: inv.name,
+        live_units=lambda: tuple(live or names),
+        _peak_reserve=lambda: frozenset(reserve))
+    s._held_pin = lambda cl, u: Controller._held_pin(s, cl, u)
+    return s
+
+
+def _hold(floor, names, reserve, extra=(), results=None):
+    """`_peak_over_hold()`: x HOLDs; returns (x's pin, resolved shares)."""
+    s = _stub(names, reserve)
+    cl = [Claim.limit("supervised", P_SAFETY, -2500.0, 2500.0, actuator=n)
+          for n in names]
+    cl += [Claim.set("maintenance", P_MAINTENANCE, 0.0, HOLD, actuator="x"),
+           Claim.limit("peak", P_PEAK, lo=floor, reason="quarter peak"),
+           *extra]
+    res = results or ([MaintResult(HOLD, True, 0.0, False, [])]
+                      + [MaintResult(IDLE, False, None, False, [])]
+                      * (len(names) - 1))
+    hi = Controller._peak_over_hold(s, cl, res)
+    if hi is not None:
+        cl.append(Claim.limit("hold", P_MAINTENANCE, hi=hi))
+    r = resolve(cl + [Claim.set("loop", 10, floor)], units=tuple(names),
+                live=set(names), reserve=reserve)
+    pin = next((c.value for c in cl
+                if c.source == "maintenance" and c.actuator == "x"), None)
+    return pin, {u: round(v) for u, v in r.shares.items()}
+
+
+_h3 = _hold(3000.0, ("x", "y", "z"), {"z"})
+_h3b = _hold(2000.0, ("x", "y", "z"), {"z"})
+print(f"    N=3 HOLD x, y able, z held: floor 3000 {_h3}, 2000 {_h3b}")
+check("⭐ HOLD at N=3 with one held: the able battery gives first, the "
+      "holding one the rest (3000 -> y 2500, x 500), the held one nothing",
+      _h3 == (500.0, {"x": 500, "y": 2500, "z": 0})
+      and _h3b == (0.0, {"x": 0, "y": 2000, "z": 0}))
+_hd = _hold(2000.0, ("x", "z"), {"z"},
+            (Claim.set("maintenance", P_MAINTENANCE, 1500.0, DRAIN,
+                       actuator="z"),),
+            [MaintResult(HOLD, True, 0.0, False, []),
+             MaintResult(DRAIN, True, 1500.0, False, [])])
+_hp = _hold(2000.0, ("x", "z"), {"z"},
+            (Claim.set("refused", P_SAFETY, -50.0, "probe", actuator="z"),))
+print(f"    held z draining 1500: {_hd}; held z charge probe: {_hp}")
+check("⭐ a held battery's own pin counts: draining 1500 W, the hold gives "
+      "only the 500 W rest (no over-placement); a charge probe leaves the "
+      "hold at 0 W (never split both ways, no park)",
+      _hd == (500.0, {"x": 500, "z": 1500}) and _hp[0] == 0.0)
+
+_unk = three((60.0, 60.0, 60.0), (10, None, 10))
+_unr = three((60.0, None, 60.0), (10, 10, 10))
+check("⭐ N=3: an unknown cut-out or an unread level holds that battery "
+      "only; the two others are asked for the need",
+      _unk.held == ("y",) and not _unk.floor_limited
+      and _unk.floor_w == 2000.0 and _unr.held == ("y",)
+      and not _unr.floor_limited and _unr.floor_w == 2000.0)
+
+_stop_row = {"ok": False, "warn": False, "id": "meter", "text": "Meter down."}
+_peak_rows = [r for r in _c65.checks() if r.get("id") == "peak_guard"]
+_c65.checks = lambda: _peak_rows[:1] + [_stop_row]
+_both = _c65.status_payload()
+del _c65.checks
+check("⭐ a row that stops control outranks the running peak alarm: the "
+      "banner says 'Not ready' for it and the Home page shows the hold",
+      len(_peak_rows) == 1 and _peak_rows[0].get("runs")
+      and _both["banner"].startswith(messages.NOT_READY)
+      and "Meter down" in _both["banner"]
+      and "until the alarm clears" in json.dumps(_both, ensure_ascii=False))
+
 print()
 if fails:
     print(f"{len(fails)} FAILED: " + ", ".join(fails))