CI-17 round 2: stale/dead fixtures take rejected_since from the pinned instant; comment made true

The two fixtures ran their cycles at _SRC_NOW but set rejected_since from the
real clock, so the cycle saw inverter-b live while live_units() saw it stale.
Both now use _SRC_NOW, and a check asserts both clocks agree. run_cycle's
comment no longer claims ctl_cycle.py:49 is the only clock read.

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

goodwe_controller/test_peakguard.py

@@ -116,6 +116,7 @@ from math import inf, isinf
 from pathlib import Path
 
 from app import messages
+from app import ctl_cycle as _ctl_cycle
 from app import peakguard
 from named_absence import first, given, nth, present
 
@@ -2787,9 +2788,22 @@ def cycle_claims(run):
     return out, (seen[-1] if seen else [])
 
 
-def run_cycle(ctl):
-    asyncio.get_event_loop_policy().new_event_loop().run_until_complete(
-        ctl.cycle())
+def run_cycle(ctl, at=None):
+    """One `Controller.cycle()`. `at` pins the `now` that cycle() takes
+    (ctl_cycle.py:49) so a fixture laid up to `at` does not age with host load.
+    Other modules still read the real clock. CI-17."""
+    real_dt = _ctl_cycle.datetime
+    if at is not None:
+        class _Pinned(datetime):
+            @classmethod
+            def now(cls, tz=None):
+                return at if tz is None else at.astimezone(tz)
+        _ctl_cycle.datetime = _Pinned
+    try:
+        asyncio.get_event_loop_policy().new_event_loop().run_until_complete(
+            ctl.cycle())
+    finally:
+        _ctl_cycle.datetime = real_dt
     return ctl
 
 
@@ -2801,11 +2815,15 @@ def run_cycle(ctl):
 EVE_MEAN_W = sum(EVE_W) / len(EVE_W)
 c_src, _ = built(TMP / "sources.db")
 c_src.auto, c_src.batt, c_src.soc = True, 0.0, 100.0
