fix(agent): stop reconnects and restarts erasing device truth from the status topic

README.md

@@ -82,9 +82,20 @@ Protocol constants, tape geometry and the day-one bring-up checklist live in
 [`HARDWARE-NOTES.md`](HARDWARE-NOTES.md). Read it before changing anything in
 `labelfab.device`.
 
-**There is no read channel.** The D30 never reports back, so "printed" means "the bytes
-were accepted and we waited the physical print duration". Out-of-tape, jams and low
-battery are undetectable.
+**There is a read channel**, contrary to the received wisdom: the D30 answers queries on
+both transports and pushes media-state changes unsolicited. The retained status topic
+carries what it said — serial, firmware, battery percentage and terminal voltage, and
+whether media is loaded — not just whether the agent is up.
+
+It says so *and when it said so*. The printer is only reachable while it is being printed
+to (it auto-powers-off, and a held-open socket just relocates that), so the agent
+remembers the last thing it heard and republishes it, stamped with `device_seen_at`.
+A consumer can then tell live truth from remembered truth instead of guessing, and no
+printer is woken to keep a status page tidy. `device.probe_on_start` surveys the printer
+once at startup when it happens to be awake.
+
+Jams remain undetectable — the media bit distinguishes loaded from not, and nothing
+observed so far distinguishes a jam from either.
 
 ## License
 

deploy/agent.toml.example

@@ -42,6 +42,10 @@ raster_width_px = 96         # 96 = 12mm head (verified); 120 = 15mm (day-1 hypo
 pace_factor = 1.2            # lower until a long strip garbles, then +50%
 density = 1                  # 1 light | 2 medium | 4 heavy; raise if codes misscan
 wake_dummy_feed = false
+# Survey the printer once at startup so the status topic carries live truth instead of
+# what was remembered from the last print. Cannot wake a sleeping unit, so a miss costs
+# one connect timeout and the stored status is published unchanged.
+probe_on_start = true
 
 [tape]
 width_mm = 15.0

src/labelfab/agent/__init__.py

@@ -11,6 +11,7 @@
 
 from labelfab.agent.coalescer import Batch, Coalescer, PendingLabel
 from labelfab.agent.config import Config, load
+from labelfab.agent.device_state import DeviceSnapshot
 from labelfab.agent.publisher import NullPublisher, Publisher, RecordingPublisher
 from labelfab.agent.spool import InsertResult, Outcome, Spool
 from labelfab.agent.worker import PrintWorker
@@ -19,6 +20,7 @@
     "Batch",
     "Coalescer",
     "Config",
+    "DeviceSnapshot",
     "InsertResult",
     "NullPublisher",
     "Outcome",

src/labelfab/agent/__main__.py

@@ -93,6 +93,11 @@ def run(self) -> int:
         signal.signal(signal.SIGINT, self.stop)
         signal.signal(signal.SIGTERM, self.stop)
 
+        # Before the source starts, so whatever these two learn is what the first
+        # retained status carries: the source republishes the last thing published,
+        # and nothing has been published yet.
+        if self.config.device.probe_on_start:
+            self.worker.probe_device()
         self.worker.recover()
         if self.dir_source:
             self.dir_source.poll()

src/labelfab/agent/config.py

@@ -95,6 +95,13 @@ class DeviceSection(BaseModel):
     #: Drop the socket between batches: the D30 auto-sleeps, so a held-open socket
     #: just relocates the failure. Reconnect-per-batch is cheaper to reason about.
     idle_disconnect: bool = True
+    #: Ask the printer what it is once at startup, so the status topic carries live
+    #: truth rather than remembered truth from the moment the agent comes up. This
+    #: cannot wake a sleeping unit -- it has powered its radio off and only the button
+    #: brings it back -- so the cost of a miss is one connect timeout, after which the
+    #: stored snapshot is published unchanged. On by default; turn it off if that
+    #: timeout is in the way of a fast start. Never runs on a broker reconnect.
+    probe_on_start: bool = True
 
     @field_validator("density")
     @classmethod

src/labelfab/agent/device_state.py

@@ -0,0 +1,123 @@
+"""What the printer last said about itself, and when it said it.
+
+The D30 only talks while it is being printed to. It auto-powers-off, and
+``device.idle_disconnect`` defaults to true precisely because a held-open socket just
+relocates the failure -- so between jobs there is nobody to ask. Every consumer of the
+status topic therefore reads *remembered* device truth most of the time, and the honest
+way to serve that is to remember it deliberately and say how old it is, rather than
+either forgetting it or waking the printer to refresh it.
+
+``seen_at`` is the whole point. Without it a consumer cannot distinguish "media is
+loaded" from "media was loaded at some unknown time in the past", and rendering the
+second as the first is exactly the failure the tri-state ``media_ok`` was introduced to
+avoid, one layer along.
+
+This also collects the six loose ``_device_*`` fields the worker used to carry and the
+two places that built a ``PrinterStatus`` out of them. That duplication is why the
+worker and the MQTT source disagreed: one published everything the printer had said,
+the other published a hardcoded blank.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+from pydantic import BaseModel, ConfigDict
+
+from labelfab.contract import PrinterStatus
+from labelfab.device.d30 import MODEL
+from labelfab.device.feedback import DeviceFeedback
+
+
+class DeviceSnapshot(BaseModel):
+    """Last-known device truth, persisted in the spool and republished on connect."""
+
+    # ``extra="ignore"`` is load-bearing rather than a default: this is stored in the
+    # spool DB, which outlives package upgrades, so a row written by a different agent
+    # version has to degrade to the fields it has in common. Failing to decode it would
+    # take the agent down at boot, and a status field is not worth an outage.
+    model_config = ConfigDict(extra="ignore", frozen=True)
+
+    serial: str | None = None
+    firmware: str | None = None
+    battery_pct: int | None = None
+    voltage_v: float | None = None
+    media_ok: bool | None = None
+    #: What the printer was complaining about the last time it was asked. Never latched
+    #: -- see ``merge``.
+    fault: str | None = None
+    #: Epoch seconds when the printer last reported anything, or ``None`` if it never
+    #: has. Stored as a float to match the rest of the spool; converted to a UTC
+    #: instant at the wire boundary.
+    seen_at: float | None = None
+
+    # -- accumulation ------------------------------------------------------- #
+
+    def merge(self, fb: DeviceFeedback, *, now: float) -> DeviceSnapshot:
+        """Fold one connection's feedback in, keeping whatever it did not report.
+
+        A connection that answered nothing must not blank out a serial we already
+        know. ``fault`` is the deliberate exception: it is recomputed from scratch
+        every connection, so a cleared media error clears the status instead of
+        latching an error forever.
+
+        ``seen_at`` only moves when the printer actually said something. Advancing it
+        on a silent connection would claim a freshness we do not have, and this
+        timestamp is the only thing standing between a consumer and treating
+        three-day-old truth as live.
+        """
+        reported = (fb.serial, fb.firmware, fb.battery_pct, fb.voltage_v, fb.paper_ok)
+        return DeviceSnapshot(
+            serial=fb.serial or self.serial,
+            firmware=fb.firmware or self.firmware,
+            battery_pct=self.battery_pct if fb.battery_pct is None else fb.battery_pct,
+            voltage_v=self.voltage_v if fb.voltage_v is None else fb.voltage_v,
+            media_ok=self.media_ok if fb.paper_ok is None else fb.paper_ok,
+            fault=fb.fault(),
+            seen_at=now if any(v is not None for v in reported) else self.seen_at,
+        )
+
+    # -- publishing --------------------------------------------------------- #
+
+    def settled_state(self) -> str:
+        """What to publish once nothing is in flight.
+
+        "idle" unless the printer is complaining. Kept apart from the status builder so
+        that "printing" and "disconnected" -- facts about the link, not about the media
+        -- are never silently rewritten into it.
+        """
+        return "error" if self.fault else "idle"
+
+    def to_status(
+        self,
+        printer_id: str,
+        *,
+        state: str,
+        tape_width_mm: float | None = None,
+        pending_labels: int = 0,
+    ) -> PrinterStatus:
+        """Build the retained status message from this snapshot."""
+        return PrinterStatus(
+            printer_id=printer_id,
+            state=state,  # type: ignore[arg-type]
+            model=MODEL,
+            serial=self.serial,
+            firmware=self.firmware,
+            battery_pct=self.battery_pct,
+            voltage_v=self.voltage_v,
+            media_ok=self.media_ok,
+            tape_width_mm=tape_width_mm,
+            pending_labels=pending_labels,
+            error=self.fault,
+            device_seen_at=self.seen_at_utc(),
+        )
+
+    def seen_at_utc(self) -> datetime | None:
+        """``seen_at`` as a UTC instant, whole seconds.
+
+        Truncated because sub-second precision on "when did the printer last speak" is
+        noise in a payload whose primary reader is a human running ``mosquitto_sub -v``.
+        """
+        if self.seen_at is None:
+            return None
+        return datetime.fromtimestamp(self.seen_at, tz=timezone.utc).replace(microsecond=0)

src/labelfab/agent/source_mqtt.py

@@ -55,35 +55,75 @@ def __init__(self, config: Config, spool: Spool, enqueue: Callable[[str], None])
         if config.mqtt.username:
             self.client.username_pw_set(config.mqtt.username, config.mqtt.password)
 
-        # Last will: if the agent drops, the retained status flips to disconnected so
-        # a producer sees the printer is unreachable before it ever clicks print.
-        will = PrinterStatus(printer_id=config.agent.printer_id, state="disconnected")
-        self.client.will_set(config.topic("status"), will.model_dump_json(), qos=1, retain=True)
+        # What the retained topic currently says. Seeded from the spool so a fresh
+        # process starts out holding the previous one's device truth, then updated by
+        # every retained publish -- which is what lets a reconnect restore the topic
+        # instead of flattening it.
+        snapshot = spool.device_snapshot()
+        self._last_status = snapshot.to_status(
+            config.agent.printer_id,
+            state=snapshot.settled_state(),
+            tape_width_mm=config.tape.width_mm,
+        )
 
         self.client.on_connect = self._on_connect
         self.client.on_message = self._on_message
 
     # -- lifecycle ---------------------------------------------------------- #
 
     def start(self) -> None:
+        # Armed here rather than in __init__ so it reflects whatever the startup probe
+        # and crash recovery learned, which happen after this object is built.
+        self._arm_will()
         self.client.connect(self.config.mqtt.host, self.config.mqtt.port, self.config.mqtt.keepalive_s)
         self.client.loop_start()
 
+    def _arm_will(self) -> None:
+        """Point the will at the current last-known truth.
+
+        The will lives in the CONNECT packet, so the broker holds a fixed payload for
+        the whole session and no amount of calling this changes what fires if *this*
+        connection drops. paho rebuilds CONNECT from these fields on every reconnect
+        though, so re-arming still moves the staleness window from "whenever the
+        process started" -- potentially weeks -- down to "the last reconnect", which
+        here is a few hours at most. Called from ``_on_connect``, i.e. from paho's own
+        network thread, which is also the thread that builds the packet, so there is
+        no window where the payload and the topic disagree.
+        """
+        self.client.will_set(
+            self.config.topic("status"), self._disconnected().model_dump_json(), qos=1, retain=True
+        )
+
     def stop(self) -> None:
         try:
-            self.publish_status(
-                PrinterStatus(printer_id=self.config.agent.printer_id, state="disconnected")
-            )
+            self.publish_status(self._disconnected())
         finally:
             self.client.loop_stop()
             self.client.disconnect()
 
+    def _disconnected(self) -> PrinterStatus:
+        """Last-known truth with the link marked down.
+
+        Backs both the will and the shutdown notice: if the agent drops, a producer
+        must see that the printer is unreachable before it ever clicks print. Losing
+        the link is news about reachability, though, not grounds for forgetting the
+        serial, the firmware and the media state -- ``device_seen_at`` rides along to
+        say how old they are.
+        """
+        return self._last_status.model_copy(update={"state": "disconnected", "pending_labels": 0})
+
     # -- Publisher protocol ------------------------------------------------- #
 
     def publish_result(self, result: JobResult) -> None:
         self.client.publish(self.config.topic("results"), result.model_dump_json(), qos=1)
 
     def publish_status(self, status: PrinterStatus, *, retain: bool = True) -> None:
+        if retain:
+            # Recorded before the publish and regardless of whether it succeeds: the
+            # startup probe and crash recovery both publish before the client is
+            # connected, so those go nowhere and the connect below is what lands them.
+            # A non-retained publish is not what the topic holds, so it is not recorded.
+            self._last_status = status
         self.client.publish(self.config.topic("status"), status.model_dump_json(), qos=1, retain=retain)
 
     def publish_progress(self, job_id: str, printed: int, total: int) -> None:
@@ -98,13 +138,15 @@ def _on_connect(self, client, userdata, flags, reason_code, properties=None) ->
             return
         client.subscribe(self.config.topic("jobs"), qos=self.config.mqtt.qos)
         client.subscribe(self.config.topic("cmd"), qos=1)
-        self.publish_status(
-            PrinterStatus(
-                printer_id=self.config.agent.printer_id,
-                state="idle",
-                tape_width_mm=self.config.tape.width_mm,
-            )
-        )
+        # Republish what the topic already said, rather than a freshly synthesised
+        # blank. This fires on every reconnect, not just the first connect, and paho
+        # reconnects silently -- the Istio route for the broker caps at a 24h timeout
+        # and flaps besides, so it runs several times a day. Publishing a hardcoded
+        # bare status here was not merely uninformative, it was lossy: it overwrote a
+        # retained message that was correct a moment earlier, and the printer is asleep
+        # and cannot be asked again.
+        self.publish_status(self._last_status)
+        self._arm_will()  # for the connect after this one; see _arm_will
 
     def _on_message(self, client, userdata, msg) -> None:
         if msg.topic.endswith("/cmd"):

src/labelfab/agent/spool.py

@@ -14,19 +14,30 @@
   its cached result and prints nothing.
 * ``dedupe_key`` deduplicates individual *labels* across different jobs, so a "print
   bin A4" that fires twice an hour apart does not waste tape.
+
+One thing here is not about jobs at all: the last :class:`DeviceSnapshot`. It is stored
+alongside them because this file is already the agent's only durable place, and because
+the printer is only reachable during a print -- so the alternative to remembering what
+it said is waking it up to ask again.
 """
 
 from __future__ import annotations
 
+import logging
 import sqlite3
 import time
 from collections.abc import Callable
 from dataclasses import dataclass
 from enum import Enum
 from pathlib import Path
 
+from pydantic import ValidationError
+
+from labelfab.agent.device_state import DeviceSnapshot
 from labelfab.contract import JobResult, PrintJob
 
+log = logging.getLogger("labelfab.spool")
+
 SCHEMA = """
 CREATE TABLE IF NOT EXISTS jobs (
     job_id          TEXT PRIMARY KEY,
@@ -51,6 +62,14 @@
     printed_at  REAL NOT NULL,
     PRIMARY KEY (job_id, label_index)
 );
+
+-- One row, holding the last DeviceSnapshot. The printer only talks while it is being
+-- printed to, so without this the agent starts every process knowing nothing about the
+-- hardware it drives until the next job happens to come along.
+CREATE TABLE IF NOT EXISTS device_state (
+    id      INTEGER PRIMARY KEY CHECK (id = 1),
+    payload TEXT NOT NULL
+);
 """
 
 
@@ -186,3 +205,34 @@ def printed_labels(self, job_id: str) -> set[int]:
             "SELECT label_index FROM printed_labels WHERE job_id = ?", (job_id,)
         ).fetchall()
         return {r["label_index"] for r in rows}
