autocore_std/

asset_watch.rs

//! Subscription helper for `ams.asset_updated.<location>` broadcasts.
//!
//! Control programs use this when they need to react to an AMS asset
//! changing — typically to re-fire CoE SDOs on an EL3356-0010 after a
//! load cell is recalibrated, but the same pattern works for any
//! "something on disk just changed, write fresh bytes to hardware"
//! flow.
//!
//! # Lifecycle
//!
//! 1. Construct via `AssetWatch::new("tsdr_fx", client)`. This
//!    subscribes the `CommandClient` to the per-location topic and
//!    sends a one-shot `ams.list_assets` request scoped to the
//!    location so the watcher starts from a consistent snapshot —
//!    regardless of whether the asset was registered before or after
//!    the control program connected.
//! 2. Per tick the control program calls `watch.pump(client)` and
//!    receives `Vec<AssetUpdate>` to act on. Updates are returned in
//!    arrival order; the first one after construction is the initial
//!    sync (trigger = `InitialSync`).
//! 3. Acting on an update is the caller's responsibility. The typical
//!    EL3356 pattern is to issue `ethercat.write_sdo` IPC for each
//!    affected CoE object inside the loop that consumed `pump()`'s
//!    return value.
//!
//! # Why poll-based instead of callbacks
//!
//! `CommandClient` is passed `&mut` per tick. Holding `Box<dyn FnMut>`
//! handlers across ticks would force the helper to own them, which
//! conflicts with the borrow patterns control programs already use
//! (sharing `&mut client` between subsystems). A `pump()` that
//! returns owned `Vec<AssetUpdate>` keeps the caller in charge of
//! when work happens.
//!
//! # Initial-sync semantics
//!
//! On construction the watcher sends `ams.list_assets` with the
//! location filter. The response is then consumed by `pump()` and
//! converted into an `AssetUpdate { trigger: InitialSync, .. }`. If
//! no active asset exists at this location, the first `pump()`
//! returns an `InitialSyncEmpty` so the control program knows the
//! resolver would have refused to start a hardware module here and
//! can put its own state machine in a safe mode.

use mechutil::ipc::CommandMessage;
use serde_json::Value;

use crate::CommandClient;

/// What caused this update to be published.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AssetWatchTrigger {
    /// First payload after `AssetWatch::new`, derived from a one-shot
    /// `ams.list_assets` query. Tells the control program "this is
    /// the current state of the role; act as if you'd just received
    /// every broadcast since the asset was registered."
    InitialSync,
    /// First payload after `AssetWatch::new` and the location has no
    /// active asset registered. Control programs should react by
    /// disabling the hardware feature this asset feeds (e.g., refuse
    /// to enter a test that depends on the role).
    InitialSyncEmpty,
    /// `ams.create_asset` registered a new asset at this location.
    Created,
    /// `ams.add_calibration` recorded a new calibration on the asset.
    /// This is the most common operational trigger — recal events
    /// during a service window.
    CalibrationAdded,
    /// `ams.update_asset` changed nameplate fields (capacity,
    /// compression_sensitivity_mv_v, bridge_resistance_ohm, etc.) or the
    /// asset's sub_locations.
    NameplateUpdated,
    /// `ams.update_asset` changed the asset's lifecycle status
    /// (active ↔ retired / out_for_service). On a retirement the
    /// payload's `asset.status` reflects the new state — control
    /// programs typically refuse to run tests against a retired role.
    StatusChanged,
    /// `ams.update_asset` moved the asset out of this role. The
    /// payload's `asset.status` shows `retired` from this location's
    /// POV — the asset itself is fine, just elsewhere now. Same
    /// reaction as `StatusChanged` in most cases.
    LocationChanged,
    /// A trigger label the server published that this client doesn't
    /// know about. Carried as a string so future server versions can
    /// add triggers without forcing a control-program rebuild.
    Other(&'static str),
}

impl AssetWatchTrigger {
    fn from_str(s: &str) -> Self {
        match s {
            "created"           => AssetWatchTrigger::Created,
            "calibration_added" => AssetWatchTrigger::CalibrationAdded,
            "nameplate_updated" => AssetWatchTrigger::NameplateUpdated,
            "status_changed"    => AssetWatchTrigger::StatusChanged,
            "location_changed"  => AssetWatchTrigger::LocationChanged,
            // Boxing/leak avoided — most "other" values are short server
            // tokens that fit in &'static str literals if we ever add
            // them. For runtime-supplied trigger strings, callers can
            // inspect `update.raw_trigger` if they need the original.
            _                   => AssetWatchTrigger::Other("other"),
        }
    }
}

