Merge release/1.0 into safety-78-drain-at-two

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

goodwe_controller/DECISIONS/MAINT-05-per-pack-maintenance.md

@@ -285,32 +285,11 @@ cross-inverter `lo > hi` contradiction at `arbiter.py:114`. The only line this
 card changed there is **which** roster entry the claim is filed on: the one whose
 state machine produced the result, rather than `inverters[0]`.
 
-## 6. FOUND AND NOT FIXED: the site loop still contradicts a drain at N>=2
+## 6. FIXED BY SAFETY-78: the site loop contradicted a drain at N>=2
 
-Measured while wiring this card, and **pre-existing on `release/1.0` at
-`2d025ee`** — the same drain filed on `inverters[0]` produces the identical
-refusal with the two unit names swapped.
-
-The maintenance SET **pins** its pack. The loop's target is a **site** claim.
-When the site target is *below* the pin — a 2000 W drain while the house imports
-less than that, which is most of a night — the only split that sums to the site
-target asks the sibling to **charge** while the maintained pack **discharges**,
-and `allocator.plan()` refuses it by name as circulating power. The refusal is
-correct and fails safe: the whole site goes to 0 W with an `ARBITER: REFUSED`
-line in the action log.
-
-**The consequence an operator would see is a misdiagnosis.** The pack is held at
-0 W while its state machine believes it is draining, so after 12 h the DRAIN
-timeout fires and reports *"drain stuck"* — §12's words for a battery that is not
-following commands. The battery is fine; the site refused to allocate.
-
-**Why it is not fixed here.** The fix is to change what the site loop's target
-*means* while a unit is pinned — regulate the meter using only the unpinned
-packs — which is loop and allocator territory and changes the integrator's
-subject. MAINT-05 moves which pack is pinned and nothing else about it. It is
-pinned as a running check (`test_maintenance.py` part 12) rather than left as a
-comment, so it cannot be forgotten or quietly tuned away, and it wants a card of
-its own.
+Found here and pre-existing: below the pin the other pack had to charge, the
+allocator refused it and the site sat at 0 W. Fixed by a site `drain` floor;
+see [[SAFETY-78]] (`DECISIONS/SAFETY-78-drain-at-two.md`).
 
 ## 7. FOUND AND NOT FIXED: `migrate_unit_keys()` has the same hole, and it is a LATCH
 

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

@@ -0,0 +1,44 @@
+# SAFETY-78 - a maintenance drain at two inverters
+
+**Status: DECIDED 2026-09-29.** Fixes MAINT-05 §6 (OPT-11 P1 F1, UX-13 P3
+check 5, OPT-11 P3 F4). Tests: `test_safety78.py`.
+
+**The defect.** At N>=2 the drain pins its pack (a unit SET at
+`P_MAINTENANCE`) and the site target is split around the pin. Whenever the
+site target was below the pin, the other pack had to charge. The allocator
+refuses opposite directions (SAFETY-25 §0a, SAFETY-31), so the site went to
+0 W and logged `ARBITER: REFUSED` every cycle. That covers a house below the
+drain rate, a grid-charge trade, and a house above it too: after a refusal the
+loop restarts from 0 W and its 400 W slew step is below the pin again.
+
+**The fix.** While a pack drains at N>=2, main.py files a site limit
+`drain`, `lo` = what the pin really places (`allocator.pin_w()`: the drain
+rate inside that pack's own hard limits), at `P_MAINTENANCE`. So the site
+target is never below the pin and the other pack only adds power in the same
+direction. Opposite splits stay forbidden.
+
+- The order is the arbiter's own: a safety SET (100) outranks the limit and
+  the pin, so everything parks; the peak floor (50) and the drain floor are
+  both applied, so the higher wins; a trade or the loop below the drain is
+  lifted to it.
+- Below the drain rate the drained pack exports the rest, as the drain does at
+  N=1 (`maintenance.py`: "forced discharge, exports the surplus"). Holding it
+  to the house load would make N=2 drain differently from N=1; that is an
+  owner question, not this fix.
+- A draining pack that can place nothing (a refusal, a stale link) adds no
+  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`). 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 (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/DECISIONS/SAFETY-81-load-too-high.md

@@ -0,0 +1,63 @@
+# SAFETY-81 — no false "Load too high for battery" alarms
+
+**Status: 2026-09-29, owner go for the card. Review kept the 10 W tolerance
+as a constant and added the 10 s hold.**
+
+## Defect
+
+The SAFETY-79 test (SAFETY-79-test-report.md, F1/F2) saw 6 false alarms in 7.
+The alarm fired whenever `decide()` clamped its floor to the rating
+(`saturated`). Three things set that while the quarter landed on target:
+
+- **Double count.** `decide()` sizes from the held meter reading plus the
+  battery reading. When the meter sample predates the battery's last increase,
+  the increase is counted twice. It happens under noise, and at each quarter
+  start, where the battery is back while the held sample is still the spike.
+- **The aim, not the cap.** `saturated` is graded against cap − margin, so at
+  margin 50 noise needs 50 W less to trip it (F2).
+- **Last seconds.** The gain 900 / remaining turns a few W over the aim into
+  thousands of W wanted, with the battery far from its rating.
+
+## Decision
+
+The alarm needs `saturated` and `peakguard.load_too_high()`: with the battery
+at full power from now on, the quarter still ends more than `ALARM_MISS_W`
+(10 W) over the real limit (`ceiling_w`). That is the miss itself, without the
+gain. The battery reading it uses is the lower of this cycle's and the last
+one's. It must hold for `ALARM_HOLD_S` (10 s). At most one notification per
+quarter, because near the threshold the test flickers with the meter; the
+ERROR log line is written for every episode.
+
+**Why 10 W:** at margin 0 the guard itself lands up to +5.7 W over (SAFETY-79
+test, check 5), and those misses are real but tiny; the alarm would fire in
+the last second of most guarded quarters. The margin (default 50) is the fix
+for them. Cost: a real miss of 10 W or less is not alarmed; at a capacity
+tariff of the order of EUR 50 per kW a year, 10 W in one month is cents.
+Review measured +13 W on a 10 kW two-unit site at margin 0, not saturated, so
+no alarm: the tolerance only matters when the battery is at its rating. A
+constant, not an option: `peak_margin_w` is the owner's knob.
+
+**Why the hold (review):** the test assumes the load now lasts the rest of
+the quarter. A load over cap + battery for ~5 s every 30 s (`sine4500_fast`,
+a cycling hob or heater) alarmed at each quarter start while the quarter
+landed under the aim. The hold stops that.
+
+**Cost:** a real alarm comes 10-12 s later than on 5a4cdc2 (`step6500` at
+1211 s, not 1201 s); a slow ramp 20 s later (`ramp6500`), as it waits for the
+real cap. While the battery refuses or its reading is lost, the alarm waits
+until the quarter is lost even at full battery (27-60 s later in review runs);
+those faults have their own alarms. A load over cap + battery for more than
+10 s still alarms even if the quarter later lands under. `ALARM_HOLD_S = 0`
+removes the hold.
+
+## Not changed: the 1 s spike itself
+
+The cycle after a quarter boundary and before the new quarter's first meter
+sample still sees 0 s left, files no claim, and an Off battery drops to 0 W for
+one cycle. That is the same edge tick as OPT-15 F1, so it is left to OPT-15.
+Control is unchanged here: every command in `tools/safety81_sim.py` is
+byte-identical to 5a4cdc2 (test_safety81.py part 1).
+
+The double count also moves control, not only the alarm: on a late sample the
+guard asks for the full rating for one cycle (edge4600 at 902 s: 2000 W where
+1150 W holds the aim). Left to a follow-up card.

goodwe_controller/DOCS.md

@@ -539,6 +539,7 @@ With battery modes on, these choices replace the *Charging and selling on price*
 - The peak limit, the safety stops and the monthly cycle come before a battery's mode.
 - A grid charge is sized so the 15-minute period stays under the peak limit less its margin (`peak_margin_w`). It starts only at 500 W or more. Once the peak limit stops it, it stays off for the rest of that 15-minute period.
 - While one battery charges from the grid, the others rest, and the house uses the grid.
+- While one battery empties for its monthly cycle, no battery charges from the grid.
 - A battery that refuses to charge or give power takes no part in a run in that direction.
 
 **In Home Assistant.** With MQTT set up, each battery also gets a *Mode* list and a *Use* switch, for example `select.goodwe_inverter_a_mode` and `switch.goodwe_inverter_a_use`. They do the same as the Batteries screen. Turn battery modes off and restart, and they are removed.
@@ -1647,6 +1648,10 @@ under `peak_cap_w`, and if the forecast goes over the cap the charge-only clamp
 is dropped so the battery can shave the peak instead. Money outranks the
 maintenance schedule.
 
+With two or more batteries, the one being emptied gives its drain rate and the
+others only add power when the house needs more; none charges while it drains
+(SAFETY-78). Below the drain rate the rest is exported, as with one battery.
+
 #### Capacity-tariff exposure of the grid-following loop
 
 ⚠️ **This section described the exposure BEFORE SAFETY-07. The loop is now
@@ -1739,7 +1744,8 @@ control action and goes to the action log and the status page. A *saturated*
 guard — asking for everything the inverter has while the quarter is still
 projected over — raises one notification (`goodwe_peak_saturated`), because that
 is the one state where the money will be spent and shedding load is the only
-thing left that changes the bill.
+thing left that changes the bill. SAFETY-81 keeps a 1 s meter spike, noise and
+a quarter's last seconds from raising it: see `peakguard.load_too_high()`.
 
 **And it sizes a price-driven charge, not only a forced discharge (SAFETY-45).**
 The same arithmetic read the other way answers *"how hard may this site charge
@@ -3211,9 +3217,12 @@ reason code `OUTPUT_CLAMP`.
 **Load too high for battery.** The capacity-peak guard (SAFETY-07) asks for
 everything the inverter can discharge and the running 15-minute average is
 still projected over the peak limit, so the month's billed peak
-(capaciteitstarief) goes up. In shadow mode the guard's projection is made on a
-plant that did not respond (SAFETY-47), and the text says so. See *The
-capacity-peak guard*.
+(capaciteitstarief) goes up. It fires only when, for 10 s, the period would
+still end more than 10 W over the real limit (not the margin's aim) with the
+battery at full power. It notifies at most once per period; later overloads
+in the same period go to the log (SAFETY-81). In shadow mode the
+guard's projection is made on a plant that did not respond (SAFETY-47), and
+the text says so. See *The capacity-peak guard*.
 
 **Battery stopped limiting the peak.** The peak guard may not discharge below
 `peak_soc_floor` (label “Peak reserve”, SAFETY-24). It also stops when the

goodwe_controller/app/allocator.py

@@ -344,6 +344,16 @@ class _Narrowed:
         self.lo, self.hi = lo, hi
 
 
+def pin_w(b):
+    """The watts a unit's own SET places once held inside its hard limits,
+    as `plan()` places a standing pin, or None when the unit has no SET."""
+    pv = getattr(b, "pinned_w", None)
+    if getattr(b, "gate", None) is None or pv is None:
+        return None
+    lo, hi = _int_envelope(b)
+    return float(pv) if lo > hi else float(_clamp_int(round(float(pv)), lo, hi))
+
+
 def loop_bound(bounds, live=None):
     """OPT-11. `(lo, hi)`: what the units can place when the LOOP decides.
 
@@ -364,11 +374,8 @@ def loop_bound(bounds, live=None):
             continue
         if getattr(b, "contradiction", False):
             continue
-        pv = getattr(b, "pinned_w", None)
-        if getattr(b, "gate", None) is not None and pv is not None:
-            lo, hi = _int_envelope(b)
-            p = float(pv) if lo > hi else float(_clamp_int(round(float(pv)),
-                                                            lo, hi))
+        p = pin_w(b)
+        if p is not None:
             lo_sum += min(0.0, p)
             hi_sum += max(0.0, p)
             continue

goodwe_controller/app/arbiter.py

@@ -184,6 +184,12 @@ CLAIM_SCOPE = {
     # the banner and the claim an install has had since SAFETY-02 shipped.
     # `Controller.cycle()` does not even arm the site detector there.
     "sign_site": SCOPE_SITE,
+    # main.py: `Claim.limit("drain", P_MAINTENANCE, lo=...)`, SAFETY-78. At
+    # N>=2, while a pack's maintenance drain is pinned, the site target may
+    # not go below what that pin places: below it a sibling would have to
+    # charge against the drain (circulation, refused, 0 W). A site `lo`,
+    # never filed at N=1. DECISIONS/SAFETY-78-drain-at-two.md.
+    "drain": SCOPE_SITE,
 
     # --- UNIT: about one pack, one inverter, one device -------------------
     # main.py: `Claim.limit("device", P_SAFETY, self.dev_min, self.dev_max,

goodwe_controller/app/homeview.py

@@ -640,6 +640,14 @@ def battery_rows(c, shadow, running, held=False):
         # carries its share (SAFETY-73)?
         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. 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
+                        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.
@@ -655,7 +663,8 @@ def battery_rows(c, shadow, running, held=False):
             phase=phase, shadow=shadow, running=running, held=held,
             refused=refusing(c, inv), peak=peak,
             giving=flow(inv.batt) == "giving", other_refuses=other,
-            sf_ahead=sf_ahead, role=role, plan=plan, has_at=bool(at))
+            sf_ahead=sf_ahead, role=role, plan=plan, has_at=bool(at),
+            other_cycle=cycling)
         rows.append({
             "name": inv.name,
             "level_pct": None if inv.soc is None else float(inv.soc),
@@ -673,7 +682,8 @@ def battery_rows(c, shadow, running, held=False):
             # UX-13 P3: why it does what "now" says, from what the add-on
             # knows this cycle. Price and forecast reasons are OPT-11 P3/P4.
             "why_key": why,
-            "why": why_words(why, name=other, at=at),
+            "why": why_words(why, name=cycling if why == "other_cycle"
+                             else other, at=at),
         })
         if modes_on:
             # OPT-11 P1: the battery's switch and mode, for the Batteries