+
+    # -- last-known device truth -------------------------------------------- #
+    #
+    # Written after every print, read once at startup. This is the only state here that
+    # is not about jobs, and it lives here because the alternative -- asking the printer
+    # on boot -- means waking a device whose whole operating model is "asleep until
+    # needed", every time the agent restarts.
+
+    def save_device_snapshot(self, snapshot: DeviceSnapshot) -> None:
+        self._db.execute(
+            "INSERT INTO device_state (id, payload) VALUES (1, ?) "
+            "ON CONFLICT(id) DO UPDATE SET payload = excluded.payload",
+            (snapshot.model_dump_json(),),
+        )
+
+    def device_snapshot(self) -> DeviceSnapshot:
+        """The last snapshot written, or an empty one.
+
+        Empty covers both "this agent has never printed" and "the stored row is no
+        longer readable". The second degrades rather than raising on purpose: an
+        unusable status field is a cosmetic problem, and taking the print agent down at
+        boot over one would be a much larger one.
+        """
+        row = self._db.execute("SELECT payload FROM device_state WHERE id = 1").fetchone()
+        if row is None:
+            return DeviceSnapshot()
+        try:
+            return DeviceSnapshot.model_validate_json(row["payload"])
+        except ValidationError:
+            log.warning("stored device snapshot is unreadable; starting without one")
+            return DeviceSnapshot()