SAFETY-52 review: a refusal is under 100 W, and any delivery ends it

- Engage (and seed from a stale stop) below REFUSED_BELOW_W, the probe's
  own 100 W answer, not below saturation_w: a battery derated to 100-499 W
  was held at 0 W ~93 % of the time.
- A reading of 100 W or more in the held direction ends the refusal, probe
  or not: at a 1 s meter SETTLE_CYCLES is 12 s, shorter than a healthy
  reading's lag, and that held a healthy step at 0 W until the next probe.
- test_refusal part 5b and a seed test; DOCS states the N>1 limit; the
  DECISIONS record carries the traced cause of the replay's extra trips.

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

STYLE.md

@@ -7,7 +7,7 @@ This applies to everything a home owner or installer reads: notifications, phone
 - **Body**: **one sentence** of at most 25 words, saying what happened and what to do. A **red alarm** may use up to **3 sentences** (70 words, each sentence at most 30 words), so that no action a person must take is lost.
 - **Details**: the last line links to the add-on documentation and names the section with the rest: `Details: add-on documentation, section "<title>" under Notifications.` Background, figures, reason codes and history go there or into the log, never into the body.
 - **Status row**: one sentence of at most 15 words. A **red** row may use up to 3 sentences (30 words), like a red alarm. It shows the live numbers the reader acts on and never an HA entity id. **Banner**: at most 12 words. **Refusal**: at most 30 words, naming the setting by its label and then its key, as `messages.setting()` writes it: “Inverter rating” (max_w), or “Inverter rating” for inverter-b (inverters.max_w) for one inverter in the list. **Option name**: at most 6 words, not counting the group prefix. **Option description**: at most 30 words. Where a value comes from belongs in DOCS.md.
-- **Log line** (WARNING/ERROR): one line, at most 25 words, with the numbers that matter. The reasoning goes on a DEBUG line. Log lines stay in English and are not translated. A reason text a line quotes (`%s`) keeps its reason code (`OUTPUT_CLAMP`, `STALE_READING`), because that is the evidence an installer searches for. The reason itself has the shape of a red alarm: at most 3 sentences of at most 30 words. The line as logged has no semicolon, banned word, capitals, "1 times" or "(s)". `test_messages.py` part 6 builds every such reason with the real code and checks it. A line that repeats every cycle is DEBUG. INFO lines carry no section numbers, ISO timestamps or capitals.
+- **Log line** (WARNING/ERROR): one line of at most 25 words **as logged**, with any built text filled in. Log lines stay in English and are not translated. A line that reports a safety reason gives one plain sentence and the reason's code (`STALE_READING`, `OUTPUT_CLAMP`), the evidence an installer searches for. The reason itself, with its figures and limits, goes on a DEBUG line right after it and into the action log, never onto the WARNING/ERROR line. It keeps the shape of a red alarm: at most 3 sentences of at most 30 words. The line as logged has no semicolon, banned word, capitals, "1 times" or "(s)". `test_messages.py` part 6 builds every such text with the real code and checks it, and lists the few non-safety families still over 25 words. A line that repeats every cycle is DEBUG. INFO lines carry no section numbers, ISO timestamps or capitals.
 - **Refusal length is counted as shown**: a setting placeholder counts as its label and key, and the long cases are driven through the validator (`test_messages.py` part 3).
 - **Event line** under the buttons: the plain words of `messages.hint()`, like a green row. The add-on's own action log stays technical and is not shown on the page. Installers read it as `events` in the `status` JSON.
 - **Sentences**: one idea each, active voice, present tense, with nothing nested in brackets and no semicolons (a semicolon is a second sentence in disguise). These translate well into Dutch and French.

goodwe_controller/DECISIONS/SAFETY-52-a-battery-that-refuses.md