@@ -692,7 +702,7 @@ def battery_rows(c, shadow, running, held=False):
 def battery_why(*, modes_on, mode, off, reading, phase, shadow, running,
                 held, refused, peak, sf_ahead, giving=False,
                 other_refuses=None, role=None, plan=None,
-                has_at=False) -> str:
+                has_at=False, other_cycle=None) -> str:
     """The key of one battery's "why" line (`home_words.why`), most urgent
     first. Pure, so every branch is tested without a controller."""
     if not reading:
@@ -712,6 +722,8 @@ def battery_why(*, modes_on, mode, off, reading, phase, shadow, running,
         return "refuses_give" if refused > 0 else "refuses_charge"
     if peak:
         return "peak"
+    if other_cycle:
+        return "other_cycle"
     if giving and other_refuses:
         return "other_refuses"
     if not modes_on:

goodwe_controller/app/main.py

@@ -71,7 +71,7 @@ from .hass import HomeAssistant
 from .inverter import (LINK_STALE_S, UNIT_A, build_roster,
                        roster_entries, roster_names)
 from .maintenance import (
-    IDLE, POWER_MOVING, MaintConfig, Maintenance, MaintResult,
+    DRAIN, IDLE, POWER_MOVING, MaintConfig, Maintenance, MaintResult,
     arbitrate as maint_arbitrate,
 )
 from .mqtt import (MODE_CONFIGS_KEY, MODE_LABELS, USE_OFF, USE_ON,
@@ -922,6 +922,11 @@ class Controller:
             "peak_margin_w", peakguard.PEAK_OPTION_DEFAULTS["peak_margin_w"]))
         self._peak_engaged = False
         self._peak_saturated = False
+        # SAFETY-81: what the alarm is graded on, beside the guard's decision.
+        self._peak_proj = None
+        self._peak_batt_prev = None
+        self._peak_alarm_block = None
+        self._peak_high_since = None
         self._peak_floor_limited = False
         self._maint_shape_suspended = False
         # OPT-11. Battery modes, off by default so an install that has not
@@ -948,6 +953,9 @@ class Controller:
         # [(pack name, why)] for packs that are due and waiting their turn.
         # Empty at a roster of one, where nothing can defer anything.
         self._maint_deferred = []
+        # SAFETY-78: names of draining packs that cannot give power now, so
+        # the "waits" line is logged once, not every cycle.
+        self._drain_waits = set()
         # Set by amain() once the transport is built, so the status page can show
         # what only the transport knows (homewizard_local's unchanged_s). Stays
         # None when P1 is off, or under a transport that has no such counter.
@@ -2161,7 +2169,8 @@ class Controller:
             headroom_w=self.arb_headroom_w, floor_pct=floor_pct,
             ceiling_pct=ceiling_pct, house_w=house, charging_now=was,
             blocked=self._charge_blocked_q == quarter,
-            done_charge=done_c, done_sell=done_s)
+            done_charge=done_c, done_sell=done_s,
+            draining=any(r.phase == DRAIN for r in maint_results or ()))
         # A charge the guard or the headroom stopped stays off this period.
         if was and self.mode_tick.action != modetrade.CHARGE and \
                 self.mode_tick.why in ("peak", "headroom"):
@@ -2383,6 +2392,42 @@ class Controller:
         except Exception:                               # noqa: BLE001
             _LOG.exception("Home Assistant command could not be handled")
 
+    def _drain_floor(self, claims, maint_results):
+        """SAFETY-78: the site `lo` for a pinned maintenance drain, or None.
+
+        At N>=2 the drain pins its pack and the site target is split around
+        the pin. Below the pin a sibling would have to charge (opposite
+        directions, refused, 0 W, and the drain never moves). So the site
+        target may not go below what the pin places: the drained pack covers
+        the house up to its drain rate and exports the rest, as at N=1, and
+        a sibling only ever adds power in the same direction. At N=1 the
+        drain is the site SET itself, so nothing is filed.
+        A draining pack whose own limits leave it nothing to give (a refusal,
+        a stale link) adds no floor, and that is logged once.
+        """
+        if len(self.inverters) < 2:
+            return None
+        live = set(self.live_units())
+        floor, waits = 0.0, set()
+        for inv, res in zip(self.inverters, maint_results):
+            if res.phase != DRAIN or not res.owns_setpoint \
+                    or not (res.setpoint_w or 0.0) > 0.0:
+                continue
+            u = self.actuator_of("maintenance", inv)
+            b = pre_resolve([c for c in claims if c.actuator == u])
+            p = (None if u not in live or b.contradiction
+                 or b.pinned_by != "maintenance" else allocator.pin_w(b))
+            if p is not None and p > 0.0:
+                floor += p
+            else:
+                waits.add(inv.name)
+        for name in sorted(waits - self._drain_waits):
+            self.log_event(f"maintenance: {name} cannot give power now, so "
+                           f"its monthly cycle waits",
+                           hint=messages.hint("maint_waits", unit=name))
+        self._drain_waits = waits
+        return floor if floor > 0.0 else None
+
     def _maint_stop_off(self, machine, now) -> None:
         """OPT-11: stop this battery's maintenance cycle because it is off."""
         if machine.phase == IDLE:
@@ -3712,8 +3757,10 @@ class Controller:
         # is not measured on that path, and the honest report is NOT COVERED -
         # the same answer, for the same reason, that the status page already
         # gives for SAFETY-18 on a transport that cannot see `unchanged_s`.
+        self._peak_proj = (None if not self.p1_enabled
+                           else self.peak.project(self.p1.averager, now))
         self.peak_guard = None if not self.p1_enabled else peakguard.decide(
-            self.peak.project(self.p1.averager, now),
+            self._peak_proj,
             # ⚠️ THE SITE SUM. The guard sizes a floor from what the battery is
             # ALREADY removing from the meter, and the meter sees every
             # inverter. One unit's power here would size the floor from a
@@ -4091,6 +4138,13 @@ class Controller:
                                           _ctl.REFUSED,
                                           actuator=self.actuator_of("refused",
                                                                     inv)))
+        # --- SAFETY-78: a pinned drain is a floor under the site target -----
+        # After every unit claim, so the floor is what the pin really places.
+        drain = self._drain_floor(claims, maint_results)
+        if drain is not None:
+            claims.append(Claim.limit("drain", P_MAINTENANCE, lo=drain,
+                                      reason="maintenance drain",
+                                      actuator=self.actuator_of("drain")))
         # --- OPT-11 P3: the site trade, after every unit claim --------------
         # Sized from what the batteries may place with their modes and hard
         # limits (the loop bound), so none is asked for watts it may not give.
@@ -4816,22 +4870,56 @@ class Controller:
         available to it. That is the one state where the money is going to be
         spent and a human is the only remaining actuator.
 
+        SAFETY-81: `g.saturated` alone is not enough. It is graded against the
+        aim, and a 1 s meter spike, noise or the last seconds of a quarter set
+        it while the quarter lands on target; `peakguard.load_too_high()` must
+        agree too, for `ALARM_HOLD_S`. The battery reading it gets is the lower
+        of this cycle's and the last one's - see there.
+
         Edge-triggered, and de-asserting arms it again - the same shape as
         FreezeWatchdog's `evaluate()`, and for the same reason: one notification
