ci: trigger a run on the rebased head

No content change. The gate never created a run for 0942228d - the newest
run stayed on d31a12b7, the pre-rebase head - so `checks / unit` had
nothing to report and sat pending. Why no run was created is CI-02 and
belongs to CI-01's owner; this commit only asks for one.

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

.gitattributes

@@ -8,3 +8,11 @@
 * text=auto eol=lf
 *.png binary
 *.gz  binary
+
+# `.gitea/workflows/*.yml` steps embed shell heredocs; a `\r` on the terminator
+# line means the heredoc never closes and the job fails on a syntax error that is
+# invisible in the diff. A Windows checkout with core.autocrlf=true would
+# reintroduce it on the next edit. CI-01.
+*.yml  text eol=lf
+*.yaml text eol=lf
+Dockerfile text eol=lf

.gitea/ci/Dockerfile

@@ -0,0 +1,22 @@
+# The image the add-on's tests run in (CI-01). This repo had NO CI at all before
+# that ticket - eight runnable test files, none of them ever run by a machine.
+#
+# ⚠️ THE DEPENDENCIES ARE THE SHIPPED ONES, FROM APK, AND THAT IS THE POINT.
+# goodwe_controller/Dockerfile installs `py3-aiohttp py3-paho-mqtt` from Alpine
+# packages rather than pip, because Alpine ships musl and there are no musl
+# wheels for aiohttp - pip would compile it from source on a client's Raspberry
+# Pi. Testing against pip-installed glibc wheels would be testing a different
+# add-on from the one that ships. Same base, same package manager, same names.
+#
+# ⚠️ NOT `FROM ghcr.io/home-assistant/...-base`. The real Dockerfile takes its
+# base from the Supervisor's `BUILD_FROM` build-arg, which is per-architecture
+# and not resolvable outside a Supervisor build. `alpine:3` is what those images
+# are, minus the s6 supervision the tests do not use - and it is already cached
+# on this runner's host, so the gate does not depend on a registry pull.
+#
+# ⚠️ The runner container itself CANNOT run these tests: it is Alpine with a bare
+# python3, no aiohttp, no pip (`No module named pip`; checked on the box, not
+# assumed). An image is not gold-plating here, it is the only option that does
+# not mutate a shared long-lived runner.
+FROM alpine:3
+RUN apk add --no-cache python3 py3-aiohttp py3-paho-mqtt

.gitea/workflows/checks.yml

@@ -0,0 +1,136 @@
+# CI-01: this repository's FIRST automated check. Before this file, admin/goodwe-addon
+# had no `.gitea/workflows/` directory at all - eight runnable test files, ~100 s
+# of real coverage including a wall-clock cadence test, and not one machine ever
+# ran them. Every merge here was a human reading a diff.
+#
+# ⚠️ NO `paths:` FILTER, AND ITS ABSENCE IS LOAD-BEARING. Branch protection on
+# release/1.0 requires the context `checks / unit (pull_request)`. Gitea leaves a
+# required context it never receives PENDING forever, so the moment this workflow
+# learns to sit a PR out, that PR becomes unmergeable. Condition a STEP if you
+# must; never the workflow.
+#
+# ⚠️ THE RUNNER IS SHARED WITH THE OTHER REPO'S RIG. `goodwe-runner` on
+# 192.168.2.147 carries the labels `self-hosted` and `goodwe-rig` and is
+# registered `:host`, and it is the only runner there is - so this job wears the
+# rig label without wanting a rig. Nothing below starts, stops, or reads the
+# `goodwe-sim` compose project or takes `sim/riglock.py`; it runs one short
+# container and exits. It does occupy the runner's single slot (`capacity: 1`),
+# so it QUEUES behind a rig campaign - a delayed verdict, never a destroyed one.
+#
+# NO JS ACTIONS: the runner image has no node, so `actions/checkout` cannot run.
+# Checkout is seven lines of git below.
+name: checks
+
+on:
+  pull_request:
+    branches: [release/1.0, master]
+  workflow_dispatch:
+
+concurrency:
+  # ⚠️ PER PULL REQUEST, NOT A BARE NAME. A shared group holds one running and
+  # one pending run, so a bare name lets a second PR supersede a first PR's run
+  # before any runner has picked it up - which is exactly what happened to the
+  # other repo's run 521 (four jobs, `started_at` at the Unix epoch,
+  # `runner_id: 0`, concluded `cancelled`, and the PR merged anyway). Do not
+  # replace this with a constant.
+  group: addon-checks-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  unit:
+    runs-on: [self-hosted, goodwe-rig]
+    timeout-minutes: 70   # ~4 to run; the rest is queueing for the single slot
+    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
+      # path that exists only INSIDE the runner container. Hand it to the host's
+      # snap-confined dockerd and the build fails with "failed to read
+      # dockerfile", or - worse, and this is the documented case - a bind mount
+      # resolves to an EMPTY DIRECTORY with no error at all. /home/homelab is
+      # mirrored 1:1 into the runner and is under $HOME, where snap confinement
+      # accepts it. The same probe under /tmp failed and under $HOME succeeded.
+      WS: /home/homelab/ci/addon-checks-${{ github.run_number }}
+    steps:
+      # ⚠️ THE CHECKOUT IS TWO LEVELS BELOW THE MOUNT ROOT, DELIBERATELY.
+      # test_freeze.py's find_hwsim() reads `HERE.parents[1] / "goodwe" / "sim"`,
+      # looking for the infra repo as a SIBLING of this one. Check out flat -
+      # goodwe_controller directly at the mount root - and `parents[1]` does not
+      # exist, so the test dies on `IndexError: 1` before its first assertion.
+      # Measured during CI-01; the flat layout was tried first and did exactly
+      # that. $WS/goodwe-addon/goodwe_controller gives parents[1] == $WS.
+      - name: checkout into a path the host shares (git; no node for actions/checkout)
+        env:
+          URL: ${{ github.server_url }}/${{ github.repository }}.git
+          TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        run: |
+          set -eu
+          rm -rf "$WS" && mkdir -p "$WS/goodwe-addon" && cd "$WS/goodwe-addon"
+          git init -q .
+          git remote add origin "$(echo "$URL" | sed "s#://#://x-access-token:${TOKEN}@#")"
+          git fetch -q --depth 1 origin "$GITHUB_REF" || git fetch -q --depth 1 origin "$GITHUB_SHA"
+          git checkout -qf FETCH_HEAD
+          echo "checked out $(git log --oneline -1)"
+
+      - name: build the test image (shipped deps, from apk - see .gitea/ci/Dockerfile)
+        run: |
+          set -eu
+          docker build -q -t goodwe-addon-checks:1 "$WS/goodwe-addon/.gitea/ci" >/dev/null
+          docker run --rm goodwe-addon-checks:1 python3 -c \
+            'import aiohttp, paho.mqtt.client as m, sys; print("python", sys.version.split()[0], "aiohttp", aiohttp.__version__)'
+
+      # ⚠️ THE FILE LIST IS DERIVED FROM THE TREE, NOT TYPED HERE. A hand-written
+      # list is how a new test file joins the repo and is never run - the exact
+      # shape of failure this ticket exists to close. The step globs `test_*.py`,
+      # prints the count, and REFUSES A COUNT OF ZERO: a loop over an empty glob
+      # exits 0 and would report a green gate that examined nothing.
+      #
+      # ⚠️ test_loop.py TAKES ~93 SECONDS OF WALL CLOCK ON PURPOSE (its own
+      # docstring says so, and says why: it drives the real `run_control()`
+      # against a real wall-clock floor, and mocking the clock would leave the
+      # assertion testing the mock). It is not hung. Do not "fix" it with a
+      # shorter timeout - a 60 s limit was tried during CI-01 and reported
+      # exactly the false hang the docstring predicts.
+      #
+      # ⚠️ WHAT THIS RUN DOES NOT COVER, stated because a pass here would
+      # otherwise over-claim. test_freeze.py's third section wants the infra
+      # repo's `sim/hwsim.py` to drive a real HTTP meter through the real
+      # transport. admin/goodwe is PRIVATE (checked via the API) and this
+      # workflow's GITHUB_TOKEN is minted for THIS repository only, so hwsim is
+      # absent here and `find_hwsim()` returns None. The rest of the file runs
+      # and passes. It does so SILENTLY - the file prints nothing to say the
+      # hwsim section was skipped - so "test_freeze.py: PASS" below is a weaker
+      # claim than it looks. That belongs to whoever owns test_freeze.py next;
+      # naming it here is the least this gate can do about it.
+      - name: the add-on's own tests
+        run: |
+          set -eu
+          docker run --rm -v "$WS:/w:ro" goodwe-addon-checks:1 sh -eu -c '
+            cd /w/goodwe-addon/goodwe_controller
+            set -- test_*.py
+            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
+            echo "$# test file(s) to run"
+            rc=0
+            for f in "$@"; do
+              start=$(date +%s)
+              if python3 "$f" > /tmp/out 2>&1; then
+                echo "PASS  $f  ($(( $(date +%s) - start ))s)"
+              else
+                echo "FAIL  $f  ($(( $(date +%s) - start ))s)"
+                tail -40 /tmp/out
+                rc=1
+              fi
+            done
+            exit $rc
+          '
+
+      - name: clean up
+        if: always()
+        run: |
+          rm -rf "$WS" || true
+          # Root-owned by this runner inside a directory the `homelab` user owns,
+          # and sudo on that box is interactive-only, so the user cannot clean
+          # this up and the job that made it has to.
+          ls -1dt /home/homelab/ci/addon-checks-*/ 2>/dev/null | tail -n +6 | xargs -r rm -rf || true

goodwe_controller/CHANGELOG.md

@@ -2,6 +2,101 @@
 
 ## Unreleased
 