+# CI-17: ONE read of the clock, and every cycle on c_src runs AT it. The
+# telegrams end here; `cycle()` used to read the real clock again, 1.3 s later
+# per cycle under --cpus 0.25, so by the 23rd cycle of the sweep the meter's
+# last telegram was older than `max_hold_s` (30 s), the guard counted the gap at
+# its ceiling and filed `peak@site` (ctl_cycle.py:49 -> p1.py max_hold_s).
+_SRC_NOW = datetime.now(timezone.utc)
 _blk_src = datetime.fromtimestamp(
-    int(datetime.now(timezone.utc).timestamp() // QUARTER_S) * QUARTER_S,
-    timezone.utc)
-for _i in range(0, int((datetime.now(timezone.utc)
-                        - _blk_src).total_seconds()) + 1, 5):
+    int(_SRC_NOW.timestamp() // QUARTER_S) * QUARTER_S, timezone.utc)
+for _i in range(0, int((_SRC_NOW - _blk_src).total_seconds()) + 1, 5):
     c_src.p1.submit(sample(_blk_src + timedelta(seconds=_i), EVE_MEAN_W))
 c_src.grid = EVE_MEAN_W
 
@@ -2817,7 +2835,7 @@ SRC_SWEEP = sorted({float(s) for s in range(0, 101, 5)}
 IDLE_SRC = {}
 for _soc in SRC_SWEEP:
     c_src.soc = _soc
-    _, _cl = cycle_claims(lambda: run_cycle(c_src))
+    _, _cl = cycle_claims(lambda: run_cycle(c_src, at=_SRC_NOW))
     IDLE_SRC[_soc] = frozenset(c.source for c in _cl)
 IDLE_SETS = set(IDLE_SRC.values())
 print(f"    a real idle cycle at {len(SRC_SWEEP)} states of charge "
@@ -3141,7 +3159,7 @@ _b2, _restore_b2 = second_inverter(c_src)
 _b2.bms_soc_floor = c_src.inverters[0].bms_soc_floor
 _n2_writes_before = len(c_src.hass.per_entity)
 try:
-    _, _cl2 = cycle_claims(lambda: run_cycle(c_src))
+    _, _cl2 = cycle_claims(lambda: run_cycle(c_src, at=_SRC_NOW))
     _n2_target, _n2_reason = c_src.target, c_src.reason
     _n2_events = list(c_src.events)
     _n2_writes = c_src.hass.per_entity[_n2_writes_before:]
@@ -3337,7 +3355,7 @@ check("...and the action log no longer carries the 'no writer' refusal, "
 _nw, _restore_nw = second_inverter(c_src, name="inverter-c", setpoint="")
 _nw_writes_before = len(c_src.hass.per_entity)
 try:
-    _, _cl_nw = cycle_claims(lambda: run_cycle(c_src))
+    _, _cl_nw = cycle_claims(lambda: run_cycle(c_src, at=_SRC_NOW))
     _nw_target, _nw_reason = c_src.target, c_src.reason
     _nw_events = list(c_src.events)
     _nw_writes = c_src.hass.per_entity[_nw_writes_before:]
@@ -3382,16 +3400,21 @@ check("...and the 0 W is NOT what the site target happened to be - the same "
 # method is asked, which is what makes this a check on `live_units()` instead of
 # a check on a lambda this file wrote.
 _bs, _restore_bs = second_inverter(c_src)
-_bs.rejected_since = datetime.now(timezone.utc) - timedelta(
-    seconds=app_main.LINK_STALE_S + 5)
+# CI-17: from `_SRC_NOW`, the instant the pinned cycles run at; the real clock
+# is later, so `live_units()` (real clock) and the cycle (pinned) agree.
+_bs.rejected_since = _SRC_NOW - timedelta(seconds=app_main.LINK_STALE_S + 5)
 _bs.last_write_ok = False
 _stale_writes_before = len(c_src.hass.per_entity)
 try:
+    check("CI-17: the pinned clock and the real clock agree that inverter B's "
+          "link is stale, so the cycle and `live_units()` see the same roster",
+          not _bs.is_live(_SRC_NOW)[0]
+          and not _bs.is_live(datetime.now(timezone.utc))[0])
     check("PRECONDITION: the SHIPPED `live_units()` - not a lambda - drops the "
           "inverter whose writes have been rejected for a whole layer-1 window, "
           "and keeps the healthy one",
           c_src.live_units() == ("inverter-a",))
-    _, _cl_stale = cycle_claims(lambda: run_cycle(c_src))
+    _, _cl_stale = cycle_claims(lambda: run_cycle(c_src, at=_SRC_NOW))
     _stale_target = c_src.target
     _stale_alloc = c_src.allocation
     _stale_events = list(c_src.events)
@@ -3433,14 +3456,13 @@ check("⚠️ AND IT DOES NOT ABSORB IT SILENTLY, which is the whole of AC 6 and
 # state `live_units()`'s own docstring warns about - the sole inverter dropping
 # takes the SITE to 0 W - so it is driven here rather than left as a sentence.
 _bd, _restore_bd = second_inverter(c_src)
-_gone = datetime.now(timezone.utc) - timedelta(
-    seconds=app_main.LINK_STALE_S + 5)
+_gone = _SRC_NOW - timedelta(seconds=app_main.LINK_STALE_S + 5)  # CI-17: as above
 _bd.rejected_since, _bd.last_write_ok = _gone, False
 c_src.inverters[0].rejected_since = _gone
 c_src.inverters[0].last_write_ok = False
 _dead_writes_before = len(c_src.hass.per_entity)
 try:
-    _, _cl_dead = cycle_claims(lambda: run_cycle(c_src))
+    _, _cl_dead = cycle_claims(lambda: run_cycle(c_src, at=_SRC_NOW))
     _dead_target, _dead_reason = c_src.target, c_src.reason
     _dead_alloc = c_src.allocation
     _dead_events = list(c_src.events)