-        per episode, not one per cycle.
+        per episode, not one per cycle. SAFETY-81: the notification and action
+        log line at most once per quarter; the ERROR line once per episode.
         """
         g = self.peak_guard
-        saturated = bool(g is not None and g.saturated)
+        batt, prev = self.site_batt_w, self._peak_batt_prev
+        if batt is None:
+            # Unknown, not "fine": the hold below neither grows nor restarts.
+            return
+        self._peak_batt_prev = batt
+        if prev is not None:
+            batt = min(float(batt), float(prev))
+        p = self._peak_proj
+        high = bool(
+            g is not None and g.saturated and p is not None
+            and p.block_start is not None
+            and peakguard.load_too_high(
+                p, batt_w=batt, ceiling_w=g.ceiling_w,
+                achievable_w=self.site_rating_w))
+        # SAFETY-81: and for ALARM_HOLD_S, on the projection's clock.
+        saturated = False
+        if high:
+            t = p.block_start + timedelta(seconds=float(p.elapsed_s))
+            if self._peak_high_since is None:
+                self._peak_high_since = t
+            saturated = ((t - self._peak_high_since).total_seconds()
+                         >= peakguard.ALARM_HOLD_S)
+        else:
+            self._peak_high_since = None
         was, self._peak_saturated = self._peak_saturated, saturated
         if not saturated or was:
             return
-        self.log_event("PEAK_GUARD_SATURATED - " + g.reason,
-                       hint=messages.hint("peak_saturated"))
         _LOG.error(
             "Switch off large loads, the battery alone cannot keep this "
             "15-minute period under the peak limit: %s.",
             messages.reason_code(g.reason))
         _LOG.debug("Evidence: %s.", g.reason)
+        # SAFETY-81: notified once per quarter; a later episode is logged only.
+        if p.block_start == self._peak_alarm_block:
+            return
+        self._peak_alarm_block = p.block_start
+        self.log_event("PEAK_GUARD_SATURATED - " + g.reason,
+                       hint=messages.hint("peak_saturated"))
         # A fixed notification_id, and a DIFFERENT one from every other alarm
         # here - check_meter_frozen() gives the reasoning. A saturated peak guard
         # is not a frozen meter and is not an out-of-range setpoint.

goodwe_controller/app/messages.py

@@ -929,6 +929,7 @@ HINTS = {
     "pack_mode": "{unit}: battery mode changed",
     "pack_sell": "{unit}: selling switched",
     "maint_off": "{unit}: maintenance stopped, the battery is switched off",
+    "maint_waits": "{unit} cannot give power now, so its monthly cycle waits",
     "mode_fault": "{unit}: battery mode ignored because of a software fault",
 }
 

goodwe_controller/app/modetrade.py

@@ -39,7 +39,9 @@ Rules that keep this safe (tests: test_modes_p3.py):
   * a battery out of a run (full, or at the floor) stays out until the run
     ends, so the ceiling and the floor do not re-arm within one run;
   * a battery that refuses a direction (SAFETY-52/73) takes no part in a
-    run in that direction: the refusal wins over the run.
+    run in that direction: the refusal wins over the run;
+  * no grid charge while another battery drains for its monthly cycle
+    (`draining`, SAFETY-78): it would charge against the drain.
 """
 
 from dataclasses import dataclass
@@ -118,7 +120,8 @@ def energy(packs, floor_pct, ceiling_pct, done_charge=(), done_sell=()):
 
 def decide(runs, now, packs, *, charge_ok, sell_ok, trade_why, price_now,
            headroom_w, floor_pct, ceiling_pct, house_w, charging_now=False,
-           blocked=False, done_charge=(), done_sell=()) -> Tick:
+           blocked=False, done_charge=(), done_sell=(),
+           draining=False) -> Tick:
     """This tick's ranges, roles and site ask.
 
     `runs` is the kept `priceruns.RunPlan` or None (no fresh prices).
@@ -129,7 +132,8 @@ def decide(runs, now, packs, *, charge_ok, sell_ok, trade_why, price_now,
     `house_w` is the measured house (grid + batteries), or None.
     `charging_now` says last tick charged (then any headroom above 0 keeps
     it going); `blocked` that the guard or the headroom stopped a charge in
-    this 15-minute period.
+    this 15-minute period. `draining` says a battery drains for its
+    monthly cycle, so none may charge from the grid (why "cycle").
     """
     plan_on = runs is not None and runs.on
     in_cheap = plan_on and runs.cheap.holds(now) and runs.charge_useful()
@@ -138,6 +142,8 @@ def decide(runs, now, packs, *, charge_ok, sell_ok, trade_why, price_now,
 
     why = None
     ch = chargers(packs, ceiling_pct, done_charge) if in_cheap else []
+    if ch and draining:
+        ch, why = [], "cycle"
     charging = False
     if ch:
         need = 0.0 if charging_now else START_W

goodwe_controller/app/peakguard.py

@@ -866,6 +866,40 @@ def decide(projection, *, batt_w, cap_w, margin_w, reading, achievable_w, soc,
                if saturated else "")))
 
 
