SAFETY-73: a battery that refuses, per inverter, with two or more

SAFETY-52's hold applied per unit at N>1 (owner, 2026-09-27, option R).
A unit that gives < 100 W for 12 cycles in one direction while asked
> 200 W, or trips its stale-reading gate on a ~0 W reading, gets a
P_SAFETY `refused` unit limit closing that direction, so the allocator
hands its share to the others within their own limits. Probe: a +-200 W
pin every 15 min, one unit at a time, only while the site asks 200 W that
way. All units refusing: the site law holds 0 W. Exits reseed the site
target from the wire. INFO plus one WARNING (REFUSES_ABOVE_CUT_OUT) above
the configured cut-out; SAFETY-42/43 no longer report a refusing unit.
N=1 unchanged.

test_refusal_units.py: cases a/b/c on flat and jitter registers against
the old behaviour, probes, recovery, exits; mutations U1-U13 all red.
TEST-19 no-fault campaign: 0 engagements in 5280 runs.

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

goodwe_controller/DECISIONS/SAFETY-73-refusal-per-inverter.md

@@ -0,0 +1,89 @@
+# SAFETY-73 - a battery that refuses, with two or more inverters
+
+**Decided 2026-09-27 by the owner, on the phase A plan (option R). All four
+new questions answered yes. SAFETY-52's four decisions carry over.**
+
+## The defect
+
+SAFETY-52 held a refusing battery at 0 W at one inverter only. At N>1
+(`main.py` passed `refused=None`) nothing closed a unit that gives ~0 W while
+its state of charge reads above its configured cut-out, so the allocator kept
+handing it its weighted half. Measured on the real `Controller`, two units,
+house 3000 W, A cut out:
+
+| register | law | B gives | grid | A's gate trips / h |
+|---|---|---|---|---|
+| flat -4 W | old | 533 W | +2471 W | 46 (each stops the whole site) |
+| flat -4 W | new | 2000 W (its rating) | +1004 W | 1 |
+| jitter | old | 1910 W | +1094 W | 0 |
+| jitter | new | 2000 W | +1004 W | 0 |
+
+House 1200 W, which B alone carries: old +671 W (flat) and +284 W (jitter)
+standing import, new -16 W. Both cut out: old keeps asking two empty batteries
+(+1910 W each on the jitter shape), new holds the site at 0 W.
+
+## What was decided
+
+1. **A refusing unit is closed in that direction** by its own P_SAFETY unit
+   claim, source `refused`. It engages when the unit gives under
+   `REFUSED_BELOW_W` (100 W) for `SETTLE_CYCLES` (12) cycles in one direction
+   while asked more than `PROBE_W` (200 W), or when its stale-reading gate
+   trips on a held reading under 100 W. The allocator then hands its share to
+   the others within their own limits (owner decision 1). It is a hard unit
+   limit that always intersects, so OPT-11's P_MODE layer fits inside it.
+2. **It binds every strategy's ask on that unit** (decision 2): peak guard,
+   arbitrage and maintenance are placed around it like the loop.
+3. **A refusal above the configured cut-out** logs INFO plus one WARNING ending
+   in `REFUSES_ABOVE_CUT_OUT` (decision 3). SAFETY-42/43 no longer report such a
+   unit: its share is 0 W, and `account()` treats it as parked.
+4. **All units refusing the same way** pass that sign to `compute()`, so the
+   site holds 0 W with SAFETY-52's law (decision 4). Probes go one unit at a
+   time.
+5. **Probe:** every 15 min for 34 s the claim becomes a pin at +-200 W and the
+   others carry the rest. The pin stands only while the resolved site target
+   asks at least 200 W its way; otherwise it would make a sibling take the
+   opposite direction, so it falls back to the closed limit for that cycle.
+6. **Exits:** a reading of 100 W or more in the held direction (probe or not);
+   100 W or more taken the other way for 12 cycles (charge at the floor; one
+   cycle of the loop overshooting through zero is not enough); stop or start.
+   On every exit the site target and integrator restart from the watts on the
+   wire, or the reopened unit is handed a parked remainder at once.
+
+**Why 200 W and not `saturation_w`.** A unit's share is often small: a 400 W
+night load split two ways asks 200 W of each. With 500 W the empty battery
+kept a phantom share that SAFETY-43's relative arm then reported as not
+delivering, the ERROR decision 3 replaces. Shares of 200 W or less that give
+under 100 W are "too small to call" in `account()`: the sibling covers the
+meter and nothing is asked of the empty battery that matters.
+
+**Why one direction throughout.** On TEST-19's flapping family a command that
+swaps sign each cycle reads ~0 through the plant's lag: 42 engagements in 21
+no-fault runs without the rule, 0 with it.
+
+**N=1 is unchanged.** Every SAFETY-73 path is gated on `len(inverters) > 1`.
+`test_refusal_units.py` part 6 runs SAFETY-52's whole suite with every
+SAFETY-73 method made to raise.
+
+## Evidence
+
+- `test_refusal_units.py`: cases a, b and c on both register shapes against
+  the old behaviour, probes and their spacing, one-at-a-time probing, the
+  recovery reseed, each exit, the small-ask and small-site-target cases, the
+  WARNING, the status row and shadow mode. Mutations U1-U13 in its docstring.
+- **No false engagements.** TEST-19's no-fault campaign (admin/goodwe
+  `sim/delivery_window.py` families, 2640 runs at N=2 and 2640 at N=1, every
+  unit executing its share) graded with the engage clause: 0 engagements, the
+  longest qualifying streak 5 cycles against 12. The grader is
+  `goodwe-session/scratch_s73/s73_nofault.py`.
+- **Recovery.** Case a, A recovering into a house that fell below the parked
+  target: worst grid +0 W with the reseed. Phase A measured -1059 W without the
+  reseed and the site hold.
+
+## Not covered
+
+- A unit that gives a steady part of its ask (100 W or more) is a limit, as at
+  N=1. SAFETY-42/43 still report it.
+- `pack_permissions` stays SoC-based, so the peak guard may size a shave with a
+  refusing battery counted. The shortfall shows as `unmet_w`.
+- A wedged unit (HW-01) reading ~0 W is now shown as a refusal, not reported as
+  not delivering. Its share still moves, which is the gain SAFETY-42 measured.

goodwe_controller/DOCS.md

@@ -295,7 +295,7 @@ When the add-on refuses to start, each reason names the setting by its label and
 - **Stuck meter alarm after** (`meter_freeze_s`): measured, not chosen (SAFETY-18). Over 2399 s of this house at its operating point, the longest run without a new reading is 15.4 s, and 300 s is 19.5 times that. It raises an alarm only and does not set 0 W.
 - **Opposing inverters threshold** (`circulation_w`): measured on 924 closed-loop test runs (SAFETY-31). The worst legitimate opposition drops from 505 s at 100 W to 5 s at 400 W, the lowest value accepted. 700 W is 1.75 times that knee, with margin because the second inverter is still a copy of the first.
 - **Opposing inverters time** (`circulation_hold_s`): derived (SAFETY-31). 75 s is fifteen control cycles, 15 times the worst legitimate episode, and 2.5 times the control board's 30 s link limit.
