CI-18: selector carries a change through a test that imports another test

A test first reached as text (a pattern reader) was never expanded again when a later path reached it by import, so a test importing it was missed. Expanded and reached are now tracked apart; the empty diff selects the safety set. Test added. Co-Authored-By: Claude <noreply@anthropic.com> Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Bdqr7PiaHcBnzPituu2yTC
CI-18: checks.yml comments carry the measured run time and the worker count
2026-11-15 21:17:26 +01:00 · 2026-11-15 21:16:05 +01:00 · 2026-11-15 20:54:00 +01:00
14 changed files with 2340 additions and 665 deletions
@@ -0,0 +1,403 @@
+#!/usr/bin/env python3
+"""CI-18: which test_*.py files does a diff affect?  For DEV and REVIEW rounds only.
+
+    python3 .gitea/ci/affected_tests.py                     # merge-base(origin/release/1.0) .. working tree
+    python3 .gitea/ci/affected_tests.py --base A --head B   # a committed range
+    python3 .gitea/ci/affected_tests.py --files f1 f2       # explicit changed paths (no git)
+    ... --names                                             # only the file names, one per line
+
+It prints what it chose and why, then the verdict line `SELECT n of N` or
+`SELECT ALL`. It never replaces the merge gate: the required context
+`checks / unit (pull_request)` runs every test_*.py (checks.yml, run-tests.sh).
+
+WHAT IS SELECTED (the union of):
+  1. every test file that reaches a changed file through the static graph of
+     the tree: imports (absolute, relative, `from pkg import mod`), plus any
+     STRING in the source that names a file (`"predeploy_check.py"`,
+     `"tools/x.py"`, a dotted `app.mqtt` as in `-m` or import_module). Tests
+     launch tools by path and read sources as text, so imports alone miss them.
+     Transitive: a module that imports a changed module carries its tests.
+  2. every changed test file;
+  3. the ALWAYS-RUN SAFETY SET below.
+  A test that walks a directory (`HERE / "app"`, `glob("test_*.py")`) names
+  that directory or a pattern in a string, so it depends on every file there.
+
+FAILS OPEN. Anything it cannot map selects ALL: a changed file that is not
+.py (config.yaml, translations, static, Dockerfile, run-tests.sh, docs, ...),
+a deleted or renamed file, a .py nothing references, a file that does not
+parse, a base ref that does not exist, a diff it cannot read. Doubt is never "skip". (An EMPTY diff selects the
+safety set only: nothing changed.)
+
+OVER-APPROXIMATION IS THE DESIGN. A false edge runs one more file; a missing
+edge would skip one. The scan favours false edges (strings, not comments;
+suffix matching of module names; any directory name a test spells out).
+"""
+
+import argparse
+import ast
+import fnmatch
+import os
+import re
+import subprocess
+import sys
+
+# --- the always-run safety set -------------------------------------------------
+# Name and reason, one each. They run on every selection, whatever changed.
+# Add a file only with a reason and say it here.
+SAFETY_SET = {
+    "test_bytecode.py": "grades the mutation harness itself; a stale-bytecode "
+                        "survivor makes every other 'it went red' claim void",
+    "test_ci_gate.py": "grades the gate's checkout/guard shell; cheap, and it "
+                       "is the file that says what a green means",
+    "test_split_patches.py": "tripwire that tests patch a Controller global "
+                             "where it is read; spans app/main.py and ctl_*.py",
+    "test_defaults_pin.py": "pins every schema default against config.yaml "
+                            "and scans all modules for literal fallbacks",
+    "test_docs.py": "operator docs must say what the code does (validator, "
+                    "entity ids)",
+}
+
+SKIP_DIRS = {".git", "__pycache__", "node_modules"}
+
+
+class Fail(Exception):
+    """The selector cannot answer; the caller selects ALL."""
+
+
+def repo_root(start):
+    p = subprocess.run(["git", "-C", start, "rev-parse", "--show-toplevel"],
+                       capture_output=True, text=True)
+    if p.returncode != 0:
+        raise Fail("not inside a git work tree")
+    return p.stdout.strip()
+
+
+def all_py(root):
+    out = []
+    for d, dirs, files in os.walk(root):
+        dirs[:] = [x for x in dirs if x not in SKIP_DIRS]
+        for f in files:
+            if f.endswith(".py"):
+                out.append(os.path.relpath(os.path.join(d, f), root))
+    return sorted(out)
+
+
+def module_names(rel):
+    """Every dotted suffix a file can be imported as: a/b/c.py -> c, b.c, a.b.c."""
+    parts = rel[:-3].split(os.sep)
+    if parts[-1] == "__init__":
+        parts = parts[:-1]
+    return {".".join(parts[i:]) for i in range(len(parts))} if parts else set()
+
+
+class Index:
+    def __init__(self, files):
+        self.files = files
+        self.by_mod = {}
+        self.by_base = {}
+        self.dirs = {}
+        # directories that hold test files (the project root): naming one is
+        # naming the whole tree, which says nothing about a dependency
+        self.test_dirs = {os.path.basename(os.path.dirname(f))
+                          for f in files
+                          if os.path.basename(f).startswith("test_")}
+        for f in files:
+            for m in module_names(f):
+                self.by_mod.setdefault(m, set()).add(f)
+            self.by_base.setdefault(os.path.basename(f), set()).add(f)
+            for comp in os.path.dirname(f).split(os.sep):
+                if comp:
+                    self.dirs.setdefault(comp, set()).add(f)
+
+    def parents(self, rel):
+        """Package __init__ files an import of `rel` also runs."""
+        out = set()
+        parts = rel.split(os.sep)[:-1]
+        while parts:
+            init = os.sep.join(parts + ["__init__.py"])
+            if init in self.files:
+                out.add(init)
+            parts.pop()
+        return out
+
+
+def docstring_nodes(tree):
+    ids = set()
+    for n in ast.walk(tree):
+        if isinstance(n, (ast.Module, ast.ClassDef, ast.FunctionDef,
+                          ast.AsyncFunctionDef)) and n.body:
+            b = n.body[0]
+            if isinstance(b, ast.Expr) and isinstance(b.value, ast.Constant) \
+                    and isinstance(b.value.value, str):
+                ids.add(id(b.value))
+    return ids
+
+
+# Calls that list a directory: a file making one depends on what the
+# directories it names contain (`ast.walk` is not one, and is excluded).
+ENUM_CALLS = {"glob", "iglob", "rglob", "listdir", "scandir", "iterdir",
+              "walk", "copytree"}
+
+# Edge reasons that mean "executes the other file" (they propagate through a
+# test file); every other reason means "reads it as data or text".
+IMPORT_KINDS = ("import ", "from ", "package of", "dynamic import")
+
+DOTTED = re.compile(r"[A-Za-z_][\w]*(?:\.[A-Za-z_][\w]*)+")
+IDENT = re.compile(r"^[A-Za-z_]\w*$")
+
+
+def edges_of(rel, root, idx):
+    """{file `rel` depends on: first reason}."""
+    path = os.path.join(root, rel)
+    try:
+        with open(path, encoding="utf-8") as fh:
+            src = fh.read()
+        tree = ast.parse(src, filename=rel)
+    except (OSError, SyntaxError, UnicodeDecodeError, ValueError) as e:
+        raise Fail("cannot parse %s: %s" % (rel, e))
+    deps = {}
+
+    def add(target, why):
+        if target != rel:
+            deps.setdefault(target, why)
+
+    def add_mod(name, why):
+        hit = idx.by_mod.get(name)
+        if hit:
+            for t in hit:
+                add(t, why)
+                for p in idx.parents(t):
+                    add(p, "package of " + t)
+
+    here = os.path.dirname(rel).split(os.sep) if os.path.dirname(rel) else []
+    dynamic = False
+    enumerates = False
+    consts = []
+    skip = docstring_nodes(tree)
+    for n in ast.walk(tree):
+        if isinstance(n, ast.Import):
+            for a in n.names:
+                add_mod(a.name, "import " + a.name)
+        elif isinstance(n, ast.ImportFrom):
+            base = n.module or ""
+            if n.level:
+                up = here[:len(here) - (n.level - 1)] if n.level > 1 else here
+                base = ".".join(up + ([base] if base else []))
+            if base:
+                add_mod(base, "from %s import" % base)
+            for a in n.names:
+                if a.name != "*":
+                    add_mod((base + "." if base else "") + a.name,
+                            "from %s import %s" % (base, a.name))
+        elif isinstance(n, ast.Call):
+            fn = n.func
+            name = fn.attr if isinstance(fn, ast.Attribute) else \
+                fn.id if isinstance(fn, ast.Name) else ""
+            obj = fn.value.id if isinstance(fn, ast.Attribute) and \
+                isinstance(fn.value, ast.Name) else ""
+            if name in ("import_module", "__import__", "run_module",
+                        "spec_from_file_location", "run_path"):
+                dynamic = True
+            if name in ENUM_CALLS and obj != "ast":
+                enumerates = True
+        elif isinstance(n, ast.Constant) and isinstance(n.value, str) \
+                and id(n) not in skip and len(n.value) <= 4000:
+            consts.append(n.value)
+    named_dirs = set()
+    for s in consts:
+        for m in re.finditer(r"[\w./-]*\.py\b", s):
+            base = os.path.basename(m.group(0))
+            for t in idx.by_base.get(base, ()):
+                add(t, "names file %r" % base)
+        for tok in DOTTED.findall(s):
+            add_mod(tok, "names module %r" % tok)
+            # `app.mqtt.Thing` or `tools.x.main`: shorten from the right
+            parts = tok.split(".")
+            for i in range(len(parts) - 1, 0, -1):
+                add_mod(".".join(parts[:i]), "names module %r" % tok)
+        if dynamic and IDENT.match(s):
+            add_mod(s, "dynamic import of %r" % s)
+        for piece in set(re.split(r"[/\\]", s)) if ("/" in s or "\\" in s) \
+                else {s}:
+            if piece in idx.dirs and piece not in idx.test_dirs:
+                named_dirs.add(piece)
+    # A file that enumerates directories depends on everything in the ones it
+    # names, and on what a glob pattern it spells matches in its own directory.
+    if enumerates:
+        for d in named_dirs:
+            for t in idx.dirs[d]:
+                add(t, "enumerates directory %r" % d)
+        for s in consts:
+            if "*" in s and s.endswith(".py"):
+                pat = os.path.basename(s)
+                # `test_*.py` globs the tests next to this file; any other
+                # pattern globs the directories this file names, and only when
+                # it names none, its own directory.
+                own = pat.startswith("test_") or not named_dirs
+                for t in idx.files:
+                    inside = (own and os.path.dirname(t)
+                              == os.path.dirname(rel)) or \
+                        any(c in named_dirs for c in os.path.dirname(t)
+                            .split(os.sep))
+                    if inside and fnmatch.fnmatch(os.path.basename(t), pat):
+                        add(t, "pattern %r" % s)
+    return deps
+
+
+def build(root, files):
+    idx = Index(set(files))
+    return {f: edges_of(f, root, idx) for f in files}
+
+
+def is_test(rel):
+    b = os.path.basename(rel)
+    return os.path.dirname(rel) == "goodwe_controller" and \
+        b.startswith("test_") and b.endswith(".py")
+
+
+def git_changes(root, base, head):
+    """[(status, path)] for base..head, or merge-base..working tree."""
+    def run(*a):
+        p = subprocess.run(["git", "-C", root] + list(a),
+                           capture_output=True, text=True)
+        if p.returncode != 0:
+            raise Fail("git %s failed: %s" % (" ".join(a), p.stderr.strip()))
+        return p.stdout
+    if head:
+        mb = run("merge-base", base, head).strip()
+        out = run("diff", "--name-status", "--no-renames", "-z", mb, head)
+    else:
+        mb = run("merge-base", base, "HEAD").strip()
+        out = run("diff", "--name-status", "--no-renames", "-z", mb)
+    toks = out.split("\0")
+    ch = []
+    i = 0
+    while i + 1 < len(toks):
+        ch.append((toks[i][:1], toks[i + 1]))
+        i += 2
+    if not head:
+        for u in run("ls-files", "-o", "--exclude-standard", "-z").split("\0"):
+            if u:
+                ch.append(("A", u))
+    return ch
+
+
+def select(root, changes):
+    """-> (names, reasons, all_reason).  all_reason is not None when ALL."""
+    files = all_py(root)
+    tests = sorted(f for f in files if is_test(f))
+    if not tests:
+        return None, None, "no test_*.py found under the tree"
+    reasons = {}
+
+    def why(t, r):
+        reasons.setdefault(t, []).append(r)
+
+    if not changes:
+        for t in SAFETY_SET:
+            if "goodwe_controller/" + t in tests:
+                why("goodwe_controller/" + t, "always-run safety set (empty "
+                    "diff): " + SAFETY_SET[t])
+        return sorted(reasons), reasons, None
+    changed_py = []
+    for st, p in changes:
+        if st not in ("M", "A"):
+            return None, None, "%s %s (deleted/renamed/type change: the old " \
+                               "importers are not in this tree)" % (st, p)
+        if not p.endswith(".py"):
+            return None, None, "changed file %s is not Python (nothing here " \
+                               "can say which tests read it)" % p
+        if not os.path.exists(os.path.join(root, p)):
+            return None, None, "changed file %s is not in the work tree" % p
+        changed_py.append(p)
+    try:
+        graph = build(root, files)
+    except Fail as e:
+        return None, None, str(e)
+    rev = {}
+    for f, deps in graph.items():
+        for d, r in deps.items():
+            rev.setdefault(d, {})[f] = r
+    for p in changed_py:
+        if is_test(p):
+            why(p, "changed test file")
+            continue
+    # reverse BFS from each changed file
+    for p in changed_py:
+        seen = {p: None}
+        expanded = {p}
+        queue = [p]
+        while queue:
+            cur = queue.pop(0)
+            for f, r in rev.get(cur, {}).items():
+                if f not in seen:
+                    seen[f] = (cur, r)
+                # A test that merely READS another file (names it, globs it)
+                # is affected by it, but is not a dependency of a third test:
+                # test_inverter reading every test_*.py as text must not make
+                # everything that imports a module "affected by test_inverter".
+                # Only an import carries the change on through a test file.
+                if f not in expanded and (not is_test(f)
+                                          or r.startswith(IMPORT_KINDS)):
+                    expanded.add(f)
+                    queue.append(f)
+        hit = [f for f in seen if is_test(f) and f != p]
+        if not hit and not is_test(p):
+            return None, None, "changed %s is referenced by no test, " \
+                               "directly or through another file" % p
+        for t in hit:
+            chain, cur = [], t
+            while seen[cur]:
+                nxt, r = seen[cur]
+                chain.append("%s (%s)" % (os.path.basename(nxt), r))
+                cur = nxt
+            why(t, "reaches changed %s via %s" % (p, " <- ".join(
+                reversed(chain)) if len(chain) > 1 else chain[0]))
+    for t, r in SAFETY_SET.items():
+        if "goodwe_controller/" + t in tests:
+            why("goodwe_controller/" + t, "always-run safety set: " + r)
+    names = sorted(reasons)
+    return names, reasons, None
+
+
+def main(argv=None):
+    ap = argparse.ArgumentParser(description=__doc__.split("\n")[0])
+    ap.add_argument("--base", default="origin/release/1.0")
+    ap.add_argument("--head", default=None,
+                    help="a ref; default: the working tree")
+    ap.add_argument("--files", nargs="*", help="changed paths, repo-relative")
+    ap.add_argument("--root", default=None)
+    ap.add_argument("--names", action="store_true",
+                    help="print only the chosen file names (one per line), "
+                         "or every test name on SELECT ALL")
+    a = ap.parse_args(argv)
+    try:
+        root = a.root or repo_root(os.getcwd())
+        changes = [("M", f) for f in a.files] if a.files is not None \
+            else git_changes(root, a.base, a.head)
+        names, reasons, all_reason = select(root, changes)
+    except Fail as e:
+        names, reasons, all_reason = None, None, str(e)
+        root = a.root or os.getcwd()
+    tests = sorted(f for f in all_py(root) if is_test(f)) \
+        if os.path.isdir(root) else []
+    if all_reason is not None:
+        if a.names:
+            print("\n".join(os.path.basename(t) for t in tests))
+        else:
+            print("SELECT ALL - %s" % all_reason)
+            print("SELECT ALL (%d of %d)" % (len(tests), len(tests)))
+        return 0
+    if a.names:
+        print("\n".join(os.path.basename(t) for t in names))
+        return 0
+    print("changed: %s" % (", ".join(p for _, p in changes) or "(nothing)"))
+    for t in names:
+        for r in reasons[t][:3]:
+            print("  %-34s %s" % (os.path.basename(t), r))
+    print("SELECT %d of %d" % (len(names), len(tests)))
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
@@ -0,0 +1,35 @@
+#!/bin/sh
+# CI-18: run the test files a diff affects, in the CI image, for a dev or review
+# round. NOT the merge gate: `checks / unit (pull_request)` runs all of them.
+#
+#   sh .gitea/ci/run-affected.sh                  # merge-base(origin/release/1.0) .. working tree
+#   sh .gitea/ci/run-affected.sh --base A --head B
+#   INFRA_DIR=/path/to/admin-goodwe@pin sh .gitea/ci/run-affected.sh
+#
+# Arguments go to affected_tests.py (which prints what it chose and why). It
+# selects ALL when it cannot map a change, so a run of this script is never a
+# silent skip. Needs python3, git and docker on the host. INFRA_DIR is the
+# admin/goodwe checkout at the commit in .gitea/infra-pin (test_freeze.py part 3
+# needs sim/hwsim.py); without it test_freeze.py fails, as it does in CI without
+# the checkout, and the script says so.
+set -eu
+repo=$(git rev-parse --show-toplevel)
+cd "$repo"
+python3 .gitea/ci/affected_tests.py "$@"
+only=$(python3 .gitea/ci/affected_tests.py --names "$@" | tr '\n' ' ')
+if [ -z "${only# }" ]; then echo "nothing selected"; exit 0; fi
+IMG=${CI_IMAGE:-goodwe-addon-checks:1}
+docker image inspect "$IMG" >/dev/null 2>&1 || docker build -q -t "$IMG" .gitea/ci >/dev/null
+infra=
+if [ -n "${INFRA_DIR:-}" ]; then
+  infra="-v $INFRA_DIR:/w/goodwe:ro"
+else
+  echo "INFRA_DIR is not set: test_freeze.py will fail if it is selected (it needs admin/goodwe at the pin)."
+fi
+# --network none: the tests need none. --memory: a runaway test must not take the host.
+# shellcheck disable=SC2086
+exec docker run --rm ${CONTAINER_NAME:+--name "$CONTAINER_NAME"} --network none \
+  --memory "${CI_MEMORY:-16g}" -e PYTHONDONTWRITEBYTECODE=1 -e GOODWE_REQUIRE_HWSIM=1 \
+  -e CI_JOBS="${CI_JOBS:-6}" -e CI_ONLY="$only" \
+  -v "$repo:/w/goodwe-addon:ro" $infra "$IMG" \
+  sh -eu /w/goodwe-addon/.gitea/ci/run-tests.sh
@@ -41,15 +41,20 @@
 # Add a file here only with a measured reason, and say it in this list.
 SERIAL="test_loop.py test_freeze.py test_safety87.py"
 #