@@ -0,0 +1,110 @@
+# SAFETY-52 - a battery that refuses is asked for 0 W
+
+**Decided 2026-09-27 by the owner, on the phase A plan (option D3). All four
+questions answered yes.**
+
+## The defect
+
+At the BMS cut-out the battery gives ~0 W whatever it is asked. The saturation
+freeze (`control.py`, `frozen`) then holds the request where it stood, and its
+SAFETY-13 exit cannot fire, because a reading of ~0 against a large ask lies
+inside the interval a limit can produce. The request stood at +5000 W for
+11.87 h on 2026-09-06 and at +1940 W for 3 h 55 m on the bus on 2026-09-19. It
+moved again only when the grid error changed sign. SAFETY-48's lead cap only
+changes where `i_w` parks (one slew step ahead instead of at the rail). When
+the request moved after the reading went flat, the stale-reading gate won
+instead and ran SAFETY-59's ~70 s trip cycle.
+
+## What was decided
+
+1. **A refused direction is held at 0 W**, reason `battery refuses`. It engages
+   in `compute()` when the freeze has held `SETTLE_CYCLES` and the reading is
+   below `REFUSED_BELOW_W` (100 W, the probe's own answer) against an ask beyond
+   `saturation_w`. The drop to 0 skips
+   the slew limit (it only removes an ask nobody meets). The direction is
+   carried like `sat_count` (`refused=`, `Decision.refused`).
+2. **A 200 W probe every 15 minutes, 34 s long** (`PROBE_W`, `PROBE_EVERY_S`,
+   `PROBE_S`; twice `BATT_OBLIGED_S`). A reading of at least 100 W in the held
+   direction ends the refusal, during a probe or not. 200 W is below `freshness.MOVE_W` (250 W), so a probe cannot arm
+   the gate. The clock is main.py's `now_mono`; `compute()` stays clockless.
+3. **A gate trip while the loop is frozen on a ~0 W reading is a refusal** for
+   control (`Controller._refusal_from_stale`). This is what stops the ~70 s
+   cycle. The log keeps SAFETY-56's two-explanation wording.
+4. **No SoC in the loop.** SAFETY-25 Option A stands. Nothing reads SoC.
+
+Other exits: the loop wants the other direction (charge is accepted at the
+floor), control stopped or started, another writer on the wire (the arbiter
+reseed). A gap in the readings keeps the refusal, because the gate's own stop is
+what seeds it.
+
+**One inverter only.** At N>1, OPT-09's `pack` limit already closes an empty
+battery's discharge end and moves its share. There, a one-unit gate trip must
+not refuse the site. `compute()` is passed `refused=None` there, and that path
+is release/1.0's law unchanged (the `i_w=None` equivalence sweep in
+`test_control.py` still holds).
+
+## Evidence
+
+`test_refusal.py` drives the real Controller and gate (the 09-06 and 09-19
+shapes, recovery with no sign change, the probe cadence, every exit).
+`tools/replay_refusal.py` replays the 16-day capture, same model as
+`replay_recorded_loop.py`, and should be read for deltas:
+
+| plant | law | gate trips | request >1 kW above delivery | longest held | import / export kWh | worst 15-min |
+|---|---|---|---|---|---|---|
+| fitted | old | 85 | 113.3 h | 16.57 h | 112.0 / 35.2 | 3434 W |
+| fitted | new | 130 (133 after review) | 11.4 h | 0.21 h | 112.3 / 35.4 | 3456 W |
+| offset 0 W | old | 1010 | 102.2 h | 16.58 h | 112.0 / 35.2 | 3434 W |
+| offset 0 W | new | 85 | 6.1 h | 0.33 h | 111.9 / 35.1 | 3434 W |
+
+**Traced in review (#141).** The +22 W worst quarter is the old law's worst
+quarter too. There the old law stands at +2060 W on a pack giving 0 W, and the
+new law asks 0 W, which the fitted plant answers with its -22 W offset: 22 W
+charged from the grid. The fit above the floor reads `batt = cmd - 28 W`, so a
+real inverter asked 0 W may do the same. Of the 71 new-law trips with no
+old-law trip within 2 min, 59 come 85-97 s after an answered probe. The offset has lifted the modelled pack
+just above its floor, the probe is answered, control ramps, the pack empties
+again mid-ramp, and the gate trips, which re-seeds the hold. The other 12
+follow an other-direction exit the same way. On a site this is one
+`STALE_READING` ERROR line per answered probe at most, and only on a battery
+that answers a probe and then cuts out again. Catching the re-cut before the
+gate needs a faster engage than `SETTLE_CYCLES`, which accuses a healthy slow
+battery, so it is left to SAFETY-58/59. After the review changes below the
+fitted-plant row reads 133 trips (the delivery exit now also catches answers
+that arrive after the probe window); the other figures are unchanged. Rejected alternatives and their figures are in
+the phase A plan: bounding the integrator by demand made the worst quarter
++541 W worse or cycled the gate 3368 times, and an SoC envelope needs the
+configured floor to be right and needs SoC.
+
+## Review changes (#141)
+
+- **"Gives nothing" is below 100 W, not below 500 W.** At `saturation_w` a
+  battery derated to 100-499 W (cold, a tapering battery) was held at 0 W for
+  ~93 % of the time: 300 W derated gave 86 Wh in 4 h against 1 200 Wh on the
+  old law. The capture has 0.11 h in that band, so the replay does not move.
+- **Any delivery ends the hold, not only an answered probe.** With
+  `meter_poll_s: 1` the 12 cycles last 12 s, less than a healthy battery's
+  reading lag (up to `BATT_OBLIGED_S`, 17 s). That engaged on a healthy step and
+  held it for 15 min. Now the late reading ends it within seconds (2 s held in
+  `test_refusal.py` part 5b).
+- **N>1 is stated as a limit in DOCS.** The pack limit there covers an empty
+  battery only when its cut-out level is configured right.
+
+## Not covered
+
+A battery that gives a steady part of the ask (e.g. a 500 W keep-alive) is a
+genuine limit (SAFETY-13 case a) and still holds its request.
+
+## What SAFETY-58 and SAFETY-59 must take into account
+
+- **SAFETY-58** (the first read after a rest trips at once): a probe never arms
+  the gate (200 < 250 W), but the refusal seeding reads the trip. If 58 delays
+  the first verdict, the seed arrives later. The 09-19 shape then needs the
+  compute-side engagement (12 cycles) to win, which it does only while the
+  request stands still. Keep `test_refusal.py` part 4 green.
+- **SAFETY-59** (DOCS says a ~70 s pattern proves a failed register): with this
+  change a refusing battery no longer produces the pattern at one inverter. The
+  DOCS text should say so, and that the `battery refuses` reason is the
+  discriminator the add-on now has. At N>1 the old pattern can still occur.
+- If either changes `MOVE_W`, `PROBE_W` must stay below it.
+  `test_refusal.py` part 2 checks the inequality.

goodwe_controller/DECISIONS/UX-08-short-notifications.md

@@ -78,3 +78,15 @@ The allowlist is empty and its cap is 0 (43 → 0). Every WARNING and ERROR line
 7. **Review: the line as logged gets the shape too.** Part 6 first checked the quoted reason alone, so a title, a 2-sentence reason and two tails still logged as 4 sentences ("Power request too high …: OUTPUT_CLAMP: … +5000 W. The command this cycle was +1000 W. Control is still running. Cut back 1 time so far."). Part 6 now also checks the whole line: at most 3 sentences, none over 30 words. The wrappers became one sentence each, the circulation and peak guard reasons two, and `STALE_READING`'s fixed limits moved from the reason to a DEBUG line beside the ERROR (its release clause stays, pinned by test_freshness). The clamp lines for one inverter are pinned at 30 words at most: `OUTPUT_CLAMP: computed +9000 W, cut to the +5000 W rating, command now +2000 W`. Several clamped inverters read as one sentence of asked/sent pairs; which limit bound each is on its own action-log row and in the notification. Totals above 30 words remain on the direction, delivery, frozen-meter and stale lines (up to 74 words). Shortening them is a follow-up: the direction and delivery reasons are also persisted in state.json, and the direction one is pinned byte-for-byte in test_watchdog.
 8. **Home Assistant 400 while it loads** now reads "Home Assistant refused number.set_value for number.x (error 400, no details). The entity may still be loading, or the value is out of range." A body with more than "400: Bad Request" is still quoted. Still WARNING, still once a minute, still a failed write.
 9. **Witness** (admin/goodwe): the resolved push says "after no answers from … to …", as the resolved card does. The YAML with every title and message stripped is identical.
+
+# UX-11: safety log lines give the code, the figures go to DEBUG
+
+**Unchanged:** every trigger, latch, threshold and alarm log level (the WARNING and ERROR lines stay), every notification id and cadence, every row colour and the banner cascade. No notification is added.
+
+1. **One plain sentence and the code.** A WARNING/ERROR line that reports a safety reason (direction, delivery, stuck meter, stale reading, peak guard, circulation, over-rating) now ends in the reason's code, taken by `messages.reason_code()`. The reason itself, figures and limits included, is on a DEBUG line logged right after it ("Evidence for inverter-a: …"), and the action log entry carries it as before. Part 6 checks that pairing in the code. Circulation and the clamp lines were not among the five named, but they break the same 25-word rule, so they follow the same pattern. The clamp numbers stay in the notification (phase 2a item 3).
+2. **The reasons and `state.json` are unchanged.** `reason` stays the evidence text, so what is saved has the same format as before. An old state file therefore loads and displays like a new one: every saved reason has opened with its code since it was first written (checked in the git history), and that code is all the ERROR line reads. The saved text goes on the DEBUG line. A stop saved without a reason shows `REASON_NOT_SAVED`. `test_messages.py` part 7 restarts from a state file holding the old long texts. `test_watchdog.py` pins the meaning (the code first, then command, reading, time and both limits, in that order) and, since the text did not change, still the bytes.
+3. **The 25-word rule is checked as logged.** Part 6 fails any WARNING+ line over 25 words with its built text filled in. Nine non-safety families still run long (tariff plan words, price feed reasons, the state migration report, write refusals and claim scope errors that only a software fault raises, maintenance, trading gate and price history reasons). They are listed with a reason in `LONG_LINE_OK`, capped at 9, and a listed family that gets shorter fails until it is removed. Shortening them is a follow-up.
+4. **The meter row's "the add-on log shows why" is now true.** When the meter goes quiet past its limit, the add-on logs one WARNING per episode, not one per cycle, and an INFO line when readings return. A meter that flaps (on `ha_signed` a healthy meter under a flat load can go 42–97 s between changes) logs at most one such pair per 10 minutes, and the episodes in between go to DEBUG. This is a new log line, not a new alarm: it has no notification and changes no decision.
+5. **The Why line past the 15 s hold** names the missing reading ("No usable meter reading for over 15 seconds: 0 W", or the battery power or level of an inverter), instead of "A reading is missing: 0 W".
+6. **Action log:** "25 engagement(s)" is "25 engagements", and "SHADOW MODE OFF - …" is "Shadow mode off: …". "SHADOW MODE ON" was not named in the card and is left as it is.
+7. **Review: after a restart the figures go on the action log too.** `log_level` defaults to `info`, so the DEBUG evidence line is not seen by default, and `events` starts empty. Each restored stop (direction per inverter and site, delivery, circulation) now also writes its saved reason to the action log ("FAILSAFE inverter-a, saved before the restart: …"), and the page's event line gets the same plain words the live stop gives. `LONG_LINE_OK` also refuses a key whose text is a safety reason, not only a `reason_code` key.

goodwe_controller/DOCS.md

@@ -526,7 +526,7 @@ accumulator is doing:
 | `Loop integrator lead` | W | that value minus **the command it was computed against**, i.e. the command the previous cycle left standing |
 | `Loop reason` | — | the law's own word for this cycle |
 
-`Loop reason` takes one of eight values:
+`Loop reason` takes one of ten values:
 
 | value | what happened |
 |---|---|
@@ -538,6 +538,27 @@ accumulator is doing:
 | `saturated-freeze` | the inverter is not following, so the command's magnitude may fall but not rise |
 | `freeze-released` | the freeze had run out of plant-shaped explanations and the command was let move again |
 | `not-running` | control is stopped **and** shadow mode is off, so the loop did not run |
+| `battery refuses` | the battery gave less than 100 W for 12 cycles (a minute at a 5-second meter) while asked for more than 500 W, so the add-on asks 0 W in that direction (see below) |
+| `battery probe` | the add-on asks 200 W to test whether a refusing battery gives power again |
+
+**When the battery refuses.** At its cut-out level the battery gives nothing,
+whatever it is asked. The add-on then asks 0 W instead of repeating its last
+request, and the `Why` row reads *The battery refuses to give power: 0 W*. Every
+15 minutes it asks 200 W for about half a minute. Control resumes as soon as
+the battery gives 100 W or more, when the grid needs charging instead, and when
+control is stopped or started. 200 W is below the 250 W the stale-reading check
+needs, so a test never trips it. A battery that gives some power, even 100 W,
+is not refusing and keeps its request.
+
+⚠️ **One inverter only.** With several inverters this hold is off. There, the
+battery level limit moves an empty battery's share to the others, but only if
+its “Battery cut-out level” matches where the battery really stops. A battery
+that refuses at another level, or for another reason (cold, a battery fault),
+can still hold its last request until the grid starts to export.
+
+*For installers:* SAFETY-52. The event log reads `BATTERY REFUSES …` and
+`BATTERY REFUSES ENDED: …`, and the INFO line ends in `BATTERY_REFUSES`.
+`DECISIONS/SAFETY-52-a-battery-that-refuses.md` has the replay figures.
 
 ⚠️ **`Loop reason` is not the `Why` row on the status page, and the two will
 often disagree.** `Why` is the *arbiter's* answer in words — *Keeping the grid
@@ -3095,8 +3116,11 @@ stop control on the add-on page.
 
 This part is for installers. The texts live in `app/messages.py` and follow
 `STYLE.md`. Every notification keeps a fixed `notification_id`, so a repeat
-replaces the card. The full figures and reason codes are on the ERROR or
-WARNING line of the add-on log.
+replaces the card. The ERROR or WARNING line of the add-on log names the
+reason code. The full figures are in `events` in the `status` JSON, also for a
+stop restored after a restart, and on the DEBUG line logged right after it.
+That DEBUG line only shows with “Log level” (log_level) set to `debug`; the
+default `info` hides it.
 
 | Title | Severity | id | Card |
 |---|---|---|---|
@@ -3223,8 +3247,9 @@ ends. That row says how to clear it.
 
 ### For installers: status page
 
-The texts are in `app/messages.py`. The full figures of any alarm are on its
-ERROR or WARNING line in the add-on log.
+The texts are in `app/messages.py`. An alarm's ERROR or WARNING line names its
+reason code, and its full figures are on the DEBUG line right after it and in
+`events` in the `status` JSON.
 
 - **Stuck-meter check: off** needs `meter_source: homewizard_local` (label
   “Where the meter reading comes from”), a `max_w` (“Inverter rating”) and a
@@ -3246,9 +3271,14 @@ ERROR or WARNING line in the add-on log.
   check** SAFETY-42.
 - **Month peak** and **15-minute periods measured** are TARIFF-01 and
   TARIFF-13. The store's state at start is the INFO line `Month peak store:
-  …`, and the peak guard's full reason is on its log line.
+  …`, and the peak guard's full reason is on the DEBUG line after its ERROR.
 - ***name*: battery power reading stopped changing** is SAFETY-37/56/57. The
-  ERROR line carries the full `STALE_READING` reason and when it clears.
+  ERROR line says when it clears and names `STALE_READING`. The full reason,
+  with both possible causes, is on the DEBUG line after it.
+- **Meter: no reading for …** (red): the add-on logs one WARNING when the
+  meter goes quiet, "No meter reading for …", and an INFO line when readings
+  return, at most once every 10 minutes. Past `stale_input_s` the **Why**
+  line names the missing reading.
 - The table at the top: **Why** is `messages.why()`, the arbiter's answer in
   words (`Resolution.explain()` stays on the log). **Tariff**, **Price
   history**, **Arbitrage** and **Shadow mode** keep every figure they had

goodwe_controller/app/control.py

@@ -265,6 +265,34 @@ _SAT_COUNT_CAP = SETTLE_CYCLES + 1                           # 13
 # `slew_limit_w()` and therefore with the rail.
 SLEW_LEAD_STEPS = 1
 
+# ---------------------------------------------------------------------------
+# SAFETY-52. A BATTERY THAT REFUSES IS ASKED FOR 0 W, NOT HELD AT ITS LAST ASK.
+#
+# At the BMS cut-out the pack delivers ~0 W whatever it is asked, so the
+# saturation freeze below holds the command where it stood (the recorded
+# 09-06 case: +5000 W for 11.87 h) until the grid error changes sign. The
+# freeze's own exit cannot help: a reading of ~0 against a large ask is inside
+# the interval a limit can produce. So a refused DIRECTION is held at 0 W,
+# carried between cycles like `sat_count`, and left again when the loop wants
+# the other direction, when a small probe is answered, or when the caller
+# resets it (stop, start, another writer). The record is DECISIONS/SAFETY-52.
+#
+# PROBE_W is below freshness.MOVE_W (250 W) on purpose: a probe that small can
+# never arm the stale-reading gate, so it cannot start the gate's ~70 s cycle.
+# test_refusal.py pins the inequality. The cadence and length are the owner's
+# 2026-09-27 answer (every 15 minutes) and freshness.BATT_OBLIGED_S twice over
+# (17 s is how long a healthy reading needs to show a command).
+REFUSED = "battery refuses"
+PROBING = "battery probe"
+PROBE_W = 200.0
+PROBE_ANSWER_W = PROBE_W / 2
+# "Gives nothing" is below the probe's own answer, not below `saturation_w`: a
+# pack derated to 100-499 W is a limit, and at 500 W it kept ~7 % of its power
+# (review of #141). The capture's refusals read -4 W.
+REFUSED_BELOW_W = PROBE_ANSWER_W
+PROBE_EVERY_S = 900.0
+PROBE_S = 34.0
+
 
 @dataclass(frozen=True)
 class Tuning:
@@ -346,6 +374,9 @@ class Decision:
     i_bound_w: float = 0.0
     requested_w: float = 0.0
     clamped_w: float = 0.0
+    # SAFETY-52. The refused direction to carry back in (+1 discharge, -1
+    # charge, 0 none), or None when the caller does not carry it.
+    refused: int | None = None
 
 
 def slew_limit_w(tuning: Tuning) -> float:
@@ -442,6 +473,8 @@ def compute(
     tuning: Tuning,
     sat_count: int = 0,
     i_w: float | None = None,
+    refused: int | None = None,
+    probe: bool = False,
 ) -> Decision:
     """One control cycle. A cycle is one meter update (~5 s on a HomeWizard P1).
 
@@ -449,6 +482,9 @@ def compute(
     `grid_w`   net grid power, + = importing
     `actual_w` what the inverter reports it is doing, + = discharging
     `i_w`      the carried integrator, or None to seed it from `prev_w`
+    `refused`  SAFETY-52's carried refused direction, or None to leave it out
+               (the `i_w=None` equivalence path does)
+    `probe`    the caller's clock says a refused direction may try PROBE_W now
     """
     reason = "tracking"
 
@@ -739,6 +775,27 @@ def compute(
             target = min(target, prev_w) if prev_w > 0 else max(target, prev_w)
             reason = "saturated-freeze"
 
+    # --- SAFETY-52: a refused direction is held at 0 W ----------------------
+    # Engages only where the freeze has held for SETTLE_CYCLES and the unit
+    # delivers ~nothing: a reading below REFUSED_BELOW_W against an ask beyond
+    # `saturation_w` is a refusal, not a slow plant. The drop to 0 skips the slew
+    # limit because it only removes an ask the battery is not meeting.
+    if refused is not None:
+        if (refused == 0 and frozen and sat_count >= SETTLE_CYCLES
+                and abs(prev_w) > tuning.saturation_w
+                and abs(actual_w) < REFUSED_BELOW_W):
+            refused = 1 if prev_w > 0 else -1
+        if refused and refused * i_w < 0:
+            refused = 0                          # the loop wants the other way
+        elif refused and refused * actual_w >= PROBE_ANSWER_W:
+            refused = 0                          # it delivers: a probe answered,
+            #                                     or a slow reading arriving
+        elif refused:
+            wants = refused * i_w > 0
+            i_w = 0.0
+            target = refused * PROBE_W if probe and wants else 0.0
+            reason = PROBING if probe and wants else REFUSED
+
     # --- quantise ----------------------------------------------------------
     # 10 W. The register is 1 W and the inverter reports at 1 W, but its
     # response lands on a coarser ladder (~17.6 W measured at ~900 W, consistent
@@ -749,7 +806,7 @@ def compute(
 
     return Decision(float(target), sat_count, frozen, reason, float(i_w),
                     i_clamped, clamped, float(i_bound_w), float(want),
-                    float(clamped_w))
+                    float(clamped_w), refused)
 
 
 def maintenance_charge_floor(

goodwe_controller/app/main.py

@@ -39,6 +39,7 @@ from .arbiter import (
     P_LOOP, P_MAINT_SHAPE, P_MAINTENANCE, P_PEAK, P_SAFETY,
     Claim, Resolution, actuator_for, resolve,
 )
+from . import control as _ctl
 from .control import Tuning, compute, maintenance_charge_floor, peak_at_risk
 from .delivery import (
     DELIVERY_OPTION_DEFAULTS, DeliveryConfig, UnitDelivery,
@@ -119,6 +120,9 @@ WRITE_BUDGET_S = 4.0
 # restart this add-on (measured on TEST-23), so a latched alarm is re-asserted
 # this often. Why: goodwe docs/DECISIONS.md, SAFETY-34.
 REANNOUNCE_S = 300.0
+# UX-11 review: a quiet meter logs at most one WARNING this often, because on
+# `ha_signed` a healthy meter under a flat load can go 42-97 s between changes.
+METER_QUIET_REPEAT_S = 600.0
 _mono = time.monotonic          # patched by the checks; asyncio keeps the real one
 
 # --------------------------------------------------------------------------- #
@@ -238,6 +242,10 @@ def why_key(winner: str, claims) -> str:
     disagreement"). A failsafe with no claim of its own is the arbiter's "no
     claim was made", or main.py's refusal to split the power ("unsplit").
     """
+    if winner == "loop":
+        # SAFETY-52: a loop holding a refused direction says so.
+        return next((c.reason for c in claims if c.source == "loop"
+                     and c.reason in (_ctl.REFUSED, _ctl.PROBING)), winner)
     if winner in ("safety", "failsafe"):
         return next((c.reason for c in claims
                      if c.source == winner and c.kind == "set"),
@@ -611,10 +619,21 @@ class Controller:
         self._grade_store_migration(store)
         for inv in self.inverters:
             if inv.sign_wd.tripped:
+                # UX-11: the plain sentence and the code. The saved reason,
+                # with its figures, is on the DEBUG line beside it.
                 _LOG.error("%s stays at 0 W after the restart because its "
                            "power went the wrong way, so find out why before "
-                           "you start control. %s.", inv.name,
+                           "you start control: %s.", inv.name,
+                           messages.reason_code(inv.sign_wd.reason))
+                _LOG.debug("Evidence for %s: %s.", inv.name,
                            inv.sign_wd.reason or "(reason not recorded)")
+                # UX-11 review: `events` starts empty and DEBUG is off by
+                # default, so the saved figures go on the action log too.
+                saved = inv.sign_wd.reason or messages.NO_REASON_SAVED
+                self.log_event(f"FAILSAFE {inv.name}, saved before the "
+                               f"restart: {saved}",
+                               hint=messages.hint("stop_restored",
+                                                  unit=inv.name))
 
         # --- SAFETY-42 + SAFETY-34 -------------------------------------------
         # ⚠️ A LATCH THAT DOES NOT ACT, WHICH IS THE ONE ASYMMETRY WITH EVERY
@@ -637,8 +656,15 @@ class Controller:
                 _LOG.error(
                     "%s is still marked as not delivering after the restart "
                     "while control keeps running, and only stopping and "
-                    "starting control clears this. %s.",
-                    inv.name, inv.delivery_wd.reason or "(reason not recorded)")
+                    "starting control clears this: %s.",
+                    inv.name, messages.reason_code(inv.delivery_wd.reason))
+                _LOG.debug("Evidence for %s: %s.", inv.name,
+                           inv.delivery_wd.reason or "(reason not recorded)")
+                saved = inv.delivery_wd.reason or messages.NO_REASON_SAVED
+                self.log_event(f"NOT DELIVERING {inv.name}, saved before the "
+                               f"restart: {saved}",
+                               hint=messages.hint("not_delivering",
+                                                  unit=inv.name))
 
         # --- SAFETY-30: THE SITE HALF OF THE SIGN CHECK ----------------------
         # ⚠️ ONE PER SITE, BESIDE THE ONE PER INVERTER, AND ARMED ONLY AT N>=2.
@@ -669,8 +695,13 @@ class Controller:
             _LOG.error(
                 "After the restart every inverter stays at 0 W, as the total "
                 "power went the wrong way, so find out why before starting "
-                "control. %s.",
-                self.site_sign_wd.reason or "(reason not recorded)")
+                "control: %s.", messages.reason_code(self.site_sign_wd.reason))
+            _LOG.debug("Evidence for the site: %s.",
+                       self.site_sign_wd.reason or "(reason not recorded)")
+            self.log_event("FAILSAFE site, saved before the restart: "
+                           + (self.site_sign_wd.reason
+                              or messages.NO_REASON_SAVED),
+                           hint=messages.hint("site_direction"))
         # And the per-unit detectors are told which question they are asking,
         # here as well as in `cycle()`. `checks()` and the status page read
         # `inv.sign_wd.cfg` and can be called before the first cycle has run, so
@@ -722,8 +753,14 @@ class Controller:
             _LOG.error(
                 "After the restart every inverter stays at 0 W, as the "
                 "inverters worked against each other, so find out why before "
-                "starting control. %s.",
-                self.circ_wd.reason or "(reason not recorded)")
+                "starting control: %s.",
+                messages.reason_code(self.circ_wd.reason))
+            _LOG.debug("Evidence for the site: %s.",
+                       self.circ_wd.reason or "(reason not recorded)")
+            self.log_event("CIRCULATING POWER, saved before the restart: "
+                           + (self.circ_wd.reason
+                              or messages.NO_REASON_SAVED),
+                           hint=messages.hint("circulating"))
 
         # SAFETY-03. The output clamp's alarm - see clamp.py for why the clamp
         # itself is the cheap half.
@@ -893,6 +930,12 @@ class Controller:
         self.target = 0.0
         self.sat_count = 0
         self.i_w = 0.0        # the loop's integrator, carried between cycles
+        # SAFETY-52. The direction the battery refuses (+1 / -1 / 0), carried
+        # like `i_w`, and when the refusal (or its last probe) started.
+        self.batt_refused = 0
+        self.batt_refused_at = None
+        self._now_mono = None
+        self._refusal_auto = False
         # TEL-10. What the last cycle's loop did, kept for publication. Seeded
         # to the SAME values the "stopped or blind" branch of `cycle()` writes,
         # because before the first cycle the loop has not run - which is the
@@ -929,6 +972,11 @@ class Controller:
         self.following = False
         self.reason = "starting"
         self.why_words = "Starting"
+        # UX-11: has this meter episode's WARNING been logged? See read_inputs.
+        self._meter_quiet_logged = False
+        # ...when the last WARNING was, and whether this episode logged one.
+        self._meter_quiet_warned = None
+        self._meter_quiet_loud = False
         # ALLOC-01. The allocation record from the last resolution, or None
         # before the first cycle. ⚠️ IT IS KEPT SO THAT SOMETHING CAN GATE ON
         # IT: SAFETY-28 puts the write path behind this record, and DEV-02 reads
@@ -1021,6 +1069,69 @@ class Controller:
         self.loop_pass_mono = None
 
     # -- helpers -------------------------------------------------------------
+    # --- SAFETY-52: a battery that refuses is asked for 0 W ------------------
+    def _refusal_probe(self) -> bool:
+        """True while a refused direction may try `PROBE_W` (every 15 min)."""
+        if (not self.batt_refused or self.batt_refused_at is None
+                or self._now_mono is None):
+            return False
+        held = self._now_mono - self.batt_refused_at
+        if held >= _ctl.PROBE_EVERY_S + _ctl.PROBE_S:
+            self.batt_refused_at = self._now_mono
+            return False
+        return held >= _ctl.PROBE_EVERY_S
+
+    def _refusal_start(self, sign: int, evidence: str) -> None:
+        self.batt_refused = sign
+        self.batt_refused_at = self._now_mono
+        self.log_event(f"BATTERY REFUSES {evidence}",
+                       hint=messages.hint("battery_refuses"))
+        _LOG.info("The battery refuses to give power, so the add-on asks 0 W "
+                  "and tries %.0f W every %.0f minutes: BATTERY_REFUSES.",
+                  _ctl.PROBE_W, _ctl.PROBE_EVERY_S / 60)
+        _LOG.debug("Evidence: BATTERY_REFUSES: %s.", evidence)
+
+    def _refusal_end(self, why: str) -> None:
+        if not self.batt_refused:
+            return
+        self.batt_refused, self.batt_refused_at = 0, None
+        self.log_event(f"BATTERY REFUSES ENDED: {why}",
+                       hint=messages.hint("battery_gives"))
+        _LOG.info("The battery refusal ends, control asks for power again: %s.",
+                  why)
+
+    def _refusal_update(self, decision, asked_w, batt_w, probe) -> None:
+        """Follow `compute()`'s carried refused direction, and say so."""
+        now = decision.refused or 0
+        if now and not self.batt_refused:
+            self._refusal_start(now, f"asked {asked_w:+.0f} W, the battery "
+                                     f"gave {batt_w:+.0f} W for "
+                                     f"{decision.sat_count} cycles")
+        elif self.batt_refused and not now:
+            gave = self.batt_refused * batt_w >= _ctl.PROBE_ANSWER_W
+            self._refusal_end("the probe was answered" if gave and probe
+                              else "the battery gives power" if gave
+                              else "the grid now asks the other direction")
+
+    def _refusal_from_stale(self, inv, now_mono) -> None:
+        """A stale-reading stop while the loop is frozen on ~0 W is a refusal.
+
+        Only at one inverter: with several, one empty battery must not stop the
+        site (OPT-09's pack limit moves its share to the others).
+        """
+        t = self.tuning
+        standing = inv.standing_w()
+        if (len(self.inverters) == 1 and (self.auto or self.shadow)
+                and not self.batt_refused
+                and self.sat_count >= t.saturation_cycles
+                and inv.batt_fresh.held is not None
+                and abs(inv.batt_fresh.held) < _ctl.REFUSED_BELOW_W
+                and abs(standing) > t.saturation_w):
+            self._now_mono = now_mono
+            self._refusal_start(1 if standing > 0 else -1,
+                                f"stale reading {inv.batt_fresh.held:+.0f} W "
+                                f"while asked {standing:+.0f} W")
+
     def log_event(self, msg: str, hint: str | None = None) -> None:
         """`msg` goes on the action log, the add-on's own technical record.
         `hint` is `messages.hint()`'s plain words for the same event, and is
@@ -1229,8 +1340,9 @@ class Controller:
         first.store_set("sign_tripped", True)
         _LOG.error(
             "%s is set to 0 W again because an older saved safety stop was "
-            "found, so find out why before you start control. %s.",
-            first.name, first.sign_wd.reason)
+            "found, so find out why before you start control: %s.",
+            first.name, messages.reason_code(first.sign_wd.reason))
+        _LOG.debug("Evidence for %s: %s.", first.name, first.sign_wd.reason)
         _LOG.debug(
             "SAFETY-02 latch guard: the pre-DEV-02 flat key says the "
             "sign-disagreement failsafe is latched and no inverter came up "
@@ -1726,6 +1838,18 @@ class Controller:
         return all(inv.soc is not None and inv.batt is not None
                    for inv in self.inverters)
 
+    def missing_reading(self) -> str:
+        """UX-11: the first reading `inputs_ok` finds missing, in words."""
+        if self.grid is None or not self.inverters:
+            return "meter"
+        for inv in self.inverters:
+            of = "" if len(self.inverters) == 1 else f" from {inv.name}"
+            if inv.batt is None:
+                return f"battery power{of}"
+            if inv.soc is None:
+                return f"battery level{of}"
+        return "input"
+
     # -- io ------------------------------------------------------------------
     @property
     def pv_roster(self) -> tuple:
@@ -1790,6 +1914,7 @@ class Controller:
         """
         o = self.o
         now_mono = freshness.monotonic() if now_mono is None else float(now_mono)
+        self._now_mono = now_mono
         if self.p1_enabled:
             # ⚠️ P1 is the only authoritative measurement of what the utility
             # sees (§5.1). When it is stale this is None, which falls into the
@@ -1885,6 +2010,34 @@ class Controller:
             horizon_h=float(price_option(o, "price_horizon_h")),
             provider=provider_configured(o), max_age_s=TARIFF_MAX_AGE_S)
 
+        # UX-11: the red meter row says "the add-on log shows why", so a meter
+        # that goes quiet gets one WARNING per episode, not one per cycle. It
+        # is a log line only: no notification, and nothing it decides.
+        # A flapping meter logs at most one WARNING per METER_QUIET_REPEAT_S,
+        # and the episodes in between go to DEBUG.
+        if self.p1_enabled:
+            if (self.p1.stale and not self._meter_quiet_logged
+                    and self.p1.published_age_s > self.p1.max_age_s):
+                self._meter_quiet_logged = True
+                now = _mono()
+                self._meter_quiet_loud = (
+                    self._meter_quiet_warned is None
+                    or now - self._meter_quiet_warned >= METER_QUIET_REPEAT_S)
+                if self._meter_quiet_loud:
+                    self._meter_quiet_warned = now
+                    _LOG.warning("No meter reading for over %s, so control "
+                                 "waits and then sets the battery to 0 W.",
+                                 messages.seconds(self.p1.max_age_s))
+                else:
+                    _LOG.debug("No meter reading for over %s again.",
+                               messages.seconds(self.p1.max_age_s))
+            elif not self.p1.stale and self._meter_quiet_logged:
+                self._meter_quiet_logged = False
+                if self._meter_quiet_loud:
+                    _LOG.info("Meter readings are back.")
+                else:
+                    _LOG.debug("Meter readings are back.")
+
         if self.inputs_ok:
             if self.inputs_bad_since is not None:
                 self.log_event("inputs recovered",
@@ -1925,6 +2078,7 @@ class Controller:
         if inv.batt_fresh.stale and not was_stale:
             self.log_event(f"STALE {inv.name}: " + inv.batt_fresh.reason,
                            hint=messages.hint("battery_stuck", unit=inv.name))
+            self._refusal_from_stale(inv, now_mono)
             # ⚠️ SAFETY-57 REWROTE THE LAST SENTENCE OF THIS LINE AND NOTHING
             # ELSE IN IT. The escape clause used to read "It clears itself the
             # moment the register reports a different value", full stop - and
@@ -1953,13 +2107,16 @@ class Controller:
             # learns to skim it.
             # UX-08 2d: plain, with SAFETY-57's escape clause kept as an
             # ordered pair (time, then power), and the reason still in `%s`.
-            # UX-10: one sentence, then the reason's two. Its limits are on
-            # the DEBUG line below, to keep the line within 3 sentences.
+            # UX-11: one sentence and the code. The reason, with SAFETY-56's
+            # two explanations and its figures, and the limits are on the
+            # DEBUG lines below and in the action log.
             _LOG.error("%s: battery power reading stopped changing, set to "
                        "0 W until it changes, or %.0f s after the command "
-                       "drops below %.0f W. %s.",
+                       "drops below %.0f W: %s.",
                        inv.name, inv.batt_fresh.bound.release_s,
-                       inv.batt_fresh.bound.move_w, inv.batt_fresh.reason)
+                       inv.batt_fresh.bound.move_w,
+                       messages.reason_code(inv.batt_fresh.reason))
+            _LOG.debug("Evidence for %s: %s.", inv.name, inv.batt_fresh.reason)
             _LOG.debug("%s: the limits are unchanged over %.0f s, a command "
                        "over %.0f W and a move by over %.0f W.", inv.name,
                        inv.batt_fresh.bound.unchanged_s,
@@ -2491,8 +2648,10 @@ class Controller:
                 self.log_event(f"FAILSAFE {inv.name}: " + inv.sign_wd.reason,
                                hint=messages.hint("direction", unit=inv.name))
                 _LOG.error("%s is set to 0 W until control is started, "
-                           "because its power went the wrong way. %s.",
-                           inv.name, inv.sign_wd.reason)
+                           "because its power went the wrong way: %s.",
+                           inv.name, messages.reason_code(inv.sign_wd.reason))
+                _LOG.debug("Evidence for %s: %s.", inv.name,
+                           inv.sign_wd.reason)
         # ⚠️ THE SITE'S OWN STANDING COMMAND, WHICH IS THE SUM OF THE UNITS'.
         # SAFETY-07 and SAFETY-18 both grade a SITE quantity - the quarter-hour
         # average at the grid connection, and a meter that has stopped moving
@@ -2540,7 +2699,10 @@ class Controller:
                                hint=messages.hint("site_direction"))
                 _LOG.error("Every inverter is set to 0 W until control is "
                            "started, because the total power went the wrong "
-                           "way. %s.", self.site_sign_wd.reason)
+                           "way: %s.",
+                           messages.reason_code(self.site_sign_wd.reason))
+                _LOG.debug("Evidence for the site: %s.",
+                           self.site_sign_wd.reason)
 
         # --- SAFETY-31, AND IT RUNS BESIDE SAFETY-02 FOR THE SAME REASON -----
         # Evaluated EVERY cycle, unconditionally, above the "inputs missing but
@@ -2572,8 +2734,9 @@ class Controller:
                            hint=messages.hint("circulating"))
             _LOG.error(
                 "Every inverter is set to 0 W until control is started, "
-                "because the inverters worked against each other. %s.",
-                self.circ_wd.reason)
+                "because the inverters worked against each other: %s.",
+                messages.reason_code(self.circ_wd.reason))
+            _LOG.debug("Evidence for the site: %s.", self.circ_wd.reason)
 
         # --- SAFETY-42, AND IT RUNS HERE FOR SAFETY-02's AND SAFETY-31's REASON
         # Evaluated EVERY cycle, unconditionally, ABOVE the "inputs missing but
