SAFETY-95: N=1 identity for every drain end; plainer DOCS line

test_safety95.py: the N=1 byte-identity now covers all six drain ends
(Off and stale link were missing); end_at()/measure() read through a
default, so test_named_absence.py passes (it failed on bdec66e); a check
that a dropped battery's start is 0 W even under a pinned charge (it
kills a mutation that survived).

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

goodwe_controller/DOCS.md

@@ -1702,9 +1702,9 @@ When a charge ends, the other batteries take over one slew step a cycle,
 and none starts charging from the grid (SAFETY-93).
 When a battery's link drops, the others take over one slew step a cycle,
 and control does not stall on the dropped battery's power (SAFETY-96).
-When a drain ends (done, Off, timeout, abort, link, refusal), the others
-cover the house at once but never take over its export; from there control
-moves one slew step a cycle (SAFETY-95).
+When a drain ends (done, Off, timeout, abort, lost link, refused), the
+other batteries cover the house at once but do not export its drain; then
+they move one slew step a cycle (SAFETY-95).
 
 #### Capacity-tariff exposure of the grid-following loop
 

goodwe_controller/test_safety95.py

@@ -240,9 +240,10 @@ HOUSES = (1200.0, 4000.0, -1500.0)
 
 
 def end_at(rows):
-    """The first cycle A is not draining, or its command left the drain."""
-    return next(r[0] for r in rows if r[0] >= T and (
-        r[4][0] != DRAIN or r[1][0] < 2000.0))
+    """The first cycle A is not draining, or its command left the drain;
+    None if its drain never ended."""
+    return next((r[0] for r in rows if r[0] >= T and (
+        r[4][0] != DRAIN or r[1][0] < 2000.0)), None)
 
 
 def measure(rows, n, house):
@@ -252,7 +253,11 @@ def measure(rows, n, house):
     grid import; the export beyond what the site exported before the end;
     the largest grid step; the others' largest command step."""
     te = end_at(rows)
-    pre = next(r for r in reversed(rows) if r[0] < te)
+    if te is None:      # no end: every measure is NaN, so every check FAILs
+        return dict(end=math.nan, site_step=math.nan, cmd_export=math.nan,
+                    imp=math.nan, exp=math.nan, gstep=math.nan,
+                    ostep=math.nan, rows=rows)
+    pre = next((r for r in reversed(rows) if r[0] < te), rows[0])
     seg = [r for r in rows if r[0] >= te - 2.0]
     aft = [r for r in rows if te <= r[0] <= te + 120.0]
     others0 = sum(pre[1][1:])
@@ -414,12 +419,11 @@ def digest(rows):
     return hashlib.sha256(repr(rows).encode()).hexdigest()
 
 
-_n1 = [functools.partial(ENDS[k], 1, h) for k in ("complete", "timeout",
-                                                  "abort", "refusal")
+_n1 = [functools.partial(ENDS[k], 1, h) for k in ENDS
        for h in (1200.0, -1500.0)]
-check("⭐ N=1, a drain that completes, times out, is aborted or refused, at "
-      "1200 W and under a surplus: every row byte-identical with and "
-      "without SAFETY-95",
+check("⭐ N=1, every drain end (complete, Off, timeout, abort, stale link, "
+      "refusal), at 1200 W and under a surplus: every row byte-identical "
+      "with and without SAFETY-95",
       [digest(f()[0]) for f in _n1] == [digest(without(f)[0]) for f in _n1])
 _wave = functools.partial(run, 2, lambda t: 2400.0 + 1500 * math.sin(t / 40),
                           store=store(2, last_completed=rig.BASE.isoformat()),
@@ -481,6 +485,13 @@ check("a dropped battery's power is the meter's: a board stuck at 2000 W "
       reseeded(2000.0, {"a": 2000.0, "b": 0.0}, {"a": 2000.0, "b": 0.0},
                -800.0, live={"inverter-b"})[0] == 0.0
       and _c._drain_tail == set())
+_chg = MaintResult(CHARGE, True, -1500.0, False, [])
+check("a drain that ends into a pinned charge: the pin is A's new command "
+      "while A is live, but 0 W once it is dropped (it is sent 0 W)",
+      reseeded(2000.0, {"a": 2000.0, "b": 0.0}, {"a": 2000.0, "b": 0.0},
+               -3500.0, res=_chg)[2] == -1500.0
+      and reseeded(2000.0, {"a": 2000.0, "b": 0.0}, {"a": 2000.0, "b": 0.0},
+                   -3500.0, res=_chg, live={"inverter-b"}) == (0.0, 0.0, 0.0))
 check("no drain ended: nothing moves (the integrator either)",
       reseeded(2000.0, {"a": 2000.0, "b": 0.0}, {"a": 2000.0, "b": 0.0},
                -800.0, ended=()) == (2000.0, 123.0, None))