-# LONGEST FIRST (SAFETY-125's CI run, job 2580; then a local parallel run, seconds:
-# modes_p4 758, safety126 634 (SAFETY-126, local, one core), safety125 508 on hotfix-0.19.2 only, modes_p3 364, safety119 328,
-# price_run_fixes 184, safety90 160, safety103 134, safety94 114, peakguard 113,
-# opt20 102, safety97 101, safety81 100). With JOBS workers pulling from one
-# queue, starting the longest files first keeps the tail short: the batch takes
-# about as long as the longest file. A sort order, not a selection.
-HEAVY="test_modes_p4.py test_safety126.py test_safety125.py test_modes_p3.py test_safety119.py test_price_run_fixes.py test_safety90.py test_safety103.py test_safety94.py test_peakguard.py test_opt20.py test_safety97.py test_safety81.py"
+# LONGEST FIRST. With JOBS workers pulling from one queue, starting the longest
+# files first keeps the tail short: the batch takes about as long as the longest
+# file or the total work divided by JOBS, whichever is more. A sort order, not a
+# selection. Seconds, CI-18 local runs at 10 workers on the 12-core host (one
+# file's time grows with the load; the order is what matters): opt36 653,
+# safety119 704, opt34 515, safety126 402, safety125 388, safety126b 364,
+# modes_p3 377, safety94 360, modes_p4c 291, modes_p4b 238, tariff24_weather 235,
+# modes_p4d 196, price_run_fixes 197, safety103 189, safety90 185, safety97 163.
+# CI-18 split test_modes_p4, test_safety126 and test_safety125 so no one file
+# is much over 400 s; test_opt34 and test_opt36 already fan out inside, and
+# test_safety119 part 4 is one coupled block, so those three stay whole.
+HEAVY="test_opt36.py test_safety119.py test_opt34.py test_safety126.py test_safety125.py test_safety126b.py test_modes_p3.py test_safety94.py test_modes_p4c.py test_modes_p4b.py test_tariff24_weather.py test_modes_p4d.py test_safety125b.py test_price_run_fixes.py test_safety103.py test_safety90.py test_safety97.py test_peakguard.py test_opt20.py test_safety81.py"
 #
-JOBS=${CI_JOBS:-6}
+JOBS=${CI_JOBS:-10}
 cd "${TESTS_DIR:-/w/goodwe-addon/goodwe_controller}"

 # one FILE: run it, keep its output in its own file, print its block whole.
@@ -78,6 +83,21 @@ if [ "$1" = "test_*.py" ]; then
  echo "NO test_*.py FOUND - the mount is empty or the layout moved. NOTHING has been checked."
  exit 1
 fi
+ALL_COUNT=$#
+# CI-18: CI_ONLY="test_a.py test_b.py" runs just those files, for a developer's
+# local round (run-affected.sh sets it). checks.yml never sets it and
+# test_affected_tests.py fails if it ever does: the merge gate is the glob above.
+# Every name must be an existing test_*.py here, and the last line says PARTIAL.
+PARTIAL=0
+if [ -n "${CI_ONLY:-}" ]; then
+  PARTIAL=1
+  for f in $CI_ONLY; do
+    case "$f" in test_*.py) ;; *) echo "CI_ONLY: '$f' is not a test_*.py name"; exit 1 ;; esac
+    [ -f "$f" ] || { echo "CI_ONLY: '$f' does not exist in $(pwd)"; exit 1; }
+  done
+  set -- $CI_ONLY
+  echo "PARTIAL RUN (CI_ONLY): $# of $ALL_COUNT test files. NOT the merge gate."
+fi
 echo "$# test file(s) to run: $JOBS at a time, then the serial group"

 RUN_DIR=$(mktemp -d /tmp/ci-tests.XXXXXX)
@@ -86,14 +106,14 @@ export RUN_DIR
 : > "$RUN_DIR/ser"
 for f in $HEAVY; do
  case " $SERIAL " in *" $f "*) continue ;; esac
-  if [ -f "$f" ]; then echo "$f" >> "$RUN_DIR/par"; fi
+  case " $* " in *" $f "*) if [ -f "$f" ]; then echo "$f" >> "$RUN_DIR/par"; fi ;; esac
 done
 for f in "$@"; do
  case " $SERIAL $HEAVY " in *" $f "*) continue ;; esac
  echo "$f" >> "$RUN_DIR/par"
 done
 for f in $SERIAL; do
-  if [ -f "$f" ]; then echo "$f" >> "$RUN_DIR/ser"; fi
+  case " $* " in *" $f "*) if [ -f "$f" ]; then echo "$f" >> "$RUN_DIR/ser"; fi ;; esac
 done
 echo "parallel: $(wc -l < "$RUN_DIR/par") file(s); serial: $(wc -l < "$RUN_DIR/ser") file(s)"

@@ -121,5 +141,8 @@ for f in "$@"; do
  fi
 done
 echo "$ran file(s), $bad failed, $(( $(date +%s) - t0 ))s wall"
+if [ "$PARTIAL" = 1 ]; then
+  echo "PARTIAL RUN: $ran of $ALL_COUNT test files ran. This is NOT the merge gate; CI runs all $ALL_COUNT."
+fi
 if [ "$bad" = 0 ] && [ "$rc" = 0 ]; then exit 0; fi
 exit 1
@@ -209,7 +209,7 @@ on:
 jobs:
  unit:
    runs-on: [self-hosted, goodwe-rig]
-    timeout-minutes: 40   # ~15 to run (CI-15: 6 files at a time; the serial suite was ~60 and hit the runner cap); the rest is margin for a busy host
+    timeout-minutes: 40   # ~21 to run, measured (CI-18: 10 files at a time; was ~27 at 6, ~60 serial, 22-37 on a busy host); the rest is margin for a busy host
    env:
      # ⚠️ UNDER /home/homelab, NOT $GITHUB_WORKSPACE (ENV-01, and re-measured
      # during CI-01). act's workspace is `/root/.cache/act/<id>/hostexecutor`, a
@@ -534,7 +534,8 @@ jobs:
      # shorter timeout - a 60 s limit was tried during CI-01 and reported
      # exactly the false hang the docstring predicts.
      #
-      # ⚠️ CI-15: THE FILES RUN SIX AT A TIME, NOT ONE AFTER ANOTHER. The serial
+      # ⚠️ CI-15: THE FILES RUN SEVERAL AT A TIME (TEN SINCE CI-18, SIX BEFORE),
+      # NOT ONE AFTER ANOTHER. The serial
      # loop summed 3 557 s on SAFETY-125's run (job 2580) and ran past the
      # runner's 60-minute cap. The loop now lives in .gitea/ci/run-tests.sh (it
      # comes from the merged tree, like the Dockerfile), so the same script runs
@@ -107,10 +107,22 @@ on real hardware, and the comments in `control.py` say which.

 ### CI and the infra pin

-CI (`.gitea/workflows/checks.yml`) runs every `test_*.py` (six at a time, the wall-clock ones alone afterwards: `.gitea/ci/run-tests.sh`), with `admin/goodwe` checked out next to this repo at the commit in `.gitea/infra-pin`, so `test_freeze.py` grades against the real `sim/hwsim.py`. If that checkout fails, the job fails. The local stub is used only in local runs, and it prints `STUB USED` when it is.
+CI (`.gitea/workflows/checks.yml`) runs every `test_*.py` (ten at a time, the wall-clock ones alone afterwards: `.gitea/ci/run-tests.sh`), with `admin/goodwe` checked out next to this repo at the commit in `.gitea/infra-pin`, so `test_freeze.py` grades against the real `sim/hwsim.py`. If that checkout fails, the job fails. The local stub is used only in local runs, and it prints `STUB USED` when it is.
 The pin tracks `admin/goodwe` **release/1.0**, not master. Two non-required checks go red when release/1.0 has moved past the pin: `infra-pin / fresh` here, on any add-on PR or push to release/1.0, and `addon-pin / fresh` in `admin/goodwe`, on every push to its release/1.0. Nothing re-runs `addon-pin` after a bump, so it stays red on that admin/goodwe commit; `infra-pin / fresh` green on the bump is the all-clear.
 **Bump the pin** whenever either is red. Set `.gitea/infra-pin` to the new release/1.0 SHA (`git ls-remote <gitea>/admin/goodwe.git refs/heads/release/1.0`) in a PR, and let `checks / unit` grade the tests against it. CI reads `admin/goodwe` with the Actions secret `INFRA_READ_TOKEN`.

+### Running only the affected tests (dev and review rounds)
+
+`checks / unit` runs every `test_*.py` and stays the only merge gate. For a local round, `sh .gitea/ci/run-affected.sh` runs just the files your diff reaches, in the same `goodwe-addon-checks:1` image and through the same `run-tests.sh`:
+
+```bash
+INFRA_DIR=~/path/to/admin-goodwe-at-the-pin sh .gitea/ci/run-affected.sh           # release/1.0 .. working tree
+sh .gitea/ci/run-affected.sh --base origin/release/1.0 --head my-branch           # a committed range
+python3 .gitea/ci/affected_tests.py --files goodwe_controller/app/price.py        # only list, with the reasons
+```
+
+It prints each chosen file and why, then `SELECT n of N`. A test file is chosen if it imports a changed file, directly or through other modules, or names it in a string (a tool launched by path); a changed test file is chosen; and so is a fixed safety set (`SAFETY_SET` in `affected_tests.py`, one reason each). Anything it cannot map (a non-Python file such as `config.yaml`, a deletion, a `.py` no test reaches) prints `SELECT ALL`. Most modules are imported by `app/main.py`, so a change to `app/` usually selects nearly everything; the saving is for test-only, tool and helper changes. A partial run ends with `PARTIAL RUN ... NOT the merge gate`. `INFRA_DIR` is the `admin/goodwe` checkout at `.gitea/infra-pin`, which `test_freeze.py` needs.
+
 ## Compatibility

 **Check the serial number, not the model name.** Characters 6-8 of the serial are
