"""Runnable check for SAFETY-119: no "Battery is not keeping up" when the
loop takes the wire back from another owner. `python3 test_safety119.py`

During a maintenance charge with no usable peak forecast, the measured peak
guard steps in during a quarter's last seconds and hands back. On the cycle
the loop gets the wire back, its integrator is re-seeded from the other
owner's command and kicked by a meter reading taken before the battery moved.
It crossed its rail for one or two cycles, the loop's ask beat the one-sided
limit, so the cycle counted as the loop's own and raised
`goodwe_integrator_bound`. The battery gave what it was told. Now the loop's
bound does not count until the plant's settle (SHORT_CYCLES cycles and
SHORT_SPAN_S) has passed since the hand-back. SAFETY-121: the settle is
HANDBACK_CYCLES (4) cycles and SHORT_SPAN_S (22 s), and across all
hand-backs the bound is hidden on at most ROLLING_BOUND_CYCLES (11) cycles in
any ROLLING_S (900 s), at every cycle length.
DECISIONS/SAFETY-119-bound-after-hand-back.md,
DECISIONS/SAFETY-121-bound-every-quarter.md.

1. The rig's shape (1 s meter that lags the battery, 1 s battery lag, house
   800 W, rating 2500 W, limit 2500 W, margin 50, charge 2000 W, no
   forecast), 2 h, three seeds: no notification, the cap held, every command
   the same as without the fix; without the fix it warns.
2. A genuine integrator bound still warns: the loop in charge with the house
   over the rating (at once, as before), and the loop at its rail after a
   hand-back (after the settle).
3. With a usable forecast: the same commands and the same alarms as without
   the fix.
4. Repeated hand-backs (N=2, the guard and the charge take turns), 2, 5, 10
   and 15 s meters: it warns in every quarter, both with the meter a cycle
   behind the battery reading and in the rig's shape (SAFETY-121, F-1: the
   meter read in the same cycle, 1 to 5 bound cycles a quarter), where
   SAFETY-119's settle of 12 cycles hid every quarter but one; the
   quarter-edge case stays silent at 2, 5, 10 and 15 s; the budget's margins
   (8 warns at the quarter edge, 14 misses quarters); a start or a blind
   cycle is not a hand-back; the delay after one hand-back.
5. The rule at its boundaries (`Controller._settling`): the settle, the
   rolling budget, and a stop that does not give the budget back.
6. test_safety84's and test_safety87's re-pinned digests: the same commands
   and claims; only OUTPUT_CLAMP log rows differ.

Mutations (MUTATION.md, purged copies) are listed in the pull request.
"""

import logging
import math
import os
import random
import sys
import time
from datetime import timedelta

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 safety84_identity as s84                                # noqa: E402
import app.main as app_main                                    # noqa: E402
from app import messages                                       # noqa: E402
from app.main import Controller                                # noqa: E402
from app.p1 import make_sample, split_signed                   # noqa: E402
from named_absence import first, nth, present                  # noqa: E402

fails = []
OPT0 = dict(s84.OPT)
CAP = 2500.0
QUARTER = 900.0


def check(name, cond):
    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
    if not cond:
        fails.append(name)


async def _call(self, *a):
    """The rig's Home Assistant, recording each notification (time, id)."""
    if len(a) >= 3 and a[1] == "create" and isinstance(a[2], dict):
        self.__dict__.setdefault("notes", []).append(
            (self.t, a[2].get("notification_id")))
    return True


def _power(self, u, lag=1.0):
    """The TEST-23 plant's battery: 800 W/s ramp, 1 s first-order lag."""
    st = self.__dict__.setdefault("_plant", {})
    a, tgt, tl = st.get(u, (0.0, 0.0, self.t))
    for _ in range(int(round((self.t - tl) / 0.1))):
        d = self.cmd[u] - tgt
        tgt += math.copysign(min(abs(d), 80.0), d)
        a += (tgt - a) * (1 - math.exp(-0.1 / lag))
    st[u] = (a, tgt, self.t)
    if self.cut(u, self.t) and self.cmd[u] >= -4.0:
        return -4.0
    return float(round(a))


_real_number, _real_power = rig._Hass.number, rig._Hass.power
_real_feed = s79._feed


async def _number(self, e, invert=False):
    """sensor.fc: a usable demand forecast of 800 W (part 3)."""
    if e == "sensor.fc":
        return 800.0
    return await _real_number(self, e, invert)


rig._Hass.call, rig._Hass.power, rig._Hass.number = _call, _power, _number
rig.TICK = 1.0
DELAY = [1]