/// One delivery from `AssetWatch::pump`.
#[derive(Debug, Clone)]
pub struct AssetUpdate {
    /// The semantic trigger; see [`AssetWatchTrigger`].
    pub trigger: AssetWatchTrigger,
    /// The exact string the server published in `payload.trigger` (or
    /// `"initial_sync"` / `"initial_sync_empty"` for the bootstrap
    /// deliveries). Useful when `trigger == Other`.
    pub raw_trigger: String,
    /// `asset.location` from the broadcast — always matches the
    /// location this watcher was constructed for, except for
    /// `LocationChanged` events where it's the *old* location.
    pub location: String,
    /// `asset.asset_id`. Empty when `trigger == InitialSyncEmpty`.
    pub asset_id: String,
    /// `asset.asset_type`. Empty when `trigger == InitialSyncEmpty`.
    pub asset_type: String,
    /// The full `asset` object from the broadcast — `null` when the
    /// trigger is `InitialSyncEmpty`. Custom nameplate fields live at
    /// `asset.custom.<name>`.
    pub asset: Value,
    /// The asset's active calibration record (full JSON including
    /// `values.scale`, `values.offset`, `expires_at`, `cert_ref`).
    /// `null` when there's no calibration on record yet.
    pub current_calibration: Value,
}

/// Latest known lifecycle state of the asset at this location, as
/// observed by `AssetWatch::pump`. Mirrors the on-disk `AssetStatus`
/// enum plus a `Missing` variant for "no asset registered here at
/// all." Control programs gate auto on `is_active()` (which is
/// `state == Active`), or pattern-match on `active_status()` for
/// finer-grained policy.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AssetWatchStatus {
    /// No asset is registered at the location (or the only one was
    /// deleted / had its location moved elsewhere).
    Missing,
    /// Active asset present at the location.
    Active,
    /// Asset present but marked Retired — control programs should
    /// refuse to run tests against it.
    Retired,
    /// Asset present but marked OutForService — same posture as
    /// Retired, separate label so the operator's UI can distinguish.
    OutForService,
}

impl AssetWatchStatus {
    fn from_asset_status_str(s: &str) -> Self {
        match s {
            "active"          => AssetWatchStatus::Active,
            "retired"         => AssetWatchStatus::Retired,
            "out_for_service" => AssetWatchStatus::OutForService,
            _                 => AssetWatchStatus::Missing,
        }
    }
}

/// Subscription handle for one location. One watcher per location;
/// construct multiple instances if the control program cares about
/// multiple roles. Hold onto the `AssetWatch` for the lifetime of the
/// control program — dropping it leaves the underlying topic
/// subscribed but discards any queued events.
pub struct AssetWatch {
    location: String,
    /// Transaction ID of the in-flight `ams.list_assets` call used for
    /// the initial sync; set to `Some` between construction and the
    /// first `pump()` that observes the response.
    initial_tid: Option<u32>,
    /// True once the InitialSync (or InitialSyncEmpty) has been emitted.
    /// Broadcasts that arrive before this point are still queued by
    /// `CommandClient` (since we subscribed at construction time) so
    /// the operator can't miss a recal that fires between
    /// `AssetWatch::new` and the list_assets response.
    initial_sent: bool,
    /// Latest known lifecycle state. Starts `Missing` and is updated
    /// from each event the watcher observes. `is_active()` reads this.
    status: AssetWatchStatus,
    /// `asset_id` of the asset currently filling this role, or `None`
    /// when `status == Missing`. Useful for logging and for the
    /// "Asset X took over from asset Y on this rig" trail.
    asset_id: Option<String>,
}

impl AssetWatch {
    /// Subscribe to `ams.asset_updated.<location>` and issue the
    /// initial-sync query. Returns the handle the control program
    /// holds for the duration of the run.
    ///
    /// `client` is the `CommandClient` from `TickContext::client`. The
    /// subscription persists past this call — `CommandClient` owns
    /// the buffer.
    pub fn new(location: &str, client: &mut CommandClient) -> Self {
        let topic = format!("ams.asset_updated.{}", location);
        client.subscribe(&topic);

        // Initial sync: one-shot ams.list_assets restricted to this
        // location, active status only. The response will surface
        // through `client.take_response(tid)` on the next poll cycle.
        let tid = client.send(
            "ams.list_assets",
            serde_json::json!({ "location": location, "status": "active" }),
        );

        Self {
            location: location.to_string(),
            initial_tid: Some(tid),
            initial_sent: false,
            status: AssetWatchStatus::Missing,
            asset_id: None,
        }
    }