+# SAFETY-81. A quarter must end this many W over the real limit before
+# "Load too high for battery" fires. Above the guard's own landing error at
+# margin 0 (up to +5.7 W, SAFETY-79 test check 5), which margin 0 accepts.
+ALARM_MISS_W = 10.0
+# ...and for this long without a break, so a load that peaks for a few seconds
+# (a hob or heater cycling) is not read as the rest of the quarter.
+ALARM_HOLD_S = 10.0
+
+
+def load_too_high(projection, *, batt_w, ceiling_w, achievable_w) -> bool:
+    """SAFETY-81. May "Load too high for battery" fire? Only if, with the
+    battery at full power from now on, the quarter still ends more than
+    `ALARM_MISS_W` over the real limit (`ceiling_w`, not the aim).
+
+    This is the miss itself, in W of the quarter average, so it does not have
+    `decide()`'s gain (900 / remaining): near a quarter's end that gain turns
+    a watt over the aim into thousands of W wanted.
+
+    The load is the held meter reading plus `batt_w`. The caller passes the
+    lower of this cycle's and the last cycle's battery reading, so a meter
+    sample taken before the battery's last increase is not counted twice
+    (DECISIONS/SAFETY-81-load-too-high.md). A real overload shows on the meter
+    before the battery moves, so that filter costs it no cycle.
+    """
+    remaining = QUARTER_S - float(projection.elapsed_s)
+    if batt_w is None or remaining <= 0.0:
+        return False
+    rest = float(projection.held_w) + float(batt_w) - float(achievable_w)
+    done_ws = (QUARTER_S * float(projection.projected_w)
+               - max(float(projection.held_w), 0.0) * remaining)
+    ends_w = (done_ws + max(rest, 0.0) * remaining) / QUARTER_S
+    return ends_w - float(ceiling_w) > ALARM_MISS_W
+
+
 def charge_headroom_w(decision) -> float | None:
     """SAFETY-45. The hardest this SITE may charge without putting the billed
     quarter over the ceiling, in W, positive - or None when nothing measured it.

goodwe_controller/app/ui_text/en.json

@@ -327,6 +327,7 @@
       "leftover_alone": "No Solar first battery is on, so it gets solar first.",
       "covers_house": "It gives power to cover the house.",
       "other_refuses": "{name} refuses to give power, so this battery takes over.",
+      "other_cycle": "{name} is emptying for its monthly cycle. This one does not charge.",
       "grid_charge": "Prices are low, so it charges from the grid.",
       "rests_charge": "Another battery charges from the grid, so it rests.",
       "selling": "Prices are high, so it sells to the grid.",

goodwe_controller/app/ui_text/nl.json

@@ -269,6 +269,7 @@
       "leftover_alone": "Geen batterij met Eerst zonne-energie staat aan, dus zij krijgt eerst zonne-energie.",
       "covers_house": "Ze levert stroom om het huis te dekken.",
       "other_refuses": "{name} weigert te leveren, dus deze batterij neemt het over.",
+      "other_cycle": "{name} loopt leeg voor haar maandelijkse cyclus. Deze laadt niet.",
       "grid_charge": "De prijs is laag, dus ze laadt van het net.",
       "rests_charge": "Een andere batterij laadt van het net, dus deze rust.",
       "selling": "De prijs is hoog, dus ze verkoopt aan het net.",

goodwe_controller/test_arbiter.py

@@ -149,7 +149,7 @@ check("...and `source` is still the first positional argument, which is what "
       Claim.limit("device", P_SAFETY, -1, 1).source == "device")
 
 print()
-print("ARB-03 2. the split: thirteen sources, each classified where it is BUILT")
+print("ARB-03 2. the split: fourteen sources, each classified where it is BUILT")
 # The classification is graded against what each claim is ABOUT, restated here
 # so that the table in arbiter.py and the expectation in a test are not the same
 # sentence read twice. test_peakguard.py part 10f pins this set against main.py's
@@ -171,6 +171,10 @@ SITE_SOURCES = {
                  "offending number is the SITE target, which no single "
                  "inverter owns - parking one of them would leave the rest "
                  "pushing the same way with the guard satisfied",
+    # SAFETY-78.
+    "drain":     "the site target may not go below a pinned maintenance "
+                 "drain, or a sibling would charge against it. Filed at N>1 "
+                 "only",
 }
 UNIT_SOURCES = {
     "device":     "one inverter's own reported min/max rating",
@@ -202,9 +206,9 @@ for _src, _why in sorted(SITE_SOURCES.items()):
     print(f"    site  {_src:<11} {_why}")
 for _src, _why in sorted(UNIT_SOURCES.items()):
     print(f"    unit  {_src:<11} {_why}")
-check("all THIRTEEN claim sources are classified, and only those thirteen",
+check("all FOURTEEN claim sources are classified, and only those fourteen",
       set(CLAIM_SCOPE) == set(SITE_SOURCES) | set(UNIT_SOURCES))
-check("the six that are about the GRID CONNECTION are site claims",
+check("the seven that are about the GRID CONNECTION are site claims",
       {s for s, v in CLAIM_SCOPE.items() if v == SCOPE_SITE} == set(SITE_SOURCES))
 check("the seven that are about a PACK, a DEVICE or one write are per-inverter",
       {s for s, v in CLAIM_SCOPE.items() if v == SCOPE_UNIT} == set(UNIT_SOURCES))
@@ -225,8 +229,8 @@ check("with one inverter every source routes to the site, per-inverter ones "
 check("...and with none configured at all, likewise",
       all(actuator_for(s, "inverter-a", ()) == ACTUATOR_SITE for s in CLAIM_SCOPE))
 check("with TWO the identity breaks and the split becomes real: the seven "
-      "per-inverter sources route to the unit this controller drives, the six "
-      "site sources stay on the grid connection",
+      "per-inverter sources route to the unit this controller drives, the "
+      "seven site sources stay on the grid connection",
       {s: actuator_for(s, "inverter-a", TWO) for s in CLAIM_SCOPE}
       == {**{s: "inverter-a" for s in UNIT_SOURCES},
           **{s: ACTUATOR_SITE for s in SITE_SOURCES}})

goodwe_controller/test_batteries_screen.py

@@ -93,7 +93,8 @@ WANT = {
     "off": ("solar_first", "off"),
     "off_peak": ("peak", "off_peak"),
     "refusal": ("refuses_give", "other_refuses"),
-    "maint": ("cycle_mode", "leftover"),
+    # SAFETY-78: the other battery says why it does not charge.
+    "maint": ("cycle_mode", "other_cycle"),
 }
 
 # ============================================================================ #
@@ -317,6 +318,14 @@ _cases = [
     (dict(role="charge", refused=-1), "refuses_charge"),
     (dict(role="sell", peak=True), "peak"),
     (dict(role="hold", off=True), "off"),
+    # SAFETY-78: another battery drains for its monthly cycle.
+    (dict(other_cycle="A"), "other_cycle"),
+    (dict(other_cycle="A", modes_on=False, giving=True), "other_cycle"),
+    (dict(other_cycle="A", mode=packmode.CHEAP_HOURS, role="charge"),
+     "other_cycle"),
+    (dict(other_cycle="A", peak=True), "peak"),
+    (dict(other_cycle="A", refused=-1), "refuses_charge"),
+    (dict(other_cycle="A", off=True), "off"),
 ]
 _wrong = [(kw, want, why(**kw)) for kw, want in _cases if why(**kw) != want]
 print(f"    {_wrong}")

goodwe_controller/test_maintenance.py

@@ -1363,50 +1363,24 @@ check("...while the second pack's write reaches only its own namespace",
 
 # --------------------------------------------------------------------------- #
 print()
-print("12. RECORDED, NOT FIXED: the site loop still contradicts a drain at N>=2")
+print("12. FIXED (SAFETY-78): a drain at N>=2 no longer parks the site")
 
-# ⚠️ THIS IS A DEFECT THIS CARD FOUND AND DELIBERATELY DID NOT FIX, AND IT IS
-# WRITTEN DOWN AS A RUNNING CHECK RATHER THAN AS A COMMENT SO IT CANNOT BE
-# FORGOTTEN OR QUIETLY "FIXED" BY A TUNING CHANGE.
-#
-# WHAT IT IS. The maintenance SET is a per-unit claim and PINS its pack. The
-# loop's target is a SITE claim. When the site target is BELOW the pin - a
-# drain of 2000 W while the house is importing less than that, which is most of
-# a night - the only split that sums to the site target asks the sibling to
-# CHARGE while the maintained pack DISCHARGES, and `allocator.plan()` refuses it
-# by name as circulating power. The refusal is correct and fails safe: the whole
-# site goes to 0 W with an ARBITER line in the action log.
-#
-# WHY IT IS NOT THIS CARD'S. It is PRE-EXISTING - measured on release/1.0 at
-# 2d025ee, where the same drain filed on `inverters[0]` produces the identical
-# refusal with the two unit names swapped - and fixing it means changing what
-# the site loop's target MEANS while a unit is pinned (regulate the meter using
-# only the unpinned packs), which is loop and allocator territory, not
-# maintenance's. MAINT-05 moves which pack is pinned and nothing else about it.
-#
-# ⚠️ AND THE CONSEQUENCE AN OPERATOR WOULD SEE IS A MISDIAGNOSIS, WHICH IS WHY
-# IT MATTERS ENOUGH TO PIN. The pack is held at 0 W while its FSM believes it is
-# draining, so after 12 h the DRAIN timeout fires and reports "drain stuck" -
-# §12's words for "the battery is not following commands, a fault". The battery
-# is fine; the site refused to allocate. That is the same class of wrong story
-# this card's own notes warn about for the deferral.
+# This section used to pin the defect: with the house below the drain, the
+# only split summing to the site target asked the sibling to CHARGE, the
+# allocator refused it and the site sat at 0 W while the FSM believed it was
+# draining. SAFETY-78 files a site `drain` floor at the pin, so the sibling
+# only ever adds power the same way. test_safety78.py grades it in full.
 _cX, _ = controller(TWO_OPTS, grid=0.0)
 _cX.maints[1].force_start(_now)
 one_cycle(_cX)
 print(f"    house at 0 W, B draining 2000 W -> reason {_cX.reason!r}; "
       f"{_cX.allocation.explain()}")
-check("a maintenance drain BELOW the site target is refused by the allocator "
-      "and the site parks at 0 W - recorded here so the next card can find it, "
-      "not asserted as desirable",
-      _cX.target == 0.0 and _cX.allocation.refused
-      and "circulating power" in _cX.allocation.explain())
-check("...and the pack that is supposedly draining is commanded 0 W, so its "
-      "12 h DRAIN timeout would eventually report 'drain stuck' - a fault "
-      "message about a healthy battery",
-      _cX.maints[1].phase == DRAIN and _cX.inverters[1].target == 0.0)
-check("...while the SAME site with the house importing more than the drain "
-      "allocates cleanly, which is what makes this a CONDITION rather than "
-      "'maintenance does not work at two inverters'",
+check("a maintenance drain above the loop's site target is allocated, not "
+      "refused: B drains at its rate and A is held at 0 W",
+      _cX.allocation.refused is None and _cX.target == 2000.0
+      and _cX.inverters[1].target == 2000.0 and _cX.inverters[0].target == 0.0)
+check("...and the SAME site with the house importing more than the drain "
+      "still allocates cleanly",
       _cB.allocation.refused is None and _cB.inverters[1].target == 2000.0)
 
 

goodwe_controller/test_modes.py

@@ -112,6 +112,11 @@ try:
 finally:
     _undo_c()
     _undo_b()
+# SAFETY-78 changes n2_maint ON PURPOSE: on a5dddad and 5a4cdc2 its N=2 drain
+# was refused on 392 of 540 cycles (site at 0 W); now none is. Printed on the
+# SAFETY-78 tree the same way. Every other digest is unchanged.
+SAFETY78_N2_MAINT = \
+    "cd4d9cb5b7e29ff4ab4fe65dab5106683d8a599f09323fbdee4018ad9b28fcd1"
 for k in A5DDDAD:
     print(f"    {k:7} {'same' if _now[k] == A5DDDAD[k] else 'DIFFERS'} "
           f"{_now[k][:16]}")
@@ -119,8 +124,13 @@ check("⭐ at N=1 and N=2, with the option absent and with it false, every "
       "command, reason and allocation record over 540 cycles each is "
       "byte-identical to a5dddad (house steps, a solar surplus, a battery "
       "at its cut-out level, at N=2 a refusal with probes, both batteries "
-      "due for maintenance), and so is the arbiter on 6000 claim lists",
-      _now == A5DDDAD)
+      "due for maintenance; n2_maint apart, below), and so is the arbiter "
+      "on 6000 claim lists",
+      {k: v for k, v in _now.items() if k != "n2_maint"}
+      == {k: v for k, v in A5DDDAD.items() if k != "n2_maint"})
+check("...and n2_maint is the SAFETY-78 digest: the N=2 drain that a5dddad "
+      "refused on 392 of 540 cycles now runs (test_safety78.py)",
+      _now["n2_maint"] == SAFETY78_N2_MAINT != A5DDDAD["n2_maint"])
 check("...and no cycle in those runs filed a `mode` claim or handed "
       "`compute()` a bound",
       _claims and not any(c.source == "mode" for cl in _claims for c in cl)

goodwe_controller/test_peakguard.py

@@ -1259,6 +1259,13 @@ check(f"the saturating fixture ({NET2:.0f} W) is a load the real ingest "
       f"accepts, not one it rejects as implausible", NET2 < PLAUSIBLE_MAX_W)
 print(f"    saturated case: {c2.peak_guard.explain()}")
 check("a load larger than the inverter saturates the guard", c2.peak_guard.saturated)
+# SAFETY-81 review: the alarm must hold ALARM_HOLD_S first. The fixture runs on
+# the wall clock, so the start of the hold is moved back instead of waited out.
+check("...but not on its first cycle: it must hold peakguard.ALARM_HOLD_S "
+      "(SAFETY-81)", hass2.notifications.count("goodwe_peak_saturated") == 0
+      and c2._peak_high_since is not None)
+c2._peak_high_since -= timedelta(seconds=peakguard.ALARM_HOLD_S)
+asyncio.get_event_loop_policy().new_event_loop().run_until_complete(c2.cycle())
 check("...and THAT raises exactly one notification, with its own id - the one "
       "state where the money will be spent and a human is the only remaining "
       "actuator", hass2.notifications.count("goodwe_peak_saturated") == 1)
@@ -2674,9 +2681,15 @@ check("...and so does the persistent notification, which outlives the page and "
 # cycles with the floor, and part 10 at N=1. The sweeps below never reach it:
 # their fixtures leave battery modes off, so they still see {device,
 # supervised, loop}. DECISIONS/OPT-11-battery-modes.md.
+# ⚠️ SAFETY-78 ADDED `drain`, AND IT IS NOT THE DEFERRED RESERVE. It is a
+# site `lo` at P_MAINTENANCE, filed only at N>=2 while a pack's maintenance
+# drain is pinned, equal to what that pin places. It is not conditioned on a
+# state of charge, it never closes discharge (a `lo`, not a `hi`), and it is
+# never filed at N=1, so the sweeps below still see {device, supervised,
+# loop}. DECISIONS/SAFETY-78-drain-at-two.md.
 CLAIM_SOURCES = {"device", "supervised", "failsafe", "safety", "peak",
                  "arbitrage", "maintenance", "loop", "circulation",
-                 "sign_site", "pack", "refused", "mode"}
+                 "sign_site", "pack", "refused", "mode", "drain"}
 # The two appends that file a claim BUILT ELSEWHERE in the method. Named here so
 # that a THIRD such indirection reads as an unrecognised append rather than as a
 # silent gap in the scan.
@@ -2718,7 +2731,7 @@ check("every `claims.append(...)` in main.py's cycle is one this scan can READ "
       "- a claim filed through a shape it cannot parse has to fail here rather "
       "than be counted as nothing, which is how a scan like this goes quietly "
       "vacuous", len(_appends) >= 10 and None not in _filed)
-check("...and the sources that cycle can file are EXACTLY these THIRTEEN. THIS "
+check("...and the sources that cycle can file are EXACTLY these FOURTEEN. THIS "
       "is what makes part 10 a tripwire: replay_loop() hand-builds the claim "
       "list, so it cannot see a claim main.py adds - and a loop-side SoC "
       "reserve is a new source. Filing one takes this line red on the line it "
@@ -2858,7 +2871,7 @@ print("    site: " + ", ".join(sorted(s for s, v in CLAIM_SCOPE.items()
       + "  |  per-inverter: "
       + ", ".join(sorted(s for s, v in CLAIM_SCOPE.items() if v == SCOPE_UNIT)))
 check("...and the split is the one ARB-03 argued, with SAFETY-31's and "
-      "SAFETY-30's and OPT-09's additions: the six about the GRID CONNECTION "
+      "SAFETY-30's, OPT-09's and SAFETY-78's additions: the seven about the GRID CONNECTION "
       "are site claims and the five about a pack, a device or one write are "
       "per-inverter. `circulation` is a site claim because its quantity is "
       "defined over the roster and is identically zero for any one unit; "
@@ -2867,7 +2880,8 @@ check("...and the split is the one ARB-03 argued, with SAFETY-31's and "
       "per-inverter because since SAFETY-30 it grades one unit's command "
       "against that unit's OWN measured power",
       {s for s, v in CLAIM_SCOPE.items() if v == SCOPE_SITE}
-      == {"loop", "peak", "arbitrage", "safety", "circulation", "sign_site"}
+      == {"loop", "peak", "arbitrage", "safety", "circulation", "sign_site",
+          "drain"}
       and {s for s, v in CLAIM_SCOPE.items() if v == SCOPE_UNIT}
       == {"device", "supervised", "failsafe", "maintenance", "pack",
           "refused", "mode"})
@@ -2971,7 +2985,8 @@ check("⚠️ AND THIS IS THE BREAKING HALF. Give the same Controller a second "
                   "mode": "inverter-a",
                   "loop": ACTUATOR_SITE, "peak": ACTUATOR_SITE,
                   "arbitrage": ACTUATOR_SITE, "safety": ACTUATOR_SITE,
-                  "circulation": ACTUATOR_SITE, "sign_site": ACTUATOR_SITE})
+                  "circulation": ACTUATOR_SITE, "sign_site": ACTUATOR_SITE,
+                  "drain": ACTUATOR_SITE})
 check("...and the roster it was restored to is the shipped one, so nothing "
       "below inherits a two-inverter Controller",
       c_src.units == (app_main.UNIT_A,) and len(c_src.inverters) == 1)
@@ -3050,13 +3065,13 @@ _unit_sources_unrouted = [
     if CLAIM_SCOPE.get(r[0]) == SCOPE_UNIT and "," not in r[1]]
 print(f"    {len(_direct)} direct appends, each read at its own line: "
       + ", ".join(f"main.py:{ln} {s} -> {a}" for s, a, ln in _routing))
-check("all FOURTEEN of `Controller.cycle()`'s DIRECT appends route through "
+check("all FIFTEEN of `Controller.cycle()`'s DIRECT appends route through "
       "`self.actuator_of(<string literal>)`, so this scan can read the routing "
       "of EVERY one of them. An append that computed its actuator, or omitted "
       "the argument, or hand-wrote a string, fails HERE rather than being "
       "silently skipped - which is how a scan like this goes quietly vacuous, "
       "and 10f is in this file because that already happened once",
-      len(_direct) == 14
+      len(_direct) == 15
       and all(a is not None and a.startswith("self.actuator_of('")
               for _s, a, _ln in _routing))
 check("⚠️ AND AT EVERY ONE OF THE ELEVEN THE LITERAL IT ROUTES THROUGH IS THE "
@@ -3073,7 +3088,7 @@ check("⚠️ AND EVERY PER-INVERTER APPEND NAMES THE INVERTER IT IS ABOUT. "
       "site-scoped sources and is the whole defect for a per-inverter one: a "
       "`device` limit built inside the roster loop and routed without `inv` "
       "files inverter B's rating on inverter A, where it binds the wrong "
-      "share. Nine of the fourteen appends are SCOPE_UNIT - `device`, "
+      "share. Nine of the fifteen appends are SCOPE_UNIT - `device`, "
       "`supervised`, `failsafe`, maintenance's two, OPT-09's `pack`, "
       "SAFETY-73's two `refused` and OPT-11's `mode` - and all nine pass one. "
       "SAFETY-30's `sign_site` is deliberately NOT one of them: it is filed on "
@@ -3090,8 +3105,8 @@ check("...and the only appends this pin does NOT reach are the two that file a "
       "arbitrage.claim() by test_arbitrage.py",
       sorted(_indirect) == sorted(CLAIM_BUILDERS)
       and len(_direct) + len(_indirect) == len(_appends))
-check("...and between them the sixteen appends account for exactly the "
-      "thirteen "
+check("...and between them the seventeen appends account for exactly the "
+      "fourteen "
       "sources 10f pins, so the routing pin and the source set are one scan "
       "and neither can drift from the other",
       {s for s, _a, _ln in _routing} | set(CLAIM_BUILDERS.values())
@@ -3152,7 +3167,7 @@ check("...and naming those three is not an accident of this fixture but the "
       {c.source for c in _cl2}
       == CLAIM_SOURCES - {"failsafe", "safety", "maintenance", "peak",
                           "arbitrage", "circulation", "sign_site",
-                          "refused", "mode"})
+                          "refused", "mode", "drain"})
 # ⚠️ AND NOW THE HALF THAT WAS EVIDENCE-SHAPED AND OBSERVED NOTHING. The three
 # checks above grade WHICH ACTUATOR each claim lands on. They say nothing about
 # WHOSE NUMBERS the bound is built from - and those are two different defects

goodwe_controller/test_safety78.py

@@ -0,0 +1,424 @@
+"""Runnable check for SAFETY-78: a maintenance drain at N=2 no longer stalls.
+`python3 test_safety78.py`
+
+At two inverters the drain pins its battery at the drain rate and the site
+target is split around the pin. Below the pin the other battery had to charge,
+the allocator refused the opposite directions, the site sat at 0 W and the
+drain never moved (OPT-11 P1 F1, UX-13 P3 check 5, OPT-11 P3 F4). Now a site
+`lo` (`drain`, P_MAINTENANCE) keeps the site target at or above what the pin
+places. Drives the REAL Controller (tools/opt11_identity.py's harness).
+DECISIONS/SAFETY-78-drain-at-two.md has the record.
+
+1. N=2, house below, at and above the drain rate, a solar surplus: the drain
+   runs at its rate, no opposite split, no refusal, no log spam, and it ends.
+2. A drain that cannot give power waits with ONE log line.
+3. Battery modes on, a Cheap hours battery in a cheap run during the drain:
+   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.
+"""
+
+import functools
+import logging
+import os
+import random
+import sys
+import tempfile
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, HERE)
+sys.path.insert(0, os.path.join(HERE, "tools"))
+logging.disable(logging.WARNING)
+
+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 allocator, arbitrage, homeview, packmode       # noqa: E402
+from app.arbiter import (Claim, P_ARBITRAGE, P_LOOP, P_MAINTENANCE,  # noqa
+                         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
+
+fails = []
+
+
+def check(name, cond):
+    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
+    if not cond:
+        fails.append(name)
+
+
+M = {"maintenance_enabled": True, "maintenance_start_hour": 11}
+A, B = "inverter-a", "inverter-b"
+
+
+def cmd(r):
+    return dict(r[2])
+
+
+def refused(rows):
+    return sum(1 for r in rows if r[7] and "REFUSED" in r[7])
+
+
+def opposite(rows):
+    return sum(1 for r in rows if cmd(r)["a"] * cmd(r)["b"] < 0)
+
+
+def drain_run(house, dur=600.0, soc=lambda u, t: 50.0, cut=None, **kw):
+    return rig.run(2, extra=dict(M, **kw.pop("extra", {})), dur=dur,
+                   load=lambda t: house(t) if callable(house) else house,
+                   soc=soc, cut=cut or (lambda u, t: False), **kw)
+
+
+def with_peak_db(fn):
+    """Run `fn()` with the Controller's peak store in a temporary file."""
+    db = os.path.join(tempfile.mkdtemp(prefix="s78t-"), "tariff.db")
+    real = rig.Controller
+    rig.Controller = functools.partial(real, peak_db_path=db)
+    try:
+        return fn()
+    finally:
+        rig.Controller = real
+
+
+# ============================================================================ #
+print("1. N=2: the drain runs whatever the house uses")
+RUNS = {h: drain_run(h) for h in (1200.0, 2000.0, 3000.0, -1500.0)}
+for h, (rows, c) in RUNS.items():
+    print(f"    house {h:+.0f} W: last A {cmd(rows[-1])['a']:+.0f} W, B "
+          f"{cmd(rows[-1])['b']:+.0f} W, grid {h - sum(cmd(rows[-1]).values()):+.0f}"
+          f" W; {rows[-1][6]}; refusals {refused(rows)}")
+check("⭐ below, at and above the drain rate and under a solar surplus, "
+      "battery A is commanded its 2000 W drain on EVERY cycle, so the drain "
+      "moves (it used to sit at 0 W from the first cycle)",
+      all(cmd(r)["a"] == 2000.0 for rows, _c in RUNS.values() for r in rows)
+      and all(c.maints[0].phase == DRAIN for _r, c in RUNS.values()))
+check("⭐ no cycle splits the site both ways and the allocator refuses "
+      "nothing: battery B never charges while A drains",
+      all(opposite(rows) == 0 and refused(rows) == 0
+          and all(cmd(r)["b"] >= 0.0 for r in rows)
+          for rows, _c in RUNS.values()))
+check("...and nothing is logged per cycle: no ARBITER line in the action log",
+      not any("ARBITER" in e for _r, c in RUNS.values() for e in c.events))
+_r3 = RUNS[3000.0][0]
+check("above the drain rate, B covers the rest of the house in the SAME "
+      "direction and the grid settles at the loop's aim",
+      990.0 <= cmd(_r3[-1])["b"] <= 1030.0
+      and abs(3000.0 - sum(cmd(_r3[-1]).values()) + 10.0) <= 30.0)
+check("at or below the drain rate, B is held at 0 W and A exports the rest, "
+      "as the drain does at one inverter ('exports the surplus')",
+      all(cmd(RUNS[h][0][-1])["b"] == 0.0 for h in (1200.0, 2000.0, -1500.0))
+      and all(RUNS[h][0][-1][6].startswith("loop -> 2000 W")
+              for h in (1200.0, 2000.0, -1500.0)))
+check("...and where the loop alone would ask less, the Why names the drain",
+      "limited by drain" in RUNS[1200.0][0][-1][6])
+
+# The drain ends: A's level falls to the floor, and the cycle moves on.
+_end_rows, _end = drain_run(1200.0, dur=500.0,
+                            soc=lambda u, t: max(5.0, 50.0 - t / 10.0)
+                            if u == "a" else 50.0)
+check("⭐ and the drain ENDS: A reaches the 11 % floor and the cycle moves "
+      "to its charge phase, with no refusal on the way",
+      _end.maints[0].phase == CHARGE and refused(_end_rows) == 0)
+
+# The check bites: without the floor, the same site stalls as it did.
+_real = Controller._drain_floor
+Controller._drain_floor = lambda self, claims, results: None
+try:
+    _bite, _bc = drain_run(1200.0, dur=120.0)
+finally:
+    Controller._drain_floor = _real
+print(f"    without the drain floor: refusals {refused(_bite)}/{len(_bite)}, "
+      f"last A {cmd(_bite[-1])['a']:+.0f} W")
+check("non-vacuous: with the floor taken out, the same fixture refuses every "
+      "cycle and parks A at 0 W (the reported defect)",
+      refused(_bite) == len(_bite) and cmd(_bite[-1])["a"] == 0.0
+      and any("ARBITER" in e for e in _bc.events))
+
+# ============================================================================ #
+print("\n2. a drain that cannot give power waits, with one log line")
+_w_rows, _w = drain_run(1500.0, cut=lambda u, t: u == "a")
+_waits = [e for e in _w.events if "cannot give power now" in e]
+print(f"    A refuses to give: {_waits}; last {_w_rows[-1][2]}")
+check("⭐ A refuses to give power (SAFETY-73): the cycle waits, the line is "
+      "logged ONCE over 120 cycles, and nothing is refused",
+      len(_waits) == 1 and _w.inverters[0].refused == 1
+      and refused(_w_rows) == 0 and _w.maints[0].phase == DRAIN)
+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)."""
+    if not getattr(c.hass, "_rej", False):
+        real = c.hass.set_number
+
+        async def set_number(e, v):
+            return False if e == "number.sp_a" else await real(e, v)
+        c.hass.set_number, c.hass._rej = set_number, True
+
+
+_l_rows, _l = drain_run(1500.0, dur=400.0, before=reject_a)
+print(f"    A's writes rejected: {_l_rows[-1][2]}, {_l_rows[-1][7][-40:]}")
+check("⭐ a draining battery that is dropped (stale link) adds no floor: B "
+      "covers the 1500 W house, not the 2000 W drain rate, and the wait is "
+      "logged once",
+      "dropped inverter-a" in _l_rows[-1][7]
+      and cmd(_l_rows[-1])["b"] == 1500.0
+      and sum("cannot give power now" in e for e in _l.events) == 1)
+
+# ============================================================================ #
+print("\n3. battery modes on: no grid charge during the drain")
+
+
+def cheap_b():
+    s = rig._Store()
+    s.data[unit_key(A, packmode.MODE_KEY)] = packmode.SOLAR_FIRST
+    s.data[unit_key(B, packmode.MODE_KEY)] = packmode.CHEAP_HOURS
+    return s
+
+
+def modes_run(maint):
+    extra = dict(s79.P1, **p3.ARB, battery_modes_enabled=True,
+                 **(M if maint else {}))
+    ticks = []
+
+    def on_cycle(c, t):
+        ticks.append(c.mode_tick)
+    rows, c = with_peak_db(lambda: rig.run(
+        2, extra=extra, dur=1800.0, load=lambda t: 800.0,
+        soc=lambda u, t: 50.0, cut=lambda u, t: False, before=p3._feed,
+        on_cycle=on_cycle, store=cheap_b()))
+    cheap = [(r, k) for r, k in zip(rows, ticks) if 900.0 <= r[0] < 1800.0]
+    return rows, c, cheap
+
+
+_base_rows, _base_c, _base_cheap = modes_run(False)
+_m_rows, _m_c, _m_cheap = modes_run(True)
+check("baseline: with no drain, B (Cheap hours only) charges from the grid "
+      "in the cheap run, so the fixture reaches the charge",
+      any(cmd(r)["b"] < 0.0 for r, _k in _base_cheap)
+      and all(k.roles[B] == "charge" for _r, k in _base_cheap))
+check("⭐ with A draining, B never charges in that run: its role is not "
+      "'charge', no trade is filed, and the modes say why ('cycle')",
+      all(cmd(r)["b"] >= 0.0 and k.roles[B] != "charge" and k.action is None
+          and k.why == "cycle" for r, k in _m_cheap)
+      and _m_c.maints[0].phase == DRAIN)
+check("...and the drain runs through the run: A at 2000 W, nothing refused",
+      all(cmd(r)["a"] == 2000.0 for r in _m_rows) and refused(_m_rows) == 0)
+
+# Battery modes off: the site trade (`arbitrage.decide()`) buys in the cheap
+# period. It is a site SET at P_ARBITRAGE, below the drain floor.
+_t_trades = []
+_t_rows, _t_c = with_peak_db(lambda: rig.run(
+    2, extra=dict(s79.P1, **p3.ARB, **M), dur=1800.0, load=lambda t: 800.0,
+    soc=lambda u, t: 50.0, cut=lambda u, t: False, before=p3._feed,
+    on_cycle=lambda c, t: _t_trades.append(c.arb and c.arb.action)))
+print(f"    modes off, trades seen: {sorted({str(a) for a in _t_trades})}")
+check("⭐ battery modes off: the site trade still asks to buy in the cheap "
+      "period, and the drain floor keeps it from charging B against the drain: "
+      "A drains at 2000 W, B never charges, nothing is refused",
+      arbitrage.CHARGE in _t_trades
+      and all(cmd(r)["a"] == 2000.0 and cmd(r)["b"] >= 0.0 for r in _t_rows)
+      and refused(_t_rows) == 0)
+_cards = homeview.battery_rows(_m_c, False, True)
+print(f"    cards: {[(b['name'], b['now'], b['why']) for b in _cards]}")
+check("⭐ B's card tells the truth: it names the draining battery and says "
+      "it does not charge (it said 'charges from the grid' while resting)",
+      _cards[1]["why_key"] == "other_cycle" and A in _cards[1]["why"]
+      and _cards[0]["why_key"] == "cycle_mode")
+
+# ============================================================================ #
+print("\n4. the peak guard during a drain, and the order of the claims")
+
+
+def off_b():
+    s = rig._Store()
+    s.data[unit_key(B, packmode.USE_KEY)] = False
+    return s
+
+
+_p_rows, _p = with_peak_db(lambda: rig.run(
+    2, extra=dict(s79.P1, battery_modes_enabled=True, **M), dur=600.0,
+    load=lambda t: 6000.0, soc=lambda u, t: 60.0, cut=lambda u, t: False,
+    before=s79._feed, store=off_b()))
+print(f"    house 6000 W, limit {s79.CAP_W} W, B Off: {_p_rows[-1][2]}, "
+      f"{_p_rows[-1][6]}")
+check("⭐ the peak guard binds during the drain: the site goes above the "
+      "drain to the guard's floor, A keeps draining at 2000 W and B (Off) "
+      "gives only the shortfall, in the same direction",
+      _p.peak_guard.engaged and _p_rows[-1][1] >= _p.peak_guard.floor_w > 2000.0
+      and all(cmd(r)["a"] == 2000.0 and cmd(r)["b"] >= 0.0 for r in _p_rows)
+      and "limited by peak" in _p_rows[-1][6] and refused(_p_rows) == 0)
+
+
+
+def blind_at(t0):
+    """The site meter stops answering at `t0`: a safety SET at P_SAFETY."""
+    def before(c, t):
+        if t >= t0 and not getattr(c.hass, "_blind", False):
+            real = c.hass.number
+
+            async def number(e, invert=False):
+                return None if e == "sensor.grid" else await real(e, invert)
+            c.hass.number, c.hass._blind = number, True
+    return before
+
+
+_s_rows, _s = drain_run(3000.0, dur=300.0, before=blind_at(150.0))
+print(f"    meter lost at 150 s: {_s_rows[-1][2]}, {_s_rows[-1][6]}")
+check("⭐ a safety stop outranks the drain at N=2: with the meter gone the "
+      "safety SET parks both batteries at 0 W, and the drain ran before",
+      cmd(_s_rows[20])["a"] == 2000.0
+      and all(cmd(r) == {"a": 0.0, "b": 0.0} for r in _s_rows if r[0] >= 175.0)
+      and _s_rows[-1][6].startswith("safety -> 0 W"))
+
+UNITS = (A, B)
+_drain = [Claim.set("maintenance", P_MAINTENANCE, 2000.0, "drain", actuator=A),
+          Claim.limit("drain", P_MAINTENANCE, lo=2000.0, reason="drain")]
+_order = {
+    "loop under the drain": ([Claim.set("loop", P_LOOP, 500.0)], 2000.0),
+    "a grid charge (trade)": ([Claim.set("arbitrage", P_ARBITRAGE, -2500.0)],
+                              2000.0),
+    "a sale bigger than the drain": ([Claim.set("arbitrage", P_ARBITRAGE,
+                                                3000.0)], 3000.0),
+    "a peak floor above the drain": ([Claim.set("loop", P_LOOP, 500.0),
+                                      Claim.limit("peak", P_PEAK, lo=3500.0)],
+                                     3500.0),
+    "a safety stop": ([Claim.set("loop", P_LOOP, 500.0),
+                       Claim.set("safety", P_SAFETY, 0.0, "stopped")], 0.0),
+}
+_got = {}
+for k, (extra, want) in _order.items():
+    r = resolve(_drain + extra, units=UNITS)
+    _got[k] = (r.target_w, dict(r.shares), r.allocated)
+    print(f"    {k}: target {r.target_w:+.0f} W, shares {dict(r.shares)}")
+check("⭐ the order holds: a safety stop parks everything; the peak floor "
+      "above the drain wins; a sale above it runs; a grid charge and the "
+      "loop below it are lifted to it; never an opposite split",
+      all(_got[k][0] == want and _got[k][2]
+          and not (min(_got[k][1].values()) < 0 < max(_got[k][1].values()))
+          for k, (_e, want) in _order.items())
+      and _got["a safety stop"][1] == {A: 0, B: 0}
+      and _got["a grid charge (trade)"][1] == {A: 2000, B: 0})
+
+# ============================================================================ #
+print("\n5. N=1: byte-identical to release/1.0 5a4cdc2")
+# Printed by `python3 tools/safety78_identity.py` on release/1.0 5a4cdc2 (the
+# same file copied in, nothing else changed), goodwe-addon-checks:1,
+# 2026-09-29. 480 cycles each, draining throughout, every claim and the
+# newest action-log line hashed.
+AT_5A4CDC2 = {
+    "n1_drain":
+        "55fcfc4a0b9251746f315cfdea607a33640d3a81faf6604c4718a04bbc7dd776",
+    "n1_drain_modes":
+        "93867eb4ee4132b47c030668bd7f07587a353bced2461524e09421125c3c4edf",
+}
+_now = s78.digests()
+for k in AT_5A4CDC2:
+    print(f"    {k} {'same' if _now[k] == AT_5A4CDC2[k] else 'DIFFERS'} "
+          f"{_now[k][:16]}")
+check("⭐ at one inverter a drain under a house below, at and above its "
+      "rate, a surplus and a load over the peak limit, with battery modes "
+      "off and on, is byte-identical to 5a4cdc2: commands, reasons, roles, "
+      "trades and every claim",
+      _now == AT_5A4CDC2)
+Controller._drain_floor = lambda self, claims, results: (
+    2000.0 if any(r.phase == DRAIN for r in results) else None)
+try:
+    _leak = s78.digest(s78.run(**s78.SCENARIOS["n1_drain"], dur=60.0)[0])
+    _same = s78.digest(s78.run(**s78.SCENARIOS["n1_drain"], dur=60.0)[0])
+finally:
+    Controller._drain_floor = _real
+_ref = s78.digest(s78.run(**s78.SCENARIOS["n1_drain"], dur=60.0)[0])
+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)}")
+    sys.exit(1)
+print("all checks passed")

goodwe_controller/test_safety81.py

@@ -0,0 +1,171 @@
+"""Runnable check for SAFETY-81, no false "Load too high for battery" alarms.
+`python3 test_safety81.py`
+
+Drives the REAL Controller through tools/safety81_sim.py (a 1 s cycle on a
+plant shaped like the SAFETY-79 rig, one sample in ten late).
+DECISIONS/SAFETY-81-load-too-high.md has the record.
+
+1. Every command is byte-identical to release/1.0 5a4cdc2.
+2. No alarm on the quarter-edge spike, the last seconds, noise (margin 50
+   and 0) or a burst the quarter absorbs.
+3. Real overloads still alarm, ALARM_HOLD_S after 5a4cdc2.
+4. peakguard.load_too_high() on its own.
+5. The hold across a cycle with no battery reading (review).
+
+Mutations (MUTATION.md, purged copies) are listed in the pull request.
+"""
+
+import asyncio
+import os
+import sys
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, os.path.join(HERE, "tools"))
+
+import safety81_sim as sim                                    # noqa: E402
+from app.main import Controller                               # noqa: E402
+from app.peakguard import load_too_high                       # noqa: E402
+
+fails = []
+
+
+def check(name, cond):
+    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
+    if not cond:
+        fails.append(name)
+
+
+# Printed by `python3 tools/safety81_sim.py` on release/1.0 5a4cdc2 (the tool
+# copied in, nothing else changed), in the CI image: (alarm times, digest).
+BASE = {
+    "const4500": ([2699.0], "6ba6b8634f0df9526adc04f03c82db7d083973268deda0d1f07c5519bc2e5e9d"),
+    "const4500_m0": ([2699.0], "2bede28f3ea6ba9294be134c9b9733e43d22a7c24deac3bf8210bbbbe09b2039"),
+    "sine4500": ([2699.0], "bf70c2c23a0eede40c34478d1354b7daa1858e0a0f920693e2dd15fb5046120f"),
+    "edge4600": ([2.0, 902.0, 1802.0, 2699.0], "3cc0c6fec0c70f138530955babc2b416fe7307373037c1b2e927b6ccdaad3a4d"),
+    "edge4600_m0": ([2.0, 902.0, 1802.0, 2699.0], "ba9ba2c042ee42b043315b996ece38b3478dc9866aaf833256c9c29c02f0fe9f"),
+    "burst_end": ([881.0, 1781.0, 2681.0], "5f70e1eb4244820f1d44b494cb80e51f1a4486dee03357c9a8e8f7d371c5d9af"),
+    "noise5000_m50": ([292.0, 2551.0, 2698.0], "48b446e1a78e6d6032b31cb68a4eadb0f74e3aec4cc1e184162e9c65449bfdc5"),
+    "noise5000_m0": ([292.0, 2552.0, 2698.0], "1a0527dee6c2b03dc95a90b4682858c7bf5edf7205c2aced249112f04328c960"),
+    "step6500": ([1201.0, 1801.0], "95e5c90e4a6814526cb4f3bea76286063b449d86a420f888c014531fb6e913a3"),
+    "miss5560": ([1201.0, 1801.0], "cd991cfe1685750598236701a5a2f2f0342d66645609c536d42ab7ba07216e6c"),
+    "ramp6500": ([1296.0, 1801.0], "8db7675535f1a01abf38bd6bfb996a6fe57711d3ecca53541bb1c0870748c46b"),
+    "step7500_reserve": ([], "21eb9c07a4582f47f347084d1c75a28a5d767bc068bc3f74f32433d3239e9ccb"),
+    # Added in review, printed the same way.
+    "sine4500_fast": (["112, about one per 30 s"], "710008fd8f2d33a0e2cbb9a626bedafd6d87bc23712bbc5b74a4af9b3ec21897"),
+    "twice7000": ([1001.0, 1502.0, 2699.0], "1cddda8cf35859efb2ad503f1683f744cb812a6325a390c4fa3df19d717af043"),
+    "step6500_gap": ([1201.0, 1801.0], "95e5c90e4a6814526cb4f3bea76286063b449d86a420f888c014531fb6e913a3"),
+}
+OUT = {k: sim.run(**v) for k, v in sim.SCENARIOS.items()}
+RUN = {k: v[:2] for k, v in OUT.items()}
+ERR = {k: v[4] for k, v in OUT.items()}
+
+# ============================================================================ #
+print("1. every command byte-identical to release/1.0 5a4cdc2")
+check("the same scenarios as 5a4cdc2", set(RUN) == set(BASE))
+for k, (alarms, digest) in RUN.items():
+    check(f"{k}: digest {digest[:12]}", digest == BASE[k][1])
+
+# ============================================================================ #
+print("2. no false alarm (5a4cdc2's are listed)")
+for k in ("edge4600", "edge4600_m0", "const4500", "const4500_m0", "sine4500",
+          "noise5000_m50", "noise5000_m0", "burst_end", "sine4500_fast"):
+    check(f"{k}: none (5a4cdc2: {BASE[k][0]})", RUN[k][0] == [])
+
+# ============================================================================ #
+print("3. real overloads still alarm, ALARM_HOLD_S after 5a4cdc2")
+# ALARM_HOLD_S (10 s) plus up to 2 s of late meter samples; at 1800 s the
+# quarter-edge tick restarts the hold.
+for k, want in (("step6500", [1211.0, 1813.0]),
+                ("miss5560", [1213.0, 1813.0])):
+    print(f"    {k}: load steps up at 1200 s; alarms at {RUN[k][0]}, "
+          f"5a4cdc2 {BASE[k][0]}")
+    check(f"{k}: after the hold, and again only at the next quarter",
+          RUN[k][0] == want)
+check("step6500_gap: cycles stopped by a lost battery reading do not "
+      "restart the hold",
+      RUN["step6500_gap"][0] == [1215.0, 1813.0])
+check("twice7000: notified once in the quarter (5a4cdc2: "
+      f"{BASE['twice7000'][0]})", RUN["twice7000"][0] == [1011.0])
+check("twice7000: the ERROR line for both overloads",
+      ERR["twice7000"] == [1011.0, 1512.0])
+# By design: until 1306 s (then the hold) this quarter would end at most
+# ALARM_MISS_W over the real 3500 W cap; 5a4cdc2 alarmed at the aim (cap - 50).
+print(f"    ramp6500: alarms at {RUN['ramp6500'][0]}, "
+      f"5a4cdc2 {BASE['ramp6500'][0]}")
+check("ramp6500: alarms once the quarter will end over the real cap by "
+      "more than ALARM_MISS_W",
+      RUN["ramp6500"][0] == [1316.0, 1813.0])
+check("step7500 under the peak reserve: not this alarm (the guard is not "
+      "allowed to discharge)", RUN["step7500_reserve"][0] == [])
+
+# ============================================================================ #
+print("4. load_too_high()")
+
+
+def proj(elapsed, done_avg, held):
+    """A projection `elapsed` s in, averaging `done_avg` so far."""
+    r = 900.0 - elapsed
+    return SimpleNamespace(elapsed_s=elapsed, held_w=held,
+                           projected_w=(done_avg * elapsed + max(held, 0) * r)
+                           / 900.0)
+
+
+def ltl(p, batt, cap=3500.0, rail=2000.0):
+    return load_too_high(p, batt_w=batt, ceiling_w=cap, achievable_w=rail)
+
+
+check("load 6000 W vs 3500 + 2000: yes", ltl(proj(300, 3450, 4000), 2000))
+check("already over, the battery can carry the load but not catch up: yes",
+      ltl(proj(450, 7200, 7200), 0, rail=5000))
+check("the last second, a 7 W miss (under ALARM_MISS_W): no",
+      not ltl(proj(899, 3509, 2500), 1000))
+check("a miss of exactly ALARM_MISS_W: no",
+      not ltl(proj(450, 3520, 3500), 2000))
+check("...and 1 W more: yes", ltl(proj(450, 3522, 3500), 2000))
+check("over cap + battery, but the quarter lands under the cap: no",
+      not ltl(proj(880, 3450, 4000), 2000))
+check("...and when it does not: yes", ltl(proj(700, 3450, 4000), 2000))
+check("no battery reading: no", not ltl(proj(300, 3450, 5000), None))
+check("no seconds left: no", not ltl(proj(900, 4000, 5000), 2000))
+check("a quarter already over is not won back by export: yes",
+      ltl(proj(800, 3950, 1000), 0))
+
+# ============================================================================ #
+print("5. the hold across a cycle with no battery reading")
+# A reading the stale detector drops (STALE_READING) reaches the alarm as None
+# while the cycle still runs. Unknown is not "fine": it must not restart the
+# hold. Driven on Controller.check_peak_saturated() itself, with stubs.
+Q0 = datetime(2026, 9, 20, 11, 0, tzinfo=timezone.utc)
+notes = []
+
+
+async def _call(domain, service, data):
+    notes.append(data.get("notification_id"))
+
+
+stub = SimpleNamespace(
+    peak_guard=SimpleNamespace(saturated=True, ceiling_w=3500.0,
+                               reason="PEAK_GUARD: fixture"),
+    site_rating_w=2000.0, shadow=False, events=[],
+    _peak_batt_prev=None, _peak_alarm_block=None, _peak_high_since=None,
+    _peak_saturated=False, hass=SimpleNamespace(call=_call),
+    log_event=lambda msg, hint=None: None)
+first = None
+for sec in range(0, 30):
+    stub.site_batt_w = None if 4 <= sec < 8 else 2000.0
+    p = proj(300 + sec, 3450, 6000)
+    p.block_start = Q0
+    stub._peak_proj = p
+    asyncio.run(Controller.check_peak_saturated(stub))
+    if notes and first is None:
+        first = sec
+check(f"a 4 s gap in the battery reading: alarm at {first} s, not 18 s",
+      first == 10)
+
+print()
+if fails:
+    print(f"FAILED: {len(fails)} check(s)")
+    sys.exit(1)
+print("all checks passed")

goodwe_controller/tools/safety78_identity.py

@@ -0,0 +1,110 @@
+#!/usr/bin/env python3
+"""SAFETY-78: at one inverter, is a maintenance drain byte-identical to before?
+
+    python3 tools/safety78_identity.py          # prints one digest per scenario
+
+tools/safety79_identity.py's harness (the REAL Controller on a P1 meter, so
+the peak guard runs) with the battery draining for its monthly cycle from the
+first cycle, under a house below, at and above the drain rate, a solar
+surplus and a load over the peak limit. Each row hashes safety79's fields,
+the modes' roles and the trade, the action log's newest line, and every claim
+main.py hands `resolve()`, so a claim or a log line added at N=1 that changes
+no watt still moves the digest. The `_modes` scenarios run battery modes on,
+Cheap hours only, over opt11p3_identity's prices. Uses only the API release/1.0 5a4cdc2 has, so the same file runs on
+both trees; test_safety78.py part 5 holds this tree to the digests printed
+there.
+"""
+
+import functools
+import hashlib
+import os
+import sys
+import tempfile
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+import opt11_identity as rig                               # noqa: E402
+import opt11p3_identity as p3                              # noqa: E402
+import safety79_identity as s79                            # noqa: E402
+import app.main as m                                       # noqa: E402
+from app import packmode                                  # noqa: E402
+from app.store import unit_key                            # noqa: E402
+
+DUR = 2400.0
+DRAIN = {"maintenance_enabled": True, "maintenance_start_hour": 11}
+
+
+def load(t):
+    """Below the 2000 W drain, at it, above it, a surplus, over the limit."""
+    k = int(t // 480)
+    return (1200.0, 2000.0, 3000.0, -1500.0, 6000.0)[min(k, 4)]
+
+
+def store_cheap(n):
+    s = rig._Store()
+    for u in ("inverter-a", "inverter-b")[:n]:
+        s.data[unit_key(u, packmode.MODE_KEY)] = packmode.CHEAP_HOURS
+    return s
+
+
+SCENARIOS = {
+    "n1_drain": dict(n=1),
+    "n1_drain_modes": dict(n=1, modes=True),
+}
+
+
+def run(n, *, modes=False, dur=DUR):
+    extra = dict(s79.P1, **DRAIN)
+    if modes:
+        extra.update(p3.ARB, battery_modes_enabled=True)
+    seen, extra_rows = [], []
+    real_resolve = m.resolve
+
+    def spy(claims, **kw):
+        seen.append(tuple((c.source, c.priority, c.kind, c.value, c.lo, c.hi,
+                           c.reason, c.actuator) for c in claims))
+        return real_resolve(claims, **kw)
+
+    def on_cycle(c, t):
+        g, tick, a = c.peak_guard, c.mode_tick, c.arb
+        extra_rows.append((
+            seen[-1] if seen else None,
+            None if g is None else (g.engaged, g.floor_w, g.reason),
+            None if tick is None else (tuple(sorted(tick.roles.items())),
+                                       tick.action, tick.why),
+            None if a is None else (a.action, a.power_w, a.reason),
+            [mm.phase for mm in c.maints],
+            len(c.events), c.events[0] if c.events else None))
+    db = os.path.join(tempfile.mkdtemp(prefix="s78-"), "tariff.db")
+    real = rig.Controller
+    rig.Controller = functools.partial(real, peak_db_path=db)
+    m.resolve = spy
+    try:
+        rows, c = rig.run(n, extra=extra, dur=dur, load=load, soc=s79.soc,
+                          cut=lambda u, t: False,
+                          before=p3._feed if modes else s79._feed,
+                          on_cycle=on_cycle,
+                          store=store_cheap(n) if modes else None)
+    finally:
+        rig.Controller = real
+        m.resolve = real_resolve
+    return [r + (x,) for r, x in zip(rows, extra_rows)], c
+
+
+def digest(rows) -> str:
+    return hashlib.sha256(repr(rows).encode()).hexdigest()
+
+
+def digests() -> dict:
+    return {k: digest(run(**v)[0]) for k, v in SCENARIOS.items()}
+
+
+if __name__ == "__main__":
+    if "--summary" in sys.argv:
+        for k, v in SCENARIOS.items():
+            rows = run(**v)[0]
+            print(f"{k}: {len(rows)} cycles, phases "
+                  f"{sorted({str(r[-1][-1]) for r in rows})}, targets "
+                  f"{sorted({r[1] for r in rows})[:6]}")
+    for k, v in digests().items():
+        print(f"{k} {v}")

goodwe_controller/tools/safety81_sim.py

@@ -0,0 +1,229 @@
+#!/usr/bin/env python3
+"""SAFETY-81: when does "Load too high for battery" fire?
+
+    python3 tools/safety81_sim.py            # alarms and a control digest per run
+
+The REAL Controller (read_inputs + cycle) at a 1 s cycle, on a plant shaped
+like the SAFETY-79 test rig: one battery switched Off (so only the peak guard
+makes it give, max 2000 W), a 3500 W cap, and a P1 meter that publishes
+0.3 s before each cycle. The battery reaches a new command 0.5 s after it is
+written. One meter sample in ten arrives late (the add-on still holds the one
+before), so the held meter reading can predate the battery's last change:
+the double count behind the rig's false alarms. The cycle on a quarter
+boundary, before the new quarter's first sample, files no peak claim, which
+is the rig's 1 s grid spike at each quarter change. Each run returns the
+alarm times and a digest of every command, so the same runs show both what
+changed and what did not.
+"""
+
+import asyncio
+import hashlib
+import logging
+import os
+import random
+import sys
+import tempfile
+import time
+from datetime import timedelta
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+import opt11_identity as rig                               # noqa: E402
+import app.main as m                                       # noqa: E402
+from app import packmode                                   # noqa: E402
+from app.main import Controller                            # noqa: E402
+from app.p1 import make_sample, split_signed               # noqa: E402
+from app.store import unit_key                             # noqa: E402
+
+TICK = 1.0
+SAMPLE_LEAD_S = 0.3     # the meter publishes this long before each cycle
+RESPONSE_S = 0.5        # a new command reaches the battery this much later
+LATE_P = 0.1            # share of meter samples that arrive late
+ALARM = "goodwe_peak_saturated"
+
+
+def constant(w):
+    return lambda t: w
+
+
+def noisy(mean, amp, seed=81):
+    """`mean` +/- `amp`, a new uniform value every second."""
+    rnd = random.Random(seed)
+    table = {}
+
+    def f(t):
+        k = int(t)
+        if k not in table:
+            table[k] = mean + rnd.uniform(-amp, amp)
+        return table[k]
+    return f
+
+
+def sine(mean, amp, period=120.0, noise=50.0):
+    import math
+    n = noisy(0.0, noise, seed=79)
+    return lambda t: mean + amp * math.sin(2 * math.pi * t / period) + n(t)
+
+
+def step(before, after, at):
+    return lambda t: before if t < at else after
+
+
+def ramp(before, after, at, secs):
+    return lambda t: before + (after - before) * min(max(t - at, 0.0) / secs, 1.0)
+
+
+class _Plant:
+    """One battery that reaches its command RESPONSE_S after the write."""
+
+    def __init__(self, load, rating, soc, batt_gap=()):
+        self.load, self.rating, self.soc = load, rating, soc
+        self.batt_gap = batt_gap      # (from, to): the battery reading is lost
+        self.t = 0.0
+        self.hist = [(-1e9, 0.0)]     # (time the command takes effect, W)
+        self.notes = []
+        self.meter = 0.0              # the last meter reading delivered
+
+    def batt(self, at):
+        w = 0.0
+        for when, v in self.hist:
+            if when <= at:
+                w = v
+        return max(-self.rating, min(self.rating, float(round(w))))
+
+    def grid(self, at):
+        return self.load(at) - self.batt(at)
+
+    async def limits(self, e):
+        return (-5000.0, 5000.0, 1.0)
+
+    async def call(self, domain, service, data):
+        if service == "create":
+            self.notes.append((self.t, data.get("notification_id")))
+        return True
+
+    async def number(self, e, invert=False):
+        if e in ("sensor.grid", "sensor.net"):
+            return self.meter
+        if e.startswith("sensor.soc"):
+            return self.soc
+        if e.startswith("sensor.batt"):
+            if self.batt_gap and self.batt_gap[0] <= self.t < self.batt_gap[1]:
+                return None
+            return self.batt(self.t)
+        return None
+
+    async def set_number(self, e, v):
+        self.hist.append((self.t + RESPONSE_S, float(v)))
+        return True
+
+
+def run(load, *, margin=50, dur=2700.0, rating=2000.0, cap=3500.0,
+        late_at=(), soc=60.0, batt_gap=()):
+    """Alarm times (s from the start of the first quarter), a digest of every
+    cycle's target and written command, the Controller, the rows, and the
+    times of the ERROR log line. A sample is also late at each `t % 900` in
+    `late_at`."""
+    opts = rig._options(1, {
+        "meter_source": "ha_signed", "p1_net_entity": "sensor.net",
+        "meter_phases": 1, "peak_cap_w": cap, "peak_margin_w": margin,
+        "battery_modes_enabled": True, "max_w": rating})
+    store = rig._Store()
+    store.data[unit_key("inverter-a", packmode.USE_KEY)] = False
+    store.data[packmode.USE_KEY] = False
+    db = os.path.join(tempfile.mkdtemp(prefix="s81-"), "tariff.db")
+
+    async def go():
+        plant = _Plant(load, rating, soc, batt_gap)
+        c = Controller(opts, plant, store, rig._Pub(), stop=asyncio.Event(),
+                       peak_db_path=db)
+        for i in c.inverters:
+            i.dev_min, i.dev_max = -5000.0, 5000.0
+        c.auto = True
+        real_dt, clock = m.datetime, [0.0]
+
+        class _DT(real_dt):
+            @classmethod
+            def now(cls, tz=None):
+                return rig.BASE + timedelta(seconds=clock[0])
+        m.datetime = _DT
+        rows, late = [], random.Random(8100)
+        errors = []
+
+        class _Err(logging.Handler):
+            def emit(self, r):
+                if "battery alone cannot keep" in r.getMessage():
+                    errors.append(plant.t)
+        h = _Err()
+        logging.getLogger("goodwe").addHandler(h)
+        try:
+            t = 0.0
+            while t < dur:
+                plant.t = clock[0] = t
+                ts = t - SAMPLE_LEAD_S
+                if late.random() >= LATE_P and t % 900 not in late_at:
+                    plant.meter = plant.grid(ts)
+                    imp, exp = split_signed(plant.meter)
+                    c.p1.submit(make_sample(
+                        "ha_signed", imp, exp, phases=1,
+                        ingest_ts=rig.BASE + timedelta(seconds=ts),
+                        ingest_mono=time.monotonic()))
+                await c.read_inputs(now_mono=t)
+                await c.cycle()
+                rows.append((t, c.target, round(plant.hist[-1][1], 6)))
+                t += TICK
+        finally:
+            m.datetime = real_dt
+            logging.getLogger("goodwe").removeHandler(h)
+        return plant, c, rows, errors
+    plant, c, rows, errors = asyncio.run(go())
+    alarms = [t for t, nid in plant.notes if nid == ALARM]
+    return (alarms, hashlib.sha256(repr(rows).encode()).hexdigest(), c, rows,
+            errors)
+
+
+# Three quarters each (2700 s); a new quarter starts at 900 s and 1800 s.
+SCENARIOS = {
+    # F1: constant and sine loads the battery carries, across quarter edges.
+    "const4500": dict(load=constant(4500.0)),
+    "const4500_m0": dict(load=constant(4500.0), margin=0),
+    "sine4500": dict(load=sine(4500.0, 400.0)),
+    # F1 at the start of a quarter: the sample after the battery's return
+    # is late, so the held reading is the 1 s spike (the battery at 0 W).
+    "edge4600": dict(load=constant(4600.0), late_at=(2.0,)),
+    "edge4600_m0": dict(load=constant(4600.0), margin=0, late_at=(2.0,)),
+    # 6000 W in a quarter's last 20 s: over cap + battery, but the quarter
+    # still lands between the aim and the cap.
+    "burst_end": dict(load=lambda t: 6000.0 if t % 900 >= 880 else 4500.0),
+    # F2: a noisy load the battery carries, at margin 50 and 0.
+    "noise5000_m50": dict(load=noisy(5000.0, 150.0)),
+    "noise5000_m0": dict(load=noisy(5000.0, 150.0), margin=0),
+    # Real: from 1200 s the house needs 1000 W more than cap + battery.
+    "step6500": dict(load=step(4500.0, 6500.0, 1200.0)),
+    # Real, small: from 1200 s, 60 W (+/- 20 W noise) more than cap + battery.
+    "miss5560": dict(load=lambda t: step(4500.0, 5560.0, 1200.0)(t)
+                     + noisy(0.0, 20.0)(t)),
+    # Real, slow: 4500 -> 6500 W over 200 s from 1200 s, through the band
+    # between the aim (cap - 50) and the real cap.
+    "ramp6500": dict(load=ramp(4500.0, 6500.0, 1200.0, 200.0)),
+    # The pack under its peak reserve (20 %), and a load far over cap +
+    # battery: that is "Battery stopped limiting the peak", not this alarm.
+    "step7500_reserve": dict(load=step(4500.0, 7500.0, 1200.0), soc=15.0),
+    # Review: a load swinging 4500 +/- 1200 W every 30 s (a hob or heater
+    # cycling). It tops cap + battery for about 5 s a swing, and the quarter
+    # still lands under the aim.
+    "sine4500_fast": dict(load=sine(4500.0, 1200.0, period=30.0, noise=200.0)),
+    # Review: two real overloads in one quarter, 1000-1100 s and 1500-1700 s.
+    "twice7000": dict(load=lambda t: 7000.0 if (1000 <= t < 1100
+                                                or 1500 <= t < 1700)
+                      else 4500.0),
+    # Review: step6500, and the battery reading is lost 1203-1215 s.
+    "step6500_gap": dict(load=step(4500.0, 6500.0, 1200.0),
+                         batt_gap=(1203.0, 1215.0)),
+}
+
+
+if __name__ == "__main__":
+    for k, v in SCENARIOS.items():
+        a, d, _, _, _ = run(**v)
+        print(f"{k:16s} alarms at {a} digest {d}")