@@ -2680,8 +2843,10 @@ class Controller:
                                                   unit=inv.name))
                 _LOG.error(
                     "%s is not delivering while control keeps running, and "
-                    "only stopping and starting control clears this. %s.",
-                    inv.name, inv.delivery_wd.reason)
+                    "only stopping and starting control clears this: %s.",
+                    inv.name, messages.reason_code(inv.delivery_wd.reason))
+                _LOG.debug("Evidence for %s: %s.", inv.name,
+                           inv.delivery_wd.reason)
                 _LOG.debug(
                     "SAFETY-42 reports and does not act: commanding a unit "
                     "that is not listening achieves nothing (HW-01), and the "
@@ -3251,6 +3416,12 @@ class Controller:
         # loop should be. It is safe because the loop's `prev_w` is
         # `self.target`, which the end of this method re-seeds to what is
         # actually standing on the wire.
+        # SAFETY-52: a refusal learned while control ran does not survive a
+        # stop or a start (shadow mode keeps the loop running when stopped).
+        if bool(self.auto) != self._refusal_auto:
+            self._refusal_auto = bool(self.auto)
+            self._refusal_end("control started" if self.auto
+                              else "control stopped")
         if (self.auto or self.shadow) and self.inputs_ok:
             # ⚠️ prev is the ARBITER's last output, not the loop's own last
             # wish. If something outranked the loop, that is what the hardware
@@ -3268,6 +3439,13 @@ class Controller:
             # stale period as power.
             if self.loop_w is None or self.target != self.loop_w:
                 self.i_w = self.target
+            # SAFETY-52: another writer overrode the loop, so a refusal learned
+            # against the loop's own ask no longer describes the wire. A gap
+            # (`loop_w` None: blind or stopped) keeps it: the stale-reading stop
+            # is what seeds it, and dropping it there restarts the ~70 s cycle.
+            if self.loop_w is not None and self.target != self.loop_w:
+                self._refusal_end("another writer")
+            probe = self._refusal_probe()
             # ⚠️ `actual_w` IS THE SITE SUM AND THAT IS DEV-02's OTHER HALF.
             # The anti-windup test inside compute() is
             # `abs(prev_w - actual_w) > saturation_w`, and `prev_w` is the SITE
@@ -3296,6 +3474,10 @@ class Controller:
                 tuning=self.tuning,
                 sat_count=self.sat_count,
                 i_w=self.i_w,
+                # SAFETY-52 at one inverter only: with several, OPT-09's pack
+                # limit already closes an empty battery's discharge end.
+                refused=self.batt_refused if len(self.inverters) == 1 else None,
+                probe=probe,
             )
             if decision.frozen and self.sat_count < self.tuning.saturation_cycles:
                 self.log_event(f"saturation freeze ({self.target:.0f} W vs "
@@ -3304,6 +3486,7 @@ class Controller:
             self.sat_count = decision.sat_count
             self.i_w = decision.i_w
             self.loop_w = decision.target_w
+            self._refusal_update(decision, loop_prev, site_batt, probe)
             # --- TEL-10 ------------------------------------------------------
             # ⚠️ THE INTEGRATOR HAS NEVER LEFT THIS PROCESS, AND THAT IS WHAT
             # THE CARD IS. `self.i_w` is carried between cycles and published
@@ -3554,9 +3737,14 @@ class Controller:
         # UX-10: `why_key()` maps a safety winner to its claim's reason, and
         # every key it can return has words in messages.WHY_WINNER.
         _why = why_key(resolution.winner, claims)
-        self.why_words = messages.why(_why, resolution.target_w,
-                                      resolution.bound_by,
-                                      contradiction=resolution.contradiction)
+        if _why == "inputs missing" and not resolution.contradiction:
+            # UX-11: past the hold, say which reading is missing.
+            self.why_words = messages.why_missing(self.missing_reading(),
+                                                  stale_after)
+        else:
+            self.why_words = messages.why(
+                _why, resolution.target_w, resolution.bound_by,
+                contradiction=resolution.contradiction)
         # --- DEV-02: EACH INVERTER IS TOLD ITS OWN SHARE ---------------------
         # ⚠️ ITS OWN SHARE, NEVER THE SITE TARGET. `resolution.shares` is the
         # allocator's answer and `sum(shares) + unmet_w == round(target_w)` is
@@ -3743,8 +3931,8 @@ class Controller:
                     shadow.why(self.o, control_started=self.auto))
             elif not first:
                 self.log_event(
-                    "SHADOW MODE OFF - the battery is being controlled again, "
-                    f"resuming from {shadow.WIRE_W:.0f} W",
+                    "Shadow mode off: the add-on controls the battery again, "
+                    f"starting from {shadow.WIRE_W:.0f} W",
                     hint=messages.hint("shadow_off"))
                 _LOG.info(
                     "Shadow mode is off, so the add-on controls the battery "
@@ -3856,8 +4044,9 @@ class Controller:
         # 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}")
+            row = (f"{tag}: "
+                   f"{messages.count(self.clamp_wd.engagements, 'engagement')}"
+                   f" - {reason}")
             if self.events and tag in self.events[0]:
                 stamp = self.events[0].split(" ", 1)[0]
                 self.events[0] = f"{stamp} {row}"
@@ -3871,8 +4060,9 @@ class Controller:
             # WARNING and a plainly titled notice, same cadence. Its OWN id, so
             # it can never replace a BUG SIGNAL notice still on screen.
             _LOG.warning("The battery gives less than asked: %s. Held back %s "
-                         "so far.", self.clamp_wd.reason,
+                         "so far.", messages.reason_code(self.clamp_wd.reason),
                          messages.times(self.clamp_wd.engagements))
+            _LOG.debug("Evidence: %s.", self.clamp_wd.reason)
             _LOG.debug("A short hold after a large load or solar step is "
                        "normal. If it lasts, check that the battery follows "
                        "its commands.")
@@ -3881,8 +4071,9 @@ class Controller:
             return
         _LOG.error(
             "Power request too high, a software fault: %s. Cut back %s so "
-            "far, control runs on.",
-            self.clamp_wd.reason, messages.times(self.clamp_wd.engagements))
+            "far, control runs on.", messages.reason_code(self.clamp_wd.reason),
+            messages.times(self.clamp_wd.engagements))
+        _LOG.debug("Evidence: %s.", self.clamp_wd.reason)
         _LOG.debug(
             "SAFETY-03: silently clipping this is how a winding integrator "
             "reached 14 768 W on this project before. The clamp makes every "
@@ -3931,7 +4122,9 @@ class Controller:
                        hint=messages.hint("peak_saturated"))
         _LOG.error(
             "Switch off large loads, the battery alone cannot keep this "
-            "15-minute period under the peak limit. %s.", g.reason)
+            "15-minute period under the peak limit: %s.",
+            messages.reason_code(g.reason))
+        _LOG.debug("Evidence: %s.", g.reason)
         # A fixed notification_id, and a DIFFERENT one from every other alarm
         # here - check_meter_frozen() gives the reasoning. A saturated peak guard
         # is not a frozen meter and is not an out-of-range setpoint.
@@ -3988,7 +4181,8 @@ class Controller:
         self.log_event(g.reason, hint=messages.hint("peak_reserve"))
         _LOG.error(
             "Switch off large loads, the peak guard may not draw the battery "
-            "lower. %s.", g.reason)
+            "lower: %s.", messages.reason_code(g.reason))
+        _LOG.debug("Evidence: %s.", g.reason)
         # A fixed notification_id, and a DIFFERENT one from every other alarm
         # here - check_meter_frozen() gives the reasoning. A battery at its
         # energy floor is not a saturated inverter and is not a frozen meter,
@@ -4148,8 +4342,9 @@ class Controller:
                        hint=messages.hint("meter_frozen"))
         _LOG.error(
             "The meter shows the same value while the battery is asked for "
-            "full power, and control still follows it, so check the meter. "
-            "%s.", self.freeze_wd.reason)
+            "full power, and control still follows it, so check the meter: "
+            "%s.", messages.reason_code(self.freeze_wd.reason))
+        _LOG.debug("Evidence: %s.", self.freeze_wd.reason)
         _LOG.debug(
             "SAFETY-18: nothing else stops this. The meter still answers, so "
             "its age is fresh, the firmware watchdog is fed and the arbiter "

goodwe_controller/app/messages.py

@@ -311,6 +311,22 @@ def seconds(value) -> str:
     return f"{n} second" if n == "1" else f"{n} seconds"
 
 
+# One token each, like a code, so the line keeps its length and stays greppable.
+NO_REASON_SAVED, NO_REASON_CODE = "REASON_NOT_SAVED", "REASON_WITHOUT_CODE"
+_REASON_CODE = re.compile(r"\b[A-Z][A-Z0-9]*(?:_[A-Z0-9]+)+\b")
+
+
+def reason_code(reason) -> str:
+    """UX-11: the code in a safety reason (`STALE_READING`), for the one plain
+    WARNING or ERROR line that reports it. The reason with its figures goes on
+    a DEBUG line beside it and into the action log. Saved reasons have always
+    opened with their code, so an old state file reads the same way."""
+    if not reason:
+        return NO_REASON_SAVED
+    m = _REASON_CODE.search(str(reason))
+    return m.group(0) if m else NO_REASON_CODE
+
+
 def times(n) -> str:
     return "1 time" if int(n) == 1 else f"{int(n)} times"
 
@@ -786,6 +802,9 @@ WHY_WINNER = {
     # failsafe winner. test_messages.py part 7 finds every one in the code.
     "sign disagreement": "Safety stop, power went the wrong way",
     "unsplit": "The power could not be split, all at 0 W",
+    # SAFETY-52: the loop's two words for a battery that gives nothing.
+    "battery refuses": "The battery refuses to give power",
+    "battery probe": "Testing if the battery gives power again",
 }
 WHY_LIMIT = {
     "peak(quarter peak)": "the peak limit",
@@ -813,6 +832,11 @@ def why(winner, target_w, bound_by=None, *, contradiction=False) -> str:
     return out
 
 
+def why_missing(what, limit_s) -> str:
+    """UX-11: past the hold, which reading is missing and what it does."""
+    return f"No usable {what} reading for over {seconds(limit_s)}: 0 W"
+
+
 def why_waiting(missing_s) -> str:
     """A reading went missing a moment ago: control holds and waits."""
     return f"A reading is missing for {seconds(missing_s)}, control waits"
@@ -831,6 +855,8 @@ HINTS = {
     "stop_restored": "{unit}: an earlier safety stop is back, it stays at 0 W",
     "battery_stuck": "{unit}: battery power reading stopped changing",
     "battery_back": "{unit}: battery power reading is used again",
+    "battery_refuses": "The battery refuses to give power, the add-on asks 0 W",
+    "battery_gives": "The battery refusal ended, control asks for power again",
     "bad_value": "The add-on made an invalid power value, inverters set to 0 W",
     "write_blocked": "{unit}: a power request was blocked, it is set to 0 W",
     "all_blocked": "A power request was blocked, inverters set to 0 W",

goodwe_controller/app/watchdog.py

@@ -670,11 +670,10 @@ class SignWatchdog:
         # ⚠️ `self.against` IS THE ONLY THING SAFETY-30 CHANGED IN THIS STRING,
         # and it defaults to the wording that shipped - "a meter reading" - so a
         # one-inverter install's `reason` is byte-identical to what an operator
-        # (and test_watchdog.py) has always read. A per-unit instance names what
-        # it actually compared instead, because "commanded +1500 W against a
-        # meter reading -1500 W" would be a false statement about which
-        # instrument produced the evidence, on the one line somebody woken by
-        # this alarm reads first.
+        # has always read (test_watchdog.py pins its meaning). A per-unit
+        # instance names what it actually compared instead, because "commanded
+        # +1500 W against a meter reading -1500 W" would be a false statement
+        # about which instrument produced the evidence.
         self.reason = (
             f"SIGN_DISAGREEMENT: commanded {commanded_w:+.0f} W against "
             f"{self.against} {against_w:+.0f} W for {held:.0f} s continuously "

goodwe_controller/test_clamp.py

@@ -497,8 +497,13 @@ check(f"AC 1: NOT ONE of the {len(emitted)} values that reached the device "
 # 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.
+# SAFETY-52 re-baseline: an inverter reporting 0 W for SETTLE_CYCLES is a
+# battery that refuses, so the command then drops to 0 W. The freeze still
+# engaged first, and the event log says so.
 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)
+      f"rather than the rail",
+      any("saturation freeze" in e for e in ctl_r.events)
+      and max(emitted) < LIVE_MAX_W)
 
 notes_r = clamp_notes(hass_r)
 print(f"    the loop clamped on {ctl_r.clamp_wd.engagements} of {60} cycles, "
@@ -521,7 +526,8 @@ 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])
+      and messages.count(ctl_r.clamp_wd.engagements, "engagement")
+      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
@@ -538,7 +544,8 @@ 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 nth(rows_l, -1))
+      messages.count(ctl_l.clamp_wd.engagements, "engagement")
+      in nth(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 flags it WHILE it is engaged - amber, since it is the "
@@ -578,8 +585,12 @@ check("UX-07 item 1: during a bound episode an earlier over-rating count is "
 
 # ⚠️ 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])
+# SAFETY-52: the loop itself now asks 0 W of an inverter that gives nothing,
+# with its own reason. That is the loop's decision, not a stop.
+check(f"it is NOT a trip - the command stands at {ctl_r.target:+.0f} W because "
+      f"the loop says '{ctl_r.loop_reason}', not because of a stop",
+      ctl_r.loop_reason == "battery refuses"
+      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)
 
@@ -2051,9 +2062,12 @@ check("it still NOTIFIES (not suppressed), under a title that claims no bug",
 check("...under its OWN notification_id, so it cannot replace a BUG SIGNAL "
       "notice still on screen",
       notes_i and all(n["notification_id"] == BOUND_ID for n in notes_i))
+# UX-11: the WARNING quotes the code. The reason is on the DEBUG line beside
+# it (test_messages.py part 6).
 check("...logged as a WARNING, not an ERROR",
-      any(x.levelno == logging.WARNING and "reached its" in x.getMessage()
-          for x in log_i)
+      any(x.levelno == logging.WARNING
+          and x.getMessage().startswith("The battery gives less than asked: "
+                                        "OUTPUT_CLAMP.") for x in log_i)
       and not any(x.levelno >= logging.ERROR and "OUTPUT_CLAMP" in x.getMessage()
                   for x in log_i))
 check("...and the status row is amber, not red",

goodwe_controller/test_delivery.py

@@ -1147,8 +1147,10 @@ async def _one():
             "soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
             "meter_source": "off", "max_w": 2000, "gain": 0.6,
             # ⚠️ THE SAME 1 s AS THE TWO-INVERTER FIXTURE, AND THAT IS WHAT
-            # MAKES THIS SECTION NON-VACUOUS. The run below is 2 s of wall clock
-            # against a 1 s window, so a hold is not what stops this firing -
+            # MAKES THIS SECTION NON-VACUOUS. The run below is 1.1 s of wall
+            # clock against a 1 s window (SAFETY-52 shortened it from 2 s, and
+            # "the window was never even opened" below is what still makes it
+            # non-vacuous), so a hold is not what stops this firing -
             # the reduction is. With a shipped-length hold "nothing was
             # reported" would be true of a fixture that simply ran out of time.
             "bms_soc_floor": _FLOOR_PCT, "delivery_hold_s": 1,
@@ -1158,7 +1160,10 @@ async def _one():
     c.inverters[0].dev_min, c.inverters[0].dev_max = -2000.0, 2000.0
     c.auto = True
     hass.ctrl = c
-    await _drive(c, n=20, dt=0.1)
+    # SAFETY-52: from SETTLE_CYCLES of delivering nothing the loop asks 0 W
+    # (the battery refuses), so the run stops just before that. 11 cycles is
+    # still 1.1 s against the 1 s hold.
+    await _drive(c, n=11, dt=0.1)
     return c, hass, store
 
 

goodwe_controller/test_freshness.py

@@ -1813,6 +1813,11 @@ _S56_ACCUSATIONS = [
 ]
 _log56 = [r.getMessage() for r in _LOGREC[_S56_LOG0:]
           if "battery power reading stopped changing" in r.getMessage()]
+# UX-11: the ERROR line quotes the code, and the reason itself is on the DEBUG
+# line logged beside it.
+_dbg56 = [r.getMessage() for r in _LOGREC[_S56_LOG0:]
+          if r.levelno == logging.DEBUG
+          and r.getMessage().startswith("Evidence for ")] or [""]
 
 
 def _s56_row(c):
@@ -1855,7 +1860,7 @@ check("PRECONDITION: all four operator-facing surfaces were really produced by "
       "two are the same string, which is 11b again on the log surface",
       len(_log56) == 2 and _log56[0] == _log56[1]
       and all(len(s) > 30 for s in _S56_SURFACES.values())
-      and _r56_ref in _log56[-1])
+      and _log56[-1].endswith(": STALE_READING.") and _r56_ref in _dbg56[-1])
 _idx56 = [_r56_ref.find(t) for _w, t in _S56_EXPLANATIONS]
 check(f"11c. BOTH EXPLANATIONS ARE NAMED, AND IN THAT ORDER: "
       f"{_S56_EXPLANATIONS[0][1]!r} at {_idx56[0]} then "

goodwe_controller/test_messages.py

@@ -877,6 +877,7 @@ DATA = {
     "main.py:messages.when(brussels(self.tariff_at))",
     "main.py:messages.duration(TARIFF_MAX_AGE_S)",
     "main.py:messages.count(len(targets), 'inverter')",
+    "main.py:messages.seconds(self.p1.max_age_s)",
     # maintenance.py, mqtt.py, p1.py
     "maintenance.py:phase", "mqtt.py:err", "p1.py:err", "p1.py:self.host",
     "p1.py:self.topic", "p1.py:poll_s", "p1.py:ingest.max_age_s",
@@ -1047,20 +1048,35 @@ def _static(spec):
     return [t for t in out if len(words(t)) > 1]
 
 
+# UX-11: a WARNING+ line quotes a safety reason by its code only. The reason,
+# with its figures, is on a DEBUG line beside it and in the action log.
+SAFETY_REASONS = {
+    "sign": _sign_texts, "circulation": _circ_texts,
+    "delivery": _delivery_texts, "clamp": _clamp_texts, "peak": _peak_texts,
+    "freeze": _freeze_texts, "stale": _stale_texts,
+}
+
+
+def _coded(maker):
+    """The codes a WARNING+ line shows for the reasons `maker` builds, plus
+    the fallback for a restored stop with no reason saved."""
+    return lambda: ([messages.reason_code(t) for t in maker()]
+                    + [messages.reason_code("")])
+
+
 TEXT = {
-    "main.py:first.sign_wd.reason": _sign_texts,
-    "main.py:inv.sign_wd.reason": _sign_texts,
-    "main.py:inv.sign_wd.reason or '(reason not recorded)'": _sign_texts,
-    "main.py:self.site_sign_wd.reason": _sign_texts,
-    "main.py:self.site_sign_wd.reason or '(reason not recorded)'": _sign_texts,
-    "main.py:self.circ_wd.reason": _circ_texts,
-    "main.py:self.circ_wd.reason or '(reason not recorded)'": _circ_texts,
-    "main.py:inv.delivery_wd.reason": _delivery_texts,
-    "main.py:inv.delivery_wd.reason or '(reason not recorded)'": _delivery_texts,
-    "main.py:self.clamp_wd.reason": _clamp_texts,
-    "main.py:g.reason": _peak_texts,
-    "main.py:self.freeze_wd.reason": _freeze_texts,
-    "main.py:inv.batt_fresh.reason": _stale_texts,
+    "main.py:messages.reason_code(first.sign_wd.reason)": _coded(_sign_texts),
+    "main.py:messages.reason_code(inv.sign_wd.reason)": _coded(_sign_texts),
+    "main.py:messages.reason_code(self.site_sign_wd.reason)":
+        _coded(_sign_texts),
+    "main.py:messages.reason_code(self.circ_wd.reason)": _coded(_circ_texts),
+    "main.py:messages.reason_code(inv.delivery_wd.reason)":
+        _coded(_delivery_texts),
+    "main.py:messages.reason_code(self.clamp_wd.reason)": _coded(_clamp_texts),
+    "main.py:messages.reason_code(g.reason)": _coded(_peak_texts),
+    "main.py:messages.reason_code(self.freeze_wd.reason)":
+        _coded(_freeze_texts),
+    "main.py:messages.reason_code(inv.batt_fresh.reason)": _coded(_stale_texts),
     "main.py:rep.explain()": _migration_texts,
     "store.py:rep.explain()": _migration_texts,
     "main.py:plan.words": _tariff_texts,
@@ -1086,6 +1102,26 @@ TEXT = {
 }
 
 
+# UX-11: a WARNING+ line, as logged with its built text filled in, is at most
+# 25 words (STYLE.md). These families are not safety reasons and were not
+# rewritten by UX-11. Each is listed with why, the list may only shrink
+# (LONG_LINE_CAP), and a listed key that no longer runs long fails.
+LONG_LINE_OK = {
+    "main.py:plan.words": "tariff plan words, TARIFF-11 pins them, follow-up",
+    "static_tariff.py:read.reason": "price feed reasons, follow-up",
+    "main.py:rep.explain()": "state migration report, names saved keys",
+    "store.py:rep.explain()": "state migration report, names saved keys",
+    "main.py:refusal": "write refusals, a software fault only",
+    "main.py:' / '.join(resolution.scope_errors)":
+        "claim scope errors, a software fault only",
+    "maintenance.py:msg": "maintenance gate reasons, follow-up",
+    "price_forecast.py:gate.reason": "trading gate reasons, follow-up",
+    "price_history.py:self.last_error": "price history errors, follow-up",
+}
+LONG_LINE_CAP = 9
+LINE_WORDS = 25
+
+
 def reason_problems(text):
     """STYLE.md's red-alarm shape, for a text a log line quotes."""
     out = []
@@ -1133,10 +1169,12 @@ def _fill(template, fmt_node, key_part, sample):
     return template  # %-style: filled by the caller
 
 
-def scan_log_lines(files, text=None, data=None):
+def scan_log_lines(files, text=None, data=None, long_ok=None, long_seen=None):
     """-> (unclassified keys, stale keys, [(key, sample, problems)])."""
     text = TEXT if text is None else text
     data = DATA if data is None else data
+    long_ok = LONG_LINE_OK if long_ok is None else long_ok
+    long_seen = set() if long_seen is None else long_seen
     seen, bad = set(), []
     for fname, tree in files:
         for key, call in dynamic_parts(tree, fname):
@@ -1162,6 +1200,11 @@ def scan_log_lines(files, text=None, data=None):
                     line = PLACEHOLDER.sub("X", line).replace("\0", sample)
                 p += [f"as logged: {x}" for x in line_problems(line)
                       + logged_shape(line) if x not in p]
+                if len(words(line)) > LINE_WORDS:
+                    long_seen.add(key)
+                    if key not in long_ok:
+                        p.append(f"as logged: over {LINE_WORDS} words "
+                                 f"({len(words(line))})")
                 if p:
                     bad.append((key, sample, p))
     unknown = sorted(seen - set(data) - set(text))
@@ -1171,14 +1214,81 @@ def scan_log_lines(files, text=None, data=None):
 
 _FILES = [(f.name, ast.parse(f.read_text(encoding="utf-8")))
           for f in sorted(APP.glob("*.py"))]
-_unknown, _stale, _bad = scan_log_lines(_FILES)
+_long_seen = set()
+_unknown, _stale, _bad = scan_log_lines(_FILES, long_seen=_long_seen)
 for k, sample, p in _bad:
-    print(f"       BAD  {k}: {'; '.join(p)} :: {sample[:110]!r}")
+    print(f"       BAD  {k}: {'; '.join(p)} :: {sample!r}")
 check(f"every dynamic part of a WARNING+ line is classified as data or as a "
       f"built text (unclassified: {_unknown})", not _unknown)
 check(f"...and no classification is left over from a line that is gone "
       f"(stale: {_stale})", not _stale)
 _n_texts = sum(len(m()) for m in TEXT.values() if m is not None)
+check(f"UX-11: no WARNING+ line runs over {LINE_WORDS} words as logged, except "
+      f"the {len(LONG_LINE_OK)} listed families, and none of them is a safety "
+      f"reason (cap {LONG_LINE_CAP})",
+      len(LONG_LINE_OK) <= LONG_LINE_CAP
+      and not any("reason_code" in k
+                  or TEXT.get(k) in SAFETY_REASONS.values()
+                  for k in LONG_LINE_OK))
+check(f"...and every listed family still runs long, so the list only shrinks "
+      f"(no longer long: {sorted(set(LONG_LINE_OK) - _long_seen)})",
+      set(LONG_LINE_OK) <= _long_seen)
+
+
+def _code_lines_without_evidence(files):
+    """WARNING+ calls quoting `messages.reason_code(X)` whose next statement
+    is not a DEBUG call that logs X itself (UX-11: nothing is lost)."""
+    out, n = [], 0
+    for fname, tree in files:
+        for node in ast.walk(tree):
+            for body in (getattr(node, f, None) for f in ("body", "orelse")):
+                if not isinstance(body, list):
+                    continue
+                for i, st in enumerate(body):
+                    call = st.value if isinstance(st, ast.Expr) else None
+                    if not (isinstance(call, ast.Call)
+                            and LOG_CALL.search(ast.unparse(call.func))):
+                        continue
+                    coded = [a.args[0] for a in call.args
+                             if isinstance(a, ast.Call) and a.args
+                             and ast.unparse(a.func) == "messages.reason_code"]
+                    if not coded:
+                        continue
+                    n += 1
+                    nxt = body[i + 1] if i + 1 < len(body) else None
+                    dbg = (nxt.value if isinstance(nxt, ast.Expr)
+                           and isinstance(nxt.value, ast.Call)
+                           and ast.unparse(nxt.value.func).endswith(".debug")
+                           else None)
+                    want = ast.unparse(coded[0])
+                    if dbg is None or not any(
+                            ast.unparse(a) == want
+                            or ast.unparse(a).startswith(want + " or ")
+                            for a in dbg.args):
+                        out.append(f"{fname}:{call.lineno}")
+    return n, out
+
+
+_n_coded, _no_evidence = _code_lines_without_evidence(_FILES)
+check(f"every WARNING+ line that quotes a reason code ({_n_coded}) has the "
+      f"reason itself on a DEBUG line right after it (missing: {_no_evidence})",
+      _n_coded >= 14 and not _no_evidence)
+_n_m, _m_bad = _code_lines_without_evidence([("x.py", ast.parse(
+    "_LOG.error('Stop: %s.', messages.reason_code(wd.reason))\n"
+    "_LOG.debug('Something else.')\n"))])
+check("...and a line whose DEBUG evidence was dropped is caught",
+      _n_m == 1 and _m_bad == ["x.py:1"])
+_evidence_bad = [(n, t, reason_problems(t)) for n, m in SAFETY_REASONS.items()
+                 for t in m() if reason_problems(t)]
+check(f"the safety reasons behind those lines, on the DEBUG line and in the "
+      f"action log, keep the red-alarm shape ({_evidence_bad})",
+      not _evidence_bad)
+_codes = {n: [messages.reason_code(t) for t in m()]
+          for n, m in SAFETY_REASONS.items()}
+check(f"...and every one carries a code the line can quote ({_codes})",
+      all(re.fullmatch(r"[A-Z]+(?:_[A-Z]+)+", c)
+          and c not in (messages.NO_REASON_SAVED, messages.NO_REASON_CODE)
+          for cs in _codes.values() for c in cs))
 check(f"every built text, run for real, obeys the rule, and so does the line "
       f"that quotes it ({_n_texts} texts, {len(_bad)} bad)", not _bad)
 _counts = [f"{f}:{n.lineno}" for f, t in _FILES for n in ast.walk(t)
@@ -1208,29 +1318,43 @@ _clamp_lines = {
     "error, write": (_clamp_line("Power request too high"), _w1),
     "warning, loop": (_clamp_line("The battery gives less than asked"), _at),
 }
-_clamp_shown = {k: (f % (r, messages.times(1)) if f else None)
+_clamp_shown = {k: (f % (messages.reason_code(r), messages.times(1))
+                    if f else None)
                 for k, (f, r) in _clamp_lines.items()}
 for k, v in _clamp_shown.items():
     print(f"       {k}: {v!r} ({len(words(v or ''))} words)")
-check("the clamp lines for one inverter are at most 3 sentences and 30 words, "
-      "and keep OUTPUT_CLAMP and the numbers",
-      all(v and sentences(v) <= 3 and len(words(v)) <= 30
+check("UX-11: the clamp lines are at most 25 words and keep OUTPUT_CLAMP and "
+      "the count, and the numbers are in the reason the DEBUG line beside "
+      "them logs",
+      all(v and len(words(v)) <= LINE_WORDS
           and "OUTPUT_CLAMP" in v and "1 time so far" in v
-          for v in _clamp_shown.values())
-      and all(x in _clamp_shown["error, loop"]
-              for x in ("+9000 W", "+5000 W", "+2000 W", "software fault")))
+          and "+9000" not in v for v in _clamp_shown.values())
+      and "software fault" in _clamp_shown["error, loop"]
+      and all(x in _over for x in ("+9000 W", "+5000 W", "+2000 W")))
 
 # Mutations: the scan catches what the 2d scan missed.
 _OLD_CLAMP = ("OUTPUT_CLAMP: the loop computed +2148 W, over the 2000 W rating, "
               "and it was clamped to +2000 W; the loop's command this cycle was "
               "+440 W. This is a BUG SIGNAL, not a limit being used: something "
               "upstream computed a setpoint the actuator cannot deliver.")
-_mut_text = dict(TEXT, **{"main.py:self.clamp_wd.reason": lambda: [_OLD_CLAMP]})
+_mut_text = dict(TEXT, **{"main.py:messages.reason_code(self.clamp_wd.reason)":
+                          lambda: [_OLD_CLAMP]})
 _u, _s, _b = scan_log_lines([f for f in _FILES if f[0] == "main.py"],
                             text=_mut_text)
 check("an injected bad reason (the 2d clamp text) is caught, on both clamp "
       f"lines ({sorted({k for k, _t, _p in _b if 'clamp' in k})})",
-      [k for k, _t, _p in _b].count("main.py:self.clamp_wd.reason") == 2)
+      [k for k, _t, _p in _b].count(
+          "main.py:messages.reason_code(self.clamp_wd.reason)") == 2)
+# UX-11: the line that quoted the whole STALE_READING reason (60 words on
+# TEST-23) is caught by the length rule alone.
+_old_stale = dict(TEXT, **{"main.py:messages.reason_code(inv.batt_fresh.reason)":
+                           _stale_texts})
+_u, _s, _b = scan_log_lines([f for f in _FILES if f[0] == "main.py"],
+                            text=_old_stale)
+check("...and the UX-10 stale line, which quoted the whole reason, is caught "
+      f"for its length ({[p for k, _t, p in _b if 'batt_fresh' in k][:1]})",
+      any("batt_fresh" in k and any(f"over {LINE_WORDS} words" in x for x in p)
+          for k, _t, p in _b))
 _new_site = ast.parse("_LOG.warning('Something broke: %s', self.new_wd.reason)")
 _u, _s, _b = scan_log_lines([("main.py", _new_site)])
 check(f"...and a new line quoting a text nobody classified is caught ({_u})",
@@ -1275,6 +1399,8 @@ import logging                                                    # noqa: E402
 from app import hass as _hass, p1 as _p1, validate as _validate, web as _web  # noqa: E402
 from app.arbiter import Claim                                     # noqa: E402
 from app.main import Controller, why_key                          # noqa: E402
+from app import control as _control                              # noqa: E402
+from app import main as _main                                    # noqa: E402
 
 
 class _Rec(logging.Handler):
@@ -1460,6 +1586,11 @@ for _f in sorted(APP.glob("*.py")):
 _keys = {why_key(s, [Claim.set(s, 0, 0.0, r)]) for s, r in _SOURCES}
 _keys |= {why_key("failsafe", []),
           why_key("failsafe", [Claim.set("loop", 0, 1.0, "tracking")])}
+# SAFETY-52: the loop's claim reason is computed, so the scan cannot see it.
+_keys |= {why_key("loop", [Claim.set("loop", 0, 0.0, _r)])
+          for _r in (_control.REFUSED, _control.PROBING)}
+check("...including the loop's two battery-refuses reasons, which get words of "
+      "their own", {_control.REFUSED, _control.PROBING} <= _keys)
 check(f"the scan found the claims ({len(_SOURCES)}), including the latched "
       f"per-inverter stop the 2d test saw as 'Other'",
       ("failsafe", "sign disagreement") in _SOURCES and len(_SOURCES) >= 8)
@@ -1487,6 +1618,9 @@ class _H:
     async def call(self, *a, **k):
         return True
 
+    async def set_number(self, *a, **k):
+        return True
+
 
 class _S:
     data = {}
@@ -1549,6 +1683,120 @@ _c.why_words = "Keeping the grid near zero: 0 W"
 check("...and once a cycle decides again, the line is that decision",
       _c.why_words == "Keeping the grid near zero: 0 W")
 
+# UX-11 item 3: past the hold, the Why line says which reading is missing,
+# not "A reading is missing: 0 W". Driven through cycle() for real.
+_c.inputs_bad_since = _dt.now(_tz.utc) - _td(seconds=20)
+asyncio.run(_c.cycle())
+check(f"past the hold, the Why line names the missing reading "
+      f"({_c.why_words!r})",
+      _c.why_words == "No usable meter reading for over 15 seconds: 0 W")
+
+# UX-11 item 2: the red meter row says the log shows why. A meter that goes
+# quiet logs one WARNING per episode, not one per cycle, and INFO when back.
+_c.p1_enabled = True
+_c.p1 = _p1.P1Ingest(max_age_s=30.0)
+
+
+def _meter_at(age_s):
+    import time as _time
+    _c.p1.submit(_p1.make_sample(_p1.SOURCE_HA, 500.0, 0.0, phases=1,
+                                 ingest_mono=_time.monotonic() - age_s))
+
+
+def _quiet_episodes():
+    # Two quiet episodes close together, then a third 10 minutes on.
+    for age in (1, 40, 45, 50, 1, 2, 60, 65, 1):
+        _meter_at(age)
+        asyncio.run(_c.read_inputs())
+    _c._meter_quiet_warned -= _main.METER_QUIET_REPEAT_S
+    for age in (40, 45, 1):
+        _meter_at(age)
+        asyncio.run(_c.read_inputs())
+
+
+_mq = [r for r in _captured("goodwe", _quiet_episodes)
+       if "meter" in r.getMessage().lower()]
+_mq_warn = [r.getMessage() for r in _mq if r.levelno == logging.WARNING]
+_mq_info = [r.getMessage() for r in _mq if r.levelno == logging.INFO]
+_mq_dbg = [r.getMessage() for r in _mq if r.levelno == logging.DEBUG]
+check(f"a meter that goes quiet logs one WARNING per episode, not per cycle, "
+      f"and one INFO when readings return, and a flapping meter at most one "
+      f"pair per {_main.METER_QUIET_REPEAT_S:.0f} s ({_mq_warn}, {_mq_info}, "
+      f"{_mq_dbg})",
+      len(_mq_warn) == 2 and _mq_info == ["Meter readings are back."] * 2
+      and len(_mq_dbg) == 2
+      and _mq_warn[0] == "No meter reading for over 30 seconds, so control "
+                         "waits and then sets the battery to 0 W."
+      and len(words(_mq_warn[0])) <= LINE_WORDS)
+
+# UX-11 item 1: a state file saved before UX-11 holds the long reasons, and
+# they load unchanged. The ERROR line at start shows the code, and the saved
+# text goes on the DEBUG line beside it. Old and new files look the same,
+# because the saved reason's format did not change and it opens with its code.
+_OLD = {
+    "sign": "SIGN_DISAGREEMENT: commanded -1000 W against a meter reading "
+            "+1500 W for 180 s continuously (limits: both over 300 W, held "
+            "over 180 s)",
+    "delivery": "UNIT_NOT_DELIVERING: inverter-a is standing at +2000 W and "
+                "delivering +0 W - 2000 W of its share (100 % of it) is not "
+                "arriving - while inverter-b is not in that state. Nothing "
+                "this add-on reads explains it - held for 121 s continuously "
+                "(limit: over 120 s)",
+    "circ": "CIRCULATING_POWER: 5800 W measured going round between the "
+            "inverters for 61 s, peaking at 6000 W, over the limits of 250 W "
+            "and 60 s. The meter cannot see it, because it nets to zero there",
+}
+
+
+class _OldStore(_S):
+    data = {"unit/inverter-a/sign_tripped": True,
+            "unit/inverter-a/sign_reason": _OLD["sign"],
+            "unit/inverter-a/delivery_tripped": True,
+            "unit/inverter-a/delivery_reason": _OLD["delivery"],
+            "circ_tripped": True, "circ_reason": _OLD["circ"],
+            "site_sign_tripped": True, "site_sign_reason": ""}
+
+
+_restart = []
+_rs = _captured("goodwe", lambda: _restart.append(Controller(
+    {"setpoint_entity": "number.sp", "meter_source": "off", "max_w": 2000},
+    _H(), _OldStore(), _P(), stop=asyncio.Event())))
+_rs_err = [r.getMessage() for r in _rs if r.levelno == logging.ERROR]
+_rs_dbg = [r.getMessage() for r in _rs if r.levelno == logging.DEBUG]
+for _m in _rs_err:
+    print(f"       {_m!r} ({len(words(_m))} words)")
+check("an old state file's long reasons load unchanged, so the stops stay on",
+      _restart and _restart[0].inverters[0].sign_wd.tripped
+      and _restart[0].inverters[0].sign_wd.reason == _OLD["sign"]
+      and _restart[0].inverters[0].delivery_wd.reason == _OLD["delivery"]
+      and _restart[0].circ_wd.reason == _OLD["circ"])
+check("...each ERROR line at start is one plain sentence with the code, at "
+      f"most {LINE_WORDS} words, and a stop saved without a reason says so",
+      len(_rs_err) == 4
+      and all(len(words(m)) <= LINE_WORDS and "(limits" not in m
+              for m in _rs_err)
+      and any(m.endswith(": SIGN_DISAGREEMENT.") for m in _rs_err)
+      and any(m.endswith(": UNIT_NOT_DELIVERING.") for m in _rs_err)
+      and any(m.endswith(": CIRCULATING_POWER.") for m in _rs_err)
+      and any(m.endswith(f": {messages.NO_REASON_SAVED}.") for m in _rs_err))
+check("...and the saved text, figures and all, is on a DEBUG line beside it",
+      all(any(t in m for m in _rs_dbg) for t in _OLD.values()))
+# UX-11 review: `log_level` defaults to info and `events` starts empty, so the
+# saved figures also go on the action log, where `status` shows them.
+_rs_ev = _restart[0].events if _restart else []
+check(f"...and in the action log, which an installer reads at the default "
+      f"log level ({len(_rs_ev)} events)",
+      all(any(t in e for e in _rs_ev) for t in _OLD.values())
+      and any(e.endswith(messages.NO_REASON_SAVED) for e in _rs_ev))
+
+# UX-11 item 4: the action log's clamp count and the shadow-off entry.
+_main_src = (APP / "main.py").read_text(encoding="utf-8")
+check("the action log counts 'engagements' with a real plural and the "
+      "shadow-off entry is plain, not 'engagement(s)' or 'SHADOW MODE OFF -'",
+      "engagement(s)" not in _main_src and "SHADOW MODE OFF" not in _main_src
+      and messages.count(25, "engagement") == "25 engagements"
+      and messages.count(1, "engagement") == "1 engagement")
+
 print()
 if fails:
     print(f"{len(fails)} FAILED")

goodwe_controller/test_price.py

@@ -4412,12 +4412,16 @@ check(f"AC, AND THE WHOLE OF DEFECT 4: 'trading is off' and 'trading is on "
       f"`off (arbitrage_enabled)` for both - naming an option that is `true` "
       f"in the options blob",
       _T11_OFF[0]["Arbitrage"] != _T11_ON_HELD[0]["Arbitrage"])
+# ⚠️ THE "Why" ROW OF THE SAME RENDER, NOT A SECOND READ OF `why_words`. Since
+# UX-10 the waiting sentence counts its seconds when it is read, so reading the
+# property again here, after the other renders above, could cross a second
+# ("1 second" in the row, "2 seconds" now) - CI run 953 on PR #141 did.
 check(f"...and the armed-but-held row carries the CONTROL LOOP'S OWN REASON "
-      f"({_T11_ON_HELD[1].why_words!r}), so the operator is told which input is "
+      f"({_T11_ON_HELD[0]['Why']!r}), so the operator is told which input is "
       f"missing rather than being sent to look at a switch that is already on. "
       f"The reason comes from main.py, not from this file",
-      _T11_ON_HELD[1].why_words
-      and _T11_ON_HELD[1].why_words in _T11_ON_HELD[0]["Arbitrage"])
+      _T11_ON_HELD[0]["Why"].startswith("A reading is missing")
+      and _T11_ON_HELD[0]["Why"] in _T11_ON_HELD[0]["Arbitrage"])
 check(f"...and a cycle that DID reach the step renders the strategy's own "
       f"sentence - {_T11_ON_OK[1].arb.explain()[:50]!r} - rather than a "
       f"re-derivation of it, which is the guarantee OPT-04 put on this row and "

goodwe_controller/test_refusal.py

@@ -0,0 +1,475 @@
+"""Runnable check for SAFETY-52.  `python3 test_refusal.py`
+
+A battery that refuses to give power (the BMS cut-out) used to leave the
+command standing at its last ask for hours: +5000 W for 11.87 h on 2026-09-06,
++1940 W for 3 h 55 m on the bus on 2026-09-19. Now a refused direction is held
+at 0 W, a 200 W probe asks every 15 minutes, and the refusal ends on charge, on
+an answered probe, on stop/start and when another writer takes the wire.
+DECISIONS/SAFETY-52 has the record and the replay figures.
+
+Mutating the modules this file grades? Purge `__pycache__` first, or run
+`tools/mutation_run.sh`, which does - see MUTATION.md.
+
+Parts 1-5 drive the REAL `Controller` (read_inputs + cycle, the real
+SAFETY-37/57 gate) against a scripted battery whose reading stands at exactly
+-4 W for any request >= -4 W at the cut-out, and follows otherwise. Time is the
+`now_mono` handed to `read_inputs()`, so 12 hours run in seconds. `OLD` runs the
+same fixture with `refused=None` forced into `compute()` - release/1.0's law -
+so every "new" figure is shown beside the one it replaces. Part 4 also runs
+`compute()` alone, as the bus saw 09-19 before the gate existed.
+
+Mutations, each on a copy of this tree purged of `__pycache__`, graded with
+`python3 -B` over test_refusal, test_control, test_clamp, test_telemetry and
+test_messages (MUTATION.md). Reds are failing checks, on this tree:
+
+  #   mutation                                                 reds
+  --  -------------------------------------------------------  ---------------
+  R1  control.py: the engage `if` -> `if False and ...`          8 here + 1 clamp
+      (release/1.0's behaviour)
+  R2  control.py: the exit on the other direction removed        1
+  R3  control.py: the exit on an answered probe removed          2
+  R4  control.py: `PROBE_W = 300.0`, above the gate's 250 W       4
+  R5  control.py: the hold keeps the old target, not 0 W          4
+  R6  main.py: `_refusal_from_stale()` returns at once           1
+  R7  main.py: another writer no longer ends the refusal         1
+  R8  main.py: stop/start no longer ends the refusal             1
+  R9  main.py: the probe window never opens                      4
+  R10 main.py: seed from a stale stop below `saturation_w` again,  1
+      not below REFUSED_BELOW_W (review of #141). The same mutation
+      in control.py's engage clause SURVIVES, and that is expected:
+      the delivery exit (R11) ends such a refusal in the same cycle.
+  R11 control.py: only an answered PROBE ends the refusal          2
+      (`and probe` back in the delivery exit; review of #141)
+"""
+
+import asyncio
+import sys
+from datetime import datetime, timedelta, timezone
+
+from app import control, freshness
+from app.control import PROBE_EVERY_S, PROBE_S, PROBE_W, REFUSED, PROBING
+from app.main import Controller, why_key
+from app.arbiter import Claim
+from app import messages
+
+fails = []
+
+
+def check(name, cond):
+    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
+    if not cond:
+        fails.append(name)
+
+
+LIVE = {"max_w": 5000.0, "gain": 0.6, "slew_w": 400.0, "target_grid_w": -10.0,
+        "deadband_w": 15, "step_w": 10}
+TICK = 5.0            # the meter's measured median tick (replay_recorded_loop)
+CUT_W = -4.0          # what the register reads at the cut-out, 2026-09-06
+H = 3600.0
+BASE = datetime(2026, 9, 6, 19, 30, tzinfo=timezone.utc)
+
+
+class _Store:
+    def __init__(self):
+        self.data = {}
+
+    def set(self, *a):
+        pass
+
+    def get_time(self, *a):
+        return None
+
+
+class _Pub:
+    enabled = False
+
+    def publish(self, values):
+        pass
+
+    def close(self):
+        pass
+
+
+class _Hass:
+    """House load, and a battery that follows unless `cut(t)` says it refuses.
+
+    At the cut-out the reading is exactly CUT_W for any request >= CUT_W (a flat
+    register, which is what arms or does not arm the stale-reading gate). A
+    charge request is followed: the pack accepts charge at the floor.
+    `cap(t)` limits discharge while not cut (the 09-06 pack gave 4570 W at 10 %).
+    """
+
+    def __init__(self, load, cut, cap=lambda t: 1e9):
+        self.load, self.cut, self.cap = load, cut, cap
+        self.t = 0.0
+        self.cmd = 0.0
+        self.writes = []
+
+    def batt(self):
+        if self.cut(self.t) and self.cmd >= CUT_W:
+            return CUT_W
+        return float(round(min(self.cmd, self.cap(self.t))))
+
+    async def limits(self, entity):
+        return (-LIVE["max_w"], LIVE["max_w"], 1.0)
+
+    async def number(self, entity, invert=False):
+        b = self.batt()
+        return {"sensor.grid": self.load(self.t) - b, "sensor.soc": 9.0,
+                "sensor.batt": b}.get(entity)
+
+    async def set_number(self, entity, value):
+        self.writes.append((self.t, float(value)))
+        self.cmd = float(value)
+        return True
+
+    async def call(self, *a):
+        return True
+
+
+def run(load, cut, dur, cap=lambda t: 1e9, old=False, events=None):
+    """Real cycles every TICK seconds of `now_mono`. `events[t] = fn(c)` pokes
+    the controller (stop, start, another writer) at that instant.
+
+    Returns [(t, grid, batt, setpoint, i_w, loop_reason, batt_refused)], the
+    number of gate trips, and the controller."""
+    async def go():
+        hass = _Hass(load, cut, cap)
+        c = Controller(dict(
+            {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
+             "soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
+             "meter_source": "off", "heartbeat_s": 10, "stale_input_s": 15},
+            **LIVE), hass, _Store(), _Pub(), stop=asyncio.Event())
+        c.auto = True
+        c.dev_min, c.dev_max = -LIVE["max_w"], LIVE["max_w"]
+        import app.main as m
+        real, real_dt = m.compute, m.datetime
+        clock = [0.0]
+
+        class _DT(real_dt):
+            """`cycle()`'s wall clock follows `now_mono`, so the 15 s input hold
+            and the gate's release run in simulated seconds too."""
+            @classmethod
+            def now(cls, tz=None):
+                return BASE + timedelta(seconds=clock[0])
+
+        m.datetime = _DT
+        if old:
+            m.compute = lambda *a, **k: real(*a, **dict(k, refused=None))
+        rows, trips, was = [], 0, False
+        try:
+            t = 0.0
+            while t < dur:
+                hass.t = clock[0] = t
+                for et, fn in list((events or {}).items()):
+                    if et <= t:
+                        fn(c)
+                        del events[et]
+                await c.read_inputs(now_mono=t)
+                stale = c.inverters[0].batt_fresh.stale
+                trips += stale and not was
+                was = stale
+                await c.cycle()
+                rows.append((t, hass.load(t) - hass.batt(), hass.batt(),
+                             hass.cmd, c.i_w, c.loop_reason, c.batt_refused))
+                t += TICK
+        finally:
+            m.compute, m.datetime = real, real_dt
+        return rows, trips, c
+    return asyncio.run(go())
+
+
+def at(rows, t):
+    return min(rows, key=lambda r: abs(r[0] - t))
+
+
+def longest_hold_h(rows, floor_w=250.0, until=None):
+    """Longest run of one identical request at or above `floor_w`, in hours."""
+    best = cur = 0.0
+    last = None
+    for r in rows:
+        if until is not None and r[0] >= until:
+            break
+        cur = cur + TICK if (r[3] == last and abs(r[3]) >= floor_w) else 0.0
+        last = r[3]
+        best = max(best, cur)
+    return best / H
+
+
+# ==============================================================================
+print("1. the 09-06 shape: a load step takes the request to 5000 W, the battery "
+      "cuts out, the load ends")
+# ==============================================================================
+def load906(t):
+    return 1020 if t < 60 else 4800 if t < 162 else 940 if t < 11.9 * H else -600
+
+
+def cut906(t):
+    return t >= 132
+
+
+def cap906(t):
+    return 4570.0
+
+
+old906, otrips906, _ = run(load906, cut906, 12.2 * H, cap906, old=True)
+new906, ntrips906, c906 = run(load906, cut906, 12.2 * H, cap906)
+for name, rows, tr in (("old", old906, otrips906), ("new", new906, ntrips906)):
+    print(f"    {name}: request at 2 min {at(rows, 120)[3]:+.0f} W, at 6 h "
+          f"{at(rows, 6 * H)[3]:+.0f} W (i_w {at(rows, 6 * H)[4]:+.0f}), longest "
+          f"hold {longest_hold_h(rows):.2f} h, gate trips {tr}")
+check("PRECONDITION: the old law holds +5000 W for the whole night on this "
+      "fixture, as it did on 2026-09-06 - the fixture reproduces the card",
+      at(old906, 6 * H)[3] == 5000.0 and longest_hold_h(old906) > 11.5)
+check("the request reaches the rail before the cut, so the new law starts from "
+      "the same place", at(new906, 120)[3] == 5000.0)
+t0 = next((r[0] for r in new906 if r[5] == REFUSED), None)
+check(f"after the cut the battery-refuses hold takes the request to 0 W within "
+      f"about a minute (at t={t0})", t0 is not None and 132 < t0 <= 132 + 90
+      and at(new906, t0)[3] == 0.0)
+check("...and it stays refused overnight, with the integrator at 0, and no "
+      "request of 250 W or more stands for longer than one probe interval",
+      at(new906, 6 * H)[6] == 1 and at(new906, 6 * H)[4] == 0.0
+      and longest_hold_h(new906, until=11.9 * H) < PROBE_EVERY_S / H)
+check(f"...and the stale-reading gate does not cycle: {ntrips906} trips in "
+      f"12 hours (the old law's latch also kept it quiet: {otrips906})",
+      ntrips906 == 0)
+morning = [r for r in new906 if r[0] >= 11.9 * H + 60]
+check("when solar makes the grid export, the refusal ends and the battery "
+      "charges - the exit on the other direction",
+      morning and morning[-1][6] == 0 and morning[-1][3] < -400)
+
+# ==============================================================================
+print("2. the probe: 200 W every 15 minutes, below the gate's 250 W")
+# ==============================================================================
+probes = []
+for r in new906:
+    if r[3] == PROBE_W and (not probes or r[0] - probes[-1][1] > TICK):
+        probes.append([r[0], r[0]])
+    elif r[3] == PROBE_W:
+        probes[-1][1] = r[0]
+starts = [p[0] for p in probes]
+gaps = [b - a for a, b in zip(starts, starts[1:])]
+print(f"    {len(probes)} probes, first at {starts[0] / 60 if starts else 0:.1f} "
+      f"min, gaps {sorted(set(round(g) for g in gaps))} s, lengths "
+      f"{sorted(set(round(p[1] - p[0] + TICK) for p in probes))} s")
+check("the probe asks exactly PROBE_W, which is below the stale-reading gate's "
+      "threshold, so a probe can never arm it",
+      PROBE_W < freshness.MOVE_W and all(r[3] in (0.0, PROBE_W)
+                                         for r in new906 if r[6] == 1))
+check("there is a probe every 15 minutes through the night (at least 40 in "
+      "11.8 hours), each about 34 s long",
+      len(probes) >= 40 and all(PROBE_EVERY_S <= g <= PROBE_EVERY_S + PROBE_S
+                                + 2 * TICK for g in gaps)
+      and all(PROBE_S - TICK <= p[1] - p[0] + TICK <= PROBE_S + TICK
+              for p in probes))
+check("...and a probe reads as a probe on the loop's own reason row",
+      any(r[5] == PROBING for r in new906))
+
+# ==============================================================================
+print("3. recovery with no sign change: the battery gives power again at 2 h")
+# ==============================================================================
+def cut_rec(t):
+    return 132 <= t < 2 * H
+
+
+oldr, _, _ = run(load906, cut_rec, 2.6 * H, cap906, old=True)
+newr, _, crec = run(load906, cut_rec, 2.6 * H, cap906)
+after_o = [r for r in oldr if 2 * H <= r[0] < 2 * H + 600]
+after_n = [r for r in newr if 2 * H <= r[0] < 2.6 * H]
+exp_o = sum(max(0.0, -r[1]) * TICK for r in after_o) / 3600
+exp_n = sum(max(0.0, -r[1]) * TICK for r in after_n) / 3600
+print(f"    old: worst grid {min(r[1] for r in after_o):+.0f} W, export "
+      f"{exp_o:.0f} Wh; new: worst grid {min(r[1] for r in after_n):+.0f} W, "
+      f"export {exp_n:.0f} Wh")
+check("PRECONDITION: the old law answers a recovering battery with its "
+      "standing +5000 W and exports kilowatts into a 940 W house",
+      min(r[1] for r in after_o) < -3000)
+check("the new law never exports more than the probe while it waits, and the "
+      "answered probe ends the refusal so control resumes",
+      min(r[1] for r in after_n) > -PROBE_W - 50
+      and after_n[-1][6] == 0 and abs(after_n[-1][1] - (-10)) < 100)
+check("...and the event log says the probe was answered",
+      any("BATTERY REFUSES ENDED: the probe was answered" in e
+          for e in crec.events))
+
+# ==============================================================================
+print("4. the 09-19 shape: a stale-reading stop on a frozen ~0 W reading "
+      "is a refusal")
+# ==============================================================================
+# The request steps +1540 -> +1940 and more as the battery cuts out mid-ramp:
+# the request moves after the reading went flat, so the gate trips before the
+# hold's 12 cycles - the ~70 s cycle SAFETY-59 measured. The trip seeds the hold.
+def load919(t):
+    return 1530 if t < 60 else 2400 if t < 1800 else 900 if t < 4 * H else -600
+
+
+def cut919(t):
+    return t >= 65
+
+
+def run919(old):
+    return run(load919, cut919, 4.2 * H, old=old)
+
+
+old919, otr919, _ = run919(True)
+new919, ntr919, c919 = run919(False)
+print(f"    old: gate trips {otr919}, longest hold {longest_hold_h(old919):.2f} h;"
+      f" new: gate trips {ntr919}, longest hold {longest_hold_h(new919):.2f} h")
+check("PRECONDITION: on the old law the gate trips over and over all night "
+      "(the SAFETY-59 cycle) or the request stands for hours",
+      otr919 > 20 or longest_hold_h(old919) > 1.0)
+check("the new law stops after at most two trips and holds 0 W, and the event "
+      "log says the battery refuses",
+      ntr919 <= 2 and at(new919, 2 * H)[3] in (0.0, PROBE_W)
+      and any("BATTERY REFUSES" in e for e in c919.events))
+
+# The same shape through `compute()` alone, as the bus saw it on 2026-09-19
+# before the gate existed: +1540 -> +1940, then a battery giving nothing.
+T = control.Tuning(**LIVE)
+res = {}
+for name, carry in (("old", None), ("new", 0)):
+    # sat_count 2: the +1540 and +1940 cycles were already short (the plan)
+    prev, i_w, sat, ref, cmds = 1940.0, 2340.0, 2, carry, []
+    for _ in range(2000):              # ~2.8 h at 5 s
+        d = control.compute(prev, 1950.0, CUT_W, T, sat, i_w, refused=ref)
+        prev, i_w, sat, ref = d.target_w, d.i_w, d.sat_count, d.refused
+        cmds.append(prev)
+    res[name] = cmds
+print(f"    compute() alone: old stands at {res['old'][-1]:+.0f} W "
+      f"({sum(v == 1940.0 for v in res['old'])} of 2000 cycles at +1940 W), "
+      f"new at {res['new'][-1]:+.0f} W from cycle "
+      f"{res['new'].index(0.0) if 0.0 in res['new'] else None}")
+check("PRECONDITION: without the gate the old law parks at +1940 W, the 7 065 "
+      "identical writes the bus recorded", set(res["old"]) == {1940.0})
+check("...and the new law holds 0 W within SETTLE_CYCLES cycles",
+      res["new"].index(0.0) < control.SETTLE_CYCLES
+      and set(res["new"][control.SETTLE_CYCLES:]) == {0.0})
+
+# ==============================================================================
+print("5. the other exits: stop, start, another writer")
+# ==============================================================================
+def stop(c):
+    c.auto = False
+
+
+def start(c):
+    c.auto = True
+
+
+rows, _, c = run(load906, cut906, 1200, cap906, events={600.0: stop,
+                                                         700.0: start})
+check("stopping control ends the refusal, and starting again begins from "
+      "nothing refused", at(rows, 590)[6] == 1 and at(rows, 650)[6] == 0
+      and at(rows, 705)[6] == 0)
+
+
+def other_writer(c):
+    c.loop_w = 1234.0      # the arbiter gave the wire something the loop did not ask
+
+
+rows, _, c = run(load906, cut906, 900, cap906, events={600.0: other_writer})
+check("another writer on the wire ends the refusal (the arbiter reseed)",
+      at(rows, 590)[6] == 1 and at(rows, 605)[6] == 0)
+
+# ==============================================================================
+print("5b. review of #141: a pack that gives SOME power, and a reading that is "
+      "only late, are not a refusal")
+# ==============================================================================
+# A pack derated to 100-499 W (cold, BMS taper) gives what it can against a
+# larger ask. Engaging below `saturation_w` (500 W) held it at 0 W for ~93 % of
+# the time; "gives nothing" is below the probe's own answer.
+# `compute()` alone: through the Controller the stale-reading gate trips on the
+# flat 300 W first, and that fixture then passes whatever this threshold is.
+def derated(carry, cap_w=300.0, n=1800):
+    prev, i_w, sat, ref, gave, held = 0.0, 0.0, 0, carry, 0.0, 0
+    b = 0.0
+    for _ in range(n):                                   # 2.5 h at 5 s
+        d = control.compute(prev, 2000.0 - b, b, T, sat, i_w, refused=ref)
+        prev, i_w, sat, ref = d.target_w, d.i_w, d.sat_count, d.refused
+        held += bool(ref)
+        b = max(0.0, min(prev, cap_w))                   # one tick late
+        gave += b
+    return gave / n, held
+
+
+T = control.Tuning(**LIVE)
+gave_old, _ = derated(None)
+gave_new, held_new = derated(0)
+print(f"    pack derated to 300 W: old law {gave_old:.0f} W average, new "
+      f"{gave_new:.0f} W, held as refusing on {held_new} cycles")
+check("a pack that gives a steady 300 W against a 2000 W ask is never held at "
+      "0 W as refusing, and keeps giving what the old law got from it",
+      held_new == 0 and gave_new == gave_old and gave_old > 290.0)
+
+
+def lagging(tick, dead_s=6.0, publish_s=10.0, phase_s=5.0, dur=600.0):
+    """`compute()` on a HEALTHY pack whose reading lags: it follows the command
+    of `dead_s` ago, published every `publish_s`. A 1 s meter (`meter_poll_s`
+    1) makes SETTLE_CYCLES only 12 s, shorter than that lag."""
+    prev, i_w, sat, ref, hist, reg, pub = 0.0, 0.0, 0, 0, [], 0.0, -phase_s
+    held = engaged = 0
+    t = 0.0
+    while t < dur:
+        then = [c for tc, c in hist if tc <= t - dead_s]
+        b = then[-1] if then else 0.0
+        if t - pub >= publish_s:
+            reg, pub = b, t
+        d = control.compute(prev, (3000.0 if t >= 10 else 0.0) - b, reg, T,
+                            sat, i_w, refused=ref)
+        engaged += bool(d.refused) and not ref
+        prev, i_w, sat, ref = d.target_w, d.i_w, d.sat_count, d.refused
+        hist.append((t, prev))
+        held += tick * bool(ref)
+        t += tick
+    return engaged, held
+
+
+lag1 = lagging(1.0)
+print(f"    1 s meter, healthy pack with a late reading: engaged {lag1[0]} "
+      f"time(s), held {lag1[1]:.0f} s")
+check("a healthy pack whose reading is merely late is let go as soon as the "
+      "reading shows it delivering - not held until the next probe",
+      lag1[1] <= 10.0)
+
+
+async def _seed(held_w):
+    """`_refusal_from_stale()` on a unit standing at +2000 W, frozen."""
+    c = Controller(dict(
+        {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
+         "soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
+         "meter_source": "off"}, **LIVE),
+        _Hass(lambda t: 0.0, lambda t: False), _Store(), _Pub(),
+        stop=asyncio.Event())
+    c.auto, c.sat_count = True, c.tuning.saturation_cycles
+    inv = c.inverters[0]
+    inv.last_write, inv.batt_fresh.held = 2000.0, held_w
+    c._refusal_from_stale(inv, 100.0)
+    return c.batt_refused
+
+
+check("a stale-reading stop on a register standing at 300 W is not a refusal "
+      "(it gives something), and one at -4 W is",
+      asyncio.run(_seed(300.0)) == 0 and asyncio.run(_seed(CUT_W)) == 1)
+d = control.compute(0.0, 1500.0, 1400.0, T, 0, 0.0, refused=1, probe=False)
+check("...i.e. a held direction that reads 100 W or more ends the refusal "
+      "without waiting for a probe", d.refused == 0)
+
+# ==============================================================================
+print("6. words")
+# ==============================================================================
+w = messages.why(why_key("loop", [Claim.set("loop", 0, 0.0, REFUSED)]), 0.0)
+wp = messages.why(why_key("loop", [Claim.set("loop", 0, PROBE_W, PROBING)]),
+                  PROBE_W)
+print(f"    {w!r} / {wp!r}")
+check("the page's Why line says the battery refuses, not 'Keeping the grid "
+      "near zero: 0 W'", w == "The battery refuses to give power: 0 W")
+check("...and a probe says what it is", wp.startswith("Testing if the battery"))
+check("...and an ordinary loop cycle still reads as the loop",
+      why_key("loop", [Claim.set("loop", 0, 10.0, "tracking")]) == "loop")
+
+print()
+if fails:
+    print(f"{len(fails)} FAILED: " + ", ".join(fails))
+    sys.exit(1)
+print("all checks passed")

goodwe_controller/test_watchdog.py

@@ -1422,14 +1422,28 @@ for _t in range(0, 400, 5):
     _wd_unit.evaluate(-1000.0, 1500.0, True, T0 + timedelta(seconds=_t))
 print(f"       site: {_wd_site.reason!r}")
 print(f"       unit: {_wd_unit.reason!r}")
-check("⚠️ THE SITE INSTANCE'S SENTENCE IS BYTE-IDENTICAL TO THE ONE THAT "
-      "SHIPPED. It is what an operator reads on the status page, what goes in "
-      "the action log and what is persisted to /data/state.json, so it is "
-      "asserted against the literal rather than against a format string",
-      _wd_site.reason == (
-          "SIGN_DISAGREEMENT: commanded -1000 W against a meter reading "
-          "+1500 W for 180 s continuously (limits: both over 300 W, held over "
-          "180 s)"))
+# UX-11: this was a byte pin. The reason is now the evidence: the action
+# log, the DEBUG line and /data/state.json carry it, and the ERROR line quotes
+# only its code (`messages.reason_code()`). So what is pinned is what it says:
+# the code in front, where every saved reason has had it and where the ERROR
+# line finds it, then the command and the reading in that order (habit 14),
+# the time held and both limits.
+_sr = _wd_site.reason
+_order = [_sr.find(t) for t in ("commanded -1000 W", "meter reading +1500 W",
+                                "for 180 s", "both over 300 W",
+                                "held over 180 s")]
+check("⚠️ THE SITE INSTANCE'S REASON SAYS WHAT SHIPPED, IN THE SAME ORDER: "
+      "the code first, then what was commanded, what the meter read, how "
+      f"long, and the two limits ({_order})",
+      _sr.startswith("SIGN_DISAGREEMENT: ")
+      and messages.reason_code(_sr) == "SIGN_DISAGREEMENT"
+      and all(i > 0 for i in _order) and _order == sorted(_order))
+# UX-11 review: the text itself did not change, and it is what state.json
+# holds, so the byte pin stays beside the meaning pin.
+check("...and it is still byte-identical to the one that shipped",
+      _sr == ("SIGN_DISAGREEMENT: commanded -1000 W against a meter reading "
+              "+1500 W for 180 s continuously (limits: both over 300 W, held "
+              "over 180 s)"))
 check("...and the per-unit instance names what it ACTUALLY compared instead. "
       "'against a meter reading' there would be a false statement about which "
       "instrument produced the evidence, on the one line somebody woken by "

goodwe_controller/tools/replay_refusal.py

@@ -0,0 +1,97 @@
+#!/usr/bin/env python3
+"""SAFETY-52's evidence: the 16-day capture replayed with and without the
+battery-refuses hold.  `python3 tools/replay_refusal.py`
+
+Same disturbance, fitted plant and pack floor as `replay_recorded_loop.py`,
+plus main.py's reseed, 15 s input hold and 0 W rules and the REAL SAFETY-37/57
+gate on a register published every 10 s. `old` is `compute()` with
+`refused=None`, which is release/1.0's law unchanged; `new` carries the refusal,
+runs the probe on main.py's clock and seeds a refusal from a gate trip.
+Modelled watt-hours: read the deltas, not the absolutes (see that tool).
+"""
+import math
+import pathlib
+import sys
+
+ADDON = pathlib.Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(ADDON))
+sys.path.insert(0, str(ADDON / "tools"))
+import replay_recorded_loop as R                            # noqa: E402
+from app import freshness, control as NEW                   # noqa: E402
+
+def go(which, rows, fit, publish_s=10.0, stale_input_s=15.0):
+    law = NEW
+    tun = law.Tuning(**R.LIVE)
+    t0, g0, b0, sp0, soc0 = rows[0]
+    pl = R.Plant(*fit[:4], soc0); pl.seed(t0, b0, sp0)
+    st = dict(target=sp0, i_w=sp0, sat=0, loop_w=sp0, ref=0, anchor=None)
+    fr = freshness.Freshness(freshness.BOUNDS["batt"]); reg, reg_t = None, -1e18
+    bad, trips, was = None, 0, False
+    M = dict(und=0.0, imp=0.0, exp=0.0, q={}, run=0.0, mx=0.0, last=None, refh=0.0, probes=0)
+    for i in range(1, len(rows)):
+        t, gr, ba, sp, soc = rows[i]
+        dt = t - rows[i - 1][0]
+        if dt > R.TICK_CAP_S:
+            pl = R.Plant(*fit[:4], soc); pl.seed(t, ba, sp)
+            st.update(target=sp, i_w=sp, sat=0, loop_w=sp, ref=0, anchor=None)
+            fr = freshness.Freshness(freshness.BOUNDS["batt"]); reg_t = -1e18; bad = None
+            continue
+        b = pl.step(t, dt, st["target"]); grid = gr + ba - b
+        if t - reg_t >= publish_s:
+            reg, reg_t = float(round(b)), t
+        fr.submit(reg, st["target"], t); batt = fr.reading()
+        if fr.stale and not was:
+            trips += 1
+            if (which == "new" and not st["ref"] and st["sat"] >= tun.saturation_cycles
+                    and abs(reg) < NEW.REFUSED_BELOW_W and abs(st["target"]) > tun.saturation_w):
+                st["ref"], st["anchor"] = (1 if st["target"] > 0 else -1), t
+        was = fr.stale
+        if batt is None:
+            bad = t if bad is None else bad
+            if t - bad >= stale_input_s:
+                st.update(target=0.0, i_w=0.0, loop_w=None)
+        else:
+            bad = None
+            if st["loop_w"] is None or st["target"] != st["loop_w"]:
+                st["i_w"] = st["target"]
+            if which == "old":
+                d = law.compute(st["target"], grid, batt, tun, st["sat"], st["i_w"],
+                                refused=None)
+            else:
+                probe = False
+                if st["ref"] and st["anchor"] is not None:
+                    h = t - st["anchor"]
+                    if h >= NEW.PROBE_EVERY_S + NEW.PROBE_S: st["anchor"] = t
+                    else: probe = h >= NEW.PROBE_EVERY_S
+                d = law.compute(st["target"], grid, batt, tun, st["sat"], st["i_w"], refused=st["ref"], probe=probe)
+                if d.refused and not st["ref"]: st["anchor"] = t
+                if not d.refused: st["anchor"] = None
+                if probe and d.target_w == NEW.PROBE_W and M["last"] != NEW.PROBE_W: M["probes"] += 1
+                st["ref"] = d.refused
+            st.update(sat=d.sat_count, i_w=d.i_w, loop_w=d.target_w, target=d.target_w)
+        tg = st["target"]
+        if tg - b > 1000: M["und"] += dt / 3600
+        if st["ref"]: M["refh"] += dt / 3600
+        M["imp"] += max(0, grid) * dt / 3.6e6; M["exp"] += max(0, -grid) * dt / 3.6e6
+        k = math.floor(t / 900); a = M["q"].setdefault(k, [0.0, 0.0]); a[0] += grid * dt; a[1] += dt
+        M["run"] = M["run"] + dt if (M["last"] == tg and abs(tg) >= 250) else 0.0
+        M["mx"] = max(M["mx"], M["run"]); M["last"] = tg
+    worst = max(max(0.0, s / d) for s, d in M["q"].values() if d > 600)
+    return trips, M, worst
+
+def main():
+    rows = R.stream()
+    fit = R.fit_plant(rows, stride=7)
+    print(f"fit {[round(x, 2) for x in fit]}, {len(rows)} meter instants")
+    for label, f in (("fitted plant", fit),
+                     ("plant offset 0 W", (fit[0], fit[1], fit[2], 0.0, fit[4]))):
+        for which in ("old", "new"):
+            tr, M, w = go(which, rows, f)
+            print(f"{label:17s} {which}: gate trips {tr:4d} | request >1 kW above "
+                  f"delivery {M['und']:6.1f} h | longest held >=250 W "
+                  f"{M['mx'] / 3600:5.2f} h | import {M['imp']:6.1f} kWh export "
+                  f"{M['exp']:5.1f} kWh | worst 15-min {w:5.0f} W | refused "
+                  f"{M['refh']:5.1f} h, probes {M['probes']}", flush=True)
+
+if __name__ == "__main__":
+    main()