+**A computed setpoint over the rating now alarms instead of being clipped in
+silence (SAFETY-03).** The clamp was never the missing half - specs.md 10.3
+records an unclamped integrator climbing ~130 W per 4 s past 10 kW and reporting
+**14 768 W** while pinned at the 5 kW rail, and a silent clamp turns exactly that
+into a flat, legal-looking 5000 W hold. `write_setpoint()` - the one path every
+write takes, cycle, heartbeat, operator button and shutdown alike - now clamps to
+the **validated** `max_w` and reports whether the raw value differed, and
+`Controller.check_output_clamp()` raises `OUTPUT_CLAMP`: a persistent
+notification (`goodwe_output_clamp`), an `ERROR`, an event row and a red status
+banner. **Not a trip**: the clamped value is written and the loop keeps running,
+per 11.2's "clamp *and alarm*".
+
+**The bound is `max_w` and nothing is `min()`ed at runtime.** Section 14 requires
+the loop clamp and the driver's `setpoint_limit_w` to agree, validated at startup
+with a refusal on mismatch - which CFG-01 already ships. The device range is
+still applied second, as defence in depth; after a successful validation it
+cannot engage, so if it ever does, that is louder than the first stage and it
+alarms too, naming the bound. No new config key: the guard's limit is the one
+the operator already set, and its cadence is a module constant.
+
+**`Decision` grew `i_clamped` / `clamped` booleans, and that is the load-bearing
+part.** `reason` is one string that later stages overwrite - a cycle that clamps
+the integrator and is then slew limited reports `slew-limited`, and one that
+clamps while saturated reports `saturated-freeze`. Both *are* the winding case,
+so an alarm keyed on the string is silent on exactly the cycles it exists for:
+mutating main.py back onto `"clamp" in decision.reason` drops the 10.3 replay
+from 2 engagements to **0**. The flags are set where the clamp happens and are
+never overwritten. On a default install (`integrator_max_w` following `max_w`)
+the integrator bound is the one that bites and the output clamp is unreachable,
+so the integrator flag is the whole signal.
+
+**Distinct from the saturation freeze, and checked in the negative.** A clamp is
+a *software* bug signal; saturation is the *hardware* failing to follow a legal
+command, and its action is to freeze the integrator, not to alarm. They share no
+reason code, notification id, status row or counter. `test_clamp.py` drives a
+loop that saturates for 10 consecutive cycles at 1.4 % of the rating and asserts
+zero clamp notifications, zero engagements, no log row and no status row - after
+first asserting that the saturation freeze *really did* engage, so the silence is
+an observation rather than a scenario that did nothing.
+
+**The action log is coalesced per episode, and that was a real bug.**
+`log_event()` keeps 40 rows; one row per engagement flushed the entire ring
+within 40 cycles of a winding loop, and the first thing it evicted was
+SAFETY-18's own `METER FROZEN?` row - this guard silencing another guard's log.
+Caught by `test_freeze.py` going red, not by inspection. One row per episode now,
+carrying the exact engagement count read off the counter.
+
+**A non-finite setpoint is refused, and it was worse than passing one through.**
+`nan` is loss of input (11.1 rule 1: command zero, keep commanding), not an
+out-of-range value. Measured: because every comparison against `nan` is False, a
+naive `max(-limit, min(limit, nan))` keeps its *first* argument and turns a `nan`
+into a command **at the full rating**, then reports it as a clamp engagement.
+
+**Review fixes (PR #12).** Three, and the first was a behavioural regression
+this PR introduced: `clamp_to_rating()` **returned early** from the `max_w`
+stage, so the device bound - the "applied second, defence in depth" half - was
+reached only when the rating bound had NOT fired. The bigger the overshoot, the
+less the guard did. At `max_w` 5000 against a +-4000 device, +9999 emitted
+**+5000 W, outside the device range**, while a mere +4500 was correctly held to
++4000. That undid the pre-PR guarantee "never send a value the device will
+reject outright" and undercut this module's own case for clamping to `max_w`
+rather than a runtime `min()`, which rests entirely on that stage catching a
+validation that did not run. Both stages now always run, the result names both
+bounds when both bite (`max_w+device`), and the property is asserted over a
+492-point sweep rather than on witness values - the defect was invisible to any
+single-value check that happened to sit inside one stage. No reachable
+production path was found for it (two validation gates and the arbiter bound sit
+in front), but latent is not harmless.
+
+Two harness defects, both of the kind where a check cannot fail for the value it
+forbids: the hysteresis check issued `range(CLEAR_CYCLES - 1)` clean cycles,
+which at `CLEAR_CYCLES = 1` issues **zero** and passes while printing
+"0 clean cycles do NOT clear it" - green for the chatter setting the constant
+exists to prevent. And the re-assert check stimulated at `REASSERT_S + 9`, green
+for **any** value including 3600, i.e. the hour of silence the module promises
+not to produce. Both are now written in wall-clock terms that do not move when
+the constant does: **one** clean cycle must not clear the alarm, and a loop
+pinned across the 900 s capacity-tariff quarter must speak at least 5 times and
+on no more than 20 % of cycles. `CLEAR_CYCLES = 1`, `REASSERT_S = 3600` and
+`REASSERT_S = 0` each go red.
+
+**`test_clamp.py`** (new, assert-based, no framework) covers the clamp
+arithmetic and its boundaries - exactly +-`max_w` is *not* an engagement - the
+alarm's edge/re-assert/hysteresis cadence, the 10.3 runaway replayed through the
+real `Controller.cycle()` against an unclamped transcription, the saturation
+negative, and the direct-write path. Six author mutations reproduced red,
+including "clamp kept, alarm deleted" (the historical build) and "alarm keyed on
+`reason`". Two of the check's own hypotheses about how the freeze and the clamp
+interact were **refused by the sweep** and rewritten to what it measured: a
+frozen cycle *can* clamp, and does so exactly when the integrator reverses.
+
+`test_freeze.py`'s notification assertions now filter by `notification_id`
+rather than by service - the add-on raises more than one kind of notification,
+and its railing scenarios legitimately raise this one too.
+
 **Cheap-window maintenance charging was unreachable at the shipped defaults, and
 now the config that makes it unreachable is refused (MAINT-04).** `config.yaml`
 shipped `max_w: 2000` against `maintenance_charge_w: 2500`. The cheap-window

goodwe_controller/DOCS.md

@@ -474,6 +474,61 @@ today is operational: set `max_w` no higher than the loop actually needs (the
 tuning was measured at 2000 W and a 1.78 kW load step is tracked fine there), fit
 the e-stop, and watch for the alarm.
 
+### The winding-loop alarm (the output clamp)
+
+Every setpoint this add-on writes is clamped to `max_w` before it leaves the
+process, and **when that clamp actually engages, it alarms.** The clamp on its
+own is not the feature - this project already clipped, and the clipping is what
+hid the defect the guard exists for: with the inverter switched off entirely an
+unclamped integrator climbed ~130 W per 4 s past 10 kW and, pinned at the 5 kW
+rail, reported **14 768 W**. Clipped silently, that is a flat, legal-looking
+5000 W hold on every dashboard, and nothing anywhere says the loop asked for
+three times it.
+
+**What "the rating" is.** The bound is `max_w`, and only `max_w`. It is already
+validated against the setpoint device's own range at startup - the add-on
+*refuses to start* if `max_w` exceeds what the device accepts (and again once
+ESPHome is up and the real rating is known), so by the time the loop runs the two
+numbers agree. The device range is applied a second time inside the write as
+defence in depth; after a successful validation it cannot bite, and if it ever
+does, that engagement alarms too and names the bound.
+
+**Clamping is not saturation, and the two never share a channel.**
+
+| | what it means | what happens |
+|---|---|---|
+| **clamp** | the *computed* value exceeded the rating | a software bug signal - something upstream is winding. **Alarm.** |
+| **saturation** | the inverter is not following a perfectly *legal* command | a hardware condition - freeze the integrator. **No alarm.** |
+
+A loop can be legally saturated all day: a cold pack, a curtailed inverter, an
+export limit. Routing the clamp through a saturation-shaped alarm puts the bug
+signal inside the noise, which is the same silence as clipping quietly. They
+share no reason code, no notification id, no status row and no counter.
+
+**It is an alarm, not a trip.** The clamped value is written and the loop keeps
+running - this is not the sign-disagreement failsafe, which commands 0 W. You
+get a persistent notification (`goodwe_output_clamp`, replaced rather than
+stacked), an `ERROR` in the log with the reason code `OUTPUT_CLAMP`, a row in
+the event list, and a red status banner while it is engaged. Once it clears, a
+warning row keeps the engagement count so the evidence does not vanish with the
+condition. Consecutive engagements share one event row carrying the exact count,
+because the event list keeps only 40 rows and a row per cycle would flush every
+other guard's entries out of it.
+
+**There is nothing to configure**, deliberately. The bound is `max_w`, which you
+already set; giving the guard a limit of its own would be a second number that
+can disagree with the first, which is exactly the failure the startup validation
+exists to prevent. The alarm's cadence - first engagement, then re-asserted every
+30 s while it persists, cleared after 3 clean cycles - is fixed for the same
+reason `RAIL_FRAC` is: it describes the notification bus, not the plant.
+
+WARNING: **on a default install the engagement you will see is the integrator's,
+not the output's.** The control law bounds the integrator to `max_w` and *then*
+derives the command from it, so the command itself can never come out over the
+rating unless `integrator_max_w` has been deliberately raised above `max_w`.
+Both are watched. A guard that watched only the emitted value would be silent on
+every standard configuration.
+
 ### The frozen-meter alarm
 
 A meter that has died is easy — it stops answering, the sample age climbs past

goodwe_controller/app/clamp.py

@@ -0,0 +1,284 @@
+"""SAFETY-03 - the output clamp, and the alarm that makes it audible.
+
+WHAT IT CATCHES, in one sentence: a control law that computes a setpoint the
+actuator cannot deliver, which every layer below it will happily clip to
+something legal-looking while the thing that produced it goes on winding.
+
+THE CLAMP IS NOT THE DELIVERABLE. specs.md 10.3 is explicit that this project
+already clipped, and that the clipping is what hid the defect:
+
+    with the inverter switched off entirely, an unclamped integrator climbed
+    ~130 W per 4 s to over 10 kW and, pinned at the 5 kW rail, reported
+    14 768 W
+
+A clamp alone turns that into a silent, perfectly flat 5000 W hold. Every
+dashboard reads "commanded 5000 W", which is a legal number, and nothing
+anywhere says the loop asked for three times it. 11.2's watchdog row says
+"Clamp AND ALARM - the loop is winding"; the second verb is this module.
+
+------------------------------------------------------------------------------
+WHAT "THE RATING" IS HERE, AND WHY IT IS ONE NUMBER AND NOT A MIN()
+
+The card names two bounds and warns against choosing between them:
+
+  - `DriverCapabilities.setpoint_limit_w` (3.2), the inverter's own magnitude
+    rating. In this add-on that arrives as the setpoint entity's `min`/`max`
+    from `hass.limits()` - `Controller.dev_min` / `dev_max`.
+  - `optimizer.loop.clamp_w` (14), the configured loop clamp. In this add-on
+    that is the `max_w` option, which `Tuning.max_w` and the arbiter's
+    `supervised` limit both already use.
+
+14 says these two must AGREE, validated at startup with a refusal on
+mismatch - and that gate already ships: `validate_startup_config()` refuses
+`max_w > dev_max` and `-max_w < dev_min` (validate.py, problem 1), and
+run_control() re-runs it once ESPHome is up and the real rating is known.
+
+So this guard clamps to `max_w`, THE VALIDATED VALUE, and does not `min()` it
+with the device rating at runtime. That is the card's instruction and it is also
+the only reading that keeps `max_w` meaningful: a silent runtime `min()` is
+precisely the "helper bounded to +-500 W under a loop clamped to 1500 W" failure
+the 14 warning box is about.
+
+WARNING - THE DEVICE BOUND IS STILL APPLIED, SECOND, AND IT IS NOT REDUNDANT.
+After a successful validation `+-max_w` lies inside `[dev_min, dev_max]`, so the
+second stage is arithmetically a no-op and can never engage. If it DOES engage,
+the validation did not run or no longer holds - `hass.limits()` returns
+(None, None) whenever ESPHome is not up yet, and the range checks are skipped
+rather than guessed at - and that is a louder fault than the first stage, not a
+quieter one. It therefore alarms too, through the same channel, with the bound
+named. This is NOT SAFETY-20; nothing here changes what `freeze.py` measures its
+rail against.
+
+------------------------------------------------------------------------------
+WHERE THE WIND ACTUALLY SHOWS UP - READ THIS BEFORE "SIMPLIFYING" THE CALLER
+
+`control.compute()` bounds the integrator to `integrator_max_w or max_w` and
+THEN derives the command from it, so on a default install (`integrator_max_w`
+unset, i.e. following `max_w`) the command can never exceed `max_w` and the
+actuator-side clamp below IS UNREACHABLE. control.py says so in as many words:
+
+    SAFETY-03 ... must watch for THIS, not for "clamped" below.
+
+An alarm wired only to this module's `ClampResult.engaged` would therefore be
+silent on the exact configuration the whole ticket is about. The winding is
+visible one layer up, at the integrator bound, and that is why `Decision` now
+carries `i_clamped` / `clamped` BOOLEANS and the caller ORs them in here.
+
+AND WHY BOOLEANS RATHER THAN `Decision.reason`. `reason` is a single string that
+later stages overwrite: a cycle that clamps the integrator and is then slew
+limited reports "slew-limited", and one that clamps while saturated reports
+"saturated-freeze". Both are exactly the winding-into-a-rail case, and on the
+string the clamp signal is destroyed before anything can alarm on it. The two
+flags are set where the clamp happens and are never overwritten. A check in
+test_clamp.py drives that precise combination.
+
+------------------------------------------------------------------------------
+DISTINCT FROM SATURATION - THIS IS ACCEPTANCE CRITERION 2, NOT A NICETY
+
+    clamp      the COMPUTED value exceeded the rating. A software bug signal:
+               something upstream is winding. Alarm, every engagement.
+    saturation |command - inverter AC power| > `saturation_w` for
+               `saturation_cycles`. The HARDWARE cannot follow a perfectly legal
+               command. Action is "freeze the integrator" (11.2), not an alarm.
+
+A loop can be legally saturated all day - a cold pack, a curtailed inverter, an
+export limit - so a saturation-shaped alarm is one an operator learns to ignore.
+Routing the clamp through it puts the bug signal inside the noise, which is the
+same silence 10.3 is complaining about. They share no reason code, no
+notification id, no status row and no counter.
+
+------------------------------------------------------------------------------
+ANTI-FLOOD, WITHOUT GOING SILENT
+
+A genuinely winding loop clamps on EVERY cycle for as long as it winds. One HA
+notification per cycle is how an operator learns to dismiss notifications
+unread, and this add-on already made that argument once (see the fixed
+`notification_id` note in main.check_meter_frozen).
+
+So the two surfaces are split, deliberately:
+
+  - EVERY engagement is recorded, never deduplicated, and that record is what
+    says how hard the loop is pushing. 15's action log is this add-on's
+    `Controller.events`, and it gets a row from `write_setpoint()` on every
+    engaged WRITE; `engagements` here counts every engaged CYCLE. The two layers
+    are counted separately on purpose - the heartbeat re-writes an
+    already-clamped value between cycles, so a per-write count would inflate and
+    a per-cycle log would miss an out-of-cycle write.
+  - THE NOTIFICATION fires on the rising edge and then RE-ASSERTS every
+    `REASSERT_S`, so a loop pinned on the rail for an hour does not go quiet.
+  - It DE-ASSERTS only after `CLEAR_CYCLES` consecutive clean cycles, which is
+    the 10.2 hysteresis: a command sitting exactly on the rail dithers in and
+    out of engagement by one quantisation step, and without this the alarm
+    chatters.
+
+`REASSERT_S` and `CLEAR_CYCLES` are module constants, not options, for the same
+reason `freeze.RAIL_FRAC` is: they describe the event bus, not the plant, and
+every knob 10.3 asked to keep exposed is a LOOP TUNING knob and is untouched by
+this module. No new config key is added anywhere - CFG-03 already has three
+places a default has to be kept in step and this ticket does not earn a fourth.
+
+------------------------------------------------------------------------------
+NOT A TRIP
+
+Clamping does not stop the loop and does not command 0 W. The clamped value is
+written and control continues, which is what 11.2 asks for ("Clamp and alarm")
+and what distinguishes this from SAFETY-01/02, whose action is 0 W. It follows
+that this must NOT drive the failsafe status - `publish()` keeps reporting
+"running", because the controller is.
+"""
+
+import math
+from dataclasses import dataclass
+from datetime import datetime
+
+# The reason code, distinct from any saturation-freeze code by construction:
+# nothing else in this add-on uses it. See "DISTINCT FROM SATURATION".
+REASON_CODE = "OUTPUT_CLAMP"
+
+# Re-assert the notification this often while the clamp stays engaged. 30 s is
+# the planning cadence (2.3), i.e. slow enough not to flood and fast enough
+# that a pinned loop is never quiet for a whole billing quarter.
+REASSERT_S = 30.0
+
+# Consecutive clean cycles before the alarm de-asserts (10.2 hysteresis).
+CLEAR_CYCLES = 3
+
+
+@dataclass(frozen=True)
+class ClampResult:
+    """What the guard did, with the RAW value preserved.
+
+    AC 2 turns on "the raw computed value differed from the clamped one", so the
+    raw value has to survive the call - a guard that returns only the applied
+    number cannot be asked the question the acceptance criterion asks.
+    """
+
+    requested_w: float
+    applied_w: float
+    limit_w: float
+    engaged: bool
+    bound: str = ""          # "" | "max_w" | "device" - which stage bit
+
+    def describe(self) -> str:
+        return (f"{REASON_CODE}: the computed setpoint {self.requested_w:+.0f} W "
+                f"was clamped to {self.applied_w:+.0f} W by the {self.bound} "
+                f"bound (rating {self.limit_w:.0f} W) - something upstream is "
+                f"winding")
+
+
+def is_commandable(value) -> bool:
+    """False for None, NaN and the infinities.
+
+    A non-finite setpoint is LOSS OF INPUT, not an out-of-range value, and
+    11.1 rule 1 says loss of input means command zero and keep commanding. It
+    must not reach the clamp arithmetic: `min(limit, nan)` returns nan on
+    CPython, so a nan would sail straight through a naive clamp and then report
+    `engaged` True on the `applied != requested` test - because nan != nan - i.e.
+    a transport bug filed in the winding-loop counter, with a nan on the wire.
+    """
+    if value is None:
+        return False
+    try:
+        return math.isfinite(float(value))
+    except (TypeError, ValueError):
+        return False
+
+
+def clamp_to_rating(requested_w, limit_w, dev_min=None, dev_max=None) -> ClampResult:
+    """Bound a computed setpoint to the validated rating. The ONLY clamp.
+
+    `limit_w`  the validated magnitude bound - `max_w`. Symmetric, because
+               3.2 calls `setpoint_limit_w` a MAGNITUDE clamp: it binds import
+               (+) and export (-) identically.
+    `dev_min` / `dev_max`  the device's own asymmetric range, applied second and
+               expected never to bite. See the module note.
+
+    Public, and the predicate lives here rather than in the test, for the reason
+    `freeze.looks_frozen()` gives: a safety expression transcribed into its own
+    check is one copy plus a future disagreement about what the check means.
+
+    EXACTLY +-`limit_w` IS NOT AN ENGAGEMENT. The bound is inclusive; a loop
+    resting on the rail is doing what it was told, and alarming there would fire
+    on every healthy full-power hour.
+    """
+    requested = float(requested_w)
+    limit = abs(float(limit_w))
+
+    # ⚠️ BOTH STAGES ALWAYS RUN. THIS USED TO `return` FROM THE FIRST ONE, AND
+    # THAT WAS A REGRESSION AGAINST THE CODE IT REPLACED. `write_setpoint()`
+    # previously clamped to the device range unconditionally, under the comment
+    # "never send a value the device will reject outright"; short-circuiting here
+    # meant the device bound was applied only when the rating bound had NOT
+    # fired, i.e. THE BIGGER THE OVERSHOOT THE LESS THE GUARD DID. Measured on
+    # the shipped function at max_w 5000 against a device of +-4000:
+    #
+    #     +9999 -> +5000  bound='max_w'   <- outside the device range
+    #     +4500 -> +4000  bound='device'
+    #
+    # It also undercut this module's own argument for clamping to `max_w` rather
+    # than a runtime `min()`, which rests entirely on this stage catching the
+    # case where the startup validation did not run.
+    applied = max(-limit, min(limit, requested))
+    bound = "max_w" if applied != requested else ""
+
+    # Second stage. Arithmetically dead after a successful startup validation -
+    # and therefore worth alarming on precisely BECAUSE it fired.
+    if dev_min is not None and dev_max is not None:
+        bounded = max(float(dev_min), min(float(dev_max), applied))
+        if bounded != applied:
+            bound = "device" if not bound else "max_w+device"
+            applied = bounded
+
+    return ClampResult(requested, applied, limit, applied != requested, bound)
+
+
+class ClampWatchdog:
+    """One instance per controller. Counts every engagement, alarms sparsely.
+
+    `note()` is called once per write with whether the clamp bit; the caller ORs
+    in the loop's own `i_clamped` / `clamped` flags, which is where the winding
+    is actually visible on a default install (see the module note).
+    """
+
+    def __init__(self):
+        self.active = False
+        self.engagements = 0      # every engagement, never deduplicated
+        self.episodes = 0         # rising edges
+        self.since: datetime | None = None
+        self.last_alarm: datetime | None = None
+        self.reason = ""
+        self.clean_cycles = 0
+
+    def note(self, engaged: bool, reason: str, now: datetime) -> bool:
+        """Advance one cycle. True when the caller should raise the alarm.
+
+        `now` is the caller's clock, the same discipline as
+        `SignWatchdog.evaluate()` and `FreezeWatchdog.evaluate()`.
+        """
+        if engaged:
+            self.engagements += 1
+            self.clean_cycles = 0
+            self.reason = reason
+            if not self.active:
+                self.active = True
+                self.since = now
+                self.episodes += 1
+                self.last_alarm = now
+                return True
+            # Still engaged: re-assert on the slow cadence, not every cycle.
+            if (self.last_alarm is None
+                    or (now - self.last_alarm).total_seconds() >= REASSERT_S):
+                self.last_alarm = now
+                return True
+            return False
+
+        if not self.active:
+            return False
+        self.clean_cycles += 1
+        if self.clean_cycles >= CLEAR_CYCLES:
+            self.active = False
+            self.since = None
+            self.last_alarm = None
+            self.reason = ""
+            self.clean_cycles = 0
+        return False

goodwe_controller/app/control.py

@@ -64,6 +64,23 @@ class Decision:
     # and quantisation. Carry it back in as `i_w` next cycle; that is what keeps
     # it a separate quantity from the command.
     i_w: float = 0.0
+    # SAFETY-03. Did a clamp actually engage this cycle - i.e. did the RAW
+    # computed value differ from the bounded one?
+    #
+    # ⚠️ THESE EXIST BECAUSE `reason` CANNOT ANSWER THAT QUESTION. `reason` is
+    # one string and every later stage overwrites it: a cycle that clamps the
+    # integrator and is then slew limited reports "slew-limited", and one that
+    # clamps while saturated reports "saturated-freeze". Both ARE the winding
+    # case, and on the string the clamp signal is gone before anything can alarm
+    # on it. These two are written where the clamp happens and are never
+    # overwritten, so they mean exactly what they say and nothing else.
+    #
+    # `i_clamped` is the one that fires on a default install: the integrator
+    # bound is reached first, so `clamped` is unreachable unless
+    # `integrator_max_w` is deliberately set above `max_w`. Alarm on BOTH - see
+    # the comment at the bound below, and clamp.py.
+    i_clamped: bool = False
+    clamped: bool = False
 
 
 def compute(
@@ -188,7 +205,8 @@ def compute(
     # bound is not. It is what makes the worst-case unwind time finite and
     # knowable instead of a function of how long the error happened to stand.
     bounded = max(-limit, min(limit, i_w))
-    if bounded != i_w:
+    i_clamped = bounded != i_w
+    if i_clamped:
         # ⚠️ SAFETY-03 (alarm whenever the loop winds into a rail) must watch
         # for THIS, not for "clamped" below. At the default limit == max_w the
         # integrator bound is reached first and the command derived from it can
@@ -209,7 +227,8 @@ def compute(
     # --- ORDER MATTERS: clamp -> slew -> freeze ----------------------------
     # An early draft applied a floor after the clamp and let demand escape it.
     target = max(-tuning.max_w, min(tuning.max_w, want))
-    if target != want:
+    clamped = target != want
+    if clamped:
         reason = "clamped"
 
     slewed = max(prev_w - tuning.slew_w, min(prev_w + tuning.slew_w, target))
@@ -235,7 +254,8 @@ def compute(
     step = max(1, int(tuning.step_w))
     target = round(target / step) * step
 
-    return Decision(float(target), sat_count, frozen, reason, float(i_w))
+    return Decision(float(target), sat_count, frozen, reason, float(i_w),
+                    i_clamped, clamped)
 
 
 def maintenance_charge_floor(

goodwe_controller/app/main.py

@@ -36,6 +36,10 @@ from .arbiter import (
     P_LOOP, P_MAINT_SHAPE, P_MAINTENANCE, P_SAFETY, Claim, resolve,
 )
 from .control import Tuning, compute, maintenance_charge_floor, peak_at_risk
+from .clamp import (
+    REASON_CODE as CLAMP_REASON, ClampResult, ClampWatchdog, clamp_to_rating,
+    is_commandable,
+)
 from .freeze import FreezeWatchdog, FreezeWatchdogConfig, rail_w
 from .hass import HomeAssistant
 from .maintenance import IDLE, MaintConfig, Maintenance
@@ -140,6 +144,19 @@ class Controller:
         self.freeze_wd = FreezeWatchdog(
             FreezeWatchdogConfig(hold_s=float(opts.get("meter_freeze_s", 300))))
 
+        # SAFETY-03. The output clamp's alarm - see clamp.py for why the clamp
+        # itself is the cheap half.
+        #
+        # ⚠️ NO CONFIG OF ITS OWN, on purpose. The bound it clamps to is `max_w`,
+        # which is already an option, already validated against the device's
+        # rating at startup, and already used by Tuning and the arbiter. Giving
+        # this a second key would be a fourth place a limit has to be kept in
+        # step (CFG-03 already tracks three) and, worse, would recreate exactly
+        # the two-numbers-that-disagree failure the §14 warning box describes.
+        # The anti-flood cadence is a module constant for the same reason
+        # freeze.RAIL_FRAC is: it describes the event bus, not the plant.
+        self.clamp_wd = ClampWatchdog()
+
         # P1 ingestion (TEL-01). `meter_source: off` keeps the original
         # single-entity meter_entity path, so an existing install is unchanged
         # until it opts in.
@@ -171,6 +188,11 @@ class Controller:
         self.last_write_ok = None
         self.inputs_bad_since = None
         self.dev_min = self.dev_max = None
+        # SAFETY-03. What the last write's clamp did, and what the loop's own
+        # bounds did last cycle. `loop_clamped` is a (engaged, note) pair so the
+        # alarm can say WHICH bound bit without re-deriving it from a string.
+        self.clamp_last: ClampResult | None = None
+        self.loop_clamped: tuple[bool, str] = (False, "")
         self.events: list[str] = []
         self.stopping = False
 
@@ -229,11 +251,37 @@ class Controller:
         entity = self.o.get("setpoint_entity", "")
         if not entity:
             return
-        # Never send a value the device will reject outright. A rejected write is
-        # silent from the controller's point of view, and the inverter then keeps
-        # doing whatever it was already doing.
-        if self.dev_min is not None and self.dev_max is not None:
-            value = max(self.dev_min, min(self.dev_max, value))
+
+        # --- SAFETY-03: THE CLAMP, AND IT IS ON THE ONLY PATH TO THE WIRE -----
+        # Every write in this add-on comes through here - the cycle, the
+        # heartbeat, the operator's stop button and shutdown() - so no optimizer,
+        # maintenance or manual path can route around it. That placement is the
+        # first acceptance criterion, and it is why the guard is not in
+        # control.py: a guard the control law owns is a guard every other claimer
+        # bypasses.
+        #
+        # ⚠️ A NON-FINITE VALUE IS LOSS OF INPUT, NOT AN OUT-OF-RANGE ONE. §11.1
+        # rule 1: command zero and keep commanding. It must not reach the clamp
+        # arithmetic - see is_commandable() for what a nan does to it - and it
+        # must not be counted as a clamp engagement, or a transport bug lands in
+        # the winding-loop counter.
+        if not is_commandable(value):
+            self.log_event(f"non-finite setpoint {value!r} - commanding 0 W")
+            _LOG.error("refusing to write a non-finite setpoint %r - commanding "
+                       "0 W and continuing to command. This is a bug upstream, "
+                       "not a limit.", value)
+            value = 0.0
+        result = clamp_to_rating(value, float(self.o.get("max_w", 2000)),
+                                 self.dev_min, self.dev_max)
+        self.clamp_last = result
+        if result.engaged:
+            # ⚠️ EVERY engagement, never deduplicated: this is the action log
+            # (§15), and it is a different surface from the notification, which
+            # IS deduplicated. The count is what says how hard the loop pushed;
+            # the notification only says that it did.
+            self.log_event(result.describe())
+        value = result.applied_w
+
         changed = (self.last_write is None or value != self.last_write)
         if not (changed or force):
             return
@@ -379,11 +427,29 @@ class Controller:
             self.sat_count = decision.sat_count
             self.i_w = decision.i_w
             self.loop_w = decision.target_w
+            # --- SAFETY-03 ---------------------------------------------------
+            # ⚠️ READ THE BOOLEANS, NOT `decision.reason`. On a default install
+            # the integrator bound is what bites and the output clamp is
+            # unreachable (control.py says so), and `reason` is then overwritten
+            # by whichever of slew-limiting or the saturation freeze ran after
+            # it - so a string test is silent on the two cases a winding loop
+            # most often presents as. These flags are set where the clamp
+            # happens and nothing downstream touches them.
+            self.loop_clamped = (
+                decision.i_clamped or decision.clamped,
+                (f"the loop asked for {decision.i_w:+.0f} W"
+                 + (" (integrator bound)" if decision.i_clamped else "")
+                 + (" (output bound)" if decision.clamped else "")),
+            )
             claims.append(Claim.set("loop", P_LOOP, decision.target_w, decision.reason))
         else:
             # Stopped or blind: no accumulation may survive the outage.
             self.i_w = 0.0
             self.loop_w = None
+            # A loop that is not running cannot be winding. Clearing this here
+            # rather than letting the last value stand is what makes the alarm
+            # de-assert when control is stopped.
+            self.loop_clamped = (False, "")
 
         resolution = resolve(claims)
         if resolution.contradiction:
@@ -406,14 +472,117 @@ class Controller:
         self.reason = resolution.explain()
         await self.write_setpoint(self.target)
 
-        # --- SAFETY-18, and it runs LAST on purpose --------------------------
-        # Both terms of the conjunction describe the state that now exists: the
-        # command that has just gone on the wire, and how long the meter that
-        # chose it has been standing still. Evaluated before write_setpoint()
-        # it would be reasoning about the PREVIOUS command against THIS cycle's
-        # meter - the one combination that is true of neither moment.
+        # --- SAFETY-03, and its boundary is the WRITE ------------------------
+        # ⚠️ NOT THE SAME BOUNDARY AS SAFETY-18's BELOW, THOUGH THE CALLS ABUT.
+        # This one reads `self.clamp_last`, which is None until write_setpoint()
+        # has actually run, so it has to follow the write. The one below grades
+        # `self.target`, which is assigned further up - see check_output_clamp()
+        # and check_meter_frozen() for each. Two guards, two orderings, adjacent
+        # lines: do not "tidy" them into one rule.
+        await self.check_output_clamp()
+
+        # --- SAFETY-18, and its boundary is the ASSIGNMENT --------------------
+        # ⚠️ THE BOUNDARY IS `self.target = resolution.target_w` ABOVE, NOT
+        # write_setpoint(). This comment used to say "the command that has just
+        # gone on the wire", which is wrong twice over and was corrected by
+        # TEST-06's reviewer: `check_meter_frozen()` grades `self.target` - the
+        # arbiter's resolved intent - and `write_setpoint()` clamps only a local
+        # `value` and can skip the write entirely, so that number may never have
+        # gone anywhere.
+        #
+        # What both terms of the conjunction must describe is the SAME moment:
+        # the command this cycle resolved, against how long the meter that chose
+        # it has been standing still. Evaluated before
+        # `self.target = resolution.target_w` it would grade the PREVIOUS
+        # cycle's command against THIS cycle's meter - the one combination that
+        # was true of neither moment. Moving it between the assignment and
+        # write_setpoint() is therefore harmless; moving it above the assignment
+        # is not, and test_freeze.py part 5 fails on exactly that.
         await self.check_meter_frozen()
 
+    async def check_output_clamp(self) -> None:
+        """SAFETY-03. Alarm whenever a clamp actually engaged this cycle.
+
+        Runs AFTER write_setpoint() for the same reason SAFETY-18 does: the
+        clamp is arithmetic done inside the write, so before the write there is
+        nothing yet to report.
+
+        ⚠️ PUBLISHING IS NOT IN THE WRITE PATH. The clamp itself is synchronous
+        inside write_setpoint(); this - the awaits, the notification, anything
+        that can block on Home Assistant - is out here. A stalled supervisor must
+        never be able to delay or drop a frame.
+
+        ⚠️ THIS IS NOT A TRIP. No claim is filed, nothing is latched, 0 W is not
+        commanded and `publish()` still reports "running". §11.2's action for
+        this row is "clamp and alarm", and the clamped value is written and the
+        loop keeps running. Turning this into a trip means coming here and
+        changing that deliberately, not discovering it in the field.
+        """
+        engaged_write = bool(self.clamp_last and self.clamp_last.engaged)
+        engaged_loop, loop_note = self.loop_clamped
+        engaged = engaged_write or engaged_loop
+        if engaged:
+            where = (self.clamp_last.describe() if engaged_write
+                     else f"{CLAMP_REASON}: {loop_note}, above the "
+                          f"{float(self.o.get('max_w', 2000)):.0f} W rating, and "
+                          f"was clamped to {self.target:+.0f} W")
+            reason = (f"{where}. This is a BUG SIGNAL, not a limit being used: "
+                      f"something upstream computed a setpoint the actuator "
+                      f"cannot deliver.")
+        else:
+            reason = ""
+        fired = self.clamp_wd.note(engaged, reason, datetime.now(timezone.utc))
+
+        # --- THE ACTION LOG, AND IT IS COALESCED PER EPISODE ------------------
+        # Every engagement is recorded, but consecutive ones are recorded by
+        # UPDATING one row's count rather than inserting a row each.
+        #
+        # ⚠️ THAT IS NOT COSMETIC, AND IT WAS FOUND THE HARD WAY. `log_event()`
+        # keeps 40 rows; a winding loop clamps on every cycle, so a row per
+        # engagement flushes the whole ring within 40 cycles - and the first
+        # thing it evicted, in test_freeze.py, was SAFETY-18's "METER FROZEN?"
+        # row. A guard whose log drowns out the other guards' logs has moved the
+        # flood rather than removed it. The count here is `clamp_wd.engagements`,
+        # so nothing is lost: the row says exactly how many, and the number is
+        # read off the counter rather than off the number of rows.
+        #
+        # This lives here rather than only in write_setpoint() because on a
+        # default install the actuator-side clamp is unreachable - compute() has
+        # already bounded the integrator - so the only engagements that exist are
+        # loop-side ones. Measured: 60 cycles of the §10.3 runaway produce 0
+        # write-path engagements and 2 loop-side ones.
+        if engaged and not engaged_write:
+            tag = f"{CLAMP_REASON} episode {self.clamp_wd.episodes}"
+            row = (f"{tag}: {self.clamp_wd.engagements} engagement(s) - "
+                   f"{reason}")
+            if self.events and tag in self.events[0]:
+                stamp = self.events[0].split(" ", 1)[0]
+                self.events[0] = f"{stamp} {row}"
+            else:
+                self.log_event(row)
+        if not fired:
+            return
+        _LOG.error(
+            "SAFETY-03 - %s Silently clipping this is how a winding integrator "
+            "reached 14 768 W on this project before: the clamp makes every "
+            "dashboard read a legal number while the thing producing it goes on "
+            "climbing. The command IS clamped and the loop is STILL RUNNING - "
+            "this is an alarm, not a trip. Engagements so far: %d.",
+            self.clamp_wd.reason, self.clamp_wd.engagements)
+        # A fixed notification_id, for the reason check_meter_frozen() gives -
+        # and a DIFFERENT one from every other alarm this add-on raises, because
+        # a clamp is not a frozen meter and neither is a saturation freeze.
+        await self.hass.call("persistent_notification", "create", {
+            "notification_id": "goodwe_output_clamp",
+            "title": "GoodWe: a computed setpoint exceeded the rating",
+            "message": (
+                f"{self.clamp_wd.reason}\n\nThe command has been clamped and "
+                f"the controller is STILL RUNNING. This is NOT the inverter "
+                f"failing to follow a legal command (that is the saturation "
+                f"freeze, which is normal and does not alarm) - the value asked "
+                f"for was itself out of range."),
+        })
+
     async def check_meter_frozen(self) -> None:
         """SAFETY-18. Alarm on a meter that answers forever without changing.
 
@@ -694,6 +863,23 @@ class Controller:
             out.append({"ok": True, "warn": False,
                         "text": f"setpoint {sp} (range {self.dev_min:g}…{self.dev_max:g} W)"})
 
+        # SAFETY-03. Red while a clamp is engaged, and the engagement COUNT is
+        # shown even after it clears - a loop that wound once and stopped is
+        # still a bug, and a row that vanishes with the condition takes the
+        # evidence with it. Reported only when there is something to report:
+        # unlike the freeze detector this guard has no "not covered" state, it
+        # runs on every write on every transport.
+        if self.clamp_wd.active:
+            out.append({"ok": False, "warn": False,
+                        "text": "OUTPUT CLAMP ENGAGED - " + self.clamp_wd.reason
+                                + f" Engagements: {self.clamp_wd.engagements}."})
+        elif self.clamp_wd.engagements:
+            out.append({"ok": False, "warn": True,
+                        "text": f"output clamp: not engaged now, but it has "
+                                f"engaged {self.clamp_wd.engagements} time(s) in "
+                                f"{self.clamp_wd.episodes} episode(s) since "
+                                f"start - see the event log"})
+
         if self.last_write_ok is False:
             out.append({"ok": False, "warn": False,
                         "text": "last write to the inverter was REJECTED - check the log"})

goodwe_controller/test_clamp.py

@@ -0,0 +1,550 @@
+"""Runnable check for SAFETY-03, the output clamp and its winding-loop alarm.
+
+    python3 test_clamp.py
+
+No framework, asserts only, same house style as test_freeze.py.
+
+Five parts, and parts 3 and 4 are the ones that carry the acceptance criteria:
+
+  1. THE CLAMP ARITHMETIC, on the shipped `clamp_to_rating()` - boundaries,
+     symmetry, the device second stage, and the non-finite cases. Table driven.
+  2. THE ALARM'S CADENCE - rising edge, re-assert, hysteresis - on the shipped
+     `ClampWatchdog`. Every engagement counted, notifications not flooded.
+  3. AC 1 + AC 3, WIRED. A computed setpoint that exceeds the rating, driven
+     through the REAL `Controller.cycle()`: nothing over the rating reaches the
+     device, and the alarm fires. Includes the historical runaway from
+     specs.md 10.3 replayed against an unclamped transcription, so the check
+     says what the clamp is actually worth in watts rather than asserting a
+     constant.
+  4. AC 2, AND IT IS THE LOAD-BEARING HALF. A loop that SATURATES WITHOUT
+     CLAMPING - the inverter refusing to follow a perfectly legal command - must
+     raise NO clamp alarm. If it does, the bug signal is inside the noise again
+     and the whole ticket has bought nothing.
+  5. NO PATH BYPASSES IT. write_setpoint() is the only route to the wire, so it
+     is called directly with an out-of-range value the way the heartbeat and the
+     operator's buttons call it.
+
+------------------------------------------------------------------------------
+WHY PART 4 ALSO ASSERTS ON `Decision`'s BOOLEANS AND NOT ON `reason`
+
+`Decision.reason` is one string and the stages after the clamp overwrite it. The
+check below drives a cycle that clamps the integrator AND is then slew limited,
+and asserts that `reason` does NOT say so while `i_clamped` does. That is not a
+style preference - an alarm keyed on the string is silent on exactly the cycles a
+winding loop produces, and this is the check that would go red if someone
+"simplified" main.py back onto it.
+
+------------------------------------------------------------------------------
+WHAT THIS CHECK DOES NOT COVER
+
+- The firmware-side magnitude clamp (the card's "two clamps, not one"). This
+  add-on's `write_setpoint()` is the add-on half; the ESP32 half is DRIVER-07's
+  and is not in this repo's Python.
+- A rig run. Nothing here needs one: the clamp is arithmetic on the write path
+  and part 3 drives the shipped `Controller.cycle()` against a stub Home
+  Assistant, which is the same surface test_freeze.py part 1-2 use. The wire-
+  level invariant the card asks for (`|emitted| <= rating` on every frame) is
+  asserted here over every value that reached `set_number()`, which is the last
+  thing this process controls; proving it at the inverter needs the witness log
+  and belongs with DRIVER-07.
+"""
+
+import asyncio
+import math
+import sys
+from datetime import datetime, timedelta, timezone
+
+from app.clamp import (
+    CLEAR_CYCLES, REASON_CODE, REASSERT_S, ClampWatchdog, clamp_to_rating,
+    is_commandable,
+)
+from app.control import Tuning, compute
+from app.main import Controller
+
+fails = []
+T0 = datetime(2026, 8, 30, 9, 0, 0, tzinfo=timezone.utc)
+
+# The live install's numbers. NOTES.md and freeze.py both say max_w: 5000 there,
+# which is also the device rating, so this is the bound the alarm is written at.
+LIVE_MAX_W = 5000.0
+CLAMP_ID = "goodwe_output_clamp"
+
+
+def check(name, cond):
+    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
+    if not cond:
+        fails.append(name)
+
+
+# ==============================================================================
+print("1. the clamp arithmetic, on the shipped expression")
+
+# (requested, limit, expected applied, expected engaged, expected bound)
+TABLE = [
+    (5001.0, 5000.0, 5000.0, True, "max_w"),      # +limit+1
+    (-5001.0, 5000.0, -5000.0, True, "max_w"),    # -limit-1, the export side
+    (5000.0, 5000.0, 5000.0, False, ""),          # exactly +limit: NOT engaged
+    (-5000.0, 5000.0, -5000.0, False, ""),        # exactly -limit: NOT engaged
+    (0.0, 5000.0, 0.0, False, ""),                # the failsafe value
+    (1234.0, 5000.0, 1234.0, False, ""),
+    (14768.0, 5000.0, 5000.0, True, "max_w"),     # 10.3's reported runaway
+]
+for requested, limit, want_applied, want_engaged, want_bound in TABLE:
+    r = clamp_to_rating(requested, limit)
+    check(f"{requested:+.0f} W against a {limit:.0f} W rating -> "
+          f"{r.applied_w:+.0f} W, engaged={r.engaged}, bound={r.bound!r}",
+          r.applied_w == want_applied and r.engaged is want_engaged
+          and r.bound == want_bound)
+
+# It is a MAGNITUDE clamp (3.2), so import and export must be bound identically.
+# Derived over a sweep rather than asserted on two hand-picked values.
+asym = [m for m in range(0, 12001, 37)
+        if clamp_to_rating(m, LIVE_MAX_W).applied_w
+        != -clamp_to_rating(-m, LIVE_MAX_W).applied_w]
+check(f"symmetric over {len(range(0, 12001, 37))} magnitudes "
+      f"({len(asym)} asymmetric)", not asym)
+
+# The engagement flag has to mean "the RAW value differed", which is the words
+# AC 2 uses. Re-derived from the result's own fields, not from the table.
+mismatch = [m for m in range(0, 12001, 37)
+            for r in [clamp_to_rating(m, LIVE_MAX_W)]
+            if r.engaged != (r.requested_w != r.applied_w)]
+check("engaged is exactly 'raw differs from applied', over the same sweep",
+      not mismatch)
+
+# The device second stage. Only reachable when max_w is ABOVE the device range,
+# i.e. when validate_startup_config() did not run or no longer holds.
+r = clamp_to_rating(3000.0, 5000.0, dev_min=-2000.0, dev_max=2000.0)
+check(f"a device narrower than max_w still bites, and says which bound "
+      f"({r.applied_w:+.0f} W, {r.bound!r})",
+      r.applied_w == 2000.0 and r.engaged and r.bound == "device")
+r = clamp_to_rating(3000.0, 5000.0, dev_min=-5000.0, dev_max=5000.0)
+check("...and is a no-op once the startup validation holds",
+      r.applied_w == 3000.0 and not r.engaged)
+
+# ⚠️ REGRESSION, REVIEW-FOUND. THE FIRST STAGE USED TO `return` AND THE DEVICE
+# STAGE WAS THEN NEVER REACHED, so the harder the overshoot the less the guard
+# did: at max_w 5000 against a +-4000 device, +9999 emitted +5000 - OUTSIDE THE
+# DEVICE RANGE - while a mere +4500 was correctly held to +4000. That inverts
+# the guard and it was a regression against the pre-PR `write_setpoint()`, whose
+# comment read "never send a value the device will reject outright".
+#
+# The property is asserted over a sweep, not on the two witness values, because
+# the defect was invisible to every single-value check that happened to sit
+# inside one stage.
+DEV_LO, DEV_HI = -4000.0, 4000.0
+escapes = [(m, clamp_to_rating(m, 5000.0, DEV_LO, DEV_HI).applied_w)
+           for m in range(-15000, 15001, 61)
+           for a in [clamp_to_rating(m, 5000.0, DEV_LO, DEV_HI).applied_w]
+           if not (DEV_LO <= a <= DEV_HI)]
+big = clamp_to_rating(9999.0, 5000.0, DEV_LO, DEV_HI)
+small = clamp_to_rating(4500.0, 5000.0, DEV_LO, DEV_HI)
+print(f"    max_w 5000 vs a device of {DEV_LO:+.0f}..{DEV_HI:+.0f}: "
+      f"+9999 -> {big.applied_w:+.0f} W ({big.bound}), "
+      f"+4500 -> {small.applied_w:+.0f} W ({small.bound})")
+check(f"BOTH stages always run - {big.requested_w:+.0f} W emits "
+      f"{big.applied_w:+.0f} W, inside the device's {DEV_HI:+.0f} W, and names "
+      f"both bounds ({big.bound!r})",
+      big.applied_w == DEV_HI and big.bound == "max_w+device")
+check(f"...and it holds over the whole range, so no overshoot escapes the "
+      f"device ({len(escapes)} escapes in {len(range(-15000, 15001, 61))})",
+      not escapes)
+check("...a big overshoot is not clamped LESS than a small one",
+      abs(big.applied_w) <= abs(small.applied_w) + 0.0)
+
+print("   non-finite input is loss of input, not an out-of-range value")
+for bad in (None, float("nan"), float("inf"), float("-inf"), "abc", object()):
+    check(f"is_commandable({bad!r:.30}) is False", is_commandable(bad) is False)
+for good in (0, -5000.0, 5000, 12.5):
+    check(f"is_commandable({good!r}) is True", is_commandable(good) is True)
+
+# ⚠️ WHY THAT GUARD EXISTS, AND THE ANSWER IS WORSE THAN "it passes a nan
+# through". MEASURED HERE, not recalled: every comparison against a nan is
+# False, so `min(limit, nan)` keeps its FIRST argument and a naive clamp turns a
+# nan into a command at the FULL RATING - and then reports engaged=True, because
+# limit != nan, so the winding-loop counter gets a transport bug filed in it.
+# The number below is computed by the same expression the guard replaces.
+nan = float("nan")
+naive = max(-LIVE_MAX_W, min(LIVE_MAX_W, nan))
+check(f"a naive clamp turns a nan into {naive:+,.0f} W and calls it "
+      f"engaged={naive != nan} - hence the guard",
+      naive == LIVE_MAX_W and (naive != nan) is True)
+
+
+# ==============================================================================
+print()
+print("2. the alarm's cadence: every engagement counted, no flood")
+
+wd = ClampWatchdog()
+t = T0
+fired = [wd.note(True, "winding", t + timedelta(seconds=i)) for i in range(10)]
+check(f"the rising edge fires ({fired[0]})", fired[0] is True)
+check(f"the next 9 one-second cycles fire {sum(fired[1:])} more times "
+      f"(re-assert is {REASSERT_S:.0f} s)", sum(fired[1:]) == 0)
+check(f"but every one of them is counted: engagements={wd.engagements}",
+      wd.engagements == 10)
+check(f"and it is one episode, not ten: episodes={wd.episodes}",
+      wd.episodes == 1)
+
+# --- THE RE-ASSERT CADENCE, PINNED IN BOTH DIRECTIONS ------------------------
+# ⚠️ REVIEW-FOUND. This used to read `wd.note(True, ..., T0 + REASSERT_S + 9)`,
+# which is green for ANY value of the constant - REASSERT_S = 3600 passed, and an
+# hour of silence is precisely what the module promises not to do. A check whose
+# stimulus is written in terms of the constant it is grading cannot grade it.
+#
+# The requirements are stated in wall-clock seconds and derived from what the
+# alarm is FOR, so they do not move when the constant does:
+#
+#   floor - a loop pinned on the rail must speak at least MIN_PER_QUARTER times
+#           inside the 900 s quarter that bills the capacity tariff. That window
+#           is the same one freeze.py and DOCS.md reason about.
+#   ceiling - it must not speak on more than MAX_FRACTION of cycles, or it is
+#           the per-cycle flood the split exists to remove.
+QUARTER_S = 900.0
+MIN_PER_QUARTER = 5
+MAX_FRACTION = 0.20
+
+pinned = ClampWatchdog()
+alarms = sum(1 for i in range(int(QUARTER_S))
+             if pinned.note(True, "pinned", T0 + timedelta(seconds=i)))
+print(f"    pinned on the rail for {QUARTER_S:.0f} s at 1 s cycles: "
+      f"{pinned.engagements} engagements, {alarms} alarm(s) "
+      f"({alarms / QUARTER_S:.1%} of cycles)")
+check(f"a loop pinned for a whole billing quarter speaks at least "
+      f"{MIN_PER_QUARTER} times (it spoke {alarms})",
+      alarms >= MIN_PER_QUARTER)
+check(f"...and on no more than {MAX_FRACTION:.0%} of cycles (it was "
+      f"{alarms / QUARTER_S:.1%})", alarms <= MAX_FRACTION * QUARTER_S)
+check("...while every single cycle is still counted",
+      pinned.engagements == int(QUARTER_S))
+
+# --- THE HYSTERESIS, PINNED AGAINST THE CHATTER DIRECTION --------------------
+# ⚠️ REVIEW-FOUND. This used to issue `range(CLEAR_CYCLES - 1)` clean cycles,
+# which at CLEAR_CYCLES = 1 issues ZERO of them and passes while printing
+# "0 clean cycles do NOT clear it". Green for every value >= 1 - including 1,
+# which IS the chatter the constant exists to prevent. The stimulus is a literal
+# now: ONE clean cycle must not clear the alarm, whatever the constant says.
+wd2 = ClampWatchdog()
+wd2.note(True, "winding", T0)
+wd2.note(False, "", T0 + timedelta(seconds=1))
+check(f"ONE clean cycle does NOT clear the alarm (active={wd2.active}) - "
+      f"a command dithering across the rail must not chatter",
+      wd2.active is True)
+survived = 1
+while wd2.active and survived < 50:
+    wd2.note(False, "", T0 + timedelta(seconds=1 + survived))
+    survived += 1
+print(f"    it took {survived} consecutive clean cycles to de-assert "
+      f"(CLEAR_CYCLES = {CLEAR_CYCLES})")
+check(f"...and it does de-assert, after {survived} of them, rather than "
+      f"latching", wd2.active is False and survived == CLEAR_CYCLES)
+check("...and the engagement count survives the clear, so the evidence does",
+      wd2.engagements == 1)
+check("a clean cycle on an idle watchdog says nothing",
+      wd2.note(False, "", T0 + timedelta(seconds=300)) is False)
+
+
+# ==============================================================================
+print()
+print("3. AC 1 + AC 3, wired: over the rating both clamps AND alarms")
+
+
+class _Store:
+    data = {}
+
+    def set(self, *a):
+        pass
+
+    def get_time(self, *a):
+        return None
+
+
+class _Pub:
+    enabled = False
+
+    def publish(self, *a, **k):
+        pass
+
+    def close(self):
+        pass
+
+
+class _Hass:
+    """Stub HA. `track` False models an inverter that is switched off entirely,
+    which is the 10.3 runaway's own setup."""
+
+    def __init__(self, grid, batt=0.0, track=True):
+        self.grid = grid
+        self.batt = batt
+        self.track = track
+        self.calls = []
+        self.writes = []
+
+    async def limits(self, entity):
+        return (-LIVE_MAX_W, LIVE_MAX_W, 1.0)
+
+    async def number(self, entity, invert=False):
+        return {"sensor.grid": self.grid, "sensor.soc": 50.0,
+                "sensor.batt": self.batt}.get(entity)
+
+    async def set_number(self, entity, value):
+        self.writes.append(float(value))
+        if self.track:
+            self.batt = float(value)
+        return True
+
+    async def call(self, domain, service, data):
+        self.calls.append((domain, service, data))
+        return True
+
+
+def wired(grid, cycles=40, batt=0.0, track=True, **over):
+    async def go():
+        hass = _Hass(grid, batt=batt, track=track)
+        c = Controller(dict(
+            {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
+             "soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
+             "meter_source": "off", "max_w": LIVE_MAX_W, "gain": 0.6,
+             "slew_w": 1000, "deadband_w": 15, "target_grid_w": 0,
+             "step_w": 10, "heartbeat_s": 10, "stale_input_s": 15},
+            **over), hass, _Store(), _Pub(), stop=asyncio.Event())
+        c.auto = True
+        c.dev_min, c.dev_max = -LIVE_MAX_W, LIVE_MAX_W
+        # The red status row exists only WHILE the clamp is engaged, and the
+        # hysteresis clears it a few cycles later - so it is captured live
+        # rather than looked for after the run, when it is correctly gone.
+        c.red_snapshot = []
+        for _ in range(cycles):
+            await c.read_inputs()
+            await c.cycle()
+            if c.clamp_wd.active and not c.red_snapshot:
+                c.red_snapshot = c.checks()
+        return hass, c
+    return asyncio.run(go())
+
+
+def clamp_notes(hass):
+    return [d for dom, svc, d in hass.calls
+            if (dom, svc) == ("persistent_notification", "create")
+            and d.get("notification_id") == CLAMP_ID]
+
+
+# THE 10.3 RUNAWAY, replayed. "With the inverter switched off entirely, an
+# unclamped integrator climbed ~130 W per 4 s to over 10 kW and, pinned at the
+# 5 kW rail, reported 14 768 W." `track=False` is the inverter switched off: it
+# reports 0 W however hard it is commanded, so the error never closes.
+hass_r, ctl_r = wired(6000.0, cycles=60, track=False)
+emitted = [abs(w) for w in hass_r.writes]
+
+# What the SAME error would have done to an accumulator with no bound - the
+# transcription is deliberately the unclamped velocity form 10.3 describes, so
+# the figure below is computed here and not quoted from the spec.
+runaway, err = 0.0, 6000.0
+for _ in range(60):
+    runaway += 0.6 * err
+print(f"    unclamped, the same 60 cycles reach {runaway:,.0f} W; "
+      f"the largest value that reached the device is {max(emitted):,.0f} W")
+check(f"an unbounded accumulator really does run away here "
+      f"({runaway:,.0f} W >> {LIVE_MAX_W:.0f} W)", runaway > 3 * LIVE_MAX_W)
+check(f"AC 1: NOT ONE of the {len(emitted)} values that reached the device "
+      f"exceeds the {LIVE_MAX_W:.0f} W rating",
+      emitted and max(emitted) <= LIVE_MAX_W)
+# ⚠️ AND THE CLAMP IS NOT THE ONLY THING HOLDING THAT LINE HERE, which is worth
+# saying out loud rather than letting the number above imply otherwise. Against
+# an inverter reporting 0 W the saturation freeze engages after
+# `saturation_cycles` and pins the command well BELOW the rail - that is why the
+# largest emitted value is what it is and not 5000 W. The clamp is what makes it
+# bounded; the freeze is what makes it small. A check that read
+# `max(emitted) == max_w` would have been asserting the freeze, not the clamp.
+check(f"...and the saturation freeze is what pins it at {max(emitted):,.0f} W "
+      f"rather than the rail", ctl_r.sat_count >= ctl_r.tuning.saturation_cycles)
+
+notes_r = clamp_notes(hass_r)
+print(f"    the loop clamped on {ctl_r.clamp_wd.engagements} of {60} cycles, "
+      f"in {ctl_r.clamp_wd.episodes} episode(s), raising {len(notes_r)} "
+      f"notification(s)")
+check("AC 3: and it ALARMED - silently clipping this is the defect",
+      ctl_r.clamp_wd.engagements > 0 and len(notes_r) >= 1)
+check("...with its own notification_id, not shared with any other guard",
+      all(d["notification_id"] == CLAMP_ID for d in notes_r))
+check("...one episode, re-asserted rather than one per cycle",
+      ctl_r.clamp_wd.episodes == 1 and len(notes_r) < ctl_r.clamp_wd.engagements)
+check("...and it names itself a bug signal, not a limit in use",
+      bool(notes_r) and "out of range" in notes_r[0]["message"])
+rows_r = [e for e in ctl_r.events if REASON_CODE in e]
+check(f"the action log records every engagement, coalesced into one row per "
+      f"episode ({len(rows_r)} row(s), {ctl_r.clamp_wd.engagements} "
+      f"engagement(s))",
+      len(rows_r) == ctl_r.clamp_wd.episodes
+      and f"{ctl_r.clamp_wd.engagements} engagement(s)" in rows_r[0])
+
+# ⚠️ REGRESSION, AND IT WAS A REAL BUG BEFORE IT WAS A CHECK. `log_event()` keeps
+# 40 rows. A row per engagement flushes that ring inside 40 cycles of a winding
+# loop, and the first thing it evicted was SAFETY-18's own "METER FROZEN?" row -
+# i.e. this guard silencing another guard's log. Driven here with a plant that
+# TRACKS the command, so the saturation freeze never engages and the integrator
+# clamps on essentially every cycle.
+hass_l, ctl_l = wired(9000.0, cycles=80, track=True)
+rows_l = [e for e in ctl_l.events if REASON_CODE in e]
+other_l = [e for e in ctl_l.events if REASON_CODE not in e]
+print(f"    80 winding cycles: {ctl_l.clamp_wd.engagements} engagement(s) -> "
+      f"{len(rows_l)} log row(s), leaving room for {len(other_l)} other event(s)")
+check(f"a long winding episode does NOT flood the 40-row event log "
+      f"({len(rows_l)} rows for {ctl_l.clamp_wd.engagements} engagements)",
+      ctl_l.clamp_wd.engagements > 40 and len(rows_l) <= ctl_l.clamp_wd.episodes)
+check("...and the count in the row is the counter's, so nothing is lost",
+      f"{ctl_l.clamp_wd.engagements} engagement(s)" in rows_l[-1])
+# The RED row exists only while engaged, and the episode above is short, so it
+# is captured mid-run rather than asserted after the hysteresis has cleared it.
+check("the status page goes red WHILE it is engaged",
+      any(not c["ok"] and not c["warn"] and "OUTPUT CLAMP ENGAGED" in c["text"]
+          for c in ctl_r.red_snapshot))
+check("...and once clear, a warn row still says how many times it engaged",
+      any(c["warn"] and f"{ctl_r.clamp_wd.engagements} time(s)" in c["text"]
+          for c in ctl_r.checks()))
+
+# ⚠️ NOT A TRIP. 11.2's action is "clamp and alarm"; the loop keeps running and
+# keeps writing the clamped value. If a later cut changes that it must come here.
+check(f"it is NOT a trip - the command stands at {ctl_r.target:+.0f} W, not 0 W",
+      ctl_r.target != 0.0 and ctl_r.target == hass_r.writes[-1])
+check("...and the controller still reports itself running, not failsafe",
+      ctl_r.auto and not ctl_r.sign_wd.tripped)
+
+# The export side, because the clamp is a magnitude clamp and a sign bug in it
+# would be invisible on the import case alone.
+hass_x, ctl_x = wired(-6000.0, cycles=60, track=False)
+check(f"the export side clamps and alarms too "
+      f"({min(hass_x.writes):+,.0f} W, {len(clamp_notes(hass_x))} note(s), "
+      f"{ctl_x.clamp_wd.engagements} engagement(s))",
+      min(hass_x.writes) < 0 and abs(min(hass_x.writes)) <= LIVE_MAX_W
+      and clamp_notes(hass_x) and ctl_x.clamp_wd.engagements > 0)
+
+
+# ==============================================================================
+print()
+print("4. AC 2: saturation WITHOUT clamping must NOT raise a clamp alarm")
+
+# A legal command the inverter will not follow. The meter sits just outside the
+# deadband so the loop moves a little and then rests; `batt` is pinned far from
+# whatever is commanded, which is the saturation condition (|cmd - actual| >
+# saturation_w) and nothing else. The integrator never approaches the rating.
+hass_s, ctl_s = wired(60.0, cycles=40, batt=2500.0, track=False)
+peak = max(abs(w) for w in hass_s.writes)
+print(f"    saturated for {ctl_s.sat_count} consecutive cycles; the command "
+      f"peaked at {peak:.0f} W, {peak / LIVE_MAX_W:.1%} of the rating")
+# ⚠️ THE POSITIVE HALF FIRST. A negative check that saw nothing happen is not an
+# observation - if the saturation freeze never engaged, "no clamp alarm" is
+# vacuous and this check is worthless.
+check(f"the saturation freeze REALLY DID engage (sat_count="
+      f"{ctl_s.sat_count} >= {ctl_s.tuning.saturation_cycles})",
+      ctl_s.sat_count >= ctl_s.tuning.saturation_cycles)
+check("...and the event log says so, in the saturation's own words",
+      any("saturation freeze" in e for e in ctl_s.events))
+check(f"...while the command stayed well inside the rating "
+      f"({peak:.0f} W < {LIVE_MAX_W:.0f} W)", peak < LIVE_MAX_W)
+# Now the negative, which is only worth anything because of the three above.
+check("AC 2: NO clamp notification was raised", not clamp_notes(hass_s))
+check("AC 2: the clamp counted ZERO engagements", ctl_s.clamp_wd.engagements == 0)
+check("AC 2: no OUTPUT_CLAMP row in the action log",
+      not [e for e in ctl_s.events if REASON_CODE in e])
+check("AC 2: and no clamp row on the status page",
+      not [c for c in ctl_s.checks() if "output clamp" in c["text"].lower()])
+
+print("   ...and the two alarms do not share a notification_id")
+ids_clamp = {d["notification_id"] for d in clamp_notes(hass_r)}
+ids_other = {d.get("notification_id") for dom, svc, d in hass_r.calls
+             if svc == "create"} - ids_clamp
+check(f"clamp ids {sorted(ids_clamp)} are disjoint from every other alarm's "
+      f"{sorted(x for x in ids_other if x)}", not (ids_clamp & ids_other))
+
+print("   ...and the alarm cannot be keyed on Decision.reason")
+# A cycle that clamps the integrator AND is then slew limited. `reason` reports
+# the LAST stage, so a string-keyed alarm is silent on it; the boolean is not.
+t = Tuning(max_w=2000.0, slew_w=100.0, gain=0.6)
+d = compute(prev_w=1000.0, grid_w=6000.0, actual_w=1000.0, tuning=t, i_w=1900.0)
+print(f"    reason={d.reason!r}  i_clamped={d.i_clamped}  clamped={d.clamped}")
+check("the integrator DID clamp on that cycle", d.i_clamped is True)
+check("...but `reason` says 'slew-limited' - the clamp was overwritten "
+      "downstream, so a string-keyed alarm is SILENT here",
+      d.reason == "slew-limited" and "clamp" not in d.reason)
+# And the same for the OUTPUT clamp, on a config that lets the integrator run
+# above the rail - then both flags fire and `reason` still reports neither.
+t2 = Tuning(max_w=2000.0, integrator_max_w=3000.0, slew_w=100.0, gain=0.6)
+d2 = compute(prev_w=1000.0, grid_w=6000.0, actual_w=1000.0, tuning=t2, i_w=2900.0)
+print(f"    reason={d2.reason!r}  i_clamped={d2.i_clamped}  clamped={d2.clamped}")
+check("both clamps engaged, and `reason` reports neither",
+      d2.i_clamped and d2.clamped and "clamp" not in d2.reason)
+# ⚠️ HOW THE FREEZE AND THE CLAMP INTERACT, MEASURED RATHER THAN ASSUMED, AND
+# THE MEASUREMENT REFUSED TWO GUESSES BEFORE THIS ONE. Written first as "a
+# frozen cycle never clamps" (false), then as "only the i_w == 0 reseed
+# exception clamps" (also false - at i_w == 0 with a matching actual the cycle
+# is not saturated at all, so it is never frozen). What the sweep actually
+# shows: a frozen cycle clamps exactly when the integrator REVERSES. control.py
+# permits that on purpose - "it may fall, cross zero, or reverse outright" -
+# because the alternative deadlocks the loop, and a reversal is a full swing
+# that can clear the bound on the far side in one step.
+#
+# That matters for this ticket rather than being trivia: it is the one shape
+# where the saturation freeze and a clamp engagement coincide, so it is exactly
+# where an alarm that shared a channel with saturation would lose the clamp.
+grid_sweep = (-9000.0, -3000.0, 3000.0, 9000.0)
+grid = [(pp, g) for pp in (-1900.0, -500.0, 0.0, 500.0, 1900.0)
+        for g in grid_sweep]
+runs = {(pp, g): compute(pp, g, 0.0, t, sat_count=5, i_w=pp) for pp, g in grid}
+frozen = [k for k, d in runs.items() if d.frozen]
+frozen_clamps = [k for k in frozen if runs[k].i_clamped]
+# The property, stated independently of the control law's own branches: the
+# correction pushes AGAINST the standing integrator, and the resulting swing
+# clears the bound.
+reversals = [(pp, g) for pp, g in frozen
+             if pp * g < 0 and abs(pp + t.gain * g) > t.max_w]
+print(f"    {len(frozen)} of {len(grid)} sweep cells are frozen; "
+      f"those that also clamp: {frozen_clamps}")
+check(f"a frozen cycle CAN clamp, and only by reversing "
+      f"({len(frozen_clamps)} of {len(frozen)} frozen cells)",
+      frozen_clamps and frozen_clamps == reversals)
+check("...and a frozen cycle pushing the way it already leans never clamps - "
+      "that is the freeze doing its job",
+      not [k for k in frozen if runs[k].i_clamped and k[0] * k[1] > 0])
+check("a healthy cycle sets neither flag",
+      not any([compute(0.0, 100.0, 0.0, t).i_clamped,
+               compute(0.0, 100.0, 0.0, t).clamped]))
+
+
+# ==============================================================================
+print()
+print("5. no path reaches the wire around the clamp")
+
+
+async def _direct():
+    hass = _Hass(0.0)
+    c = Controller({"setpoint_entity": "number.sp", "max_w": 2000,
+                    "meter_source": "off"},
+                   hass, _Store(), _Pub(), stop=asyncio.Event())
+    c.dev_min, c.dev_max = -LIVE_MAX_W, LIVE_MAX_W
+    # The shape the heartbeat, shutdown() and handle_action() all use.
+    await c.write_setpoint(9999.0, force=True)
+    await c.write_setpoint(-9999.0, force=True)
+    await c.write_setpoint(float("nan"), force=True)
+    await c.write_setpoint(1500.0, force=True)
+    return hass, c
+
+
+hass_d, ctl_d = asyncio.run(_direct())
+print(f"    asked for 9999 / -9999 / nan / 1500 W; the device saw "
+      f"{hass_d.writes}")
+check("a direct out-of-range write is clamped, not passed through",
+      hass_d.writes[:2] == [2000.0, -2000.0])
+check("a nan is refused and becomes 0 W, per 11.1 rule 1",
+      hass_d.writes[2] == 0.0)
+check("...and is NOT counted as a clamp engagement - it is a transport bug",
+      sum(1 for e in ctl_d.events if REASON_CODE in e) == 2)
+check("an in-range value is untouched", hass_d.writes[3] == 1500.0)
+check("every emitted frame satisfies |emitted| <= the rating",
+      all(abs(w) <= 2000.0 for w in hass_d.writes))
+
+print()
+if fails:
+    print(f"{len(fails)} FAILED: {', '.join(fails)}")
+    sys.exit(1)
+print("all checks passed")

goodwe_controller/test_freeze.py

@@ -2,7 +2,10 @@
 
     python3 test_freeze.py
 
-Three parts, and the last two are the ones that matter:
+Six parts. Parts 2 and 3 are the acceptance argument; parts 5 and 6 exist
+because TEST-06 found three mutations that survived parts 1-4 green - the
+evaluation POSITION inside cycle(), the NOT COVERED status string, and the
+shipped `meter_freeze_s` itself:
 
   1. the predicate's boundaries and the edge-trigger, table-driven;
   2. SILENT ON REAL CONVERGED DATA. `scenarios/p1-age-2026-08-26.jsonl.gz` is
@@ -18,6 +21,14 @@ Three parts, and the last two are the ones that matter:
      repo is next to this one (or `$GOODWE_SIM`), and an equivalent local stub
      otherwise. Both are exercised; the run says which one it used.
 
+  4. THE WIRING: the real `Controller.cycle()`, asserted on the service call
+     that leaves the process;
+  5. WHERE IN `cycle()` the detector is evaluated - by consequence, not by
+     source order. It runs after `write_setpoint()`, and that is what makes two
+     compositions safe (see the section for both);
+  6. WHAT THE STATUS PAGE SAYS about an uncovered transport, and the shipped
+     threshold read back out of `config.yaml`.
+
 ⚠️ Part 3 runs in WALL CLOCK - `unchanged_s` is `time.monotonic()` and the
 transport really polls - so it drives the detector at `hold_s=10`, not the
 shipped 300. The THRESHOLD is config; the MECHANISM is what is under test, and
@@ -57,6 +68,7 @@ import asyncio
 import gzip
 import json
 import os
+import re
 import subprocess
 import sys
 import time
@@ -79,6 +91,29 @@ HERE = Path(__file__).resolve().parent
 # The live install's numbers, which are the ones the exposure is written at.
 LIVE_MAX_W = 5000.0
 
+# --------------------------------------------------------------------------- #
+# THE SHIPPED THRESHOLD, READ FROM config.yaml (TEST-06 / AC3)
+# --------------------------------------------------------------------------- #
+# ⚠️ `meter_freeze_s: 300` used to appear in this file only as the literal 300,
+# retyped six times, so `300 -> 3600` in config.yaml left every check here
+# green. Everything below is now driven off the SHIPPED value, and the value
+# itself is pinned in test_validate.py the way MAINT-04 pins the other defaults.
+#
+# Regex, not PyYAML: the add-on image installs py3-aiohttp/py3-paho-mqtt only,
+# and a check that skips on ImportError is not a check.
+_CFG_TXT = (HERE / "config.yaml").read_text(encoding="utf-8")
+
+
+def _cfg_default(key):
+    """The single `  key: <number>` default line in config.yaml's options block."""
+    found = re.findall(rf"^  {key}: (-?\d+(?:\.\d+)?)\s*$", _CFG_TXT, re.M)
+    return float(found[0]) if len(found) == 1 else None
+
+
+SHIPPED_HOLD_S = _cfg_default("meter_freeze_s")
+SHIPPED_DEADBAND_W = _cfg_default("deadband_w")
+SHIPPED_MAX_W = _cfg_default("max_w")
+
 
 def check(name, cond):
     print(f"  {'ok  ' if cond else 'FAIL'} {name}")
@@ -135,6 +170,35 @@ check("the reason names both terms",
       "unchanged over 300" in wd.reason and "command over 4750" in wd.reason)
 
 
+# --------------------------------------------------------------------------- #
+# 1b. the SHIPPED threshold, and the two properties it has to have (AC3)
+# --------------------------------------------------------------------------- #
+# ⚠️ This is the number the whole ticket is an argument about, and it was pinned
+# nowhere: `meter_freeze_s: 300 -> 3600` used to leave this file and
+# test_validate.py both green. test_validate.py pins the literal in MAINT-04's
+# manner; here it is pinned by its CONSEQUENCES, so a schema-legal value that
+# gives up the argument goes red even though it is a legal value.
+print("the shipped meter_freeze_s (AC3)")
+check(f"config.yaml's numeric defaults were read (freeze {SHIPPED_HOLD_S}, "
+      f"deadband {SHIPPED_DEADBAND_W}, max_w {SHIPPED_MAX_W})",
+      None not in (SHIPPED_HOLD_S, SHIPPED_DEADBAND_W, SHIPPED_MAX_W))
+check("the code default and the shipped default are the same number - "
+      "FreezeWatchdogConfig() is a default, config.yaml is the ship",
+      FreezeWatchdogConfig().hold_s == SHIPPED_HOLD_S)
+check("this file's `tuning()` really is config.yaml's deadband, so the "
+      "mutual-exclusion arithmetic below is about the shipped law",
+      tuning().deadband_w == SHIPPED_DEADBAND_W)
+
+# The upper bound is money: config.yaml's own comment says 300 s is "still
+# inside the 900 s quarter that bills the capacity tariff". A hold that outlives
+# the quarter notifies after the peak is already billed.
+QUARTER_S = 900.0
+check(f"the alarm still lands inside the {QUARTER_S:.0f} s billing quarter "
+      f"({SHIPPED_HOLD_S:.0f} s hold)", SHIPPED_HOLD_S < QUARTER_S)
+# The lower bound is the conjunction itself - see the derivation below, which
+# now runs off the shipped numbers rather than off two retyped literals.
+
+
 # --------------------------------------------------------------------------- #
 # 2. silent on 40 minutes of real converged behaviour
 # --------------------------------------------------------------------------- #
@@ -183,10 +247,10 @@ def replay(hold_s, max_w, cycle_floor_s=15.0):
 
 
 for mw in (2000.0, LIVE_MAX_W):
-    wd_r, peak_cmd, peak_unchanged, fired = replay(300.0, mw)
+    wd_r, peak_cmd, peak_unchanged, fired = replay(SHIPPED_HOLD_S, mw)
     print(f"    max_w {mw:.0f}: worst |command| {peak_cmd:.0f} W (rail "
           f"{rail_w(mw):.0f} W), worst unchanged {peak_unchanged:.1f} s "
-          f"(hold 300 s), episodes {wd_r.episodes}")
+          f"(hold {SHIPPED_HOLD_S:.0f} s), episodes {wd_r.episodes}")
     check(f"max_w {mw:.0f}: never alarms on the converged capture", fired is None)
     check(f"max_w {mw:.0f}: and the rail term had margin to spare",
           peak_cmd < rail_w(mw))
@@ -194,7 +258,7 @@ for mw in (2000.0, LIVE_MAX_W):
 # The margin on the OTHER term, as numbers rather than as trust. Both figures
 # are quoted in config.yaml and freeze.py, so they are re-measured every run and
 # a capture swapped underneath them cannot leave the prose saying the old thing.
-_, _, worst_unchanged, _ = replay(300.0, LIVE_MAX_W)
+_, _, worst_unchanged, _ = replay(SHIPPED_HOLD_S, LIVE_MAX_W)
 gaps, last, prev = [], rows[0]["t"], None
 for r in rows:
     if prev is None or r["pwr"] != prev:
@@ -208,7 +272,8 @@ check("worst observed unchanged_s is the 8 s the docs quote",
       7.5 < worst_unchanged < 8.5)
 check("longest distinct-reading interval is the 15.4 s the docs quote",
       15.0 < max(gaps) < 16.0)
-check("300 s is at least 15x the worst of the two", 300.0 >= 15 * max(gaps))
+check(f"the shipped {SHIPPED_HOLD_S:.0f} s is at least 15x the worst of the two",
+      SHIPPED_HOLD_S >= 15 * max(gaps))
 
 # --------------------------------------------------------------------------- #
 # 2b. TEL-05's case, reconstructed, because the capture cannot contain it
@@ -226,22 +291,26 @@ check("300 s is at least 15x the worst of the two", 300.0 >= 15 * max(gaps))
 # -14.2 W) rather than invented: one hour of a perfectly healthy converged meter
 # that happens not to move.
 print("TEL-05's low-power case - a converged meter that legitimately repeats")
-tel05 = FreezeWatchdog(FreezeWatchdogConfig(hold_s=300))
+tel05 = FreezeWatchdog(FreezeWatchdogConfig(hold_s=SHIPPED_HOLD_S))
 t5 = tuning(LIVE_MAX_W)
 tgt, iw, sc = 0.0, 0.0, 0
 bare_would_fire = False
 tel05_cmd = 0.0
-for k in range(3600):
+# Long enough that the REJECTED bare threshold has to get its chance even if
+# someone raises meter_freeze_s - the mutation must be caught by the pins above,
+# not by this case going red for an incidental reason.
+TEL05_S = max(3600, int(SHIPPED_HOLD_S) + 300)
+for k in range(TEL05_S):
     if k % 15 == 0:                       # cycle_floor_s, the only trigger left
         d = compute(prev_w=tgt, grid_w=-14.2, actual_w=tgt, tuning=t5,
                     sat_count=sc, i_w=iw)
         tgt, iw, sc = d.target_w, d.i_w, d.sat_count
     tel05_cmd = max(tel05_cmd, abs(tgt))
     # The REJECTED design, written out so the case can be shown to be real.
-    bare_would_fire = bare_would_fire or float(k) >= 300
+    bare_would_fire = bare_would_fire or float(k) >= SHIPPED_HOLD_S
     tel05.evaluate(float(k), tgt, LIVE_MAX_W, T0 + timedelta(seconds=k))
-print(f"    an hour at -14.2 W: worst |command| {tel05_cmd:.0f} W, "
-      f"unchanged_s reaches 3599 s, episodes {tel05.episodes}")
+print(f"    {TEL05_S / 60:.0f} min at -14.2 W: worst |command| {tel05_cmd:.0f} W, "
+      f"unchanged_s reaches {TEL05_S - 1} s, episodes {tel05.episodes}")
 check("the bare unchanged_s threshold TEL-05 rejected WOULD fire here",
       bare_would_fire)
 check("the conjunction does not - this is what the rail term buys",
@@ -255,8 +324,11 @@ check("...and the command stayed nowhere near the rail",
 # figure is inside the deadband, so the command there is parked at 0 W. If
 # someone lowers meter_freeze_s or narrows the deadband, this is the check that
 # should stop them.
-DEADBAND_W = 15.0
-slowest_healthy_w = 3600.0 / 300.0
+DEADBAND_W = SHIPPED_DEADBAND_W          # config.yaml's, not a retyped 15
+slowest_healthy_w = 3600.0 / SHIPPED_HOLD_S
+print(f"    at the shipped {SHIPPED_HOLD_S:.0f} s hold, the slowest healthy "
+      f"meter that still looks unchanged is {slowest_healthy_w:.1f} W, against "
+      f"a {DEADBAND_W:.0f} W deadband")
 check("a healthy meter slow enough to look unchanged is inside the deadband",
       slowest_healthy_w < DEADBAND_W)
 _park = compute(prev_w=0.0, grid_w=-slowest_healthy_w, actual_w=0.0,
@@ -277,11 +349,19 @@ check("...and the same replay DOES alarm once both terms are made reachable",
 # 3. a genuinely frozen meter, over real HTTP, through the real transport
 # --------------------------------------------------------------------------- #
 def find_hwsim():
-    """The rig's own fake meter, if the infra repo is reachable."""
+    """The rig's own fake meter, if the infra repo is reachable.
+
+    ⚠️ `HERE.parents[1]` only exists when this package sits at least two levels
+    below the filesystem root - it raised IndexError and took the WHOLE suite
+    down on a checkout at `C:/goodwe_controller`, which is a failure of the
+    lookup, not of the thing being tested. Guarded, so a shallow checkout falls
+    through to the stub like any other missing sibling repo.
+    """
     cands = []
     if os.environ.get("GOODWE_SIM"):
         cands.append(Path(os.environ["GOODWE_SIM"]) / "hwsim.py")
-    cands.append(HERE.parents[1] / "goodwe" / "sim" / "hwsim.py")
+    if len(HERE.parents) > 1:
+        cands.append(HERE.parents[1] / "goodwe" / "sim" / "hwsim.py")
     for c in cands:
         if c.exists():
             return c
@@ -379,7 +459,15 @@ def with_server(argv, port, fn):
 
 
 hw = find_hwsim()
-print(f"frozen meter over HTTP - using {'the rig hwsim at ' + str(hw) if hw else 'the local stub'}")
+# ⚠️ WHICH TRANSPORT BACKED PART 3 IS PART OF THE RESULT, and it is repeated in
+# the summary at the end of the run. The two are not equivalent: pointed at the
+# real `sim/hwsim.py` the rig path failed 1 run in 3 (a 16 s window against a
+# 10 s hold, when hwsim.serve() is slow to start), while the stub is tuned never
+# to fail. A green log that does not say which one ran is a green log that can
+# have quietly substituted the weaker stand-in - the same shape as an assertion
+# that cannot fail, from the other end.
+PART3_TRANSPORT = f"the rig hwsim at {hw}" if hw else "the local stub"
+print(f"frozen meter over HTTP - using {PART3_TRANSPORT}")
 if hw:
     def argv_for(mode):
         return [HWSIM_DRIVER, "8791" if mode == "freeze" else "8792", mode,
@@ -454,11 +542,17 @@ class _Hass:
         # runaway under test never happens - which would have made this a check
         # that passes for the wrong reason.
         self.batt = 0.0
+        # ⚠️ The dead-poll-task case (AC1's second half): every input reads
+        # None, exactly as it does when the transport task has died while the
+        # controller keeps cycling.
+        self.dead = False
 
     async def limits(self, entity):
         return (-5000.0, 5000.0, 1.0)
 
     async def number(self, entity, invert=False):
+        if self.dead:
+            return None
         return {"sensor.grid": self.grid, "sensor.soc": 50.0,
                 "sensor.batt": self.batt}.get(entity)
 
@@ -479,29 +573,113 @@ class _Source:
         self.unchanged_s = unchanged_s
 
 
+# ⚠️ meter_freeze_s is deliberately NOT in here. A controller built with the key
+# absent must land on the shipped number by itself, and that is asserted below -
+# supplying it here would test this dict instead of the shipped default.
+STALE_INPUT_S = 15.0
+
+
+def _opts(**over):
+    o = {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
+         "soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
+         "meter_source": "off", "max_w": LIVE_MAX_W, "gain": 0.6,
+         "slew_w": 1000, "deadband_w": SHIPPED_DEADBAND_W,
+         "target_grid_w": -10, "step_w": 10, "heartbeat_s": 10,
+         "stale_input_s": STALE_INPUT_S}
+    o.update(over)
+    return o
+
+
+# ⚠️ TWO SURFACES, AND EVERY ASSERTION BELOW SAYS WHICH ONE IT IS ABOUT.
+# The first cut of this helper selected by SERVICE, which measured "any guard's
+# notification" while every assertion built on it was named for THIS guard's.
+# That was true only because `check_meter_frozen()` was the sole
+# `persistent_notification.create` in `app/`; SAFETY-03 (#12) adds a second one
+# (`goodwe_output_clamp`) raised by scenarios that rail the command on purpose,
+# and eight assertions here went red against it - AC1's core check included.
+#
+# Blanket id-keying is NOT the repair. Three of those eight are claims about the
+# WHOLE notification surface, and narrowing them to this guard's id would have
+# silently shrunk a claim rather than fixed a conflation - which is the defect
+# class this card exists to close. So both surfaces are available by name, and
+# each assertion picks one deliberately.
+FREEZE_NOTE_ID = "goodwe_meter_frozen"
+
+
+def all_notes_of(hass):
+    """EVERY HA persistent notification raised this run, whoever raised it."""
+    return [d for dom, svc, d in hass.calls
+            if (dom, svc) == ("persistent_notification", "create")]
+
+
+def _is_freeze_note(d):
+    """Is this notification the freeze detector's?
+
+    ⚠️ THE ID **OR** THE TITLE, and the `or` is measured rather than tidy.
+    Keyed on the id alone, renaming the id empties every list in this file and
+    SIX assertions collapse together - so the one check whose subject IS the id
+    could no longer fail on its own, in the file about assertions that cannot
+    fail. Recognising the guard by what it says as well leaves the rename
+    exactly one red: the id's own check. Measured both ways:
+
+        selector                      vs #12's app/    id renamed
+        by service (the first cut)      8 red            1 red
+        by id alone                     3 red            6 red
+        by id or title (this)           3 red            1 red
+
+    The union cannot re-admit another guard's alarm unless that guard also
+    calls itself frozen, which would be a confusable alarm on its own account.
+    """
+    return (d.get("notification_id") == FREEZE_NOTE_ID
+            or "frozen" in str(d.get("title", "")).lower())
+
+
+def notes_of(hass):
+    """Only THIS guard's - see _is_freeze_note()."""
+    return [d for d in all_notes_of(hass) if _is_freeze_note(d)]
+
+
+def notes_by_text_of(hass):
+    """This guard's, recognised by what it SAYS and never by its id.
+
+    Exists for the two assertions whose subject IS the id: selected by the id,
+    both would be true by construction.
+    """
+    return [d for d in all_notes_of(hass)
+            if "frozen" in str(d.get("title", "")).lower()]
+
+
 def wired(unchanged_s, grid, cycles=60, **over):
+    """Returns (hass, controller, trace).
+
+    `trace` is one row per cycle - (command after the cycle, the watchdog's
+    level, notifications raised so far) - which is what AC1 needs: the
+    QUESTION IS NOT WHETHER THE ALARM HAPPENS BUT ON WHICH CYCLE.
+    """
     async def go():
         hass = _Hass(grid)
-        c = Controller(dict(
-            {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
-             "soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
-             "meter_source": "off", "max_w": LIVE_MAX_W, "gain": 0.6,
-             "slew_w": 1000, "deadband_w": 15, "target_grid_w": -10,
-             "step_w": 10, "heartbeat_s": 10, "stale_input_s": 15},
-            **over), hass, _Store(), _Pub(), stop=asyncio.Event())
+        c = Controller(_opts(**over), hass, _Store(), _Pub(),
+                       stop=asyncio.Event())
         c.auto = True
         c.dev_min, c.dev_max = -5000.0, 5000.0
         c.p1_source = _Source(unchanged_s)
+        trace = []
         for _ in range(cycles):
             await c.read_inputs()
             await c.cycle()
-        return hass, c
+            trace.append((c.target, c.freeze_wd.active, len(notes_of(hass))))
+        return hass, c, trace
     return asyncio.run(go())
 
 
-hass_f, ctl_f = wired(3600.0, 4000.0)
-notes = [d for dom, svc, d in hass_f.calls
-         if (dom, svc) == ("persistent_notification", "create")]
+# ⚠️ SAFETY-03 (#12) arrived at the same diagnosis independently and added a
+# `frozen_notes()` keyed on the id. `notes_of()` above subsumes it - same id,
+# plus the title fallback that keeps the id's own check independently
+# falsifiable, plus `all_notes_of()` for the claims that are about the whole
+# surface rather than about this guard.
+hass_f, ctl_f, trace_f = wired(3600.0, 4000.0)
+notes = notes_of(hass_f)                      # THIS guard's
+all_notes = all_notes_of(hass_f)              # the whole surface
 # The status row lives behind `p1_enabled`, which is what a homewizard_local
 # install has. Flipped here rather than plumbing a whole P1 transport in - the
 # row's own content is what is under test.
@@ -509,9 +687,28 @@ ctl_f.p1_enabled = True
 rows = [c for c in ctl_f.checks() if "METER MAY BE FROZEN" in c["text"]]
 print(f"    frozen at +4000 W: command ran to {ctl_f.target:.0f} W, "
       f"{len(notes)} notification(s), {len(rows)} red status row(s)")
-check("the shipped cycle() raises exactly one HA notification", len(notes) == 1)
+# SURFACE-WIDE, deliberately, and the second guard has now landed. SAFETY-03
+# ships `goodwe_output_clamp`, and this scenario - a runaway loop winding into
+# the rail - raises it legitimately: the loop asks for more than the rating and
+# is clamped. So the answer to "is this scenario still meant to raise exactly
+# one alarm" is NO, it raises exactly two, and BOTH ARE NAMED here rather than
+# the claim being narrowed until it went green again. A third guard has to come
+# back to this line, which is the whole point of keeping one assertion in the
+# file that accounts for the controller's entire alarm output.
+_ids = sorted(d.get("notification_id") for d in all_notes)
+print(f"    the whole notification surface this run: {_ids}")
+check("the whole notification surface carried exactly two alarms this run - "
+      "SAFETY-18's and SAFETY-03's, both named",
+      _ids == ["goodwe_meter_frozen", "goodwe_output_clamp"])
+# THIS GUARD'S, keyed on the id.
+check("...and the freeze detector is the one that raised it", len(notes) == 1)
+# ⚠️ Selected by TEXT, not by id - see notes_by_text_of(). One of the two
+# assertions whose subject IS the id (the other is the stability check in part
+# 5), so it must not be selected by the thing it is checking, and it has to be
+# able to fail while everything else stays green.
 check("...with a fixed notification_id so repeats replace, not stack",
-      notes and notes[0].get("notification_id") == "goodwe_meter_frozen")
+      [d.get("notification_id") for d in notes_by_text_of(hass_f)]
+      == [FREEZE_NOTE_ID])
 check("...and it says the controller is still running",
       bool(notes) and "STILL RUNNING" in notes[0]["message"])
 check("the status page carries a failed row saying so",
@@ -522,17 +719,293 @@ check("the event log names it", any("METER FROZEN" in e for e in ctl_f.events))
 check("and it does NOT command 0 W - this cut is an alarm only",
       abs(ctl_f.target) >= rail_w(LIVE_MAX_W))
 
-hass_n, ctl_n = wired(3600.0, -14.2)
-check("wired up, TEL-05's converged case still raises nothing",
-      not [1 for dom, svc, _ in hass_n.calls if svc == "create"])
-hass_m, ctl_m = wired(2.0, 8000.0)
-check("wired up, a real railed load against a moving meter raises nothing",
-      not [1 for dom, svc, _ in hass_m.calls if svc == "create"]
-      and abs(ctl_m.target) >= rail_w(LIVE_MAX_W))
-check("an uncovered transport (no unchanged_s) raises nothing, and says so",
-      not [1 for dom, svc, _ in wired(None, 4000.0)[0].calls if svc == "create"])
+# ⚠️ THE THREE SCENARIOS THAT SAFETY-03 FORCED A DECISION ON, decided one at a
+# time and not reflexively narrowed to green. Measured on the merged tree:
+#
+#   scenario                              surface  ids                    freeze
+#   frozen meter, railed command (above)     2      frozen + clamp          1
+#   TEL-05 converged, -14.2 W                0      -                       0
+#   moving meter, a real 8 kW load           1      clamp                   0
+#   uncovered transport, no unchanged_s      1      clamp                   0
+#
+# CONVERGED stays SURFACE-WIDE. At the operating point this controller exists to
+# hold, the correct behaviour is that the whole add-on has nothing to say - not
+# merely that the freeze guard has nothing to say. Nothing about SAFETY-03
+# weakens that claim, and it still holds, so narrowing it would give away a real
+# property for nothing.
+#
+# The other two NARROW to notes_of(), and the reason is the same in both: the
+# clamp alarm there is CORRECT and is SAFETY-03's subject, not this file's. A
+# genuine 8 kW load really does drive the loop past the rating; an uncovered
+# transport really does not stop it doing so. Asserting silence across the whole
+# surface would make this file fail whenever a different guard correctly does
+# its job - a claim about somebody else's behaviour wearing this file's name.
+# What each still owns is the load-bearing half: THE FREEZE DETECTOR SAID
+# NOTHING. The surface is printed rather than asserted, so a reader can still
+# see what else fired without this file grading it.
+hass_n, ctl_n, _ = wired(3600.0, -14.2)
+check("wired up, TEL-05's converged case leaves the WHOLE SURFACE silent - at "
+      "the operating point the add-on should have nothing at all to say",
+      not all_notes_of(hass_n))
+
+hass_m, ctl_m, _ = wired(2.0, 8000.0)
+print(f"    real 8 kW load, moving meter: surface "
+      f"{sorted(d.get('notification_id') for d in all_notes_of(hass_m))}, "
+      f"freeze {len(notes_of(hass_m))}")
+check("wired up, a real railed load against a moving meter raises nothing FROM "
+      "THIS GUARD (the clamp alarm there is SAFETY-03's and is correct)",
+      not notes_of(hass_m) and abs(ctl_m.target) >= rail_w(LIVE_MAX_W))
+
+hass_u = wired(None, 4000.0)[0]
+print(f"    uncovered transport: surface "
+      f"{sorted(d.get('notification_id') for d in all_notes_of(hass_u))}, "
+      f"freeze {len(notes_of(hass_u))}")
+check("an uncovered transport (no unchanged_s) raises nothing FROM THIS GUARD, "
+      "which is the honest report of a detector that cannot see",
+      not notes_of(hass_u))
+
+
+# --------------------------------------------------------------------------- #
+# 5. WHERE in cycle() the detector is evaluated (TEST-06 / AC1)
+# --------------------------------------------------------------------------- #
+# ⚠️ WHERE THE BOUNDARY ACTUALLY IS - AND IT IS NOT write_setpoint().
+# The first version of this comment said the detector runs "after
+# write_setpoint()" and grades "the command that has just gone on the wire".
+# TEST-06's reviewer disproved both halves by mutation:
+#
+#     N3  move check_meter_frozen() to immediately BEFORE write_setpoint()
+#         -> green, and CORRECTLY green
+#
+# because `self.target = resolution.target_w` is assigned two lines above the
+# write, `write_setpoint()` clamps only a local `value`, and it can skip the
+# write entirely (`if not (changed or force): return`). So what the detector
+# grades is `self.target` - the arbiter's resolved intent, PRE device clamp and
+# possibly never emitted - and the invariant these checks pin is:
+#
+#     check_meter_frozen() runs AFTER `self.target = resolution.target_w`
+#
+# That distinction is not pedantry. A check more robust than its own
+# explanation is how the next person moves the call ABOVE the assignment
+# believing they had preserved the stated rule.
+#
+# Both properties below are asserted by an OBSERVABLE CONSEQUENCE rather than by
+# source order, so semantic-preserving refactors stay green (measured by the
+# reviewer: renaming the method, and calling it from a helper that itself runs
+# last, both survive - the suite drives `await c.cycle()` and never names the
+# method):
+#
+#   (a) BOTH TERMS MUST DESCRIBE THE SAME MOMENT. The command it grades has to
+#       be the one THIS cycle resolved, judged against how long the meter that
+#       chose it has stood still. Evaluated above the assignment it grades the
+#       PREVIOUS cycle's command against THIS cycle's meter - a pairing that was
+#       never true of any moment - and the alarm arrives a whole cycle late.
+#   (b) THE DEAD-POLL-TASK COMPOSITION. When the poll task dies the inputs go
+#       stale, the arbiter resolves 0 W, and the detector - evaluated after that
+#       resolution lands in self.target - correctly sees an unrailed command and
+#       CLEARS. Evaluated above the assignment it would still be looking at the
+#       railed command from before the failsafe, and would keep the alarm
+#       standing over a controller that is already commanding 0 W.
+print("WHERE the detector is evaluated in cycle() - the ordering AC1 protects")
+
+_railed = [i for i, (t, _, _) in enumerate(trace_f)
+           if abs(t) >= rail_w(LIVE_MAX_W)]
+_alarmed = [i for i, (_, _, n) in enumerate(trace_f) if n >= 1]
+_first_railed = _railed[0] if _railed else None
+_first_alarm = _alarmed[0] if _alarmed else None
+print(f"    ramp to the rail: first railed command on cycle {_first_railed}, "
+      f"first notification on cycle {_first_alarm}, over {len(trace_f)} cycles")
+# Without this the equality below could be satisfied on cycle 0, where "the same
+# cycle" and "one cycle later" are indistinguishable.
+check("the command takes several cycles to reach the rail, so 'same cycle' and "
+      "'one cycle later' are different observations",
+      _first_railed is not None and _first_railed >= 1
+      and len(trace_f) > _first_railed + 1)
+check("the alarm lands on the SAME cycle the command reaches the rail - "
+      "evaluated after `self.target = resolution.target_w` it grades the "
+      "command THIS cycle resolved; evaluated above that assignment it grades "
+      "the previous one and arrives a cycle late",
+      _first_alarm is not None and _first_alarm == _first_railed)
+
+
+def dead_poll(after_cycles=20, dead_cycles=4, revived_cycles=20):
+    """Alarm on a railed command, then kill the poll task and keep cycling.
+
+    The real thing: the transport task dies, every input reads None, and after
+    `stale_input_s` the arbiter's P_SAFETY stop resolves 0 W. `inputs_bad_since`
+    is backdated rather than slept through - the clock is wall clock, and the
+    mechanism rather than the wait is what is under test.
+    """
+    async def go():
+        hass = _Hass(4000.0)
+        c = Controller(_opts(), hass, _Store(), _Pub(), stop=asyncio.Event())
+        c.auto = True
+        c.dev_min, c.dev_max = -5000.0, 5000.0
+        c.p1_source = _Source(3600.0)
+        for _ in range(after_cycles):
+            await c.read_inputs()
+            await c.cycle()
+        before = (c.target, c.freeze_wd.active, len(notes_of(hass)))
+        hass.dead = True
+        trace = []
+        for _ in range(dead_cycles):
+            await c.read_inputs()
+            if c.inputs_bad_since is not None:
+                c.inputs_bad_since = (datetime.now(timezone.utc)
+                                      - timedelta(seconds=STALE_INPUT_S + 1))
+            await c.cycle()
+            trace.append((c.target, c.freeze_wd.active, len(notes_of(hass))))
+        # ...and then the transport comes back, against a meter that is still
+        # frozen. The loop re-rails and this is a NEW episode - the alarm is
+        # allowed, indeed required, to speak again.
+        hass.dead = False
+        revived = []
+        for _ in range(revived_cycles):
+            await c.read_inputs()
+            await c.cycle()
+            revived.append((c.target, c.freeze_wd.active, len(notes_of(hass))))
+        # ⚠️ by TEXT: "the second episode REPLACED the first" is a claim about
+        # the id, so the list it is checked against must not be selected by the
+        # id - that would make it true by construction.
+        return before, trace, revived, notes_by_text_of(hass), c
+    return asyncio.run(go())
+
+
+_before, _dead, _revived, _freeze_notes, _ctl_d = dead_poll()
+_zeroed = [i for i, (t, _, _) in enumerate(_dead) if t == 0.0]
+print(f"    poll task killed: alarming at {_before[0]:+.0f} W "
+      f"(active={_before[1]}, {_before[2]} notification(s)) -> "
+      + ", ".join(f"cycle {i}: {t:+.0f} W active={a} notes={n}"
+                  for i, (t, a, n) in enumerate(_dead)))
+check("the episode really was standing before the poll task died - otherwise "
+      "the release below is vacuous",
+      _before[1] and _before[2] == 1 and abs(_before[0]) >= rail_w(LIVE_MAX_W))
+check("a dead poll task drives the command to 0 W",
+      bool(_zeroed) and _zeroed[0] == 0)
+check("...and on THAT SAME cycle the alarm is already clear, because the "
+      "detector ran after 0 W was resolved into self.target, not before it",
+      bool(_zeroed) and not _dead[_zeroed[0]][1])
+check("...and it stays clear for as long as the failsafe holds it there",
+      _dead and not any(a for _, a, _ in _dead))
+# ⚠️ "and it raises nothing new while parked" is NOT asserted here, and the
+# reason matters more than the deletion. It was dropped as REDUNDANT, not as
+# undecidable: TEST-06's reviewer falsified it in one line - move the
+# `hass.call` above `if not fired: return` so the detector notifies every cycle,
+# and notifications appear in exactly these parked cycles (that mutation takes
+# this file RED 9). My first note here claimed no mutation could break it, which
+# was the very inference this file exists to distrust, made about this file.
+# The deletion stands because the exact count below catches that same mutation
+# nine times over - no coverage was lost - and the honest form of the claim is
+# "I tried to break it and here is what happened", never "I could not, so it
+# cannot".
+
+# ⚠️ AND THE OTHER HALF, or "no second notification" is a claim no mutation can
+# break: when the transport comes back against a meter that is STILL frozen the
+# loop re-rails, and that is a new episode which MUST speak. What must not
+# happen is a second notification stacking up beside the first.
+_re_railed = [i for i, (t, _, _) in enumerate(_revived)
+              if abs(t) >= rail_w(LIVE_MAX_W)]
+print(f"    transport revived: command re-rails on cycle "
+      f"{_re_railed[0] if _re_railed else None}, "
+      f"{_ctl_d.freeze_wd.episodes} episode(s), {len(_freeze_notes)} "
+      f"notification(s) from this guard, "
+      f"ids {sorted({n.get('notification_id') for n in _freeze_notes})}")
+# A PRECONDITION, not a discriminating check - the same role as "the command
+# takes several cycles to reach the rail" above, and labelled the same way:
+# without it the two checks below are satisfied by a controller that never came
+# back. The crudest falsifier (never file the loop claim) takes most of this
+# file red, which is the signature of a precondition rather than a check; the
+# narrowest one measured is a stale-input stop that never releases.
+check("PRECONDITION: a revived transport against a still-frozen meter re-rails "
+      "the command",
+      bool(_re_railed))
+check("...which is a SECOND episode, and it alarms again on the cycle it rails",
+      _ctl_d.freeze_wd.episodes == 2
+      and bool(_re_railed) and _revived[_re_railed[0]][2] == _before[2] + 1)
+# ⚠️ STABILITY, not the value - the value is pinned once, above. Renaming the
+# id must fail that check and NOT this one; the two say different things and
+# each has to be able to go red on its own.
+check("...on the same notification_id, so the second REPLACES the first "
+      "instead of stacking a second one beside it",
+      len(_freeze_notes) == 2
+      and len({n.get("notification_id") for n in _freeze_notes}) == 1)
+
+
+# --------------------------------------------------------------------------- #
+# 6. what the status page SAYS about coverage (TEST-06 / AC2)
+# --------------------------------------------------------------------------- #
+# ⚠️ The check above only ever asserted that an uncovered transport is SILENT,
+# which a status page claiming `armed` on a guard that cannot run also satisfies.
+# The AC was that the detector reports uncovered transports *and says so*, so
+# the string is what gets asserted here, in both directions: the uncovered row
+# must not claim `armed`, and the covered row must still claim it - or "does not
+# say armed" would be satisfiable by deleting the armed branch outright.
+print("what the status page SAYS about coverage")
+
+
+def freeze_rows(unchanged_s, grid, **over):
+    """The freeze-detector rows of the real checks(), on a p1 install.
+
+    ⚠️ WHAT THIS DOES NOT COVER. The rows live behind `if self.p1_enabled:`, and
+    the flag is set here by hand rather than by building a real P1 transport -
+    inherited from part 4's fixture. So everything below proves "WHEN the row is
+    produced it says the right thing", and nothing below proves "the row is
+    produced on homewizard_local and not otherwise". That gate is uncovered;
+    AC2 is about the string, and widening the fixture belongs to whoever covers
+    the gate.
+    """
+    _h, ctl, _t = wired(unchanged_s, grid, cycles=3, **over)
+    # p1_enabled is what a homewizard_local install has; the row's own content
+    # is what is under test, not the transport that feeds it.
+    ctl.p1_enabled = True
+    return ctl, [r for r in ctl.checks()
+                 if r["text"].startswith("freeze detector")]
+
+
+_ctl_u, _rows_u = freeze_rows(None, 4000.0, meter_source="ha_dsmr")
+_ctl_c, _rows_c = freeze_rows(2.0, -14.2)
+print(f"    uncovered: {_rows_u[0]['text'] if _rows_u else '(no row at all)'}")
+print(f"    covered:   {_rows_c[0]['text'] if _rows_c else '(no row at all)'}")
+check("an uncovered transport gets exactly one freeze-detector row",
+      len(_rows_u) == 1)
+check("...and it says NOT COVERED",
+      bool(_rows_u) and "NOT COVERED" in _rows_u[0]["text"])
+check("...and it does NOT claim the guard is armed - that would be the status "
+      "page lying on behalf of a guard that cannot run",
+      bool(_rows_u) and "armed" not in _rows_u[0]["text"])
+check("...and it names the meter_source it is not covering, so the operator "
+      "knows which install this is",
+      bool(_rows_u) and "ha_dsmr" in _rows_u[0]["text"])
+check("...and it names homewizard_local as the transport that would be covered",
+      bool(_rows_u) and "homewizard_local" in _rows_u[0]["text"])
+check("...flagged as a warning rather than a silent ok",
+      bool(_rows_u) and not _rows_u[0]["ok"] and _rows_u[0]["warn"])
+
+check("a COVERED transport gets exactly one freeze-detector row",
+      len(_rows_c) == 1)
+check("...and that one DOES say armed - without this, 'the uncovered row does "
+      "not say armed' is satisfied by deleting the armed branch",
+      bool(_rows_c) and "armed" in _rows_c[0]["text"]
+      and "NOT COVERED" not in _rows_c[0]["text"])
+check("...and it quotes the hold and the rail it is armed at, derived from the "
+      "config the controller was actually built with",
+      bool(_rows_c)
+      and f"{_ctl_c.freeze_wd.cfg.hold_s:.0f} s unchanged" in _rows_c[0]["text"]
+      and f"{rail_w(LIVE_MAX_W):.0f} W" in _rows_c[0]["text"])
+check("...as an ok row, since a covered armed guard is not a warning",
+      bool(_rows_c) and _rows_c[0]["ok"] and not _rows_c[0]["warn"])
+
+# AC3's other half, through the built object rather than through a literal: the
+# opts dict does NOT carry meter_freeze_s, so this is the number an install that
+# leaves the key at its config.yaml default actually lands on.
+print(f"    a Controller built with meter_freeze_s ABSENT holds "
+      f"{_ctl_c.freeze_wd.cfg.hold_s:.0f} s (config.yaml ships "
+      f"{SHIPPED_HOLD_S:.0f} s)")
+check("a Controller built with the key absent lands on the SHIPPED hold, so "
+      "config.yaml and main.py's fallback cannot drift apart unnoticed",
+      _ctl_c.freeze_wd.cfg.hold_s == SHIPPED_HOLD_S)
 
 print()
+print(f"part 3 (the live HTTP transport) ran against: {PART3_TRANSPORT}")
 if fails:
     print(f"{len(fails)} FAILED: {', '.join(fails)}")
     sys.exit(1)

goodwe_controller/test_validate.py

@@ -253,6 +253,91 @@ check("...and the shipped charge rate is at or under the shipped ceiling",
 check("...and the shipped discharge rate too, so it warns about nothing",
       _shipped["maintenance_discharge_w"] <= _shipped["max_w"])
 
+# --------------------------------------------------------------------------- #
+print("the shipped meter_freeze_s, pinned (TEST-06 / AC3)")
+# ⚠️ WHY A NUMBER GETS ITS OWN CHECK. `meter_freeze_s: 300` is the threshold
+# SAFETY-18's entire argument is about, and before this it was pinned NOWHERE:
+# 300 -> 3600 left test_freeze.py AND this file green, because the schema
+# accepts int(60,3600) and no check ever read the shipped value. The schema
+# bounds what is CONFIGURABLE; this pins what is SHIPPED, which is a different
+# question and the one the docs answer.
+#
+# Read by regex like _shipped above, for the same reason: no PyYAML in the
+# image, and a check that skips on ImportError is not a check.
+_freeze_s = [float(v) for v in re.findall(
+    r"^  meter_freeze_s: (\d+(?:\.\d+)?)\s*$", _txt, re.M)]
+_freeze_schema = re.findall(r"^  meter_freeze_s: int\((\d+),(\d+)\)\s*$",
+                            _txt, re.M)
+check(f"config.yaml carries exactly one meter_freeze_s default {_freeze_s}",
+      len(_freeze_s) == 1)
+check("...and exactly one schema range for it", len(_freeze_schema) == 1)
+_hold_s = _freeze_s[0] if _freeze_s else None
+_lo, _hi = ((float(_freeze_schema[0][0]), float(_freeze_schema[0][1]))
+            if _freeze_schema else (None, None))
+check("the shipped meter_freeze_s is still the measured 300 s that freeze.py, "
+      "config.yaml's own comment and test_freeze.py are all written at",
+      _hold_s == 300.0)
+check(f"...and it is inside its own schema range ({_lo}-{_hi} s)",
+      None not in (_lo, _hi, _hold_s) and _lo <= _hold_s <= _hi)
+
+# The two properties the number has to have, DERIVED rather than restated - so
+# they go red on a value that is merely schema-legal. Both come straight from
+# config.yaml's own comment on the key.
+#
+#   upper: the alert must land inside the 900 s quarter that bills the capacity
+#          tariff, or the notification arrives after the money is spent;
+#   lower: the slowest a HEALTHY meter can tick a 1 Wh register and still look
+#          unchanged for hold_s is 3600/hold_s W, and that must stay inside the
+#          deadband or the conjunction's two terms stop being mutually
+#          exclusive (freeze.py, "WHERE hold_s COMES FROM").
+_QUARTER_S = 900.0
+_shipped_deadband = [float(v) for v in re.findall(
+    r"^  deadband_w: (\d+(?:\.\d+)?)\s*$", _txt, re.M)]
+check(f"config.yaml carries exactly one deadband_w default {_shipped_deadband}",
+      len(_shipped_deadband) == 1)
+_db = _shipped_deadband[0] if _shipped_deadband else None
+_slowest_healthy_w = 3600.0 / _hold_s if _hold_s else None
+print(f"    shipped meter_freeze_s {_hold_s} s: alert inside the "
+      f"{_QUARTER_S:.0f} s billing quarter, and a healthy meter slow enough to "
+      f"look unchanged for it is {_slowest_healthy_w} W against a "
+      f"{_db} W deadband")
+check(f"...the alarm still lands inside the {_QUARTER_S:.0f} s billing quarter",
+      _hold_s is not None and _hold_s < _QUARTER_S)
+check("...and a healthy meter that looks unchanged for it is still inside the "
+      "deadband, so the two terms stay mutually exclusive",
+      _slowest_healthy_w is not None and _db is not None
+      and _slowest_healthy_w < _db)
+
+# ⚠️ AND THE RANGE'S OWN BOUNDS. "300 sits inside the range" is worth nothing
+# while the range itself is unpinned: widened to `int(1,86400)` an operator can
+# configure `meter_freeze_s: 86400` and the guard is dead for a day at a time,
+# with every check in this repo still green. That mutation survived the first
+# cut of this PR and is what this block closes. Both numbers are argued for in
+# config.yaml's own comment and in freeze.py, and each is re-derived here:
+#
+#   60 s   the floor freeze.py calls NOT A RECOMMENDATION - there a healthy
+#          meter can look unchanged at 60 W, four times the shipped deadband, so
+#          the conjunction's two terms stop being mutually exclusive and only
+#          the margin is left;
+#   3600 s the ceiling - already four times the billing quarter, so the top of
+#          the range has given up detecting a freeze inside the quarter that
+#          bills it. It is the last bound between a supported configuration and
+#          a disabled guard, and moving it is a decision rather than a tidy-up.
+print(f"    schema range {_lo}-{_hi} s: at the floor a healthy meter that looks "
+      f"unchanged is "
+      + (f"{3600.0 / _lo:.0f} W" if _lo else "?")
+      + f" against a {_db} W deadband; the ceiling is "
+      + (f"{_hi / _QUARTER_S:.1f}x" if _hi else "?")
+      + " the billing quarter")
+check(f"the schema range itself is still the argued 60-3600 s (found "
+      f"{_lo}-{_hi})", (_lo, _hi) == (60.0, 3600.0))
+check("...a floor that gives up the mutual exclusion and keeps only the "
+      "margin, which is why config.yaml calls it not a recommendation",
+      None not in (_lo, _db) and 3600.0 / _lo > _db)
+check("...and a ceiling already past the billing quarter, so no value the "
+      "schema still accepts can alarm inside the quarter that bills it",
+      _hi is not None and _hi > _QUARTER_S)
+
 # --------------------------------------------------------------------------- #
 print("the KNOWN cross-file default divergence, pinned rather than commented")
 # ⚠️ WHAT THIS GUARDS, AND WHY IT ASSERTS THE BUG RATHER THAN THE FIX.

goodwe_controller/test_watchdog.py

@@ -274,6 +274,16 @@ from app.main import Controller  # noqa: E402
 class _Hass:
     def __init__(self):
         self.writes = []
+        self.calls = []
+
+    # ⚠️ ADDED FOR SAFETY-03, AND THE REASON IS WORTH READING. The e2e below
+    # runs a DELIBERATELY INVERTED loop (gain -0.6) into a steady 2000 W import,
+    # which is a loop that winds straight into the rail - so the output-clamp
+    # alarm fires here, and the stub not having `call` is how that was
+    # discovered. This scenario is now a second, independent witness that the
+    # clamp alarm sees a real winding loop; test_clamp.py asserts it explicitly.
+    async def call(self, domain, service, data):
+        self.calls.append((domain, service, data))
 
     async def limits(self, entity):
         return (-5000.0, 5000.0, 1.0)