SAFETY-02: the P_PEAK assertion was pinning the test's copy of the gate

Review finding 2, and it is the same duplication `disagrees()` just removed, so
it gets the same fix rather than a follow-up card.

`following_from()` in test_watchdog.py was a TRANSCRIPTION of the gate in
main.py, carrying "keep the two in step" as an instruction - which is exactly the
arrangement that does not keep two things in step. Measured: editing the shipped
gate alone to `winner in ("loop", "peak")` left every suite green while the
watchdog went blind for the duration of every peak event. The row whose whole job
is to force whoever lands peak shaving to come and change an expectation did not
do that.

The expression is now `watchdog.loop_is_following()`, called by main.py and by
the test. One definition, two callers, same reasoning as `disagrees()`: the copy
that drifts is always the one in the harness doing the grading.

⚠️ AND EXTRACTING IT WAS NOT SUFFICIENT - the drift proof still came out green
after the extraction, which is why it was worth running rather than assuming. The
realistic P_PEAK row has peak shaving asking for MORE than the loop, so the
magnitude term rejects it whatever `winner` says and masks the term under test. A
second row with peak asking for LESS makes `winner` the only deciding term. With
that row, widening the shipped gate fails two checks; without it, none.

Also enforcing the write ordering that was documented and not asserted: swapping
the two `store.set` calls passed everything. A store that fails the second write
now proves the property the ordering exists for - a crash mid-trip leaves a
reason with no latch, never a latch with no explanation. Swapping them now turns
that check red.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Du77usMj8XNKNFZGmUiWDa

goodwe_controller/test_watchdog.py

@@ -13,7 +13,9 @@ table-driven half: boundaries, timers, the latch, and the arming gate.
 import sys
 from datetime import datetime, timedelta, timezone
 
-from app.watchdog import SignWatchdog, SignWatchdogConfig
+from app.watchdog import (
+    SignWatchdog, SignWatchdogConfig, loop_is_following,
+)
 
 fails = []
 T0 = datetime(2026, 8, 20, 15, 0, 0, tzinfo=timezone.utc)
@@ -164,11 +166,18 @@ from app.arbiter import (  # noqa: E402
 
 
 def following_from(claims, asked):
-    """Exactly main.py's expression. Keep the two in step."""
+    """THE SHIPPED gate, not a copy of it.
+
+    ⚠️ This used to transcribe main.py's expression with a comment saying
+    "keep the two in step", which is exactly the arrangement that does not keep
+    two things in step. Measured: editing the shipped gate alone to
+    `winner in ("loop", "peak")` left this suite - including the P_PEAK row below,
+    whose entire job is to catch that edit - completely green, while the watchdog
+    went blind for the duration of every peak event. Import it or the P_PEAK
+    assertion is decoration.
+    """
     r = resolve(claims)
-    return r, (r.winner == "loop" and asked is not None
-               and r.target_w * asked > 0
-               and abs(r.target_w) <= abs(asked))
+    return r, loop_is_following(r, asked)
 
 
 MAXW = Claim.limit("supervised", P_SAFETY, -2000, 2000, "max_w")
@@ -211,10 +220,21 @@ TABLE = [
     ("the loop asked for zero",
      [Claim.set("loop", P_LOOP, 0.0), MAXW], 0.0, False),
     # The forward hazard, asserted as it stands TODAY rather than as it should
-    # be. When peak shaving lands, this expectation is what has to change.
-    ("peak shaving at P_PEAK disarms the gate (see main.py's forward hazard)",
+    # be. When peak shaving lands, these expectations are what have to change.
+    ("peak shaving at P_PEAK disarms the gate (loop_is_following's forward hazard)",
      [Claim.set("loop", P_LOOP, 900),
       Claim.set("peak", P_PEAK, 1800, "shaving"), MAXW], 900, False),
+    # ⚠️ AND THE ROW ABOVE CANNOT DETECT A WIDENED GATE ON ITS OWN, which was
+    # found by trying it: with the shipped gate edited to
+    # `winner in ("loop", "peak")` that case STILL comes out False, because peak
+    # shaving asked for MORE than the loop and the magnitude term rejects it
+    # anyway. The winner term is masked. This row isolates it - a peak claim
+    # smaller than the loop's ask and pointing the same way, so `winner` is the
+    # only term that can decide - and it is what actually turns red if the gate
+    # is widened without anyone revisiting the hazard.
+    ("...and with peak BELOW the loop's ask, only the winner term separates them",
+     [Claim.set("loop", P_LOOP, 1800),
+      Claim.set("peak", P_PEAK, 900, "shaving"), MAXW], 1800, False),
 ]
 for name, claims, asked, expect in TABLE:
     r, got = following_from(claims, asked)
@@ -414,6 +434,55 @@ fresh = Controller({"setpoint_entity": "number.sp", "meter_source": "off"},
 check("...so the next restart comes back armed and untripped, as it should",
       not fresh.sign_wd.tripped)
 
+# ⚠️ THE WRITE ORDERING IS AN INVARIANT, SO IT GETS AN ASSERTION. main.py writes
+# `sign_reason` and THEN `sign_tripped`, each one its own `save()`, so that a
+# process that dies between the two leaves a reason nothing reads - harmless -
+# rather than a latch with no explanation for the operator who has to decide
+# whether to clear it. Documented reasoning is not enforcement: swapping those two
+# lines used to pass every check in this file.
+#
+# The death is modelled as the second write raising. In the field it is a power
+# cut or a container kill (Store.save() swallows OSError rather than raising), so
+# what is asserted here is the PROPERTY - what survives on disk - not the
+# mechanism.
+
+
+class _CrashOnFlag(_MemStore):
+    def set(self, k, v):
+        if k == "sign_tripped" and v:
+            raise RuntimeError("power cut, mid-trip")
+        super().set(k, v)
+
+
+async def _crash_mid_write():
+    store = _CrashOnFlag()
+    hass = _Hass()
+    opts = {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
+            "soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
+            "meter_source": "off", "max_w": 2000, "gain": -0.6,
+            "sign_disagree_w": 300, "sign_disagree_s": 1, "auto_start": True}
+    c = Controller(opts, hass, store, _P(), stop=asyncio.Event())
+    c.auto = True
+    c.dev_min, c.dev_max = -5000.0, 5000.0
+    died = False
+    for _ in range(12):
+        try:
+            await c.read_inputs()
+            await c.cycle()
+        except RuntimeError:
+            died = True
+            break
+        await asyncio.sleep(0.25)
+    return store, died
+
+
+crashed_store, died = asyncio.run(_crash_mid_write())
+check("the second write is the one that sets the flag", died)
+check("a crash between the two writes leaves the REASON on disk",
+      "SIGN_DISAGREEMENT" in crashed_store.data.get("sign_reason", ""))
+check("...and never a latch with nothing to explain it",
+      not crashed_store.data.get("sign_tripped"))
+
 print()
 if fails:
     print(f"{len(fails)} FAILED: {', '.join(fails)}")