MAINT-15 round 2: in-repo differential check, full-hold with-reading check

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

goodwe_controller/DECISIONS/MAINT-14-no-soc-in-maintenance.md

@@ -0,0 +1,68 @@
+# MAINT-14 - no SoC reading in a maintenance cycle
+
+**Status: for review.** Tests: `test_maint14.py`. No option added; the one new
+number is a constant in `app/maintenance.py`.
+
+**The fault.** `Maintenance.tick()` returned early when the SoC was `None`,
+before any timeout check. (1) A charge on a pack whose SoC reading was lost
+never ended: no timeout, no back-off, no notice, and the 7-day backstop of
+MAINT-13 sat behind the same return. (2) A drain kept commanding `discharge_w`
+with an unknown level, which breaks the TEL-08 rule that an unknown level never
+allows discharge.
+
+## Rules
+
+1. **Charge, no SoC.** The charge timeout (`CHARGE_TIMEOUT_MIN`, gate-open
+   minutes) and the 7-day wall backstop still run. When one fires: if the last
+   SoC seen was >= target - 3 the charge counts as done (the MAINT-13 "close
+   enough to full" rule, HOLD follows); otherwise it fails: MAINT-13 back-off
+   and its one notification. If no SoC was ever seen, it fails.
+2. **Drain, no SoC.** DRAIN holds at 0 W and owns the setpoint, exactly as HOLD
+   does. The 12 h drain timeout still runs and fails as before.
+3. **SoC comes back.** Nothing is recorded; the drain continues at the drain
+   rate, the charge carries on, and the stall wait of MAINT-13 restarts from
+   what it was: it paused while the reading was missing, so a gap never counts
+   as "no more charge".
+4. **HOLD** is unchanged (0 W).
+5. **Log.** One line when a drain starts holding for this reason ("the battery
+   level is unknown, so the drain holds at 0 W until it is back"), and one line
+   when the level has been missing for `NOSOC_NOTE_MIN` = 15 min. Both are said
+   once per episode; the old line that repeated every tick is gone.
+
+## Why
+
+- **"Last seen", not "best seen".** A pack that was at 99 and then dropped to 50
+  before the reading went is not close to full. Only the last reading counts.
+  It is saved with the charge record (`charge_track.last`), and saved at once
+  when the reading first goes missing, so a restart mid-gap still knows it
+  (within the usual 5 min save interval otherwise).
+- **Failure when nothing was seen.** With no level we cannot call a pack full.
+  The failure costs a 7-day wait and one notice; calling it done costs a
+  missed balance and a month-long false "complete". The first is the honest one.
+- **Timeouts count the same minutes as with a reading.** Gate-open charging
+  minutes only; the wall backstop for a gate that never opens.
+- **15 min.** A judgement. A poll glitch of a few minutes should not write a
+  line; a reading lost for a quarter of an hour is worth one. Raise it if the
+  action log shows noise.
+- **0 W, not "stop the phase".** Aborting a drain on a short gap would throw
+  away the part already drained. Holding at 0 W loses nothing and the 12 h
+  timeout is the limit.
+
+## Cost
+
+- A drain that loses its reading stops draining, and the 12 h clock keeps
+  running. A gap of many hours can therefore fail a drain that was working. The
+  back-off then applies (7 days).
+- A charge whose reading is lost early (last SoC below target - 3) fails at the
+  timeout, even if the pack was in fact filling. The same 7-day wait applies.
+
+## What this does not cover
+
+- **HOLD with no SoC**: closed by MAINT-15 (`MAINT-15-hold-ends-without-soc.md`).
+- It does not tell the operator why the reading is missing (sensor, register,
+  link). It does not reuse the stale-input gate; it is the cycle's own rule.
+- The charge itself is still driven by the control loop, which has its own
+  handling of a missing SoC (TEL-08 for the loop is a separate card).
+- A restart loses at most 5 min of accounting, as in MAINT-13. The "unknown for
+  long" line may be said again after a restart: the episode start is in memory.
+- Text added is two log lines, English only, like the other log lines.

goodwe_controller/DECISIONS/MAINT-15-hold-ends-without-soc.md

@@ -0,0 +1,47 @@
+# MAINT-15 - the balancing hold ends on time with no SoC reading
+
+**Status: for review.** Tests: `test_maint15.py`. No option added.
+
+**The fault.** `step()` sent every tick with no SoC to `_no_soc()`, and for
+HOLD that returned 0 W without the time check. A hold that lost its reading
+never ended: `last_completed` was never written and the next due date never
+moved. Harmless for power (0 W), but the cycle was stuck. This was left open
+in MAINT-14.
+
+## Rule
+
+1. **HOLD ends at `hold_min` whether or not there is a reading.** One helper,
+   `Maintenance._hold()`, holds the time check and is used by both paths. When
+   the time is up it writes `last_completed`, goes IDLE and logs the same
+   "cycle COMPLETE" line. Before the time is up it commands 0 W as before.
+2. **Text.** The 15-min "no battery level" line in HOLD now reads "The hold
+   still ends at its normal time." Still once per episode.
+
+## Why
+
+HOLD is reached only after the charge counted as done. Every entry into HOLD
+is in `step()` or `_no_soc()` and follows a charge that reached the target,
+stalled near full (MAINT-13) or timed out near full (MAINT-13/14). Nothing else
+sets the phase to HOLD (`_enter(HOLD` has four call sites, all after a done
+charge). The hold is a wait for the BMS to balance; it never needed the level.
+So a hold that ends with no reading is a completed cycle. The "cycle COMPLETE"
+line is true without a reading: it names minutes of hold, not a level.
+
+## Cost
+
+- If the level was lost during the hold, the cycle is recorded as complete
+  without anyone seeing the final level. The charge had counted as done, so this
+  is the same trust MAINT-14 already gives a charge that ends with no reading.
+
+## What this does not cover
+
+- It does not check that the pack is still full at the end of the hold.
+- It does not say why the reading was lost.
+- With a reading, nothing changes. `test_maint15.py` section 7 compares the new
+  HOLD tick by tick with a frozen copy of the release/1.0 HOLD branch (phase,
+  setpoint, owner, events, `last_completed`) at four `hold_min` values. A wider
+  trace of the whole cycle (800 seeded runs against db9e231) was run outside the
+  repo; its script and log are kept under `scratch/MAINT-15/` of the dev
+  workspace, not in this repository.
+- A state file written by an older version with `phase: hold` loads and ends on
+  time; the file is not changed in format.

goodwe_controller/DECISIONS/SAFETY-107-refusal-under-another-writer.md

@@ -0,0 +1,96 @@
+# SAFETY-107 - a refusal held under another writer is learned once
+
+**Status: for review.** Tests: `test_safety107.py`. One inverter only; nothing
+changes at N>1 (SAFETY-73's unit refusal has no "another writer" end).
+
+**The defect** (MAINT-13 F5, MAINT-14 "seen again"). A full pack under a
+maintenance charge in a cheap window takes no charge. At N=1 the action log then
+showed "BATTERY REFUSES asked -2500 W ..." and "BATTERY REFUSES ENDED: another
+writer" every cycle: 548 + 548 lines in 15 min on the MAINT-13 rig head,
+652 + 651 in 12 min on its base, 33 + 32 in 60 s in MAINT-14's c2b. The 40-line
+log flooded and the hint under the buttons changed every second, so the
+MAINT-13 notice was gone at once.
+
+**Cause** (`main.py`, the SAFETY-52 block before `compute()`). The end tested
+`self.target != self.loop_w`: the wire differs from the loop's ask. A held
+refusal makes the loop ask 0 W, so any claim that binds (the maintenance
+charge's `hi`, the peak guard's `lo`, a trade's SET) differs from it on every
+cycle. The refusal was ended against the same writer it had been learned
+against. `sat_count` is not reset by that end, so `compute()` learned the same
+refusal again in the same cycle. Nothing about the battery or the wire had
+changed; only the log and the hint did.
+
+**Other writers.** The same flap comes from any writer that holds the wire
+while the loop is refused, as long as the evidence lasts: a trade (a SET at
+P_ARBITRAGE) into a full pack, and the peak guard over an empty pack: 335
+refusal lines in 15 min each at a 5 s meter (`test_safety107.py` part 2, run on
+release/1.0).
+A maintenance drain or hold does not flap: its SET resets `sat_count` each cycle.
+
+**The rule** (`_refusal_update()`, `rejudged`). When another writer held the
+wire, the held refusal is judged afresh: `compute()` gets `refused=0` and no
+probe, exactly as before the change. Then:
+
+1. it finds the same refusal against that writer's ask: the same episode
+   continues, with no line and no hint. The probe clock restarts, as it did on
+   the old end and start: a probe is the loop's ask, and it is not on the wire
+   while another writer is;
+2. it does not: the refusal ends, logged "another writer", or "the battery
+   gives power" when the reading shows it (it was "another writer" before);
+3. it finds the other direction: end, then a new start, as before. In
+   practice 2 comes first: SAFETY-94's evidence restarts when the ask changes
+   sign, so a writer that turns round ends the refusal at once and the other
+   way is learned ~12 cycles later (`test_safety107.py` part 3e).
+
+**Why this rule.** The end now means what SAFETY-52 meant by it: the refusal no
+longer describes what is on the wire, judged by the same test that learns it.
+`compute()` sees the same inputs as before and the state after each cycle is
+the same (`batt_refused`, `batt_refused_at`), so every command is unchanged.
+Keeping the held refusal into `compute()` instead (mutation M5) would also
+stop the lines, but it holds the loop at 0 W where the old law let it ask past
+a writer's limit, which changes commands; it also removes SAFETY-52's
+owner-decided "another writer" exit.
+
+**Evidence.**
+- Differential trace (`goodwe/scratch/SAFETY-107/probe/trace.py`, not in this
+  repo; the real Controller in the CI image, release/1.0 bf3ac79 against this
+  tree): 300 seeds, N=1 (153) and
+  N=2 (147), 1/2/5 s meter, full/empty/dead/healthy battery windows, a stepping
+  house, maintenance charge with the cheap window opening and closing, P1 and
+  the peak guard, trade windows, stop/start; 427 680 cycles, 294 seeds with a
+  refusal held. Every command, site target, integrator, `sat_count`, refusal
+  state and reason is identical. Refusal lines: 100 748 before, 2 188 after.
+- The card's case: 1 line in 15 min at N=1 and N=2, 1/2/5 s meter, house
+  importing or exporting (release/1.0: 335-1 753 at N=1; N=2 was already 1).
+
+**Mutations** (`tools/mutation_run.sh` on purged copies; reds in
+test_safety107 / test_refusal / test_refusal_units; trace = seeds of 300 whose
+commands differ from release/1.0):
+
+| | mutation | reds | trace |
+|---|---|---|---|
+| M1 | `now == rejudged` -> `False` (every such cycle ends: the old flap) | 26 / 0 / 0 | 0 |
+| M2 | `rejudged` always 0 (another writer never ends it) | 5 / 1 / 2 | 29 |
+| M3 | the probe clock is not restarted on a continued episode | 1 / 0 / 0 | 2 |
+| M4 | a probe may open while rejudged | 0 / 0 / 0 | 0 |
+| M5 | the held refusal goes into `compute()` while rejudged | 2 / 1 / 2 | 27 |
+| M6 | an end while rejudged always says "another writer" | 1 / 0 / 0 | 0 |
+
+M4 is equivalent: a probe matters only if `compute()` engages in that call,
+which needs `SETTLE_CYCLES` saturated against the writer, and each of those
+cycles restarted the clock. The line stays because it states the rule.
+
+**Cost.** None in power. A takeover by a different writer while the battery
+still refuses is logged as an end and, ~12 cycles later, a new start against
+that writer (3 lines for the episode, `test_safety107.py` part 3a).
+
+**Not covered.**
+- A perfectly flat reading under a writer. The stale-reading gate trips, stops
+  the site and seeds the refusal (SAFETY-52 decision 3). After the stop the
+  writer's ask returns fresh, `compute()` has no saturation history against it,
+  so the refusal ends at the first cycle and the next trip seeds it again: 2
+  lines per gate cycle (29 in 15 min, house exporting), the same as release/1.0.
+  The gate cycle itself (0 W, then the writer's ask) is the cause; stopping it
+  changes commands. None of the rig runs showed it: real readings move.
+- At N=1 a refusal shapes only the loop. A writer above it still asks the
+  battery; at N>1 SAFETY-73's P_SAFETY claim binds every writer.

goodwe_controller/DOCS.md

@@ -605,6 +605,10 @@ still moving towards the request, such as one turning from charging to
 giving (SAFETY-94). "Gives" is counted from what the battery's reading shows
 at rest, which the add-on learns after two quiet minutes and keeps: some
 inverters show about -100 W at rest (SAFETY-97).
+While another task sets the power, such as a maintenance charge, the peak
+guard or a trade, the refusal stays as long as the battery still refuses that
+task's request. It is logged once, and the 200 W test waits until control sets
+the power again (SAFETY-107).
 
 **With two or more inverters** the same rule works per inverter. An inverter
 whose battery gives less than 100 W for 12 cycles, and at least 22 s, while
@@ -618,7 +622,7 @@ If the battery refuses to give power 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 and SAFETY-73. The event log reads
+*For installers:* SAFETY-52, SAFETY-73 and SAFETY-107. The event log reads
 `BATTERY REFUSES …` and `BATTERY REFUSES ENDED …` (with the inverter's name at
 N>1), and the INFO line ends in `BATTERY_REFUSES`.
 `DECISIONS/SAFETY-52-a-battery-that-refuses.md` has the replay figures and

goodwe_controller/app/main.py

@@ -1279,9 +1279,26 @@ class Controller:
         _LOG.info("The battery refusal ends, control asks for power again: %s.",
                   why)
 
-    def _refusal_update(self, decision, asked_w, batt_w, probe) -> None:
-        """Follow `compute()`'s carried refused direction, and say so."""
+    def _refusal_update(self, decision, asked_w, batt_w, probe,
+                        rejudged: int = 0) -> None:
+        """Follow `compute()`'s carried refused direction, and say so.
+
+        `rejudged`: the refusal held while another writer had the wire, so
+        `compute()` judged it afresh against that writer's ask (SAFETY-107).
+        """
         now = decision.refused or 0
+        if rejudged:
+            if now == rejudged:
+                # Still refused against the ask on the wire: the same episode.
+                # The probe clock restarts, as it did when this was an end and
+                # a new start: a probe is the loop's ask, and it is not on the
+                # wire while another writer is.
+                self.batt_refused_at = self._now_mono
+                return
+            gave = (rejudged * (batt_w - self._site_idle_w())
+                    >= _ctl.PROBE_ANSWER_W)
+            self._refusal_end("the battery gives power" if gave
+                              else "another writer")
         if now and not self.batt_refused:
             self._refusal_start(now, f"asked {asked_w:+.0f} W, the battery "
                                      f"gave {batt_w:+.0f} W for "
@@ -4973,11 +4990,15 @@ class Controller:
             if trade_was and not mode_trade:
                 self._reseed_run_edge(claims)
             # SAFETY-52: another writer overrode the loop, so a refusal learned
-            # against the loop's own ask no longer describes the wire. A gap
+            # against the loop's own ask may no longer describe the wire. A gap
             # (`loop_w` None: blind or stopped) keeps it: the stale-reading stop
             # is what seeds it, and dropping it there restarts the ~70 s cycle.
+            # SAFETY-107: it ends only if compute() no longer finds it against
+            # that writer's ask (`_refusal_update()`). A held refusal makes the
+            # loop ask 0 W, so any claim that binds differs from it every cycle.
+            rejudged = 0
             if self.loop_w is not None and self.target != self.loop_w:
-                self._refusal_end("another writer")
+                rejudged = self.batt_refused
             # SAFETY-96: a battery that dropped out of the split leaves the
             # loop's start point (SAFETY-93's own if its charge ended too).
             self._reseed_drop(_live96)
@@ -4986,7 +5007,8 @@ class Controller:
             # SAFETY-95: then a drain that ended, last (it reads the start
             # the others left).
             self._reseed_drain_end(maint_results, _live96)
-            probe = self._refusal_probe()
+            # No probe while it is judged afresh (the loop does not hold the wire).
+            probe = False if rejudged else 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.
@@ -5029,7 +5051,8 @@ class Controller:
                 # limit already closes an empty battery's discharge end.
                 # 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
+                refused=(0 if rejudged else self.batt_refused)
+                if len(self.inverters) == 1
                 else self._site_refused(),
                 probe=probe,
                 # SAFETY-94: at N=1 the engage also needs the reading to have
@@ -5055,7 +5078,8 @@ class Controller:
             self.loop_w = decision.target_w
             self._loop_decision = decision
             if len(self.inverters) == 1:
-                self._refusal_update(decision, loop_prev, site_batt, probe)
+                self._refusal_update(decision, loop_prev, site_batt, probe,
+                                     rejudged)
             # --- 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

goodwe_controller/app/maintenance.py

@@ -84,6 +84,10 @@ FAIL_BACKOFF_DAYS = 7
 CHARGE_TICK_MAX_MIN = 5
 # The charge accounting is written to the state file at most this often.
 CHARGE_SAVE_MIN = 5
+# MAINT-14 (DECISIONS/MAINT-14-no-soc-in-maintenance.md): with no SoC reading
+# the timeouts still run, a drain holds at 0 W, and the log says once, after this
+# many minutes, that the level has been unknown for long.
+NOSOC_NOTE_MIN = 15
 
 # ⚠️ CFG-03. THE ONE IN-CODE MIRROR OF config.yaml FOR EVERY `maintenance_*`
 # OPTION, and the only place in the add-on where one of these numbers is typed.
@@ -367,6 +371,10 @@ class Maintenance:
         # written to the state file every CHARGE_SAVE_MIN minutes.
         self._trk = None
         self._trk_saved = None
+        # MAINT-14: when the SoC reading went missing (None while it is there),
+        # and whether the "unknown for long" line was said in this episode.
+        self._nosoc_since = None
+        self._nosoc_said = False
 
     # -- state ---------------------------------------------------------------
     @property
@@ -374,6 +382,7 @@ class Maintenance:
         return self.store.get("phase", IDLE)
 
     def _enter(self, phase: str, now: datetime, events: list, msg: str) -> None:
+        self._nosoc_since = None
         self._trk = None
         self._trk_saved = None
         if self.store.get("charge_track") is not None:
@@ -473,7 +482,8 @@ class Maintenance:
         self._enter(DRAIN, now, events, "maintenance: forced start -> drain")
         return events
 
-    def _charge_account(self, now: datetime, soc: float, gate_open: bool) -> dict:
+    def _charge_account(self, now: datetime, soc: float | None,
+                        gate_open: bool) -> dict:
         """MAINT-13: what this charge has used so far, as minutes of ALLOWED time.
 
         `open_min` counts only ticks with the price gate open; `flat_min` counts
@@ -481,6 +491,9 @@ class Maintenance:
         both. One tick adds at most CHARGE_TICK_MAX_MIN, so a gap (add-on down)
         counts for nothing. A state file from before MAINT-13, or from another
         charge, starts from the wall-clock minutes so far: the old behaviour.
+
+        MAINT-14: `soc` may be None (no reading). Time still counts, the stall
+        wait pauses, and `last` keeps the last SoC seen (saved with the rest).
         """
         since = self.store.get("phase_started")
         trk = self._trk
@@ -490,7 +503,9 @@ class Maintenance:
                 trk = dict(raw)
             else:
                 trk = {"since": since, "open_min": self._elapsed_min(now),
-                       "flat_min": 0.0, "best": float(soc), "at": None}
+                       "flat_min": 0.0,
+                       "best": None if soc is None else float(soc),
+                       "last": None if soc is None else float(soc), "at": None}
             self._trk = trk
         at = trk.get("at")
         try:
@@ -500,10 +515,14 @@ class Maintenance:
         dt = 0.0 if last is None else min(
             CHARGE_TICK_MAX_MIN, max(0.0, (now - last).total_seconds() / 60.0))
         trk["at"] = now.isoformat()
-        if soc > trk["best"]:
+        if soc is None:
+            pass
+        elif trk.get("best") is None or soc > trk["best"]:
             trk["best"], trk["flat_min"] = float(soc), 0.0
         elif gate_open:
             trk["flat_min"] += dt
+        if soc is not None:
+            trk["last"] = float(soc)
         if gate_open:
             trk["open_min"] += dt
         saved = self._trk_saved
@@ -522,6 +541,81 @@ class Maintenance:
         events.append(msg)
         return MaintResult(IDLE, True, 0.0, False, events, failed=kind)
 
+    def _hold(self, now: datetime, events: list) -> MaintResult:
+        """HOLD for one tick, with or without a SoC reading (MAINT-15).
+
+        HOLD needs no level to end: it is reached only after the charge counted
+        as done (MAINT-13/14), so its end is a time check and nothing else.
+        """
+        elapsed = self._elapsed_min(now)
+        if elapsed >= self.cfg.hold_min:
+            self.store.set_time("last_completed", now)
+            self._enter(IDLE, now, events,
+                        f"maintenance: cycle COMPLETE ({elapsed:.0f} min hold) "
+                        "- handing back to the loop")
+            return MaintResult(IDLE, False, None, False, events)
+        # 0 W, not "do nothing": commanding zero is what stops the loop
+        # pulling the pack straight back down the moment it is full, which
+        # is exactly when the BMS needs it parked at the top.
+        return MaintResult(HOLD, True, 0.0, False, events)
+
+    def _no_soc(self, now: datetime, phase: str, gate_open: bool,
+                events: list) -> MaintResult:
+        """MAINT-14: one tick of a pack in a moving phase with no SoC reading.
+
+        HOLD stays at 0 W and ends at its normal time (MAINT-15). DRAIN holds at 0 W, like HOLD, and its 12 h
+        timeout still runs. CHARGE keeps its timeout and wall backstop: out of
+        time is a failure, except when the last SoC seen was within
+        CHARGE_NEAR_FULL_PCT of the target, which counts as done (MAINT-13).
+        Nothing here may start a phase change that needs the level.
+        """
+        if self._nosoc_since is None:
+            self._nosoc_since = now
+            self._nosoc_said = False
+            self._trk_saved = None          # save the last SoC seen, now
+            if phase == DRAIN:
+                msg = ("maintenance: the battery level is unknown, so the "
+                       "drain holds at 0 W until it is back.")
+                _LOG.warning("%s", msg)
+                events.append(msg)
+        if (not self._nosoc_said and (now - self._nosoc_since).total_seconds()
+                >= NOSOC_NOTE_MIN * 60):
+            self._nosoc_said = True
+            tail = ("The hold still ends at its normal time."
+                    if phase == HOLD else
+                    f"The time limit of the {phase} still applies.")
+            msg = (f"maintenance: no battery level for {NOSOC_NOTE_MIN} min. "
+                   f"{tail}")
+            _LOG.warning("%s", msg)
+            events.append(msg)
+        if phase == HOLD:
+            return self._hold(now, events)
+        elapsed = self._elapsed_min(now)
+        if phase == DRAIN:
+            if elapsed > DRAIN_TIMEOUT_MIN:
+                return self._fail(now, "drain",
+                                  f"drain stuck, battery level unknown, after "
+                                  f"{elapsed / 60:.0f} h")
+            return MaintResult(DRAIN, True, 0.0, False, events)
+        trk = self._charge_account(now, None, gate_open)
+        if (trk["open_min"] > CHARGE_TIMEOUT_MIN
+                or elapsed > CHARGE_WALL_TIMEOUT_MIN):
+            last = trk.get("last")
+            if last is not None and last >= self.cfg.soc_target - CHARGE_NEAR_FULL_PCT:
+                self._enter(HOLD, now, events,
+                            f"maintenance: charge time is up and the last level "
+                            f"seen was {last:.0f} %, close enough to full, so the "
+                            f"charge counts as done -> holding "
+                            f"{self.cfg.hold_min} min to balance")
+                return MaintResult(HOLD, True, 0.0, False, events)
+            if elapsed > CHARGE_WALL_TIMEOUT_MIN:
+                lim = f"the {CHARGE_WALL_TIMEOUT_MIN // 1440}-day limit"
+            else:
+                lim = f"{trk['open_min'] / 60:.0f} h of charging time"
+            return self._fail(now, "charge",
+                              f"charge stuck, battery level unknown, after {lim}")
+        return MaintResult(CHARGE, False, None, True, events)
+
     def abort(self, now: datetime, why: str) -> list:
         events = []
         self._enter(IDLE, now, events, f"maintenance: ABORT ({why})")
@@ -609,13 +703,13 @@ class Maintenance:
         # transitions below mean anything, and a stale "100 %" would end the
         # charge phase having charged nothing. Hold the current phase's command
         # and wait; the phase timeout is the backstop.
+        #
+        # MAINT-14: "hold and wait" is not "wait for ever". The timeouts run
+        # here too, and a drain holds at 0 W (an unknown level never allows
+        # discharge, TEL-08). See `_no_soc`.
         if soc is None:
-            _LOG.warning("maintenance: no SoC reading, holding phase %s", phase)
-            if phase == DRAIN:
-                return MaintResult(DRAIN, True, self.cfg.discharge_w, False, events)
-            if phase == HOLD:
-                return MaintResult(HOLD, True, 0.0, False, events)
-            return MaintResult(CHARGE, False, None, True, events)
+            return self._no_soc(now, phase, gate_open, events)
+        self._nosoc_since = None
 
         elapsed = self._elapsed_min(now)
 
@@ -660,16 +754,7 @@ class Maintenance:
             return MaintResult(CHARGE, False, None, True, events)
 
         if phase == HOLD:
-            if elapsed >= self.cfg.hold_min:
-                self.store.set_time("last_completed", now)
-                self._enter(IDLE, now, events,
-                            f"maintenance: cycle COMPLETE ({elapsed:.0f} min hold) "
-                            "- handing back to the loop")
-                return MaintResult(IDLE, False, None, False, events)
-            # 0 W, not "do nothing": commanding zero is what stops the loop
-            # pulling the pack straight back down the moment it is full, which
-            # is exactly when the BMS needs it parked at the top.
-            return MaintResult(HOLD, True, 0.0, False, events)
+            return self._hold(now, events)
 
         _LOG.error("maintenance: unknown phase %r, forcing idle", phase)
         events += self.abort(now, f"unknown phase {phase!r}")

goodwe_controller/test_maint14.py

@@ -0,0 +1,403 @@
+"""MAINT-14: maintenance with no SoC reading.
+`python3 test_maint14.py`
+
+Graded here: charge and drain with SoC None at one and two packs - the charge
+timeout and 7 day backstop still run (failure, or done when the last SoC was
+>= target - 3), the drain holds at 0 W and times out, SoC coming back resumes
+normally with no back-off, the log lines (once per episode), a restart while the
+SoC is missing, and one pack's missing SoC leaving the other pack alone.
+See DECISIONS/MAINT-14-no-soc-in-maintenance.md.
+
+Mutating what this file grades? Read MUTATION.md first (stale bytecode).
+"""
+
+import os
+import sys
+import tempfile
+from datetime import datetime, timedelta, timezone
+
+from app.maintenance import (
+    CHARGE, CHARGE_NEAR_FULL_PCT, CHARGE_TIMEOUT_MIN, CHARGE_WALL_TIMEOUT_MIN,
+    DRAIN, DRAIN_TIMEOUT_MIN, FAIL_BACKOFF_DAYS, HOLD, IDLE, NOSOC_NOTE_MIN,
+    MaintConfig, Maintenance, arbitrate,
+)
+from app.store import DEFAULTS, Store
+
+UTC = timezone.utc
+fails = []
+
+
+def check(name, cond):
+    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
+    if not cond:
+        fails.append(name)
+
+
+CFG = MaintConfig(enabled=True, interval_days=28, start_hour=19,
+                  discharge_w=2000, charge_w=2000, soc_floor=11,
+                  soc_target=100, hold_min=120)
+T0 = datetime(2026, 10, 4, 18, 24, tzinfo=UTC)
+MIN = timedelta(minutes=1)
+
+
+class Mem:
+    def __init__(self, data=None):
+        self.data = dict(DEFAULTS, **(data or {}))
+
+    def set(self, k, v):
+        self.data[k] = v
+
+    def set_time(self, k, when):
+        self.data[k] = when.astimezone(UTC).isoformat() if when else None
+
+    def get_time(self, k):
+        raw = self.data.get(k)
+        return datetime.fromisoformat(raw) if raw else None
+
+    def save(self):
+        pass
+
+
+def in_drain(store=None, unit=None, at=T0):
+    m = Maintenance(CFG, store or Mem(), unit=unit, name=unit or "")
+    m.force_start(at)
+    assert m.phase == DRAIN
+    return m
+
+
+def in_charge(store=None, unit=None, at=T0):
+    m = in_drain(store, unit, at)
+    m.tick(at, 11.0)
+    assert m.phase == CHARGE
+    return m
+
+
+def run(m, start, minutes, soc, gate=True, step=1):
+    """Tick every `step` min; returns (last result, end time, first result that
+    is not the starting phase with its time, all events)."""
+    ph, t, res, out, ev = m.phase, start, None, None, []
+    for i in range(0, minutes, step):
+        t = start + i * MIN
+        s = soc(i) if callable(soc) else soc
+        res = m.tick(t, s, gate_open=gate)
+        ev += res.events
+        if out is None and res.phase != ph:
+            out = (res, t)
+            break
+    return res, start + minutes * MIN, out, ev
+
+
+print("1. charge with no SoC")
+m = in_charge()
+res, _, out, _ = run(m, T0, 600, None)
+check("it keeps charging: the loop owns the setpoint, no start of a new phase",
+      out is None and res.phase == CHARGE and res.charge_only
+      and not res.owns_setpoint)
+
+m = in_charge()
+run(m, T0, 60, 60.0)
+res, _, out, ev = run(m, T0 + 60 * MIN, CHARGE_TIMEOUT_MIN + 10, None)
+check("the charge timeout still runs: it fails once the time is up (last SoC 60)",
+      out and out[0].failed == "charge" and out[0].phase == IDLE)
+check("...with the MAINT-13 back-off and its one plain line",
+      m.store.get_time("last_failed") is not None
+      and sum("did not finish" in e for e in ev) == 1
+      and f"{FAIL_BACKOFF_DAYS} days" in ev[-1])
+check("...it took 36 h of allowed time from the charge start, not more",
+      out and out[1] <= T0 + timedelta(minutes=CHARGE_TIMEOUT_MIN + 10))
+check("...and the pack does not want a start in the back-off",
+      not m.wants_start((T0 + timedelta(days=3)).replace(hour=19)))
+
+m = in_charge()
+res, _, out, _ = run(m, T0, CHARGE_TIMEOUT_MIN + 10, None)
+check("never saw a SoC: the timeout is a failure, not 'done'",
+      out and out[0].failed == "charge")
+
+for last in (97.0, 99.0):
+    m = in_charge()
+    run(m, T0, 10, last)
+    res, _, out, ev = run(m, T0 + 10 * MIN, CHARGE_TIMEOUT_MIN + 20, None)
+    check(f"last SoC {last:.0f} (>= target - {CHARGE_NEAR_FULL_PCT}): time up counts as done",
+          out and out[0].phase == HOLD and out[0].failed == ""
+          and m.store.get_time("last_failed") is None)
+    check("...one plain line says so",
+          sum("counts as done" in e for e in ev) == 1
+          and f"{last:.0f} %" in " ".join(ev))
+m = in_charge()
+run(m, T0, 10, 96.0)
+res, _, out, _ = run(m, T0 + 10 * MIN, CHARGE_TIMEOUT_MIN + 20, None)
+check("last SoC 96 (below target - 3): time up is a failure",
+      out and out[0].failed == "charge")
+m = in_charge()
+run(m, T0, 10, 99.0)
+run(m, T0 + 10 * MIN, 10, 50.0)
+res, _, out, _ = run(m, T0 + 20 * MIN, CHARGE_TIMEOUT_MIN + 20, None)
+check("'last seen' is the last reading, not the best (99 then 50: failure)",
+      out and out[0].failed == "charge")
+
+print("2. the 7 day backstop, with the gate shut")
+m = in_charge()
+run(m, T0, 10, 70.0)
+res, _, out, _ = run(m, T0 + 10 * MIN, CHARGE_WALL_TIMEOUT_MIN + 30, None,
+                     gate=False, step=5)
+check("gate never opens, last SoC 70: failure at 7 days",
+      out and out[0].failed == "charge"
+      and out[1] >= T0 + timedelta(minutes=CHARGE_WALL_TIMEOUT_MIN)
+      and out[1] <= T0 + timedelta(minutes=CHARGE_WALL_TIMEOUT_MIN + 10))
+m = in_charge()
+run(m, T0, 10, 98.0)
+res, _, out, _ = run(m, T0 + 10 * MIN, CHARGE_WALL_TIMEOUT_MIN + 30, None,
+                     gate=False, step=5)
+check("gate never opens, last SoC 98: done at 7 days",
+      out and out[0].phase == HOLD and out[0].failed == ""
+      and out[1] >= T0 + timedelta(minutes=CHARGE_WALL_TIMEOUT_MIN))
+m = in_charge()
+run(m, T0, 10, 70.0)
+res, _, out, _ = run(m, T0 + 10 * MIN, 3000, None, gate=False, step=5)
+check("gate shut, no SoC: the open-time clock does not run (no early end)",
+      out is None)
+
+print("3. SoC comes back mid-charge")
+m = in_charge()
+run(m, T0, 30, 80.0)
+run(m, T0 + 30 * MIN, 600, None)
+res, _, out, _ = run(m, T0 + 630 * MIN, 5, 80.0)
+check("normal behaviour resumes: still charging, nothing recorded",
+      out is None and res.phase == CHARGE
+      and m.store.get_time("last_failed") is None)
+res, _, out, _ = run(m, T0 + 640 * MIN, 10, 100.0)
+check("...and it completes at the target as usual",
+      out and out[0].phase == HOLD and m.store.get_time("last_failed") is None)
+m = in_charge()
+run(m, T0, 30, 98.0)
+run(m, T0 + 30 * MIN, 600, None)
+res, _, out, _ = run(m, T0 + 630 * MIN, 30, 98.0)
+check("the stall wait paused while the SoC was missing (30 min flat after the "
+      "gap is not 60)", out is None)
+res, _, out, _ = run(m, T0 + 660 * MIN, 40, 98.0)
+check("...and completes once 60 min flat are in", out and out[0].phase == HOLD)
+m = in_charge()
+run(m, T0, 10, 70.0)
+run(m, T0 + 10 * MIN, CHARGE_TIMEOUT_MIN - 100, None)
+res, _, out, _ = run(m, T0 + (CHARGE_TIMEOUT_MIN - 90) * MIN, 5, 70.0)
+check("a gap just before the limit then a reading: no failure, no back-off",
+      out is None and m.store.get_time("last_failed") is None)
+
+print("4. drain with no SoC")
+m = in_drain()
+run(m, T0, 30, 60.0)
+res, _, out, ev = run(m, T0 + 30 * MIN, 60, None)
+check("DRAIN holds at 0 W, owned by maintenance, the same as HOLD",
+      out is None and res.phase == DRAIN and res.owns_setpoint
+      and res.setpoint_w == 0.0 and not res.charge_only)
+hold = in_charge()
+hold.tick(T0 + MIN, 100.0)
+hr = hold.tick(T0 + 2 * MIN, None)
+check("...the same answer as HOLD gives",
+      (hr.owns_setpoint, hr.setpoint_w) == (res.owns_setpoint, res.setpoint_w))
+check("never above 0 W on any missing tick",
+      all(m.tick(T0 + (100 + i) * MIN, None).setpoint_w == 0.0 for i in range(30)))
+check("one plain line that the drain holds, once, not per tick",
+      sum("drain holds at 0 W" in e for e in ev) == 1)
+check("one plain line after the threshold, once",
+      sum("no battery level for" in e for e in ev) == 1
+      and f"{NOSOC_NOTE_MIN} min" in " ".join(ev))
+m = in_drain()
+res, _, out, ev = run(m, T0, NOSOC_NOTE_MIN - 1, None)
+check("under the threshold the 'no level for long' line is not said",
+      not any("no battery level for" in e for e in ev))
+m = in_drain()
+res, _, out, ev = run(m, T0, DRAIN_TIMEOUT_MIN + 10, None)
+check("the 12 h drain timeout still runs: failure with back-off",
+      out and out[0].failed == "drain" and out[0].phase == IDLE
+      and out[1] > T0 + timedelta(minutes=DRAIN_TIMEOUT_MIN)
+      and m.store.get_time("last_failed") is not None)
+m = in_drain()
+run(m, T0, 60, None)
+res, _, out, _ = run(m, T0 + 60 * MIN, 5, 60.0)
+check("SoC back: the drain continues at the drain rate",
+      out is None and res.phase == DRAIN and res.owns_setpoint
+      and res.setpoint_w == CFG.discharge_w)
+res, _, out, _ = run(m, T0 + 70 * MIN, 5, 10.0)
+check("...and ends at the floor as usual",
+      out and out[0].phase == CHARGE and m.store.get_time("last_failed") is None)
+m = in_drain()
+run(m, T0, 60, None)
+run(m, T0 + 60 * MIN, 5, 60.0)
+res, _, out, ev = run(m, T0 + 70 * MIN, 5, None)
+check("a second episode says its lines again",
+      any("drain holds at 0 W" in e for e in ev))
+
+print("5. the log is once per episode")
+m = in_charge()
+res, _, out, ev = run(m, T0, 300, None)
+check("a 5 h charge episode: exactly one 'no level for long' line, none per tick",
+      len(ev) == 1 and "no battery level for" in ev[0])
+run(m, T0 + 300 * MIN, 3, 70.0)
+res, _, out, ev = run(m, T0 + 303 * MIN, 60, None)
+check("...and one more for a new episode", len(ev) == 1)
+m = in_charge()
+res, _, out, ev = run(m, T0, 5, None)
+check("a 5 min gap says nothing", ev == [])
+
+print("5b. the words per phase")
+m = in_charge()
+m.tick(T0 + MIN, 100.0)
+_, _, _, ev = run(m, T0 + 2 * MIN, 30, None)
+note = [e for e in ev if "no battery level for" in e]
+check("HOLD: the line exists and says the hold ends at its normal time (MAINT-15)",
+      len(note) == 1 and "normal time" in note[0] and "waits" not in note[0])
+m = in_charge()
+_, _, _, ev = run(m, T0, 30, None)
+check("charge keeps its wording", any("time limit of the charge" in e for e in ev))
+m = in_charge()
+run(m, T0, 10, 70.0)
+_, _, out, ev = run(m, T0 + 10 * MIN, CHARGE_WALL_TIMEOUT_MIN + 30, None,
+                    gate=False, step=5)
+fail = [e for e in ev if "ABORT" in e]
+check("7-day backstop: the reason names the 7-day limit, not '0 h'",
+      len(fail) == 1 and "7-day limit" in fail[0] and "0 h" not in fail[0])
+m = in_charge()
+_, _, out, ev = run(m, T0, CHARGE_TIMEOUT_MIN + 10, None)
+fail = [e for e in ev if "ABORT" in e]
+check("charge-time limit: the reason gives the hours of charging time",
+      len(fail) == 1 and "36 h of charging time" in fail[0])
+m = in_charge()
+run(m, T0, 10, 98.0)
+_, _, out, _ = run(m, T0 + 10 * MIN, CHARGE_TIMEOUT_MIN + 20, None)
+_, _, _, ev = run(m, out[1] + MIN, 30, None)
+check("charge done with no SoC, then HOLD with no SoC: a new episode, so the "
+      "HOLD line is said", out[0].phase == HOLD
+      and sum("normal time" in e for e in ev) == 1)
+
+print("6. restart while the SoC is missing")
+with tempfile.TemporaryDirectory() as d:
+    path = os.path.join(d, "state.json")
+    m = Maintenance(CFG, Store(path))
+    m.force_start(T0)
+    m.tick(T0, 11.0)
+    run(m, T0, 97, lambda i: 90.0 if i < 96 else 98.0)   # 98 seen 1 min ago
+    run(m, T0 + 97 * MIN, 3, None)          # SoC lost; state saved at once
+    t = T0 + 100 * MIN                         # restart 3 min after the loss, before the next 5 min save
+    m2 = Maintenance(CFG, Store(path))
+    r = m2.tick(t, None)
+    check("a fresh process reads charge, and still charges", m2.phase == CHARGE
+          and r.phase == CHARGE and r.charge_only)
+    check("...with the open minutes and the last SoC seen from the file",
+          m2._trk is not None and m2._trk["last"] == 98.0
+          and 92 <= m2._trk["open_min"] <= 105)
+    res, _, out, _ = run(m2, t + MIN, CHARGE_TIMEOUT_MIN, None, step=5)
+    check("...so the timeout counts from the first charge minute and, with the "
+          "last SoC 98, completes (no failure)",
+          out and out[0].phase == HOLD and out[0].failed == ""
+          and out[1] < T0 + timedelta(minutes=CHARGE_TIMEOUT_MIN + 60))
+    # drain: restart keeps it at 0 W and keeps the clock
+    path2 = os.path.join(d, "s2.json")
+    md = Maintenance(CFG, Store(path2))
+    md.force_start(T0)
+    md.tick(T0 + 100 * MIN, None)
+    md2 = Maintenance(CFG, Store(path2))
+    r = md2.tick(T0 + 101 * MIN, None)
+    check("a drain restarted with no SoC holds at 0 W", md2.phase == DRAIN
+          and r.setpoint_w == 0.0 and r.owns_setpoint)
+    res, _, out, _ = run(md2, T0 + 102 * MIN, DRAIN_TIMEOUT_MIN, None, step=5)
+    check("...and the 12 h clock is from the drain start, not the restart",
+          out and out[0].failed == "drain"
+          and out[1] <= T0 + timedelta(minutes=DRAIN_TIMEOUT_MIN + 10))
+    # a charge restarted with no record of any SoC (old state file)
+    old = {"phase": "charge", "phase_started": (T0 - timedelta(hours=35)).isoformat(),
+           "last_completed": None, "last_start_attempt": None, "auto": True}
+    import json
+    path3 = os.path.join(d, "s3.json")
+    json.dump(old, open(path3, "w"))
+    mo = Maintenance(CFG, Store(path3))
+    res, _, out, _ = run(mo, T0, 120, None)
+    check("an old state file with no SoC seen: the charge fails at 36 h",
+          out and out[0].failed == "charge")
+    # a MAINT-13 state file: charge_track with a best but no `last`
+    since = (T0 - timedelta(hours=35)).isoformat()
+    m13 = dict(old, phase_started=since, charge_track={
+        "since": since, "open_min": 2100.0, "flat_min": 0.0, "best": 99.0,
+        "at": (T0 - MIN).isoformat()})
+    path4 = os.path.join(d, "s4.json")
+    json.dump(m13, open(path4, "w"))
+    mo = Maintenance(CFG, Store(path4))
+    res, _, out, _ = run(mo, T0, 120, None)
+    check("a MAINT-13 state file (no last SoC): no crash, the charge fails at "
+          "36 h (best is not last)", out and out[0].failed == "charge")
+
+print("7. two packs")
+a, b = (Maintenance(CFG, Mem(), unit="inverter-a", name="inverter-a"),
+        Maintenance(CFG, Mem(), unit="inverter-b", name="inverter-b"))
+a.force_start(T0)
+# the twin site: the same two packs, A healthy (SoC 60, still above the floor)
+a_ok, b_alone = (Maintenance(CFG, Mem(), unit="inverter-a", name="inverter-a"),
+                 Maintenance(CFG, Mem(), unit="inverter-b", name="inverter-b"))
+a_ok.force_start(T0)
+same = True
+for i in range(0, 120, 10):
+    t = T0 + i * MIN
+    plan = arbitrate([a, b], t)
+    plan0 = arbitrate([a_ok, b_alone], t)
+    same = same and plan[1] == plan0[1]
+    ra = a.tick(t, None, blocked_by=plan[0])
+    rb = b.tick(t, 55.0, blocked_by=plan[1])
+    a_ok.tick(t, 60.0, blocked_by=plan0[0])
+    rb0 = b_alone.tick(t, 55.0, blocked_by=plan0[1])
+    same = same and (rb.phase, rb.setpoint_w, rb.deferred) == \
+        (rb0.phase, rb0.setpoint_w, rb0.deferred)
+check("A drains with no SoC at 0 W while B (SoC 55, idle) is as it is when A "
+      "is healthy", ra.setpoint_w == 0.0 and ra.owns_setpoint and rb.phase == IDLE
+      and rb.setpoint_w is None and not rb.owns_setpoint)
+check("...B's answers are unchanged tick by tick", same)
+res = None
+for i in range(0, DRAIN_TIMEOUT_MIN + 30, 10):
+    t = T0 + 120 * MIN + i * MIN
+    plan = arbitrate([a, b], t)
+    res = a.tick(t, None, blocked_by=plan[0])
+    b.tick(t, 55.0, blocked_by=plan[1])
+    if res.failed:
+        break
+check("A's drain times out and A backs off (A's keys only)",
+      res.failed == "drain" and a.store.get_time("last_failed") is not None
+      and b.store.get_time("last_failed") is None)
+slot = (T0 + timedelta(days=1)).replace(hour=19, minute=5)
+check("...then B, with its SoC, takes the slot as usual",
+      arbitrate([a, b], slot) == ["", ""] and b.wants_start(slot)
+      and not a.wants_start(slot))
+b.force_start(slot)
+rb = b.tick(slot + 5 * MIN, 50.0)
+check("...and B's drain runs at the drain rate",
+      rb.phase == DRAIN and rb.setpoint_w == CFG.discharge_w)
+# the other way: B charges with no SoC, A has a SoC
+a = Maintenance(CFG, Mem(), unit="inverter-a", name="inverter-a")
+b = Maintenance(CFG, Mem(), unit="inverter-b", name="inverter-b")
+b.force_start(T0)
+b.tick(T0, 11.0)
+run(b, T0, 5, 97.0)
+a_alone = Maintenance(CFG, Mem(), unit="inverter-a", name="inverter-a")
+same, t = True, T0 + 10 * MIN
+while t < T0 + timedelta(minutes=CHARGE_TIMEOUT_MIN + 30):
+    plan = arbitrate([a, b], t)
+    rb = b.tick(t, None, blocked_by=plan[1])
+    ra = a.tick(t, 60.0, blocked_by=plan[0])
+    ra0 = a_alone.tick(t, 60.0, blocked_by=plan[0])
+    same = same and (ra.phase, ra.setpoint_w, ra.deferred) == \
+        (ra0.phase, ra0.setpoint_w, ra0.deferred)
+    if rb.phase != CHARGE:
+        break
+    t += 5 * MIN
+check("B charges with no SoC (last 97): A (SoC 60) is unchanged tick by tick",
+      same)
+check("...B ends as done (HOLD), A untouched, no failure on either",
+      rb.phase == HOLD and rb.failed == ""
+      and b.store.get_time("last_failed") is None
+      and a.store.get_time("last_failed") is None and a.phase == IDLE)
+
+print()
+if fails:
+    print(f"{len(fails)} FAILED")
+    for f in fails:
+        print("  -", f)
+    sys.exit(1)
+print("all passed")

goodwe_controller/test_maint15.py

@@ -0,0 +1,299 @@
+"""MAINT-15: the HOLD ends on time with no SoC reading.
+`python3 test_maint15.py`
+
+Graded here: a HOLD that loses its SoC (at the start, mid-hold, just before
+hold_min) ends on time, writes last_completed and says the COMPLETE line; SoC
+back mid-hold changes nothing; HOLD stays 0 W; one pack's loss leaves the other
+pack alone; a restart in HOLD with no SoC and an old state file.
+See DECISIONS/MAINT-15-hold-ends-without-soc.md.
+
+Mutating what this file grades? Read MUTATION.md first (stale bytecode).
+"""
+
+import os
+import sys
+import tempfile
+from datetime import datetime, timedelta, timezone
+
+from app.maintenance import (
+    CHARGE, DRAIN, HOLD, IDLE, NOSOC_NOTE_MIN, MaintConfig, Maintenance,
+)
+from app.store import DEFAULTS, Store
+
+UTC = timezone.utc
+fails = []
+
+
+def check(name, cond):
+    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
+    if not cond:
+        fails.append(name)
+
+
+CFG = MaintConfig(enabled=True, interval_days=28, start_hour=19,
+                  discharge_w=2000, charge_w=2000, soc_floor=11,
+                  soc_target=100, hold_min=120)
+T0 = datetime(2026, 10, 4, 18, 24, tzinfo=UTC)
+MIN = timedelta(minutes=1)
+
+
+class Mem:
+    def __init__(self, data=None):
+        self.data = dict(DEFAULTS, **(data or {}))
+
+    def set(self, k, v):
+        self.data[k] = v
+
+    def set_time(self, k, when):
+        self.data[k] = when.astimezone(UTC).isoformat() if when else None
+
+    def get_time(self, k):
+        raw = self.data.get(k)
+        return datetime.fromisoformat(raw) if raw else None
+
+    def save(self):
+        pass
+
+
+def in_drain(store=None, unit=None, at=T0):
+    m = Maintenance(CFG, store or Mem(), unit=unit, name=unit or "")
+    m.force_start(at)
+    assert m.phase == DRAIN
+    return m
+
+
+def in_charge(store=None, unit=None, at=T0):
+    m = in_drain(store, unit, at)
+    m.tick(at, 11.0)
+    assert m.phase == CHARGE
+    return m
+
+
+def run(m, start, minutes, soc, gate=True, step=1):
+    """Tick every `step` min; returns (last result, end time, first result that
+    is not the starting phase with its time, all events)."""
+    ph, t, res, out, ev = m.phase, start, None, None, []
+    for i in range(0, minutes, step):
+        t = start + i * MIN
+        s = soc(i) if callable(soc) else soc
+        res = m.tick(t, s, gate_open=gate)
+        ev += res.events
+        if out is None and res.phase != ph:
+            out = (res, t)
+            break
+    return res, start + minutes * MIN, out, ev
+
+
+
+HOLD_MIN = CFG.hold_min
+
+
+def in_hold(store=None, unit=None, at=T0):
+    """A pack in HOLD that started at `at` + 1 min (charge counted as done)."""
+    m = in_charge(store, unit, at)
+    m.tick(at + MIN, 100.0)
+    assert m.phase == HOLD
+    return m
+
+
+H0 = T0 + MIN            # the hold's start
+print("1. SoC lost in HOLD: the hold ends on time")
+for label, lost_at in (("at the start", 0), ("mid-hold", 60),
+                       ("just before hold_min", HOLD_MIN - 1)):
+    m = in_hold()
+    run(m, H0, lost_at, 100.0) if lost_at else None
+    res, _, out, ev = run(m, H0 + lost_at * MIN, HOLD_MIN + 5 - lost_at, None)
+    check(f"lost {label}: ends at hold_min, to IDLE, not failed",
+          out is not None and out[0].phase == IDLE and out[0].failed == ""
+          and out[1] == H0 + HOLD_MIN * MIN)
+    check("...last_completed is written at that time",
+          m.store.get_time("last_completed") == H0 + HOLD_MIN * MIN)
+    check("...the same COMPLETE line, once, and no last_failed",
+          sum("cycle COMPLETE (120 min hold)" in e for e in ev) == 1
+          and m.store.get_time("last_failed") is None)
+    check("...the pack is idle and hands the setpoint back",
+          m.phase == IDLE and not out[0].owns_setpoint)
+
+m = in_hold()
+res, _, out, _ = run(m, H0, HOLD_MIN - 1, None)
+check("before hold_min: still HOLD at 0 W, owned, nothing written",
+      out is None and res.phase == HOLD and res.owns_setpoint
+      and res.setpoint_w == 0.0 and m.store.get_time("last_completed") is None)
+check("...the pack is not due again (next due date moved after the end)",
+      (run(m, H0 + (HOLD_MIN - 1) * MIN, 3, None), m.due(H0 + 130 * MIN))[1] is False)
+
+print("2. the text")
+m = in_hold()
+_, _, out, ev = run(m, H0, HOLD_MIN + 5, None)
+note = [e for e in ev if "no battery level for" in e]
+check("one 15 min line, saying the hold ends at its normal time",
+      len(note) == 1 and "normal time" in note[0] and "waits" not in note[0]
+      and f"{NOSOC_NOTE_MIN} min" in note[0])
+check("...and the other events are the one COMPLETE line only", len(ev) == 2)
+m = in_hold()
+_, _, _, ev = run(m, H0, NOSOC_NOTE_MIN - 2, None)
+check("under the threshold the line is not said", ev == [])
+
+print("3. SoC comes back mid-hold")
+ends = {}
+for label, soc in (("no SoC", lambda i: None if i < 40 else 100.0),
+                   ("SoC the whole time", 100.0),
+                   ("a low SoC after the gap", lambda i: None if i < 40 else 55.0)):
+    m = in_hold()
+    res, _, out, ev = run(m, H0, HOLD_MIN + 5, soc)
+    ends[label] = (out[1], m.store.get_time("last_completed")) if out else None
+    check(f"{label}: ends at hold_min", out is not None
+          and out[1] == H0 + HOLD_MIN * MIN)
+check("...the end time is the same with or without the gap",
+      len(set(ends.values())) == 1)
+m = in_hold()
+run(m, H0, 40, None)
+res, _, out, _ = run(m, H0 + 40 * MIN, 30, 100.0)
+check("back mid-hold: still HOLD at 0 W until the time is up",
+      out is None and res.phase == HOLD and res.setpoint_w == 0.0)
+
+print("4. restart in HOLD with no SoC")
+with tempfile.TemporaryDirectory() as d:
+    path = os.path.join(d, "state.json")
+    m = Maintenance(CFG, Store(path))
+    m.force_start(T0)
+    m.tick(T0, 11.0)
+    m.tick(H0, 100.0)
+    run(m, H0, 50, None)
+    m2 = Maintenance(CFG, Store(path))
+    r = m2.tick(H0 + 51 * MIN, None)
+    check("a fresh process reads HOLD and holds 0 W", m2.phase == HOLD
+          and r.phase == HOLD and r.owns_setpoint and r.setpoint_w == 0.0)
+    res, _, out, _ = run(m2, H0 + 52 * MIN, HOLD_MIN, None)
+    check("...and the hold ends at hold_min from the hold start, not the restart",
+          out and out[0].phase == IDLE and out[1] == H0 + HOLD_MIN * MIN
+          and m2.store.get_time("last_completed") == H0 + HOLD_MIN * MIN)
+    import json
+    old = {"phase": "hold", "phase_started": H0.isoformat(),
+           "last_completed": None, "last_start_attempt": None, "auto": True}
+    p2 = os.path.join(d, "old.json")
+    json.dump(old, open(p2, "w"))
+    mo = Maintenance(CFG, Store(p2))
+    r = mo.tick(H0 + 10 * MIN, None)
+    check("an old state file in HOLD loads and holds 0 W", mo.phase == HOLD
+          and r.setpoint_w == 0.0)
+    res, _, out, _ = run(mo, H0 + 11 * MIN, HOLD_MIN, None)
+    check("...and ends on time, writing last_completed",
+          out and out[1] == H0 + HOLD_MIN * MIN
+          and mo.store.get_time("last_completed") == H0 + HOLD_MIN * MIN)
+    mr = Maintenance(CFG, Store(p2))
+    check("...the written file reloads as a finished cycle",
+          mr.phase == IDLE and mr.store.get_time("last_completed") is not None)
+
+print("5. two packs")
+a, b = (Maintenance(CFG, Mem(), unit="inverter-a", name="inverter-a"),
+        Maintenance(CFG, Mem(), unit="inverter-b", name="inverter-b"))
+a_ok, b_ok = (Maintenance(CFG, Mem(), unit="inverter-a", name="inverter-a"),
+              Maintenance(CFG, Mem(), unit="inverter-b", name="inverter-b"))
+for x in (a, a_ok):
+    x.force_start(T0)
+    x.tick(T0, 11.0)
+    x.tick(H0, 100.0)
+# B is idle; A's hold loses its SoC (twin site: A keeps it)
+same, a_end = True, None
+for i in range(0, HOLD_MIN + 10):
+    t = H0 + i * MIN
+    ra = a.tick(t, None)
+    rb = b.tick(t, 60.0)
+    ra2 = a_ok.tick(t, 100.0)
+    rb2 = b_ok.tick(t, 60.0)
+    same = same and (rb.phase, rb.owns_setpoint, rb.setpoint_w) == \
+        (rb2.phase, rb2.owns_setpoint, rb2.setpoint_w) and b.phase == b_ok.phase
+    if a_end is None and a.phase == IDLE:
+        a_end = t
+check("A loses SoC in HOLD: A ends on time (as the twin with a reading)",
+      a_end == H0 + HOLD_MIN * MIN and a_ok.phase == IDLE
+      and a.store.get_time("last_completed") == a_ok.store.get_time("last_completed"))
+check("...B is the same as in the twin site where A kept its SoC", same)
+# B in its own HOLD with SoC, A in HOLD without
+a, b = (Maintenance(CFG, Mem(), unit="inverter-a", name="inverter-a"),
+        Maintenance(CFG, Mem(), unit="inverter-b", name="inverter-b"))
+a.force_start(T0); a.tick(T0, 11.0); a.tick(H0, 100.0)
+b_start = H0 + 30 * MIN
+b.force_start(b_start - MIN); b.tick(b_start - MIN, 11.0); b.tick(b_start, 100.0)
+b_end = None
+for i in range(0, HOLD_MIN + 40):
+    t = H0 + i * MIN
+    a.tick(t, None)
+    if t >= b_start:
+        b.tick(t, 100.0)
+        if b_end is None and b.phase == IDLE:
+            b_end = t
+check("B (SoC present) ends at its own hold_min, A's loss changes nothing",
+      b_end == b_start + HOLD_MIN * MIN and a.phase == IDLE
+      and b.store.get_time("last_completed") == b_start + HOLD_MIN * MIN)
+check("...A's end is not delayed by B", a.store.get_time("last_completed")
+      == H0 + HOLD_MIN * MIN)
+
+print("6. other phases unchanged")
+m = in_drain()
+res, _, out, _ = run(m, T0, 60, None)
+check("DRAIN with no SoC still holds 0 W and does not end", out is None
+      and res.phase == DRAIN and res.setpoint_w == 0.0)
+m = in_charge()
+res, _, out, _ = run(m, T0, 600, None)
+check("CHARGE with no SoC still keeps charging", out is None
+      and res.phase == CHARGE and res.charge_only)
+
+
+def old_hold_branch(phase_started, last_completed, now, hold_min):
+    """FROZEN copy of release/1.0 (db9e231) HOLD branch of step(), with a
+    reading: returns (phase, owns_setpoint, setpoint_w, events,
+    last_completed). Kept as a reference for the differential check below."""
+    elapsed = (now - phase_started).total_seconds() / 60.0
+    if elapsed >= hold_min:
+        return (IDLE, False, None,
+                [f"maintenance: cycle COMPLETE ({elapsed:.0f} min hold) "
+                 "- handing back to the loop"], now)
+    return (HOLD, True, 0.0, [], last_completed)
+
+
+print("7. with a reading: the new HOLD against the frozen release/1.0 branch")
+for hm in (0, 5, 60, 120):
+    cfg = MaintConfig(enabled=True, interval_days=28, start_hour=19,
+                      discharge_w=2000, charge_w=2000, soc_floor=11,
+                      soc_target=100, hold_min=hm)
+    m = Maintenance(cfg, Mem())
+    m.force_start(T0)
+    m.tick(T0, 11.0)
+    m.tick(H0, 100.0)
+    started, last, same, end = H0, None, True, None
+    for i in range(0, hm + 15):
+        t = H0 + i * MIN
+        if m.phase != HOLD:
+            break
+        want = old_hold_branch(started, last, t, hm)
+        r = m.tick(t, 100.0 - (i % 3))
+        got = (r.phase, r.owns_setpoint, r.setpoint_w, r.events,
+               m.store.get_time("last_completed"))
+        same = same and got == want
+        last = want[4]
+        if r.phase == IDLE:
+            end = t
+    check(f"hold_min {hm}: every tick identical (phase, owns, W, events, "
+          "last_completed)", same)
+    check("...ends at hold_min, last_completed = that tick, one COMPLETE line",
+          end == H0 + hm * MIN and m.store.get_time("last_completed") == end
+          and m.phase == IDLE)
+
+m = in_hold()
+res, _, out, ev = run(m, H0, HOLD_MIN + 5, 100.0)
+check("HOLD with a reading: 0 W until the end, then IDLE at hold_min",
+      out is not None and out[1] == H0 + HOLD_MIN * MIN and out[0].phase == IDLE)
+check("...last_completed is that tick and the COMPLETE line is said once",
+      m.store.get_time("last_completed") == H0 + HOLD_MIN * MIN
+      and ev == ["maintenance: cycle COMPLETE (120 min hold) "
+                 "- handing back to the loop"])
+
+print()
+if fails:
+    print(f"{len(fails)} FAILED:")
+    for f in fails:
+        print("  -", f)
+    sys.exit(1)
+print("all ok")

goodwe_controller/test_maintenance.py

@@ -122,7 +122,8 @@ m = Maintenance(cfg, FakeStore())
 m.force_start(t0)
 r = m.tick(t0 + timedelta(minutes=5), soc=None)
 check("no SoC: holds the phase rather than guessing", r.phase == DRAIN)
-check("no SoC: still commands the drain rate", r.setpoint_w == 2500)
+check("no SoC: MAINT-14 the drain holds at 0 W, never discharges",
+      r.owns_setpoint and r.setpoint_w == 0.0)
 
 m = Maintenance(cfg, FakeStore())
 m.store.set("phase", "banana")

goodwe_controller/test_safety107.py

@@ -0,0 +1,307 @@
+"""Runnable check for SAFETY-107: a refusal held while another writer has the
+wire is learned once. `python3 test_safety107.py`
+
+A full pack under a maintenance charge in a cheap window took no charge. At
+one inverter SAFETY-52's "another writer" end then fired every cycle, and
+compute() learned the refusal again in the same cycle: "BATTERY REFUSES" and
+"BATTERY REFUSES ENDED: another writer" each second on the rig (MAINT-13 F5).
+Now another writer ends a refusal only if compute(), judging it afresh
+against that writer's ask, no longer finds it. DECISIONS/SAFETY-107.
+
+Drives the REAL Controller (read_inputs + cycle) on opt11_identity's rig.
+1. The card's case: 15 min of a full pack under a maintenance charge, N=1
+   and N=2, at a 1, 2 and 5 s meter: at most 2 refusal lines, held throughout.
+2. Other writers: a trade and the peak guard over a battery that gives
+   nothing. One episode each, no flap.
+3. A refusal still ends: a different writer really takes over, the battery
+   delivers again, a writer asks the other way, stop. The probe clock.
+
+4. Not covered, pinned: a perfectly flat reading under a writer.
+
+Mutations (MUTATION.md, purged copies) are listed in DECISIONS/SAFETY-107.
+"""
+
+import functools
+import logging
+import os
+import sys
+import tempfile
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, HERE)
+sys.path.insert(0, os.path.join(HERE, "tools"))
+logging.disable(logging.WARNING)
+
+import opt11_identity as rig                                   # noqa: E402
+import safety79_identity as s79                                # noqa: E402
+import app.main as app_main                                    # noqa: E402
+from app import control                                        # noqa: E402
+from app.arbiter import Claim, P_ARBITRAGE                     # noqa: E402
+from app.main import Controller                                # noqa: E402
+from named_absence import nth, present                         # noqa: E402
+
+fails = []
+QUARTER = 900.0
+
+
+def check(name, cond):
+    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
+    if not cond:
+        fails.append(name)
+
+
+def run(n=1, *, plant, house, dur=QUARTER, tick=5.0, maint=True,
+        cheap=lambda c, t: True, trade=lambda t: None, p1=False, cap=2500,
+        auto=lambda t: True):
+    """Rows (t, commands, site target, loop reason, refused, refused_at,
+    unit refusals) and every action-log line as (t, line, hint).
+
+    `plant(u, t, cmd)` is battery u's reading. `cheap(c, t)` opens the cheap
+    window. `trade(t)` is a SET claim at P_ARBITRAGE added to the claim list
+    (a writer above the loop), or None."""
+    log, rows, now = [], [], [0.0]
+    real_power, real_tick = rig._Hass.power, rig.TICK
+    real_log, real_resolve, real_c = (Controller.log_event, app_main.resolve,
+                                      rig.Controller)
+
+    def power(self, u):
+        return float(plant(u, self.t, self.cmd[u]))
+
+    def log_event(self, msg, hint=None):
+        log.append((now[0], msg, hint))
+        return real_log(self, msg, hint)
+
+    def resolve(claims, **kw):
+        v = trade(now[0])
+        if v is not None:
+            claims.append(Claim.set("arbitrage", P_ARBITRAGE, v, "trade"))
+        return real_resolve(claims, **kw)
+
+    def before(c, t):
+        now[0] = t
+        if not getattr(c.hass, "_s107", False):
+            read = c.hass.number
+
+            async def number(e, invert=False):
+                if e == "sensor.pnow":
+                    return 0.05 if cheap(c, c.hass.t) else 0.30
+                if e == "sensor.pavg":
+                    return 0.20
+                return await read(e, invert)
+            c.hass.number, c.hass._s107 = number, True
+        if p1:
+            s79._feed(c, t)
+        c.auto = auto(t)
+
+    def on_cycle(c, t):
+        rows.append((t, tuple(c.hass.cmd[u] for u in "ab"[:n]), c.target,
+                     c.loop_reason, c.batt_refused, c.batt_refused_at,
+                     tuple(i.refused for i in c.inverters)))
+
+    extra = {"price_now_entity": "sensor.pnow",
+             "price_avg_entity": "sensor.pavg"}
+    if maint:
+        extra.update({"maintenance_enabled": True,
+                      "maintenance_start_hour": 11,
+                      "maintenance_charge_w": 2000})
+    if p1:
+        extra.update(s79.P1, peak_cap_w=cap)
+    db = os.path.join(tempfile.mkdtemp(prefix="s107-"), "tariff.db")
+    rig._Hass.power, rig.TICK = power, tick
+    Controller.log_event, app_main.resolve = log_event, resolve
+    rig.Controller = functools.partial(real_c, peak_db_path=db)
+    try:
+        _, c = rig.run(n, extra=extra, dur=dur, load=house,
+                       soc=lambda u, t: 50.0, cut=lambda u, t: False,
+                       before=before, on_cycle=on_cycle,
+                       store=s79.store_charging(n) if maint else rig._Store())
+    finally:
+        rig._Hass.power, rig.TICK = real_power, real_tick
+        Controller.log_event, app_main.resolve = real_log, real_resolve
+        rig.Controller = real_c
+    return rows, log, c
+
+
+def full_a(u, t, cmd):
+    """A takes no charge (a full pack), its reading jittering by a few watts
+    as a real one does; B follows."""
+    return -1.0 - int(t) % 3 if u == "a" and cmd < 0 else round(cmd)
+
+
+def flat_a(u, t, cmd):
+    """`full_a` with a perfectly flat reading (the stale-reading gate trips)."""
+    return -1.0 if u == "a" and cmd < 0 else round(cmd)
+
+
+def empty_a(u, t, cmd):
+    """A gives nothing (an empty pack), its reading jittering by a few watts
+    (SAFETY-73's jitter shape, so the stale-reading gate stays out of it);
+    B follows."""
+    return -4.0 - int(t) % 3 if u == "a" and cmd >= -4.0 else round(cmd)
+
+
+def lines(log, word="REFUSES"):
+    return [x for x in log if word in x[1]]
+
+
+def ends(log):
+    return [x for x in log if "REFUSES ENDED" in x[1]]
+
+
+def refused(r):
+    """Row r holds a refusal (site at N=1, any unit at N>1)."""
+    return bool(r[4]) or any(r[6])
+
+
+def held_from(rows):
+    """True when, once learned, the refusal is held on every later row."""
+    first = next((k for k, r in enumerate(rows) if refused(r)), None)
+    return first is not None and all(refused(r) for r in rows[first:])
+
+
+# ==============================================================================
+print("1. the card: a full pack under a maintenance charge, 15 min")
+# ==============================================================================
+for n, tick, house in ((n, k, h) for n in (1, 2) for k in (1.0, 2.0, 5.0)
+                   for h in (800.0, -1500.0)):
+    rows, log, c = run(n, plant=full_a, house=lambda t: house, tick=tick)
+    ref = lines(log)
+    tag = f"N={n}, {tick:.0f} s meter, house {house:+.0f} W"
+    print(f"    {tag}: {len(ref)} refusal line(s): "
+          + " / ".join(x[1][:60] for x in ref))
+    check(f"{tag}: at most 2 refusal lines in 15 min (rig: 548 + 548)",
+          1 <= len(ref) <= 2)
+    check(f"{tag}: no end line", not ends(log))
+    check(f"{tag}: learned once and held to the end", held_from(rows))
+    # N=1: the charge's ceiling stays on the wire. N=2: SAFETY-73 closes A
+    # in the refused direction, as before.
+    check(f"{tag}: still in its charge, the wire as before",
+          c.maints[0].phase == "charge"
+          and all(r[1][0] == (-2000.0 if n == 1 else 0.0)
+                  for r in rows[-10:]))
+    # The hint under the buttons is set at most once for the refusal.
+    check(f"{tag}: the refusal hint is set once",
+          sum(1 for x in ref if x[2] is not None) == 1)
+
+# The probe clock: while the charge holds the wire no probe is tried (the
+# loop's ask is not on the wire), so the action log stays quiet.
+rows, log, _ = run(1, plant=full_a, house=lambda t: 800.0, dur=3600.0)
+check("N=1, an hour: still one line, no probe tried under the charge",
+      len(lines(log)) == 1
+      and not any(r[3] == control.PROBING for r in rows))
+
+# ==============================================================================
+print("\n2. other writers: a trade and the peak guard, no flap")
+# ==============================================================================
+# A trade (a SET above the loop) asks a full pack to charge for 15 min.
+rows, log, _ = run(1, plant=full_a, house=lambda t: 800.0, maint=False,
+                   trade=lambda t: -1500.0)
+print(f"    trade into a full pack: {len(lines(log))} line(s)")
+check("a trade into a full pack: learned once, held, no end",
+      len(lines(log)) == 1 and not ends(log) and held_from(rows))
+# The peak guard asks an empty pack to give while the house is over the limit.
+rows, log, c = run(1, plant=empty_a, house=lambda t: 4500.0, maint=False,
+                   p1=True)
+print(f"    peak guard on an empty pack: {len(lines(log))} line(s), "
+      f"guard floor {c.peak_guard.floor_w if c.peak_guard else None}")
+check("the peak guard on an empty pack: the guard holds the wire",
+      c.peak_guard is not None and c.peak_guard.engaged)
+check("the peak guard on an empty pack: at most 2 lines, held to the end",
+      1 <= len(lines(log)) <= 2 and held_from(rows))
+
+# ==============================================================================
+print("\n3. a refusal still ends")
+# ==============================================================================
+# a. Learned by the loop on an empty pack; at 300 s a trade really takes the
+#    wire and asks it to give 2000 W. The refusal was learned against the
+#    loop's ask: it ends, then is learned against the trade, once.
+rows, log, _ = run(1, plant=empty_a, house=lambda t: 2000.0, maint=False,
+                   trade=lambda t: 2000.0 if 300.0 <= t < 900.0 else None,
+                   dur=1200.0)
+e = ends(log)
+print("    " + " / ".join(f"{x[0]:.0f} s {x[1][:45]}" for x in lines(log)))
+check("a different writer takes over: the refusal ends, 'another writer'",
+      len(e) == 1 and "another writer" in e[0][1] and 300.0 < e[0][0] < 320.0)
+check("...and is learned again against the trade, once, and held",
+      len(lines(log)) == 3 and held_from([r for r in rows if r[0] > 400.0]))
+check("...and the loop keeps it when the trade leaves (no line)",
+      all(x[0] < 900.0 for x in lines(log)) and refused(nth(rows, -1)))
+# The writer asks the other way, which the empty pack takes.
+rows, log, _ = run(1, plant=empty_a, house=lambda t: 2000.0, maint=False,
+                   trade=lambda t: -1500.0 if t >= 300.0 else None)
+e = ends(log)
+check("a writer that asks the other way ends it ('another writer')",
+      len(e) == 1 and "another writer" in e[0][1] and present(nth(rows, -1))
+      and not refused(nth(rows, -1)))
+# b. The full pack starts taking charge again under the same writer.
+rows, log, _ = run(1, plant=lambda u, t, cmd: (
+    round(cmd) if t >= 400.0 else full_a(u, t, cmd)), house=lambda t: 800.0)
+e = ends(log)
+print("    " + " / ".join(f"{x[0]:.0f} s {x[1][:45]}" for x in lines(log)))
+check("the battery takes charge again: the refusal ends once, on its reading",
+      len(e) == 1 and "the battery gives power" in e[0][1]
+      and 400.0 <= e[0][0] < 430.0 and present(nth(rows, -1))
+      and not refused(nth(rows, -1)))
+# c. Stop and start still end it.
+rows, log, _ = run(1, plant=full_a, house=lambda t: 800.0,
+                   auto=lambda t: not 500.0 <= t < 520.0)
+e = ends(log)
+check("stopping control still ends it", any("control stopped" in x[1]
+                                            for x in e))
+# d. The probe clock. Learned under the charge; the cheap window closes when
+#    the refusal is 910 s old, and the loop holds the wire (a surplus, so it
+#    wants to charge). The first probe is a full PROBE_EVERY_S after the
+#    window closed, as before: never at once.
+close = [None]
+
+
+def window(c, t):
+    if close[0] is None and c.batt_refused:
+        close[0] = t + 910.0
+    return close[0] is None or t < close[0]
+
+
+rows, log, _ = run(1, plant=full_a, house=lambda t: -1500.0, cheap=window,
+                   dur=3000.0)
+probes = [r[0] for r in rows if r[3] == control.PROBING]
+print("    " + " / ".join(f"{x[0]:.0f} s {x[1][:45]}" for x in lines(log)))
+print(f"    window closed at {close[0]} s, first probe at "
+      f"{probes[0] if probes else None} s")
+check("the refusal carries over to the loop with no line",
+      len(lines(log)) == 1 and held_from(rows))
+check("the first probe after the writer left comes PROBE_EVERY_S later",
+      close[0] is not None and bool(probes)
+      and probes[0] >= close[0] + control.PROBE_EVERY_S - 10.0)
+# e. A writer turns round on a dead pack with no gap. The evidence restarts
+#    on the new sign (SAFETY-94), so the old refusal ends at once and the
+#    other way is learned once, ~12 cycles later.
+rows, log, _ = run(1, plant=lambda u, t, cmd: (
+    -2.0 - int(t) % 3 if u == "a" else round(cmd)), house=lambda t: 2000.0,
+    maint=False, trade=lambda t: 2000.0 if t < 300.0 else -2000.0)
+e = ends(log)
+print("    " + " / ".join(f"{x[0]:.0f} s {x[1][:45]}" for x in lines(log)))
+check("a writer turns round on a dead pack: one end, then the other way",
+      len(lines(log)) == 3 and len(e) == 1 and 300.0 < e[0][0] < 400.0
+      and present(nth(rows, -1)) and nth(rows, -1)[4] == -1)
+
+# ==============================================================================
+print("\n4. not covered: a perfectly flat reading under a writer")
+# ==============================================================================
+# The stale-reading gate trips on a flat reading while the command moves,
+# stops the site and seeds the refusal (SAFETY-52 decision 3). After the stop
+# the writer's ask comes back fresh, compute() has no saturation history
+# against it yet, so the refusal ends at the first cycle and the next trip
+# seeds it again: one pair per gate cycle, as on release/1.0 (the commands
+# are the same; DECISIONS/SAFETY-107 "Not covered"). Pinned so a change shows.
+rows, log, _ = run(1, plant=flat_a, house=lambda t: -1500.0)
+trips = len(lines(log, "STALE inverter"))
+print(f"    flat reading, house -1500 W: {trips} gate trips, "
+      f"{len(lines(log))} refusal lines in 15 min")
+check("flat reading: no per-cycle flap, at most one pair per gate trip",
+      trips >= 1 and len(lines(log)) <= 2 * trips)
+
+print()
+if fails:
+    print(f"{len(fails)} FAILED: " + ", ".join(fails))
+    sys.exit(1)
+print("all SAFETY-107 checks passed")