def _feed(c, t):
    """The P1 meter `DELAY` ticks behind the battery, as on the rig (the
    meter reached HA after the battery reading). Records the true grid."""
    h = c.hass.__dict__.setdefault("grid_true", [])
    h.append((t, c.hass.grid()))
    g = h[max(0, len(h) - 1 - DELAY[0])][1]
    imp, exp = split_signed(g)
    c.p1.submit(make_sample("ha_signed", imp, exp, phases=1,
                            ingest_ts=rig.BASE + timedelta(seconds=t),
                            ingest_mono=time.monotonic()))


s79._feed = _feed


def house(seed, base=800.0, step=None):
    """800 W +- 30 W, a new value each second; `step` = (t, W) from then."""
    rnd = random.Random(seed)
    noise = [rnd.uniform(-30.0, 30.0) for _ in range(4000)]

    def load(t):
        if step is not None and t >= step[0]:
            return step[1]
        return base + noise[int(t) % 4000]
    return load


def run(load, *, n=1, dur=7200.0, delay=1, extra=None, cheap=True, cut=None,
        charging=True, blind=0.0):
    """One run of `n` inverters of 2500 W; `charging=False`: no maintenance,
    the loop in charge; `blind`: no meter reading until that second."""
    extra = dict(extra or {})
    s84.OPT = dict(OPT0, peak_cap_w=int(CAP * n), peak_margin_w=50,
                   maintenance_charge_w=2000)
    if not charging:
        extra.update(maintenance_enabled=False, peak_cap_w=20000)
    for k in [k for k in extra if k in s84.OPT]:
        s84.OPT[k] = extra.pop(k)
    DELAY[0] = delay
    try:
        feed = s84.feed(cheap)

        def before(c, t):
            if t >= blind:                  # before that: no meter reading
                return feed(c, t)
        return s84.run(n, load=load, cheap=cheap,
                       store=(s79.store_charging(n) if charging
                              else rig._Store()),
                       soc=lambda u, t: 60.0, dur=dur, cut=cut,
                       before=before if blind else None,
                       extra=dict({"max_w": 2500, "slew_w": 0}, **extra))
    finally:
        s84.OPT = OPT0


def patched(obj, name, value):
    def wrap(fn):
        real = getattr(obj, name)
        setattr(obj, name, value)
        try:
            return fn()
        finally:
            setattr(obj, name, real)
    return wrap


no_fix = patched(Controller, "_settling", lambda self, owns, now: False)


def notes(c):
    return [t for t, i in getattr(c.hass, "notes", [])
            if i == messages.ID_LOOP_BOUND]


def quarters(c, tick=1.0):
    """Each whole quarter's true average import (W)."""
    q = {}
    for t, g in c.hass.grid_true:
        q.setdefault(int(t // QUARTER), []).append(g)
    return [sum(v) / len(v) for v in q.values()
            if len(v) == QUARTER / tick]


def commands(rows):
    return [(r[0], r[1], r[2]) for r in rows]


def guard_steps(c):
    return sum(1 for e in c.events if "PEAK_GUARD:" in e)


# ============================================================================ #
print("1. a maintenance charge with no forecast, the guard steps in at each "
      "quarter's end, 2 h")
NEW, OLD = {}, {}
for seed, delay in ((1, 1), (2, 1), (3, 2)):
    NEW[seed] = run(house(seed), delay=delay)
    OLD[seed] = no_fix(lambda: run(house(seed), delay=delay))
    print(f"    seed {seed} meter {delay} s behind: notifications "
          f"{notes(NEW[seed][1])} (without the fix {notes(OLD[seed][1])}), "
          f"highest quarter {max(quarters(NEW[seed][1])):.0f} W, guard "
          f"step-ins in the last 40 log rows {guard_steps(NEW[seed][1])}")
check("⭐ no 'Battery is not keeping up' notification, cycle or episode",
      all(notes(c) == [] and c.clamp_wd.bound == 0
          and c.clamp_wd.bound_episodes == 0 for _r, c in NEW.values()))
check("⭐ the peak cap is held: every quarter's average is under 2500 W",
      all(len(quarters(c)) == 8 and max(quarters(c)) < CAP
          for _r, c in NEW.values()))
check("⭐ control is unchanged: every command the same as without the fix",
      all(commands(NEW[k][0]) == commands(OLD[k][0]) for k in NEW))
check("...the charge ran and the guard stepped in near each quarter's end",
      all(any(r[1] == -2000.0 for r in rows) and any(
          r[-1][1] and r[-1][1][0] and r[0] % QUARTER >= QUARTER - 40
          for r in rows) for rows, _c in NEW.values()))
check("non-vacuous: without the fix every run warns, at a quarter's edge",
      all(len(notes(c)) >= 1 for _r, c in OLD.values())
      and all(min(t % QUARTER, QUARTER - t % QUARTER) <= 40
              for _r, c in OLD.values() for t in notes(c)))

# ============================================================================ #
print("\n2. a genuine integrator bound still warns")
_rail = run(house(1, base=4000.0), dur=600.0,
            charging=False)
_rail_old = no_fix(lambda: run(house(1, base=4000.0), dur=600.0,
                                 charging=False))
print(f"    loop in charge, house 4000 W over the 2500 W rating: "
      f"{notes(_rail[1])}, without the fix {notes(_rail_old[1])}")
check("⭐ the loop in charge at its rail warns (log and notification), at "
      "the same cycle as before",
      len(notes(_rail[1])) >= 1 and notes(_rail[1]) == notes(_rail_old[1])
      and _rail[1].clamp_wd.bound == _rail_old[1].clamp_wd.bound
      and any("OUTPUT_CLAMP episode" in e for e in _rail[1].events))

# The charge sets the wire until 600 s, then a solar surplus of 6000 W: the
# loop takes the wire back and stands at its -2500 W rail.
_hb = run(house(1, step=(600.0, -6000.0)), dur=1200.0,
          extra={"peak_cap_w": 20000})
_hb_old = no_fix(lambda: run(house(1, step=(600.0, -6000.0)), dur=1200.0,
                             extra={"peak_cap_w": 20000}))
_first = notes(_hb[1])[:1]
print(f"    hand-back to a loop at its rail at 600 s: first notification "
      f"{_first}, without the fix {notes(_hb_old[1])[:1]}; command at the "
      f"end {_hb[0][-1][1]}")
MB = 11                       # the budget (at 1 s it ends the settle)
check(f"⭐ the loop at its rail after a hand-back still warns, once "
      f"{MB} of its bound cycles are hidden (1 s meter: {MB} s)",
      _first and 600.0 + MB <= _first[0] <= 600.0 + MB + 4.0
      and _hb[0][-1][1] == -2500.0
      and _hb[1].clamp_wd.bound > 100)
check("...without the fix it warned at once (non-vacuous)",
      notes(_hb_old[1])[:1] and notes(_hb_old[1])[0] < 600.0 + MB)
# The same charge with no forecast; at 1795 s, under the guard's step-in at
# the quarter's end, the house goes to 4000 W, over the 2500 W rating.
_up = run(house(1, step=(1795.0, 4000.0)), dur=1900.0)
_up_old = no_fix(lambda: run(house(1, step=(1795.0, 4000.0)), dur=1900.0))
print(f"    guard step-in, then the house over the rating from 1795 s: "
      f"{notes(_up[1])}, without the fix {notes(_up_old[1])}")
check("⭐ a genuine bound in the rig's own scenario still warns, within the "
      "settle of the guard's hand-back",
      notes(_up[1]) and 1795.0 <= notes(_up[1])[0] <= 1795.0 + 40.0
      and any("OUTPUT_CLAMP episode" in e for e in _up[1].events))

# ============================================================================ #
print("\n3. with a usable forecast (800 W): unchanged")
FC = {"peak_forecast_entity": "sensor.fc"}
_fc = run(house(1), extra=FC)
_fc_old = no_fix(lambda: run(house(1), extra=FC))
print(f"    notifications {notes(_fc[1])} (without the fix "
      f"{notes(_fc_old[1])}), highest quarter "
      f"{max(quarters(_fc[1])):.0f} W, charge "
      f"{min(r[1] for r in _fc[0][:300]):.0f} W")
check("⭐ the same commands and the same alarms as without the fix, and "
      "the cap held",
      commands(_fc[0]) == commands(_fc_old[0])
      and notes(_fc[1]) == notes(_fc_old[1])
      and _fc[1].clamp_wd.engagements == _fc_old[1].clamp_wd.engagements
      and max(quarters(_fc[1])) < CAP)
check("...the forecast was used (the charge is shaped under -2000 W)",
      -2000.0 < min(r[1] for r in _fc[0][:300]) < 0.0)

# ============================================================================ #
print("\n4. repeated hand-backs: the guard and the charge take turns")


def round1(self, owns, now):
    """Round 1's rule, kept here to show what it hid: every hand-back
    restarts the window, so a loss within the settle never lets it end."""
    d = self.__dict__
    if not owns:
        d["_r1"], d["_r1_n"] = True, None
        return False
    if d.get("_r1"):
        d["_r1"], d["_r1_n"] = False, (now, 0)
    if d.get("_r1_n") is None:
        return False
    t0, n = d["_r1_n"]
    d["_r1_n"] = (t0, n + 1)
    if n + 1 >= app_main.SHORT_CYCLES and (
            now is None or t0 is None or now - t0 >= app_main.SHORT_SPAN_S):
        d["_r1_n"] = None
        return False
    return True


# N=2 (2 x 2500 W), a cheap-window charge, no usable forecast, the house 50 W
# over the 5000 W rating from 300 s: the loop never held the wire for more
# than 10 cycles. Round 1: no notification in 2 h; round 2: none at 10 s or
# 15 s meters.
def never(tick, hw=5050.0, fix=True, seed=0, delay=1):
    """`seed` 0: a constant house; else 800 W +- 30 W, `hw` +- 30 W from
    300 s. `delay` 0 is the rig's shape: the meter read in the same cycle
    as the battery (SAFETY-121)."""
    rig.TICK = tick
    if seed:
        load = (lambda t, h=house(seed):
                h(t) + (hw - 800.0 if t >= 300.0 else 0.0))
    else:
        load = house(1, step=(300.0, hw))
    try:
        f = lambda: run(load, n=2, delay=delay)
        return f() if fix else no_fix(f)
    finally:
        rig.TICK = 1.0


def per_quarter(c):
    return [sum(1 for t in notes(c) if int(t // QUARTER) == q)
            for q in range(8)]


def every_quarter(c):
    return all(n >= 1 for n in per_quarter(c)[1:])


_nv = {t: never(t) for t in (2.0, 5.0, 10.0, 15.0)}
_nv_old = {t: never(t, fix=False) for t in (2.0, 5.0, 10.0, 15.0)}
_nv_r1 = patched(Controller, "_settling", round1)(lambda: never(5.0))
for _t in _nv:
    print(f"    N=2, {_t:.0f} s meter, house 5050 W: notifications "
          f"{len(notes(_nv[_t][1]))} (per quarter "
          f"{per_quarter(_nv[_t][1])}), without the fix "
          f"{len(notes(_nv_old[_t][1]))}, highest quarter "
          f"{max(quarters(_nv[_t][1], _t)):.0f} W")
print(f"    round 1's rule at 5 s: {len(notes(_nv_r1[1]))}")
check("⭐ 2, 5, 10 and 15 s meters: it warns in every quarter after the "
      "first (round 2 gave none at 10 and 15 s)",
      all(every_quarter(_nv[t][1]) for t in _nv))
check("non-vacuous: round 1's rule (a window per hand-back) never warned "
      "here, and the fix still hides some of what was raised without it",
      notes(_nv_r1[1]) == [] and all(
          _nv[t][1].clamp_wd.bound < _nv_old[t][1].clamp_wd.bound
          for t in _nv))
check("...every command is the same as without the fix, at 2, 5, 10 and "
      "15 s",
      all(commands(_nv[t][0]) == commands(_nv_old[t][0]) for t in _nv))
check("...and at 5 s the cap held",
      len(quarters(_nv[5.0][1], 5.0)) == 8
      and max(quarters(_nv[5.0][1], 5.0)) < 5000.0)

# SAFETY-121, the rig's shape (TEST-23, F-1): the meter is read in the same
# cycle as the battery, so the guard's charge headroom is right after it
# hands back and the guard and the charge do not chase each other. The loop
# then holds the wire once a quarter, at its start, and is at its bound on
# 1 to 5 cycles (5 s), not 20. House 5100 W +- 30 W (5300 W at 15 s, where
# 5100 W is over the rating in one quarter only, without the fix too).
RIG = {1.0: 5100.0, 2.0: 5100.0, 5.0: 5100.0, 10.0: 5100.0, 15.0: 5300.0}
_rg = {t: never(t, hw, seed=1, delay=0) for t, hw in RIG.items()}
_rg_old = {t: never(t, hw, fix=False, seed=1, delay=0)
           for t, hw in RIG.items()}
# SAFETY-119 as merged (bee7431) at 5 s: a settle of 12 cycles, budget 8.
_rg_119 = patched(app_main, "HANDBACK_CYCLES", app_main.SHORT_CYCLES)(
    lambda: patched(app_main, "ROLLING_BOUND_CYCLES", 8)(
        lambda: never(5.0, 5100.0, seed=1, delay=0)))
for _t in _rg:
    print(f"    rig shape, {_t:.0f} s meter, house {RIG[_t]:.0f} W: per "
          f"quarter {per_quarter(_rg[_t][1])}, without the fix "
          f"{per_quarter(_rg_old[_t][1])}, highest quarter "
          f"{max(quarters(_rg[_t][1], _t)):.0f} W")
print(f"    rig shape, 5 s, SAFETY-119's settle and budget: per quarter "
      f"{per_quarter(_rg_119[1])}")
check("⭐ the rig's shape at 1, 2, 5, 10 and 15 s meters: it warns in every "
      "quarter after the first, as without the fix",
      all(every_quarter(_rg[t][1]) and every_quarter(_rg_old[t][1])
          for t in _rg))
check("non-vacuous (F-1): with SAFETY-119's settle of 12 cycles and budget "
      "of 8, the rig's shape at 5 s warns in at most one quarter",
      sum(1 for n in per_quarter(_rg_119[1])[1:] if n) <= 1)
check("...every command the same as without the fix, and the cap held at "
      "1 to 10 s",
      all(commands(_rg[t][0]) == commands(_rg_old[t][0]) for t in _rg)
      and all(max(quarters(_rg[t][1], t)) < 5000.0
              for t in (1.0, 2.0, 5.0, 10.0)))

# The quarter-edge case (part 1) at the slower meters: still no warning.
# The meter's own lag limits the cap there, the same with and without the fix.
_qe = {}
for _t in (2.0, 5.0, 10.0, 15.0):         # with part 1: 1, 2, 5, 10, 15 s
    rig.TICK = _t
    for _seed, _dl in ((1, 1), (2, 1), (3, 2)):
        _qe[_t, _seed] = (run(house(_seed), delay=_dl),
                          no_fix(lambda: run(house(_seed), delay=_dl)))
    rig.TICK = 1.0
    print(f"    {_t:.0f} s meter, quarter edge: notifications "
          f"{[len(notes(r[0][1])) for k, r in _qe.items() if k[0] == _t]} "
          f"(without the fix "
          f"{[len(notes(r[1][1])) for k, r in _qe.items() if k[0] == _t]})")
check("⭐ the quarter-edge case at 2, 5, 10 and 15 s meters, seeds 1 to 3: "
      "no notification",
      all(notes(r[0][1]) == [] and r[0][1].clamp_wd.bound == 0
          for r in _qe.values()))
check("...every command the same as without the fix; without it the false "
      "alarm is there at each meter (non-vacuous)",
      all(commands(r[0][0]) == commands(r[1][0]) for r in _qe.values())
      and all(any(len(notes(_qe[_t, k][1][1])) >= 1 for k in (1, 2, 3))
              for _t in (2.0, 5.0, 10.0, 15.0)))

# The budget's margins (DECISIONS/SAFETY-121): 9 to 13 met both cases in the
# sweep. 8 warns at the quarter edge (2 s, seed 25); 14 misses quarters in
# the rig's shape (1 s, house 5050 W, seed 1).
rig.TICK = 2.0
_q25 = (run(house(25), delay=2),
        patched(app_main, "ROLLING_BOUND_CYCLES", 8)(
            lambda: run(house(25), delay=2)))
rig.TICK = 1.0
_m14 = patched(app_main, "ROLLING_BOUND_CYCLES", 14)(
    lambda: never(1.0, 5050.0, seed=1, delay=0))
_m11 = never(1.0, 5050.0, seed=1, delay=0)
print(f"    2 s, quarter edge, seed 25: {notes(_q25[0][1])}, budget 8 "
      f"{notes(_q25[1][1])}; rig shape 1 s 5050 W per quarter "
      f"{per_quarter(_m11[1])}, budget 14 {per_quarter(_m14[1])}")
check("...the budget's margins: 11 is silent at 2 s seed 25 and warns every "
      "quarter in the rig's shape at 1 s and 5050 W; 8 warns there, 14 "
      "misses a quarter there",
      notes(_q25[0][1]) == [] and len(notes(_q25[1][1])) >= 1
      and every_quarter(_m11[1]) and not every_quarter(_m14[1]))
check("the rule is the one the sweep chose. Changing it means re-running "
      "the sweep in DECISIONS/SAFETY-121-bound-every-quarter.md",
      app_main.HANDBACK_CYCLES == 4 and app_main.ROLLING_BOUND_CYCLES == 11
      and app_main.ROLLING_S == 900.0 and app_main.SHORT_SPAN_S == 22.0)

# A start, or a blind cycle, is not a hand-back: the loop at its rail after
# 20 s of not running warns at once (R2: a restart counted as a hand-back).
_bl = run(house(1, base=4000.0), dur=120.0, charging=False, blind=20.0)
_bl_old = no_fix(lambda: run(house(1, base=4000.0), dur=120.0,
                             charging=False, blind=20.0))
print(f"    the loop not running until 20 s, then at its rail: "
      f"{notes(_bl[1])}, without the fix {notes(_bl_old[1])}")
check("⭐ a start after a blind spell is not a hand-back: it warns at the "
      "same cycle as without the fix",
      notes(_bl[1])[:1] and 20.0 <= notes(_bl[1])[0] <= 26.0
      and notes(_bl[1]) == notes(_bl_old[1]))

# One hand-back to a loop at its rail at 600 s: the first warning comes when
# the settle ends, 22 s and 4 cycles after it (SAFETY-119: 12 cycles, 40 s at
# 5 s and 80 s at 10 s).
_sl = {}
for _t in (5.0, 10.0):
    rig.TICK = _t
    _sl[_t] = run(house(1, step=(600.0, -6000.0)), dur=900.0,
                  extra={"peak_cap_w": 20000})
    rig.TICK = 1.0
_sl1 = {t: [x for x in notes(r[1]) if x >= 600.0][:1] for t, r in _sl.items()}
print(f"    hand-back to a loop at its rail at 600 s, first notification: "
      f"5 s {_sl1[5.0]}, 10 s {_sl1[10.0]}")
check("⭐ one hand-back (the loop has the wire from the cycle after the "
      "step): the first warning comes 25 s after it at a 5 s meter (the "
      "22 s span), 30 s after it at 10 s (4 cycles)",
      _sl1[5.0] == [630.0] and _sl1[10.0] == [640.0])

# ============================================================================ #
print("\n5. the rule at its boundaries")


class _C:
    _settling = Controller._settling

    def __init__(self):
        self._wire_lost, self._handback, self._hidden = False, None, ()
        self.loop_clamped = (False, "", False)


def seq(c, steps):
    """(owned, time[, the loop's bound that cycle]) -> does it count not."""
    out = []
    for st in steps:
        c.loop_clamped = (len(st) > 2 and st[2], "", False)
        out.append(c._settling(st[0], st[1]))
    return out


SPAN = app_main.SHORT_SPAN_S
K, M, ROLL = 4, 11, 900.0     # the chosen rule, written out on purpose
_c = _C()
check("from the start, owned every cycle: never settling (no hand-back)",
      seq(_c, [(True, float(i)) for i in range(30)]) == [False] * 30)
_c = _C()
_out = seq(_c, [(False, 0.0)] + [(True, 2.0 * i) for i in range(1, 15)])
check(f"after another owner at 2 s: settling for 11 owned cycles, then not "
      f"(the span, {SPAN:.0f} s, decides)",
      _out == [False] + [True] * 11 + [False] * 3)
_c = _C()
_out = seq(_c, [(False, 0.0)] + [(True, 0.5 * i) for i in range(1, 60)])
_end = _out.index(False, 1) if False in _out[1:] else -1
check(f"at a 0.5 s meter the span decides: settling until {SPAN:.0f} s",
      _end == 1 + int(SPAN / 0.5) and all(_out[1:_end]))
_set = {}
for _tick in (10.0, 30.0, 60.0):
    _c = _C()
    _set[_tick] = seq(_c, [(False, 0.0)]
                      + [(True, _tick * i) for i in range(1, K + 2)])
check(f"at 10, 30 and 60 s the cycles decide: settling for {K - 1} owned "
      f"cycles, then not (SAFETY-119: 11)",
      all(o == [False] + [True] * (K - 1) + [False] * 2
          for o in _set.values()))
_c = _C()
_out = seq(_c, [(False, 0.0), (True, 2.0), (True, 4.0), (False, 6.0),
                (True, 8.0), (True, 10.0)])
check("losing the wire again restarts the settle",
      _out == [False, True, True, False, True, True]
      and _c._handback == (8.0, 2))
_c = _C()
check("no monotonic clock: the cycle count alone decides",
      seq(_c, [(False, None)] + [(True, None)] * (K + 1))
      == [False] + [True] * (K - 1) + [False] * 2)

# One hand-back, the loop's bound on every cycle: the settle (5 s) or the
# budget (1 s) ends the hiding.
_c = _C()
_o5 = seq(_c, [(False, 0.0)] + [(True, 5.0 * i, True) for i in range(1, 9)])
_c = _C()
_o1 = seq(_c, [(False, 0.0)] + [(True, 1.0 * i, True) for i in range(1, 21)])
check("one hand-back, the bound on every cycle: 5 hidden at 5 s (22 s), "
      f"then counted; {M} hidden at 1 s (the budget), then counted",
      _o5 == [False] + [True] * 5 + [False] * 3
      and _o1 == [False] + [True] * M + [False] * (20 - M))


def turns(c, tick, lost, periods, first=0.0, bound=True):
    """A hand-back, then `periods` x (1 owned cycle with the loop's bound,
    `lost` cycles away). Returns [(time, hidden)] of the owned cycles."""
    t, out = first, []
    for _ in range(lost):
        c._settling(False, t)
        t += tick
    for _ in range(periods):
        c.loop_clamped = (bound, "", False)
        out.append((t, c._settling(True, t)))
        t += tick
        for _ in range(lost):
            c._settling(False, t)
            t += tick
    return out, t


# The rolling budget: the loop's bound is hidden on at most M cycles in any
# ROLL seconds, across all hand-backs.
_c = _C()
_o, _ = turns(_c, 5.0, 3, 60)
_hid = [t for t, h in _o if h]
check(f"across hand-backs (one bound cycle per turn, 5 s): the first {M} "
      "are hidden, then every one counts",
      [h for _t, h in _o[:M + 5]] == [True] * M + [False] * 5)
check(f"...never more than {M} hidden in any {ROLL:.0f} s, and they come "
      f"back as the oldest ages out (exactly {M} in the first {ROLL:.0f} s "
      "after the first)",
      all(sum(1 for x in _hid if t <= x < t + ROLL) <= M for t in _hid)
      and sum(1 for x in _hid if x < _hid[0] + ROLL) == M
      and any(x >= _hid[0] + ROLL for x in _hid))
_c = _C()
_c._hidden = (0.0,) * M
_c._wire_lost, _c._handback = False, (ROLL - 30.0, 0)
_c.loop_clamped = (True, "", False)
_a = _c._settling(True, ROLL - 1.0)
_c.loop_clamped = (True, "", False)
_b = _c._settling(True, ROLL)
check(f"the budget refills {ROLL:.0f} s after the oldest hidden cycle: "
      f"counted at {ROLL - 1:.0f} s, hidden again at {ROLL:.0f} s",
      _a is False and _b is True)
_c = _C()
_o1, _t = turns(_c, 2.0, 3, 20)
for _ in range(12):                             # the wire away a whole settle
    _c._settling(False, _t)
    _t += 2.0
_o2, _ = turns(_c, 2.0, 3, 5, first=_t)
check("the budget is not per hand-back or per run: after a whole settle "
      "with the wire away, a new hand-back within the same "
      f"{ROLL:.0f} s hides nothing more",
      [h for _t2, h in _o1[:M]] == [True] * M
      and not any(h for _t2, h in _o2))
_c = _C()
_o1, _t = turns(_c, 2.0, 3, 20)
_ = [_c._settling(True, _t + 2.0 * i) for i in range(13)]
_c._settling(False, _t + 40.0)
_c.loop_clamped = (True, "", False)
check("the loop holding the wire a whole settle ends the window: the next "
      "hand-back hides nothing more while the budget is spent",
      _c._settling(True, _t + 42.0) is False)
_c = _C()
_c._wire_lost = True
_t = 0.0
for _ in range(M + 4):                          # an over-rating: always counts
    _c.loop_clamped = (True, "", True)
    _c._settling(True, _t)
    _c._wire_lost = True
    _t += 5.0
_c._settling(False, _t)
_c.loop_clamped = (True, "", False)
check("an over-rating does not use up the budget",
      _c._settling(True, _t + 5.0) is True and len(_c._hidden) == 1)

# The budget is one number at every cycle length (SAFETY-119 scaled it).
_bud = {}
for _tick in (1.0, 2.0, 3.0, 5.0, 10.0, 15.0, 20.0, 30.0):
    _c = _C()
    _o, _ = turns(_c, _tick, 1, 14)
    _bud[_tick] = sum(1 for _t2, h in _o if h)
_c = _C()
_o60, _ = turns(_c, 60.0, 1, 14)
print(f"    hidden before the first counts, one bound cycle per turn: {_bud}; "
      f"at 60 s, turns of 2 cycles: {sum(1 for _t2, h in _o60 if h)} of 14 "
      "hidden (the budget never runs out: DECISIONS limit)")
check(f"the budget does not follow the cycle length: {M} hidden at 1 to "
      "30 s, then counted",
      set(_bud.values()) == {M})

# No monotonic clock: the budget cannot refill, so a bound that stays is
# counted after the budget however often the wire changes hands.
_c = _C()
_nc = []
for _ in range(40):
    for _ in range(3):
        _c._settling(False, None)
    _c.loop_clamped = (True, "", False)
    _nc.append(_c._settling(True, None))
check(f"no monotonic clock, repeated hand-backs: the budget holds ({M} "
      "hidden, then every bound counts)", _nc == [True] * M + [False] * (40 - M))

# An over-rating (a command over `max_w`) is a bug, never settled away.
_ctl = _hb[1]
_ctl.loop_clamped = (True, "x", True)
_ctl._wire_lost, _ctl._handback, _ctl._hidden = True, None, ()


class _Res:
    winner, bound_by, target_w = "loop", "", 0.0
    unit_bounds = {}


class _D:
    held, target_w = False, 0.0


_ctl._loop_decision, _ctl.allocation = _D(), None
_ctl._loop_alarm_input(_Res())
check("an over-rating on a hand-back cycle still counts (bug=True kept)",
      _ctl.loop_clamped == (True, "x", True) and _ctl._handback is not None)
_ctl.loop_clamped = (True, "y", False)
_ctl._loop_alarm_input(_Res())
check("...while the integrator bound on the next cycle is not counted",
      _ctl.loop_clamped[0] is False)

# A stop ends the window but does not give the budget back (review F2, X3).
_now = _ctl._now_mono
_ctl._hidden = (_now,) * M
_ctl._wire_lost, _ctl._handback = True, (1.0, 3)
_ctl._loop_decision = None                    # stopped or blind
_ctl._loop_alarm_input(_Res())
_st, _kept = (_ctl._wire_lost, _ctl._handback), _ctl._hidden
_ctl._settling(False, _now + 1.0)
_ctl.loop_clamped = (True, "", False)
_after = _ctl._settling(True, _now + 2.0)
check("a stop clears the window (the next start is not a hand-back) and "
      f"keeps the budget: with {M} hidden, the bound after the next "
      "hand-back counts",
      _st == (False, None) and _kept == (_now,) * M and _after is False)
print(f"    after the stop: hidden kept {len(_kept)}, the next hand-back's "
      f"bound hidden {_after}")

# ============================================================================ #
print("\n6. the pinned digests of test_safety84 and test_safety87: only the "
      "log changes")
# Those harnesses run at the 5 s tick with a battery that follows at once.
rig._Hass.power, rig.TICK, s79._feed = _real_power, 5.0, _real_feed
rig._Hass.number = _real_number
import safety87_identity as s87                               # noqa: E402

RUNS = dict({f"s84 {k}": (lambda v=v: s84.run(**v))
             for k, v in s84.SCENARIOS.items()},
            **{f"s87 {k}": (lambda v=v: s87_run(v))
               for k, v in s87.SCENARIOS.items()})


def s87_run(v):
    v = dict(v)
    return s84.run(1, dur=s87.DUR, store=s87.store(v.pop("store", None)), **v)


_moved, _gone = [], 0
for k, fn in RUNS.items():
    new, old = fn()[0], no_fix(fn)[0]
    same = all(a[:-1] == b[:-1] and a[-1][:4] == b[-1][:4]
               for a, b in zip(new, old)) and len(new) == len(old)
    logs = {e for r in old for e in [r[-1][5]] if e} - {
        e for r in new for e in [r[-1][5]] if e}
    _gone += len(logs)
    if not same or any("OUTPUT_CLAMP episode" not in e for e in logs):
        _moved.append(k)
    print(f"    {k}: commands and claims "
          f"{'the same' if same else 'DIFFER'}, log rows differ {len(logs)}, "
          f"digest {s84.digest(new)[:16]}")
check("⭐ in all twelve, every command, loop field, claim, guard decision "
      "and phase is the same as without the fix; the only log rows that "
      "differ are OUTPUT_CLAMP rows (gone, or a lower count)",
      not _moved and _gone > 0)

print()
if fails:
    print(f"FAILED: {len(fails)} check(s)")
    for f in fails:
        print(f"  - {f}")
    sys.exit(1)
print("all SAFETY-119 checks passed")