TEL-01 review fixes: stop the age sensor tripping installs that have no meter

T-1, and it was a fleet-wide trip to zero. publish() emitted p1_age
unconditionally, and published_age_s counts from P1Ingest.__init__ when no
sample has ever arrived. With meter_source defaulting to off, every existing
install would have published sensor.p1_sample_age_s climbing without bound; the
ESP32 does `has_state() && state >= max_age_s` and forces the layer-1 failsafe,
so each of them would have pinned its inverter at 0 W within 30 s. Exactly the
opposite of the zero-regression the off default was for. The key is now omitted
from the payload AND from MQTT discovery when P1 is off, so the entity does not
exist at all - which is the status quo, and what has_state() is testing for.
The predicate is one function, is_enabled(), because the grid reading, the task
start and the discovery announcement have to agree or this comes back.

T-2, connect no longer manufactures a sample. get_states returns whatever HA
currently holds, which after a Core restart is a RestoreEntity value of unknown
age; stamping it with ingest_ts=now reset the age and reported a fresh meter
that could have been dead for an hour. run()'s own docstring already said a
reconnect must emit nothing - the code disagreed with it, and a test asserted
the violation. The cache is still primed, so the first real state_changed
builds a complete sample; the age just stays honest until one arrives.

T-3, gaps are no longer filled with the last held value. The averager held a
sample forward across any interval, so a meter dying at 5 kW and returning ten
minutes later credited 5 kW x 600 s to the capacity-tariff accumulator - a
fabricated peak on a permanent record. The hold is capped at max_age_s: past
that the stretch is walked so block boundaries still land correctly, but
nothing accumulates and elapsed does not grow, which is what finally makes the
comment about a gap dragging the billed average down true. Same threshold for
control and billing: a reading too old to steer by is too old to bill by.

T-4, the out-of-order/duplicate guard is covered. It was untested, and the
reason is worth recording: the obvious assertion passes without the guard,
because the negative interval is separately refused by the covered > 0 test.
What the guard prevents is the timestamp REWIND, which only shows up one sample
later as a re-integrated window. The test now goes one sample later.

T-6, DOCS was wrong about latency. meter_max_age_s and stale_input_s stack, so
meter death to 0 W is 45 s and not 30. Documented as a table with both clocks.

Also documented the T-5 asymmetry rather than papering over it: the age
measures arrival, not change, so a stuck MQTT bridge republishing its last
telegram still looks fresh. Correct on ha_dsmr, not detectable on mqtt_p1
without a change-detector. Written up as a known limit.

Writing the T-1 test caught a second defect in the test itself: it recorded
only MQTT topics, and object_id lives in the payload, so "the age sensor is not
announced" had been passing for the wrong reason.

test_p1.py: 99 -> 122 checks. 14 mutations run, all 14 red, files restored
byte-identical - including one per fix above.

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

goodwe_controller/DOCS.md

@@ -60,13 +60,30 @@ that subtraction into the add-on, where it is done once and tested, and replaces
 |---|---|---|
 | `meter_source` | `off` | `off` keeps `meter_entity`. `ha_dsmr` subscribes to the DSMR integration over the HA WebSocket; `mqtt_p1` reads a topic |
 | `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: grid power reads as *missing*, and the existing failsafe commands 0 W |
+| `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 |
 
+#### 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
@@ -102,6 +119,18 @@ emits nothing, which is indistinguishable — to anything watching the value —
 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
+> `ha_dsmr` path 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 `ha_dsmr` where
+> both are available.
+
 ### Control
 
 | option | default | meaning |
@@ -116,7 +145,7 @@ telegrams stop and resets when they arrive, whatever the reading says.
 | `saturation_cycles` | 3 | How many consecutive cycles before freezing. **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) |
 
 #### Why `target_grid_w` is not zero

goodwe_controller/app/main.py

@@ -39,7 +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
+from .p1 import P1Ingest, build_source, is_enabled
 from . import web
 
 OPTIONS_PATH = "/data/options.json"
@@ -96,7 +96,7 @@ class Controller:
         # 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 = str(opts.get("meter_source", "off")) not in ("off", "")
+        self.p1_enabled = is_enabled(opts)
 
         # live state
         self.auto = bool(store.data.get("auto", opts.get("auto_start", False)))
@@ -324,16 +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.
-            "p1_age": round(self.p1.published_age_s, 1),
-        })
+            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."""
@@ -501,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)

goodwe_controller/app/mqtt.py

@@ -60,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:
@@ -94,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

@@ -213,8 +213,16 @@ class QuarterAverager:
     halves credited to the two blocks, never attributed wholly to either.
     """
 
-    def __init__(self, phases: int = 1):
+    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
@@ -273,16 +281,22 @@ class QuarterAverager:
             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)
-            dt = stop - cursor
-            if dt > 0:
-                self._acc += max(self._last_net, 0.0) * dt
+            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) * dt
-                self._elapsed += dt
+                        self._pp_acc[i] += max(v, 0.0) * covered
+                self._elapsed += covered
             cursor = stop
             if stop < end:
                 break
@@ -321,7 +335,9 @@ class P1Ingest:
     def __init__(self, phases: int = 1, max_age_s: float = 30.0):
         self.phases = phases
         self.max_age_s = float(max_age_s)
-        self.averager = QuarterAverager(phases)
+        # 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
@@ -477,9 +493,17 @@ class HaDsmrSource:
                     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"))
-                    self._schedule()
                 elif payload.get("type") == "event":
                     data = (payload.get("event") or {}).get("data") or {}
                     if data.get("entity_id") not in self.ids:
@@ -647,6 +671,18 @@ class MqttP1Source:
 # --------------------------------------------------------------------------- #
 # 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.
 
@@ -654,7 +690,7 @@ def build_source(opts: dict, ingest: P1Ingest, session, broker: dict | None):
     changing transport is a config edit, never a code path.
     """
     source = str(opts.get("meter_source", "off") or "off").strip()
-    if source in ("off", ""):
+    if not is_enabled(opts):
         return None
     if source == SOURCE_HA:
         return HaDsmrSource(session, ingest, {

goodwe_controller/test_p1.py

@@ -229,6 +229,47 @@ 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")
@@ -482,9 +523,15 @@ async def _e2e():
 
 live, wire = asyncio.run(_e2e())
 check("the websocket handshake and subscription complete", live.samples >= 1)
-# Six state_changed events arrived (two per telegram). The debounce is what
-# makes that three consistent samples instead of six half-updated ones.
-check("three telegrams produce three samples, not six", live.samples == 3)
+# ⚠️ 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",
@@ -492,9 +539,102 @@ check("the sample was built over the wire, tagged with its transport",
 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 == 3)
+      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.5)
+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("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.p1 import is_enabled          # noqa: E402
+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))
 
 print()
 if fails: