TEL-04: make a missing build() guard fail legibly instead of aborting

Review point from the non-vacuity run. Deleting build()'s "is the net entity
cached at all" guard makes build() raise KeyError rather than return False.
That still failed the suite, but by aborting it with a traceback at whichever
check ran first - a red that costs the next person ten minutes deciding
whether the suite is broken or the code is.

New `built()` helper in test_p1.py wraps build() and turns an escaping
exception into a returned value, so the comparison against True/False fails
by name. Applied only to the ha_signed section; TEL-01's own checks are
untouched.

Mutation 4 before: 0 named checks red, aborted at check 123 of 174.
Mutation 4 after:  2 named checks red - "nothing cached yet builds nothing"
and "an unavailable entity does not build a sample" - all 174 reached.

Still 174 checks, all green, and the other nine mutations are unchanged.

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

.gitattributes

@@ -0,0 +1,10 @@
+# This add-on is deployed to a Linux container. core.autocrlf=true on the
+# authoring box gave the checkout CRLF, so a plain copy shipped CRLF files -
+# run.sh with CRLF is a "bad interpreter" failure, and any hash-based drift
+# check between repo and deployment fails for a reason that has nothing to do
+# with the code. Deploy with:
+#     git -c core.autocrlf=false archive release/1.0 goodwe_controller | tar -x
+# which is how 0.3.0 went out, byte-identical to the blobs.
+* text=auto eol=lf
+*.png binary
+*.gz  binary

goodwe_controller/CHANGELOG.md

@@ -1,5 +1,61 @@
 # Changelog
 
+## Unreleased
+
+**TEL-04.** A third `meter_source`, `ha_signed`, reading **one signed** Home
+Assistant entity: positive = import, negative = export. That is the shape a
+HomeWizard P1 publishes (`sensor.p1_meter_active_power`), and it is the meter
+actually fitted here - which neither TEL-01 transport can read, because
+`ha_dsmr` needs two unsigned registers and refuses a negative one, i.e. every
+exporting telegram. Set `p1_net_entity`, and `p1_phase_net_entities` for the
+per-phase capacity-tariff figures on a three-phase connection.
+
+Everything TEL-01 established is inherited rather than re-implemented - the
+new transport is a subclass of the `ha_dsmr` one overriding only which
+entities it wants and how they become a sample. So ingest timestamping,
+`meter_max_age_s`, `sensor.p1_sample_age_s` recomputed against the clock and
+republished once a second, the plausibility bounds, and `unavailable` /
+`unknown` treated as a *missing reading and never 0 W* all behave identically
+across the three sources.
+
+Still defaults to `off`; an existing install is unaffected until it opts in.
+
+## 0.3.0
+
+**SAFETY-04.** The control law's integrator is now an explicit accumulator,
+bounded independently of the output clamp instead of inheriting whatever
+headroom the clamp happened to leave. It also freezes while the inverter is
+not tracking, rather than continuing to wind up against a command nothing is
+acting on. `integrator_max_w` (default `0`) governs the bound; `0` means
+"follow `max_w`", which is the existing behaviour.
+
+Behaviour is unchanged at the defaults - a 4,928-case equivalence sweep
+against the previous control law confirms it decides identically at
+`integrator_max_w: 0`.
+
+**TEL-01.** P1 meter ingestion, so a Belgian P1's two unsigned registers
+(consumption, injection) no longer need a hand-written signed template
+sensor: the subtraction moves into the add-on, done once and tested. Two
+transports, chosen with the new `meter_source` option: `ha_dsmr` subscribes
+to the DSMR integration over the HA WebSocket, `mqtt_p1` reads a topic.
+Defaults to `off`, which keeps the existing `meter_entity` path untouched -
+nothing changes for an install that does not opt in.
+
+Enabling it publishes `sensor.p1_sample_age_s`: seconds since the newest
+accepted telegram, recomputed against the clock and republished roughly once
+a second rather than only when a telegram lands. That is deliberate - Home
+Assistant only pushes a state on change, so a meter sitting at a genuinely
+constant reading would otherwise look identical to a dead one. Watching the
+age instead means a frozen meter shows a climbing age, not a flat line. The
+firmware watchdog subscribes to this exact entity id.
+
+Known limits, both already in DOCS.md: on `mqtt_p1`, a bridge stuck
+republishing its last telegram still "arrives", so the age cannot detect
+that particular failure - prefer `ha_dsmr` where both are available. And a
+dead P1 meter takes 45 s to reach 0 W commanded (30 s for `meter_max_age_s`
+to call the reading stale, then 15 s of `stale_input_s` on top), which is
+`meter_max_age_s` and `stale_input_s` stacking, not either one alone.
+
 ## 0.2.1
 
 `target_grid_w` (default -10 W): what the meter should rest at. The deadband

goodwe_controller/DOCS.md

@@ -48,6 +48,105 @@ Use the ESP32's readings rather than the inverter's cloud or dongle sensors:
 those serve cached values, and a stale reading here ends the maintenance charge
 phase having charged nothing.
 
+### P1 meter ingestion
+
+`meter_entity` above expects one signed sensor, which usually means a template
+someone wrote by hand. Setting `meter_source` moves the whole derivation into
+the add-on, where it is done once and tested, and replaces `meter_entity`
+entirely.
+
+Which mode you want depends on what your P1 reader publishes, and there are two
+shapes in the wild:
+
+- **Two unsigned registers**, consumption and injection, which is what a Belgian
+  P1 read over DSMR gives you → `ha_dsmr`, or `mqtt_p1` for a bridge. The add-on
+  subtracts them.
+- **One signed figure**, positive = import and negative = export, which is what
+  a HomeWizard P1 gives you (`sensor.p1_meter_active_power`) → `ha_signed`. The
+  add-on splits it. `ha_dsmr` **cannot** read this: it wants two registers and
+  rejects a negative one outright, which is every exporting telegram.
+
+Either way, do not build the missing shape out of template sensors. The point of
+`meter_source` is that the sign convention is derived in one tested place rather
+than in YAML nobody reviews underneath a safety input.
+
+| option | default | meaning |
+|---|---|---|
+| `meter_source` | `off` | `off` keeps `meter_entity`. `ha_dsmr` subscribes to the DSMR integration over the HA WebSocket; `mqtt_p1` reads a topic; `ha_signed` subscribes to one signed entity over the HA WebSocket |
+| `meter_phases` | 1 | 1 or 3. Must match the telegram, or every telegram is rejected and logged |
+| `meter_max_age_s` | 30 | Beyond this the reading is stale and grid power reads as *missing*. On its own it does **not** command 0 W — see the timing note below. It is also the longest a reading is held forward into the 15-minute average |
+| `meter_mqtt_topic` | | `mqtt_p1` only |
+| `p1_import_entity` | | The **unsigned** consumption sensor. Do not point this at a signed template |
+| `p1_export_entity` | | The **unsigned** injection sensor |
+| `p1_phase_import_entities` | `[]` | L1..L3, in order. Needed for the capacity-tariff peak on a three-phase connection |
+| `p1_phase_export_entities` | `[]` | L1..L3, in order |
+| `p1_net_entity` | | `ha_signed` only. The **signed** net-power sensor: `+` import, `-` export |
+| `p1_phase_net_entities` | `[]` | `ha_signed` only. L1..L3, in order, each signed the same way. Needed for the capacity-tariff peak on a three-phase connection |
+
+#### How long a dead meter takes to reach 0 W
+
+`meter_max_age_s` and `stale_input_s` **stack**. They are two different clocks
+and neither one is the whole answer:
+
+| step | option | default |
+|---|---|---|
+| telegrams stop, P1 sample goes stale, grid power starts reading *missing* | `meter_max_age_s` | 30 s |
+| inputs have been missing long enough for the loop to command 0 W | `stale_input_s` | 15 s |
+| **total, meter death → 0 W commanded by this add-on** | | **45 s** |
+
+So in P1 mode `stale_input_s` is *not* "how long inputs may be missing before
+commanding 0 W" measured from the meter dying — it is measured from the moment
+the P1 sample already went stale. Size the pair together: the ESP32's own
+watchdog commands 0 W after ~30 s of silence from this add-on regardless, and
+that layer is unaffected by either option.
+
+There is **no fallback to an inverter-side power figure**, deliberately. The
+inverter's own AC power tracks its battery almost perfectly and the real meter
+hardly at all, so a controller that failed over to it would be regulating
+against its own output while looking healthy.
+
+The `mqtt_p1` payload is one JSON object per telegram, and the schema is strict —
+a key it does not recognise is a telegram from something other than what was
+tested, and guessing a key here means guessing a kilowatt:
+
+```json
+{"import_w": 1234.0,
+ "export_w": 0.0,
+ "phases": [{"import_w": 500, "export_w": 0},
+            {"import_w": 400, "export_w": 0},
+            {"import_w": 334, "export_w": 0}],
+ "timestamp": "2026-08-24T18:00:05+02:00"}
+```
+
+`phases` and `timestamp` are optional; `timestamp` must carry a UTC offset. Where
+it is present it is used for the age, which is what stops a retained message
+replayed on reconnect from presenting a ten-minute-old reading as current.
+
+#### `sensor.p1_sample_age_s`
+
+Published over MQTT discovery whenever a broker is available: **seconds since the
+newest accepted telegram**, refreshed every second rather than only when a
+telegram lands. The ESP32's stale-input watchdog subscribes to this exact entity
+id, so do not rename it.
+
+The reason it is recomputed against the clock is that Home Assistant only pushes
+a state when the state *changes*. A meter sitting at a genuinely constant reading
+emits nothing, which is indistinguishable — to anything watching the value — from
+a meter that has died. Watching the age instead separates the two: it climbs when
+telegrams stop and resets when they arrive, whatever the reading says.
+
+The entity is only created when `meter_source` is not `off`. With P1 ingestion
+disabled there is nothing feeding it, and an age sensor climbing with no ingester
+behind it would trip the firmware watchdog on a system that is working fine.
+
+> **Known limit, `mqtt_p1` only.** The age measures *arrival*, not change. On the
+> two HA WebSocket paths (`ha_dsmr`, `ha_signed`) that is exactly right: a frozen
+> meter emits no `state_changed`, so nothing arrives and the age climbs. On the
+> MQTT path a bridge that is stuck republishing its last telegram keeps arriving,
+> so the age stays near zero and a frozen meter still looks fresh. Detecting
+> *that* needs a change-detector rather than an arrival-detector, and it is not
+> in this version. Prefer an HA WebSocket source where both are available.
+
 ### Control
 
 | option | default | meaning |
@@ -59,11 +158,35 @@ phase having charged nothing.
 | `target_grid_w` | -10 | What the meter should rest at. Negative = a slight export |
 | `step_w` | 10 | Quantisation |
 | `saturation_w` | 500 | Divergence that counts as "the inverter is at a limit" |
-| `saturation_cycles` | 3 | How many consecutive cycles before freezing. **Do not set to 1** |
+| `saturation_cycles` | 3 | How many consecutive cycles before freezing. A cycle is one *changed* meter reading, not a fixed period - see the note below. **Do not set to 1** |
+| `integrator_max_w` | 0 | Bound on the loop's accumulator, and 0 means "same as `max_w`". Caps how much stale error can be waiting to unwind when the sign flips. **Do not raise it above `max_w`** - the output clamp already bounds what is commanded, so the only thing extra headroom buys is more cycles of wrong-direction power after every saturation event. Lowering it below `max_w` is the useful direction |
 | `heartbeat_s` | 10 | Refresh interval; must stay well under the firmware watchdog |
-| `stale_input_s` | 15 | How long inputs may be missing before commanding 0 W |
+| `stale_input_s` | 15 | How long inputs may be missing before commanding 0 W. In P1 mode this clock starts only *after* `meter_max_age_s` has already expired — the two stack, see "How long a dead meter takes to reach 0 W" |
 | `auto_start` | false | Start controlling on boot (only after commissioning) |
 
+#### Saturation is counted in cycles, not seconds
+
+The specification states the saturation window as **"> 10 s"**. This add-on counts
+**cycles** instead, and that is a deliberate, accepted deviation rather than an
+oversight - the acceptance criterion is not met as literally written.
+
+A cycle here is one *changed* meter reading: the controller only runs the loop when the
+meter value differs from the previous poll. At the reference P1's ~5 s update rate the
+default of 3 cycles is usually around 15 s, but there is **no guaranteed wall-clock
+window** - a meter that repeats the same value stalls the counter for as long as it
+repeats.
+
+Two reasons that is acceptable:
+
+- the control law is a pure function with no clock, which is what makes it testable
+  without hardware, and a seconds-based window would have to live in the controller;
+- a stalled counter is a detection-latency limit and not a runaway risk. The condition
+  that stalls it - an unchanging meter - stops the whole loop, so nothing accumulates
+  while it is stalled.
+
+If a guaranteed window matters on your site, raise `saturation_cycles` for a fast meter,
+and treat the figure as "N meter updates" rather than "N seconds".
+
 #### Why `target_grid_w` is not zero
 
 The deadband is a one-way ratchet: any resting point inside it holds until

goodwe_controller/app/control.py

@@ -28,6 +28,22 @@ class Tuning:
     step_w: int = 10
     saturation_w: float = 500.0
     saturation_cycles: int = 3
+    # The integrator's own bound. None means "follow max_w", which is the
+    # default and the recommended setting.
+    #
+    # ⚠️ DO NOT RAISE THIS ABOVE max_w without a measurement to justify it.
+    # Every watt of integrator above the rail is a watt of wind that has to be
+    # burned off before the command can start moving the other way, i.e. extra
+    # cycles of discharge into an already-exporting meter after every
+    # saturation event. Measured on the closed-loop sim, 4000 W load dropped to
+    # 0: at integrator_max_w == max_w the command is 1000 W two cycles later; at
+    # 1.5x max_w it is 1800 W. The output clamp already bounds what reaches the
+    # wire, so headroom here buys nothing but unwind latency.
+    #
+    # It is a separate key because it has to be able to be SMALLER than max_w,
+    # which is the only direction that buys anything: it caps unwind latency
+    # below what the rail implies. Merging it into max_w would take that away.
+    integrator_max_w: float | None = None
     # What the meter should rest at, in W. Negative = a slight export.
     # ⚠️ The deadband is a one-way ratchet: any resting point inside it holds
     # forever, and the meter's IMPORT register counts every positive one with
@@ -44,6 +60,10 @@ class Decision:
     sat_count: int
     frozen: bool
     reason: str
+    # The integrator AFTER this cycle, before the output clamp, the slew limit
+    # and quantisation. Carry it back in as `i_w` next cycle; that is what keeps
+    # it a separate quantity from the command.
+    i_w: float = 0.0
 
 
 def compute(
@@ -52,12 +72,14 @@ def compute(
     actual_w: float,
     tuning: Tuning,
     sat_count: int = 0,
+    i_w: float | None = None,
 ) -> Decision:
     """One control cycle. A cycle is one meter update (~5 s on a HomeWizard P1).
 
     `prev_w`   what we last commanded
     `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`
     """
     reason = "tracking"
 
@@ -71,6 +93,21 @@ def compute(
     # on EVERY large correction, because the plant itself needs 3-6 s to settle
     # while a cycle is ~5 s. Requiring N consecutive saturated cycles is what
     # lets slew be larger than saturation_w.
+    #
+    # The spec states this window twice and differently: "> 10 s" (§11.2) and
+    # "3 samples" (§10.3). This counts CYCLES, and a cycle is not a unit of
+    # time: run_control() calls cycle() only when the meter value CHANGES
+    # (`if self.grid != last_grid`), so three cycles is three distinct meter
+    # readings and nothing more. At the reference P1's ~5 s update rate that is
+    # usually ~15 s, but there is no upper bound on it - a meter that repeats a
+    # value stalls the counter.
+    #
+    # That is a detection-latency limit, not a windup hazard: the same
+    # condition that stalls the counter stalls the whole loop, so nothing
+    # accumulates in the meantime either. If a wall-clock window is ever
+    # required, it belongs in Controller (which has a clock) and not here.
+    # ponytail: this function is worth keeping clockless; the ceiling is that
+    # saturation_cycles cannot express a guaranteed number of seconds.
     saturated_now = abs(prev_w - actual_w) > tuning.saturation_w
     sat_count = min(sat_count + 1, 10) if saturated_now else 0
     frozen = sat_count >= tuning.saturation_cycles
@@ -83,11 +120,82 @@ def compute(
     # 70 s" is how a healthy loop is recognised at a glance, and a command
     # frozen where it should not be is how two real bugs were caught.
     error = grid_w - tuning.target_grid_w
+
+    # --- the integrator ----------------------------------------------------
+    # This loop is in velocity form: the accumulator IS the commanded power, so
+    # "the integrator" and "the output" were one variable and could not be
+    # bounded apart. `i_w` is that accumulator made explicit; main.py carries it
+    # between cycles, which is what turns the two clamps into two limits.
+    #
+    # Passing i_w=None re-seeds it from the last command every cycle. With
+    # integrator_max_w following max_w that reduces this function to the exact
+    # velocity form it replaced, frozen branch included - asserted by an
+    # exhaustive comparison against a transcription of the old law in
+    # test_control.py, not by inspection. Break either the gate or the bound
+    # below and that test is what tells you the equivalence went with it.
+    if i_w is None:
+        i_w = float(prev_w)
+    limit = tuning.max_w if tuning.integrator_max_w is None else tuning.integrator_max_w
+
     if abs(error) < tuning.deadband_w:
-        want = prev_w
         reason = "deadband"
     else:
-        want = prev_w + tuning.gain * error
+        moved = i_w + tuning.gain * error
+        # ⚠️ Freeze means "may not wind FURTHER in the direction it is already
+        # pushing". It may fall, cross zero, or reverse outright.
+        #
+        # It must NOT be encoded as "only corrections that shrink |i_w|": that
+        # is unsatisfiable for BOTH signs of error whenever the correction is
+        # larger than twice the integrator, i.e. every time the integrator is
+        # near zero. The loop then sits at its last value forever, because what
+        # clears the freeze is the inverter tracking again and not-tracking is
+        # the definition of saturation. Measured on that encoding: 0 W held
+        # indefinitely into a 2 kW import, where this form recovers next cycle.
+        #
+        # This is the same asymmetric rule the output freeze uses below, which
+        # has been in service on real hardware. It is applied here as well
+        # because the requirement is that the INTEGRATOR stop accumulating, not
+        # only the command.
+        #
+        # ⚠️ EXACTLY ZERO IS ITS OWN CASE, and it must be handled explicitly
+        # rather than falling into one of the two branches. "May not wind
+        # further in the direction it is already pushing" has no referent at
+        # zero: nothing is wound, and neither direction is "further". Writing
+        # this as `if i_w > 0 ... else ...` silently files zero under
+        # rising-only and permanently blocks the first push toward charging -
+        # the same deadlock as the shrink-only encoding above, mirrored in sign,
+        # and reachable because main.py resets i_w to exactly 0.0 on every stop
+        # and every reseed. Measured before the fix: 12 800 of 25 920 frozen
+        # ticks at i_w == 0.0 held the integrator, 8 304 of them changing the
+        # emitted command, worst case abandoning a 2 kW charge into a 4 kW
+        # export.
+        #
+        # Freezing at zero would also be pointless: the freeze exists to stop
+        # accumulation running away, and a first step from zero is bounded by
+        # the gain, the output clamp and the slew limit like any other.
+        if not frozen or i_w == 0.0:
+            i_w = moved
+        elif i_w > 0:
+            i_w = min(moved, i_w)
+        else:
+            i_w = max(moved, i_w)
+
+    # ⚠️ Applied EVERY cycle, frozen or not: the freeze is conditional, this
+    # bound is not. It is what makes the worst-case unwind time finite and
+    # knowable instead of a function of how long the error happened to stand.
+    bounded = max(-limit, min(limit, i_w))
+    if bounded != i_w:
+        # ⚠️ SAFETY-03 (alarm whenever the loop winds into a rail) must watch
+        # for THIS, not for "clamped" below. At the default limit == max_w the
+        # integrator bound is reached first and the command derived from it can
+        # then never exceed max_w, so "clamped" is unreachable on a default
+        # install - it survives only for a configuration that deliberately lets
+        # the integrator run above the rail. Two reasons rather than one
+        # because the two events want different alarms: "i-clamped" is the loop
+        # winding, "clamped" is a command that came out over the rating anyway.
+        reason = "i-clamped"
+    i_w = bounded
+    want = i_w
 
     # ⚠️ Maintenance shaping (charge-only, cheap-window floor) used to live
     # here. It now belongs to arbiter.py as limit claims, so that precedence
@@ -123,7 +231,7 @@ def compute(
     step = max(1, int(tuning.step_w))
     target = round(target / step) * step
 
-    return Decision(float(target), sat_count, frozen, reason)
+    return Decision(float(target), sat_count, frozen, reason, float(i_w))
 
 
 def maintenance_charge_floor(

goodwe_controller/app/main.py

@@ -39,6 +39,7 @@ from .control import Tuning, compute, maintenance_charge_floor, peak_at_risk
 from .hass import HomeAssistant
 from .maintenance import IDLE, MaintConfig, Maintenance
 from .mqtt import MqttPublisher
+from .p1 import P1Ingest, build_source, is_enabled
 from . import web
 
 OPTIONS_PATH = "/data/options.json"
@@ -70,6 +71,11 @@ class Controller:
             step_w=int(opts.get("step_w", 10)),
             saturation_w=float(opts.get("saturation_w", 500)),
             saturation_cycles=int(opts.get("saturation_cycles", 3)),
+            # 0 / unset means "follow max_w", which is the recommended
+            # value. Read the note in control.py before raising it above
+            # max_w: every watt above the rail is unwind latency.
+            integrator_max_w=(float(opts["integrator_max_w"])
+                              if opts.get("integrator_max_w") else None),
         )
         self.maint = Maintenance(
             MaintConfig(
@@ -85,10 +91,19 @@ class Controller:
             store,
         )
 
+        # P1 ingestion (TEL-01). `meter_source: off` keeps the original
+        # single-entity meter_entity path, so an existing install is unchanged
+        # until it opts in.
+        self.p1 = P1Ingest(phases=int(opts.get("meter_phases", 1)),
+                           max_age_s=float(opts.get("meter_max_age_s", 30)))
+        self.p1_enabled = is_enabled(opts)
+
         # live state
         self.auto = bool(store.data.get("auto", opts.get("auto_start", False)))
         self.target = 0.0
         self.sat_count = 0
+        self.i_w = 0.0        # the loop's integrator, carried between cycles
+        self.loop_w = None    # what the loop asked for last cycle, or None
         self.reason = "starting"
         self.grid = self.soc = self.batt = None
         self.peak_fc = None
@@ -114,6 +129,16 @@ class Controller:
     # -- io ------------------------------------------------------------------
     async def read_inputs(self) -> None:
         o = self.o
+        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
+            # existing "inputs missing -> command 0 W" path below. There is
+            # deliberately NO fallback to an inverter-side figure: the
+            # inverter's own AC power correlates 0.998 with battery power and
+            # 0.09 with the real meter, so a controller that failed over to it
+            # would be regulating against its own output.
+            self.grid = self.p1.net_w
+        else:
             self.grid = await self.hass.number(o.get("meter_entity", ""),
                                                bool(o.get("meter_invert")))
         self.soc = await self.hass.number(o.get("soc_entity", ""))
@@ -231,17 +256,36 @@ class Controller:
             # wish. If something outranked the loop, that is what the hardware
             # actually did, and the controller must track reality or it jumps
             # the moment it regains control.
+            #
+            # ⚠️ The integrator has to track the same reality, and it is no
+            # longer prev_w, so it needs saying out loud: if the arbiter did not
+            # give the loop what it asked for last cycle, the loop's
+            # accumulated error belongs to a command that never happened.
+            # Re-seed from what the hardware was actually told. This is the
+            # failsafe case too - every layer-1 stop resolves to 0 W, so
+            # entering failsafe re-seeds the integrator to zero and the first
+            # cycle after release starts from zero instead of dumping the whole
+            # stale period as power.
+            if self.loop_w is None or self.target != self.loop_w:
+                self.i_w = self.target
             decision = compute(
                 prev_w=self.target,
                 grid_w=self.grid,
                 actual_w=self.batt,
                 tuning=self.tuning,
                 sat_count=self.sat_count,
+                i_w=self.i_w,
             )
             if decision.frozen and self.sat_count < self.tuning.saturation_cycles:
                 self.log_event(f"saturation freeze ({self.target:.0f} W vs {self.batt:.0f} W)")
             self.sat_count = decision.sat_count
+            self.i_w = decision.i_w
+            self.loop_w = decision.target_w
             claims.append(Claim.set("loop", P_LOOP, decision.target_w, decision.reason))
+        else:
+            # Stopped or blind: no accumulation may survive the outage.
+            self.i_w = 0.0
+            self.loop_w = None
 
         resolution = resolve(claims)
         if resolution.contradiction:
@@ -280,14 +324,26 @@ class Controller:
             await asyncio.sleep(1)
 
     def publish(self) -> None:
-        self.mqtt.publish({
+        values = {
             "setpoint": self.target,
             "grid": self.grid,
             "battery": self.batt,
             "soc": self.soc,
             "phase": self.maint.phase,
             "status": "running" if self.auto else "stopped",
-        })
+        }
+        # ⚠️ ONLY when P1 ingestion is actually running. The ESP32's stale-input
+        # watchdog subscribes to sensor.p1_sample_age_s and forces the layer-1
+        # failsafe once it reaches max_age_s. With meter_source off there is no
+        # ingester feeding it, so published_age_s would be time-since-startup
+        # climbing without bound - i.e. every existing install would cross the
+        # threshold within 30 s and pin its inverter at 0 W forever. Publishing
+        # nothing leaves the entity non-existent, which is the status quo and
+        # what has_state() in the firmware is checking for.
+        if self.p1_enabled:
+            # Recomputed here, once a second, on purpose - see P1Ingest.
+            values["p1_age"] = round(self.p1.published_age_s, 1)
+        self.mqtt.publish(values)
 
     async def shutdown(self) -> None:
         """Deterministic wind-down. Do not skip this."""
@@ -300,11 +356,27 @@ class Controller:
     def checks(self) -> list:
         o = self.o
         out = []
-        for label, value, entity in (
-            ("grid power", self.grid, o.get("meter_entity")),
-            ("battery SoC", self.soc, o.get("soc_entity")),
-            ("battery power", self.batt, o.get("batt_entity")),
-        ):
+        if self.p1_enabled:
+            age = self.p1.published_age_s
+            if self.p1.stale:
+                out.append({"ok": False, "warn": False,
+                            "text": f"P1 meter ({o.get('meter_source')}): no reading for "
+                                    f"{age:.0f} s (limit {self.p1.max_age_s:.0f} s)"
+                                    + (f" - last error: {self.p1.last_error}"
+                                       if self.p1.last_error else "")})
+            else:
+                out.append({"ok": True, "warn": False,
+                            "text": f"P1 meter ({o.get('meter_source')}): {self.p1.net_w:g} W, "
+                                    f"{age:.0f} s old, {self.p1.samples} telegrams, "
+                                    f"{self.p1.parse_errors} rejected"})
+        rows = [("battery SoC", self.soc, o.get("soc_entity")),
+                ("battery power", self.batt, o.get("batt_entity"))]
+        if not self.p1_enabled:
+            # In P1 mode the check above replaces this one; leaving both in
+            # would report "no entity configured" for a meter_entity that is
+            # correctly unused, i.e. a permanent false NOT READY.
+            rows.insert(0, ("grid power", self.grid, o.get("meter_entity")))
+        for label, value, entity in rows:
             if not entity:
                 out.append({"ok": False, "warn": False, "text": f"{label}: no entity configured"})
             elif value is None:
@@ -431,6 +503,7 @@ async def amain() -> None:
         # observability, and the battery does not care. Caught broadly and on
         # purpose: this crashed the add-on once already (paho 1.x vs 2.x) and
         # took the control loop down with it.
+        broker = None
         try:
             broker = await hass.mqtt_service()
             pub = MqttPublisher(
@@ -438,6 +511,7 @@ async def amain() -> None:
                 broker.get("port", 1883) if broker else 1883,
                 broker.get("username") if broker else None,
                 broker.get("password") if broker else None,
+                omit=() if is_enabled(opts) else ("p1_age",),
             )
         except Exception as err:  # noqa: BLE001
             _LOG.warning("MQTT unavailable (%s) - continuing without status entities", err)
@@ -458,11 +532,22 @@ async def amain() -> None:
             with contextlib.suppress(NotImplementedError):
                 loop.add_signal_handler(sig, stop.set)
 
-        task = asyncio.create_task(controller.run_control())
+        tasks = [asyncio.create_task(controller.run_control())]
+
+        # P1 ingestion runs as its own long-lived task. ⚠️ It must not be driven
+        # off the control loop: telegrams arrive every ~5 s and the loop would
+        # decimate them, so the 15-minute average - the capacity-tariff billing
+        # unit - would be computed from a fraction of the data.
+        p1_source = build_source(opts, controller.p1, session, broker)
+        if p1_source is not None:
+            tasks.append(asyncio.create_task(p1_source.run()))
+
         await stop.wait()
 
         await controller.shutdown()
+        for task in tasks:
             task.cancel()
+        for task in tasks:
             with contextlib.suppress(asyncio.CancelledError):
                 await task
         await runner.cleanup()

goodwe_controller/app/mqtt.py

@@ -45,6 +45,14 @@ SENSORS = [
     ("soc", "goodwe_battery_soc", "Battery SoC", "%", "battery", "measurement", None),
     ("phase", "goodwe_maintenance_phase", "Maintenance phase", None, None, None, "mdi:battery-sync"),
     ("status", "goodwe_controller_status", "Controller status", None, None, None, "mdi:heart-pulse"),
+    # ⚠️ This one deliberately breaks the goodwe_ prefix above: the entity id
+    # must be exactly `sensor.p1_sample_age_s`, because SAFETY-01's firmware
+    # watchdog subscribes to that literal id and the ENV-01 simulation rig
+    # asserts on it. Renaming it silently disarms a safety layer. It is seconds
+    # since the newest accepted P1 telegram, republished every second so that a
+    # meter frozen at a constant value still shows a climbing age - which is the
+    # false-trip that this entity exists to remove.
+    ("p1_age", "p1_sample_age_s", "P1 sample age", "s", "duration", "measurement", None),
 ]
 
 BASE = "goodwe_ctl"
@@ -52,7 +60,12 @@ AVAILABILITY = f"{BASE}/availability"
 
 
 class MqttPublisher:
-    def __init__(self, host, port, username=None, password=None):
+    def __init__(self, host, port, username=None, password=None, omit=()):
+        # `omit` drops sensor keys from discovery entirely. ⚠️ Announcing a
+        # sensor that nothing will ever publish to is not harmless here:
+        # p1_sample_age_s is a watchdog input, and an entity that exists but is
+        # never fed is a worse signal than one that does not exist at all.
+        self.omit = set(omit)
         self.enabled = mqtt is not None and bool(host)
         self.client = None
         if not self.enabled:
@@ -86,6 +99,8 @@ class MqttPublisher:
 
     def _announce(self) -> None:
         for key, object_id, name, unit, dev_class, state_class, icon in SENSORS:
+            if key in self.omit:
+                continue
             cfg = {
                 "name": name,
                 "object_id": object_id,

goodwe_controller/app/p1.py

@@ -0,0 +1,826 @@
+"""P1 meter ingestion - the only authoritative measurement of real grid exchange.
+
+Everything downstream trusts this module: the safety checks, the capacity-tariff
+peak, the optimizer, the control loop's sign. So three things happen here and
+nowhere else.
+
+  1. The IMPORT/EXPORT DERIVATION. A Belgian P1 read over DSMR exposes two
+     UNSIGNED registers - consumption and injection. Net power is
+     `import_w - export_w`, positive = import, and that subtraction is done
+     exactly once, here (spec §5.2: "the derivation is the EMS's job, not a
+     template the user has to write"). A second copy of it somewhere else is a
+     second chance to invert the control loop.
+
+     Some P1 readers - the HomeWizard P1 among them - publish the OTHER shape:
+     one SIGNED figure, positive = import, and no unsigned registers at all.
+     `split_signed()` fans that back out into the same two magnitudes, so there
+     is still exactly one internal representation and one sign convention. ⚠️ It
+     lives here, next to the subtraction, for the same reason the subtraction
+     does: the moment a user is asked to write two template sensors that split a
+     signed value, the sign convention is back in unreviewed YAML underneath a
+     safety input, which is precisely what §5.2 moved into the EMS.
+
+  2. THE INGEST TIMESTAMP. Every accepted sample is stamped on arrival. A value
+     with no age is a value that cannot be trusted (§5.2), and staleness is the
+     failsafe trigger (§11.2).
+
+  3. VALIDATION. This is untrusted external data at the edge of a safety chain.
+     A malformed telegram must not become a plausible-looking number, and it
+     must never resolve to 0 W - a fabricated zero is indistinguishable from a
+     balanced house and defeats the very staleness trigger this module feeds.
+
+⚠️ There is deliberately NO fallback to an inverter-side power figure. The
+inverter's own AC power correlates 0.998 with battery power and 0.09 with the
+real meter (§5.1) - regulating on it means regulating against your own output.
+When the transport dies the correct behaviour is a gap: no sample, a growing
+age, and the existing "inputs missing -> command 0 W" path in main.py.
+"""
+
+import asyncio
+import json
+import logging
+import math
+import os
+import time
+from dataclasses import dataclass
+from datetime import datetime, timezone
+
+import aiohttp
+
+try:
+    import paho.mqtt.client as mqtt
+except ImportError:  # pragma: no cover - container always has it
+    mqtt = None
+
+_LOG = logging.getLogger("goodwe.p1")
+
+SOURCE_HA = "ha_dsmr"
+SOURCE_MQTT = "mqtt_p1"
+SOURCE_HA_SIGNED = "ha_signed"
+
+QUARTER_S = 900
+
+# ⚠️ Plausibility ceiling, not a clamp - anything above it is rejected as an
+# anomaly rather than averaged in. Chosen to sit above the largest Belgian
+# residential connection (3x63 A ~ 43 kW) and BELOW 65535: §20 open question 5
+# records an HA sensor reporting 64954 for -582 W, i.e. an unsigned 16-bit
+# register decoded without its sign. That corruption reads as a perfectly
+# plausible 65 kW if you only bound it at "some big number".
+PLAUSIBLE_MAX_W = 50_000.0
+
+
+# --------------------------------------------------------------------------- #
+# the sample
+# --------------------------------------------------------------------------- #
+class P1Error(ValueError):
+    """A telegram that must be rejected rather than believed."""
+
+
+@dataclass(frozen=True)
+class P1Sample:
+    """One telegram, validated, derived and stamped.
+
+    Frozen on purpose: this object is handed to readers on other tasks (and,
+    for the MQTT transport, produced on paho's network thread). Immutability is
+    what makes "read the latest sample" safe without a lock.
+    """
+
+    ingest_ts: datetime            # tz-aware UTC, set at ingest
+    ingest_mono: float             # time.monotonic() at ingest - see age_s()
+    telegram_ts: datetime | None   # from the telegram, where the source has one
+    source: str                    # SOURCE_HA | SOURCE_MQTT | SOURCE_HA_SIGNED
+    import_w: float                # unsigned magnitude, as the meter reports it
+    export_w: float                # unsigned magnitude
+    net_w: float                   # import_w - export_w  (+ import, - export)
+    per_phase_w: tuple[float, ...] | None          # signed net, len == phases
+    per_phase_import_w: tuple[float, ...] | None   # offtake only, for the tariff
+
+    def age_s(self, now_mono: float | None = None,
+              now_utc: datetime | None = None) -> float:
+        """Seconds since this sample was ingested, never negative.
+
+        ⚠️ Measured with time.monotonic(), not the wall clock. An NTP step on a
+        Pi that just booted moves the wall clock by minutes; using it here would
+        either fake a stale meter or, worse, hide a real one.
+
+        Where the telegram carries its own timestamp we take the WORSE of the
+        two ages. That is what stops an MQTT retained message - replayed on
+        reconnect with a fresh receive time - from presenting a ten-minute-old
+        reading as brand new.
+        """
+        now_mono = time.monotonic() if now_mono is None else now_mono
+        age = max(0.0, now_mono - self.ingest_mono)
+        if self.telegram_ts is not None:
+            now_utc = datetime.now(timezone.utc) if now_utc is None else now_utc
+            age = max(age, (now_utc - self.telegram_ts).total_seconds())
+        return max(0.0, age)
+
+
+def _watts(value, what: str) -> float:
+    """Parse one power figure, or raise. Never returns a substituted default.
+
+    ⚠️ Strings are refused even when float() would happily take them. A JSON
+    telegram carrying "1200" where a number belongs is a payload from a source
+    that is not the one we validated against, and the next surprise it has may
+    not be a benign one. Transports that legitimately deal in text (HA entity
+    states are always strings) convert before they get here, so this stays the
+    strict edge for structured payloads.
+    """
+    if isinstance(value, (bool, str, bytes)) or value is None:
+        raise P1Error(f"{what}: not a number ({value!r})")
+    try:
+        out = float(value)
+    except (TypeError, ValueError):
+        raise P1Error(f"{what}: not a number ({value!r})") from None
+    if not math.isfinite(out):
+        raise P1Error(f"{what}: not finite ({value!r})")
+    if abs(out) > PLAUSIBLE_MAX_W:
+        raise P1Error(f"{what}: {out:g} W is outside plausible meter range")
+    return out
+
+
+def split_signed(net_w) -> tuple[float, float]:
+    """One signed figure -> the (import, export) magnitudes the module speaks.
+
+    The inverse of make_sample's subtraction, and the easy direction: no second
+    register to disagree with, so there is nothing to mix a fresh reading with a
+    stale one. `+` is import, `-` is export - verified in test_p1.py against real
+    captured readings from this house's own meter, not against a datasheet.
+
+    ⚠️ Exactly one of the two comes out non-zero. Splitting into `(max(v,0),
+    max(-v,0))` rather than clamping keeps `import_w - export_w == v` exactly, so
+    the signed value the meter published survives the round trip bit for bit -
+    a control loop must not be steered by a number that changed on the way in.
+
+    ⚠️ Validation is `_watts`, the same gate the unsigned path uses: NaN,
+    infinity, non-numbers and the §20 open-question-5 unsigned-decode
+    contamination (64954 for -582 W) are all refused here rather than believed.
+    A signed source makes that check MORE important, not less - on this path
+    64954 is not obviously wrong the way a negative "unsigned" register is.
+    """
+    v = _watts(net_w, "net")
+    return (v, 0.0) if v >= 0 else (0.0, -v)
+
+
+def make_sample(source: str, import_w, export_w, *, phases: int,
+                phase_import_w=None, phase_export_w=None,
+                telegram_ts: datetime | None = None,
+                ingest_ts: datetime | None = None,
+                ingest_mono: float | None = None) -> P1Sample:
+    """Validate, derive net power, stamp. Raises P1Error on anything doubtful.
+
+    `import_w`/`export_w` are the two unsigned Belgian registers. Per-phase
+    figures are equally unsigned and equally split, so each phase gets the same
+    derivation.
+    """
+    imp = _watts(import_w, "import")
+    exp = _watts(export_w, "export")
+    # ⚠️ Both registers are magnitudes. A negative one means the upstream
+    # already applied a sign we are about to apply again - reject it rather
+    # than silently double-signing the control loop.
+    if imp < 0 or exp < 0:
+        raise P1Error(f"unsigned registers cannot be negative (import={imp:g} export={exp:g})")
+
+    per_phase = per_phase_import = None
+    if phase_import_w is not None or phase_export_w is not None:
+        pi = list(phase_import_w or [])
+        pe = list(phase_export_w or [0.0] * len(pi))
+        if len(pi) != phases or len(pe) != phases:
+            raise P1Error(
+                f"phase count mismatch: telegram has {len(pi)} import / {len(pe)} export "
+                f"phases, meter_phases is {phases}")
+        vals = [_watts(a, f"L{i + 1} import") - _watts(b, f"L{i + 1} export")
+                for i, (a, b) in enumerate(zip(pi, pe))]
+        per_phase = tuple(vals)
+        per_phase_import = tuple(max(v, 0.0) for v in vals)
+
+    if telegram_ts is not None and telegram_ts.tzinfo is None:
+        raise P1Error("telegram timestamp has no timezone")
+
+    return P1Sample(
+        ingest_ts=ingest_ts or datetime.now(timezone.utc),
+        ingest_mono=time.monotonic() if ingest_mono is None else ingest_mono,
+        telegram_ts=telegram_ts,
+        source=source,
+        import_w=imp,
+        export_w=exp,
+        net_w=imp - exp,
+        per_phase_w=per_phase,
+        per_phase_import_w=per_phase_import,
+    )
+
+
+# --------------------------------------------------------------------------- #
+# the 15-minute average
+# --------------------------------------------------------------------------- #
+@dataclass(frozen=True)
+class QuarterBlock:
+    start: datetime                             # UTC, aligned to :00/:15/:30/:45
+    offtake_avg_w: float                        # billed figure: net offtake only
+    per_phase_offtake_avg_w: tuple[float, ...] | None
+
+
+class QuarterAverager:
+    """Time-weighted average of net offtake over clock-aligned 15-min blocks.
+
+    Samples arrive irregularly (~1-10 s), so a plain mean over samples would
+    weight a burst of fast telegrams the same as a slow one and produce a figure
+    that is not the billed quantity. Each sample's value is therefore HELD until
+    the next arrives and integrated over that interval: sum(value * dt) / dt.
+
+    ⚠️ Only OFFTAKE is accumulated (§9.1) - the capacity tariff bills the highest
+    quarter-hour average offtake, and a quarter of pure export averages to 0 kW,
+    not to a negative one. The signed series stays available for control; this
+    accumulator is for the meter's bill.
+
+    ⚠️ Blocks are found by flooring epoch seconds to 900. That IS clock-aligned
+    and DST-proof for Belgium, because every offset in that tz is a whole number
+    of hours, so a 900 s grid in UTC lands on :00/:15/:30/:45 local before and
+    after a transition - no tz database, no DST special case.
+    ponytail: the ceiling is a timezone with a sub-hour offset (India +05:30,
+    Nepal, Chatham). Those need real tz-aware boundary maths; upgrade path is to
+    compute the boundary with zoneinfo instead of the modulo, everything else
+    here is unchanged.
+
+    A sample that straddles a boundary is split at the boundary and its two
+    halves credited to the two blocks, never attributed wholly to either.
+    """
+
+    def __init__(self, phases: int = 1, max_hold_s: float = 30.0):
+        self.phases = phases
+        # ⚠️ How long one sample may be held forward before the series is
+        # treated as a gap rather than a plateau. Without this the meter can die
+        # while importing 5 kW, come back ten minutes later, and the hold-forward
+        # credits 5 kW x 600 s to the capacity-tariff accumulator - a fabricated
+        # peak, on a permanent record, from data that was never measured. Set
+        # from meter_max_age_s: the point past which the reading is not trusted
+        # for control is the point past which it must not be billed either.
+        self.max_hold_s = float(max_hold_s)
+        self._block: int | None = None       # epoch seconds of the block start
+        self._acc = 0.0                      # W*s of offtake in the open block
+        self._pp_acc = [0.0] * phases
+        self._elapsed = 0.0                  # seconds integrated in the open block
+        self._last_t: float | None = None    # epoch seconds of the held sample
+        self._last_net = 0.0
+        self._last_pp: tuple[float, ...] | None = None
+
+    # -- reading ------------------------------------------------------------
+    @property
+    def block_start(self) -> datetime | None:
+        if self._block is None:
+            return None
+        return datetime.fromtimestamp(self._block, timezone.utc)
+
+    @property
+    def elapsed_s(self) -> float:
+        """Seconds already integrated into the open block.
+
+        Exposed alongside the partial accumulator because SAFETY-07 projects the
+        end-of-quarter average and cannot do that from a finished average.
+        """
+        return self._elapsed
+
+    @property
+    def partial_ws(self) -> float:
+        """Offtake watt-seconds accumulated in the open block so far."""
+        return self._acc
+
+    @property
+    def offtake_avg_w(self) -> float:
+        """Average offtake over the part of the open block seen so far."""
+        return self._acc / self._elapsed if self._elapsed > 0 else 0.0
+
+    @property
+    def per_phase_offtake_avg_w(self) -> tuple[float, ...] | None:
+        if self._last_pp is None or self._elapsed <= 0:
+            return None
+        return tuple(a / self._elapsed for a in self._pp_acc)
+
+    # -- writing ------------------------------------------------------------
+    def add(self, sample: P1Sample) -> list[QuarterBlock]:
+        """Integrate up to this sample, then hold its value. Returns any blocks
+        that closed in the process (usually none, occasionally one)."""
+        t = sample.ingest_ts.timestamp()
+        closed: list[QuarterBlock] = []
+
+        if self._last_t is None:
+            self._block = int(t // QUARTER_S) * QUARTER_S
+            self._last_t, self._last_net = t, sample.net_w
+            self._last_pp = sample.per_phase_w
+            return closed
+        if t <= self._last_t:
+            # Out-of-order or duplicate arrival: integrating a negative dt would
+            # subtract energy that really happened. Drop it, keep the held value.
+            return closed
+
+        cursor = self._last_t
+        # Beyond this instant the held value stops being evidence of anything.
+        # The stretch from here to `t` is walked so the block boundaries are
+        # still crossed correctly, but nothing is accumulated and `_elapsed`
+        # does not grow - which is what makes a closed block, always divided by
+        # the full 900 s, actually get dragged down by the missing coverage.
+        hold_end = self._last_t + self.max_hold_s
+        while True:
+            end = self._block + QUARTER_S
+            stop = min(t, end)
+            covered = max(0.0, min(stop, hold_end) - cursor)
+            if covered > 0:
+                self._acc += max(self._last_net, 0.0) * covered
+                if self._last_pp is not None:
+                    for i, v in enumerate(self._last_pp[: self.phases]):
+                        self._pp_acc[i] += max(v, 0.0) * covered
+                self._elapsed += covered
+            cursor = stop
+            if stop < end:
+                break
+            closed.append(QuarterBlock(
+                start=datetime.fromtimestamp(self._block, timezone.utc),
+                # A closed block is always divided by the full 900 s, never by
+                # the seconds we happened to observe - a gap in coverage must
+                # drag the billed average down, not be averaged away.
+                offtake_avg_w=self._acc / QUARTER_S,
+                per_phase_offtake_avg_w=(
+                    tuple(a / QUARTER_S for a in self._pp_acc)
+                    if self._last_pp is not None else None),
+            ))
+            self._block = end
+            self._acc = 0.0
+            self._pp_acc = [0.0] * self.phases
+            self._elapsed = 0.0
+
+        self._last_t, self._last_net = t, sample.net_w
+        self._last_pp = sample.per_phase_w
+        return closed
+
+
+# --------------------------------------------------------------------------- #
+# what the rest of the add-on talks to
+# --------------------------------------------------------------------------- #
+class P1Ingest:
+    """Holds the latest sample and the rolling quarter-hour average.
+
+    ⚠️ Staleness is DERIVED from the stored sample, not carried as a separate
+    flag. That is what makes "flag before the value is visible" free: there is
+    one immutable object and a single attribute rebind to publish it, so a
+    reader can never see a fresh value with a stale flag or the reverse.
+    """
+
+    def __init__(self, phases: int = 1, max_age_s: float = 30.0):
+        self.phases = phases
+        self.max_age_s = float(max_age_s)
+        # The same threshold governs control and billing: a reading too old to
+        # steer by is too old to bill by. See QuarterAverager.max_hold_s.
+        self.averager = QuarterAverager(phases, max_hold_s=self.max_age_s)
+        self.blocks: list[QuarterBlock] = []
+        self.samples = 0
+        self.parse_errors = 0
+        self.last_error: str | None = None
+        self.started_mono = time.monotonic()
+        self._last: P1Sample | None = None
+
+    @property
+    def last(self) -> P1Sample | None:
+        return self._last
+
+    def submit(self, sample: P1Sample) -> None:
+        self._last = sample
+        self.samples += 1
+        for block in self.averager.add(sample):
+            self.blocks.append(block)
+            del self.blocks[:-96]   # a day of quarters; STATE-01 owns real retention
+
+    def reject(self, err: Exception | str) -> None:
+        """A malformed telegram or an unavailable entity.
+
+        ⚠️ The last good sample and ITS timestamp are left untouched. The reading
+        does not become 0 W and it does not become fresh - the age keeps growing,
+        which is precisely the signal a rejected telegram should produce.
+        """
+        self.parse_errors += 1
+        self.last_error = str(err)
+        _LOG.warning("P1 telegram rejected: %s", err)
+
+    # -- what consumers read -------------------------------------------------
+    def age_s(self) -> float | None:
+        """Age of the newest accepted sample, or None if there has never been one."""
+        return None if self._last is None else self._last.age_s()
+
+    @property
+    def published_age_s(self) -> float:
+        """The figure behind `sensor.p1_sample_age_s`.
+
+        Seconds since the newest accepted telegram, or since this ingester
+        started when none has ever arrived.
+
+        ⚠️ Always a number and never `unknown`, because SAFETY-01's firmware
+        watchdog subscribes to it: an entity that simply stops existing is
+        indistinguishable, from the firmware's side, from a meter that is fine.
+        And ⚠️ it is recomputed against the clock on every publish rather than
+        stamped once per telegram, so a meter that freezes at a constant reading
+        still produces a visibly climbing age. That is the whole point of this
+        entity - HA pushes state changes, so a genuinely constant P1 value emits
+        nothing at all, and a watchdog watching the value would sit there
+        believing the last update was recent.
+        """
+        age = self.age_s()
+        return max(0.0, time.monotonic() - self.started_mono) if age is None else age
+
+    @property
+    def stale(self) -> bool:
+        """True when there is no sample, or the newest one is past max_age_s."""
+        age = self.age_s()
+        return age is None or age > self.max_age_s
+
+    @property
+    def net_w(self) -> float | None:
+        """Signed net grid power, or None when stale. Never a substituted zero."""
+        return None if self.stale else self._last.net_w
+
+    @property
+    def per_phase_import_w(self) -> tuple[float, ...] | None:
+        if self.stale or self._last is None:
+            return None
+        return self._last.per_phase_import_w
+
+
+# --------------------------------------------------------------------------- #
+# transport 1: Home Assistant WebSocket (the DSMR integration's entities)
+# --------------------------------------------------------------------------- #
+WS_URL = "ws://supervisor/core/websocket"
+BAD_STATES = ("unknown", "unavailable", "none", "")
+
+
+class HaDsmrSource:
+    """Subscribes to state_changed for the configured DSMR entities.
+
+    ⚠️ WebSocket, not REST polling. REST returns states, but polling at the
+    30 s planning tick decimates a 5 s telegram stream and the quarter-hour
+    average would then be computed from a sixth of the data (§5.3). "Consume
+    every telegram" means event-driven.
+
+    ⚠️ One telegram updates several entities, and HA emits one state_changed per
+    entity. Building a sample on each event would mix a new import reading with
+    a stale export one for a few milliseconds every 5 s. A short debounce
+    coalesces the burst back into the single telegram it came from.
+    """
+
+    DEBOUNCE_S = 0.35
+
+    def __init__(self, session: aiohttp.ClientSession, ingest: P1Ingest,
+                 entities: dict, token: str | None = None):
+        self.session = session
+        self.ingest = ingest
+        self.entities = entities          # {"import": id, "export": id, "phase_import": [...], ...}
+        self.token = token or os.environ.get("SUPERVISOR_TOKEN", "")
+        self.ids = self._wanted()
+        self.cache: dict[str, float] = {}
+        self.connected = False
+        self._pending: asyncio.Task | None = None
+
+    def _wanted(self) -> set[str]:
+        out = set()
+        for key in ("import", "export"):
+            if self.entities.get(key):
+                out.add(self.entities[key])
+        for key in ("phase_import", "phase_export"):
+            out.update(e for e in self.entities.get(key) or [] if e)
+        return out
+
+    async def run(self) -> None:
+        """Long-lived task: connect, subscribe, reconnect with backoff, forever.
+
+        ⚠️ A reconnect emits nothing. A gap must stay a gap - a synthetic sample
+        on reconnect would reset the age and hide the outage from the very
+        watchdog that exists to catch it.
+        """
+        backoff = 1.0
+        while True:
+            try:
+                await self._session_once()
+                backoff = 1.0
+            except asyncio.CancelledError:
+                raise
+            except Exception as err:  # noqa: BLE001 - any transport fault retries
+                _LOG.warning("P1 HA websocket: %s - reconnecting in %.0fs", err, backoff)
+            finally:
+                self.connected = False
+            await asyncio.sleep(backoff)
+            backoff = min(backoff * 2, 30.0)
+
+    async def _session_once(self) -> None:
+        async with self.session.ws_connect(WS_URL, heartbeat=30) as ws:
+            hello = await ws.receive_json()
+            if hello.get("type") == "auth_required":
+                await ws.send_json({"type": "auth", "access_token": self.token})
+                reply = await ws.receive_json()
+                if reply.get("type") != "auth_ok":
+                    raise RuntimeError(f"auth rejected: {reply.get('message', reply)}")
+            await ws.send_json({"id": 1, "type": "subscribe_events",
+                                "event_type": "state_changed"})
+            await ws.send_json({"id": 2, "type": "get_states"})
+            self.connected = True
+            _LOG.info("P1 ingest: subscribed to %s", ", ".join(sorted(self.ids)))
+
+            async for msg in ws:
+                if msg.type is not aiohttp.WSMsgType.TEXT:
+                    continue
+                payload = json.loads(msg.data)
+                if payload.get("id") == 2 and payload.get("type") == "result":
+                    # ⚠️ Prime the cache, but do NOT build a sample from it.
+                    # get_states returns whatever HA currently holds, which
+                    # after a Core restart is a RestoreEntity value of unknown
+                    # age. Stamping that with ingest_ts=now resets the age to
+                    # zero and reports a fresh meter that may have been dead for
+                    # an hour - a synthetic sample hiding the outage from the
+                    # watchdog that exists to catch it. The cache is what lets
+                    # the FIRST real state_changed build a complete sample; the
+                    # age stays honest until one arrives.
+                    for obj in payload.get("result") or []:
+                        self._absorb(obj.get("entity_id"), obj.get("state"))
+                elif payload.get("type") == "event":
+                    data = (payload.get("event") or {}).get("data") or {}
+                    if data.get("entity_id") not in self.ids:
+                        continue
+                    new = data.get("new_state") or {}
+                    self._absorb(data.get("entity_id"), new.get("state"))
+                    self._schedule()
+            raise RuntimeError("websocket closed")
+
+    def _absorb(self, entity_id: str | None, state) -> None:
+        if not entity_id or entity_id not in self.ids:
+            return
+        raw = str(state).strip().lower()
+        if raw in BAD_STATES:
+            # ⚠️ An `unavailable` DSMR entity is a missing reading, not 0 W.
+            # Forget the cached value so no sample can be built from a mixture
+            # of a live register and one that stopped reporting.
+            self.cache.pop(entity_id, None)
+            self.ingest.reject(f"{entity_id} is {raw}")
+            return
+        try:
+            self.cache[entity_id] = float(raw)
+        except ValueError:
+            self.cache.pop(entity_id, None)
+            self.ingest.reject(f"{entity_id} is not numeric: {raw!r}")
+
+    def _schedule(self) -> None:
+        if self._pending and not self._pending.done():
+            return
+        self._pending = asyncio.get_running_loop().create_task(self._after_debounce())
+
+    async def _after_debounce(self) -> None:
+        await asyncio.sleep(self.DEBOUNCE_S)
+        self.build()
+
+    def build(self) -> bool:
+        """Assemble one sample from the cache. Returns True if one was accepted."""
+        imp_id, exp_id = self.entities.get("import"), self.entities.get("export")
+        if imp_id not in self.cache or exp_id not in self.cache:
+            return False
+        pi = [self.cache.get(e) for e in self.entities.get("phase_import") or []]
+        pe = [self.cache.get(e) for e in self.entities.get("phase_export") or []]
+        if pi and (None in pi or (pe and None in pe)):
+            return False           # incomplete phase set: wait, do not guess
+        try:
+            self.ingest.submit(make_sample(
+                SOURCE_HA, self.cache[imp_id], self.cache[exp_id],
+                phases=self.ingest.phases,
+                phase_import_w=pi or None,
+                phase_export_w=pe or None,
+                # ⚠️ No telegram_ts: HA's last_changed is when the STATE changed,
+                # which for a constant reading is minutes ago even though the
+                # telegram is current. Using it as a telegram time would fake
+                # staleness on a genuinely steady meter.
+            ))
+            return True
+        except P1Error as err:
+            self.ingest.reject(err)
+            return False
+
+
+# --------------------------------------------------------------------------- #
+# transport 2: MQTT
+# --------------------------------------------------------------------------- #
+def parse_mqtt_payload(raw: bytes | str, phases: int, *,
+                       now: datetime | None = None) -> P1Sample:
+    """One JSON telegram from the configured topic. Raises P1Error.
+
+    The accepted document, documented in DOCS.md:
+
+        {"import_w": 1234.0, "export_w": 0.0,
+         "phases": [{"import_w": 500, "export_w": 0}, ...],   # optional
+         "timestamp": "2026-08-24T18:00:05+02:00"}            # optional
+
+    ponytail: one strict schema rather than sniffing the half-dozen P1-bridge
+    dialects in the wild. The upgrade path is a `meter_mqtt_format` option
+    selecting a parser; a lenient parser is the wrong default at a safety
+    boundary, where guessing a key means guessing a kilowatt.
+    """
+    try:
+        doc = json.loads(raw)
+    except (ValueError, TypeError) as err:
+        raise P1Error(f"payload is not JSON: {err}") from None
+    if not isinstance(doc, dict):
+        raise P1Error(f"payload is not a JSON object ({type(doc).__name__})")
+
+    ts = None
+    if doc.get("timestamp"):
+        try:
+            ts = datetime.fromisoformat(str(doc["timestamp"]))
+        except ValueError:
+            raise P1Error(f"unparseable timestamp {doc['timestamp']!r}") from None
+        if ts.tzinfo is None:
+            raise P1Error("timestamp has no UTC offset")
+
+    pi = pe = None
+    if "phases" in doc:
+        rows = doc["phases"]
+        if not isinstance(rows, list) or not all(isinstance(r, dict) for r in rows):
+            raise P1Error("'phases' must be a list of objects")
+        pi = [r.get("import_w") for r in rows]
+        pe = [r.get("export_w", 0.0) for r in rows]
+
+    return make_sample(SOURCE_MQTT, doc.get("import_w"), doc.get("export_w"),
+                       phases=phases, phase_import_w=pi, phase_export_w=pe,
+                       telegram_ts=ts, ingest_ts=now)
+
+
+class MqttP1Source:
+    """Subscribes to one topic and submits every message that parses.
+
+    Uses paho's own reconnect loop on its own thread, then hops back onto the
+    event loop with call_soon_threadsafe so the ingest state is only ever
+    mutated from one thread.
+    """
+
+    def __init__(self, ingest: P1Ingest, topic: str, host, port=1883,
+                 username=None, password=None):
+        self.ingest = ingest
+        self.topic = topic
+        self.host, self.port = host, int(port or 1883)
+        self.username, self.password = username, password
+        self.client = None
+        self.loop = None
+
+    async def run(self) -> None:
+        if mqtt is None or not self.host or not self.topic:
+            _LOG.error("P1 MQTT source not usable (broker=%s topic=%r) - no meter data",
+                       self.host, self.topic)
+            return
+        self.loop = asyncio.get_running_loop()
+        try:
+            self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
+                                      client_id="goodwe_p1_ingest")
+        except AttributeError:      # paho 1.x, which is what Alpine ships
+            self.client = mqtt.Client(client_id="goodwe_p1_ingest")
+        if self.username:
+            self.client.username_pw_set(self.username, self.password or "")
+        self.client.on_connect = lambda *_a, **_k: self.client.subscribe(self.topic, qos=0)
+        self.client.on_message = self._on_message
+        self.client.reconnect_delay_set(min_delay=1, max_delay=30)
+        self.client.connect_async(self.host, self.port, keepalive=60)
+        self.client.loop_start()
+        _LOG.info("P1 ingest: MQTT %s:%s topic %s", self.host, self.port, self.topic)
+        try:
+            while True:
+                await asyncio.sleep(3600)
+        finally:
+            self.client.loop_stop()
+            self.client.disconnect()
+
+    def _on_message(self, _client, _userdata, msg) -> None:
+        # Runs on paho's network thread.
+        if self.loop is None:
+            return
+        self.loop.call_soon_threadsafe(self._handle, msg.payload)
+
+    def _handle(self, payload) -> None:
+        try:
+            self.ingest.submit(parse_mqtt_payload(payload, self.ingest.phases))
+        except P1Error as err:
+            self.ingest.reject(err)
+
+
+# --------------------------------------------------------------------------- #
+# transport 3: Home Assistant WebSocket, one signed entity
+# --------------------------------------------------------------------------- #
+class HaSignedSource(HaDsmrSource):
+    """The same websocket, subscribed to ONE signed power entity.
+
+    For readers that publish net power as a single signed figure - a HomeWizard
+    P1's `sensor.p1_meter_active_power`, positive = import - rather than the two
+    unsigned DSMR registers. This is the meter actually installed at the house,
+    and `ha_dsmr` cannot read it: it needs two registers and refuses a negative
+    one outright, which is every exporting telegram.
+
+    ⚠️ A subclass, not a copy. The connect / auth / subscribe / reconnect /
+    `_absorb` machinery above is transport, not shape, and it has already been
+    debugged once - notably "prime the cache from get_states but never build a
+    sample out of it" and "a reconnect emits nothing". Only `_wanted` (which
+    entity ids) and `build` (how they become a sample) differ, so only those two
+    are overridden. Everything TEL-01 established therefore applies unchanged:
+    ingest timestamping, meter_max_age_s, the clock-recomputed age sensor, and
+    `unavailable` treated as a missing reading rather than 0 W.
+
+    ⚠️ The debounce is inherited but does nothing useful here, and that is fine:
+    one telegram is one entity, so there is no burst of per-entity events to
+    coalesce and no window in which a new reading sits beside a stale one. It
+    costs one scheduled sleep per telegram at ~0.2 Hz. Left in place rather than
+    special-cased, because a second code path through build() is a second place
+    for the sign to go wrong.
+    """
+
+    def _wanted(self) -> set[str]:
+        out = set()
+        if self.entities.get("net"):
+            out.add(self.entities["net"])
+        out.update(e for e in self.entities.get("phase_net") or [] if e)
+        return out
+
+    def build(self) -> bool:
+        """Assemble one sample from the cache. Returns True if one was accepted."""
+        net_id = self.entities.get("net")
+        if net_id not in self.cache:
+            return False
+        pn = [self.cache.get(e) for e in self.entities.get("phase_net") or []]
+        if pn and None in pn:
+            return False           # incomplete phase set: wait, do not guess
+        try:
+            imp, exp = split_signed(self.cache[net_id])
+            pi = pe = None
+            if pn:
+                # ponytail: this split is arithmetically redundant today -
+                # make_sample subtracts the two lists again and does not
+                # sign-check per-phase figures, so handing it the signed values
+                # with a zero export list produces the identical tuple. Verified:
+                # mutating it that way leaves all 174 checks green, i.e. no test
+                # can tell the difference, and it is recorded here rather than
+                # left as a silent equivalent mutant for the next reviewer to
+                # rediscover. Kept because `phase_import_w` means a MAGNITUDE:
+                # a negative in it is the double-signing that make_sample refuses
+                # outright for the connection-level registers, and the day that
+                # check is extended per-phase the shortcut breaks the meter, not
+                # the test.
+                pairs = [split_signed(v) for v in pn]
+                pi = [a for a, _ in pairs]
+                pe = [b for _, b in pairs]
+            self.ingest.submit(make_sample(
+                SOURCE_HA_SIGNED, imp, exp,
+                phases=self.ingest.phases,
+                phase_import_w=pi, phase_export_w=pe,
+                # ⚠️ No telegram_ts, for the same reason as ha_dsmr: HA's
+                # last_changed is when the VALUE changed, which on a steady meter
+                # is minutes ago while the telegram is current. sensor.
+                # p1_sample_age_s is what covers a genuinely frozen meter.
+            ))
+            return True
+        except P1Error as err:
+            self.ingest.reject(err)
+            return False
+
+
+# --------------------------------------------------------------------------- #
+# selection
+# --------------------------------------------------------------------------- #
+def is_enabled(opts: dict) -> bool:
+    """Whether P1 ingestion is switched on at all.
+
+    ⚠️ One definition, because three places depend on it and they MUST agree:
+    where the grid reading comes from, whether the ingest task is started, and
+    whether sensor.p1_sample_age_s is announced over MQTT discovery. An age
+    sensor announced with no ingester behind it is a watchdog input nobody is
+    feeding, and the ESP32 trips on it.
+    """
+    return str(opts.get("meter_source", "off") or "off").strip() not in ("off", "")
+
+
+def build_source(opts: dict, ingest: P1Ingest, session, broker: dict | None):
+    """Return the transport named by `meter_source`, or None if disabled.
+
+    This is the whole of AC 1's "switchable": every consumer reads P1Ingest, so
+    changing transport is a config edit, never a code path.
+    """
+    source = str(opts.get("meter_source", "off") or "off").strip()
+    if not is_enabled(opts):
+        return None
+    if source == SOURCE_HA:
+        return HaDsmrSource(session, ingest, {
+            "import": opts.get("p1_import_entity", ""),
+            "export": opts.get("p1_export_entity", ""),
+            "phase_import": opts.get("p1_phase_import_entities") or [],
+            "phase_export": opts.get("p1_phase_export_entities") or [],
+        })
+    if source == SOURCE_HA_SIGNED:
+        return HaSignedSource(session, ingest, {
+            "net": opts.get("p1_net_entity", ""),
+            "phase_net": opts.get("p1_phase_net_entities") or [],
+        })
+    if source == SOURCE_MQTT:
+        broker = broker or {}
+        return MqttP1Source(ingest, str(opts.get("meter_mqtt_topic", "")),
+                            broker.get("host"), broker.get("port", 1883),
+                            broker.get("username"), broker.get("password"))
+    if source:
+        _LOG.error("meter_source %r is not one of %s - P1 ingestion disabled",
+                   source, ", ".join((SOURCE_HA, SOURCE_MQTT, SOURCE_HA_SIGNED)))
+    return None

goodwe_controller/config.yaml

@@ -1,5 +1,5 @@
 name: GoodWe RS485 Controller
-version: "0.2.1"
+version: "0.3.0"
 slug: goodwe_controller
 description: >-
   Drives a GoodWe ES/BP battery inverter over RS485 by emulating its smart
@@ -39,6 +39,34 @@ options:
   batt_invert: false
   setpoint_entity: ""
 
+  # --- P1 meter ingestion (specs §5.2 / §14 `meter:`) -------------------------
+  # `off` keeps the original single meter_entity path above, so an existing
+  # install is untouched until it opts in. ha_dsmr subscribes to the DSMR
+  # integration's entities over the HA WebSocket; mqtt_p1 reads the topic below;
+  # ha_signed reads ONE signed HA entity (+ import / - export), which is what a
+  # HomeWizard P1 publishes and what ha_dsmr cannot consume.
+  meter_source: "off"
+  meter_phases: 1
+  meter_max_age_s: 30
+  meter_mqtt_topic: ""
+  # The two UNSIGNED Belgian registers. The EMS derives net power from them
+  # (import - export); do NOT point these at a signed template sensor.
+  p1_import_entity: ""
+  p1_export_entity: ""
+  # Optional, in L1..L3 order. Required for the capacity-tariff peak on a
+  # three-phase connection; the list length must equal meter_phases.
+  p1_phase_import_entities: []
+  p1_phase_export_entities: []
+  # ha_signed only. ONE signed net-power sensor: positive = import from the
+  # grid, negative = export to it. Do NOT split it into two template sensors -
+  # the split is done in the add-on (p1.split_signed) precisely so the sign
+  # convention is tested rather than living in unreviewed YAML.
+  p1_net_entity: ""
+  # Optional, in L1..L3 order, each one signed the same way. Same role as
+  # p1_phase_import_entities: the capacity-tariff peak on a three-phase
+  # connection. The list length must equal meter_phases.
+  p1_phase_net_entities: []
+
   # --- control ---------------------------------------------------------------
   max_w: 2000
   gain: 0.6
@@ -48,6 +76,7 @@ options:
   step_w: 10
   saturation_w: 500
   saturation_cycles: 3
+  integrator_max_w: 0
   heartbeat_s: 10
   stale_input_s: 15
   auto_start: false
@@ -80,6 +109,24 @@ schema:
   batt_invert: bool
   setpoint_entity: str
 
+  meter_source: list(off|ha_dsmr|mqtt_p1|ha_signed)
+  # ⚠️ 2 is accepted by this range but is not a real Belgian connection. A
+  # telegram whose phase count disagrees is rejected at ingest and logged, so a
+  # mis-set 2 shows up immediately as "0 telegrams accepted" rather than as a
+  # quietly wrong number.
+  meter_phases: int(1,3)
+  meter_max_age_s: int(5,300)
+  meter_mqtt_topic: str?
+  p1_import_entity: str?
+  p1_export_entity: str?
+  p1_phase_import_entities:
+    - str
+  p1_phase_export_entities:
+    - str
+  p1_net_entity: str?
+  p1_phase_net_entities:
+    - str
+
   max_w: int(100,5000)
   gain: float(0.05,1.0)
   slew_w: int(50,5000)
@@ -88,6 +135,7 @@ schema:
   step_w: int(1,100)
   saturation_w: int(100,2000)
   saturation_cycles: int(1,10)
+  integrator_max_w: int(0,15000)
   heartbeat_s: int(2,25)
   stale_input_s: int(5,120)
   auto_start: bool

goodwe_controller/test_control.py

@@ -24,6 +24,63 @@ def check(name, cond):
         fails.append(name)
 
 
+# ---------------------------------------------------------------------------
+# COVERAGE AUDIT - measured, not executed. Read this before adding a mechanism.
+#
+# THE INVARIANT: every mechanism in compute() must be noticed by AT LEAST TWO
+# checks when it is deleted. If you add a mechanism to compute(), re-run the
+# audit and add it to the table. If a figure here drops, a check has started
+# passing for a reason other than the one it names.
+#
+# THE TECHNIQUE, because there is no script to run: replace one mechanism in
+# control.py with a no-op, run this file, count the failures, restore. That is
+# the converse of the usual mutation - not "does a wrong value fail?" but "does
+# anyone notice when the mechanism is GONE?". It is kept as a comment rather
+# than as tooling on purpose: the only cheap way to automate it is to key on
+# source lines, which goes stale silently, and a green audit that has quietly
+# stopped testing anything is precisely the failure this ticket exists to fix.
+# A comment cannot go stale-green, because it never claims to be running.
+#
+# Measured at 389d9ec. Numbers are the lead's independent reproduction.
+#
+#   mechanism in compute()                     checks that fail when deleted
+#   ------------------------------------------ -----------------------------
+#   integrator freeze (AC 3)                                               2
+#   integrator clamp (AC 1)                                                6
+#   integrator bound follows max_w                                         4
+#   output clamp                                                           3
+#   slew limit                                                             4
+#   output freeze                                                          2
+#   deadband                                                               5
+#   quantisation                                                           2
+#   saturation detector, `saturated_now = False`                          11
+#   saturation duration (AC 2), fires instantly                            2
+#   sat counter reset on a good cycle                                      6
+#   target_grid_w bias                                                     3
+#   i_w=None seeding from prev_w                                           6
+#
+# The detector figure is for the `saturated_now = False` form specifically;
+# disabling it further down as `frozen = False` is a weaker mutation and gives
+# 10. Reproduce the same form or the number will not match.
+#
+# ⚠️ IT HAS FOUND A DEAD MECHANISM TWICE, BOTH THE SAME WAY: a clamp standing in
+# for the mechanism under test. Deleting the integrator freeze once failed
+# NOTHING, because the fixtures sat at max_w 2000 and the integrator bound
+# truncated a wound value back to exactly 2000 - the assertion passed on the
+# clamp. The output clamp was masked the same way by the integrator bound.
+# Hence: A FIXTURE MUST SIT CLEAR OF EVERY RAIL IT IS NOT TESTING. Where a test
+# names one mechanism, make that mechanism the binding one (see TCLAMP and TF).
+#
+# ⚠️ RUN MUTATIONS WITH `python -B` AND CLEAR app/__pycache__. CPython
+# invalidates a .pyc on (source mtime in whole seconds, source size), so a
+# same-second rewrite that also preserves the file size reuses stale bytecode
+# and the suite reports on code you are no longer running. It under-reported one
+# mutation here as 2 where the true figure is 6. The error is one-directional -
+# stale bytecode can only under-report - so every figure above is a lower bound
+# at worst, and the two zeros ever recorded were both confirmed by fixing them
+# and watching the count rise, which a caching artefact cannot do.
+# ---------------------------------------------------------------------------
+
 print("control law")
 
 # Deadband: inside meter noise, hold exactly - do not drift.
@@ -41,13 +98,22 @@ check("proportional step (gain 0.6)", d.target_w == 300)
 d = compute(prev_w=0, grid_w=-500, actual_w=0, tuning=T)
 check("export drives charging", d.target_w == -300)
 
-# Clamp
-d = compute(prev_w=1900, grid_w=1000, actual_w=1900, tuning=Tuning(max_w=2000, slew_w=5000))
+# Clamp.
+# ⚠️ integrator_max_w is lifted clear of max_w so that the OUTPUT clamp is the
+# mechanism under test. Left at the default the integrator bound truncates
+# first, these two assertions pass on that alone, and deleting the output clamp
+# fails nothing - the same masking that hid the integrator freeze.
+TCLAMP = Tuning(max_w=2000, slew_w=5000, integrator_max_w=5000)
+d = compute(prev_w=1900, grid_w=1000, actual_w=1900, tuning=TCLAMP)
 check("clamped to max_w", d.target_w == 2000)
 
 # Slew: from 0 with a huge error, no more than slew_w in one cycle.
 d = compute(prev_w=0, grid_w=5000, actual_w=0, tuning=Tuning(max_w=5000, slew_w=1000))
 check("slew limits one cycle", d.target_w == 1000)
+d = compute(prev_w=-1900, grid_w=-1000, actual_w=-1900, tuning=TCLAMP)
+check("clamped to -max_w", d.target_w == -2000)
+d = compute(prev_w=0, grid_w=-5000, actual_w=0, tuning=Tuning(max_w=5000, slew_w=1000))
+check("slew limits one cycle, charging", d.target_w == -1000)
 
 # Saturation needs DURATION: one diverging cycle must NOT freeze.
 t = Tuning(saturation_w=500, saturation_cycles=3)
@@ -73,6 +139,251 @@ check("freeze still allows magnitude to fall", d.target_w < 2000)
 d = compute(prev_w=0, grid_w=7, actual_w=0, tuning=Tuning(deadband_w=1, step_w=10))
 check("quantised to step_w", d.target_w % 10 == 0)
 
+print("SAFETY-04: the integrator is bounded apart from the output")
+
+# The historical runaway, with its real numbers. A commercial controller on
+# this site, with the inverter switched OFF, wound ~130 W every 4 s past 10 kW
+# and reported 14 768 W while its output clamp sat at 5 kW. At gain 0.6 that
+# rate is a standing error of 130/0.6 = 217 W that never resolves, because the
+# inverter is not there to resolve it. 150 cycles is past the ~113 it took to
+# reach 14 768 W at that rate.
+RUNAWAY_ERROR = 130.0 / 0.6
+RUNAWAY_CYCLES = 150
+HISTORICAL_W = 14768.0
+
+
+def runaway(tuning, sign=1):
+    """Inverter off: it reports 0 W forever, the error never clears."""
+    prev, i_w, sat = 0.0, 0.0, 0
+    worst_i, worst_cmd = 0.0, 0.0
+    for _ in range(RUNAWAY_CYCLES):
+        d = compute(prev_w=prev, grid_w=sign * RUNAWAY_ERROR, actual_w=0.0,
+                    tuning=tuning, sat_count=sat, i_w=i_w)
+        prev, i_w, sat = d.target_w, d.i_w, d.sat_count
+        worst_i = max(worst_i, abs(i_w))
+        worst_cmd = max(worst_cmd, abs(prev))
+    return worst_i, worst_cmd
+
+
+TR = Tuning(max_w=2000)          # integrator_max_w unset => follows max_w
+wi, wc = runaway(TR)
+check(f"runaway: integrator plateaus at {wi:.0f} W (<= 2000)", wi <= TR.max_w)
+check(f"runaway: emitted command peaks at {wc:.0f} W (<= 2000)", wc <= TR.max_w)
+check("runaway: nowhere near the historical 14 768 W", wc < HISTORICAL_W / 4)
+
+# ...and with the saturation detector deliberately defeated, so that only the
+# clamp is holding. Kill one mechanism, the other still bounds it.
+TD = Tuning(max_w=2000, saturation_w=1e9)
+wi, wc = runaway(TD)
+check(f"runaway with the detector defeated: integrator still bounded ({wi:.0f} W)",
+      wi <= TD.max_w)
+check("runaway with the detector defeated: command still <= max_w", wc <= TD.max_w)
+
+# The mirror: the same runaway driving the other way. An export that never
+# clears winds the integrator negative just as hard.
+wi, wc = runaway(Tuning(max_w=2000), sign=-1)
+check(f"runaway (export direction): integrator bounded at {wi:.0f} W", wi <= 2000)
+check("runaway (export direction): emitted command <= max_w", wc <= 2000)
+
+# The bound is a separate quantity, and the useful direction is BELOW max_w:
+# there it binds first and caps unwind latency tighter than the rail does.
+d = compute(prev_w=0, grid_w=6000, actual_w=0,
+            tuning=Tuning(max_w=2000, integrator_max_w=1000, slew_w=5000))
+check("integrator bound binds independently of the output clamp",
+      d.i_w == 1000 and d.target_w == 1000)
+
+# Freeze = may not wind further in the direction it is already pushing.
+# ⚠️ max_w is raised WELL above the fixtures on purpose. At the default 2000
+# the integrator bound truncates a wound value back to exactly 2000 and
+# satisfies these assertions on its own, so deleting the freeze outright
+# failed nothing - the clamp was standing in for the mechanism under test.
+# Any fixture here must sit clear of every rail, or it tests the rail.
+TF = Tuning(saturation_w=500, saturation_cycles=3, max_w=5000)
+f1 = compute(prev_w=2000, grid_w=800, actual_w=0, tuning=TF, sat_count=3, i_w=2000.0)
+check("frozen: integration does not wind further", f1.i_w == 2000.0 and f1.frozen)
+f2 = compute(prev_w=2000, grid_w=-800, actual_w=0, tuning=TF, sat_count=3, i_w=2000.0)
+check("frozen: unwinding is still allowed", f2.i_w < 2000.0)
+# ...and the same two on the charging side. Every freeze rule in this file has
+# a mirror, because the one that did not is the defect that got through review.
+f3 = compute(prev_w=-2000, grid_w=-800, actual_w=0, tuning=TF, sat_count=3, i_w=-2000.0)
+check("frozen (charging): integration does not wind further", f3.i_w == -2000.0)
+f4 = compute(prev_w=-2000, grid_w=800, actual_w=0, tuning=TF, sat_count=3, i_w=-2000.0)
+check("frozen (charging): unwinding is still allowed", f4.i_w > -2000.0)
+
+# ⚠️ REGRESSION, and the reason the first cut of SAFETY-04 was rejected. A
+# freeze encoded as "only corrections that shrink |i_w|" is unsatisfiable for
+# BOTH signs of error whenever |correction| > 2*|i_w|, so near zero the loop
+# stops moving forever - the freeze cannot clear, because clearing it needs the
+# inverter to track and not-tracking is what saturation means. Measured on that
+# encoding: 0 W held into a 2 kW import for as long as the sim ran.
+z = compute(prev_w=0, grid_w=2000, actual_w=600, tuning=T, sat_count=3, i_w=0.0)
+check("frozen at i_w=0: a 2 kW import still moves the command",
+      z.frozen and z.target_w == 1000)
+# ...and the next cycle the inverter is inside saturation_w of the command, so
+# the freeze clears on its own. Deadlock would show up here as frozen=True.
+z2 = compute(prev_w=1000, grid_w=1000, actual_w=600, tuning=T,
+             sat_count=z.sat_count, i_w=z.i_w)
+check("frozen at i_w=0: the freeze then clears", not z2.frozen)
+# Same stranding on the other side: a small positive integrator against export.
+z3 = compute(prev_w=100, grid_w=-1000, actual_w=800, tuning=T, sat_count=3, i_w=100.0)
+check("frozen at i_w=+100: a 1 kW export still moves the command",
+      z3.frozen and z3.target_w < 0)
+z4 = compute(prev_w=-100, grid_w=1000, actual_w=-800, tuning=T, sat_count=3, i_w=-100.0)
+check("frozen at i_w=-100: a 1 kW import still moves the command",
+      z4.frozen and z4.target_w > 0)
+
+# ⚠️ EXACTLY ZERO, BOTH DIRECTIONS. This boundary has a history: the first cut
+# deadlocked here under import, and the fix for it deadlocked here under export
+# because `if i_w > 0 ... else ...` files 0.0 under rising-only. main.py resets
+# i_w to exactly 0.0 on every stop and every reseed, so it is a normal state,
+# not a corner.
+zi = compute(prev_w=0, grid_w=2000, actual_w=600, tuning=T, sat_count=3, i_w=0.0)
+check("frozen at i_w=0.0: an import push moves the integrator",
+      zi.frozen and zi.i_w > 0)
+ze = compute(prev_w=0, grid_w=-2000, actual_w=-600, tuning=T, sat_count=3, i_w=0.0)
+check("frozen at i_w=0.0: an export push moves the integrator",
+      ze.frozen and ze.i_w < 0)
+# The COMMAND still holds at 0 W in that second case, and that is release/1.0's
+# rule, not a leftover: at prev_w == 0 the output freeze forbids starting to
+# charge while saturated, because commanding 0 while the inverter reports
+# hundreds of watts means something else is driving the bus. Asserted so that
+# nobody "fixes" it by accident - the integrator moving is what this ticket
+# owns, the command rule belongs to the output freeze.
+check("frozen at i_w=0.0: the output freeze still blocks a charge from 0 W",
+      ze.target_w == 0.0)
+# Where prev_w is already charging the output freeze does NOT block, and there
+# the difference reaches the wire: held at 0.0 the integrator abandons the
+# charge mid-export.
+zc = compute(prev_w=-2000, grid_w=-4000, actual_w=-600, tuning=T, sat_count=3, i_w=0.0)
+check("frozen at i_w=0.0: a charge is not abandoned during heavy export",
+      zc.target_w == -2000.0)
+
+# The general property, rather than another handful of points: while frozen the
+# integrator may be held ONLY when the correction would push it further from
+# zero on the side it already sits. Any other hold is a deadlock.
+stuck = []
+for i0 in [x * 25.0 for x in range(-80, 81)]:
+    for g in [x * 100.0 for x in range(-40, 41)]:
+        err = g - T.target_grid_w
+        if abs(err) < T.deadband_w:
+            continue
+        dd = compute(prev_w=0.0, grid_w=g, actual_w=1500.0, tuning=T, sat_count=3, i_w=i0)
+        if dd.i_w == i0 and not ((i0 > 0 and err > 0) or (i0 < 0 and err < 0)):
+            stuck.append((i0, g))
+check(f"frozen integrator never deadlocks, over {161*81} states"
+      + (f" (e.g. {stuck[0]})" if stuck else ""), not stuck)
+
+# False-positive guard: a normal 2 kW load step must not trip the detector,
+# because the plant needs several cycles to catch up on every one of them.
+prev, actual, sat, i_w, froze = 0.0, 0.0, 0, 0.0, False
+for _ in range(12):
+    d = compute(prev, 2000.0 - actual, actual, T, sat, i_w)
+    prev, sat, i_w = d.target_w, d.sat_count, d.i_w
+    actual = actual + 0.94 * (prev - actual)
+    froze = froze or d.frozen
+check("a normal 2 kW load step does not trip the saturation freeze", not froze)
+
+# The convergence sim below runs WITHOUT a carried integrator. This is the same
+# 2 kW step in the configuration that actually ships, where main.py carries it.
+prev, actual, sat, i_w = 0.0, 0.0, 0, 0.0
+carried = 0
+for _ in range(12):
+    d = compute(prev, 2000.0 - actual, actual, T, sat, i_w)
+    prev, sat, i_w = d.target_w, d.sat_count, d.i_w
+    actual = actual + 0.94 * (prev - actual)
+    carried += 1
+    if abs(2000.0 - actual) < T.deadband_w:
+        break
+check(f"carried integrator converges in {carried} cycles (<=6)", carried <= 6)
+check("carried integrator does not overshoot the load", actual <= 2000.0 + T.deadband_w)
+
+# ⚠️ REGRESSION: an integrator allowed to wind past the rail buys nothing (the
+# output clamp already bounds the wire) and costs extra cycles of
+# wrong-direction power after every saturation event. 4000 W load held to
+# saturation, then dropped to 0; the figure is the command on the first cycle
+# after the drop. This is what makes the DOCS advice checkable.
+def unwind(t):
+    prev, actual, sat, i_w, load = 0.0, 0.0, 0, 0.0, 4000.0
+    for c in range(16):
+        if c == 15:
+            load = 0.0
+        d = compute(prev, load - actual, actual, t, sat, i_w)
+        prev, sat, i_w = d.target_w, d.sat_count, d.i_w
+        actual = actual + 0.94 * (prev - actual)
+    return prev
+
+
+tight, loose = unwind(Tuning(max_w=2000)), unwind(Tuning(max_w=2000, integrator_max_w=3000))
+check(f"after saturation ends the command is {tight:.0f} W (<= 1000)", tight <= 1000)
+check(f"headroom above max_w makes that worse ({loose:.0f} W) - hence the default",
+      loose > tight)
+
+print("SAFETY-04: the i_w=None path is still release/1.0, exactly")
+
+
+def legacy(prev, grid, actual, t, sat_count):
+    """release/1.0's control law, transcribed. Do not 'improve' this."""
+    reason = "tracking"
+    sc = min(sat_count + 1, 10) if abs(prev - actual) > t.saturation_w else 0
+    frozen = sc >= t.saturation_cycles
+    error = grid - t.target_grid_w
+    if abs(error) < t.deadband_w:
+        want, reason = prev, "deadband"
+    else:
+        want = prev + t.gain * error
+    target = max(-t.max_w, min(t.max_w, want))
+    if target != want:
+        reason = "clamped"
+    slewed = max(prev - t.slew_w, min(prev + t.slew_w, target))
+    if slewed != target:
+        reason = "slew-limited"
+    target = slewed
+    if frozen:
+        target = min(target, prev) if prev > 0 else max(target, prev)
+        reason = "saturated-freeze"
+    step = max(1, int(t.step_w))
+    return float(round(target / step) * step), sc, frozen, reason
+
+
+# ⚠️ Compare EVERYTHING observable, not just the number. A previous version of
+# this sweep compared (target_w, sat_count) only and passed 3024 cases while
+# `reason` had silently lost a value - which is the kind of thing a sweep this
+# broad exists to catch. `frozen` and `reason` are both in the tuple now.
+#
+# The one deliberate rename: what release/1.0 called "clamped" is now
+# "i-clamped", because the truncation happens on the integrator before the
+# command is derived from it. Aliased here rather than papered over - if any
+# OTHER reason ever diverges, this check goes red.
+ALIAS = {"i-clamped": "clamped"}
+diffs = []
+seen = set()
+for tune in (Tuning(), Tuning(target_grid_w=-10.0), Tuning(max_w=5000, slew_w=5000)):
+    for prev in (-2000.0, -500.0, -100.0, 0.0, 100.0, 500.0, 2000.0):
+        for grid in (-6000.0, -1000.0, -500.0, -14.0, 0.0, 14.0, 500.0, 1000.0, 6000.0):
+            for actual in (-2000.0, 0.0, 600.0, 2000.0):
+                for sc in (0, 2, 3, 9):
+                    d = compute(prev, grid, actual, tune, sc)   # i_w defaults to None
+                    seen.add(d.reason)
+                    got = (d.target_w, d.sat_count, d.frozen,
+                           ALIAS.get(d.reason, d.reason))
+                    if got != legacy(prev, grid, actual, tune, sc):
+                        diffs.append((prev, grid, actual, sc, got,
+                                      legacy(prev, grid, actual, tune, sc)))
+check(f"i_w=None reproduces release/1.0 over {3*7*9*4*4} cases, reason included"
+      + (f" (first diff {diffs[0]})" if diffs else ""), not diffs)
+
+# ...and the rename is not a quiet deletion: the signal SAFETY-03 alarms on has
+# to actually occur in that sweep, or its hook is dead.
+check("the integrator clamp reports itself as 'i-clamped'", "i-clamped" in seen)
+
+# "clamped" stays reachable, but only where the integrator is deliberately
+# allowed above the rail - then BOTH fire and the output clamp, which describes
+# the value actually emitted, is the one reported.
+dc = compute(prev_w=0, grid_w=6000, actual_w=0,
+             tuning=Tuning(max_w=2000, integrator_max_w=3000, slew_w=5000))
+check("the output clamp still reports 'clamped' when it is the binding one",
+      dc.reason == "clamped" and dc.i_w == 3000 and dc.target_w == 2000)
+
 print("capacity tariff")
 check("no forecast means no cap", maintenance_charge_floor(2500, None, 3500) == 2500)
 check("headroom caps the charge", maintenance_charge_floor(2500, 2000, 3500) == 1500)

goodwe_controller/test_p1.py

@@ -0,0 +1,936 @@
+"""Runnable check for P1 ingestion.  `python3 test_p1.py`
+
+No framework, no fixtures, no meter - it has to run on a tech's laptop and in CI
+with nothing installed and nothing plugged in (§17: "usable without real
+hardware").
+
+Every assert here is a rule whose absence poisons something downstream: an
+inverted sign inverts the control loop, a fabricated zero hides a dead meter, a
+naive per-phase sum misbills the capacity tariff, and a plain mean of samples
+computes the wrong quarter-hour figure whenever the telegram cadence changes.
+"""
+
+import asyncio
+import sys
+import time
+from datetime import datetime, timedelta, timezone
+
+import aiohttp   # already required by app.p1, so this adds no new dependency
+
+from app.p1 import (
+    P1Error, P1Ingest, HaDsmrSource, HaSignedSource, QuarterAverager,
+    SOURCE_HA, SOURCE_HA_SIGNED, SOURCE_MQTT,
+    build_source, is_enabled, make_sample, parse_mqtt_payload, split_signed,
+)
+
+fails = []
+total = 0
+
+TZ_BE_SUMMER = timezone(timedelta(hours=2))
+TZ_BE_WINTER = timezone(timedelta(hours=1))
+BASE = datetime(2026, 8, 24, 10, 0, 0, tzinfo=timezone.utc)   # a quarter boundary
+
+
+def check(name, cond):
+    global total
+    total += 1
+    if cond:
+        print(f"  ok   {name}")
+    else:
+        print(f"  FAIL {name}")
+        fails.append(name)
+
+
+def raises(name, fn):
+    global total
+    total += 1
+    try:
+        fn()
+    except P1Error:
+        print(f"  ok   {name}")
+        return
+    except Exception as err:  # noqa: BLE001
+        print(f"  FAIL {name} (raised {type(err).__name__}, wanted P1Error)")
+        fails.append(name)
+        return
+    print(f"  FAIL {name} (no error raised)")
+    fails.append(name)
+
+
+def built(src):
+    """`src.build()`, with any escaping exception turned into a visible value.
+
+    ⚠️ Legibility of a RED, not leniency. build() is contracted to return a bool
+    and to funnel every bad telegram through ingest.reject() - a guard that goes
+    missing (say the "is the net entity cached at all" one) makes it raise
+    instead. That still fails the suite, but by aborting it with a traceback at
+    whichever check happened to run first, which costs the next person ten
+    minutes deciding whether the suite is broken or the code is. Returning the
+    exception makes it compare unequal to True/False, so the NAMED check goes red
+    and says which rule died.
+    """
+    try:
+        return src.build()
+    except Exception as err:  # noqa: BLE001 - a raise here is itself the failure
+        print(f"       build() raised {type(err).__name__}: {err}")
+        return err
+
+
+def sample(net_import, net_export=0.0, at=BASE, phases=1, pi=None, pe=None):
+    return make_sample(SOURCE_HA, net_import, net_export, phases=phases,
+                       phase_import_w=pi, phase_export_w=pe,
+                       ingest_ts=at, ingest_mono=at.timestamp())
+
+
+# --------------------------------------------------------------------------- #
+print("import/export -> signed net (the derivation the EMS owns)")
+
+s = sample(1500.0, 0.0)
+check("pure import is positive", s.net_w == 1500.0)
+
+s = sample(0.0, 900.0)
+check("pure export is negative", s.net_w == -900.0)
+
+# The single test that catches an inverted control loop.
+s = sample(120.0, 2000.0)
+check("export-dominant telegram yields negative net", s.net_w == -1880.0)
+
+s = sample(2000.0, 120.0)
+check("import-dominant telegram yields positive net", s.net_w == 1880.0)
+
+# Both registers non-zero at once is real: a three-phase house can import on one
+# phase and export on another in the same telegram.
+s = sample(400.0, 400.0)
+check("both registers equal nets to exactly zero", s.net_w == 0.0)
+
+s = sample(0.0, 0.0)
+check("both registers zero is a valid balanced reading", s.net_w == 0.0
+      and s.import_w == 0.0 and s.export_w == 0.0)
+
+check("the magnitudes survive the derivation",
+      sample(300.0, 50.0).import_w == 300.0 and sample(300.0, 50.0).export_w == 50.0)
+
+# --------------------------------------------------------------------------- #
+print("rejecting a telegram instead of believing it")
+
+raises("negative 'unsigned' import is rejected", lambda: sample(-100.0, 0.0))
+raises("negative 'unsigned' export is rejected", lambda: sample(0.0, -100.0))
+raises("a non-numeric register is rejected", lambda: sample("n/a", 0.0))
+raises("None is rejected, not read as zero", lambda: sample(None, 0.0))
+raises("NaN is rejected", lambda: sample(float("nan"), 0.0))
+raises("infinity is rejected", lambda: sample(float("inf"), 0.0))
+# §20 open question 5: an HA sensor reporting 64954 for -582 W, i.e. an unsigned
+# 16-bit register decoded without its sign. Must not average in as 65 kW.
+raises("the 64954 signed-decode contamination is rejected",
+       lambda: sample(64954.0, 0.0))
+raises("a naive timestamp is rejected",
+       lambda: make_sample(SOURCE_MQTT, 100, 0, phases=1,
+                           telegram_ts=datetime(2026, 8, 24, 10, 0, 0)))
+
+# --------------------------------------------------------------------------- #
+print("single- and three-phase")
+
+s = sample(800.0, 0.0, phases=1, pi=[800.0], pe=[0.0])
+check("single phase accepts one phase", s.per_phase_w == (800.0,))
+check("per-phase tuple, not list", isinstance(s.per_phase_w, tuple))
+
+raises("three phases configured, one delivered -> rejected",
+       lambda: sample(800.0, 0.0, phases=3, pi=[800.0], pe=[0.0]))
+raises("one phase configured, three delivered -> rejected",
+       lambda: sample(800.0, 0.0, phases=1, pi=[300.0, 300.0, 200.0],
+                      pe=[0.0, 0.0, 0.0]))
+
+s = sample(600.0, 0.0)
+check("no phase data leaves per-phase None, not a fabricated tuple",
+      s.per_phase_w is None and s.per_phase_import_w is None)
+
+# §17's three-phase unbalanced-load regression case. L1 imports hard, L2 exports,
+# L3 idles: the connection nets to 600 W of offtake while 2100 W is drawn across
+# the phases. This is the case a naive per-phase sum gets wrong.
+s = sample(600.0, 0.0, phases=3, pi=[2000.0, 0.0, 100.0], pe=[0.0, 1500.0, 0.0])
+check("unbalanced: per-phase net keeps the export phase negative",
+      s.per_phase_w == (2000.0, -1500.0, 100.0))
+check("unbalanced: per-phase IMPORT clamps the exporting phase to zero",
+      s.per_phase_import_w == (2000.0, 0.0, 100.0))
+check("unbalanced: per-phase import sums to 2100 W, the connection nets 600 W",
+      sum(s.per_phase_import_w) == 2100.0 and s.net_w == 600.0)
+check("unbalanced: the naive sum is NOT the billed figure",
+      sum(s.per_phase_import_w) != s.net_w)
+
+# --------------------------------------------------------------------------- #
+print("rolling 15-minute average (time-weighted, clock-aligned, offtake only)")
+
+# Irregular spacing, hand-computed:
+#   0->3 s  held at 1000 W ->  3000 Ws
+#   3->13 s held at    0 W ->      0 Ws
+#  13->20 s held at 2000 W -> 14000 Ws
+#  total 17000 Ws over 20 s -> 850 W
+a = QuarterAverager(1)
+a.add(sample(1000.0, at=BASE))
+a.add(sample(0.0, at=BASE + timedelta(seconds=3)))
+a.add(sample(2000.0, at=BASE + timedelta(seconds=13)))
+a.add(sample(2000.0, at=BASE + timedelta(seconds=20)))
+check("irregular spacing integrates to the hand-computed 850 W",
+      abs(a.offtake_avg_w - 850.0) < 1e-9)
+check("elapsed-seconds-in-block is exposed for SAFETY-07", a.elapsed_s == 20.0)
+check("partial accumulator is exposed for SAFETY-07", a.partial_ws == 17000.0)
+# A plain mean over the four samples would be 1250 W. The whole point of the
+# time weighting is that these two numbers differ.
+check("a plain mean of those samples would have said 1250 W, not 850",
+      abs((1000 + 0 + 2000 + 2000) / 4 - 1250.0) < 1e-9 and a.offtake_avg_w != 1250.0)
+
+# Cadence change mid-block: 1 s telegrams for a minute, then 10 s telegrams.
+#   0->60 s   held at 1000 W ->  60000 Ws
+#  60->600 s  held at  100 W ->  54000 Ws
+#  114000 Ws over 600 s -> 190 W
+a = QuarterAverager(1)
+for i in range(0, 61):
+    a.add(sample(1000.0 if i < 60 else 100.0, at=BASE + timedelta(seconds=i)))
+for i in range(70, 601, 10):
+    a.add(sample(100.0, at=BASE + timedelta(seconds=i)))
+check("a cadence change does not bias the average (190 W)",
+      abs(a.offtake_avg_w - 190.0) < 1e-9)
+check("elapsed tracks the whole 600 s despite the cadence change", a.elapsed_s == 600.0)
+# The decimation trap: the fast minute contributes 60 of 114 samples but only
+# 10 % of the time, so a per-sample mean lands near 574 W - three times high.
+naive = (60 * 1000 + 54 * 100) / 114
+check("a per-sample mean would have said ~574 W", 570 < naive < 578)
+
+# Straddling the boundary: 890 s into a block, next telegram 20 s later. Ten
+# seconds belong to each block and must be split, not attributed to one.
+a = QuarterAverager(1)
+a.add(sample(1000.0, at=BASE + timedelta(seconds=890)))
+closed = a.add(sample(1000.0, at=BASE + timedelta(seconds=910)))
+check("crossing a boundary closes exactly one block", len(closed) == 1)
+check("the closed block keeps only its own 10 s (10000/900 W)",
+      abs(closed[0].offtake_avg_w - 10000.0 / 900.0) < 1e-9)
+check("the closed block divides by the full 900 s, so a gap drags it down",
+      closed[0].offtake_avg_w < 1000.0)
+check("the new block carries the other 10 s", a.elapsed_s == 10.0
+      and abs(a.offtake_avg_w - 1000.0) < 1e-9)
+check("the closed block starts on a quarter boundary",
+      closed[0].start == BASE and closed[0].start.minute % 15 == 0)
+
+# A whole block of pure export: the capacity tariff bills offtake, so this is
+# 0 kW, never a negative peak.
+a = QuarterAverager(1)
+a.add(sample(0.0, 3000.0, at=BASE))
+closed = a.add(sample(0.0, 3000.0, at=BASE + timedelta(seconds=900)))
+check("a block of pure export averages to 0 W of offtake",
+      len(closed) == 1 and closed[0].offtake_avg_w == 0.0)
+check("...while the signed sample itself stays negative",
+      sample(0.0, 3000.0).net_w == -3000.0)
+
+# Clock alignment holds either side of a DST change, because every Belgian UTC
+# offset is a whole number of hours and the block grid is 900 s of UTC.
+for label, tz in (("summer (+02:00)", TZ_BE_SUMMER), ("winter (+01:00)", TZ_BE_WINTER)):
+    a = QuarterAverager(1)
+    odd = datetime(2026, 8, 24, 13, 7, 23, tzinfo=tz)
+    a.add(sample(500.0, at=odd))
+    local = a.block_start.astimezone(tz)
+    check(f"block boundary is local :00/:15/:30/:45 in {label}",
+          local.minute in (0, 15, 30, 45) and local.second == 0 and local.microsecond == 0)
+
+# Three-phase averaging keeps the phases apart.
+a = QuarterAverager(3)
+a.add(sample(600.0, 0.0, at=BASE, phases=3, pi=[2000.0, 0.0, 100.0], pe=[0.0, 1500.0, 0.0]))
+a.add(sample(600.0, 0.0, at=BASE + timedelta(seconds=100), phases=3,
+             pi=[2000.0, 0.0, 100.0], pe=[0.0, 1500.0, 0.0]))
+check("per-phase offtake averages are held separately",
+      a.per_phase_offtake_avg_w == (2000.0, 0.0, 100.0))
+check("the block's own average is the connection net, not the phase sum",
+      abs(a.offtake_avg_w - 600.0) < 1e-9)
+
+# An out-of-order arrival must not subtract energy that really happened.
+a = QuarterAverager(1)
+a.add(sample(1000.0, at=BASE))
+a.add(sample(1000.0, at=BASE + timedelta(seconds=10)))
+before = a.partial_ws
+a.add(sample(1000.0, at=BASE + timedelta(seconds=5)))
+check("an out-of-order telegram is dropped, not integrated backwards",
+      a.partial_ws == before and a.elapsed_s == 10.0)
+# ⚠️ The assertion above is NOT sufficient on its own, and that is the whole
+# lesson: deleting the guard still passes it, because the negative interval is
+# separately refused by the `covered > 0` test. What the guard actually prevents
+# is the REWIND - without it the held timestamp moves back to +5 s and the next
+# telegram re-integrates the 5..10 s window that was already counted. The damage
+# only becomes visible one sample later, so the test has to go one sample later.
+a.add(sample(1000.0, at=BASE + timedelta(seconds=20)))
+check("...and the held timestamp is not rewound, so the next telegram "
+      "cannot double-count", a.elapsed_s == 20.0 and a.partial_ws == 20000.0)
+
+# A duplicate telegram (identical timestamp) is the same rule.
+a = QuarterAverager(1)
+a.add(sample(1000.0, at=BASE))
+a.add(sample(1000.0, at=BASE + timedelta(seconds=10)))
+a.add(sample(4000.0, at=BASE + timedelta(seconds=10)))
+a.add(sample(1000.0, at=BASE + timedelta(seconds=20)))
+check("a duplicate timestamp neither re-integrates nor replaces the held value",
+      a.elapsed_s == 20.0 and a.partial_ws == 20000.0)
+
+# A gap must not be filled with the last held value. The meter dies at 5 kW and
+# returns ten minutes later; hold-forward would credit 5 kW x 600 s to the
+# capacity-tariff accumulator - a fabricated peak, on a permanent record, from
+# data nobody measured.
+a = QuarterAverager(1, max_hold_s=30.0)
+a.add(sample(5000.0, at=BASE))
+a.add(sample(5000.0, at=BASE + timedelta(seconds=600)))
+check("a 600 s gap is held for at most max_hold_s, not for the whole gap",
+      a.partial_ws == 5000.0 * 30.0)
+check("the unobserved stretch does not count as elapsed time", a.elapsed_s == 30.0)
+closed = a.add(sample(5000.0, at=BASE + timedelta(seconds=900)))
+check("the outage drags the billed quarter down instead of inventing a peak",
+      len(closed) == 1 and abs(closed[0].offtake_avg_w - 300000.0 / 900.0) < 1e-9)
+check("...nowhere near the 5000 W a hold-forward would have billed",
+      closed[0].offtake_avg_w < 400.0)
+
+# The cap must not disturb a normally-spaced stream.
+a = QuarterAverager(1, max_hold_s=30.0)
+for i in range(0, 121, 5):          # a healthy 5 s telegram cadence
+    a.add(sample(2000.0, at=BASE + timedelta(seconds=i)))
+check("a healthy 5 s cadence is untouched by the hold cap",
+      a.elapsed_s == 120.0 and abs(a.offtake_avg_w - 2000.0) < 1e-9)
+
+# --------------------------------------------------------------------------- #
+print("ingest timestamp, age and staleness")
+
+now = time.monotonic()
+s = make_sample(SOURCE_HA, 1000, 0, phases=1, ingest_mono=now)
+check("a sample carries a tz-aware ingest timestamp",
+      s.ingest_ts.tzinfo is not None)
+check("age is ~0 immediately after ingest", s.age_s(now_mono=now) == 0.0)
+check("age grows with elapsed time", s.age_s(now_mono=now + 12.5) == 12.5)
+check("age never goes negative on a clock step",
+      s.age_s(now_mono=now - 100.0) == 0.0)
+
+ing = P1Ingest(phases=1, max_age_s=30.0)
+check("no sample yet is stale, not zero", ing.stale is True and ing.net_w is None)
+check("no sample yet has no age at all", ing.age_s() is None)
+
+ing.submit(make_sample(SOURCE_HA, 1234, 0, phases=1, ingest_mono=time.monotonic()))
+check("a fresh sample is not stale", ing.stale is False)
+check("a fresh sample exposes signed net power", ing.net_w == 1234.0)
+check("a fresh sample's age is small", 0 <= ing.age_s() < 1.0)
+
+ing.submit(make_sample(SOURCE_HA, 1234, 0, phases=1,
+                       ingest_mono=time.monotonic() - 29.0))
+check("29 s old with max_age_s 30 is still usable", ing.stale is False)
+ing.submit(make_sample(SOURCE_HA, 1234, 0, phases=1,
+                       ingest_mono=time.monotonic() - 31.0))
+check("31 s old with max_age_s 30 is stale", ing.stale is True)
+check("a stale sample reads as None, never as 0 W", ing.net_w is None)
+check("...and its per-phase import is None too", ing.per_phase_import_w is None)
+
+# The MQTT retained-message trap: replayed on reconnect with a fresh receive
+# time but a ten-minute-old telegram time. Fresh ingest must not launder it.
+old = datetime.now(timezone.utc) - timedelta(minutes=10)
+ing = P1Ingest(phases=1, max_age_s=30.0)
+ing.submit(make_sample(SOURCE_MQTT, 1000, 0, phases=1, telegram_ts=old,
+                       ingest_mono=time.monotonic()))
+check("a retained telegram is stale on arrival despite a fresh receive time",
+      ing.stale is True and ing.age_s() > 590)
+
+# A rejected telegram must not refresh anything and must not become 0 W.
+ing = P1Ingest(phases=1, max_age_s=30.0)
+ing.submit(make_sample(SOURCE_HA, 1500, 0, phases=1,
+                       ingest_mono=time.monotonic() - 10.0))
+stamp = ing.last.ingest_mono
+ing.reject("malformed telegram")
+check("a rejected telegram counts as a parse error", ing.parse_errors == 1)
+check("a rejected telegram leaves the last good value in place",
+      ing.last.net_w == 1500.0)
+check("a rejected telegram does not refresh the timestamp",
+      ing.last.ingest_mono == stamp)
+check("a rejected telegram does not resolve to 0 W", ing.net_w == 1500.0)
+
+# Meter goes stale but stays connected (§17 failure injection): no new sample
+# ever arrives, and the age keeps climbing past max_age_s on its own.
+ing = P1Ingest(phases=1, max_age_s=30.0)
+ing.submit(make_sample(SOURCE_HA, 800, 0, phases=1,
+                       ingest_mono=time.monotonic() - 120.0))
+check("connected-but-silent meter trips staleness with no new telegram",
+      ing.stale is True and ing.age_s() > 100)
+
+# sensor.p1_sample_age_s: SAFETY-01's firmware subscribes to this, so it must
+# always be a number and must climb while nothing arrives.
+ing = P1Ingest(phases=1, max_age_s=30.0)
+ing.started_mono = time.monotonic() - 45.0
+check("the published age is a number before the first telegram ever arrives",
+      isinstance(ing.published_age_s, float) and ing.published_age_s > 44)
+check("...and it is never None, unlike the raw age",
+      ing.age_s() is None and ing.published_age_s is not None)
+ing.submit(make_sample(SOURCE_HA, 500, 0, phases=1, ingest_mono=time.monotonic()))
+check("a telegram resets the published age", ing.published_age_s < 1.0)
+# The false-trip this entity exists to remove: the meter keeps sending, the
+# VALUE never changes, and the age must still reflect that it is being sent.
+for _ in range(3):
+    ing.submit(make_sample(SOURCE_HA, 500, 0, phases=1, ingest_mono=time.monotonic()))
+check("an unchanging meter value still reads as fresh while telegrams arrive",
+      ing.stale is False and ing.published_age_s < 1.0)
+ing.submit(make_sample(SOURCE_HA, 500, 0, phases=1,
+                       ingest_mono=time.monotonic() - 90.0))
+check("the same unchanging value reads as stale once the telegrams stop",
+      ing.stale is True and ing.published_age_s > 89)
+
+# --------------------------------------------------------------------------- #
+print("MQTT transport: parsing a telegram off the topic")
+
+good = '{"import_w": 1200.5, "export_w": 0}'
+s = parse_mqtt_payload(good, 1)
+check("a well-formed payload parses", s.net_w == 1200.5 and s.source == SOURCE_MQTT)
+
+s = parse_mqtt_payload('{"import_w": 0, "export_w": 2500}', 1)
+check("an export payload parses to a negative net", s.net_w == -2500.0)
+
+s = parse_mqtt_payload(
+    '{"import_w": 600, "export_w": 0, "phases":'
+    ' [{"import_w":2000,"export_w":0},{"import_w":0,"export_w":1500},'
+    '  {"import_w":100,"export_w":0}]}', 3)
+check("a three-phase payload parses per-phase",
+      s.per_phase_w == (2000.0, -1500.0, 100.0))
+
+s = parse_mqtt_payload(
+    '{"import_w": 100, "export_w": 0, "timestamp": "2026-08-24T12:00:00+02:00"}', 1)
+check("a telegram timestamp is kept when the payload has one",
+      s.telegram_ts == datetime(2026, 8, 24, 12, 0, 0, tzinfo=TZ_BE_SUMMER))
+
+raises("a non-JSON payload is rejected", lambda: parse_mqtt_payload("not json", 1))
+raises("a JSON array is rejected", lambda: parse_mqtt_payload("[1,2,3]", 1))
+raises("a payload missing import_w is rejected",
+       lambda: parse_mqtt_payload('{"export_w": 0}', 1))
+raises("a payload with a null register is rejected",
+       lambda: parse_mqtt_payload('{"import_w": null, "export_w": 0}', 1))
+raises("a payload with a string register is rejected",
+       lambda: parse_mqtt_payload('{"import_w": "1200", "export_w": 0}', 1))
+raises("a timestamp with no UTC offset is rejected",
+       lambda: parse_mqtt_payload(
+           '{"import_w":1,"export_w":0,"timestamp":"2026-08-24T12:00:00"}', 1))
+raises("a phase count that disagrees with meter_phases is rejected",
+       lambda: parse_mqtt_payload(
+           '{"import_w":1,"export_w":0,"phases":[{"import_w":1,"export_w":0}]}', 3))
+
+# --------------------------------------------------------------------------- #
+print("HA DSMR transport: building a sample out of entity states")
+
+ENT = {"import": "sensor.p1_import", "export": "sensor.p1_export",
+       "phase_import": [], "phase_export": []}
+ing = P1Ingest(phases=1, max_age_s=30.0)
+src = HaDsmrSource(None, ing, ENT, token="x")
+
+src._absorb("sensor.p1_import", "1000")
+check("one entity alone does not build a sample", src.build() is False
+      and ing.last is None)
+src._absorb("sensor.p1_export", "0")
+check("both entities present builds one sample", src.build() is True
+      and ing.net_w == 1000.0)
+
+# ⚠️ The reason the debounce exists: HA emits one state_changed per entity, so
+# mid-telegram the cache briefly holds a new import with the old export.
+src._absorb("sensor.p1_import", "0")
+src._absorb("sensor.p1_export", "2500")
+src.build()
+check("a coalesced telegram lands as one consistent sample", ing.net_w == -2500.0)
+
+before = ing.last
+src._absorb("sensor.p1_export", "unavailable")
+check("an unavailable entity is a parse error", ing.parse_errors == 1)
+check("an unavailable entity does not build a sample from a stale half",
+      src.build() is False)
+check("an unavailable entity leaves the last good sample untouched",
+      ing.last is before and ing.net_w == -2500.0)
+
+src._absorb("sensor.p1_export", "unknown")
+check("an unknown entity is treated the same way", ing.parse_errors == 2)
+src._absorb("sensor.p1_export", "banana")
+check("a non-numeric entity state is a parse error, not 0 W",
+      ing.parse_errors == 3 and ing.net_w == -2500.0)
+
+src._absorb("sensor.p1_export", "64954")
+src._absorb("sensor.p1_import", "0")
+check("the contaminated signed decode is refused at the transport too",
+      src.build() is False and ing.parse_errors == 4)
+
+ing3 = P1Ingest(phases=3, max_age_s=30.0)
+ENT3 = {"import": "sensor.p1_import", "export": "sensor.p1_export",
+        "phase_import": ["sensor.l1_i", "sensor.l2_i", "sensor.l3_i"],
+        "phase_export": ["sensor.l1_e", "sensor.l2_e", "sensor.l3_e"]}
+src3 = HaDsmrSource(None, ing3, ENT3, token="x")
+for eid, val in (("sensor.p1_import", "600"), ("sensor.p1_export", "0"),
+                 ("sensor.l1_i", "2000"), ("sensor.l2_i", "0")):
+    src3._absorb(eid, val)
+check("an incomplete phase set waits instead of guessing", src3.build() is False)
+for eid, val in (("sensor.l3_i", "100"), ("sensor.l1_e", "0"),
+                 ("sensor.l2_e", "1500"), ("sensor.l3_e", "0")):
+    src3._absorb(eid, val)
+check("a complete three-phase set builds", src3.build() is True)
+check("three-phase entities produce the unbalanced per-phase tuple",
+      ing3.last.per_phase_w == (2000.0, -1500.0, 100.0))
+check("the sample's per-phase tuple length matches meter_phases",
+      len(ing3.last.per_phase_w) == ing3.phases == 3)
+
+# --------------------------------------------------------------------------- #
+print("HA DSMR transport: end to end against a fake Home Assistant")
+# Everything above pokes at build()/_absorb() directly. This one drives the real
+# thing over a real websocket - auth handshake, subscribe_events, get_states,
+# per-entity events - because "the transport works" is otherwise an untested
+# claim about a protocol nobody re-reads.
+
+
+async def _e2e():
+    from aiohttp import web
+    import app.p1 as p1mod
+
+    done = asyncio.Event()
+
+    async def fake_ha(request):
+        ws = web.WebSocketResponse()
+        await ws.prepare(request)
+        await ws.send_json({"type": "auth_required", "ha_version": "2026.8"})
+        auth = await ws.receive_json()
+        assert auth["type"] == "auth" and auth["access_token"] == "tok"
+        await ws.send_json({"type": "auth_ok"})
+        sub = await ws.receive_json()
+        assert sub["type"] == "subscribe_events"
+        assert sub["event_type"] == "state_changed"
+        await ws.send_json({"id": sub["id"], "type": "result", "success": True})
+        get = await ws.receive_json()
+        assert get["type"] == "get_states"
+        await ws.send_json({"id": get["id"], "type": "result", "success": True, "result": [
+            {"entity_id": "sensor.p1_import", "state": "1000"},
+            {"entity_id": "sensor.p1_export", "state": "0"},
+            {"entity_id": "sensor.something_else", "state": "hello"},
+        ]})
+        # One telegram, two state_changed events - exactly how HA emits it.
+        for imp, exp in ((1500, 0), (0, 800)):
+            await asyncio.sleep(0.5)
+            for eid, val in (("sensor.p1_import", imp), ("sensor.p1_export", exp)):
+                await ws.send_json({"type": "event", "event": {"data": {
+                    "entity_id": eid,
+                    "new_state": {"entity_id": eid, "state": str(val)}}}})
+        await asyncio.sleep(0.5)
+        await ws.send_json({"type": "event", "event": {"data": {
+            "entity_id": "sensor.p1_export",
+            "new_state": {"entity_id": "sensor.p1_export", "state": "unavailable"}}}})
+        await asyncio.sleep(0.5)
+        done.set()
+        return ws
+
+    srv = web.Application()
+    srv.router.add_get("/ws", fake_ha)
+    runner = web.AppRunner(srv)
+    await runner.setup()
+    site = web.TCPSite(runner, "127.0.0.1", 0)
+    await site.start()
+    port = site._server.sockets[0].getsockname()[1]
+    p1mod.WS_URL = f"http://127.0.0.1:{port}/ws"
+
+    ing = P1Ingest(phases=1, max_age_s=30.0)
+    async with aiohttp.ClientSession() as sess:
+        src = HaDsmrSource(sess, ing, dict(ENT), token="tok")
+        task = asyncio.get_running_loop().create_task(src.run())
+        try:
+            await asyncio.wait_for(done.wait(), 20)
+            await asyncio.sleep(0.5)
+        finally:
+            task.cancel()
+            try:
+                await task
+            except asyncio.CancelledError:
+                pass
+    await runner.cleanup()
+    return ing, src
+
+
+live, wire = asyncio.run(_e2e())
+check("the websocket handshake and subscription complete", live.samples >= 1)
+# ⚠️ TWO, not three. get_states primes the cache but must NOT build a sample:
+# HA returns whatever it currently holds, which after a Core restart is a
+# RestoreEntity value of unknown age, and stamping that with ingest_ts=now
+# resets the age and reports a fresh meter that may have been dead for an hour.
+# Only the two real state_changed telegrams become samples. Four state_changed
+# events arrived (two per telegram); the debounce is what makes those two
+# consistent samples rather than four half-updated ones.
+check("connecting does not manufacture a sample from cached HA state",
+      live.samples == 2)
+check("the final export-dominant telegram nets negative",
+      live.last.net_w == -800.0)
+check("the sample was built over the wire, tagged with its transport",
+      live.last.source == SOURCE_HA)
+check("an entity we did not subscribe to is never cached",
+      "sensor.something_else" not in wire.cache and len(wire.cache) == 1)
+check("a mid-stream unavailable is a parse error, not a sample",
+      live.parse_errors == 1 and live.samples == 2)
+check("the last good reading survives the unavailable", live.net_w == -800.0)
+check("the averager integrated the live stream", live.averager.elapsed_s > 0.2)
+
+# The reason get_states still matters: it is what lets the FIRST real telegram
+# build a complete sample instead of waiting for every entity to change once.
+check("the primed cache let the first telegram build immediately",
+      live.samples == 2 and live.last.import_w == 0.0)
+
+# --------------------------------------------------------------------------- #
+print("ha_signed: one signed entity -> the same two magnitudes")
+# The meter actually fitted at this house is a HomeWizard P1 publishing ONE
+# signed sensor. ha_dsmr cannot read it - it wants two unsigned registers and
+# refuses a negative one, which is every exporting telegram.
+
+check("a positive reading is import", split_signed(1500.0) == (1500.0, 0.0))
+check("a negative reading is export", split_signed(-900.0) == (0.0, 900.0))
+check("zero is a balanced reading, not a missing one",
+      split_signed(0.0) == (0.0, 0.0))
+check("exactly one magnitude is ever non-zero",
+      all(a == 0.0 or b == 0.0 for a, b in
+          (split_signed(v) for v in (-5710.0, -1.0, 0.0, 1.0, 4384.0))))
+# ⚠️ The split must not change the number. A control loop steered by a value
+# that was rounded or clamped on the way in is steered by a different meter.
+check("the split round-trips the signed value exactly",
+      all(make_sample(SOURCE_HA_SIGNED, *split_signed(v), phases=1).net_w == v
+          for v in (-11763.0, -5710.0, -0.5, 0.0, 0.5, 775.0, 4384.0)))
+
+raises("a non-numeric signed reading is rejected", lambda: split_signed("n/a"))
+raises("a signed None is rejected, not read as zero", lambda: split_signed(None))
+raises("a signed NaN is rejected", lambda: split_signed(float("nan")))
+raises("a signed infinity is rejected", lambda: split_signed(float("inf")))
+# ⚠️ This one matters MORE on the signed path than on the unsigned one. On
+# ha_dsmr the §20 contamination is also caught by "unsigned cannot be negative";
+# here 64954 arrives as a perfectly well-formed positive signed reading and the
+# plausibility ceiling is the only thing standing in front of it.
+raises("the 64954 signed-decode contamination is still rejected",
+       lambda: split_signed(64954.0))
+raises("...and its negative twin too", lambda: split_signed(-64954.0))
+
+# --------------------------------------------------------------------------- #
+print("ha_signed: the sign convention, against real captured readings")
+# ⚠️ Not a datasheet claim. These are literal values out of
+# sim/scenarios/ha-p1_meter_active_power-2026-08-{20,23}.json, HA recorder
+# exports of sensor.p1_meter_active_power at this house, copied here rather than
+# read from that repo so this file still runs on a laptop with nothing installed
+# (§17). If the convention were inverted, the physics below would be absurd.
+
+# 2026-08-23T11:46:52Z - the day's most negative reading, 13:46 local, full sun.
+s = make_sample(SOURCE_HA_SIGNED, *split_signed(-5710.0), phases=1)
+check("the midday PV peak (-5710 W) is EXPORT, not a 5.7 kW draw",
+      s.net_w == -5710.0 and s.export_w == 5710.0 and s.import_w == 0.0)
+# 2026-08-19T22:00:00Z - midnight local, 20 Aug's file starts here. No sun.
+s = make_sample(SOURCE_HA_SIGNED, *split_signed(775.0), phases=1)
+check("the overnight base load (+775 W) is IMPORT",
+      s.net_w == 775.0 and s.import_w == 775.0 and s.export_w == 0.0)
+# 2026-08-20T12:20:58Z - 14:20 local, the largest export in either capture.
+s = make_sample(SOURCE_HA_SIGNED, *split_signed(-11763.0), phases=1)
+check("the -11763 W midday extreme is export and survives the ceiling",
+      s.net_w == -11763.0 and s.export_w == 11763.0)
+# 2026-08-23T10:18:28Z - the largest import in the healthy capture.
+s = make_sample(SOURCE_HA_SIGNED, *split_signed(4384.0), phases=1)
+check("the +4384 W peak is import", s.net_w == 4384.0 and s.import_w == 4384.0)
+# The whole convention in one line: night draws, midday feeds back.
+check("night is positive and midday is negative, which is the convention",
+      split_signed(775.0)[0] > 0 and split_signed(-5710.0)[1] > 0)
+
+# --------------------------------------------------------------------------- #
+print("ha_signed transport: building a sample out of one entity state")
+
+NET = {"net": "sensor.p1_meter_active_power", "phase_net": []}
+ing = P1Ingest(phases=1, max_age_s=30.0)
+sig = HaSignedSource(None, ing, NET, token="x")
+
+check("nothing cached yet builds nothing", built(sig) is False and ing.last is None)
+sig._absorb("sensor.p1_meter_active_power", "1000")
+check("one signed entity is a complete telegram on its own",
+      built(sig) is True and ing.net_w == 1000.0)
+check("the sample is tagged with its own transport",
+      ing.last.source == SOURCE_HA_SIGNED)
+sig._absorb("sensor.p1_meter_active_power", "-2500")
+built(sig)
+check("a negative state lands as a negative net", ing.net_w == -2500.0)
+
+before = ing.last
+sig._absorb("sensor.p1_meter_active_power", "unavailable")
+check("an unavailable signed entity is a parse error", ing.parse_errors == 1)
+check("an unavailable entity does not build a sample", built(sig) is False)
+# ⚠️ The rule the whole ticket turns on: a missing reading is MISSING. Resolving
+# it to 0 W would read as a perfectly balanced house and defeat the staleness
+# trigger that FW-01's watchdog is built on.
+check("an unavailable entity leaves the last good sample untouched, not 0 W",
+      ing.last is before and ing.net_w == -2500.0)
+sig._absorb("sensor.p1_meter_active_power", "unknown")
+check("an unknown signed entity is treated the same way", ing.parse_errors == 2)
+sig._absorb("sensor.p1_meter_active_power", "banana")
+check("a non-numeric signed state is a parse error, not 0 W",
+      ing.parse_errors == 3 and ing.net_w == -2500.0)
+sig._absorb("sensor.p1_meter_active_power", "64954")
+check("64954 is refused at the signed transport too",
+      built(sig) is False and ing.parse_errors == 4)
+sig._absorb("sensor.not_ours", "123")
+check("an unsubscribed entity is never cached by the signed transport",
+      "sensor.not_ours" not in sig.cache)
+
+# A rejected reading must not make the age look fresh - the age is what the
+# firmware watchdog reads, and a rejection is exactly when it must keep climbing.
+ing = P1Ingest(phases=1, max_age_s=30.0)
+sig = HaSignedSource(None, ing, dict(NET), token="x")
+ing.submit(make_sample(SOURCE_HA_SIGNED, 1200, 0, phases=1,
+                       ingest_mono=time.monotonic() - 20.0))
+sig._absorb("sensor.p1_meter_active_power", "unavailable")
+built(sig)
+check("a rejected reading does not reset the published age",
+      ing.published_age_s > 19 and ing.net_w == 1200.0)
+ing.submit(make_sample(SOURCE_HA_SIGNED, 1200, 0, phases=1,
+                       ingest_mono=time.monotonic() - 40.0))
+check("...and the age keeps climbing past max_age_s on its own",
+      ing.stale is True and ing.net_w is None)
+
+# The three-phase reading the TEL-04 survey recorded at this house: L1 +2301 W,
+# L2 +468 W, L3 -2582 W, netting +187 W. A signed per-phase set splits the same
+# way, and the exporting phase must still clamp out of the billed figure.
+ing3 = P1Ingest(phases=3, max_age_s=30.0)
+NET3 = {"net": "sensor.p1_meter_active_power",
+        "phase_net": ["sensor.p1_l1", "sensor.p1_l2", "sensor.p1_l3"]}
+sig3 = HaSignedSource(None, ing3, NET3, token="x")
+for eid, val in (("sensor.p1_meter_active_power", "187"), ("sensor.p1_l1", "2301")):
+    sig3._absorb(eid, val)
+check("an incomplete signed phase set waits instead of guessing", built(sig3) is False)
+sig3._absorb("sensor.p1_l2", "468")
+sig3._absorb("sensor.p1_l3", "-2582")
+check("a complete signed three-phase set builds", built(sig3) is True)
+check("signed per-phase entities keep the exporting phase negative",
+      ing3.last.per_phase_w == (2301.0, 468.0, -2582.0))
+check("per-phase IMPORT clamps the exporting phase to zero",
+      ing3.last.per_phase_import_w == (2301.0, 468.0, 0.0))
+check("the phase import sum is 2769 W while the connection nets 187 W",
+      sum(ing3.last.per_phase_import_w) == 2769.0 and ing3.last.net_w == 187.0)
+check("the signed per-phase tuple length matches meter_phases",
+      len(ing3.last.per_phase_w) == ing3.phases == 3)
+
+sig_bad = HaSignedSource(None, P1Ingest(phases=3, max_age_s=30.0),
+                         {"net": "sensor.net", "phase_net": ["sensor.a", "sensor.b"]},
+                         token="x")
+for eid in ("sensor.net", "sensor.a", "sensor.b"):
+    sig_bad._absorb(eid, "100")
+check("two phases delivered against meter_phases 3 is rejected, not padded",
+      built(sig_bad) is False and sig_bad.ingest.last is None
+      and sig_bad.ingest.parse_errors == 1)
+
+# --------------------------------------------------------------------------- #
+print("ha_signed transport: end to end against a fake Home Assistant")
+# The transport is a subclass, so this is what proves the INHERITED machinery -
+# auth, subscribe, the get_states priming rule, the reconnect-emits-nothing
+# rule - still behaves when only _wanted() and build() were replaced.
+
+
+async def _e2e_signed():
+    from aiohttp import web
+    import app.p1 as p1mod
+
+    done = asyncio.Event()
+    eid = "sensor.p1_meter_active_power"
+
+    async def fake_ha(request):
+        ws = web.WebSocketResponse()
+        await ws.prepare(request)
+        await ws.send_json({"type": "auth_required", "ha_version": "2026.8"})
+        auth = await ws.receive_json()
+        assert auth["type"] == "auth" and auth["access_token"] == "tok"
+        await ws.send_json({"type": "auth_ok"})
+        sub = await ws.receive_json()
+        assert sub["type"] == "subscribe_events"
+        await ws.send_json({"id": sub["id"], "type": "result", "success": True})
+        get = await ws.receive_json()
+        assert get["type"] == "get_states"
+        await ws.send_json({"id": get["id"], "type": "result", "success": True, "result": [
+            {"entity_id": eid, "state": "775.0"},
+            {"entity_id": "sensor.something_else", "state": "hello"},
+        ]})
+        # Two real telegrams, both literal captured values: overnight import,
+        # then the midday export peak.
+        for val in ("775.0", "-5710.0"):
+            await asyncio.sleep(0.5)
+            await ws.send_json({"type": "event", "event": {"data": {
+                "entity_id": eid,
+                "new_state": {"entity_id": eid, "state": val}}}})
+        await asyncio.sleep(0.5)
+        await ws.send_json({"type": "event", "event": {"data": {
+            "entity_id": eid,
+            "new_state": {"entity_id": eid, "state": "unavailable"}}}})
+        await asyncio.sleep(0.5)
+        done.set()
+        return ws
+
+    srv = web.Application()
+    srv.router.add_get("/ws", fake_ha)
+    runner = web.AppRunner(srv)
+    await runner.setup()
+    site = web.TCPSite(runner, "127.0.0.1", 0)
+    await site.start()
+    port = site._server.sockets[0].getsockname()[1]
+    p1mod.WS_URL = f"http://127.0.0.1:{port}/ws"
+
+    ing = P1Ingest(phases=1, max_age_s=30.0)
+    async with aiohttp.ClientSession() as sess:
+        src = HaSignedSource(sess, ing, dict(NET), token="tok")
+        task = asyncio.get_running_loop().create_task(src.run())
+        try:
+            await asyncio.wait_for(done.wait(), 20)
+            await asyncio.sleep(0.5)
+        finally:
+            task.cancel()
+            try:
+                await task
+            except asyncio.CancelledError:
+                pass
+    await runner.cleanup()
+    return ing, src
+
+
+live, wire = asyncio.run(_e2e_signed())
+# ⚠️ TWO, not three - the same rule as the ha_dsmr e2e. get_states primes the
+# cache but must never become a sample: after a Core restart it is a
+# RestoreEntity value of unknown age, and stamping it with ingest_ts=now reports
+# a fresh meter that may have been dead for an hour.
+check("ha_signed does not manufacture a sample from cached HA state",
+      live.samples == 2)
+check("the signed telegrams arrived over a real websocket",
+      live.last.source == SOURCE_HA_SIGNED)
+check("the final export telegram nets negative, over the wire",
+      live.last.net_w == -5710.0 and live.last.export_w == 5710.0)
+check("ha_signed subscribes to the one entity and caches nothing else",
+      wire.ids == {"sensor.p1_meter_active_power"}
+      and "sensor.something_else" not in wire.cache)
+check("a mid-stream unavailable signed state is a parse error, not a sample",
+      live.parse_errors == 1 and live.samples == 2)
+# ⚠️ And the cached half is DROPPED, so no later telegram can be assembled out
+# of a value that stopped reporting.
+check("an unavailable entity is evicted from the cache", wire.cache == {})
+check("the last good reading survives the unavailable, and is not 0 W",
+      live.net_w == -5710.0)
+check("the averager integrated the live signed stream", live.averager.elapsed_s > 0.2)
+# ⚠️ The entity FW-01 waits on. It must be a number here exactly as it is on the
+# other transports - the house P1 went 51.1 s and 36.2 s between state changes
+# overnight, and without this the watchdog false-trips the battery to 0 W.
+check("sensor.p1_sample_age_s is a live number on this transport too",
+      isinstance(live.published_age_s, float) and live.published_age_s >= 0.0)
+
+# --------------------------------------------------------------------------- #
+print("ha_signed: selection by config")
+
+check("ha_signed is enabled", is_enabled({"meter_source": SOURCE_HA_SIGNED}) is True)
+built = build_source({"meter_source": SOURCE_HA_SIGNED,
+                      "p1_net_entity": "sensor.p1_meter_active_power"},
+                     P1Ingest(), None, None)
+check("meter_source ha_signed selects the signed transport",
+      isinstance(built, HaSignedSource))
+check("...wired to p1_net_entity, and subscribed to exactly that one entity",
+      built.ids == {"sensor.p1_meter_active_power"})
+# ⚠️ The three modes must not bleed into each other: ha_dsmr must keep ignoring
+# p1_net_entity, or a half-configured install silently reads the wrong sensor.
+plain = build_source({"meter_source": SOURCE_HA,
+                      "p1_import_entity": "sensor.i", "p1_export_entity": "sensor.e",
+                      "p1_net_entity": "sensor.p1_meter_active_power"},
+                     P1Ingest(), None, None)
+check("ha_dsmr still selects the unsigned transport and ignores p1_net_entity",
+      type(plain) is HaDsmrSource and plain.ids == {"sensor.i", "sensor.e"})
+check("meter_source off still selects nothing",
+      build_source({"meter_source": "off"}, P1Ingest(), None, None) is None)
+check("an unrecognised meter_source selects nothing rather than guessing",
+      build_source({"meter_source": "ha_signd"}, P1Ingest(), None, None) is None)
+
+# --------------------------------------------------------------------------- #
+print("the age sensor must not exist when P1 is off")
+# ⚠️ This is a fleet-wide regression guard, not a nicety. The ESP32 watchdog
+# does `id(p1_age_s).has_state() && id(p1_age_s).state >= max_age_s` and forces
+# the layer-1 failsafe. published_age_s counts from P1Ingest.__init__, so if the
+# age were published with meter_source off it would climb past 30 s on every
+# existing install within half a minute and pin the inverter at 0 W forever.
+
+from app.mqtt import SENSORS, MqttPublisher   # noqa: E402
+
+check("meter_source off is disabled", is_enabled({"meter_source": "off"}) is False)
+check("a missing meter_source is disabled", is_enabled({}) is False)
+check("an empty meter_source is disabled", is_enabled({"meter_source": ""}) is False)
+check("ha_dsmr is enabled", is_enabled({"meter_source": SOURCE_HA}) is True)
+check("mqtt_p1 is enabled", is_enabled({"meter_source": SOURCE_MQTT}) is True)
+
+# The entity id SAFETY-01's firmware subscribes to, pinned by object_id.
+row = [s for s in SENSORS if s[0] == "p1_age"]
+check("the age sensor is declared exactly once", len(row) == 1)
+check("its object_id pins entity_id to sensor.p1_sample_age_s",
+      row[0][1] == "p1_sample_age_s")
+check("it is published in seconds", row[0][3] == "s")
+
+
+class _RecordingClient:
+    def __init__(self):
+        self.sent = []
+
+    def publish(self, topic, payload=None, retain=False):
+        # Topic AND payload: object_id, the thing that actually pins the entity
+        # id, only appears in the discovery payload. Recording topics alone made
+        # the "is not announced" check pass for the wrong reason.
+        self.sent.append(f"{topic} {payload}")
+
+
+def _announced(omit):
+    pub = MqttPublisher(None, 1883, omit=omit)   # host None -> never connects
+    pub.client = _RecordingClient()
+    pub._announce()
+    return " ".join(pub.client.sent)
+
+
+check("with P1 off the age sensor is never announced",
+      "p1_sample_age_s" not in _announced(("p1_age",)))
+check("the other status entities are still announced with P1 off",
+      "goodwe_grid_power" in _announced(("p1_age",)))
+check("with P1 on the age sensor IS announced",
+      "p1_sample_age_s" in _announced(()))
+
+# And the publish dict itself, through the real Controller.
+from app.main import Controller            # noqa: E402
+
+
+class _Store:
+    data = {}
+
+    def set(self, *a):
+        pass
+
+    def get_time(self, *a):
+        return None
+
+
+class _Pub:
+    def __init__(self):
+        self.last = {}
+
+    def publish(self, values):
+        self.last = values
+
+    def close(self):
+        pass
+
+
+pub_off = _Pub()
+Controller({"meter_source": "off"}, None, _Store(), pub_off).publish()
+check("with P1 off, p1_age is absent from the published payload",
+      "p1_age" not in pub_off.last)
+check("...while the normal status keys are still published",
+      "setpoint" in pub_off.last and "grid" in pub_off.last)
+
+pub_on = _Pub()
+Controller({"meter_source": SOURCE_HA}, None, _Store(), pub_on).publish()
+check("with P1 on, p1_age is published", "p1_age" in pub_on.last)
+check("...as a number, so has_state() becomes true only once we feed it",
+      isinstance(pub_on.last["p1_age"], float))
+
+# ⚠️ And on ha_signed identically - this is the whole reason TEL-04 exists. The
+# age sensor is a hard prerequisite for the FW-01 flash, and it has to appear on
+# the transport that can actually read the meter in this house.
+pub_sig = _Pub()
+Controller({"meter_source": SOURCE_HA_SIGNED}, None, _Store(), pub_sig).publish()
+check("with ha_signed selected, p1_age is published too",
+      "p1_age" in pub_sig.last and isinstance(pub_sig.last["p1_age"], float))
+
+print()
+if fails:
+    print(f"{len(fails)} of {total} FAILED: {', '.join(fails)}")
+    sys.exit(1)
+print(f"{total} checks")
+print("all checks passed")