@@ -0,0 +1,268 @@
+"""Runnable check for CI-18's selector.  `python3 test_affected_tests.py`
+
+`.gitea/ci/affected_tests.py` says which test_*.py files a diff affects, so a
+developer's round can skip the rest. It is used ONLY for local rounds; the merge
+gate (`checks / unit (pull_request)`) runs every file. So the properties graded
+here are the ones whose failure would be a missed test:
+
+  1. A synthetic tree: transitive imports, a tool launched by path, a module
+     named in a string, and what must NOT be chosen.
+  2. FAIL OPEN: a non-Python file, a deletion, a .py nothing references, a
+     file that does not parse, a base that does not exist, all select ALL.
+  3. The real tree: the safety set names files that exist, config.yaml and
+     run-tests.sh select ALL, a changed test file is chosen.
+  4. THE MUTANT: break a real module in a private copy of the tree; the
+     selector must choose a test that goes red, and the unbroken copy is green.
+  5. The merge gate cannot be partial: checks.yml never mentions CI_ONLY and
+     still runs the unfiltered run-tests.sh line.
+Read MUTATION.md before you report a survivor.
+"""
+
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+REPO = os.path.dirname(HERE)
+SEL = os.path.join(REPO, ".gitea", "ci", "affected_tests.py")
+sys.path.insert(0, os.path.dirname(SEL))
+import affected_tests as sel                                    # noqa: E402
+
+fails = []
+tmps = []
+
+
+def check(name, cond):
+    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
+    if not cond:
+        fails.append(name)
+
+
+def mk(root, files):
+    for rel, text in files.items():
+        p = os.path.join(root, rel)
+        os.makedirs(os.path.dirname(p), exist_ok=True)
+        with open(p, "w", encoding="utf-8") as fh:
+            fh.write(text)
+
+
+def tmp():
+    d = tempfile.mkdtemp(prefix="test_affected.")
+    tmps.append(d)
+    return d
+
+
+def choose(root, paths, status="M"):
+    names, _r, why = sel.select(root, [(status, p) for p in paths])
+    return ([os.path.basename(n) for n in names] if names is not None else None,
+            why)
+
+
+# ---------------------------------------------------------------- 1. synthetic
+print("1. a synthetic tree")
+R = tmp()
+P = "goodwe_controller/"
+mk(R, {
+    P + "app/__init__.py": "",
+    P + "app/low.py": "def f():\n    return 1\n",
+    P + "app/mid.py": "from app import low\n\ndef g():\n    return low.f()\n",
+    P + "app/top.py": "from app.mid import g\n",
+    P + "app/lonely.py": "X = 1\n",
+    P + "app/viastr.py": "Y = 2\n",
+    P + "tools/launched.py": "print('hi')\n",
+    P + "tools/reader.py": "print('r')\n",
+    P + "test_a.py": "from app import top\n",
+    P + "test_b.py": "from app import lonely\n",
+    P + "test_c.py": "import subprocess\nsubprocess.run(['python3', 'tools/launched.py'])\n",
+    P + "test_d.py": "import importlib\nm = importlib.import_module('app.viastr')\n",
+    # reads every test_*.py as text: affected by a changed test, but it is a
+    # reader, not a dependency of anything that imports a module
+    P + "test_reader.py": "import glob\nfor f in glob.glob('test_*.py'):\n    open(f).read()\n",
+    P + "test_unrelated.py": "print('x')\n",
+    # a test imported by another test carries the change on (an import runs it)
+    P + "test_base.py": "from app import lonely\n",
+    P + "test_chain.py": "import test_base\n",
+})
+n, why = choose(R, [P + "app/low.py"])
+check("a change two imports down chooses the test that imports the top (transitive)",
+      n is not None and "test_a.py" in n)
+check("...and not tests that do not reach it",
+      n is not None and not {"test_b.py", "test_c.py", "test_unrelated.py"} & set(n))
+n, why = choose(R, [P + "app/lonely.py"])
+check("a direct importer is chosen", n is not None and "test_b.py" in n)
+check("a test that imports another test is chosen with it, even though "
+      "test_reader (a pattern reader) also reaches test_base as text",
+      n is not None and "test_chain.py" in n and "test_base.py" in n)
+n, why = choose(R, [P + "tools/launched.py"])
+check("a tool launched by path in a string chooses the launching test",
+      n is not None and "test_c.py" in n)
+n, why = choose(R, [P + "app/viastr.py"])
+check("a module imported dynamically by name chooses its test",
+      n is not None and "test_d.py" in n)
+n, why = choose(R, [P + "test_unrelated.py"])
+check("a changed test file is chosen", n is not None and "test_unrelated.py" in n)
+check("a test that only READS test files by pattern is chosen with it",
+      n is not None and "test_reader.py" in n)
+check("...but a reader does not drag in the importers of an unrelated module",
+      n is not None and "test_a.py" not in n and "test_b.py" not in n)
+n, why = choose(R, [P + "app/low.py", P + "tools/launched.py"])
+check("two changes choose the union",
+      n is not None and {"test_a.py", "test_c.py"} <= set(n))
+
+# the safety set is always in, whatever changed
+R2 = tmp()
+files = {P + "app/__init__.py": "", P + "app/m.py": "A = 1\n",
+         P + "test_m.py": "from app import m\n", P + "test_zzz.py": "print(1)\n"}
+for name in sel.SAFETY_SET:
+    files[P + name] = "print('safety')\n"
+mk(R2, files)
+n, why = choose(R2, [P + "app/m.py"])
+check("every safety-set file is chosen on a one-module change",
+      n is not None and set(sel.SAFETY_SET) <= set(n))
+check("...with a reason printed for each", all(sel.SAFETY_SET.values()))
+check("...and a test outside the set that does not reach it stays out",
+      n is not None and "test_zzz.py" not in n)
+
+# ------------------------------------------------------------ 2. fails open
+print("\n2. it fails open")
+for label, paths, status in (
+        ("config.yaml (not Python)", [P + "config.yaml"], "M"),
+        ("a translation file", [P + "translations/en.yaml"], "M"),
+        ("run-tests.sh itself", [".gitea/ci/run-tests.sh"], "M"),
+        ("a Dockerfile", [P + "Dockerfile"], "M"),
+        ("a markdown file", ["README.md"], "M"),
+        ("a deleted module", [P + "app/low.py"], "D"),
+        ("a renamed module", [P + "app/low.py"], "R"),
+        ("a .py nothing references", [P + "app/never_imported.py"], "M")):
+    if label.startswith("a .py"):
+        mk(R, {P + "app/never_imported.py": "Z = 1\n"})
+    n, why = choose(R, paths, status)
+    check(f"{label}: SELECT ALL", n is None and why)
+mk(R, {P + "app/broken.py": "def (:\n"})
+n, why = choose(R, [P + "app/low.py"])
+check("a file in the tree that does not parse: SELECT ALL", n is None and why)
+os.remove(os.path.join(R, P + "app/broken.py"))
+n, why = choose(tmp(), [P + "app/low.py"])
+check("a tree with no test files: SELECT ALL", n is None and why)
+
+# the CLI: --names prints every test on SELECT ALL, and a bad base is ALL
+g = tmp()
+mk(g, {P + "app/__init__.py": "", P + "app/m.py": "A = 1\n",
+       P + "test_m.py": "from app import m\n"})
+env = {**os.environ, "GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@t",
+       "GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@t"}
+
+
+def git(*a):
+    return subprocess.run(["git", "-C", g] + list(a), env=env,
+                          capture_output=True, text=True)
+
+
+git("init", "-q", "-b", "base")
+git("add", "-A")
+git("commit", "-q", "-m", "base")
+git("checkout", "-q", "-b", "work")
+mk(g, {P + "app/m.py": "A = 2\n"})
+git("commit", "-q", "-am", "work")
+r = subprocess.run([sys.executable, SEL, "--root", g, "--base", "base",
+                    "--head", "work", "--names"], capture_output=True, text=True)
+check("a committed range via git: the importer is chosen",
+      r.returncode == 0 and r.stdout.split() == ["test_m.py"])
+r = subprocess.run([sys.executable, SEL, "--root", g, "--base", "nosuchref",
+                    "--head", "work"], capture_output=True, text=True)
+check("a base that does not exist: SELECT ALL, not an error and not nothing",
+      r.returncode == 0 and "SELECT ALL" in r.stdout)
+mk(g, {P + "app/m.py": "A = 3\n"})
+r = subprocess.run([sys.executable, SEL, "--root", g, "--base", "base",
+                    "--names"], capture_output=True, text=True)
+check("uncommitted work is part of the diff (no --head)",
+      r.returncode == 0 and r.stdout.split() == ["test_m.py"])
+mk(g, {P + "app/newmod.py": "B = 1\n", P + "test_new.py": "from app import newmod\n"})
+r = subprocess.run([sys.executable, SEL, "--root", g, "--base", "base",
+                    "--names"], capture_output=True, text=True)
+check("an untracked new module and its new test are in the diff",
+      r.returncode == 0 and set(r.stdout.split()) == {"test_m.py", "test_new.py"})
+
+# ---------------------------------------------------------------- 3. real tree
+print("\n3. the real tree")
+tests = sorted(f for f in os.listdir(HERE)
+               if f.startswith("test_") and f.endswith(".py"))
+check("every safety-set name is a test file that exists",
+      all(f in tests for f in sel.SAFETY_SET))
+for p in ("goodwe_controller/config.yaml", ".gitea/ci/run-tests.sh",
+          ".gitea/workflows/checks.yml", "goodwe_controller/DOCS.md",
+          "goodwe_controller/translations/en.yaml"):
+    n, why = choose(REPO, [p])
+    check(f"{p}: SELECT ALL", n is None and why)
+n, why = choose(REPO, ["goodwe_controller/app/maintenance.py"])
+check("app/maintenance.py chooses test_maint25.py (it imports it through main)",
+      n is not None and "test_maint25.py" in n)
+n, why = choose(REPO, ["goodwe_controller/test_maint25.py"])
+check("a changed test chooses itself and the safety set",
+      n is not None and "test_maint25.py" in n and set(sel.SAFETY_SET) <= set(n))
+check("...and far fewer than all",
+      n is not None and len(n) < len(tests) // 2)
+n, why = choose(REPO, [".gitea/ci/affected_tests.py"])
+check("the selector's own file chooses this file",
+      n is not None and "test_affected_tests.py" in n)
+
+# ------------------------------------------------------------------ 4. mutant
+print("\n4. the mutant: break a module, the selector must choose a red test")
+M = tmp()
+ign = shutil.ignore_patterns(".git", "scenarios", "__pycache__", "*.pyc")
+shutil.copytree(REPO, os.path.join(M, "repo"), ignore=ign, symlinks=True)
+mroot = os.path.join(M, "repo")
+cwd = os.path.join(mroot, "goodwe_controller")
+env2 = {**os.environ, "PYTHONDONTWRITEBYTECODE": "1"}
+
+
+def run_test(name):
+    r = subprocess.run([sys.executable, "-B", name], cwd=cwd, env=env2,
+                       capture_output=True, text=True, timeout=600)
+    return r.returncode
+
+
+check("control: test_maint25.py is green on the unbroken copy",
+      run_test("test_maint25.py") == 0)
+path = os.path.join(cwd, "app", "maintenance.py")
+with open(path, encoding="utf-8") as fh:
+    src = fh.read()
+old = "        return until if now < until else None\n"
+check("the line to break is where this test expects it", src.count(old) == 1)
+with open(path, "w", encoding="utf-8") as fh:
+    fh.write(src.replace(old, "        return None\n"))
+n, why = choose(mroot, ["goodwe_controller/app/maintenance.py"])
+check("the selector chooses test_maint25.py for the broken module",
+      n is not None and "test_maint25.py" in n)
+check("...and running the chosen file is RED", run_test("test_maint25.py") != 0)
+with open(path, "w", encoding="utf-8") as fh:
+    fh.write(src)
+check("put back: GREEN again", run_test("test_maint25.py") == 0)
+
+# --------------------------------------------- 5. the gate cannot be partial
+print("\n5. the merge gate runs everything")
+with open(os.path.join(REPO, ".gitea", "workflows", "checks.yml"),
+          encoding="utf-8") as fh:
+    wf = fh.read()
+check("checks.yml never mentions CI_ONLY", "CI_ONLY" not in wf)
+check("checks.yml runs run-tests.sh with no file list",
+      "sh -eu /w/goodwe-addon/.gitea/ci/run-tests.sh\n" in wf)
+check("checks.yml never calls the selector or the local wrapper",
+      "affected_tests" not in wf and "run-affected" not in wf)
+with open(os.path.join(REPO, ".gitea", "ci", "run-tests.sh"),
+          encoding="utf-8") as fh:
+    rt = fh.read()
+check("run-tests.sh takes CI_ONLY from the environment only and says PARTIAL",
+      "PARTIAL RUN" in rt and 'set -- test_*.py' in rt)
+
+for d in tmps:
+    shutil.rmtree(d, ignore_errors=True)
+print()
+if fails:
+    print(f"{len(fails)} FAILED:")
+    for f in fails:
+        print(f"  - {f}")
+    sys.exit(1)
+print("all checks passed")
@@ -22,7 +22,11 @@ record.
 8. Options: validation of the trust and the entities.

 Mutations (MUTATION.md, purged copies) are listed in the pull request.
-"""
+
+This file is parts 1-5 and 8. Part 6 (the real controller over a night and a day,
+the slow one) and part 7 are in test_modes_p4b.py, test_modes_p4c.py and
+test_modes_p4d.py, one group of scenarios each, so CI can run them side by side
+(CI-18). Every check is in exactly one of the four files."""

 import asyncio
 import logging
@@ -549,340 +553,6 @@ e = modetrade.energy([pk(B, packmode.LEFTOVER_SOLAR, 30.0)], 20.0, 95.0,
 check("the cheap run is sized by the shortfall (1.5 kWh), not the room to "
      "95 % (6.5 kWh)", abs(e[0] - 1.5) < 1e-9)

-# ============================================================================ #
-print("6. the real controller over a night and a day")
-AIM = rp.PEAK_CAP - 50
-FC_KWH = [None]             # today's forecast in the fake sensor (None: down)
-
-
-async def _sensor(self, e):
-    now = rp_start[0] + timedelta(seconds=self.t)
-    kwh = FC_KWH[0](now) if callable(FC_KWH[0]) else FC_KWH[0]
-    if kwh is None:
-        return {"state": "unavailable", "attributes": {},
-                "last_updated": now.isoformat()}
-    if e.endswith("tomorrow"):
-        return None
-    loc = now.astimezone(L)
-    d0 = loc.replace(hour=0, minute=0, second=0, microsecond=0)
-    wh = {(d0 + timedelta(hours=h)).astimezone(timezone.utc).isoformat():
-          (kwh * 100.0 if 8 <= h < 18 else 0.0) for h in range(24)}
-    left = sum(v for k, v in wh.items()
-               if datetime.fromisoformat(k) >= now) / 1000.0
-    return {"state": f"{left:.3f}",
-            "attributes": {"unit_of_measurement": "kWh", "wh_period": wh},
-            "last_updated": now.isoformat(), "last_reported": now.isoformat()}
-
-
-rp.Hass.state = _sensor
-rp_start = [UTC0]
-FC_OPTS = {"pv_forecast_entity": "sensor.energy_production_today_remaining",
-           "cycle_cost_threshold_eur": 0.20}
-# TARIFF-24 round 2: with a forecast read, a price mode needs the cycle cost
-# itself (0.20 here), no longer 1.5 times it (0.30). These scenarios were
-# written for a 0.30 gap, and a midday 0.21 gap now plans a grid top-up they
-# are not about, so they ask for 0.30 with the forecast; the no-forecast leg
-# (`rows_nf`) keeps 0.20, which is 0.30 widened: the same gate on both sides.
-GATE_030 = {"cycle_cost_threshold_eur": 0.30}
-
-
-def seed_profile(c):
-    if c.house_profile is not None:
-        c.house_profile = profile(day_w=600.0, evening=3000.0)
-
-
-def house_flat(w):
-    return [(UTC0.timestamp() - 10, float(w))]
-
-
-def run_day(choice, *, fc, soc0, hours=21.0, start_h=0.75, house_w=800.0,
-            refuse=(), extra=None, patch=seed_profile):
-    FC_KWH[0] = fc
-    start = UTC0 + timedelta(hours=start_h)
-    rp_start[0] = start
-    rows = []
-
-    def cb(c, h, now, pr):
-        sh = dict(c.allocation.shares) if c.allocation is not None and \
-            c.allocation.shares else {}
-        rows.append(dict(
-            now=now, grid=h.grid(), shares=sh,
-            action=getattr(c.arb, "action", None),
-            roles=dict(c.mode_tick.roles) if c.mode_tick else {},
-            soc={u: p.soc for u, p in h.plants.items()},
-            targets=c.plan_targets(), plan=c.day_plan, runs=c.mode_runs))
-    st_, c = rp.run("p3_ls", house_flat(house_w), WPTS, start, 3600 * hours,
-                    check=cb, soc0=soc0, choice=choice, refuse=refuse,
-                    extra=dict(FC_OPTS, **(extra or {})), patch=patch)
-    return rows, st_, c
-
-
-SF_LS = {"a": (True, "solar_first", False), "b": (True, "leftover_solar",
-                                                  False)}
-SF_CH = {"a": (True, "solar_first", False), "b": (True, "cheap_hours",
-                                                  False)}
-
-
-def invariants(tag, rows, c):
-    charge = [r for r in rows if r["action"] == arbitrage.CHARGE]
-    mixed = [r for r in rows if any(v > 0 for v in r["shares"].values())
-             and any(v < 0 for v in r["shares"].values())]
-    check(f"{tag}: no opposite-direction split on any cycle", not mixed)
-    check(f"{tag}: Solar first never charges from the grid (A's share >= 0 "
-          "while the site charges)",
-          all(r["shares"].get(A, 0) >= 0 for r in charge))
-    q = {int(r["now"].timestamp() // 900) for r in charge}
-    banked = [b.offtake_avg_w for b in c.p1.blocks
-              if int(b.start.timestamp() // 900) in q]
-    check(f"{tag}: every banked quarter with a grid charge <= cap - margin "
-          f"(max {max(banked, default=0):.1f} W, aim {AIM})",
-          not banked or max(banked) <= AIM + 0.5)
-    return charge
-
-
-# Sunny: 40 kWh forecast. B (Leftover solar) at 30 % needs nothing.
-rows, st_, c = run_day(SF_LS, fc=40.0, soc0={"a": 60.0, "b": 30.0},
-                       extra=GATE_030)
-charge = invariants("sunny", rows, c)
-_last = nth(rows, -1, "a cycle")
-print(f"    sunny: verdict {present(_last) and _last['plan'] and _last['plan'].verdict}, "
-      f"{len(charge)} charge cycles")
-check("sunny: a day plan is made, and says sunny",
-      any(r["plan"] is not None and r["plan"].verdict == dayplan.SUNNY
-          for r in rows))
-check("sunny: no grid charge at all", not charge)
-
-# TARIFF-24 round 2: the same day at the cycle cost 0.20 itself. With the
-# forecast read the gap needed is 0.20 (not 1.5 x = 0.30), so the midday
-# quarter 0.339 against the evening 0.551 (gap 0.21, between 1x and 1.5x) is
-# a planned top-up for B (Leftover solar) up to its planned level.
-rows_g, st_, c_g = run_day(SF_LS, fc=40.0, soc0={"a": 60.0, "b": 30.0})
-charge_g = invariants("gap between 1x and 1.5x", rows_g, c_g)
-_first_g = nth(charge_g, 0, "a charge cycle")
-_gaps = [min(g.mean_eur_kwh for g in r["runs"].give)
-         - r["runs"].cheap.mean_eur_kwh for r in charge_g
-         if r["runs"] is not None and r["runs"].give]
-print(f"    gap 0.20 <= gap < 0.30: {len(charge_g)} charge cycles, run gap "
-      f"{min(_gaps, default=0):.3f}, B to "
-      f"{max((r['soc']['b'] for r in rows_g), default=0):.1f} %")
-check("a forecast day whose price gap lies between 1x and 1.5x the cycle "
-      "cost (0.20 < 0.21 < 0.30) plans a cheap run: B charges from the grid "
-      "inside it, only B, and only up to its planned level",
-      present(_first_g) and len(_gaps) == len(charge_g)
-      and all(0.20 <= g < 0.30 for g in _gaps)
-      and all(r["runs"].cheap.holds(r["now"]) for r in charge_g)
-      and all(r["roles"][B] == modetrade.CHARGE for r in charge_g)
-      and all(r["roles"][A] != modetrade.CHARGE for r in charge_g)
-      and max(r["soc"]["b"] for r in rows_g) <= 45.64 + 2.0)
-check("...and the same day with no forecast read does not (it needs 0.30): "
-      "the old gate, kept for a day without a forecast",
-      not [r for r in run_day(SF_LS, fc=None,
-                              soc0={"a": 60.0, "b": 30.0})[0]
-           if r["action"] == arbitrage.CHARGE
-           and r["now"].astimezone(L).hour == 12])
-
-# Dark: 1 kWh. B buys the evening's shortfall only, in the cheap run.
-rows, st_, c = run_day(SF_LS, fc=1.0, soc0={"a": 60.0, "b": 30.0})
-charge = invariants("dark", rows, c)
-tgt = nth([r["targets"][B] for r in rows if r["targets"]], 0, "a level")
-top = max(r["soc"]["b"] for r in rows)
-print(f"    dark: B level {tgt:.1f} %, B reached {top:.1f} %, "
-      f"{len(charge)} charge cycles")
-check("dark: Leftover solar charges from the grid in the cheap run",
-      charge and all(r["runs"].cheap.holds(r["now"]) for r in charge)
-      and all(r["roles"][B] == modetrade.CHARGE for r in charge))
-check("dark: ...only up to its level (the shortfall), not 95 %",
-      present(tgt) and tgt < 60.0 and top <= tgt + 2.0)
-check("dark: A rests while B charges", all(r["roles"][A] == modetrade.REST
-                                           for r in charge))
-
-# Cheap hours on a partly sunny day: ignores the plan, fills as in P3.
-rows, st_, c = run_day(SF_CH, fc=14.0, soc0={"a": 60.0, "b": 30.0},
-                       extra=GATE_030)
-charge = invariants("cheap hours, partly", rows, c)
-top = max(r["soc"]["b"] for r in rows)
-print(f"    cheap hours partly: B reached {top:.1f} %")
-check("Cheap hours gets no planned level",
-      not any(r["targets"] and B in r["targets"] for r in rows))
-rows_nf, _s, c_nf = run_day(SF_CH, fc=None, soc0={"a": 60.0, "b": 30.0})
-check("Cheap hours with a sun forecast charges as with none (P3)",
-      charge and abs(max(r["soc"]["b"] for r in rows_nf) - top) < 1.0
-      and top > 80.0)
-
-# The forecast goes away mid-run: a charge under way goes on to its level
-# (OPT-34; it stopped at once, P3 option A, before). With no charge under way
-# the levels go with the forecast (below, and test_opt27.py).
-# OPT-27: a sensor that is unavailable is covered by its last fresh reading
-# for up to 3 h (`solarfc.Hold`), so "lost" here is a hold that never covers;
-# the next scenario is the same loss with the hold (test_opt27.py).
-DROP = UTC0 + timedelta(hours=1, minutes=30)
-
-
-class _NoHold(solarfc.Hold):
-    def stand_in(self, *a, **k):
-        return None
-
-
-def seed_profile_no_hold(c):
-    seed_profile(c)
-    c.fc_hold = _NoHold()
-
-
-rows, st_, c = run_day(SF_LS, fc=lambda now: 1.0 if now < DROP else None,
-                       soc0={"a": 60.0, "b": 10.0}, patch=seed_profile_no_hold)
-invariants("forecast lost mid-run", rows, c)
-before = [r for r in rows if r["now"] < DROP and r["action"]
-          == arbitrage.CHARGE]
-late = [r for r in rows if r["now"] > DROP + timedelta(
-    seconds=2 * 60 + 10) and r["action"] == arbitrage.CHARGE]
-check("forecast lost mid-run (OPT-34): Leftover solar was charging, and goes "
-      "on charging past a forecast read (60 s) and a cycle, to its level",
-      before and late)
-rows, st_, c = run_day(SF_LS, fc=lambda now: 1.0 if now < DROP else None,
-                       soc0={"a": 60.0, "b": 10.0})
-invariants("forecast unavailable mid-run, covered", rows, c)
-check("forecast unavailable mid-run (OPT-27): the last reading covers it, so "
-      "Leftover solar goes on charging past the point where it was lost",
-      [r for r in rows if r["now"] < DROP and r["action"]
-       == arbitrage.CHARGE]
-      and [r for r in rows if r["now"] > DROP + timedelta(
-          seconds=2 * 60 + 10) and r["action"] == arbitrage.CHARGE])
-
-# The forecast is far too high (40 kWh, and there is no sun at all): no grid
-# charge, so B covers less of the evening - a money risk, never a safety one.
-rows, st_, c = run_day(SF_LS, fc=40.0, soc0={"a": 30.0, "b": 30.0},
-                       extra=GATE_030)
-invariants("forecast too high", rows, c)
-check("forecast too high: no grid charge, and B still keeps its power for "
-      "the dear run", not [r for r in rows if r["action"]
-                           == arbitrage.CHARGE]
-      and all(r["roles"].get(B) != modetrade.GIVE for r in rows
-              if r["now"].astimezone(L).hour not in (19,)))
-
-# A refusal wins over the plan.
-rows, st_, c = run_day(SF_LS, fc=1.0, soc0={"a": 60.0, "b": 30.0},
-                       refuse=("b",), extra=GATE_030)
-invariants("refusal", rows, c)
-check("refusal: B refuses to charge, so the plan's charge ends and B is "
-      "never a charger after it", rows and not [
-          r for r in rows[len(rows) // 3:]
-          if r["roles"].get(B) == modetrade.CHARGE])
-
-# Off wins over the plan.
-rows, st_, c = run_day({"a": (True, "solar_first", False),
-                        "b": (False, "leftover_solar", False)},
-                       fc=1.0, soc0={"a": 60.0, "b": 30.0})
-check("Off: an Off battery never charges from the grid, whatever the plan",
-      not [r for r in rows if r["action"] == arbitrage.CHARGE]
-      and all(abs(r["shares"].get(B, 0)) < 1e-6 for r in rows))
-
-# No house profile yet (first day): no plan, P3 option A.
-rows, st_, c = run_day(SF_LS, fc=1.0, soc0={"a": 60.0, "b": 30.0},
-                       patch=None)
-check("no house profile yet: no plan, and Leftover solar takes solar only",
-      all(r["plan"] is None for r in rows)
-      and not [r for r in rows if r["action"] == arbitrage.CHARGE])
-check("...while the profile learns and the store keeps each closed hour",
-      c.house_profile is not None and len(
-          c.store.data.get("house_profile") or {}) >= 1)
-check("selling stays off (option A)", packmode.SELLING_ENABLED is False)
-
-# A forecast arriving mid-period makes the plan at once, not at the next
-# 15-minute period.
-COME = datetime(2026, 9, 21, 1, 3, tzinfo=L).astimezone(timezone.utc)
-rows, st_, c = run_day(SF_LS, fc=lambda now: None if now < COME else 1.0,
-                       soc0={"a": 60.0, "b": 30.0}, hours=0.5)
-first = next((r["now"] for r in rows if r["targets"]), None)
-check("a forecast that arrives at 01:03 makes the plan before 01:15",
-      first is not None and first < COME + timedelta(minutes=3))
-
-# Battery modes off: the forecast is never read.
-READS = [0]
-_real_sensor = rp.Hass.state
-
-
-async def _counted(self, e):
-    READS[0] += 1
-    return await _real_sensor(self, e)
-rp.Hass.state = _counted
-FC_KWH[0] = 1.0
-rp.run("modes_off", house_flat(800.0), WPTS, UTC0 + timedelta(hours=1),
-       600.0, extra=dict(FC_OPTS))
-rp.Hass.state = _real_sensor
-check("battery modes off: the forecast sensor is never read", READS[0] == 0)
-
-# ============================================================================ #
-print("7. the Batteries screen and the Simple view")
-rows, st_, c = run_day(SF_LS, fc=1.0, soc0={"a": 60.0, "b": 30.0},
-                       hours=0.6)
-main_files.live.datetime = datetime
-sb = homeview.solar_block(c, c.solar_fc.at)
-check("the forecast line names the forecast, what is counted and when",
-      sb["line"].startswith("Solar still to come today:")
-      and "we count on" in sb["line"] and sb["state"] == "plan")
-why = homeview.day_key(c, B)
-check("B's Why key (plan, dark) is tops_up", why == "tops_up")
-check("...and says so in plain words, with the level and what it covers "
-      "(OPT-28)", homeview.why_words(why, pct="82 %", kwh="9.4 kWh")
-      == "It charges from the grid to 82 %. That covers 9.4 kWh of "
-         "expensive hours the sun will not.")
-_d0, _d1 = homeview.local_day(c.solar_fc.at)
-_blocks, _ = homeview.mode_runs(c, _d0, _d1, c.solar_fc.at, False)
-_grid = [b for b in _blocks if b["kind"] == "grid" and b["battery"] == 1]
-_need = homeview.plan_need(c, B)
-check(f"Today: B's grid block is the planned shortfall ({_need:.2f} kWh), "
-      "not its room to 95 %", _grid and abs(_grid[0]["kwh"] - _need) < 0.01
-      and _need < 3.0)
-_run = c.mode_runs.cheap
-_packs = c._mode_packs()
-_m1 = c._run_mark("charge", _run, _packs, lambda p: True, ((B, 40.0),))
-_m2 = c._run_mark("charge", _run, _packs, lambda p: False, ((B, 50.0),))
-check("a level the hysteresis moved opens the run to that battery again",
-      _m1 and not _m2)
-latch, c._charge_latch = c._charge_latch, None
-c.solar_fc = None
-check("with no charge under way, the levels are dropped as soon as the "
-      "forecast goes, even before the plan is rebuilt",
-      c._plan_targets and c.plan_targets() is None)
-check("a lost forecast says so: no forecast, Leftover solar solar only",
-      homeview.solar_block(c, datetime.now(timezone.utc))["state"]
-      == "no_forecast" and homeview.day_key(c, B) == "no_forecast")
-c._charge_latch = latch
-# the latch ends at its end time; this replay ran in September, so ask at a
-# moment inside it
-_real_dt = main_files.live.datetime
-main_files.live.datetime = type("_DT", (datetime,), {"now": classmethod(
-    lambda cls, tz=None: latch[0] - timedelta(hours=1))})
-check("OPT-34: with a charge under way (B charged in this run), the level "
-      "stays: plan_targets() gives it (the line: test_opt34.py)",
-      latch is not None and c.plan_targets() is not None
-      and B in c.plan_targets() and c.plan_targets()[B] == c._plan_targets[B])
-main_files.live.datetime = _real_dt
-check("the Why line wins over 'saves its power' only while holding",
-      homeview.battery_why(modes_on=True, mode="leftover_solar", off=False,
-                           reading=True, phase="idle", shadow=False,
-                           running=True, held=False, refused=0, peak=False,
-                           sf_ahead=True, role="hold", has_at=True,
-                           day="sun_fills") == "sun_fills"
-      and homeview.battery_why(modes_on=True, mode="leftover_solar",
-                               off=False, reading=True, phase="idle",
-                               shadow=False, running=True, held=False,
-                               refused=0, peak=False, sf_ahead=True,
-                               role="charge", has_at=True, day="tops_up")
-      == "grid_charge")
-for lang in ("en", "nl"):
-    homeview._LANG[0] = lang
-    t = [homeview.why_words(k, pct="82 %", kwh="9.4 kWh")
-         for k in ("sun_fills", "tops_up", "no_forecast")]
-    t += [homeview._t("solar.no_forecast"), homeview._t("solar.learning"),
-          homeview.line_solar(9.2, 6.4, "10:15")]
-    check(f"{lang}: every P4 sentence is there and short (<= 2 x 12 words)",
-          all(t) and all(len(s.split()) <= 24 for s in t))
-homeview._LANG[0] = "en"
-from app import uilayout                                        # noqa: E402
-simple = [f.key for _g, _p, f in uilayout.fields() if f.simple]
-check("the Simple view has the solar forecast card",
-      "pv_forecast_entity" in simple)

 # ============================================================================ #
 print("8. options")
@@ -0,0 +1,225 @@
+"""Runnable check for OPT-11 P4, part 6a: the sunny day, the 0.20 gap between 1x and 1.5x, the dark day.
+`python3 test_modes_p4b.py`
+
+Split out of test_modes_p4.py by CI-18 so the slow part 6 runs in pieces (CI
+runs files side by side; the one file took ~13 min). The docstring there describes
+the whole. Same fake Home Assistant, same engine (tools/opt11p3_replay.py), same
+checks, none dropped: the four files print exactly the check lines the one file did.
+Read MUTATION.md before you report a survivor."""
+
+import asyncio
+import logging
+import os
+import random
+import sys
+from datetime import datetime, timedelta, timezone
+
+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)
+
+from named_absence import nth, present                         # noqa: E402
+import main_files                                              # noqa: E402
+import opt11p3_replay as rp                                    # noqa: E402
+from app import arbitrage, dayplan, homeview, modetrade, packmode  # noqa
+from app import priceruns, solarfc, validate                    # noqa: E402
+from app.tariff import BILLING_TZ                               # noqa: E402
+
+fails = []
+
+
+def check(name, cond):
+    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
+    if not cond:
+        fails.append(name)
+
+
+L = timezone(timedelta(hours=2))
+A, B = "inverter-a", "inverter-b"
+D0 = datetime(2026, 9, 21, tzinfo=L)
+UTC0 = D0.astimezone(timezone.utc)
+
+# ============================================================================ #
+def profile(day_w=None, days=1, evening=None, start=UTC0 - timedelta(days=1)):
+    hp = dayplan.HouseProfile()
+    t = start
+    end = start + timedelta(days=days)
+    while t < end:
+        h = t.astimezone(L).hour
+        w = (evening if evening is not None and 17 <= h < 22 else
+             day_w if day_w is not None else 600.0)
+        hp.add(t, w)
+        t += timedelta(seconds=5)
+    hp.add(t, None)
+    return hp
+
+P = modetrade.Pack
+
+def pk(name, mode, soc, cap=10.0, **kw):
+    return P(name, True, mode, False, 2500.0, soc, cap, **kw)
+
+
+# Prices: cheap 01:00-03:00, dear 19:00-20:00 local (as in test_modes_p3).
+DAYW = [0.20] * 96 * 2
+for i in list(range(4, 12)) + list(range(100, 108)):
+    DAYW[i] = 0.02
+for i in list(range(76, 80)) + list(range(172, 176)):
+    DAYW[i] = 0.40
+WPTS = [(UTC0 + timedelta(minutes=15 * i), 900, p) for i, p in enumerate(DAYW)]
+from app.price import PricePoint                                # noqa: E402
+
+# ============================================================================ #
+print("6. the real controller over a night and a day")
+AIM = rp.PEAK_CAP - 50
+FC_KWH = [None]             # today's forecast in the fake sensor (None: down)
+
+
+async def _sensor(self, e):
+    now = rp_start[0] + timedelta(seconds=self.t)
+    kwh = FC_KWH[0](now) if callable(FC_KWH[0]) else FC_KWH[0]
+    if kwh is None:
+        return {"state": "unavailable", "attributes": {},
+                "last_updated": now.isoformat()}
+    if e.endswith("tomorrow"):
+        return None
+    loc = now.astimezone(L)
+    d0 = loc.replace(hour=0, minute=0, second=0, microsecond=0)
+    wh = {(d0 + timedelta(hours=h)).astimezone(timezone.utc).isoformat():
+          (kwh * 100.0 if 8 <= h < 18 else 0.0) for h in range(24)}
+    left = sum(v for k, v in wh.items()
+               if datetime.fromisoformat(k) >= now) / 1000.0
+    return {"state": f"{left:.3f}",
+            "attributes": {"unit_of_measurement": "kWh", "wh_period": wh},
+            "last_updated": now.isoformat(), "last_reported": now.isoformat()}
+
+
+rp.Hass.state = _sensor
+rp_start = [UTC0]
+FC_OPTS = {"pv_forecast_entity": "sensor.energy_production_today_remaining",
+           "cycle_cost_threshold_eur": 0.20}
+# TARIFF-24 round 2: with a forecast read, a price mode needs the cycle cost
+# itself (0.20 here), no longer 1.5 times it (0.30). These scenarios were
+# written for a 0.30 gap, and a midday 0.21 gap now plans a grid top-up they
+# are not about, so they ask for 0.30 with the forecast; the no-forecast leg
+# (`rows_nf`) keeps 0.20, which is 0.30 widened: the same gate on both sides.
+GATE_030 = {"cycle_cost_threshold_eur": 0.30}
+
+
+def seed_profile(c):
+    if c.house_profile is not None:
+        c.house_profile = profile(day_w=600.0, evening=3000.0)
+
+
+def house_flat(w):
+    return [(UTC0.timestamp() - 10, float(w))]
+
+
+def run_day(choice, *, fc, soc0, hours=21.0, start_h=0.75, house_w=800.0,
+            refuse=(), extra=None, patch=seed_profile):
+    FC_KWH[0] = fc
+    start = UTC0 + timedelta(hours=start_h)
+    rp_start[0] = start
+    rows = []
+
+    def cb(c, h, now, pr):
+        sh = dict(c.allocation.shares) if c.allocation is not None and \
+            c.allocation.shares else {}
+        rows.append(dict(
+            now=now, grid=h.grid(), shares=sh,
+            action=getattr(c.arb, "action", None),
+            roles=dict(c.mode_tick.roles) if c.mode_tick else {},
+            soc={u: p.soc for u, p in h.plants.items()},
+            targets=c.plan_targets(), plan=c.day_plan, runs=c.mode_runs))
+    st_, c = rp.run("p3_ls", house_flat(house_w), WPTS, start, 3600 * hours,
+                    check=cb, soc0=soc0, choice=choice, refuse=refuse,
+                    extra=dict(FC_OPTS, **(extra or {})), patch=patch)
+    return rows, st_, c
+
+
+SF_LS = {"a": (True, "solar_first", False), "b": (True, "leftover_solar",
+                                                  False)}
+SF_CH = {"a": (True, "solar_first", False), "b": (True, "cheap_hours",
+                                                  False)}
+
+
+def invariants(tag, rows, c):
+    charge = [r for r in rows if r["action"] == arbitrage.CHARGE]
+    mixed = [r for r in rows if any(v > 0 for v in r["shares"].values())
+             and any(v < 0 for v in r["shares"].values())]
+    check(f"{tag}: no opposite-direction split on any cycle", not mixed)
+    check(f"{tag}: Solar first never charges from the grid (A's share >= 0 "
+          "while the site charges)",
+          all(r["shares"].get(A, 0) >= 0 for r in charge))
+    q = {int(r["now"].timestamp() // 900) for r in charge}
+    banked = [b.offtake_avg_w for b in c.p1.blocks
+              if int(b.start.timestamp() // 900) in q]
+    check(f"{tag}: every banked quarter with a grid charge <= cap - margin "
+          f"(max {max(banked, default=0):.1f} W, aim {AIM})",
+          not banked or max(banked) <= AIM + 0.5)
+    return charge
+
+
+# Sunny: 40 kWh forecast. B (Leftover solar) at 30 % needs nothing.
+rows, st_, c = run_day(SF_LS, fc=40.0, soc0={"a": 60.0, "b": 30.0},
+                       extra=GATE_030)
+charge = invariants("sunny", rows, c)
+_last = nth(rows, -1, "a cycle")
+print(f"    sunny: verdict {present(_last) and _last['plan'] and _last['plan'].verdict}, "
+      f"{len(charge)} charge cycles")
+check("sunny: a day plan is made, and says sunny",
+      any(r["plan"] is not None and r["plan"].verdict == dayplan.SUNNY
+          for r in rows))
+check("sunny: no grid charge at all", not charge)
+
+# TARIFF-24 round 2: the same day at the cycle cost 0.20 itself. With the
+# forecast read the gap needed is 0.20 (not 1.5 x = 0.30), so the midday
+# quarter 0.339 against the evening 0.551 (gap 0.21, between 1x and 1.5x) is
+# a planned top-up for B (Leftover solar) up to its planned level.
+rows_g, st_, c_g = run_day(SF_LS, fc=40.0, soc0={"a": 60.0, "b": 30.0})
+charge_g = invariants("gap between 1x and 1.5x", rows_g, c_g)
+_first_g = nth(charge_g, 0, "a charge cycle")
+_gaps = [min(g.mean_eur_kwh for g in r["runs"].give)
+         - r["runs"].cheap.mean_eur_kwh for r in charge_g
+         if r["runs"] is not None and r["runs"].give]
+print(f"    gap 0.20 <= gap < 0.30: {len(charge_g)} charge cycles, run gap "
+      f"{min(_gaps, default=0):.3f}, B to "
+      f"{max((r['soc']['b'] for r in rows_g), default=0):.1f} %")
+check("a forecast day whose price gap lies between 1x and 1.5x the cycle "
+      "cost (0.20 < 0.21 < 0.30) plans a cheap run: B charges from the grid "
+      "inside it, only B, and only up to its planned level",
+      present(_first_g) and len(_gaps) == len(charge_g)
+      and all(0.20 <= g < 0.30 for g in _gaps)
+      and all(r["runs"].cheap.holds(r["now"]) for r in charge_g)
+      and all(r["roles"][B] == modetrade.CHARGE for r in charge_g)
+      and all(r["roles"][A] != modetrade.CHARGE for r in charge_g)
+      and max(r["soc"]["b"] for r in rows_g) <= 45.64 + 2.0)
+check("...and the same day with no forecast read does not (it needs 0.30): "
+      "the old gate, kept for a day without a forecast",
+      not [r for r in run_day(SF_LS, fc=None,
+                              soc0={"a": 60.0, "b": 30.0})[0]
+           if r["action"] == arbitrage.CHARGE
+           and r["now"].astimezone(L).hour == 12])
+
+# Dark: 1 kWh. B buys the evening's shortfall only, in the cheap run.
+rows, st_, c = run_day(SF_LS, fc=1.0, soc0={"a": 60.0, "b": 30.0})
+charge = invariants("dark", rows, c)
+tgt = nth([r["targets"][B] for r in rows if r["targets"]], 0, "a level")
+top = max(r["soc"]["b"] for r in rows)
+print(f"    dark: B level {tgt:.1f} %, B reached {top:.1f} %, "
+      f"{len(charge)} charge cycles")
+check("dark: Leftover solar charges from the grid in the cheap run",
+      charge and all(r["runs"].cheap.holds(r["now"]) for r in charge)
+      and all(r["roles"][B] == modetrade.CHARGE for r in charge))
+check("dark: ...only up to its level (the shortfall), not 95 %",
+      present(tgt) and tgt < 60.0 and top <= tgt + 2.0)
+check("dark: A rests while B charges", all(r["roles"][A] == modetrade.REST
+                                           for r in charge))
+
+print()
+if fails:
+    print(f"FAILED: {len(fails)}")
+    for f in fails:
+        print("  -", f)
+    sys.exit(1)
+print("all checks passed")
@@ -0,0 +1,232 @@
+"""Runnable check for OPT-11 P4, part 6b: cheap hours on a partly sunny day, the forecast lost mid-run, a forecast far too high.
+`python3 test_modes_p4c.py`
+
+Split out of test_modes_p4.py by CI-18 so the slow part 6 runs in pieces (CI
+runs files side by side; the one file took ~13 min). The docstring there describes
+the whole. Same fake Home Assistant, same engine (tools/opt11p3_replay.py), same
+checks, none dropped: the four files print exactly the check lines the one file did.
+Read MUTATION.md before you report a survivor."""
+
+import asyncio
+import logging
+import os
+import random
+import sys
+from datetime import datetime, timedelta, timezone
+
+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)
+
+from named_absence import nth, present                         # noqa: E402
+import main_files                                              # noqa: E402
+import opt11p3_replay as rp                                    # noqa: E402
+from app import arbitrage, dayplan, homeview, modetrade, packmode  # noqa
+from app import priceruns, solarfc, validate                    # noqa: E402
+from app.tariff import BILLING_TZ                               # noqa: E402
+
+fails = []
+
+
+def check(name, cond):
+    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
+    if not cond:
+        fails.append(name)
+
+
+L = timezone(timedelta(hours=2))
+A, B = "inverter-a", "inverter-b"
+D0 = datetime(2026, 9, 21, tzinfo=L)
+UTC0 = D0.astimezone(timezone.utc)
+
+# ============================================================================ #
+def profile(day_w=None, days=1, evening=None, start=UTC0 - timedelta(days=1)):
+    hp = dayplan.HouseProfile()
+    t = start
+    end = start + timedelta(days=days)
+    while t < end:
+        h = t.astimezone(L).hour
+        w = (evening if evening is not None and 17 <= h < 22 else
+             day_w if day_w is not None else 600.0)
+        hp.add(t, w)
+        t += timedelta(seconds=5)
+    hp.add(t, None)
+    return hp
+
+P = modetrade.Pack
+
+def pk(name, mode, soc, cap=10.0, **kw):
+    return P(name, True, mode, False, 2500.0, soc, cap, **kw)
+
+
+# Prices: cheap 01:00-03:00, dear 19:00-20:00 local (as in test_modes_p3).
+DAYW = [0.20] * 96 * 2
+for i in list(range(4, 12)) + list(range(100, 108)):
+    DAYW[i] = 0.02
+for i in list(range(76, 80)) + list(range(172, 176)):
+    DAYW[i] = 0.40
+WPTS = [(UTC0 + timedelta(minutes=15 * i), 900, p) for i, p in enumerate(DAYW)]
+from app.price import PricePoint                                # noqa: E402
+
+# ============================================================================ #
+print("6. the real controller over a night and a day")
+AIM = rp.PEAK_CAP - 50
+FC_KWH = [None]             # today's forecast in the fake sensor (None: down)
+
+
+async def _sensor(self, e):
+    now = rp_start[0] + timedelta(seconds=self.t)
+    kwh = FC_KWH[0](now) if callable(FC_KWH[0]) else FC_KWH[0]
+    if kwh is None:
+        return {"state": "unavailable", "attributes": {},
+                "last_updated": now.isoformat()}
+    if e.endswith("tomorrow"):
+        return None
+    loc = now.astimezone(L)
+    d0 = loc.replace(hour=0, minute=0, second=0, microsecond=0)
+    wh = {(d0 + timedelta(hours=h)).astimezone(timezone.utc).isoformat():
+          (kwh * 100.0 if 8 <= h < 18 else 0.0) for h in range(24)}
+    left = sum(v for k, v in wh.items()
+               if datetime.fromisoformat(k) >= now) / 1000.0
+    return {"state": f"{left:.3f}",
+            "attributes": {"unit_of_measurement": "kWh", "wh_period": wh},
+            "last_updated": now.isoformat(), "last_reported": now.isoformat()}
+
+
+rp.Hass.state = _sensor
+rp_start = [UTC0]
+FC_OPTS = {"pv_forecast_entity": "sensor.energy_production_today_remaining",
+           "cycle_cost_threshold_eur": 0.20}
+# TARIFF-24 round 2: with a forecast read, a price mode needs the cycle cost
+# itself (0.20 here), no longer 1.5 times it (0.30). These scenarios were
+# written for a 0.30 gap, and a midday 0.21 gap now plans a grid top-up they
+# are not about, so they ask for 0.30 with the forecast; the no-forecast leg
+# (`rows_nf`) keeps 0.20, which is 0.30 widened: the same gate on both sides.
+GATE_030 = {"cycle_cost_threshold_eur": 0.30}
+
+
+def seed_profile(c):
+    if c.house_profile is not None:
+        c.house_profile = profile(day_w=600.0, evening=3000.0)
+
+
+def house_flat(w):
+    return [(UTC0.timestamp() - 10, float(w))]
+
+
+def run_day(choice, *, fc, soc0, hours=21.0, start_h=0.75, house_w=800.0,
+            refuse=(), extra=None, patch=seed_profile):
+    FC_KWH[0] = fc
+    start = UTC0 + timedelta(hours=start_h)
+    rp_start[0] = start
+    rows = []
+
+    def cb(c, h, now, pr):
+        sh = dict(c.allocation.shares) if c.allocation is not None and \
+            c.allocation.shares else {}
+        rows.append(dict(
+            now=now, grid=h.grid(), shares=sh,
+            action=getattr(c.arb, "action", None),
+            roles=dict(c.mode_tick.roles) if c.mode_tick else {},
+            soc={u: p.soc for u, p in h.plants.items()},
+            targets=c.plan_targets(), plan=c.day_plan, runs=c.mode_runs))
+    st_, c = rp.run("p3_ls", house_flat(house_w), WPTS, start, 3600 * hours,
+                    check=cb, soc0=soc0, choice=choice, refuse=refuse,
+                    extra=dict(FC_OPTS, **(extra or {})), patch=patch)
+    return rows, st_, c
+
+
+SF_LS = {"a": (True, "solar_first", False), "b": (True, "leftover_solar",
+                                                  False)}
+SF_CH = {"a": (True, "solar_first", False), "b": (True, "cheap_hours",
+                                                  False)}
+
+
+def invariants(tag, rows, c):
+    charge = [r for r in rows if r["action"] == arbitrage.CHARGE]
+    mixed = [r for r in rows if any(v > 0 for v in r["shares"].values())
+             and any(v < 0 for v in r["shares"].values())]
+    check(f"{tag}: no opposite-direction split on any cycle", not mixed)
+    check(f"{tag}: Solar first never charges from the grid (A's share >= 0 "
+          "while the site charges)",
+          all(r["shares"].get(A, 0) >= 0 for r in charge))
+    q = {int(r["now"].timestamp() // 900) for r in charge}
+    banked = [b.offtake_avg_w for b in c.p1.blocks
+              if int(b.start.timestamp() // 900) in q]
+    check(f"{tag}: every banked quarter with a grid charge <= cap - margin "
+          f"(max {max(banked, default=0):.1f} W, aim {AIM})",
+          not banked or max(banked) <= AIM + 0.5)
+    return charge
+
+
+# Cheap hours on a partly sunny day: ignores the plan, fills as in P3.
+rows, st_, c = run_day(SF_CH, fc=14.0, soc0={"a": 60.0, "b": 30.0},
+                       extra=GATE_030)
+charge = invariants("cheap hours, partly", rows, c)
+top = max(r["soc"]["b"] for r in rows)
+print(f"    cheap hours partly: B reached {top:.1f} %")
+check("Cheap hours gets no planned level",
+      not any(r["targets"] and B in r["targets"] for r in rows))
+rows_nf, _s, c_nf = run_day(SF_CH, fc=None, soc0={"a": 60.0, "b": 30.0})
+check("Cheap hours with a sun forecast charges as with none (P3)",
+      charge and abs(max(r["soc"]["b"] for r in rows_nf) - top) < 1.0
+      and top > 80.0)
+
+# The forecast goes away mid-run: a charge under way goes on to its level
+# (OPT-34; it stopped at once, P3 option A, before). With no charge under way
+# the levels go with the forecast (below, and test_opt27.py).
+# OPT-27: a sensor that is unavailable is covered by its last fresh reading
+# for up to 3 h (`solarfc.Hold`), so "lost" here is a hold that never covers;
+# the next scenario is the same loss with the hold (test_opt27.py).
+DROP = UTC0 + timedelta(hours=1, minutes=30)
+
+
+class _NoHold(solarfc.Hold):
+    def stand_in(self, *a, **k):
+        return None
+
+
+def seed_profile_no_hold(c):
+    seed_profile(c)
+    c.fc_hold = _NoHold()
+
+
+rows, st_, c = run_day(SF_LS, fc=lambda now: 1.0 if now < DROP else None,
+                       soc0={"a": 60.0, "b": 10.0}, patch=seed_profile_no_hold)
+invariants("forecast lost mid-run", rows, c)
+before = [r for r in rows if r["now"] < DROP and r["action"]
+          == arbitrage.CHARGE]
+late = [r for r in rows if r["now"] > DROP + timedelta(
+    seconds=2 * 60 + 10) and r["action"] == arbitrage.CHARGE]
+check("forecast lost mid-run (OPT-34): Leftover solar was charging, and goes "
+      "on charging past a forecast read (60 s) and a cycle, to its level",
+      before and late)
+rows, st_, c = run_day(SF_LS, fc=lambda now: 1.0 if now < DROP else None,
+                       soc0={"a": 60.0, "b": 10.0})
+invariants("forecast unavailable mid-run, covered", rows, c)
+check("forecast unavailable mid-run (OPT-27): the last reading covers it, so "
+      "Leftover solar goes on charging past the point where it was lost",
+      [r for r in rows if r["now"] < DROP and r["action"]
+       == arbitrage.CHARGE]
+      and [r for r in rows if r["now"] > DROP + timedelta(
+          seconds=2 * 60 + 10) and r["action"] == arbitrage.CHARGE])
+
+# The forecast is far too high (40 kWh, and there is no sun at all): no grid
+# charge, so B covers less of the evening - a money risk, never a safety one.
+rows, st_, c = run_day(SF_LS, fc=40.0, soc0={"a": 30.0, "b": 30.0},
+                       extra=GATE_030)
+invariants("forecast too high", rows, c)
+check("forecast too high: no grid charge, and B still keeps its power for "
+      "the dear run", not [r for r in rows if r["action"]
+                           == arbitrage.CHARGE]
+      and all(r["roles"].get(B) != modetrade.GIVE for r in rows
+              if r["now"].astimezone(L).hour not in (19,)))
+
+print()
+if fails:
+    print(f"FAILED: {len(fails)}")
+    for f in fails:
+        print("  -", f)
+    sys.exit(1)
+print("all checks passed")
@@ -0,0 +1,294 @@
+"""Runnable check for OPT-11 P4, part 6c and 7: a refusal, Off, no house profile, a forecast arriving mid-period, battery modes off; then the Batteries screen and the Simple view.
+`python3 test_modes_p4d.py`
+
+Split out of test_modes_p4.py by CI-18 so the slow part 6 runs in pieces (CI
+runs files side by side; the one file took ~13 min). The docstring there describes
+the whole. Same fake Home Assistant, same engine (tools/opt11p3_replay.py), same
+checks, none dropped: the four files print exactly the check lines the one file did.
+Read MUTATION.md before you report a survivor."""
+
+import asyncio
+import logging
+import os
+import random
+import sys
+from datetime import datetime, timedelta, timezone
+
+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)
+
+from named_absence import nth, present                         # noqa: E402
+import main_files                                              # noqa: E402
+import opt11p3_replay as rp                                    # noqa: E402
+from app import arbitrage, dayplan, homeview, modetrade, packmode  # noqa
+from app import priceruns, solarfc, validate                    # noqa: E402
+from app.tariff import BILLING_TZ                               # noqa: E402
+
+fails = []
+
+
+def check(name, cond):
+    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
+    if not cond:
+        fails.append(name)
+
+
+L = timezone(timedelta(hours=2))
+A, B = "inverter-a", "inverter-b"
+D0 = datetime(2026, 9, 21, tzinfo=L)
+UTC0 = D0.astimezone(timezone.utc)
+
+# ============================================================================ #
+def profile(day_w=None, days=1, evening=None, start=UTC0 - timedelta(days=1)):
+    hp = dayplan.HouseProfile()
+    t = start
+    end = start + timedelta(days=days)
+    while t < end:
+        h = t.astimezone(L).hour
+        w = (evening if evening is not None and 17 <= h < 22 else
+             day_w if day_w is not None else 600.0)
+        hp.add(t, w)
+        t += timedelta(seconds=5)
+    hp.add(t, None)
+    return hp
+
+P = modetrade.Pack
+
+def pk(name, mode, soc, cap=10.0, **kw):
+    return P(name, True, mode, False, 2500.0, soc, cap, **kw)
+
+
+# Prices: cheap 01:00-03:00, dear 19:00-20:00 local (as in test_modes_p3).
+DAYW = [0.20] * 96 * 2
+for i in list(range(4, 12)) + list(range(100, 108)):
+    DAYW[i] = 0.02
+for i in list(range(76, 80)) + list(range(172, 176)):
+    DAYW[i] = 0.40
+WPTS = [(UTC0 + timedelta(minutes=15 * i), 900, p) for i, p in enumerate(DAYW)]
+from app.price import PricePoint                                # noqa: E402
+
+# ============================================================================ #
+print("6. the real controller over a night and a day")
+AIM = rp.PEAK_CAP - 50
+FC_KWH = [None]             # today's forecast in the fake sensor (None: down)
+
+
+async def _sensor(self, e):
+    now = rp_start[0] + timedelta(seconds=self.t)
+    kwh = FC_KWH[0](now) if callable(FC_KWH[0]) else FC_KWH[0]
+    if kwh is None:
+        return {"state": "unavailable", "attributes": {},
+                "last_updated": now.isoformat()}
+    if e.endswith("tomorrow"):
+        return None
+    loc = now.astimezone(L)
+    d0 = loc.replace(hour=0, minute=0, second=0, microsecond=0)
+    wh = {(d0 + timedelta(hours=h)).astimezone(timezone.utc).isoformat():
+          (kwh * 100.0 if 8 <= h < 18 else 0.0) for h in range(24)}
+    left = sum(v for k, v in wh.items()
+               if datetime.fromisoformat(k) >= now) / 1000.0
+    return {"state": f"{left:.3f}",
+            "attributes": {"unit_of_measurement": "kWh", "wh_period": wh},
+            "last_updated": now.isoformat(), "last_reported": now.isoformat()}
+
+
+rp.Hass.state = _sensor
+rp_start = [UTC0]
+FC_OPTS = {"pv_forecast_entity": "sensor.energy_production_today_remaining",
+           "cycle_cost_threshold_eur": 0.20}
+# TARIFF-24 round 2: with a forecast read, a price mode needs the cycle cost
+# itself (0.20 here), no longer 1.5 times it (0.30). These scenarios were
+# written for a 0.30 gap, and a midday 0.21 gap now plans a grid top-up they
+# are not about, so they ask for 0.30 with the forecast; the no-forecast leg
+# (`rows_nf`) keeps 0.20, which is 0.30 widened: the same gate on both sides.
+GATE_030 = {"cycle_cost_threshold_eur": 0.30}
+
+
+def seed_profile(c):
+    if c.house_profile is not None:
+        c.house_profile = profile(day_w=600.0, evening=3000.0)
+
+
+def house_flat(w):
+    return [(UTC0.timestamp() - 10, float(w))]
+
+
+def run_day(choice, *, fc, soc0, hours=21.0, start_h=0.75, house_w=800.0,
+            refuse=(), extra=None, patch=seed_profile):
+    FC_KWH[0] = fc
+    start = UTC0 + timedelta(hours=start_h)
+    rp_start[0] = start
+    rows = []
+
+    def cb(c, h, now, pr):
+        sh = dict(c.allocation.shares) if c.allocation is not None and \
+            c.allocation.shares else {}
+        rows.append(dict(
+            now=now, grid=h.grid(), shares=sh,
+            action=getattr(c.arb, "action", None),
+            roles=dict(c.mode_tick.roles) if c.mode_tick else {},
+            soc={u: p.soc for u, p in h.plants.items()},
+            targets=c.plan_targets(), plan=c.day_plan, runs=c.mode_runs))
+    st_, c = rp.run("p3_ls", house_flat(house_w), WPTS, start, 3600 * hours,
+                    check=cb, soc0=soc0, choice=choice, refuse=refuse,
+                    extra=dict(FC_OPTS, **(extra or {})), patch=patch)
+    return rows, st_, c
+
+
+SF_LS = {"a": (True, "solar_first", False), "b": (True, "leftover_solar",
+                                                  False)}
+SF_CH = {"a": (True, "solar_first", False), "b": (True, "cheap_hours",
+                                                  False)}
+
+
+def invariants(tag, rows, c):
+    charge = [r for r in rows if r["action"] == arbitrage.CHARGE]
+    mixed = [r for r in rows if any(v > 0 for v in r["shares"].values())
+             and any(v < 0 for v in r["shares"].values())]
+    check(f"{tag}: no opposite-direction split on any cycle", not mixed)
+    check(f"{tag}: Solar first never charges from the grid (A's share >= 0 "
+          "while the site charges)",
+          all(r["shares"].get(A, 0) >= 0 for r in charge))
+    q = {int(r["now"].timestamp() // 900) for r in charge}
+    banked = [b.offtake_avg_w for b in c.p1.blocks
+              if int(b.start.timestamp() // 900) in q]
+    check(f"{tag}: every banked quarter with a grid charge <= cap - margin "
+          f"(max {max(banked, default=0):.1f} W, aim {AIM})",
+          not banked or max(banked) <= AIM + 0.5)
+    return charge
+
+
+# A refusal wins over the plan.
+rows, st_, c = run_day(SF_LS, fc=1.0, soc0={"a": 60.0, "b": 30.0},
+                       refuse=("b",), extra=GATE_030)
+invariants("refusal", rows, c)
+check("refusal: B refuses to charge, so the plan's charge ends and B is "
+      "never a charger after it", rows and not [
+          r for r in rows[len(rows) // 3:]
+          if r["roles"].get(B) == modetrade.CHARGE])
+
+# Off wins over the plan.
+rows, st_, c = run_day({"a": (True, "solar_first", False),
+                        "b": (False, "leftover_solar", False)},
+                       fc=1.0, soc0={"a": 60.0, "b": 30.0})
+check("Off: an Off battery never charges from the grid, whatever the plan",
+      not [r for r in rows if r["action"] == arbitrage.CHARGE]
+      and all(abs(r["shares"].get(B, 0)) < 1e-6 for r in rows))
+
+# No house profile yet (first day): no plan, P3 option A.
+rows, st_, c = run_day(SF_LS, fc=1.0, soc0={"a": 60.0, "b": 30.0},
+                       patch=None)
+check("no house profile yet: no plan, and Leftover solar takes solar only",
+      all(r["plan"] is None for r in rows)
+      and not [r for r in rows if r["action"] == arbitrage.CHARGE])
+check("...while the profile learns and the store keeps each closed hour",
+      c.house_profile is not None and len(
+          c.store.data.get("house_profile") or {}) >= 1)
+check("selling stays off (option A)", packmode.SELLING_ENABLED is False)
+
+# A forecast arriving mid-period makes the plan at once, not at the next
+# 15-minute period.
+COME = datetime(2026, 9, 21, 1, 3, tzinfo=L).astimezone(timezone.utc)
+rows, st_, c = run_day(SF_LS, fc=lambda now: None if now < COME else 1.0,
+                       soc0={"a": 60.0, "b": 30.0}, hours=0.5)
+first = next((r["now"] for r in rows if r["targets"]), None)
+check("a forecast that arrives at 01:03 makes the plan before 01:15",
+      first is not None and first < COME + timedelta(minutes=3))
+
+# Battery modes off: the forecast is never read.
+READS = [0]
+_real_sensor = rp.Hass.state
+
+
+async def _counted(self, e):
+    READS[0] += 1
+    return await _real_sensor(self, e)
+rp.Hass.state = _counted
+FC_KWH[0] = 1.0
+rp.run("modes_off", house_flat(800.0), WPTS, UTC0 + timedelta(hours=1),
+       600.0, extra=dict(FC_OPTS))
+rp.Hass.state = _real_sensor
+check("battery modes off: the forecast sensor is never read", READS[0] == 0)
+
+# ============================================================================ #
+print("7. the Batteries screen and the Simple view")
+rows, st_, c = run_day(SF_LS, fc=1.0, soc0={"a": 60.0, "b": 30.0},
+                       hours=0.6)
+main_files.live.datetime = datetime
+sb = homeview.solar_block(c, c.solar_fc.at)
+check("the forecast line names the forecast, what is counted and when",
+      sb["line"].startswith("Solar still to come today:")
+      and "we count on" in sb["line"] and sb["state"] == "plan")
+why = homeview.day_key(c, B)
+check("B's Why key (plan, dark) is tops_up", why == "tops_up")
+check("...and says so in plain words, with the level and what it covers "
+      "(OPT-28)", homeview.why_words(why, pct="82 %", kwh="9.4 kWh")
+      == "It charges from the grid to 82 %. That covers 9.4 kWh of "
+         "expensive hours the sun will not.")
+_d0, _d1 = homeview.local_day(c.solar_fc.at)
+_blocks, _ = homeview.mode_runs(c, _d0, _d1, c.solar_fc.at, False)
+_grid = [b for b in _blocks if b["kind"] == "grid" and b["battery"] == 1]
+_need = homeview.plan_need(c, B)
+check(f"Today: B's grid block is the planned shortfall ({_need:.2f} kWh), "
+      "not its room to 95 %", _grid and abs(_grid[0]["kwh"] - _need) < 0.01
+      and _need < 3.0)
+_run = c.mode_runs.cheap
+_packs = c._mode_packs()
+_m1 = c._run_mark("charge", _run, _packs, lambda p: True, ((B, 40.0),))
+_m2 = c._run_mark("charge", _run, _packs, lambda p: False, ((B, 50.0),))
+check("a level the hysteresis moved opens the run to that battery again",
+      _m1 and not _m2)
+latch, c._charge_latch = c._charge_latch, None
+c.solar_fc = None
+check("with no charge under way, the levels are dropped as soon as the "
+      "forecast goes, even before the plan is rebuilt",
+      c._plan_targets and c.plan_targets() is None)
+check("a lost forecast says so: no forecast, Leftover solar solar only",
+      homeview.solar_block(c, datetime.now(timezone.utc))["state"]
+      == "no_forecast" and homeview.day_key(c, B) == "no_forecast")
+c._charge_latch = latch
+# the latch ends at its end time; this replay ran in September, so ask at a
+# moment inside it
+_real_dt = main_files.live.datetime
+main_files.live.datetime = type("_DT", (datetime,), {"now": classmethod(
+    lambda cls, tz=None: latch[0] - timedelta(hours=1))})
+check("OPT-34: with a charge under way (B charged in this run), the level "
+      "stays: plan_targets() gives it (the line: test_opt34.py)",
+      latch is not None and c.plan_targets() is not None
+      and B in c.plan_targets() and c.plan_targets()[B] == c._plan_targets[B])
+main_files.live.datetime = _real_dt
+check("the Why line wins over 'saves its power' only while holding",
+      homeview.battery_why(modes_on=True, mode="leftover_solar", off=False,
+                           reading=True, phase="idle", shadow=False,
+                           running=True, held=False, refused=0, peak=False,
+                           sf_ahead=True, role="hold", has_at=True,
+                           day="sun_fills") == "sun_fills"
+      and homeview.battery_why(modes_on=True, mode="leftover_solar",
+                               off=False, reading=True, phase="idle",
+                               shadow=False, running=True, held=False,
+                               refused=0, peak=False, sf_ahead=True,
+                               role="charge", has_at=True, day="tops_up")
+      == "grid_charge")
+for lang in ("en", "nl"):
+    homeview._LANG[0] = lang
+    t = [homeview.why_words(k, pct="82 %", kwh="9.4 kWh")
+         for k in ("sun_fills", "tops_up", "no_forecast")]
+    t += [homeview._t("solar.no_forecast"), homeview._t("solar.learning"),
+          homeview.line_solar(9.2, 6.4, "10:15")]
+    check(f"{lang}: every P4 sentence is there and short (<= 2 x 12 words)",
+          all(t) and all(len(s.split()) <= 24 for s in t))
+homeview._LANG[0] = "en"
+from app import uilayout                                        # noqa: E402
+simple = [f.key for _g, _p, f in uilayout.fields() if f.simple]
+check("the Simple view has the solar forecast card",
+      "pv_forecast_entity" in simple)
+
+print()
+if fails:
+    print(f"FAILED: {len(fails)}")
+    for f in fails:
+        print("  -", f)
+    sys.exit(1)
+print("all checks passed")
@@ -30,7 +30,12 @@ live), inverter-b Solar first, a house of 1 kW, cap 3500 W, margin 50 W.
   quarter's last second.

 Mutations (MUTATION.md, purged copies) are listed in the pull request.
-"""
+
+CI-18: this file holds parts 1, 2 and 8 (part 8 grades part 2's runs). Parts 3-7 are in
+test_safety125b.py, split out so CI can run the pieces side by side (the one
+file took ~10 min). Every check is in exactly one file; the two print exactly the
+check lines the one file did."""
+

 import asyncio
 import logging
@@ -319,201 +324,7 @@ check("the forecast entity is read, in kW, as the live template (SAFETY-118), "
               "month peak 5413 W banked, P1 projection"][1]])


-# ============================================================================ #
-print("\n3. the same for a price trade (modes off) and a maintenance charge")
-_ca, _ra, _fa = drive(opts(modes=False, arb=True), peak=PEAK, late=30.0)[:3]
-check("modes off, price trading on, a reading 30 s late: every quarter at "
-      "or under the aim", graded("trade", _fa, AIM_PEAK)
-      and any(r[3] == "ARBITRAGE_CHARGE" for r in _ra))
-
-
-def late_feed(gap, every=0.0, look=None):
-    """safety84's cheap-window feed, no telegram in each quarter's last
-    `gap` s, and one only every `every` s (0: every cycle). `look(c, t)`
-    runs first, on the state the previous cycle left."""
-    feed = s84.feed(True)
-
-    def before(c, t):
-        if look is not None:
-            look(c, t)
-        if ((rig.BASE.timestamp() + t) % QUARTER_S < QUARTER_S - gap
-                and (not every or t % every < 1e-9)) \
-                or not getattr(c.hass, "_s84", False):
-            feed(c, t)
-    return before
-
-
-def rig_quarters(rows, load):
-    """The safety84 rig's whole quarters: the true average import (W)."""
-    qs = {}
-    for r in rows:
-        q = int((rig.BASE.timestamp() + r[0]) // QUARTER_S)
-        a = qs.setdefault(q, [0.0, 0])
-        a[0] += max(0.0, load(r[0]) - sum(v for _u, v in r[2]))
-        a[1] += 1
-    return {datetime.fromtimestamp(q * QUARTER_S, L).strftime("%H:%M"):
-            s / k for q, (s, k) in sorted(qs.items())
-            if k * rig.TICK >= QUARTER_S}
-
-
-for peak, house, aim in ((5413.0, 4000.0, 5363.0), (None, 2500.0, AIM_CAP)):
-    _mr, _mc = s84.run(2, load=lambda t, h=house: h, dur=2700.0,
-                       month_peak=peak, before=late_feed(30.0))
-    _mf = rig_quarters(_mr, lambda t, h=house: h)
-    check(f"N=2 maintenance charge, house {house:.0f} W, "
-          f"{'peak banked' if peak else 'no peak banked'}, a reading 30 s "
-          f"late: every quarter at or under the aim",
-          graded("maintenance", _mf, aim)
-          and any(mm.phase == "charge" for mm in _mc.maints))
-
-
-# ============================================================================ #
-print("\n4. a healthy meter: the quarters, and the words")
-for name, kw, aim, kw_text in (("month peak banked", dict(peak=PEAK),
-                                AIM_PEAK, "5.4 kW"),
-                               ("no peak banked", dict(), AIM_CAP, "3.5 kW")):
-    _c, _rows, _full, _s = drive(opts(), every=10, **kw)
-    check(f"{name}: every quarter at or under the aim",
-          graded(name, _full, aim))
-    over = [s for s in _s if not s[2] and s[0] > s[1]]
-    under = [s for s in _s if not s[2] and s[0] <= s[1]]
-    _o, _u = first(iter(over), "an over sample"), \
-        first(iter(under), "an under sample")
-    print(f"    over the limit at this rate ({len(over)} samples): "
-          f"{_o[3] if present(_o) else _o!s} / "
-          f"{_o[4] if present(_o) else _o!s}")
-    print(f"    under it ({len(under)}): "
-          f"{_u[3] if present(_u) else _u!s} / "
-          f"{_u[4] if present(_u) else _u!s}")
-    check(f"{name}: over the limit at this rate the guard row never says "
-          f"'under'; it says the grid charge is limited",
-          present(_o) and all(
-              "under" not in s[3] and "Peak guard: grid charge limited, "
-              "at this rate" in s[3] and " over " in s[3] for s in over))
-    check(f"{name}: ...and the month-peak line says the rate, and that the "
-          f"grid charge is cut back to stay under {kw_text}",
-          present(_o) and all(
-              s[4].startswith("At this rate this 15-minute period reaches")
-              and s[4].endswith("The grid charge is cut back to stay "
-                                f"under {kw_text}.") for s in over))
-    check(f"{name}: under the limit both read as before",
-          present(_u) and all(
-              s[3].startswith("Peak guard: idle, heading for")
-              and " under " in s[3]
-              and s[4].startswith("This 15-minute period is heading for")
-              and s[4].endswith(f"The batteries step in above {kw_text}.")
-              for s in under))
-
-
-# ============================================================================ #
-print("\n5. the words on their own")
-_en = messages.row_peak_guard_idle(5814.0, 5413.0, month_peak=True)
-_cut = messages.row_peak_guard_idle(5814.0, 5413.0, month_peak=True,
-                                    cut=True)
-_ok = messages.row_peak_guard_idle(5200.0, 5413.0, month_peak=True)
-print(f"    {_en}\n    {_cut}\n    {_ok}")
-check("over, nothing cut: 'idle, at this rate ..., over ...'",
-      _en == "Peak guard: idle, at this rate 5 814 W, over 5 413 W")
-check("over, a grid charge cut",
-      _cut == "Peak guard: grid charge limited, at this rate 5 814 W, "
-              "over 5 413 W")
-check("under: the 0.19.1 sentence, unchanged",
-      _ok == "Peak guard: idle, heading for 5 200 W, under this month's "
-             "5 413 W peak")
-check("the Dutch forms say the same",
-      _cut.nl == "Piekbewaking: laden beperkt, aan dit tempo 5 814 W, "
-                 "boven 5 413 W"
-      and _en.nl.startswith("Piekbewaking: in rust, aan dit tempo"))
-check("the month-peak line in shadow mode says 'would'",
-      homeview.line_period(5814.0, 5413.0, True, cut=True).endswith(
-          "The grid charge would be cut back to stay under 5.4 kW.")
-      and homeview.line_period(5814.0, 5413.0, False).endswith(
-          "The batteries step in above 5.4 kW."))
-
-
-# ============================================================================ #
-print("\n6. one inverter: the maintenance charge is sized to the room too")
-# The start-up quarter on a 10 s meter: the charge's first step is not seen
-# until the second reading, so it may land up to START_TOL over the aim; the
-# same at N=2 and before this fix. Every later quarter is graded at TOL.
 START_TOL = 15.0
-N1 = {}
-for peak, house, aim in ((5413.0, 4000.0, 5363.0), (None, 2500.0, AIM_CAP)):
-    for late, every in ((0.0, 0.0), (30.0, 0.0), (20.0, 10.0)):
-        _r, _c = s84.run(1, load=lambda t, h=house: h, dur=2700.0,
-                         month_peak=peak, before=late_feed(late, every))
-        _q = rig_quarters(_r, lambda t, h=house: h)
-        N1[peak, late, every] = (_r, _c)
-        _vals = list(_q.values())
-        _tag = (f"N=1, {'peak banked' if peak else 'no peak banked'}, "
-                f"{'10 s' if every else '5 s'} meter, "
-                f"{'late ' + format(late, '.0f') + ' s' if late else 'healthy'}")
-        print(f"    {_tag}: quarters "
-              f"{ {k: round(v, 1) for k, v in _q.items()} } W, aim {aim:.0f}")
-        check(f"{_tag}: every quarter at or under the aim",
-              len(_vals) == 3
-              and all(v <= aim + TOL for v in _vals[1:])
-              and nth(_vals, 0, "a first quarter")
-              <= aim + (START_TOL if every else TOL)
-              and any(mm.phase == "charge" for mm in _c.maints))
-_hr, _hc = N1[5413.0, 0.0, 0.0]
-_eng = sum(1 for r in _hr if r[-1][1] and r[-1][1][0])
-print(f"    healthy, peak banked: guard engaged on {_eng} of {len(_hr)} "
-      f"cycles; A charges on "
-      f"{sum(1 for r in _hr if dict(r[2])['a'] < -1.0)} cycles")
-check("N=1: the guard no longer steps in during the charge (no hand-back), "
-      "and the battery charges on every cycle, at no more than its rate",
-      _eng == 0 and all(-2000.0 <= dict(r[2])["a"] < -1.0 for r in _hr)
-      and not any("suspended" in e for e in _hc.events))
-
-
-# ============================================================================ #
-print("\n7. the row during an N=2 maintenance charge and in shadow mode")
-_rows7 = []
-
-
-def _look(c, t):
-    pg = c.peak_guard
-    if t >= QUARTER_S and pg is not None and pg.projected_w is not None \
-            and not pg.engaged and pg.projected_w > pg.ceiling_w:
-        _rows7.append(first((r["text"] for r in c.checks()
-                             if r["id"] == "peak_guard"), "guard row"))
-
-
-def _steps(t):
-    """4000 W, 800 W more for the first 30 s of every minute."""
-    return 4000.0 + (800.0 if t % 60.0 < 30.0 else 0.0)
-
-
-_m2r, _m2c = s84.run(2, load=_steps, dur=1800.0, month_peak=5413.0,
-                     before=late_feed(0.0, look=_look))
-_m2q = rig_quarters(_m2r, _steps)
-print(f"    N=2 maintenance charge, house steps 4000/4800 W: "
-      f"{len(_rows7)} over-the-limit rows, e.g. "
-      f"{first(iter(_rows7), 'a row')}; quarters "
-      f"{ {k: round(v, 1) for k, v in _m2q.items()} } W")
-check("N=2 maintenance charge, over the limit at this rate: the row says the "
-      "grid charge is limited",
-      present(first(iter(_rows7), "a row")) and all(
-          r.startswith("Peak guard: grid charge limited, at this rate")
-          for r in _rows7)
-      and all(v <= 5363.0 + TOL for v in _m2q.values()))
-_sh_o = dict(opts(), shadow_mode="on")
-_csh, _rsh, _fsh, _ssh = drive(_sh_o, peak=PEAK, minutes=17, every=10,
-                               plant=OnItsOwn)
-_sh_over = [s for s in _ssh if not s[2] and s[0] > s[1]]
-_sh1 = first(iter(_sh_over), "an over sample")
-print(f"    shadow mode, the inverter charging 5000 W on its own: "
-      f"{len(_sh_over)} over samples, e.g. "
-      f"{_sh1[3] if present(_sh1) else _sh1!s} / "
-      f"{_sh1[4] if present(_sh1) else _sh1!s}")
-check("shadow mode: nothing is cut on the wire, so the row says 'idle', "
-      "never 'grid charge limited', and the line says 'would'",
-      _csh.shadow and present(_sh1) and all(
-          s[3].startswith("Peak guard: idle, at this rate")
-          and "would" in s[4] for s in _sh_over))
-
-
 # ============================================================================ #
 print("\n8. edge cases")
 _e45 = drive(opts(), peak=PEAK, late=45.0)
@@ -0,0 +1,436 @@
+"""Runnable check for SAFETY-125, a part.  `python3 test_safety125b.py`
+
+Split out of test_safety125.py by CI-18 so CI can run the pieces side by side (the one file
+took ~10 min). The docstring there describes the whole. It holds parts 3-7 (a price trade and a maintenance charge, a healthy meter, the words, one inverter, N=2 and shadow). Same helpers,
+same checks, none dropped: the pieces print exactly the check lines the one file did.
+Read MUTATION.md before you report a survivor."""
+
+
+import asyncio
+import logging
+import os
+import sys
+import tempfile
+import types
+from datetime import datetime, timedelta, timezone
+
+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 opt11p3_replay as rp                                    # noqa: E402
+import safety84_identity as s84                                # noqa: E402
+import main_files  # noqa: E402
+import app.main as m                                           # noqa: E402
+import app.p1 as p1mod                                         # noqa: E402
+from named_absence import first, nth, present                  # noqa: E402
+from app import homeview, messages, peakguard, tariff          # noqa: E402
+from app import price as _price, static_tariff as _st          # noqa: E402
+from app.main import Controller                                # noqa: E402
+from app.p1 import make_sample, split_signed                   # noqa: E402
+from app.store import unit_key                                 # noqa: E402
+from app.tariff import QUARTER_S, Projection                   # noqa: E402
+
+fails = []
+
+
+def check(name, cond):
+    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
+    if not cond:
+        fails.append(name)
+
+
+PEAK = 5413.17          # October's banked peak on the live site
+AIM_PEAK = PEAK - 50.0
+AIM_CAP = 3500.0 - 50.0
+TOL = 10.0              # W over the aim a quarter may land (SAFETY-81)
+FC = "sensor.piek_prognose_kwartier"
+L = timezone(timedelta(hours=1))
+START = datetime(2026, 10, 29, 0, 0, tzinfo=L).astimezone(timezone.utc)
+PROFILE = ('{"periods": ['
+           '{"name": "night", "days": "all", "from": "22:00", "to": "07:00",'
+           ' "eur_per_kwh": 0.2333},'
+           '{"name": "day", "days": "all", "from": "07:00", "to": "22:00",'
+           ' "eur_per_kwh": 0.2801}]}')
+
+
+def opts(*, fc=False, modes=True, arb=False):
+    o = dict(rp.options(modes, arb, 2))
+    o.update(cycle_cost_threshold_eur=0.03, static_profile=PROFILE,
+             peak_cap_w=3500, peak_margin_w=50)
+    o["inverters"][0].update(max_w=5000, capacity_kwh=15.36)
+    o["inverters"][1].update(max_w=4600, capacity_kwh=20.48)
+    if fc:
+        o["peak_forecast_entity"] = FC
+    return o
+
+
+class Harder(rp.Plant):
+    """Charges harder than asked, as inverter-a did live (asked -4271 W,
+    measured about -4750 W; asked -5000 W, about -5210 W)."""
+
+    def step(self, dt, floor):
+        c = self.cmd
+        if c < 0:
+            self.cmd = max(c * 1.11, -5216.0)
+        try:
+            super().step(dt, floor)
+        finally:
+            self.cmd = c
+
+
+async def _no_fetch(*a, **k):
+    raise AssertionError("the tariff task reached for a price feed")
+
+
+class OnItsOwn(rp.Plant):
+    """Charges 5000 W whatever it is asked (shadow mode: nothing is
+    written, the inverter charges on its own)."""
+
+    def step(self, dt, floor):
+        self.cmd = -5000.0
+        super().step(dt, floor)
+
+
+def drive(o, *, peak=None, late=0.0, minutes=31, every=0, start=None,
+          period=0.0, fc_unit="kW", plant=Harder):
+    """The real Controller from `start` (00:00 local) for `minutes` at a 1 s
+    cycle (OPT-15's lag). `late`: no P1 telegram in each quarter's last
+    `late` s; `period`: one telegram every `period` s (0: every cycle).
+    -> (controller, rows, {quarter start: true grid average}, samples), the
+    samples taken every `every` cycles from the second quarter (the first
+    starts before the first reading): (projection, ceiling, engaged,
+    peak_guard row, month-peak line)."""
+    store = rp._Store()
+    for u, mode in (("a", "cheap_hours"), ("b", "solar_first")):
+        store.data[unit_key(f"inverter-{u}", "pack_use")] = True
+        store.data[unit_key(f"inverter-{u}", "pack_mode")] = mode
+        store.data[unit_key(f"inverter-{u}", "pack_sell")] = False
+    rows, samples, sums = [], [], {}
+    t0 = start or START
+
+    async def go():
+        lag = rp.Lag()
+        hass = rp.Hass(("a", "b"), lambda t: 1000.0, {"a": 11.0, "b": 10.0},
+                       limits={"a": (-5000.0, 5000.0),
+                               "b": (-4600.0, 4600.0)}, lag=lag)
+        hass.plants["a"].__class__ = plant
+        fc_kw = [0.0]
+
+        async def state(e):
+            # The HA quarter projection, in kW like the live template.
+            return None if e != FC else {
+                "state": (f"{fc_kw[0]:.3f}" if fc_unit == "kW"
+                          else f"{fc_kw[0] * 1000.0:.0f}"),
+                "attributes": {"unit_of_measurement": fc_unit}}
+        hass.state = state
+        tmp = tempfile.mkdtemp(prefix="s125-")
+        db = os.path.join(tmp, "tariff.db")
+        if peak is not None:
+            tariff.PeakStore(db)._db.execute(
+                "INSERT INTO monthly_peak VALUES (?, ?, ?, NULL)",
+                (tariff.month_key(t0), peak,
+                 "2026-10-21T20:15:00+00:00")).connection.commit()
+        c = Controller(dict(o), hass, store, rp._Pub(), stop=asyncio.Event(),
+                       peak_db_path=db,
+                       price_history_path=os.path.join(tmp, "ph.json"))
+        c.inverters[0].dev_min, c.inverters[0].dev_max = -5000.0, 5000.0
+        c.inverters[1].dev_min, c.inverters[1].dev_max = -4600.0, 4600.0
+        c.auto = True
+        real = main_files.live.datetime, homeview.datetime, p1mod.time
+        clock, simt = [t0], [0.0]
+
+        class _DT(real[0]):
+            @classmethod
+            def now(cls, tz=None):
+                return clock[0] if tz is None else clock[0].astimezone(tz)
+        main_files.live.datetime = homeview.datetime = _DT
+        p1mod.time = types.SimpleNamespace(monotonic=lambda: 1e6 + simt[0])
+        real_load = _price.load_prices, _st.load_prices
+        _price.load_prices = _st.load_prices = _no_fetch
+        task = asyncio.create_task(c.run_tariff_refresh())
+        try:
+            hass.advance(0.0, 10.0)
+            t, n = 0.0, 0
+            while t < minutes * 60:
+                now = t0 + timedelta(seconds=t)
+                hass.t, simt[0], clock[0] = t, t, now
+                if t == 0.0:
+                    for _ in range(500):
+                        if c.tariff_plan is not None:
+                            break
+                        await asyncio.sleep(0)
+                q = int(now.timestamp() // QUARTER_S)
+                acc = sums.setdefault(q, [0.0, 0])
+                g_now = hass.grid()
+                fc_kw[0] = (acc[0] + g_now * (QUARTER_S - acc[1])) \
+                    / QUARTER_S / 1000.0
+                if now.timestamp() % QUARTER_S < QUARTER_S - late \
+                        and t >= lag.p1_s \
+                        and (not period or t % period < 1e-9):
+                    imp, exp = split_signed(hass._past(lag.meter_s
+                                                       + lag.p1_s)[0])
+                    c.p1.submit(make_sample(
+                        "ha_signed", imp, exp, phases=1,
+                        ingest_ts=now - timedelta(seconds=lag.p1_s),
+                        ingest_mono=1e6 + t))
+                await c.read_inputs(now_mono=t)
+                await c.cycle()
+                hass.advance(lag.tick, 10.0)
+                acc[0] += max(0.0, hass.grid())
+                acc[1] += 1
+                rows.append((now, c.target, hass.plants["a"].p,
+                             getattr(c.arb, "action", None)))
+                pg = c.peak_guard
+                if every and n % every == 0 and t >= QUARTER_S \
+                        and pg is not None and pg.projected_w is not None:
+                    row = first((r["text"] for r in c.checks(now)
+                                 if r["id"] == "peak_guard"), "guard row")
+                    samples.append((pg.projected_w, pg.ceiling_w, pg.engaged,
+                                    row, homeview.month_peak(c, now)["line"]))
+                t += lag.tick
+                n += 1
+        finally:
+            task.cancel()
+            try:
+                await task
+            except asyncio.CancelledError:
+                pass
+            main_files.live.datetime, homeview.datetime, p1mod.time = real
+            _price.load_prices, _st.load_prices = real_load
+        return c
+    c = asyncio.run(go())
+    full = {datetime.fromtimestamp(q * QUARTER_S, L).strftime("%H:%M"):
+            s / k for q, (s, k) in sorted(sums.items()) if k >= QUARTER_S}
+    return c, rows, full, samples
+
+
+def into(now):
+    return now.timestamp() % QUARTER_S
+
+
+# ============================================================================ #
+def graded(label, full, aim):
+    print(f"    {label}: quarters "
+          f"{ {k: round(v, 1) for k, v in full.items()} } W, aim {aim:.0f}")
+    return len(full) >= 2 and all(v <= aim + TOL for v in full.values())
+
+
+def cut_and_back(rows):
+    """The charge is off in each quarter's last 5 s, and back (A charging
+    over 2 kW) within the next quarter's first minute."""
+    ends = [r for r in rows if into(r[0]) >= QUARTER_S - 5]
+    starts = {}
+    for r in rows:
+        if into(r[0]) < 60:
+            q = int(r[0].timestamp() // QUARTER_S)
+            starts[q] = starts.get(q, False) or r[2] < -2000.0
+    return (present(first(iter(ends), "a quarter end"))
+            and all(r[1] > -1.0 for r in ends)
+            and len(starts) >= 2 and all(starts.values()))
+
+
+# ============================================================================ #
+print("\n3. the same for a price trade (modes off) and a maintenance charge")
+_ca, _ra, _fa = drive(opts(modes=False, arb=True), peak=PEAK, late=30.0)[:3]
+check("modes off, price trading on, a reading 30 s late: every quarter at "
+      "or under the aim", graded("trade", _fa, AIM_PEAK)
+      and any(r[3] == "ARBITRAGE_CHARGE" for r in _ra))
+
+
+def late_feed(gap, every=0.0, look=None):
+    """safety84's cheap-window feed, no telegram in each quarter's last
+    `gap` s, and one only every `every` s (0: every cycle). `look(c, t)`
+    runs first, on the state the previous cycle left."""
+    feed = s84.feed(True)
+
+    def before(c, t):
+        if look is not None:
+            look(c, t)
+        if ((rig.BASE.timestamp() + t) % QUARTER_S < QUARTER_S - gap
+                and (not every or t % every < 1e-9)) \
+                or not getattr(c.hass, "_s84", False):
+            feed(c, t)
+    return before
+
+
+def rig_quarters(rows, load):
+    """The safety84 rig's whole quarters: the true average import (W)."""
+    qs = {}
+    for r in rows:
+        q = int((rig.BASE.timestamp() + r[0]) // QUARTER_S)
+        a = qs.setdefault(q, [0.0, 0])
+        a[0] += max(0.0, load(r[0]) - sum(v for _u, v in r[2]))
+        a[1] += 1
+    return {datetime.fromtimestamp(q * QUARTER_S, L).strftime("%H:%M"):
+            s / k for q, (s, k) in sorted(qs.items())
+            if k * rig.TICK >= QUARTER_S}
+
+
+for peak, house, aim in ((5413.0, 4000.0, 5363.0), (None, 2500.0, AIM_CAP)):
+    _mr, _mc = s84.run(2, load=lambda t, h=house: h, dur=2700.0,
+                       month_peak=peak, before=late_feed(30.0))
+    _mf = rig_quarters(_mr, lambda t, h=house: h)
+    check(f"N=2 maintenance charge, house {house:.0f} W, "
+          f"{'peak banked' if peak else 'no peak banked'}, a reading 30 s "
+          f"late: every quarter at or under the aim",
+          graded("maintenance", _mf, aim)
+          and any(mm.phase == "charge" for mm in _mc.maints))
+
+
+# ============================================================================ #
+print("\n4. a healthy meter: the quarters, and the words")
+for name, kw, aim, kw_text in (("month peak banked", dict(peak=PEAK),
+                                AIM_PEAK, "5.4 kW"),
+                               ("no peak banked", dict(), AIM_CAP, "3.5 kW")):
+    _c, _rows, _full, _s = drive(opts(), every=10, **kw)
+    check(f"{name}: every quarter at or under the aim",
+          graded(name, _full, aim))
+    over = [s for s in _s if not s[2] and s[0] > s[1]]
+    under = [s for s in _s if not s[2] and s[0] <= s[1]]
+    _o, _u = first(iter(over), "an over sample"), \
+        first(iter(under), "an under sample")
+    print(f"    over the limit at this rate ({len(over)} samples): "
+          f"{_o[3] if present(_o) else _o!s} / "
+          f"{_o[4] if present(_o) else _o!s}")
+    print(f"    under it ({len(under)}): "
+          f"{_u[3] if present(_u) else _u!s} / "
+          f"{_u[4] if present(_u) else _u!s}")
+    check(f"{name}: over the limit at this rate the guard row never says "
+          f"'under'; it says the grid charge is limited",
+          present(_o) and all(
+              "under" not in s[3] and "Peak guard: grid charge limited, "
+              "at this rate" in s[3] and " over " in s[3] for s in over))
+    check(f"{name}: ...and the month-peak line says the rate, and that the "
+          f"grid charge is cut back to stay under {kw_text}",
+          present(_o) and all(
+              s[4].startswith("At this rate this 15-minute period reaches")
+              and s[4].endswith("The grid charge is cut back to stay "
+                                f"under {kw_text}.") for s in over))
+    check(f"{name}: under the limit both read as before",
+          present(_u) and all(
+              s[3].startswith("Peak guard: idle, heading for")
+              and " under " in s[3]
+              and s[4].startswith("This 15-minute period is heading for")
+              and s[4].endswith(f"The batteries step in above {kw_text}.")
+              for s in under))
+
+
+# ============================================================================ #
+print("\n5. the words on their own")
+_en = messages.row_peak_guard_idle(5814.0, 5413.0, month_peak=True)
+_cut = messages.row_peak_guard_idle(5814.0, 5413.0, month_peak=True,
+                                    cut=True)
+_ok = messages.row_peak_guard_idle(5200.0, 5413.0, month_peak=True)
+print(f"    {_en}\n    {_cut}\n    {_ok}")
+check("over, nothing cut: 'idle, at this rate ..., over ...'",
+      _en == "Peak guard: idle, at this rate 5 814 W, over 5 413 W")
+check("over, a grid charge cut",
+      _cut == "Peak guard: grid charge limited, at this rate 5 814 W, "
+              "over 5 413 W")
+check("under: the 0.19.1 sentence, unchanged",
+      _ok == "Peak guard: idle, heading for 5 200 W, under this month's "
+             "5 413 W peak")
+check("the Dutch forms say the same",
+      _cut.nl == "Piekbewaking: laden beperkt, aan dit tempo 5 814 W, "
+                 "boven 5 413 W"
+      and _en.nl.startswith("Piekbewaking: in rust, aan dit tempo"))
+check("the month-peak line in shadow mode says 'would'",
+      homeview.line_period(5814.0, 5413.0, True, cut=True).endswith(
+          "The grid charge would be cut back to stay under 5.4 kW.")
+      and homeview.line_period(5814.0, 5413.0, False).endswith(
+          "The batteries step in above 5.4 kW."))
+
+
+# ============================================================================ #
+print("\n6. one inverter: the maintenance charge is sized to the room too")
+# The start-up quarter on a 10 s meter: the charge's first step is not seen
+# until the second reading, so it may land up to START_TOL over the aim; the
+# same at N=2 and before this fix. Every later quarter is graded at TOL.
+START_TOL = 15.0
+N1 = {}
+for peak, house, aim in ((5413.0, 4000.0, 5363.0), (None, 2500.0, AIM_CAP)):
+    for late, every in ((0.0, 0.0), (30.0, 0.0), (20.0, 10.0)):
+        _r, _c = s84.run(1, load=lambda t, h=house: h, dur=2700.0,
+                         month_peak=peak, before=late_feed(late, every))
+        _q = rig_quarters(_r, lambda t, h=house: h)
+        N1[peak, late, every] = (_r, _c)
+        _vals = list(_q.values())
+        _tag = (f"N=1, {'peak banked' if peak else 'no peak banked'}, "
+                f"{'10 s' if every else '5 s'} meter, "
+                f"{'late ' + format(late, '.0f') + ' s' if late else 'healthy'}")
+        print(f"    {_tag}: quarters "
+              f"{ {k: round(v, 1) for k, v in _q.items()} } W, aim {aim:.0f}")
+        check(f"{_tag}: every quarter at or under the aim",
+              len(_vals) == 3
+              and all(v <= aim + TOL for v in _vals[1:])
+              and nth(_vals, 0, "a first quarter")
+              <= aim + (START_TOL if every else TOL)
+              and any(mm.phase == "charge" for mm in _c.maints))
+_hr, _hc = N1[5413.0, 0.0, 0.0]
+_eng = sum(1 for r in _hr if r[-1][1] and r[-1][1][0])
+print(f"    healthy, peak banked: guard engaged on {_eng} of {len(_hr)} "
+      f"cycles; A charges on "
+      f"{sum(1 for r in _hr if dict(r[2])['a'] < -1.0)} cycles")
+check("N=1: the guard no longer steps in during the charge (no hand-back), "
+      "and the battery charges on every cycle, at no more than its rate",
+      _eng == 0 and all(-2000.0 <= dict(r[2])["a"] < -1.0 for r in _hr)
+      and not any("suspended" in e for e in _hc.events))
+
+
+# ============================================================================ #
+print("\n7. the row during an N=2 maintenance charge and in shadow mode")
+_rows7 = []
+
+
+def _look(c, t):
+    pg = c.peak_guard
+    if t >= QUARTER_S and pg is not None and pg.projected_w is not None \
+            and not pg.engaged and pg.projected_w > pg.ceiling_w:
+        _rows7.append(first((r["text"] for r in c.checks()
+                             if r["id"] == "peak_guard"), "guard row"))
+
+
+def _steps(t):
+    """4000 W, 800 W more for the first 30 s of every minute."""
+    return 4000.0 + (800.0 if t % 60.0 < 30.0 else 0.0)
+
+
+_m2r, _m2c = s84.run(2, load=_steps, dur=1800.0, month_peak=5413.0,
+                     before=late_feed(0.0, look=_look))
+_m2q = rig_quarters(_m2r, _steps)
+print(f"    N=2 maintenance charge, house steps 4000/4800 W: "
+      f"{len(_rows7)} over-the-limit rows, e.g. "
+      f"{first(iter(_rows7), 'a row')}; quarters "
+      f"{ {k: round(v, 1) for k, v in _m2q.items()} } W")
+check("N=2 maintenance charge, over the limit at this rate: the row says the "
+      "grid charge is limited",
+      present(first(iter(_rows7), "a row")) and all(
+          r.startswith("Peak guard: grid charge limited, at this rate")
+          for r in _rows7)
+      and all(v <= 5363.0 + TOL for v in _m2q.values()))
+_sh_o = dict(opts(), shadow_mode="on")
+_csh, _rsh, _fsh, _ssh = drive(_sh_o, peak=PEAK, minutes=17, every=10,
+                               plant=OnItsOwn)
+_sh_over = [s for s in _ssh if not s[2] and s[0] > s[1]]
+_sh1 = first(iter(_sh_over), "an over sample")
+print(f"    shadow mode, the inverter charging 5000 W on its own: "
+      f"{len(_sh_over)} over samples, e.g. "
+      f"{_sh1[3] if present(_sh1) else _sh1!s} / "
+      f"{_sh1[4] if present(_sh1) else _sh1!s}")
+check("shadow mode: nothing is cut on the wire, so the row says 'idle', "
+      "never 'grid charge limited', and the line says 'would'",
+      _csh.shadow and present(_sh1) and all(
+          s[3].startswith("Peak guard: idle, at this rate")
+          and "would" in s[4] for s in _sh_over))
+
+
+print()
+if fails:
+    print(f"{len(fails)} FAILED:")
+    for f in fails:
+        print(f"  - {f}")
+    sys.exit(1)
+print("all checks passed")
@@ -25,7 +25,11 @@ quarter's edge as on the rig.
 4. The page and the "Why" on a decision within the aim.

 Mutations (MUTATION.md, purged copies) are listed in the pull request.
-"""
+
+CI-18: this file holds parts 1 and 4 (part 4 grades part 1's nights). Parts 2 and 3 are in test_safety126b.py, split out so
+CI can run the pieces side by side (the one file took ~13 min). Every check is in
+exactly one file; the two print exactly the check lines the one file did."""
+

 import asyncio
 import dataclasses
@@ -307,127 +311,6 @@ check("fix reverted, at 0.01 s: the ERROR, the notification and the "
      _b["errors"] and _b["notes"] and any(s[3] for s in _b["seen"]))


-# ============================================================================ #
-print("\n2. a real overrun under the reserve still alarms")
-
-
-def evening(t):
-    """600 W, then an oven and a car charger from 20 minutes on."""
-    return 600.0 if t < 1200 else 6500.0
-
-
-for tag, onset, kw in (
-        ("night grid charge, then 6500 W", 1200.0, dict(house=evening)),
-        ("modes off, 5500 W all along", 0.0,
-         dict(house=5500.0, modes=None, socs=(15.0, 12.0)))):
-    o = opts(modes=kw.get("modes", OWNER) is not None)
-    head = drive(o, minutes=35, **kw)
-    base = without_fix(lambda: drive(o, minutes=35, **kw))
-    show("head " + tag, head)
-    show("fix reverted", base)
-    # From the load's onset: before it, the fix reverted shows the night's
-    # false alarm (part 1).
-    first_h = next((s[0] for s in head["seen"] if s[6] and s[0] >= onset),
-                   None)
-    first_b = next((s[0] for s in base["seen"] if s[6] and s[0] >= onset),
-                   None)
-    check(f"⭐ {tag}: the ERROR and the notification, naming both "
-          f"batteries",
-          head["errors"] and head["notes"]
-          and all("inverter-a and inverter-b" in n for _t, n in head["notes"]))
-    _th = [t for t, _n in head["notes"]]
-    _tb = [t for t, _n in base["notes"] if t >= onset]
-    print(f"    notifications at {_th} s; with the fix reverted {_tb} s")
-    check(f"{tag}: the first at the same cycle as before the fix "
-          f"({first_h} s, {first_b} s), and none that was not raised before",
-          first_h is not None and first_h == first_b
-          and present(nth(_th, 0)) and nth(_th, 0) == nth(_tb, 0)
-          and set(_th) <= set(_tb))
-    check(f"{tag}: the red peak row and its banner while it stands",
-          any(s[4] for s in head["seen"]))
-
-
-def burst(t):
-    """600 W, a 6 500 W load from 20 to 25 minutes, then 600 W again while
-    the night charge goes on."""
-    return 6500.0 if 1200 <= t < 1500 else 600.0
-
-
-# F3 (review): `_peak_landing()` resets its latch when the guard lets go, so
-# a real overrun does not keep the later landing errors alarming.
-_bh = drive(opts(), house=burst, off=0.99, minutes=61)
-_bb = without_fix(lambda: drive(opts(), house=burst, off=0.99,
-                                     minutes=61))
-show("head real overrun, then the night again", _bh)
-show("fix reverted", _bb)
-# The burst banked a new month peak, which the later quarters stay under.
-_new_peak = _bh["c"].peak_guard.ceiling_w
-
-
-def _late(r):
-    """From the next quarter on: the overrun's own quarter ends over."""
-    return [x for x in r["seen"] if x[0] >= 1800]
-
-
-print(f"    notifications at {[t for t, _n in _bh['notes']]} s; with the fix "
-      f"reverted {[t for t, _n in _bb['notes']]} s; limit after the burst "
-      f"{_new_peak:.0f} W")
-check("⭐ after a real overrun ends, the latch is reset: the alarm came "
-      "during the overrun, and the later quarters' landings are quiet "
-      "(no notification, no 'Why' peak, no red row) and under the new peak",
-      any(1200 <= t < 1500 for t, _n in _bh["notes"])
-      and not [t for t, _n in _bh["notes"] if t >= 1800]
-      and not any(x[3] or x[4] or x[5] for x in _late(_bh))
-      and present(nth([v for k, v in _bh["quarters"].items()
-                       if k >= "22:30"], 0))
-      and all(v <= _new_peak for k, v in _bh["quarters"].items()
-              if k >= "22:30"))
-check("...not vacuous: the guard wanted power at those landings, and with "
-      "the fix reverted they show the 'Why' peak or an alarm",
-      any(x[2] for x in _late(_bh))
-      and (any(x[3] or x[4] or x[5] for x in _late(_bb))
-           or bool([t for t, _n in _bb["notes"] if t >= 1800])))
-
-
-# ============================================================================ #
-print("\n3. misses_aim() on its own")
-START = datetime(2026, 10, 30, 21, 15, tzinfo=timezone.utc)
-
-
-def proj(projected, elapsed, held):
-    return Projection(projected, projected, elapsed, 1.0, True, START, held)
-
-
-check("the rig's case: 3452 W against 3450 W, 0.3 s left, charging "
-      "1250 W from the grid: not a miss",
-      not peakguard.misses_aim(proj(3452.0, 899.7, 3452.0), batt_w=-1250.0,
-                               target_w=AIM_CAP))
-check("...although decide() wants power there (its gain, 900 / 0.3 s)",
-      (peakguard.decide(proj(3452.0, 899.7, 3452.0), batt_w=-1250.0,
-                        cap_w=3500, margin_w=50, reading=None,
-                        achievable_w=0.0, soc=10.0, soc_floor=20.0)
-       .wanted_w or 0.0) > 1000.0)
-check("a quarter heading 11 W over the aim with the batteries at 0 W is a "
-      "miss", peakguard.misses_aim(proj(3461.0, 899.7, 3461.0), batt_w=0.0,
-                                   target_w=AIM_CAP))
-check("...and 10 W is not (ALARM_MISS_W, as for 'Load too high')",
-      not peakguard.misses_aim(proj(3460.0, 899.7, 3460.0), batt_w=0.0,
-                               target_w=AIM_CAP))
-check("the add-on's own grid charge counts as stoppable: 4000 W on the "
-      "meter mid-quarter, 3000 W of it charging, is not a miss",
-      not peakguard.misses_aim(proj(2500.0, 450.0, 4000.0), batt_w=-3000.0,
-                               target_w=AIM_CAP))
-check("...and a house of 6000 W the batteries are covering is one",
-      peakguard.misses_aim(proj(1000.0, 450.0, 0.0), batt_w=6000.0,
-                           target_w=AIM_CAP))
-check("no battery reading, or no seconds left: not a miss (decide() files "
-      "nothing there)",
-      not peakguard.misses_aim(proj(9000.0, 450.0, 9000.0), batt_w=None,
-                               target_w=AIM_CAP)
-      and not peakguard.misses_aim(proj(9000.0, 900.0, 9000.0), batt_w=0.0,
-                                   target_w=AIM_CAP))
-
-
 # ============================================================================ #
 print("\n4. the page and the 'Why' on a decision within the aim")
 _c = NIGHTS["aim 3450, cycle 0.01 s before the edge"][0]["c"]
@@ -0,0 +1,382 @@
+"""Runnable check for SAFETY-126, a part.  `python3 test_safety126b.py`
+
+Split out of test_safety126.py by CI-18 so CI can run the pieces side by side (the one file
+took ~13 min). The docstring there describes the whole. It holds parts 2 and 3 (a real overrun still alarms, misses_aim()). Same helpers,
+same checks, none dropped: the pieces print exactly the check lines the one file did.
+Read MUTATION.md before you report a survivor."""
+
+
+import asyncio
+import dataclasses
+import io
+import logging
+import os
+import sys
+import tempfile
+import types
+from datetime import datetime, timedelta, timezone
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, HERE)
+sys.path.insert(0, os.path.join(HERE, "tools"))
+
+import opt11p3_replay as rp                                    # noqa: E402
+from named_absence import first, nth, present                  # noqa: E402
+import main_files  # noqa: E402
+import app.main as m                                           # noqa: E402
+import app.p1 as p1mod                                         # noqa: E402
+from app import homeview, messages, peakguard, tariff          # noqa: E402
+from app import price as _price, static_tariff as _st          # noqa: E402
+from app.main import Controller                                # noqa: E402
+from app.p1 import make_sample, split_signed                   # noqa: E402
+from app.store import unit_key                                 # noqa: E402
+from app.tariff import QUARTER_S, Projection                   # noqa: E402
+
+logging.disable(logging.NOTSET)
+logging.getLogger().setLevel(logging.CRITICAL)
+_ERRORS = io.StringIO()
+_h = logging.StreamHandler(_ERRORS)
+_h.setLevel(logging.ERROR)
+logging.getLogger("goodwe").addHandler(_h)
+logging.getLogger("goodwe").setLevel(logging.ERROR)
+logging.getLogger("goodwe").propagate = False
+
+fails = []
+
+
+def check(name, cond):
+    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
+    if not cond:
+        fails.append(name)
+
+
+PEAK = 5413.17          # October's banked peak on the live site
+AIM_CAP = 3500.0 - 50.0
+AIM_PEAK = PEAK - 50.0
+TOL = 10.0              # W over the aim a quarter may land (SAFETY-81)
+L = timezone(timedelta(hours=1))
+NIGHT = datetime(2026, 10, 30, 22, 0, tzinfo=L).astimezone(timezone.utc)
+PROFILE = ('{"periods": ['
+           '{"name": "night", "days": "all", "from": "22:00", "to": "07:00",'
+           ' "eur_per_kwh": 0.2333},'
+           '{"name": "day", "days": "all", "from": "07:00", "to": "22:00",'
+           ' "eur_per_kwh": 0.2801}]}')
+OWNER = {"a": "leftover_solar", "b": "cheap_hours"}
+
+
+def opts(*, modes=True, arb=False):
+    o = dict(rp.options(modes, arb, 2))
+    o.update(cycle_cost_threshold_eur=0.03, static_profile=PROFILE,
+             peak_cap_w=3500, peak_margin_w=50)
+    o["inverters"][0].update(max_w=5000, capacity_kwh=15.36)
+    o["inverters"][1].update(max_w=4600, capacity_kwh=20.48)
+    return o
+
+
+class Harder(rp.Plant):
+    """Charges harder than asked, as inverter-a did live (test_safety125.py:
+    asked -4271 W, measured about -4750 W; asked -5000 W, about -5210 W), up
+    to `cap` W. So a charge sized to the aim lands a few W over it, as on the
+    rig (worst +7 W)."""
+    cap = -5216.0
+
+    def step(self, dt, floor):
+        c = self.cmd
+        if c < 0:
+            self.cmd = max(c * 1.11, self.cap)
+        try:
+            super().step(dt, floor)
+        finally:
+            self.cmd = c
+
+
+async def _no_fetch(*a, **k):
+    raise AssertionError("the tariff task reached for a price feed")
+
+
+def without_fix(fn):
+    """release/1.0 cc9bbeb: no decision is ever within the aim."""
+    real = getattr(Controller, "_peak_landing", None)
+    if real is None:
+        return fn()
+    Controller._peak_landing = lambda self, decision: decision
+    try:
+        return fn()
+    finally:
+        Controller._peak_landing = real
+
+
+def drive(o, *, modes=OWNER, socs=(10.0, 10.0), house=600.0, minutes=46,
+          start=NIGHT, off=0.7, peak=None):
+    """The real Controller from `start` + `off` s for `minutes` at a 1 s
+    cycle with OPT-15's lag, so a cycle lands `1 - off` s before each
+    quarter's end. `house` is W or a function of t. Returns a dict: the
+    quarters' true grid averages, the commands, the reserve notifications,
+    the ERROR lines, and per cycle where the guard engaged: (t, floor
+    limited, within the aim, battery "Why" peak, red reserve row, red
+    saturated row, alarm latched)."""
+    store = rp._Store()
+    for u, mode in (modes or {}).items():
+        store.data[unit_key(f"inverter-{u}", "pack_use")] = True
+        store.data[unit_key(f"inverter-{u}", "pack_mode")] = mode
+        store.data[unit_key(f"inverter-{u}", "pack_sell")] = False
+    load = house if callable(house) else (lambda t: float(house))
+    t0 = start + timedelta(seconds=off)
+    out = {"cmds": [], "notes": [], "seen": [], "sums": {}}
+    _ERRORS.seek(0)
+    _ERRORS.truncate()
+
+    async def go():
+        lag = rp.Lag()
+        hass = rp.Hass(("a", "b"), load, {"a": socs[0], "b": socs[1]},
+                       limits={"a": (-5000.0, 5000.0),
+                               "b": (-4600.0, 4600.0)}, lag=lag)
+        for u, cap in (("a", -5216.0), ("b", -4800.0)):
+            hass.plants[u].__class__ = Harder
+            hass.plants[u].cap = cap
+        real_set = hass.set_number
+
+        async def set_number(e, v):
+            out["cmds"].append((round(hass.t, 3), e, float(v)))
+            return await real_set(e, v)
+
+        async def call(*a):
+            if a[:2] == ("persistent_notification", "create") \
+                    and a[2]["notification_id"] == messages.ID_PEAK_RESERVE:
+                out["notes"].append((hass.t, a[2]["message"]))
+            return True
+        hass.set_number, hass.call = set_number, call
+        tmp = tempfile.mkdtemp(prefix="s126-")
+        db = os.path.join(tmp, "tariff.db")
+        if peak is not None:
+            tariff.PeakStore(db)._db.execute(
+                "INSERT INTO monthly_peak VALUES (?, ?, ?, NULL)",
+                (tariff.month_key(t0), peak,
+                 "2026-10-21T20:15:00+00:00")).connection.commit()
+        c = Controller(dict(o), hass, store, rp._Pub(), stop=asyncio.Event(),
+                       peak_db_path=db,
+                       price_history_path=os.path.join(tmp, "ph.json"))
+        c.inverters[0].dev_min, c.inverters[0].dev_max = -5000.0, 5000.0
+        c.inverters[1].dev_min, c.inverters[1].dev_max = -4600.0, 4600.0
+        c.auto = True
+        real = main_files.live.datetime, homeview.datetime, p1mod.time
+        clock, simt = [t0], [0.0]
+
+        class _DT(real[0]):
+            @classmethod
+            def now(cls, tz=None):
+                return clock[0] if tz is None else clock[0].astimezone(tz)
+        main_files.live.datetime = homeview.datetime = _DT
+        p1mod.time = types.SimpleNamespace(monotonic=lambda: 1e6 + simt[0])
+        real_load = _price.load_prices, _st.load_prices
+        _price.load_prices = _st.load_prices = _no_fetch
+        task = (asyncio.create_task(c.run_tariff_refresh())
+                if o.get("battery_modes_enabled") or o.get("arbitrage_enabled")
+                else None)
+        try:
+            hass.advance(0.0, 10.0)
+            t = 0.0
+            while t < minutes * 60:
+                now = t0 + timedelta(seconds=t)
+                hass.t, simt[0], clock[0] = t, t, now
+                if t == 0.0 and task is not None:
+                    for _ in range(500):
+                        if c.tariff_plan is not None:
+                            break
+                        await asyncio.sleep(0)
+                q = int(now.timestamp() // QUARTER_S)
+                acc = out["sums"].setdefault(q, [0.0, 0])
+                if t >= lag.p1_s:
+                    imp, exp = split_signed(hass._past(lag.meter_s
+                                                       + lag.p1_s)[0])
+                    c.p1.submit(make_sample(
+                        "ha_signed", imp, exp, phases=1,
+                        ingest_ts=now - timedelta(seconds=lag.p1_s),
+                        ingest_mono=1e6 + t))
+                await c.read_inputs(now_mono=t)
+                await c.cycle()
+                hass.advance(lag.tick, 10.0)
+                acc[0] += max(0.0, hass.grid())
+                acc[1] += 1
+                g = c.peak_guard
+                if g is not None and g.engaged and g.blind is None:
+                    rows = {r["id"]: r for r in c.checks(now)}
+                    row = rows.get("peak_guard", {})
+                    red = row.get("ok") is False and not row.get("warn")
+                    out["seen"].append((
+                        t, g.floor_limited, getattr(g, "within_aim", False),
+                        homeview.peak_engaged(c),
+                        red and "banner" in row,
+                        red and "banner" not in row,
+                        c._peak_floor_limited))
+                t += lag.tick
+        finally:
+            if task is not None:
+                task.cancel()
+                try:
+                    await task
+                except asyncio.CancelledError:
+                    pass
+            main_files.live.datetime, homeview.datetime, p1mod.time = real
+            _price.load_prices, _st.load_prices = real_load
+        return c
+    out["c"] = asyncio.run(go())
+    out["quarters"] = {
+        datetime.fromtimestamp(q * QUARTER_S, L).strftime("%H:%M"): s / k
+        for q, (s, k) in sorted(out["sums"].items()) if k >= QUARTER_S - 1}
+    out["errors"] = [x for x in _ERRORS.getvalue().splitlines()
+                     if "PEAK_GUARD_FLOOR_LIMITED" in x]
+    return out
+
+
+def show(tag, r):
+    print(f"    {tag}: quarters "
+          f"{ {k: round(v, 1) for k, v in r['quarters'].items()} }, "
+          f"guard cycles {len(r['seen'])}, floor-limited "
+          f"{sum(1 for s in r['seen'] if s[1])}, within-aim "
+          f"{sum(1 for s in r['seen'] if s[2])}, Why peak "
+          f"{sum(1 for s in r['seen'] if s[3])}, notes {len(r['notes'])}, "
+          f"errors {len(r['errors'])}")
+
+
+def quiet(r):
+    """No reserve alarm in any form: ERROR, notification, red row, latch,
+    "Why" peak, and no red saturated row."""
+    return (not r["errors"] and not r["notes"]
+            and not any(s[3] or s[4] or s[5] or s[6] for s in r["seen"]))
+
+
+def on_aim(r, aim):
+    q = r["quarters"]
+    return len(q) >= 3 and all(v <= aim + TOL for v in q.values())
+
+
+# ============================================================================ #
+print("\n2. a real overrun under the reserve still alarms")
+
+
+def evening(t):
+    """600 W, then an oven and a car charger from 20 minutes on."""
+    return 600.0 if t < 1200 else 6500.0
+
+
+for tag, onset, kw in (
+        ("night grid charge, then 6500 W", 1200.0, dict(house=evening)),
+        ("modes off, 5500 W all along", 0.0,
+         dict(house=5500.0, modes=None, socs=(15.0, 12.0)))):
+    o = opts(modes=kw.get("modes", OWNER) is not None)
+    head = drive(o, minutes=35, **kw)
+    base = without_fix(lambda: drive(o, minutes=35, **kw))
+    show("head " + tag, head)
+    show("fix reverted", base)
+    # From the load's onset: before it, the fix reverted shows the night's
+    # false alarm (part 1).
+    first_h = next((s[0] for s in head["seen"] if s[6] and s[0] >= onset),
+                   None)
+    first_b = next((s[0] for s in base["seen"] if s[6] and s[0] >= onset),
+                   None)
+    check(f"⭐ {tag}: the ERROR and the notification, naming both "
+          f"batteries",
+          head["errors"] and head["notes"]
+          and all("inverter-a and inverter-b" in n for _t, n in head["notes"]))
+    _th = [t for t, _n in head["notes"]]
+    _tb = [t for t, _n in base["notes"] if t >= onset]
+    print(f"    notifications at {_th} s; with the fix reverted {_tb} s")
+    check(f"{tag}: the first at the same cycle as before the fix "
+          f"({first_h} s, {first_b} s), and none that was not raised before",
+          first_h is not None and first_h == first_b
+          and present(nth(_th, 0)) and nth(_th, 0) == nth(_tb, 0)
+          and set(_th) <= set(_tb))
+    check(f"{tag}: the red peak row and its banner while it stands",
+          any(s[4] for s in head["seen"]))
+
+
+def burst(t):
+    """600 W, a 6 500 W load from 20 to 25 minutes, then 600 W again while
+    the night charge goes on."""
+    return 6500.0 if 1200 <= t < 1500 else 600.0
+
+
+# F3 (review): `_peak_landing()` resets its latch when the guard lets go, so
+# a real overrun does not keep the later landing errors alarming.
+_bh = drive(opts(), house=burst, off=0.99, minutes=61)
+_bb = without_fix(lambda: drive(opts(), house=burst, off=0.99,
+                                     minutes=61))
+show("head real overrun, then the night again", _bh)
+show("fix reverted", _bb)
+# The burst banked a new month peak, which the later quarters stay under.
+_new_peak = _bh["c"].peak_guard.ceiling_w
+
+
+def _late(r):
+    """From the next quarter on: the overrun's own quarter ends over."""
+    return [x for x in r["seen"] if x[0] >= 1800]
+
+
+print(f"    notifications at {[t for t, _n in _bh['notes']]} s; with the fix "
+      f"reverted {[t for t, _n in _bb['notes']]} s; limit after the burst "
+      f"{_new_peak:.0f} W")
+check("⭐ after a real overrun ends, the latch is reset: the alarm came "
+      "during the overrun, and the later quarters' landings are quiet "
+      "(no notification, no 'Why' peak, no red row) and under the new peak",
+      any(1200 <= t < 1500 for t, _n in _bh["notes"])
+      and not [t for t, _n in _bh["notes"] if t >= 1800]
+      and not any(x[3] or x[4] or x[5] for x in _late(_bh))
+      and present(nth([v for k, v in _bh["quarters"].items()
+                       if k >= "22:30"], 0))
+      and all(v <= _new_peak for k, v in _bh["quarters"].items()
+              if k >= "22:30"))
+check("...not vacuous: the guard wanted power at those landings, and with "
+      "the fix reverted they show the 'Why' peak or an alarm",
+      any(x[2] for x in _late(_bh))
+      and (any(x[3] or x[4] or x[5] for x in _late(_bb))
+           or bool([t for t, _n in _bb["notes"] if t >= 1800])))
+
+
+# ============================================================================ #
+print("\n3. misses_aim() on its own")
+START = datetime(2026, 10, 30, 21, 15, tzinfo=timezone.utc)
+
+
+def proj(projected, elapsed, held):
+    return Projection(projected, projected, elapsed, 1.0, True, START, held)
+
+
+check("the rig's case: 3452 W against 3450 W, 0.3 s left, charging "
+      "1250 W from the grid: not a miss",
+      not peakguard.misses_aim(proj(3452.0, 899.7, 3452.0), batt_w=-1250.0,
+                               target_w=AIM_CAP))
+check("...although decide() wants power there (its gain, 900 / 0.3 s)",
+      (peakguard.decide(proj(3452.0, 899.7, 3452.0), batt_w=-1250.0,
+                        cap_w=3500, margin_w=50, reading=None,
+                        achievable_w=0.0, soc=10.0, soc_floor=20.0)
+       .wanted_w or 0.0) > 1000.0)
+check("a quarter heading 11 W over the aim with the batteries at 0 W is a "
+      "miss", peakguard.misses_aim(proj(3461.0, 899.7, 3461.0), batt_w=0.0,
+                                   target_w=AIM_CAP))
+check("...and 10 W is not (ALARM_MISS_W, as for 'Load too high')",
+      not peakguard.misses_aim(proj(3460.0, 899.7, 3460.0), batt_w=0.0,
+                               target_w=AIM_CAP))
+check("the add-on's own grid charge counts as stoppable: 4000 W on the "
+      "meter mid-quarter, 3000 W of it charging, is not a miss",
+      not peakguard.misses_aim(proj(2500.0, 450.0, 4000.0), batt_w=-3000.0,
+                               target_w=AIM_CAP))
+check("...and a house of 6000 W the batteries are covering is one",
+      peakguard.misses_aim(proj(1000.0, 450.0, 0.0), batt_w=6000.0,
+                           target_w=AIM_CAP))
+check("no battery reading, or no seconds left: not a miss (decide() files "
+      "nothing there)",
+      not peakguard.misses_aim(proj(9000.0, 450.0, 9000.0), batt_w=None,
+                               target_w=AIM_CAP)
+      and not peakguard.misses_aim(proj(9000.0, 900.0, 9000.0), batt_w=0.0,
+                                   target_w=AIM_CAP))
+
+
+# ============================================================================ #
+print()
+if fails:
+    print(f"{len(fails)} SAFETY-126 check(s) FAILED:")
+    for f in fails:
+        print(f"  - {f}")
+    sys.exit(1)
+print("all SAFETY-126 checks passed")