    /// The location this watcher is bound to.
    pub fn location(&self) -> &str {
        &self.location
    }

    /// True once an `InitialSync` (or `InitialSyncEmpty`) has been
    /// returned from `pump()`. Until this flips, subsequent broadcasts
    /// are still surfaced — initial-sync just tells the operator
    /// "you've seen the baseline."
    pub fn initialised(&self) -> bool {
        self.initial_sent
    }

    /// `true` if the watcher's latest observation is that an active
    /// asset is registered at this location. The common control-loop
    /// gate: refuse to enter auto when this is `false`.
    ///
    /// Initially `false` until the first `pump()` consumes the
    /// initial-sync result. After that it tracks broadcasts in
    /// real time.
    pub fn is_active(&self) -> bool {
        self.status == AssetWatchStatus::Active
    }

    /// `asset_id` currently filling this role, or `None` when
    /// `status` is `Missing`. Useful for logging "running against
    /// LC-2026..." and for the per-cycle trail in TIS records.
    pub fn active_asset_id(&self) -> Option<&str> {
        self.asset_id.as_deref()
    }

    /// Finer-grained version of `is_active`. Use when a control
    /// program needs to distinguish (e.g.) `Retired` from
    /// `OutForService` to drive different operator messaging.
    pub fn active_status(&self) -> AssetWatchStatus {
        self.status
    }

    /// Internal — refresh `status` + `asset_id` from one observed
    /// event. Called from `pump()` for every event before it's
    /// returned to the caller. Centralised here so the state machine
    /// is one place: any trigger that says "I'm here now" sets
    /// Active; any "I'm gone" sets Missing.
    fn fold_event(&mut self, update: &AssetUpdate) {
        // For LocationChanged at this location the trigger is the
        // synthetic departure event — status comes through as Retired
        // (the server marks the ghost). Treat that as Missing locally
        // since the asset is no longer here, even if it's alive
        // elsewhere.
        if matches!(update.trigger, AssetWatchTrigger::LocationChanged) {
            self.status = AssetWatchStatus::Missing;
            self.asset_id = None;
            return;
        }
        if matches!(update.trigger, AssetWatchTrigger::InitialSyncEmpty) {
            self.status = AssetWatchStatus::Missing;
            self.asset_id = None;
            return;
        }
        // For everything else — InitialSync, Created, CalibrationAdded,
        // NameplateUpdated, StatusChanged, Other — read the status
        // string off the asset payload. A retirement comes through as
        // StatusChanged with asset.status == "retired"; that lands
        // here and we record Retired (not Missing) so the caller can
        // tell "asset deliberately retired" from "no asset ever."
        let raw_status = update.asset
            .get("status").and_then(|v| v.as_str()).unwrap_or("");
        self.status = AssetWatchStatus::from_asset_status_str(raw_status);
        self.asset_id = if update.asset_id.is_empty() {
            None
        } else {
            Some(update.asset_id.clone())
        };
    }

