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/DECISIONS/SAFETY-87-guard-uses-able-batteries.md

@@ -45,7 +45,15 @@ graded on its own floor, and an unread SoC or unknown cut-out counts as held.
 The min/max pairing still decides permission, over the able batteries. The
 roster verdict `decide()` echoes (`soc`, `soc_floor`) is kept, so SAFETY-32's
 call-site and reduction checks still bite. What changed is only that one held
-battery no longer stops the others.
+battery no longer stops the others. One exception, with battery modes on: see
+Open.
+
+**Maintenance pins beside a held battery** (review). A draining battery that
+may still give is raised from its drain rate towards its rating for the peak
+floor, after the others (`_drain_over_reserve()`; peak > drain), else the
+floor's rest landed on the held battery. In a HOLD, a held battery counts its
+own pin (a drain) among the others, and a held battery pinned to charge stops
+the raise (never split both ways).
 
 **Before -> after** (test_safety87.py, rig of 2 x 2500 W, aim 3450 W):
 
@@ -66,7 +74,9 @@ battery no longer stops the others.
   says "cannot keep … alone", not "will go over", so it stays true.
 - Battery modes on, A Off and B held: when B alone can place the floor, OPT-11
   keeps A resting and B gives it. When B cannot, A gives first. Making Off give
-  before a held battery in the first case needs a third allocator tier.
+  before a held battery in the first case needs a third allocator tier. The
+  same at N=3: A Off, B held, C able, floor 4000 W gives C 2500, B 1500, A 0.
+  Until then, with modes on, a held battery can give for the peak.
 - The notification and row text live in `messages.py`, which has no NL layer
   yet (UX-19). The Home page text, which has NL, is unchanged.
 

goodwe_controller/app/main.py

@@ -2562,9 +2562,11 @@ class Controller:
         (P_PEAK), so the floor did not bind and the house imported what the
         drain left. Where the floor asks for more than the drain rate, it is
         refiled at P_MAINTENANCE and wins; nothing else changes. Safety claims
-        (P_SAFETY) still outrank it. At N>=2 the `drain` floor does this.
+        (P_SAFETY) still outrank it. At N>=2 the `drain` floor does this, and
+        `_drain_over_reserve()` when a battery is held at its reserve.
         """
         if len(self.inverters) != 1:
+            self._drain_over_reserve(claims, maint_results)
             return
         res = maint_results[0]
         if res.phase != DRAIN or not res.owns_setpoint:
@@ -2574,6 +2576,59 @@ class Controller:
             if c.source == peakguard.SOURCE and c.lo > rate:
                 claims[i] = dataclasses.replace(c, priority=P_MAINTENANCE)
 
+    def _held_pin(self, claims, u):
+        """SAFETY-87: what held unit `u` places anyway (its own pin), or None
+        when that pin charges, so the site is never split both ways."""
+        b = pre_resolve([c for c in claims if c.actuator == u])
+        p = None if b.contradiction else allocator.pin_w(b)
+        return None if p is not None and p < 0.0 else max(p or 0.0, 0.0)
+
+    def _drain_over_reserve(self, claims, maint_results) -> None:
+        """SAFETY-87, N>=2: while a battery is held at its peak reserve, a
+        draining battery that may still give is raised from its drain rate
+        towards its rating for the peak floor, after the others. Otherwise
+        the floor's rest lands on the held battery (its weight 0 only orders
+        the split; the drain pin binds). A site stop still decides alone."""
+        reserve = self._peak_reserve()
+        floor = max((c.lo for c in claims if c.source == peakguard.SOURCE),
+                    default=0.0)
+        if not reserve or not floor > 0.0 or any(
+                c.kind == SET and c.actuator == ACTUATOR_SITE
+                and c.priority > P_PEAK for c in claims):
+            return
+        live = set(self.live_units())
+        drains, others = [], 0.0
+        for inv, res in zip(self.inverters, maint_results):
+            u = self.actuator_of("maintenance", inv)
+            if u not in live:
+                continue
+            if inv.name in reserve:
+                p = self._held_pin(claims, u)
+                if p is None:
+                    return
+                others += p
+                continue
+            b = pre_resolve([c for c in claims if c.actuator == u])
+            if b.contradiction:
+                continue
+            p = allocator.pin_w(b)
+            if p is not None and p < 0.0:
+                return          # a pack pinned to charge: never split both ways
+            if res.phase == DRAIN and b.pinned_by == "maintenance" \
+                    and p is not None and p > 0.0:
+                drains.append((u, p, allocator.hi_w(b)))
+            others += allocator.hi_w(b) if p is None else p
+        give = floor - others
+        for u, p, hi in drains:
+            if give <= 0.0:
+                break
+            up = max(p, min(hi, p + give))
+            give -= up - p
+            for i, c in enumerate(claims):
+                if c.source == "maintenance" and c.kind == SET \
+                        and c.actuator == u:
+                    claims[i] = dataclasses.replace(c, value=up)
+
     def _peak_over_hold(self, claims, maint_results):
         """SAFETY-86: a peak floor outranks a maintenance HOLD (0 W).
 
@@ -2603,7 +2658,13 @@ class Controller:
         others = 0.0
         for inv in self.inverters:
             u = self.actuator_of("maintenance", inv)
-            if u in hold or u not in live or inv.name in reserve:
+            if u in hold or u not in live:
+                continue
+            if inv.name in reserve:     # only its own pin, never both ways
+                p = self._held_pin(claims, u)
+                if p is None:
+                    return
+                others += p
                 continue
             b = pre_resolve([c for c in claims if c.actuator == u])
             if b.contradiction:

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))