CI-17: test_peakguard's real-cycle fixture reads the clock once, so host load cannot add peak@site

c_src's telegrams were laid up to one datetime.now(); each cycle() then read
the real clock again (ctl_cycle.py:49). Under --cpus 0.25 a cycle takes ~1.3 s,
so by the 23rd sweep cycle the last telegram was older than max_hold_s (30 s),
the guard counted the gap and filed peak@site. The cycles now run AT the
fixture's own instant. Test only; the exact-set checks are unchanged.

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 instant `cycle()` reads as "now"
+    (ctl_cycle.py:49, its only clock read), so a fixture laid up to `at` is the
+    same fixture however slowly the host runs the cycles after it. 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:]
@@ -3391,7 +3409,7 @@ try:
           "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)
@@ -3440,7 +3458,7 @@ 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)