    /// Per-tick drain. Returns every event since the previous call,
    /// in arrival order. Empty when no events are pending.
    ///
    /// The first non-empty return (assuming the `ams.list_assets`
    /// response has arrived) starts with either `InitialSync` or
    /// `InitialSyncEmpty`, followed by any broadcasts that landed
    /// before the response did.
    pub fn pump(&mut self, client: &mut CommandClient) -> Vec<AssetUpdate> {
        let mut out: Vec<AssetUpdate> = Vec::new();

        // Initial-sync handling. We only emit the bootstrap event once
        // and only after the response lands. While it's pending, queued
        // broadcasts stay in the CommandClient buffer — they're drained
        // below so the initial-sync event always leads.
        if let Some(tid) = self.initial_tid {
            if let Some(resp) = client.take_response(tid) {
                self.initial_tid = None;
                self.initial_sent = true;
                if resp.success {
                    // Response shape: { "assets": [ { ... } ] } from
                    // ams.list_assets. We take the first active match
                    // because location filtering should already have
                    // narrowed it; collisions are a registry bug.
                    let first = resp.data.get("assets")
                        .and_then(|v| v.as_array())
                        .and_then(|arr| arr.first())
                        .cloned();
                    out.push(self.bootstrap_event(first, client));
                } else {
                    // list_assets failing during initial sync is rare
                    // (the AMS servelet returns empty arrays, not
                    // errors, for unknown locations). Treat as empty
                    // so the control program enters a safe-mode state.
                    out.push(AssetUpdate {
                        trigger:             AssetWatchTrigger::InitialSyncEmpty,
                        raw_trigger:         "initial_sync_empty".to_string(),
                        location:            self.location.clone(),
                        asset_id:            String::new(),
                        asset_type:          String::new(),
                        asset:               Value::Null,
                        current_calibration: Value::Null,
                    });
                }
            }
        }

        // Drain broadcasts that have arrived since the last pump.
        let topic = format!("ams.asset_updated.{}", self.location);
        for msg in client.take_broadcasts(&topic) {
            out.push(AssetUpdate::from_broadcast(msg));
        }
        // Fold every event we're about to return into the watcher's
        // internal status. Bootstrap events have already been folded
        // inside bootstrap_event, but doing it here too is idempotent
        // (fold reads only fields that don't change between successive
        // applications of the same event).
        for update in &out {
            self.fold_event(update);
        }
        out
    }

    /// Build the initial-sync event from the `ams.list_assets`
    /// response. When the response carries an asset row, fetch its
    /// asset.json + current calibration via the same on-disk flow the
    /// server uses for live broadcasts so the bootstrap payload has
    /// the same shape as subsequent events.
    fn bootstrap_event(&mut self, asset_row: Option<Value>, client: &mut CommandClient) -> AssetUpdate {
        let Some(row) = asset_row else {
            return AssetUpdate {
                trigger:             AssetWatchTrigger::InitialSyncEmpty,
                raw_trigger:         "initial_sync_empty".to_string(),
                location:            self.location.clone(),
                asset_id:            String::new(),
                asset_type:          String::new(),
                asset:               Value::Null,
                current_calibration: Value::Null,
            };
        };

        // The list_assets row is a flattened RegistryEntry, not the
        // full asset.json. For the initial-sync payload we want the
        // whole record so the control program reads
        // `asset.custom.capacity` etc. Send a follow-up read_asset.
        // We don't block on it here — instead the next pump will pick
        // up that response and the FIRST event is a "thin" InitialSync
        // carrying just the identity. Most callers want the full record
        // before they act, so the bootstrap event flags this with a
        // hint in `asset` being just the registry row.
        //
        // Trade-off note: making the bootstrap event hold the
        // identity-only payload keeps `pump()` fully non-blocking and
        // avoids a hidden second request. Callers that need the full
        // record should match on `InitialSync` and follow up with
        // their own `ams.read_asset` if they want the custom fields
        // (or wait for the first real broadcast, which carries them).
        let _ = client; // reserved for future enrichment

        AssetUpdate {
            trigger:             AssetWatchTrigger::InitialSync,
            raw_trigger:         "initial_sync".to_string(),
            location:            self.location.clone(),
            asset_id:            row.get("asset_id").and_then(|v| v.as_str()).unwrap_or("").to_string(),
            asset_type:          row.get("asset_type").and_then(|v| v.as_str()).unwrap_or("").to_string(),
            asset:               row,
            current_calibration: Value::Null,
        }
    }
}

