Merge pull request 'Deploy 0.14.0 to master' (#170) from deploy-0.14.0 into master

Deploy 0.14.0 to master (#170)

.gitea/infra-pin

@@ -1,2 +1,2 @@
 # The admin/goodwe commit CI tests against. Bump: set to the full SHA of admin/goodwe release/1.0 HEAD (README.md, "CI and the infra pin").
-d637d0b1b13f1b863dc47e445e2ca4cf4a82fe10
+0c04488a9194039c2bf851b680a8ffc53beff671

goodwe_controller/CHANGELOG.md

@@ -1,5 +1,42 @@
 # Changelog
 
+## 0.14.0
+
+**A minor bump: new behaviour with two inverters, and no option changes.**
+No option was added, removed or renamed, and no default changed (83
+options). Options that start 0.13.0 also start 0.14.0. With one inverter,
+control is the same as in 0.13.0.
+
+⭐ **Two inverters are now supported** for the monthly maintenance (drain,
+charge, hold) and the peak guard. To add a second inverter, follow
+`docs/commissioning-b.md` in the `admin/goodwe` repo. Its Stage 5, where the
+add-on takes control of it, needs 0.14.0. Read the known issues below first:
+three of them are about two inverters.
+
+Contents of this release, in merge order since 0.13.0 (`8c438ab`):
+
+| | |
+|---|---|
+| `f3e409c` | Two inverters: a battery at or under its peak reserve gives last, after the others give their full power. It is freed again above reserve + 1 point (e.g. above 21 % for a 20 % reserve). At a quarter's edge the peak guard no longer drops out for a second ([[SAFETY-91]]) |
+| `858f77c` | Two inverters: when a maintenance charge ends, for any reason, the other battery no longer starts charging from the grid; control hands over in small steps ([[SAFETY-93]]) |
+| `e51e3e0` | CI tests against a newer `admin/goodwe` (CI only) ([[CI-13]]) |
+
+### Known issues
+
+- A false "Battery is not keeping up" can still show when the peak guard pauses a maintenance charge ([[SAFETY-92]]).
+- A healthy battery that turns from charging to giving can be held at 0 W for up to 15 min ([[SAFETY-94]]).
+- After a gap in the P1 meter, the peak guard is blind for the rest of that quarter hour ([[SAFETY-90]]).
+- Two inverters: if one battery's drain stops early (e.g. switched Off), the other battery can jump to about 1.6 kW at once, and the surplus goes to the grid ([[SAFETY-95]]).
+- Two inverters: after one battery's link drops, control can stall until a safety stop ([[SAFETY-96]]).
+- Two inverters: the maintenance charge can switch on and off while the house is above the peak aim ([[SAFETY-88]]).
+
+### If you are crossing from a version older than 0.10.0
+
+`## 0.10.0` below can refuse an old config: `slew_w` above its limit, and the
+withdrawn `price_provider` ids `engie_dynamic`, `totalenergies_dynamic` and
+`eneco_eddy`. Read it before you update. Run `tools/predeploy_check.py` with the
+bare options (`jq '.data.options'`) first.
+
 ## 0.13.0
 
 **A minor bump: new behaviour, a new add-on page (Home, Settings, Batteries,

goodwe_controller/DECISIONS/SAFETY-93-charge-end.md

@@ -0,0 +1,60 @@
+# SAFETY-93 - the maintenance charge end hands back smoothly
+
+**Status: implemented 2026-10-02, for review.** Tests: `test_safety93.py`.
+
+**The defect** (SAFETY-85 rig test, check 4; also on 9358781). At N=2, when
+A's maintenance charge ended, B went 0 -> -1600 W in one cycle and charged
+from the grid with the house at 6000 W. The HOLD end was already smooth
+(SAFETY-85).
+
+**Cause.** During the charge the site target holds A's charge (SAFETY-84's
+`charge` ceiling, -2000 W). At the end A gets its HOLD pin (0 W), but the
+loop starts its next step from last cycle's site target. The slew limit
+moves the site total by 400 W (-2000 -> -1600), and the only battery left
+to place it on is B. The same happens when the charge times out or the
+battery is switched Off (A free, its -2000 W shared out). With the charge
+suspended by the guard and A giving under the loop, B took A's share in
+one step (675 W).
+
+**The fix** (`main.py`, `_reseed_charge_end()`, N>=2 only). On the cycle a
+battery leaves CHARGE, the loop's start point (`self.target`, and the
+integrator with it) is what the other batteries stood at plus that
+battery's new command: its pin (HOLD or abort, 0 W), or, if it is free, its
+last discharge (a charge stops). The loop then moves from there one slew
+step a cycle. The ending battery's own step (-2000 -> 0 W) is maintenance's
+pin, as before; it lowers the grid import. The end is kept until the loop
+next runs, so a missed poll on that cycle (the input hold) does not lose it.
+
+Review: a charge also ends, for the site, when the battery is still in
+CHARGE but can no longer be placed: a stale link or a refusal (SAFETY-73)
+drops it, and its -2000 W landed on B the same way (B 0 -> -1600 W at a
+1200 W house). So the end is noted after the unit claims are filed: a
+battery leaves the charge when it leaves CHARGE, is not live, or its own
+claims contradict. The guard suspending the charge (SAFETY-84) is no end.
+
+Why a start-point reseed, and not a per-battery slew in the allocator: the
+jump comes from one stale number at one edge, the kind of edge SAFETY-73
+(`_unit_refusal_end`) and OPT-11 P3 (`_reseed_run_edge`) already reseed.
+The change stays at that edge; nothing else at N>=2 moves.
+
+**Before -> after** (`test_safety93.py`, 2 x 2500 W, 10 s lag, 2 s meter):
+
+| case | before | after |
+|---|---|---|
+| rig case: house 6000, A charges 2000 to the end | B 0 -> -1600, charges 4 cycles | B 0 -> 400 -> 800 ..., never charges |
+| house 1200 | B 0 -> -1600 (2800 W from the grid) | B steps of 400, never charges |
+| charge times out / A switched Off | B 0 -> -1600 | B 0 -> 400 |
+| guard suspended the charge, A gives 1825 | B step 675 W | B step 395 W |
+| missed poll on the end cycle | B 0 -> -1600 a cycle later | B 0 -> 400 |
+| A dropped mid-charge (stale link, refusal), house 1200 | B 0 -> -1600 | B 0 -> 400 |
+| N=1, both charge ends above | | byte-identical |
+
+No pinned digest moved: the 46 digests of `tools/*identity*.py` (N=1 and
+N=2) are byte-identical to f3e409c; none of their scenarios reaches a
+charge end.
+
+**Not changed.** The peak guard's floor still moves the site in one step
+when it engages (a house step, a quarter edge): the guard outranks the slew
+(owner order). A HOLD pin still drops the holding battery to 0 W at once
+and SAFETY-86 gives way to the others first. SAFETY-88 (charge flapping
+above the aim) is untouched.

goodwe_controller/DOCS.md

@@ -1677,6 +1677,8 @@ While a full battery rests after its charge (the hold), it gives just enough
 to keep the 15-minute peak at that aim if the house needs it; with two
 batteries the other one gives first. The hold still ends on its clock
 (SAFETY-86).
+When a charge ends, the other batteries take over one slew step a cycle,
+and none starts charging from the grid (SAFETY-93).
 
 #### Capacity-tariff exposure of the grid-following loop
 

goodwe_controller/app/main.py

@@ -75,8 +75,8 @@ from .hass import HomeAssistant
 from .inverter import (LINK_STALE_S, UNIT_A, build_roster,
                        roster_entries, roster_names)
 from .maintenance import (
-    DRAIN, HOLD, IDLE, POWER_MOVING, MaintConfig, Maintenance, MaintResult,
-    arbitrate as maint_arbitrate,
+    CHARGE, DRAIN, HOLD, IDLE, POWER_MOVING, MaintConfig, Maintenance,
+    MaintResult, arbitrate as maint_arbitrate,
 )
 from .mqtt import (MODE_CONFIGS_KEY, MODE_LABELS, USE_OFF, USE_ON,
                    MqttPublisher, announced_units, mode_configs,
@@ -994,6 +994,10 @@ class Controller:
         self._drain_waits = set()
         # SAFETY-91: last cycle's held-back batteries (`_held_back()`).
         self._held_was = frozenset()
+        # SAFETY-93: the batteries placed in their maintenance charge last
+        # cycle, and those whose charge ended since the loop last ran.
+        self._charging_was = frozenset()
+        self._charge_ended = frozenset()
         # 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.
@@ -2368,6 +2372,39 @@ class Controller:
         self.target = want
         self.i_w = want - self.tuning.gain * (want - float(batt))
 
+    def _charge_track(self, claims, maint_results) -> None:
+        """SAFETY-93, N>=2: a charge ends when its battery leaves CHARGE or can
+        no longer be placed (stale link, refusal). Kept until the loop runs."""
+        if len(self.inverters) < 2:
+            return
+        live = set(self.live_units())
+        now = set()
+        for inv, res in zip(self.inverters, maint_results):
+            u = self.actuator_of("maintenance", inv)
+            if res.phase == CHARGE and u in live and not pre_resolve(
+                    [c for c in claims if c.actuator == u]).contradiction:
+                now.add(inv.name)
+        self._charge_ended |= self._charging_was - now
+        self._charging_was = frozenset(now)
+
+    def _reseed_charge_end(self, maint_results) -> None:
+        """SAFETY-93, N>=2: when a battery's maintenance charge ends, the loop
+        starts from what the others stood at plus that battery's new command
+        (its HOLD pin, or its last discharge if it is free; a charge stops).
+        Else the charge's watts stay in the site target and land on the
+        others: B 0 -> -1600 W in one cycle. DECISIONS/SAFETY-93-charge-end.md.
+        """
+        ended, self._charge_ended = self._charge_ended, frozenset()
+        if len(self.inverters) < 2 or not ended:
+            return
+        for inv, res in zip(self.inverters, maint_results):
+            if inv.name not in ended:
+                continue
+            now_w = (float(res.setpoint_w or 0.0) if res.owns_setpoint
+                     else max(float(inv.target), 0.0))
+            self.target += now_w - float(inv.target)
+        self.i_w = self.target
+
     def _mode_fault_log(self, resolution) -> None:
         """OPT-11: say once, per battery, that its mode was dropped as a fault.
 
@@ -4556,6 +4593,7 @@ class Controller:
         if self.modes_on:
             mode_trade = self._mode_trade(claims, now)
         self._mode_trade_was = mode_trade
+        self._charge_track(claims, maint_results)
         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
@@ -4582,6 +4620,8 @@ class Controller:
             # is what seeds it, and dropping it there restarts the ~70 s cycle.
             if self.loop_w is not None and self.target != self.loop_w:
                 self._refusal_end("another writer")
+            # SAFETY-93: a charge that ended leaves the loop's start point.
+            self._reseed_charge_end(maint_results)
             probe = self._refusal_probe()
             if len(self.inverters) > 1:
                 probe = any(_probing.values())

goodwe_controller/config.yaml

@@ -1,5 +1,5 @@
 name: GoodWe RS485 Controller
-version: "0.13.0"
+version: "0.14.0"
 slug: goodwe_controller
 description: >-
   Drives a GoodWe ES/BP battery inverter over RS485 by emulating its smart

goodwe_controller/test_safety93.py

@@ -0,0 +1,422 @@
+"""Runnable check for SAFETY-93: the maintenance charge end hands back
+smoothly, with no grid charging at a high house. `python3 test_safety93.py`
+
+At N=2, when battery A's maintenance charge ended, its watts stayed in the
+loop's site target and landed on B: B went 0 -> -1600 W in one cycle and
+charged from the grid with the house at 6000 W (SAFETY-85 rig test, check 4).
+Now the loop starts from what B stood at plus A's new command. Drives the
+REAL Controller on a lagging battery (10 s, 2 s meter, as on the rig).
+DECISIONS/SAFETY-93-charge-end.md.
+
+1. The rig case: A charges 2000 W late in a quarter with the house at 6000 W,
+   then its charge ends (HOLD).
+2. The same at a low house.
+3. The charge ends another way: a timeout (abort) and the battery switched
+   Off.
+4. The charge ends while the guard has suspended it (A gives under the loop).
+5. N=1 is untouched; a missed poll on the end cycle; the rule at its
+   boundaries.
+
+Mutations (MUTATION.md, purged copies) are listed in the pull request.
+"""
+
+import hashlib
+import logging
+import math
+import os
+import sys
+from datetime import timedelta
+
+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 safety79_identity as s79                                # noqa: E402
+import safety84_identity as s84                                # noqa: E402
+from app import packmode                                       # noqa: E402
+from app.main import Controller                                # noqa: E402
+from app.maintenance import (CHARGE_TIMEOUT_MIN, HOLD, IDLE,    # noqa: E402
+                             MaintResult)
+from app.store import unit_key                                 # noqa: E402
+from named_absence import first, present                       # noqa: E402
+
+fails = []
+A = "inverter-a"
+SLEW = 400.0            # the rig's slew_w
+AIM = 3500 - 50         # peak_cap_w - peak_margin_w
+OPT0 = dict(s84.OPT)
+
+
+def check(name, cond):
+    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
+    if not cond:
+        fails.append(name)
+
+
+def lagging(fn, lag=10.0):
+    """plantsim's battery (800 W/s ramp, `lag` s first-order lag) at the
+    live 2 s meter tick (test_safety85.py part 7)."""
+    real_p, real_t = rig._Hass.power, rig.TICK
+
+    def power(self, u):
+        st = self.__dict__.setdefault("_plant", {})
+        a, tgt, tl = st.get(u, (0.0, 0.0, self.t))
+        for _ in range(int(round((self.t - tl) / 0.1))):
+            d = self.cmd[u] - tgt
+            tgt += math.copysign(min(abs(d), 80.0), d)
+            a += (tgt - a) * (1 - math.exp(-0.1 / lag))
+        st[u] = (a, tgt, self.t)
+        return float(round(a))
+    rig._Hass.power, rig.TICK = power, 2.0
+    try:
+        return fn()
+    finally:
+        rig._Hass.power, rig.TICK = real_p, real_t
+
+
+def run(n, house, end, *, step=None, dur=None, store=None, extra=None,
+        before=None):
+    """A charging in a cheap window from the first cycle (quarters start at
+    0, 900, 1800 s); the house is 1200 W, then `house` from `step`; A reads
+    99 % (its charge target) from `end`. Returns (rows, controller)."""
+    step = end if step is None else step
+    s84.OPT = dict(OPT0, peak_cap_w=3500)
+    feed = s84.feed(True)
+
+    def both(c, t):
+        if before is not None:
+            before(c, t)
+        feed(c, t)
+    try:
+        return lagging(lambda: s84.run(
+            n, load=lambda t: house if t >= step else 1200.0, cheap=True,
+            store=store or s79.store_charging(n),
+            soc=lambda u, t: (99.0 if t >= end else 90.0) if u == "a"
+            else 60.0, dur=dur or end + 300.0, before=both,
+            extra=dict({"maintenance_hold_min": 600}, **(extra or {}))))
+    finally:
+        s84.OPT = OPT0
+
+
+def cmd(r):
+    return dict(r[2])
+
+
+def moves(rows, u, lo, hi):
+    """The largest change in battery `u`'s command between cycles in
+    (lo, hi]."""
+    return max(abs(cmd(b)[u] - cmd(a)[u]) for a, b in zip(rows, rows[1:])
+               if lo < b[0] <= hi)
+
+
+def charging_hi(rows, lo, hi, house_at):
+    """Cycles in [lo, hi] where a battery charges while the house is above
+    the peak aim."""
+    return sum(1 for r in rows if lo <= r[0] <= hi and house_at(r[0]) > AIM
+               and min(cmd(r).values()) < 0.0)
+
+
+# None on a tree without the fix (f3e409c), where this file runs red.
+no_fix = getattr(Controller, "_reseed_charge_end", None)
+
+
+def without(fn):
+    """Run `fn` with the SAFETY-93 reseed taken out."""
+    if no_fix is None:
+        return fn()
+    Controller._reseed_charge_end = lambda self, ended: None
+    try:
+        return fn()
+    finally:
+        Controller._reseed_charge_end = no_fix
+
+
+def phase_at(rows, t):
+    return first((r[-1][3][0] for r in rows if r[0] == t), f"row {t}")
+
+
+def at(rows, t, u):
+    """Battery `u`'s command at `t` s."""
+    r = first((r for r in rows if r[0] == t), f"row {t}")
+    return dict(r[2])[u] if present(r) else r
+
+
+# ============================================================================ #
+print("1. the rig case: A charges 2000 W late in a quarter, house 6000 W")
+# House 1200 W until 1760 s, then 6000 W; the quarter (900-1800 s) has room,
+# so the charge stays at 2000 W. A is full at 1780 s. Graded to 1798 s: the
+# guard grades a new quarter from 1800 s.
+R1 = run(2, 6000.0, 1780.0, step=1760.0)
+R1_old = without(lambda: run(2, 6000.0, 1780.0, step=1760.0))
+
+
+def hi1(t):
+    return 6000.0 if t >= 1760.0 else 1200.0
+
+
+for name, (rows, _c) in (("now", R1), ("without the fix", R1_old)):
+    print(f"    {name}: " + ", ".join(
+        f"{r[0]:.0f}s A {cmd(r)['a']:.0f} B {cmd(r)['b']:.0f}"
+        for r in rows if 1776 <= r[0] <= 1786))
+_r, _ = R1
+check("the case is the rig's: A charges 2000 W with the house at 6000 W the "
+      "cycle before its charge ends, then holds",
+      at(_r, 1778.0, "a") == -2000.0
+      and phase_at(_r, 1778.0) == "charge" and phase_at(_r, 1780.0) == HOLD)
+check("⭐ no battery charges from the grid after the charge ends (house above "
+      "the aim)", charging_hi(_r, 1780, 1798, hi1) == 0)
+check("⭐ B moves at most one slew step a cycle across the charge end, and A "
+      "after its HOLD pin", moves(_r, "b", 1776, 1798) <= SLEW
+      and moves(_r, "a", 1780, 1798) <= SLEW)
+check("...and B gives toward the house at once (+400 W at the end cycle)",
+      at(_r, 1780.0, "b") == SLEW)
+check("non-vacuous: without the fix B jumps 0 -> -1600 W and charges from "
+      "the grid for 4 cycles", moves(R1_old[0], "b", 1776, 1798) >= 1600
+      and charging_hi(R1_old[0], 1780, 1798, hi1) >= 4)
+
+# ============================================================================ #
+print("\n2. the same at a low house")
+R2 = run(2, 1200.0, 1300.0)
+R2_old = without(lambda: run(2, 1200.0, 1300.0))
+print(f"    B's largest move: {moves(R2[0], 'b', 1296, 1600):.0f} W, without "
+      f"the fix {moves(R2_old[0], 'b', 1296, 1600):.0f} W; B's least "
+      f"command after the end "
+      f"{min(cmd(r)['b'] for r in R2[0] if r[0] >= 1300):.0f}"
+      f" W, without {min(cmd(r)['b'] for r in R2_old[0] if r[0] >= 1300):.0f}"
+      " W")
+check("⭐ house 1200 W: B moves at most one slew step a cycle and never "
+      "charges from the grid after A's charge ends",
+      moves(R2[0], "b", 1296, 1600) <= SLEW
+      and min(cmd(r)["b"] for r in R2[0] if r[0] >= 1300) >= 0.0)
+check("non-vacuous: without the fix B jumps to -1600 W (2800 W from the grid)",
+      min(cmd(r)["b"] for r in R2_old[0] if r[0] >= 1300) <= -1600.0)
+
+# ============================================================================ #
+print("\n3. the charge ends another way")
+
+
+def timed_out(end):
+    """A store whose charge passes CHARGE_TIMEOUT_MIN at `end` s."""
+    s = s79.store_charging(2)
+    s.data[unit_key(A, "phase_started")] = (
+        rig.BASE + timedelta(seconds=end - CHARGE_TIMEOUT_MIN * 60 - 1)
+    ).isoformat()
+    return s
+
+
+def b_done():
+    """A charging; B finished its monthly cycle, so it starts none here."""
+    s = s79.store_charging(2)
+    s.data[unit_key("inverter-b", "last_completed")] = rig.BASE.isoformat()
+    return s
+
+
+def switch_off(at):
+    """Battery A switched Off at `at` s (OPT-11 modes on)."""
+    def before(c, t):
+        if t >= at:
+            c.store.data[unit_key(A, packmode.USE_KEY)] = False
+    return before
+
+
+ENDS = {
+    "timeout": lambda: run(2, 6000.0, 9999.0, step=1760.0, dur=2080.0,
+                           store=timed_out(1780.0)),
+    "Off": lambda: run(2, 6000.0, 9999.0, step=1760.0, dur=2080.0,
+                       store=b_done(), before=switch_off(1780.0),
+                       extra={"battery_modes_enabled": True}),
+}
+for k, fn in ENDS.items():
+    rows, c = fn()
+    old = without(fn)[0]
+    print(f"    {k}: A {[cmd(r)['a'] for r in rows if 1776 <= r[0] <= 1784]}"
+          f" B {[cmd(r)['b'] for r in rows if 1776 <= r[0] <= 1784]}; "
+          f"without the fix B "
+          f"{[cmd(r)['b'] for r in old if 1776 <= r[0] <= 1784]}")
+    check(f"{k}: the charge ends at 1780 s",
+          phase_at(rows, 1778.0) == "charge"
+          and phase_at(rows, 1780.0) == IDLE)
+    check(f"⭐ {k}: no battery charges from the grid after it, and B moves at "
+          f"most one slew step a cycle",
+          charging_hi(rows, 1780, 1798, hi1) == 0
+          and moves(rows, "b", 1776, 1798) <= SLEW)
+    check(f"non-vacuous ({k}): without the fix B jumps to -1600 W, from "
+          f"the grid", at(old, 1780.0, "b") == -1600.0
+          and charging_hi(old, 1780, 1798, hi1) >= 1)
+
+# ============================================================================ #
+print("\n4. the charge ends while the guard has suspended it")
+# House 6000 W from 1500 s: the guard suspends the shaping, the loop has both
+# batteries give. A is full at 1530 s; its HOLD pin takes it to 0 W.
+R4 = run(2, 6000.0, 1530.0, step=1500.0)
+R4_old = without(lambda: run(2, 6000.0, 1530.0, step=1500.0))
+_a4 = at(R4[0], 1528.0, "a")
+print(f"    A gave {_a4:.0f} W before its HOLD; B's largest move "
+      f"{moves(R4[0], 'b', 1526, 1798):.0f} W, without the fix "
+      f"{moves(R4_old[0], 'b', 1526, 1798):.0f} W")
+check("A gives under the loop before its charge ends",
+      present(_a4) and _a4 > SLEW and R4[1].maints[0].phase == HOLD)
+check("⭐ B takes A's share at one slew step a cycle (A's own give stops "
+      "with its HOLD pin, as before)", moves(R4[0], "b", 1526, 1798) <= SLEW
+      and charging_hi(R4[0], 1530, 1798, lambda t: 6000.0) == 0)
+check("non-vacuous: without the fix B jumps more than a slew step",
+      moves(R4_old[0], "b", 1526, 1798) > SLEW)
+
+# ============================================================================ #
+print("\n5. N=1 is untouched; a missed poll; the rule at its boundaries")
+
+
+def digest(rows):
+    return hashlib.sha256(repr(rows).encode()).hexdigest()
+
+
+_n1 = [digest(run(1, h, e, step=s)[0]) for h, e, s in
+       ((6000.0, 1780.0, 1760.0), (1200.0, 1300.0, None))]
+_n1_old = [digest(without(lambda: run(1, h, e, step=s))[0]) for h, e, s in
+           ((6000.0, 1780.0, 1760.0), (1200.0, 1300.0, None))]
+check("⭐ N=1, the charge ending at a high and a low house: every row "
+      "byte-identical with and without the reseed", _n1 == _n1_old)
+
+if no_fix is None:
+    print(f"\n{len(fails)} FAILED (no _reseed_charge_end): " + ", ".join(fails))
+    sys.exit(1)
+
+
+def missed_poll(at_t):
+    """Battery B's power reading is missing on the cycle at `at_t` s."""
+    def before(c, t):
+        if not getattr(c.hass, "_s93", False):
+            read = c.hass.number
+
+            async def number(e, invert=False):
+                if e == "sensor.batt_b" and c.hass.t == at_t:
+                    return None
+                return await read(e, invert)
+            c.hass.number, c.hass._s93 = number, True
+    return before
+
+
+R6 = run(2, 6000.0, 1780.0, step=1760.0, before=missed_poll(1780.0))
+R6_old = without(lambda: run(2, 6000.0, 1780.0, step=1760.0,
+                             before=missed_poll(1780.0)))
+print(f"    a missed poll on the end cycle: B "
+      f"{[cmd(r)['b'] for r in R6[0] if 1778 <= r[0] <= 1786]}, without "
+      f"the fix {[cmd(r)['b'] for r in R6_old[0] if 1778 <= r[0] <= 1786]}")
+check("a missed poll on the cycle the charge ends: the loop holds that cycle",
+      phase_at(R6[0], 1780.0) == HOLD
+      and "inputs missing" in str(first(r[6] for r in R6[0] if r[0] == 1780.0)))
+check("⭐ ...and the end still reaches the loop on the next cycle: no grid "
+      "charge (A's charge stands through the held cycle), one slew step a "
+      "cycle", charging_hi(R6[0], 1782, 1798, hi1) == 0
+      and moves(R6[0], "b", 1776, 1798) <= SLEW)
+check("non-vacuous: without the fix B charges from the grid",
+      charging_hi(R6_old[0], 1782, 1798, hi1) >= 1)
+_c = R1[1]
+
+
+def reseeded(target, shares, res, ended=frozenset({A})):
+    """`_reseed_charge_end` on the rig's controller, A's charge ended and
+    its result `res`: (target, i_w)."""
+    _c.target, _c.i_w, _c._charge_ended = target, 123.0, ended
+    for inv in _c.inverters:
+        inv.target = shares[inv.name[-1]]
+    _c._reseed_charge_end([res, _free])
+    return _c.target, _c.i_w
+
+
+_pin = MaintResult(HOLD, True, 0.0, False, [])
+_free = MaintResult(IDLE, False, None, False, [])
+check("a HOLD pin: the start is B's command plus the pin, integrator with it",
+      reseeded(-2000.0, {"a": -2000.0, "b": 0.0}, _pin) == (0.0, 0.0)
+      and reseeded(2770.0, {"a": 1385.0, "b": 1385.0}, _pin)
+      == (1385.0, 1385.0) and _c._charge_ended == frozenset())
+check("a free battery: its charge leaves the start, its discharge stays",
+      reseeded(-2000.0, {"a": -2000.0, "b": 0.0}, _free) == (0.0, 0.0)
+      and reseeded(2770.0, {"a": 1385.0, "b": 1385.0}, _free)
+      == (2770.0, 2770.0))
+check("no charge ended: nothing moves (the integrator either)",
+      reseeded(-2000.0, {"a": -2000.0, "b": 0.0}, _pin, frozenset())
+      == (-2000.0, 123.0))
+_n1c = run(1, 6000.0, 1780.0, step=1760.0)[1]
+_n1c.target, _n1c.i_w = -2000.0, 5.0
+_n1c.inverters[0].target, _n1c._charge_ended = -2000.0, frozenset({A})
+_n1c._reseed_charge_end([_pin])
+check("N=1: nothing moves", (_n1c.target, _n1c.i_w) == (-2000.0, 5.0))
+_seen = []
+
+
+def _spy(self, maint_results):
+    _seen.append(sorted(self._charge_ended))
+    return no_fix(self, maint_results)
+
+
+Controller._reseed_charge_end = _spy
+try:
+    run(2, 6000.0, 1780.0, step=1760.0, dur=2400.0,
+        extra={"maintenance_hold_min": 5})
+finally:
+    Controller._reseed_charge_end = no_fix
+check("it fires once, at the charge end: not at the HOLD end or on any "
+      "other cycle", [x for x in _seen if x] == [[A]]
+      and len(_seen) > 300)
+
+# ============================================================================ #
+print("\n6. A still in CHARGE but no longer placed: stale link, refusal")
+
+
+def link_drop(at):
+    """A's writes rejected from `at` s, so it is dropped as a stale link."""
+    def before(c, t):
+        a = c.inverters[0]
+        if t >= at and a.rejected_since is None:
+            a.rejected_since = rig.BASE + timedelta(seconds=at)
+    return before
+
+
+def refuses(at):
+    """A stops charging from `at` s (reads 0 W); SAFETY-73 drops it."""
+    def before(c, t):
+        if not getattr(c.hass, "_s93r", False):
+            real = c.hass.power
+
+            def power(u):
+                p = real(u)
+                return 0.0 if u == "a" and c.hass.t >= at and p < 0 else p
+            c.hass.power, c.hass._s93r = power, True
+    return before
+
+
+for k, hook in (("stale link", link_drop), ("refusal", refuses)):
+    def fn(hook=hook):
+        return run(2, 1200.0, 9999.0, dur=1500.0, before=hook(1300.0))
+    rows, _c6 = fn()
+    old = without(fn)[0]
+    drop = first((r[0] for r in rows if r[0] >= 1300
+                  and "dropped inverter-a" in str(r[7])), "A dropped")
+    print(f"    {k}: A dropped at {drop:.0f} s; B "
+          f"{[cmd(r)['b'] for r in rows if drop - 2 <= r[0] <= drop + 4]}, "
+          f"without the fix {[cmd(r)['b'] for r in old if drop - 2 <= r[0] <= drop + 4]}")
+    check(f"⭐ {k}: A is dropped while still in CHARGE, and B never charges "
+          f"from the grid and moves at most one slew step a cycle",
+          phase_at(rows, drop) == "charge"
+          and min(cmd(r)["b"] for r in rows if r[0] >= 1300) >= 0.0
+          and moves(rows, "b", 1296, 1500) <= SLEW)
+    check(f"non-vacuous ({k}): without the fix B jumps to -1600 W",
+          min(cmd(r)["b"] for r in old if r[0] >= 1300) <= -1600.0)
+
+_seen.clear()
+Controller._reseed_charge_end = _spy
+try:
+    run(2, 6000.0, 1530.0, step=1500.0)
+finally:
+    Controller._reseed_charge_end = no_fix
+check("the guard suspending the charge is no end: it fires once, at the HOLD",
+      [x for x in _seen if x] == [[A]])
+
+print()
+if fails:
+    print(f"{len(fails)} FAILED: " + ", ".join(fails))
+    sys.exit(1)
+print("all SAFETY-93 checks passed")