-- **Not delivering alarm after** (`delivery_hold_s`): two or more inverters only. It reports only (SAFETY-42): nothing is commanded differently. Derived from 2640 closed-loop test runs with nothing faulty. The worst legitimate episode is 35.0 s and 185 s is 5.29 times that. The schema allows 30 s, below that, so expect false reports there. A battery at or below its own cut-out level is never reported. Re-derive with admin/goodwe `sim/delivery_window.py --campaign --blag`.
+- **Not delivering alarm after** (`delivery_hold_s`): two or more inverters only. It reports only (SAFETY-42): nothing is commanded differently. Derived from 2640 closed-loop test runs with nothing faulty. The worst legitimate episode is 35.0 s and 185 s is 5.29 times that. The schema allows 30 s, below that, so expect false reports there. A battery at or below its own cut-out level is never reported, and neither is one that gives under 100 W (SAFETY-73: it is a battery that refuses). Re-derive with admin/goodwe `sim/delivery_window.py --campaign --blag`.
 - **Shadow mode** (`shadow_mode`): the whole decision path runs and nothing reaches the inverter. The status page and action log show what would have been sent. A safety stop that is already up stays up. See [Shadow mode](#shadow-mode-shadow_mode).
 - **Control gain** (`gain`): measured on real hardware and already at the limit.
 - **Largest change per cycle** (`slew_w`): 0 means follow the rating: about a ninth of it, derived (SAFETY-39) from a measured ~18 s step response against a 2 s meter tick. A fixed value faster than the rating allows is refused. At 1000 W against a 5000 W rating control swung the meter from 3.4 kW export to 2.8 kW import on 2026-09-05.
@@ -550,15 +550,23 @@ control is stopped or started. 200 W is below the 250 W the stale-reading check
 needs, so a test never trips it. A battery that gives some power, even 100 W,
 is not refusing and keeps its request.
 
-⚠️ **One inverter only.** With several inverters this hold is off. There, the
+**With two or more inverters** the same rule works per inverter. An inverter
-battery level limit moves an empty battery's share to the others, but only if
+whose battery gives less than 100 W for 12 cycles while asked for more than
-its “Battery cut-out level” matches where the battery really stops. A battery
+200 W is asked 0 W, and the other inverters take over its share, each up to
-that refuses at another level, or for another reason (cold, a battery fault),
+its own limits. The status page names it: *inverter-a: the battery refuses to
-can still hold its last request until the grid starts to export.
+give power, the others take over*. It is tested at 200 W every 15 minutes, one
+inverter at a time, and only while the site asks at least 200 W that way.
+It still accepts charge, and a minute of charging ends the refusal. When every
+inverter refuses the same way, the whole site asks 0 W as with one inverter.
+If the battery refuses while its level reads above its “Battery cut-out level”,
+the log adds one warning ending in `REFUSES_ABOVE_CUT_OUT`: check that setting.
+A battery that refuses is not reported as *not delivering*.
 
-*For installers:* SAFETY-52. The event log reads `BATTERY REFUSES …` and
+*For installers:* SAFETY-52 and SAFETY-73. The event log reads
-`BATTERY REFUSES ENDED: …`, and the INFO line ends in `BATTERY_REFUSES`.
+`BATTERY REFUSES …` and `BATTERY REFUSES ENDED …` (with the inverter's name at
-`DECISIONS/SAFETY-52-a-battery-that-refuses.md` has the replay figures.
+N>1), and the INFO line ends in `BATTERY_REFUSES`.
+`DECISIONS/SAFETY-52-a-battery-that-refuses.md` has the replay figures and
+`DECISIONS/SAFETY-73-refusal-per-inverter.md` the two-inverter ones.
 
 ⚠️ **`Loop reason` is not the `Why` row on the status page, and the two will
 often disagree.** `Why` is the *arbiter's* answer in words — *Keeping the grid

goodwe_controller/app/arbiter.py

@@ -226,6 +226,12 @@ CLAIM_SCOPE = {
     # reaches the allocator, so a unit-scoped claim is the shape that sentence
     # asks for rather than an evasion of it.
     "pack": SCOPE_UNIT,
+    # main.py: `Claim.limit("refused", P_SAFETY, ...)` and, during a probe,
+    # `Claim.set("refused", P_SAFETY, +-PROBE_W)`. SAFETY-73: THIS battery gives
+    # nothing in one direction, measured on THIS unit's own reading. A hard unit
+    # limit that always intersects, so the share goes to the siblings. Filed at
+    # N>1 only; at N=1 SAFETY-52's law in compute() does the same job.
+    "refused": SCOPE_UNIT,
     # main.py: `Claim.set("maintenance", P_MAINTENANCE, ...)` when the cycle owns
     # the setpoint, and `Claim.limit("maintenance", P_MAINT_SHAPE, hi=...)` when
     # it only shapes. §12's monthly recalibration is a PACK procedure - drain

goodwe_controller/app/inverter.py

@@ -317,6 +317,16 @@ class Inverter:
             tripped=bool(data.get(self.key("delivery_tripped"), False)),
             reason=str(data.get(self.key("delivery_reason"), "") or ""))
 
+        # SAFETY-73. The direction this unit's battery refuses (+1 give, -1
+        # take, 0 none), when that refusal or its last probe started, the
+        # current probe's start, and the cycles of evidence so far. Memory
+        # only, like SAFETY-52's: stop and start end it. Used at N>1 only.
+        self.refused = 0
+        self.refused_at = None
+        self.probe_from = None
+        self.ref_count = 0
+        self.ref_sign = 0
+
     # -- store ---------------------------------------------------------------
     def key(self, name: str) -> str:
         """This unit's namespaced Store key for `name`. See store.unit_key()."""
@@ -411,6 +421,13 @@ class Inverter:
         return (messages.row_battery_stuck(self.name) if self.batt_fresh.stale
                 else "")
 
+    def refusal_note(self) -> str:
+        """SAFETY-73's status row while this unit's battery refuses."""
+        if not self.refused:
+            return ""
+        return messages.row_battery_refuses(self.name, self.refused,
+                                            self.probe_from is not None)
+
     def link_note(self, now: datetime) -> str:
         """One line for the status page. Empty when there is nothing to say.
 

goodwe_controller/app/main.py

@@ -28,6 +28,7 @@ import contextlib
 import functools
 import json
 import logging
+import math
 import signal
 import time
 from datetime import datetime, timedelta, timezone
@@ -234,6 +235,11 @@ def pack_energy_of(inv):
                                 inv.capacity_kwh, inv.max_w)
 
 
+def _refused_interval(sign: int) -> tuple:
+    """SAFETY-73: the (lo, hi) a refusing unit may be asked: 0 W that way."""
+    return (-math.inf, 0.0) if sign > 0 else (0.0, math.inf)
+
+
 def why_key(winner: str, claims) -> str:
     """The key of `messages.WHY_WINNER` for a resolution's winner.
 
@@ -1121,6 +1127,19 @@ class Controller:
         """
         t = self.tuning
         standing = inv.standing_w()
+        if len(self.inverters) > 1:
+            # SAFETY-73: at N>1 the same trip refuses THIS unit only.
+            if ((self.auto or self.shadow) and not inv.refused
+                    and inv.ref_count >= t.saturation_cycles
+                    and inv.batt_fresh.held is not None
+                    and abs(inv.batt_fresh.held) < _ctl.REFUSED_BELOW_W
+                    and abs(standing) > _ctl.PROBE_W):
+                self._now_mono = now_mono
+                self._unit_refusal_start(
+                    inv, 1 if standing > 0 else -1,
+                    f"stale reading {inv.batt_fresh.held:+.0f} W while asked "
+                    f"{standing:+.0f} W")
+            return
         if (len(self.inverters) == 1 and (self.auto or self.shadow)
                 and not self.batt_refused
                 and self.sat_count >= t.saturation_cycles
@@ -1132,6 +1151,111 @@ class Controller:
                                 f"stale reading {inv.batt_fresh.held:+.0f} W "
                                 f"while asked {standing:+.0f} W")
 
+    # --- SAFETY-73: the same refusal, per inverter, at N>1 -------------------
+    # SAFETY-52's decisions applied to one unit of several: a unit that gives
+    # < REFUSED_BELOW_W for SETTLE_CYCLES while asked > PROBE_W is closed in
+    # that direction by a P_SAFETY unit limit, so the allocator hands its share
+    # to the others; it is probed at PROBE_W every PROBE_EVERY_S, one unit at a
+    # time. The ask is PROBE_W, not `saturation_w`: a unit's share is often
+    # small (a 400 W night load split two ways) and SAFETY-43 would otherwise
+    # name the empty battery as not delivering. DECISIONS/SAFETY-73.
+    def _unit_refusal_start(self, inv, sign: int, evidence: str) -> None:
+        inv.refused, inv.refused_at = sign, self._now_mono
+        inv.probe_from, inv.ref_count = None, 0
+        self.log_event(f"BATTERY REFUSES {inv.name}: {evidence}",
+                       hint=messages.hint("unit_refuses", unit=inv.name))
+        _LOG.info("%s: the battery refuses power, so it is asked 0 W and the "
+                  "other inverters take over: BATTERY_REFUSES.", inv.name)
+        _LOG.debug("Evidence for %s: BATTERY_REFUSES: %s.", inv.name, evidence)
+        if (inv.soc is not None and inv.bms_soc_floor is not None
+                and float(inv.soc) > float(inv.bms_soc_floor)):
+            # Owner decision 3: a refusal above the configured cut-out level
+            # may mean that level is set wrong. One line, no safety stop.
+            _LOG.warning("%s refuses power above its battery cut-out level, "
+                         "check its cut-out level: REFUSES_ABOVE_CUT_OUT.",
+                         inv.name)
+            _LOG.debug("Evidence for %s: REFUSES_ABOVE_CUT_OUT: battery level "
+                       "%.0f %%, cut-out level %.0f %%.", inv.name,
+                       float(inv.soc), float(inv.bms_soc_floor))
+
+    def _unit_refusal_end(self, inv, why: str) -> None:
+        if not inv.refused:
+            return
+        inv.refused, inv.refused_at, inv.probe_from = 0, None, None
+        inv.ref_count = 0
+        self.log_event(f"BATTERY REFUSES ENDED {inv.name}: {why}",
+                       hint=messages.hint("unit_gives", unit=inv.name))
+        _LOG.info("%s: the battery refusal ends, it is asked for power again: "
+                  "%s.", inv.name, why)
+        # The site target may stand above what was placed (the refused share
+        # was unmet). Restart it from the watts on the wire, or the reopened
+        # unit is handed the parked remainder at once (phase A: -1059 W).
+        placed = float(sum(i.standing_w() for i in self.inverters))
+        self.target = self.i_w = placed
+        self.sat_count = 0
+
+    def _unit_probe(self, inv) -> bool:
+        """True while `inv` may try PROBE_W. One unit probes at a time."""
+        now = self._now_mono
+        if not inv.refused or inv.refused_at is None or now is None:
+            return False
+        if inv.probe_from is not None:
+            if now - inv.probe_from < _ctl.PROBE_S:
+                return True
+            inv.probe_from, inv.refused_at = None, now
+            return False
+        if now - inv.refused_at < _ctl.PROBE_EVERY_S or any(
+                i.probe_from is not None for i in self.inverters):
+            return False
+        inv.probe_from = now
+        return True
+
+    def _unit_refusals(self) -> None:
+        """Engage and end each unit's refusal from its own reading."""
+        for inv in self.inverters:
+            b = inv.batt
+            if inv.refused:
+                if b is not None and inv.refused * b >= _ctl.PROBE_ANSWER_W:
+                    self._unit_refusal_end(
+                        inv, "the probe was answered"
+                        if inv.probe_from is not None
+                        else "the battery gives power")
+                    continue
+                # The other direction: the battery takes PROBE_ANSWER_W or more
+                # the other way (charge at the floor) for SETTLE_CYCLES. Counted
+                # on its own reading, so one cycle of the loop overshooting
+                # through zero does not hand it its share back.
+                inv.ref_count = (inv.ref_count + 1 if b is not None and
+                                 -inv.refused * b >= _ctl.PROBE_ANSWER_W else 0)
+                if inv.ref_count >= _ctl.SETTLE_CYCLES:
+                    self._unit_refusal_end(
+                        inv, "the battery takes power the other way")
+                continue
+            if b is None:
+                continue        # a gap keeps the count: the gate's trip seeds
+            standing = inv.standing_w()
+            sign = 1 if standing > 0 else -1
+            # One direction throughout: a command that swaps sign every cycle
+            # (a flapping load) reads ~0 through the plant's own lag, and that
+            # is a battery following, not refusing (TEST-19 flapping family).
+            if abs(standing) > _ctl.PROBE_W and abs(b) < _ctl.REFUSED_BELOW_W:
+                inv.ref_count = (min(inv.ref_count + 1, _ctl.SETTLE_CYCLES)
+                                 if sign == inv.ref_sign else 1)
+            else:
+                inv.ref_count = 0
+            inv.ref_sign = sign
+            if inv.ref_count >= _ctl.SETTLE_CYCLES:
+                self._unit_refusal_start(
+                    inv, sign,
+                    f"asked {standing:+.0f} W, the battery gave {b:+.0f} W "
+                    f"for {_ctl.SETTLE_CYCLES} cycles")
+
+    def _site_refused(self):
+        """Owner decision 4: every unit refuses the same way, so the site law
+        holds 0 W (SAFETY-52's, in compute()). Otherwise None, the old law."""
+        signs = {inv.refused for inv in self.inverters}
+        return signs.pop() if len(signs) == 1 and 0 not in signs else None
+
     def log_event(self, msg: str, hint: str | None = None) -> None:
         """`msg` goes on the action log, the add-on's own technical record.
         `hint` is `messages.hint()`'s plain words for the same event, and is
@@ -2516,7 +2640,8 @@ class Controller:
         now = datetime.now(timezone.utc)
         return [n for n in
                 [x for inv in self.inverters
-                 for x in (inv.link_note(now), inv.stale_note())]
+                 for x in (inv.link_note(now), inv.stale_note(),
+                           inv.refusal_note())]
                 if n]
 
     async def cycle(self) -> None:
@@ -2810,7 +2935,13 @@ class Controller:
                     # what part 6 grades instead is that WIDENING the window
                     # changes the answer.
                     live=inv.is_live(now)[0],
-                    parked=inv.sign_wd.tripped,
+                    # SAFETY-73: a refusing unit is asked 0 W on purpose, and
+                    # one asked under PROBE_W that gives under 100 W is a
+                    # refusal too small to call (its sibling covers the meter).
+                    parked=inv.sign_wd.tripped or bool(inv.refused) or (
+                        n_units > 1 and inv.batt is not None
+                        and abs(float(standing[inv.name])) <= _ctl.PROBE_W
+                        and abs(float(inv.batt)) < _ctl.REFUSED_BELOW_W),
                     # ⚠️ THIS UNIT'S OWN CUT-OUT. Not the shipped 10 %, and not
                     # the site's `peak_soc_floor` either - that one bounds a
                     # money-saving discharge only, and grading a pack against it
@@ -3422,6 +3553,16 @@ class Controller:
             self._refusal_auto = bool(self.auto)
             self._refusal_end("control started" if self.auto
                               else "control stopped")
+            for inv in (self.inverters if len(self.inverters) > 1 else ()):
+                self._unit_refusal_end(inv, "control started" if self.auto
+                                       else "control stopped")
+        # SAFETY-73: each unit's own refusal, engaged and ended on its own
+        # reading, before the loop reads `self.target` (an end reseeds it).
+        _probing = {}
+        if len(self.inverters) > 1 and self.inputs_ok:
+            self._unit_refusals()
+            _probing = {inv.name: self._unit_probe(inv)
+                        for inv in self.inverters if inv.refused}
         if (self.auto or self.shadow) and self.inputs_ok:
             # ⚠️ prev is the ARBITER's last output, not the loop's own last
             # wish. If something outranked the loop, that is what the hardware
@@ -3446,6 +3587,8 @@ class Controller:
             if self.loop_w is not None and self.target != self.loop_w:
                 self._refusal_end("another writer")
             probe = self._refusal_probe()
+            if len(self.inverters) > 1:
+                probe = any(_probing.values())
             # ⚠️ `actual_w` IS THE SITE SUM AND THAT IS DEV-02's OTHER HALF.
             # The anti-windup test inside compute() is
             # `abs(prev_w - actual_w) > saturation_w`, and `prev_w` is the SITE
@@ -3476,7 +3619,10 @@ class Controller:
                 i_w=self.i_w,
                 # SAFETY-52 at one inverter only: with several, OPT-09's pack
                 # limit already closes an empty battery's discharge end.
-                refused=self.batt_refused if len(self.inverters) == 1 else None,
+                # SAFETY-73: at N>1 the site law holds 0 W only while every
+                # unit refuses the same way. Otherwise None, the old law.
+                refused=self.batt_refused if len(self.inverters) == 1
+                else self._site_refused(),
                 probe=probe,
             )
             if decision.frozen and self.sat_count < self.tuning.saturation_cycles:
@@ -3486,7 +3632,8 @@ class Controller:
             self.sat_count = decision.sat_count
             self.i_w = decision.i_w
             self.loop_w = decision.target_w
-            self._refusal_update(decision, loop_prev, site_batt, probe)
+            if len(self.inverters) == 1:
+                self._refusal_update(decision, loop_prev, site_batt, probe)
             # --- TEL-10 ------------------------------------------------------
             # ⚠️ THE INTEGRATOR HAS NEVER LEFT THIS PROCESS, AND THAT IS WHAT
             # THE CARD IS. `self.i_w` is carried between cycles and published
@@ -3585,8 +3732,44 @@ class Controller:
         # ability to DELIVER rests on this argument - a weight cannot stop a
         # unit being commanded and cannot move a failing unit's load to its
         # sibling. The `pack` claim filed above is the half that constrains.
-        resolution = resolve(claims, units=self.units, live=self.live_units(),
+        # --- SAFETY-73: each refusing unit's own P_SAFETY claim ---------------
-                             weights=self.pack_weights, tick=self.alloc_tick)
+        # A hard unit limit that always intersects (OPT-11's P_MODE layer sits
+        # inside it): closed in the refused direction, so the allocator hands
+        # the share to the others within their own limits. During its probe it
+        # is a pin at PROBE_W and the others carry the rest.
+        for inv in self.inverters:
+            if not inv.refused:
+                continue
+            if _probing.get(inv.name):
+                claims.append(Claim.set("refused", P_SAFETY,
+                                        inv.refused * _ctl.PROBE_W,
+                                        _ctl.PROBING,
+                                        actuator=self.actuator_of("refused",
+                                                                  inv)))
+            else:
+                claims.append(Claim.limit("refused", P_SAFETY,
+                                          *_refused_interval(inv.refused),
+                                          _ctl.REFUSED,
+                                          actuator=self.actuator_of("refused",
+                                                                    inv)))
+        _live = self.live_units()
+        while True:
+            resolution = resolve(claims, units=self.units, live=_live,
+                                 weights=self.pack_weights, tick=self.alloc_tick)
+            # SAFETY-73: a probe pin stands only while the resolved site target
+            # asks at least PROBE_W its way. Below that a sibling would carry
+            # the opposite direction (circulation), so the pin goes back to the
+            # closed limit and the split is resolved once more.
+            _void = [k for k, c in enumerate(claims)
+                     if c.source == "refused" and c.kind == "set"
+                     and c.value * resolution.target_w < _ctl.PROBE_W ** 2]
+            if not _void:
+                break
+            for k in _void:
+                claims[k] = Claim.limit(
+                    "refused", P_SAFETY,
+                    *_refused_interval(1 if claims[k].value > 0 else -1),
+                    _ctl.REFUSED, actuator=claims[k].actuator)
         self.allocation = resolution.allocation
         # --- THE TWO REFUSALS, AND THEY ARE NOT THE SAME REFUSAL --------------
         # Both do the same thing - REPLACE the resolution with a 0 W failsafe
@@ -5114,7 +5297,8 @@ class Controller:
         _now = datetime.now(timezone.utc)
         for inv in self.inverters:
             for note, red in ((inv.link_note(_now), not inv.is_live(_now)[0]),
-                              (inv.stale_note(), False)):
+                              (inv.stale_note(), False),
+                              (inv.refusal_note(), False)):
                 if note:
                     out.append({"ok": False, "warn": not red, "text": note})
 

goodwe_controller/app/messages.py

@@ -786,6 +786,14 @@ def row_battery_stuck(unit) -> str:
             f"for now")
 
 
+def row_battery_refuses(unit, sign, probing) -> str:
+    """SAFETY-73. One inverter's battery refuses, the others take its share."""
+    way = "give" if sign > 0 else "take"
+    if probing:
+        return f"{unit}: testing if the battery will {way} power again"
+    return f"{unit}: the battery refuses to {way} power, the others take over"
+
+
 # The top of the page: why the battery does what it does (the "Why" line).
 WHY_WINNER = {
     "loop": "Keeping the grid near zero",
@@ -805,6 +813,8 @@ WHY_WINNER = {
     # SAFETY-52: the loop's two words for a battery that gives nothing.
     "battery refuses": "The battery refuses to give power",
     "battery probe": "Testing if the battery gives power again",
+    # SAFETY-73: a refusing inverter's own claim, closed or probing.
+    "refused": "A battery refuses power, the others take over",
 }
 WHY_LIMIT = {
     "peak(quarter peak)": "the peak limit",
@@ -815,6 +825,7 @@ WHY_LIMIT = {
     "pack": "the battery level",
     "reserve": "the battery reserve",
     "maintenance": "battery maintenance",
+    "refused": "a battery that refuses power",
 }
 
 
@@ -857,6 +868,8 @@ HINTS = {
     "battery_back": "{unit}: battery power reading is used again",
     "battery_refuses": "The battery refuses to give power, the add-on asks 0 W",
     "battery_gives": "The battery refusal ended, control asks for power again",
+    "unit_refuses": "{unit}: the battery refuses power, the others take over",
+    "unit_gives": "{unit}: the battery refusal ended, it is asked for power again",
     "bad_value": "The add-on made an invalid power value, inverters set to 0 W",
     "write_blocked": "{unit}: a power request was blocked, it is set to 0 W",
     "all_blocked": "A power request was blocked, inverters set to 0 W",
@@ -1027,6 +1040,12 @@ def status_samples() -> list:
         ("row_peak_guard_on", AMBER, row_peak_guard_on(1200, 2500,
                                                        shadow=True)),
         ("row_battery_stuck", AMBER, row_battery_stuck("inverter-b")),
+        ("row_battery_refuses", AMBER, row_battery_refuses("inverter-b", 1,
+                                                           False)),
+        ("row_battery_refuses", AMBER, row_battery_refuses("inverter-b", -1,
+                                                           False)),
+        ("row_battery_refuses", AMBER, row_battery_refuses("inverter-b", 1,
+                                                           True)),
         ("row_circulating_check", GREEN, row_circulating_check(0, 0, 60)),
     ]
 

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: eleven sources, each classified where it is BUILT")
+print("ARB-03 2. the split: twelve 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
@@ -189,16 +189,20 @@ UNIT_SOURCES = {
                   "is added, `peak_soc_floor` is not consulted, the SITE target "
                   "is untouched, and main.py does not file it at all at a "
                   "roster of one",
+    # SAFETY-73.
+    "refused":    "THIS battery's own reading gave nothing against THIS unit's "
+                  "own standing command, so that direction is closed on THIS "
+                  "unit and its share goes to the others. Filed at N>1 only",
 }
 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 ELEVEN claim sources are classified, and only those eleven",
+check("all TWELVE claim sources are classified, and only those twelve",
       set(CLAIM_SCOPE) == set(SITE_SOURCES) | set(UNIT_SOURCES))
 check("the six 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 five that are about a PACK, a DEVICE or one write are per-inverter",
+check("the six 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))
 check("a source nobody classified cannot be routed at all - actuator_for() "
       "raises rather than defaulting to the site, because defaulting to the "
@@ -216,7 +220,7 @@ check("with one inverter every source routes to the site, per-inverter ones "
       all(actuator_for(s, "inverter-a", ONE) == ACTUATOR_SITE for s in CLAIM_SCOPE))
 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 five "
+check("with TWO the identity breaks and the split becomes real: the six "
       "per-inverter sources route to the unit this controller drives, the six "
       "site sources stay on the grid connection",
       {s: actuator_for(s, "inverter-a", TWO) for s in CLAIM_SCOPE}

goodwe_controller/test_delivery.py

@@ -33,7 +33,7 @@ from datetime import datetime, timedelta, timezone
 
 from named_absence import given, present
 
-from app import messages
+from app import control, messages
 from app.control import Tuning
 from app.delivery import (
     DELIVERY_OPTION_DEFAULTS, DELIVERY_SHARE_FRACTION, REASON_CODE,
@@ -592,7 +592,21 @@ check("...and the SHARES now differ, which is OPT-09 and not a weakening of the
       "cards meet here from opposite sides: OPT-09 stops the ASK, this file "
       "grades the ALARM, and neither can do the other's half",
       _pw != _pf
-      and _sf[-1]["shares"][1] == 0.0 and _sw[-1]["shares"][1] == 2000.0)
+      and _sf[-1]["shares"][1] == 0.0
+      and max(r["shares"][1] for r in _sw) == 2000.0)
+# SAFETY-73. A unit that gives < 100 W for 12 cycles while asked > 500 W is a
+# battery that refuses (owner, 2026-09-27): it is closed, its share moves, and
+# SAFETY-42 is left with units that deliver something but not enough. This
+# fixture compresses `delivery_hold_s` to 1 s at 0.2 s cycles, so the report
+# still comes first here. On a site 185 s against 60 s puts the refusal first.
+_ref_at = next((i for i, r in enumerate(_sw) if r["shares"][1] == 0.0), None)
+check("SAFETY-73: the wedged arm's B, giving 0 W against 2000 W, is refused "
+      "after 12 cycles of that evidence and asked 0 W from then, the report "
+      "having come first only because this fixture's hold is 1 s",
+      _ref_at is not None and _ref_at >= control.SETTLE_CYCLES
+      and all(r["shares"][1] > 0 for r in _sw[:_ref_at])
+      and all(r["shares"][1] == 0.0 for r in _sw[_ref_at:])
+      and _cw.inverters[1].refused == 1)
 check("...and the state of charge is still what tells them apart",
       [r["soc"] for r in _sw] != [r["soc"] for r in _sf])
 print(f"       both arms, last cycle: grid {_sw[-1]['grid']:.0f} W, shares "
@@ -640,10 +654,10 @@ check("...and `account()` STILL turns this pack away if the ask does arrive - "
       account(live=True, parked=False, soc=_cold_b.soc,
               bms_floor=_cold_b.bms_soc_floor) is not None)
 check("...and the WEDGED arm's B differs from it in exactly that one place - "
-      "which is why it is still asked for its share, and still named",
+      "which is why it was asked for its share until the refusal, and named",
       account(live=True, parked=False, soc=_cw.inverters[1].soc,
               bms_floor=_cw.inverters[1].bms_soc_floor) is None
-      and float(_cw.inverters[1].standing_w()) > 0)
+      and _sw[0]["shares"][1] > 0 and _sw[-1]["named"] == ("inverter-b",))
 
 # ⚠️ WHICH FLOOR, AND THIS IS A WIRING CHECK THAT THE PURE ONES ABOVE CANNOT
 # MAKE. `peakguard.pack_floor_pct()` conjoins a pack's own cut-out with the
@@ -916,9 +930,13 @@ check("...and EVERY WATT IS THE SAME. Grid, site target, both shares and both "
       "reported or not - which is what 'it does not act' means, stated as a "
       "measurement instead of as an intention",
       _power_domain(_sw) == _power_domain(_sn))
-check("...the wedged unit is still being commanded its share after the trip, "
+_trip_at = next((i for i, r in enumerate(_sw) if r["named"]), None)
-      "not parked: this is NOT SAFETY-02's remedy and must not become it",
+check("...the wedged unit is still commanded its share after the trip, not "
-      abs(float(_cw.inverters[1].target)) > 0)
+      "parked by this guard: this is NOT SAFETY-02's remedy. What later takes "
+      "the share is SAFETY-73's refusal, which runs the same in the arm where "
+      "this guard never trips (the line above)",
+      _trip_at is not None and _ref_at is not None and _trip_at < _ref_at
+      and all(r["shares"][1] > 0 for r in _sw[_trip_at:_ref_at]))
 check("...and nothing named `delivery` reaches the arbiter - the resolution's "
       "own sentence is unchanged by the trip",
       "deliver" not in _cw.reason.lower() and _cw.reason == _cn.reason)

goodwe_controller/test_peakguard.py

@@ -2655,7 +2655,7 @@ check("...and so does the persistent notification, which outlives the page and "
 # wants from you, not an edit.
 CLAIM_SOURCES = {"device", "supervised", "failsafe", "safety", "peak",
                  "arbitrage", "maintenance", "loop", "circulation",
-                 "sign_site", "pack"}
+                 "sign_site", "pack", "refused"}
 # 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.
@@ -2697,7 +2697,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 ELEVEN. THIS "
+check("...and the sources that cycle can file are EXACTLY these TWELVE. 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 "
@@ -2848,7 +2848,8 @@ check("...and the split is the one ARB-03 argued, with SAFETY-31's and "
       {s for s, v in CLAIM_SCOPE.items() if v == SCOPE_SITE}
       == {"loop", "peak", "arbitrage", "safety", "circulation", "sign_site"}
       and {s for s, v in CLAIM_SCOPE.items() if v == SCOPE_UNIT}
-      == {"device", "supervised", "failsafe", "maintenance", "pack"})
+      == {"device", "supervised", "failsafe", "maintenance", "pack",
+          "refused"})
 
 # The shipped roster is ONE inverter, so §6.2's identity holds and every claim a
 # real cycle files is a site claim. Observed on the claims main.py's own resolve
@@ -2945,7 +2946,7 @@ check("⚠️ AND THIS IS THE BREAKING HALF. Give the same Controller a second "
       "rest of the roster goes on pushing the same way - and this line goes red",
       _routed == {"device": "inverter-a", "supervised": "inverter-a",
                   "failsafe": "inverter-a", "maintenance": "inverter-a",
-                  "pack": "inverter-a",
+                  "pack": "inverter-a", "refused": "inverter-a",
                   "loop": ACTUATOR_SITE, "peak": ACTUATOR_SITE,
                   "arbitrage": ACTUATOR_SITE, "safety": ACTUATOR_SITE,
                   "circulation": ACTUATOR_SITE, "sign_site": ACTUATOR_SITE})
@@ -3027,13 +3028,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 ELEVEN of `Controller.cycle()`'s DIRECT appends route through "
+check("all THIRTEEN 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) == 11
+      len(_direct) == 13
       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 "
@@ -3050,15 +3051,15 @@ 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. Six of the eleven appends are SCOPE_UNIT - `device`, "
+      "share. Eight of the thirteen appends are SCOPE_UNIT - `device`, "
-      "`supervised`, `failsafe`, maintenance's two and OPT-09's `pack` - and "
+      "`supervised`, `failsafe`, maintenance's two, OPT-09's `pack` and "
-      "all six pass one. "
+      "SAFETY-73's two `refused` - and all eight pass one. "
       "SAFETY-30's `sign_site` is deliberately NOT one of them: it is filed on "
       "the site, beside `circulation`, because the number that tripped it is "
       "the site target",
       _unit_sources_unrouted == []
       and len([r for r in _routing
-               if CLAIM_SCOPE.get(r[0]) == SCOPE_UNIT]) == 6)
+               if CLAIM_SCOPE.get(r[0]) == SCOPE_UNIT]) == 8)
 check("...and the only appends this pin does NOT reach are the two that file a "
       "claim BUILT ELSEWHERE - `peak_claim` and `arb_claim` - which have no "
       "`actuator=` at the call site to mis-type. Their routing is pinned where "
@@ -3067,7 +3068,7 @@ 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 twelve appends account for exactly the eleven "
+check("...and between them the fifteen appends account for exactly the twelve "
       "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())
@@ -3119,7 +3120,7 @@ check("⚠️ AND THE RATING LIMITS ARE FILED ONCE PER INVERTER, ON ITS OWN "
           ("supervised", "inverter-a"),
           ("supervised", "inverter-b")])
 check("...and naming those three is not an accident of this fixture but the "
-      "reason the static pin exists - SEVEN of the eleven sources need a state "
+      "reason the static pin exists - EIGHT of the twelve sources need a state "
       "no check in this file reaches. SAFETY-31's `circulation` is one of the "
       "seven: it is filed only while the site is MEASURED to be circulating, "
       "which no fixture here is. SAFETY-30's `sign_site` is another, and it "
@@ -3127,7 +3128,8 @@ check("...and naming those three is not an accident of this fixture but the "
       "standing opposed to the meter for its whole window",
       {c.source for c in _cl2}
       == CLAIM_SOURCES - {"failsafe", "safety", "maintenance", "peak",
-                          "arbitrage", "circulation", "sign_site"})
+                          "arbitrage", "circulation", "sign_site",
+                          "refused"})
 # ⚠️ 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_refusal_units.py

@@ -0,0 +1,480 @@
+"""Runnable check for SAFETY-73.  `python3 test_refusal_units.py`
+
+SAFETY-52 per inverter. With two or more inverters, a battery that refuses
+(gives < 100 W for 12 cycles while asked > 500 W, or trips the stale-reading
+gate on a ~0 W reading) is closed in that direction by its own P_SAFETY limit,
+so the allocator hands its share to the others. It is probed at 200 W every
+15 minutes, one inverter at a time. When every inverter refuses the same way
+the site holds 0 W. When a refusal ends the site target restarts from the watts
+on the wire. DECISIONS/SAFETY-73 has the record.
+
+Drives the REAL `Controller` (read_inputs + cycle, the real freshness gate,
+allocator, OPT-09 pack claim and SAFETY-42 detector) with two inverters in
+simulated time. At the cut-out a battery reads -4 W for any request >= -4 W
+and accepts charge. Two register shapes: FLAT at -4 W (the 09-06 capture) and
+JITTER +-3 W (the gate cannot arm). `OLD` runs the same fixture with the
+per-unit refusal switched off, which is 652097a's behaviour at N>1.
+
+Mutations (MUTATION.md, `python3 -B`, copies purged of `__pycache__`), graded
+over this file / test_delivery / test_refusal. Every one is red here:
+
+  #    mutation                                                   reds
+  ---  ---------------------------------------------------------  -------
+  U1   main.py: the per-unit engage clause never fires              7/2/0
+  U2   main.py: the N>1 stale-reading seed returns at once          14/0/0
+  U3   main.py: no `refused` claim is filed                         13/2/0
+  U4   main.py: a probe never pins (always the closed limit)        7/0/0
+  U5   main.py: a probe pin stands whatever the site target         1/0/0
+  U6   main.py: no reseed of the site target when a refusal ends    1/0/0
+  U7   main.py: no site hold when every inverter refuses            2/0/0
+  U8   main.py: the other-direction exit removed (charge counter)   1/0/0
+  U9   main.py: the delivery exit removed                           3/0/0
+  U10  main.py: probes are no longer one at a time                  1/0/0
+  U11  main.py: the above-cut-out WARNING condition inverted        1/0/0
+  U12  main.py: SAFETY-42's account no longer sees a small ask      1/0/0
+  U13  main.py: a unit engages above `saturation_w`, not PROBE_W    1/0/0
+
+On 652097a itself this file stops at the first sample (`Inverter` has no
+`refused`); its OLD arms reproduce that tree's N>1 behaviour instead, and
+every PRECONDITION line is graded against them.
+"""
+
+import asyncio
+import logging
+import sys
+from datetime import datetime, timedelta, timezone
+
+from app import control
+from app.control import PROBE_EVERY_S, PROBE_S, PROBE_W
+from app.main import Controller
+from app.arbiter import P_LOOP, P_PEAK, P_SAFETY, Claim, resolve
+import app.main as m
+
+fails = []
+
+
+def check(name, cond):
+    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
+    if not cond:
+        fails.append(name)
+
+
+TICK = 5.0
+CUT_W = -4.0
+H = 3600.0
+BASE = datetime(2026, 9, 6, 19, 30, tzinfo=timezone.utc)
+UNITS = ("a", "b")
+
+
+class _Store:
+    def __init__(self):
+        from app.store import DEFAULTS
+        self.data = dict(DEFAULTS)
+
+    def set(self, k, v):
+        self.data[k] = v
+
+    def get_time(self, *a):
+        return None
+
+
+class _Pub:
+    enabled = False
+
+    def publish(self, *a, **k):
+        pass
+
+    def close(self):
+        pass
+
+
+class _Hass:
+    """A house, one meter, and two batteries that follow unless cut out."""
+
+    def __init__(self, load, cut, soc, jitter):
+        self.load, self.cut, self.soc, self.jitter = load, cut, soc, jitter
+        self.cmd = {u: 0.0 for u in UNITS}
+        self.t = 0.0
+
+    def batt(self, u):
+        c = self.cmd[u]
+        if self.cut[u](self.t) and c >= CUT_W:
+            if self.jitter:
+                return CUT_W + ((int(self.t / TICK) % 3) - 1) * 3.0
+            return CUT_W
+        return float(round(c))
+
+    def grid(self):
+        return self.load(self.t) - sum(self.batt(u) for u in UNITS)
+
+    async def limits(self, e):
+        return (-2000.0, 2000.0, 1.0)
+
+    async def call(self, *a):
+        return True
+
+    async def number(self, e, invert=False):
+        if e == "sensor.grid":
+            return self.grid()
+        u = e[-1]
+        if e.startswith("sensor.soc"):
+            v = self.soc[u]
+            return v(self.t) if callable(v) else v
+        if e.startswith("sensor.batt"):
+            return self.batt(u)
+        return None
+
+    async def set_number(self, e, v):
+        self.cmd[e[-1]] = float(v)
+        return True
+
+
+def _off(c):
+    """652097a's N>1 behaviour: nothing refuses per unit."""
+    c._unit_refusals = lambda: None
+    c._unit_refusal_start = lambda *a, **k: None
+
+
+def run(load, cut, dur, *, soc=None, jitter=False, old=False, events=None,
+        extra=None):
+    """Real cycles every TICK of simulated time. Rows are dicts per cycle."""
+    async def go():
+        hass = _Hass(load, cut, soc or {"a": 50.0, "b": 50.0}, jitter)
+        inv = [{"name": f"inverter-{u}", "setpoint_entity": f"number.sp_{u}",
+                "soc_entity": f"sensor.soc_{u}",
+                "batt_entity": f"sensor.batt_{u}",
+                "max_w": 2000, "bms_soc_floor": 10} for u in UNITS]
+        o = {"meter_entity": "sensor.grid", "meter_source": "off",
+             "inverters": inv, "gain": 0.6, "slew_w": 400,
+             "target_grid_w": -10, "deadband_w": 15, "step_w": 10,
+             "heartbeat_s": 10, "stale_input_s": 15, "auto_start": True}
+        o.update(extra or {})
+        c = Controller(o, hass, _Store(), _Pub(), stop=asyncio.Event())
+        for i in c.inverters:
+            i.dev_min, i.dev_max = -2000.0, 2000.0
+        c.auto = True
+        if old:
+            _off(c)
+        real_dt = m.datetime
+        clock = [0.0]
+
+        class _DT(real_dt):
+            @classmethod
+            def now(cls, tz=None):
+                return BASE + timedelta(seconds=clock[0])
+
+        m.datetime = _DT
+        rows = []
+        trips = {u: 0 for u in UNITS}
+        was = {u: False for u in UNITS}
+        pending = dict(events or {})
+        try:
+            t = 0.0
+            while t < dur:
+                hass.t = clock[0] = t
+                for et in sorted(pending):
+                    if et <= t:
+                        pending.pop(et)(c)
+                await c.read_inputs(now_mono=t)
+                for u, i in zip(UNITS, c.inverters):
+                    s = i.batt_fresh.stale
+                    trips[u] += s and not was[u]
+                    was[u] = s
+                await c.cycle()
+                rows.append(dict(
+                    t=t, grid=hass.grid(), site=c.target, reason=c.loop_reason,
+                    a=hass.cmd["a"], b=hass.cmd["b"],
+                    ra=c.inverters[0].refused, rb=c.inverters[1].refused,
+                    da=c.inverters[0].delivery_wd.tripped,
+                    db=c.inverters[1].delivery_wd.tripped))
+                t += TICK
+        finally:
+            m.datetime = real_dt
+        return rows, trips, c
+    return asyncio.run(go())
+
+
+def after(rows, t0, t1=None):
+    return [r for r in rows if r["t"] >= t0 and (t1 is None or r["t"] < t1)]
+
+
+def mean(rows, k):
+    return sum(r[k] for r in rows) / len(rows) if rows else float("nan")
+
+
+def never(t):
+    return False
+
+
+def cut_at(t0, t1=None):
+    return lambda t: t >= t0 and (t1 is None or t < t1)
+
+
+def probes(rows, k):
+    """Start times of runs where refused unit `k` is asked exactly PROBE_W."""
+    out, prev = [], False
+    for r in rows:
+        on = abs(r[k]) == PROBE_W and r["r" + k] != 0
+        if on and not prev:
+            out.append(r["t"])
+        prev = on
+    return out
+
+
+SHAPES = (("flat", False), ("jitter", True))
+
+# ==============================================================================
+print("1. case a: house 3000 W, both give power, A cuts out at 120 s")
+# ==============================================================================
+for shape, jit in SHAPES:
+    new, ntr, cn = run(lambda t: 3000.0, {"a": cut_at(120), "b": never}, H,
+                       jitter=jit)
+    old, otr, _ = run(lambda t: 3000.0, {"a": cut_at(120), "b": never}, H,
+                      jitter=jit, old=True)
+    tail_n, tail_o = after(new, 300), after(old, 300)
+    held = [r for r in tail_n if abs(r["a"]) != PROBE_W]
+    print(f"    {shape}: old B {mean(tail_o, 'b'):.0f} W grid "
+          f"{mean(tail_o, 'grid'):+.0f} W trips A {otr['a']}; new B "
+          f"{mean(tail_n, 'b'):.0f} W grid {mean(tail_n, 'grid'):+.0f} W "
+          f"trips A {ntr['a']}")
+    check(f"{shape}: PRECONDITION: on the old law B never takes over (mean "
+          f"under 1950 W) while A gives nothing",
+          mean(tail_o, "b") < 1950 and mean(tail_o, "grid") > 1050)
+    check(f"{shape}: A is refused and asked 0 W outside its probes, B carries "
+          f"its rating, and the grid imports only what B cannot give",
+          cn.inverters[0].refused == 1 and all(r["a"] == 0.0 for r in held)
+          and min(r["b"] for r in tail_n) == 2000.0
+          and abs(mean(tail_n, "grid") - 1004) < 20)
+    check(f"{shape}: A's stale-reading gate trips at most once, the site is "
+          f"never stopped after the first 5 minutes, and SAFETY-42 does not "
+          f"report A as not delivering",
+          ntr["a"] <= 1
+          and all(r["reason"] != "not-running" for r in tail_n)
+          and not any(r["da"] for r in new))
+
+# ==============================================================================
+print("2. case b: house 1200 W, which B alone can carry, A cuts out")
+# ==============================================================================
+for shape, jit in SHAPES:
+    new, ntr, cn = run(lambda t: 1200.0, {"a": cut_at(120), "b": never}, H,
+                       jitter=jit)
+    old, _, _ = run(lambda t: 1200.0, {"a": cut_at(120), "b": never}, H,
+                    jitter=jit, old=True)
+    tail_n, tail_o = after(new, 300), after(old, 300)
+    imp = sum(max(0.0, r["grid"]) for r in tail_n) * TICK / H
+    print(f"    {shape}: old grid {mean(tail_o, 'grid'):+.0f} W, new grid "
+          f"{mean(tail_n, 'grid'):+.0f} W, new import {imp:.0f} Wh")
+    check(f"{shape}: PRECONDITION: the old law leaves a standing import",
+          mean(tail_o, "grid") > 250)
+    check(f"{shape}: B takes over the whole house and the grid sits near its "
+          f"target", abs(mean(tail_n, "grid") + 10) < 25 and imp < 10)
+    pa = probes(new, "a")
+    in_probe = [r for r in new
+                if any(p <= r["t"] < p + PROBE_S for p in pa)]
+    check(f"{shape}: A is probed at 200 W every 15 minutes while B carries "
+          f"the rest, and the two never pull opposite ways",
+          len(pa) >= 3
+          and all(abs(q - p - PROBE_EVERY_S - PROBE_S) <= 2 * TICK
+                  for p, q in zip(pa, pa[1:]))
+          and all(r["b"] > 0 for r in in_probe)
+          and all(r["a"] * r["b"] >= 0 for r in new))
+
+# ==============================================================================
+print("3. case c: both cut out; the site holds 0 W, probes one at a time")
+# ==============================================================================
+for shape, jit in SHAPES:
+    new, ntr, cn = run(lambda t: 3000.0, {"a": cut_at(120), "b": cut_at(150)},
+                       1.5 * H, jitter=jit)
+    old, _, _ = run(lambda t: 3000.0, {"a": cut_at(120), "b": cut_at(150)},
+                    1.5 * H, jitter=jit, old=True)
+    tail_n, tail_o = after(new, 600), after(old, 600)
+    pa, pb = probes(new, "a"), probes(new, "b")
+    print(f"    {shape}: old mean asks A/B {mean(tail_o, 'a'):.0f}/"
+          f"{mean(tail_o, 'b'):.0f} W; new {mean(tail_n, 'a'):.0f}/"
+          f"{mean(tail_n, 'b'):.0f} W, probes A {len(pa)} B {len(pb)}")
+    check(f"{shape}: PRECONDITION: the old law keeps asking two empty "
+          f"batteries", mean(tail_o, "a") > 150 and mean(tail_o, "b") > 150)
+    check(f"{shape}: the site holds 0 W: the loop says `battery refuses` or "
+          f"`battery probe` and each battery is asked 0 W or its probe",
+          all(r["reason"] in (control.REFUSED, control.PROBING)
+              for r in tail_n)
+          and all(r["a"] in (0.0, PROBE_W) and r["b"] in (0.0, PROBE_W)
+                  for r in tail_n))
+    check(f"{shape}: both batteries are probed, never at the same time",
+          len(pa) >= 3 and len(pb) >= 3
+          and not any(r["a"] and r["b"] for r in new if r["t"] > 300))
+
+# ==============================================================================
+print("4. recovery: the reseed stops the parked site target reaching the grid")
+# ==============================================================================
+# Both cut, the house drops to 940 W at 1 h, A recovers at 1.5 h with no sign
+# change. Phase A, with the envelope but no reseed and no site hold: -1059 W.
+ld = (lambda t: 3000.0 if t < H else 940.0)
+for shape, jit in SHAPES:
+    rows, _, c = run(ld, {"a": cut_at(120, 1.5 * H), "b": cut_at(150)},
+                     2 * H, jitter=jit)
+    win = after(rows, 1.5 * H, 1.5 * H + 900)
+    worst = min(r["grid"] for r in win)
+    print(f"    {shape}: worst grid after A recovers {worst:+.0f} W; A "
+          f"refused {c.inverters[0].refused}")
+    check(f"{shape}: A's recovery is found (at its next probe at the latest) "
+          f"and never exports more than 250 W",
+          c.inverters[0].refused == 0 and worst > -250
+          and any("BATTERY REFUSES ENDED inverter-a" in e for e in c.events))
+
+# Case a: B at its rating, the site target parked at 3600 W while the house
+# still imports; the house falls to 2500 W and A recovers. Without the reseed A
+# is handed the parked 1600 W at once, 1100 W more than the house needs.
+ld2 = (lambda t: 3000.0 if t < 0.5 * H else 2500.0)
+rows, _, c = run(ld2, {"a": cut_at(120, 0.75 * H), "b": never}, 1.5 * H)
+win = after(rows, 0.75 * H, 0.75 * H + 900)
+print(f"    a: worst grid when A recovers {min(r['grid'] for r in win):+.0f} W")
+check("case a: a recovery hands A no parked site target (worst export under "
+      "250 W)", min(r["grid"] for r in win) > -250
+      and any("BATTERY REFUSES ENDED inverter-a" in e for e in c.events))
+
+# ==============================================================================
+print("5. exits, words and the circulation guard")
+# ==============================================================================
+# Solar: the house exports from 30 min, so the loop wants to charge.
+rows, _, c = run(lambda t: 3000.0 if t < 0.5 * H else -2500.0,
+                 {"a": cut_at(120), "b": never}, H)
+check("the grid asking the other direction lets A charge at once (its limit "
+      "closes only the refused direction) and a minute of charge ends the "
+      "refusal",
+      c.inverters[0].refused == 0 and min(r["a"] for r in after(rows, 0.6 * H))
+      < -500 and any("the other way" in e for e in c.events))
+
+
+def _stop(c):
+    c.auto = False
+
+
+rows, _, c = run(lambda t: 3000.0, {"a": cut_at(120), "b": never}, 0.5 * H,
+                 events={1200.0: _stop})
+check("stopping control ends the refusal",
+      c.inverters[0].refused == 0
+      and any("control stopped" in e for e in c.events))
+
+# A small house: the site asks under 200 W, so a probe may not pin A (B would
+# have to charge while A gives power). The probe waits instead.
+rows, _, c = run(lambda t: 3000.0 if t < 600 else 120.0,
+                 {"a": cut_at(120), "b": never}, 1.2 * H)
+check("with the site asking under 200 W no probe is sent and no inverter is "
+      "ever asked the opposite way to the other",
+      c.inverters[0].refused == 1
+      and probes(after(rows, 700), "a") == []
+      and all(r["a"] * r["b"] >= 0 for r in rows)
+      and not any("ARBITER" in e for e in c.events))
+
+# The hold binds every strategy's ask: a peak-guard SET at P_PEAK on the site
+# is still placed around a refusing unit's closed limit.
+_res = resolve([Claim.set("peak", P_PEAK, 3000.0, "quarter peak"),
+                Claim.set("loop", P_LOOP, 500.0, "tracking"),
+                Claim.limit("supervised", P_SAFETY, -2000.0, 2000.0,
+                            actuator="inverter-a"),
+                Claim.limit("supervised", P_SAFETY, -2000.0, 2000.0,
+                            actuator="inverter-b"),
+                Claim.limit("refused", P_SAFETY, hi=0.0,
+                            reason=control.REFUSED, actuator="inverter-a")],
+               units=("inverter-a", "inverter-b"),
+               live=("inverter-a", "inverter-b"))
+check("a P_SAFETY refusal limit binds a higher-priority strategy too: the "
+      "peak guard's 3000 W goes to B, A gets 0 W",
+      _res.shares.get("inverter-a") == 0 and _res.shares.get("inverter-b")
+      == 2000)
+
+
+class _Grab(logging.Handler):
+    def __init__(self):
+        super().__init__()
+        self.lines = []
+
+    def emit(self, record):
+        self.lines.append((record.levelno, record.getMessage()))
+
+
+_g = _Grab()
+_lg = logging.getLogger("goodwe")
+_lv = _lg.level
+_lg.setLevel(logging.INFO)
+_lg.addHandler(_g)
+run(lambda t: 3000.0, {"a": cut_at(120), "b": never}, 600)
+run(lambda t: 3000.0, {"a": cut_at(120), "b": never}, 600,
+    soc={"a": 10.0, "b": 50.0})
+_lg.removeHandler(_g)
+_lg.setLevel(_lv)
+warns = [x for lv, x in _g.lines if lv == logging.WARNING
+         and "REFUSES_ABOVE_CUT_OUT" in x]
+infos = [x for lv, x in _g.lines if lv == logging.INFO
+         and "BATTERY_REFUSES" in x]
+print(f"    {warns} / {infos}")
+check("a refusal above the configured cut-out level logs INFO and one WARNING "
+      "naming the inverter, and a battery at its cut-out logs neither "
+      "(OPT-09's limit closes it first)",
+      len(warns) == 1 and warns[0].startswith("inverter-a ")
+      and len(infos) == 1 and infos[0].startswith("inverter-a:")
+      and len(warns[0].split()) <= 25)
+
+# A night load of 400 W: each unit is asked about 400 W once the loop has
+# made up A's missing half, under `saturation_w`. Engaging at PROBE_W is what
+# refuses A here; at 500 W it would keep a phantom share that SAFETY-43 names.
+rows, _, c = run(lambda t: 400.0, {"a": cut_at(0), "b": never}, 1200)
+check("a 400 W night load: A is refused at a share under 500 W, B carries the "
+      "house and nothing reports A as not delivering",
+      c.inverters[0].refused == 1 and not any(r["da"] for r in rows)
+      and abs(mean(after(rows, 300), "grid") + 10) < 25
+      and mean(after(rows, 300), "b") > 380)
+
+# A night load of 150 W: A's share never passes PROBE_W, so it is never
+# refused, and SAFETY-43 must not name it as not delivering either: a battery
+# asked under 200 W that gives under 100 W is a refusal too small to call.
+rows, _, c = run(lambda t: 150.0, {"a": cut_at(0), "b": never}, 1200)
+check("a small share on an empty battery is neither refused nor reported as "
+      "not delivering, and the grid stays near its target",
+      c.inverters[0].refused == 0 and not any(r["da"] for r in rows)
+      and abs(mean(after(rows, 300), "grid") + 10) < 25)
+
+rows, _, c = run(lambda t: 3000.0, {"a": cut_at(120), "b": never}, 600)
+notes = c.liveness_notes()
+check("the status page names the refusing inverter",
+      any(n.startswith("inverter-a: the battery refuses to give power")
+          for n in notes))
+
+# Shadow mode writes nothing, so nothing is asked and nothing can refuse.
+rows, _, c = run(lambda t: 3000.0, {"a": cut_at(120), "b": never}, 1200,
+                 extra={"shadow_mode": True})
+check("shadow mode: no inverter is refused and both stay at 0 W",
+      c.inverters[0].refused == 0 and c.inverters[1].refused == 0
+      and all(r["a"] == 0.0 and r["b"] == 0.0 for r in rows))
+
+# ==============================================================================
+print("6. one inverter is untouched")
+# ==============================================================================
+
+
+def _boom(*a, **k):
+    raise AssertionError("SAFETY-73 code ran at N=1")
+
+
+_saved = {k: getattr(Controller, k) for k in (
+    "_unit_refusals", "_unit_refusal_start", "_unit_refusal_end",
+    "_unit_probe", "_site_refused")}
+for k in _saved:
+    setattr(Controller, k, _boom)
+try:
+    import test_refusal as _t1
+    _one_ok = not _t1.fails
+except (AssertionError, SystemExit) as e:
+    _one_ok = isinstance(e, SystemExit) and not e.code
+finally:
+    for k, v in _saved.items():
+        setattr(Controller, k, v)
+check("SAFETY-52's whole single-inverter suite passes with every SAFETY-73 "
+      "method made to raise, so none of it runs at N=1", _one_ok)
+
+print()
+if fails:
+    print(f"{len(fails)} FAILED: " + ", ".join(fails))
+    sys.exit(1)
+print("all checks passed")