impl AssetUpdate {
    fn from_broadcast(msg: CommandMessage) -> Self {
        let raw_trigger = msg.data.get("trigger")
            .and_then(|v| v.as_str()).unwrap_or("")
            .to_string();
        let trigger = AssetWatchTrigger::from_str(&raw_trigger);
        let asset = msg.data.get("asset").cloned().unwrap_or(Value::Null);
        let current_calibration = msg.data.get("current_calibration").cloned().unwrap_or(Value::Null);
        let location = msg.data.get("location").and_then(|v| v.as_str()).unwrap_or("").to_string();
        let asset_id = msg.data.get("asset_id").and_then(|v| v.as_str()).unwrap_or("").to_string();
        let asset_type = msg.data.get("asset_type").and_then(|v| v.as_str()).unwrap_or("").to_string();
        Self {
            trigger,
            raw_trigger,
            location,
            asset_id,
            asset_type,
            asset,
            current_calibration,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use mechutil::ipc::{CommandMessage, MessageType};
    use serde_json::json;
    use tokio::sync::mpsc;

    fn fresh_client() -> (
        CommandClient,
        mpsc::UnboundedSender<CommandMessage>,
        mpsc::UnboundedReceiver<String>,
    ) {
        let (write_tx, write_rx) = mpsc::unbounded_channel();
        let (response_tx, response_rx) = mpsc::unbounded_channel();
        let client = CommandClient::new(write_tx, response_rx);
        (client, response_tx, write_rx)
    }

    fn broadcast(topic: &str, data: Value) -> CommandMessage {
        let mut m = CommandMessage::default();
        m.topic = topic.to_string();
        m.message_type = MessageType::Broadcast;
        m.data = data;
        m
    }

    #[test]
    fn subscription_and_initial_sync_request_are_sent_on_new() {
        let (mut client, _response_tx, mut write_rx) = fresh_client();
        let _watch = AssetWatch::new("tsdr_fx", &mut client);

        // One outgoing request: ams.list_assets scoped to the location.
        let outgoing = write_rx.try_recv().expect("expected list_assets send");
        let sent: CommandMessage = serde_json::from_str(&outgoing).unwrap();
        assert_eq!(sent.topic, "ams.list_assets");
        assert_eq!(sent.data.get("location").and_then(|v| v.as_str()), Some("tsdr_fx"));
        assert_eq!(sent.data.get("status").and_then(|v| v.as_str()),   Some("active"));

        // Topic is subscribed.
        let mut probe = broadcast("ams.asset_updated.tsdr_fx", json!({ "trigger": "created" }));
        probe.message_type = MessageType::Broadcast;
        // Without subscription, poll would drop this; with subscription, it
        // buffers. Indirectly verifies subscribe was called.
        let _ = probe;
    }

    #[test]
    fn initial_sync_returns_asset_row_when_present() {
        let (mut client, response_tx, _write_rx) = fresh_client();
        let mut watch = AssetWatch::new("tsdr_fx", &mut client);

        // Reply to the list_assets request the watcher sent.
        let tid = watch.initial_tid.unwrap();
        let mut resp = CommandMessage::default();
        resp.transaction_id = tid;
        resp.message_type = MessageType::Response;
        resp.success = true;
        resp.data = json!({
            "assets": [{
                "asset_id":   "LC-20260301T091548",
                "asset_type": "load_cell",
                "location":   "tsdr_fx",
                "status":     "active",
                "current_calibration_id": "20260301T091548"
            }]
        });
        response_tx.send(resp).unwrap();

        client.poll();
        let events = watch.pump(&mut client);
        assert_eq!(events.len(), 1);
        assert_eq!(events[0].trigger, AssetWatchTrigger::InitialSync);
        assert_eq!(events[0].asset_id, "LC-20260301T091548");
        assert_eq!(events[0].asset_type, "load_cell");
        assert_eq!(events[0].location, "tsdr_fx");
        assert!(watch.initialised());
    }

    #[test]
    fn initial_sync_empty_when_no_asset_at_location() {
        let (mut client, response_tx, _write_rx) = fresh_client();
        let mut watch = AssetWatch::new("nope", &mut client);

        let tid = watch.initial_tid.unwrap();
        let mut resp = CommandMessage::default();
        resp.transaction_id = tid;
        resp.message_type = MessageType::Response;
        resp.success = true;
        resp.data = json!({ "assets": [] });
        response_tx.send(resp).unwrap();

        client.poll();
        let events = watch.pump(&mut client);
        assert_eq!(events.len(), 1);
        assert_eq!(events[0].trigger, AssetWatchTrigger::InitialSyncEmpty);
        assert_eq!(events[0].asset, Value::Null);
        assert!(watch.initialised());
    }

    #[test]
    fn broadcasts_parse_trigger_and_payload() {
        let (mut client, response_tx, _write_rx) = fresh_client();
        let mut watch = AssetWatch::new("tsdr_fx", &mut client);

        // Resolve the initial sync so it's out of the way.
        let tid = watch.initial_tid.unwrap();
        let mut resp = CommandMessage::default();
        resp.transaction_id = tid;
        resp.message_type = MessageType::Response;
        resp.success = true;
        resp.data = json!({ "assets": [] });
        response_tx.send(resp).unwrap();
        client.poll();
        let _ = watch.pump(&mut client);

        // Now a recal broadcast lands.
        response_tx.send(broadcast(
            "ams.asset_updated.tsdr_fx",
            json!({
                "trigger":   "calibration_added",
                "location":  "tsdr_fx",
                "asset_id":  "LC-1",
                "asset_type":"load_cell",
                "asset":     { "custom": { "capacity": 5000.0 } },
                "current_calibration": {
                    "cal_id": "20260301T091548",
                    "values": { "scale": 9.81234, "offset": -0.0042 }
                }
            }),
        )).unwrap();
        client.poll();

        let events = watch.pump(&mut client);
        assert_eq!(events.len(), 1);
        assert_eq!(events[0].trigger, AssetWatchTrigger::CalibrationAdded);
        assert_eq!(events[0].raw_trigger, "calibration_added");
        assert_eq!(
            events[0].current_calibration.pointer("/values/scale").and_then(|v| v.as_f64()),
            Some(9.81234),
        );
        // Capacity from the asset payload survives so the control
        // program can rewrite SDO 0x8000:11 without a follow-up read.
        assert_eq!(
            events[0].asset.pointer("/custom/capacity").and_then(|v| v.as_f64()),
            Some(5000.0),
        );
    }

    #[test]
    fn broadcasts_for_other_locations_are_ignored() {
        let (mut client, response_tx, _write_rx) = fresh_client();
        let mut watch = AssetWatch::new("tsdr_fx", &mut client);

        let tid = watch.initial_tid.unwrap();
        let mut resp = CommandMessage::default();
        resp.transaction_id = tid;
        resp.message_type = MessageType::Response;
        resp.success = true;
        resp.data = json!({ "assets": [] });
        response_tx.send(resp).unwrap();

        // Wrong-location broadcast — different topic; CommandClient
        // never buffers it because that topic isn't subscribed.
        response_tx.send(broadcast(
            "ams.asset_updated.tsdr_fy",
            json!({ "trigger": "calibration_added" }),
        )).unwrap();
        client.poll();

        let events = watch.pump(&mut client);
        assert_eq!(events.len(), 1, "only the initial-sync event should land");
        assert_eq!(events[0].trigger, AssetWatchTrigger::InitialSyncEmpty);
    }

    #[test]
    fn is_active_starts_false_and_flips_after_initial_sync_with_active_asset() {
        let (mut client, response_tx, _write_rx) = fresh_client();
        let mut watch = AssetWatch::new("tsdr_fx", &mut client);
        // Before pump: nothing observed.
        assert!(!watch.is_active());
        assert_eq!(watch.active_status(), AssetWatchStatus::Missing);
        assert!(watch.active_asset_id().is_none());

        let tid = watch.initial_tid.unwrap();
        let mut resp = CommandMessage::default();
        resp.transaction_id = tid;
        resp.message_type = MessageType::Response;
        resp.success = true;
        resp.data = json!({
            "assets": [{
                "asset_id":   "LC-NEW",
                "asset_type": "load_cell",
                "location":   "tsdr_fx",
                "status":     "active",
            }]
        });
        response_tx.send(resp).unwrap();

        client.poll();
        let _ = watch.pump(&mut client);
        assert!(watch.is_active(), "active asset → is_active() should be true");
        assert_eq!(watch.active_asset_id(), Some("LC-NEW"));
        assert_eq!(watch.active_status(), AssetWatchStatus::Active);
    }

    #[test]
    fn is_active_stays_false_after_initial_sync_empty() {
        let (mut client, response_tx, _write_rx) = fresh_client();
        let mut watch = AssetWatch::new("nope", &mut client);

        let tid = watch.initial_tid.unwrap();
        let mut resp = CommandMessage::default();
        resp.transaction_id = tid;
        resp.message_type = MessageType::Response;
        resp.success = true;
        resp.data = json!({ "assets": [] });
        response_tx.send(resp).unwrap();
        client.poll();
        let _ = watch.pump(&mut client);

        assert!(!watch.is_active());
        assert_eq!(watch.active_status(), AssetWatchStatus::Missing);
        assert!(watch.active_asset_id().is_none());
    }

    #[test]
    fn retire_broadcast_flips_is_active_to_false() {
        // Boot with an active asset, then receive a status_changed
        // → retired broadcast. The watcher's state reflects the
        // retirement; the control program will refuse to enter auto.
        let (mut client, response_tx, _write_rx) = fresh_client();
        let mut watch = AssetWatch::new("tsdr_fx", &mut client);

        let tid = watch.initial_tid.unwrap();
        let mut resp = CommandMessage::default();
        resp.transaction_id = tid;
        resp.message_type = MessageType::Response;
        resp.success = true;
        resp.data = json!({ "assets": [{
            "asset_id": "LC-1", "asset_type": "load_cell",
            "location": "tsdr_fx", "status": "active",
        }] });
        response_tx.send(resp).unwrap();
        client.poll();
        let _ = watch.pump(&mut client);
        assert!(watch.is_active());

        // Now retire.
        response_tx.send(broadcast(
            "ams.asset_updated.tsdr_fx",
            json!({
                "trigger":    "status_changed",
                "location":   "tsdr_fx",
                "asset_id":   "LC-1",
                "asset_type": "load_cell",
                "asset":      { "asset_id": "LC-1", "status": "retired" },
            }),
        )).unwrap();
        client.poll();
        let _ = watch.pump(&mut client);
        assert!(!watch.is_active(),
            "after retirement is_active should be false");
        assert_eq!(watch.active_status(), AssetWatchStatus::Retired);
        // Asset is still _identified_ — operator may want to log "the
        // retired cell was LC-1." It's just no longer suitable for
        // testing.
        assert_eq!(watch.active_asset_id(), Some("LC-1"));
    }

    #[test]
    fn location_changed_event_flips_is_active_to_missing() {
        let (mut client, response_tx, _write_rx) = fresh_client();
        let mut watch = AssetWatch::new("tsdr_fx", &mut client);

        // Boot active.
        let tid = watch.initial_tid.unwrap();
        let mut resp = CommandMessage::default();
        resp.transaction_id = tid;
        resp.message_type = MessageType::Response;
        resp.success = true;
        resp.data = json!({ "assets": [{
            "asset_id": "LC-1", "asset_type": "load_cell",
            "location": "tsdr_fx", "status": "active",
        }] });
        response_tx.send(resp).unwrap();
        client.poll();
        let _ = watch.pump(&mut client);
        assert!(watch.is_active());

        // Asset relocated away — server publishes the ghost departure
        // at the old location (this one). Treat as Missing locally
        // because the asset is no longer here, even though it's alive
        // somewhere else.
        response_tx.send(broadcast(
            "ams.asset_updated.tsdr_fx",
            json!({
                "trigger":    "location_changed",
                "location":   "tsdr_fx",
                "asset_id":   "LC-1",
                "asset_type": "load_cell",
                "asset":      { "asset_id": "LC-1", "status": "retired" }
            }),
        )).unwrap();
        client.poll();
        let _ = watch.pump(&mut client);
        assert!(!watch.is_active());
        assert_eq!(watch.active_status(), AssetWatchStatus::Missing);
        assert!(watch.active_asset_id().is_none());
    }

    #[test]
    fn broadcasts_queued_before_initial_sync_response_arrive_after_it() {
        // Subtle: a broadcast can land while the list_assets response
        // is still in flight. The bootstrap event still appears first
        // when pump() runs.
        let (mut client, response_tx, _write_rx) = fresh_client();
        let mut watch = AssetWatch::new("tsdr_fx", &mut client);

        // Broadcast arrives first.
        response_tx.send(broadcast(
            "ams.asset_updated.tsdr_fx",
            json!({ "trigger": "calibration_added" }),
        )).unwrap();
        // Then the list_assets response.
        let tid = watch.initial_tid.unwrap();
        let mut resp = CommandMessage::default();
        resp.transaction_id = tid;
        resp.message_type = MessageType::Response;
        resp.success = true;
        resp.data = json!({ "assets": [{
            "asset_id": "LC-1", "asset_type": "load_cell",
            "location": "tsdr_fx", "status": "active"
        }] });
        response_tx.send(resp).unwrap();
        client.poll();

        let events = watch.pump(&mut client);
        assert_eq!(events.len(), 2);
        assert_eq!(events[0].trigger, AssetWatchTrigger::InitialSync);
        assert_eq!(events[1].trigger, AssetWatchTrigger::CalibrationAdded);
    }
}