rustuya 0.3.0-rc.1

//! Tuya device communication and state management.
//!
//! Handles TCP connections, handshakes, heartbeats, and command-response flows.

use crate::crypto::TuyaCipher;
use crate::crypto::hex_encode;
use crate::error::{
    ERR_DEVTYPE, ERR_JSON, ERR_OFFLINE, ERR_PAYLOAD, ERR_STATE, ERR_SUCCESS, Result, TuyaError,
    get_error_message,
};
use crate::protocol::{
    CommandType, DeviceType, PREFIX_55AA, PREFIX_6699, TuyaHeader, TuyaMessage, Version,
    get_protocol, pack_message, parse_header, unpack_message,
};
use crate::scanner::get as get_scanner;
use futures_core::stream::Stream;
use log::{debug, error, info, trace, warn};
use parking_lot::RwLock;
use rand::Rng;
use serde::Serialize;
use serde_json::Value;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpStream;
use tokio::sync::{mpsc, oneshot};
use tokio::time::{Interval, MissedTickBehavior, interval, sleep, timeout};
use tokio_util::sync::CancellationToken;

const SLEEP_HEARTBEAT_DEFAULT: Duration = Duration::from_secs(7);
const SLEEP_HEARTBEAT_CHECK: Duration = Duration::from_secs(5);
const SLEEP_RECONNECT_MIN: Duration = Duration::from_secs(16);
const SLEEP_RECONNECT_MAX: Duration = Duration::from_secs(4096);
const SLEEP_INACTIVITY_TIMEOUT: Duration = Duration::from_secs(30);

/// Base cooldown between scanner-triggered backoff bypasses when the
/// discovered IP matches what the device already knows. Doubles on each
/// useless bypass and resets on a successful connect.
const SCANNER_BYPASS_BASE_COOLDOWN: Duration = Duration::from_secs(60);

const ADDR_AUTO: &str = "Auto";
const DATA_UNVALID: &str = "data unvalid";

const CHAN_BROADCAST_CAPACITY: usize = 128;
const CHAN_MPSC_CAPACITY: usize = 64;

/// Commands that must return data (payload) and should not return on empty ACK.
const MANDATORY_DATA_CMDS: &[u32] = &[CommandType::LanExtStream as u32];

/// Commands that do not produce a response we should wait for
/// (handshake handshake-internal commands and heartbeats).
const NO_RESPONSE_CMDS: &[u32] = &[
    CommandType::SessKeyNegStart as u32,
    CommandType::SessKeyNegResp as u32,
    CommandType::SessKeyNegFinish as u32,
    CommandType::HeartBeat as u32,
];

mod keys {
    pub const REQ_TYPE: &str = "reqType";

    // Response keys
    pub const ERR_CODE: &str = "errorCode";
    pub const ERR_MSG: &str = "errorMsg";
    pub const ERR_PAYLOAD_OBJ: &str = "errorPayload";
    pub const PAYLOAD_STR: &str = "payloadStr";
    pub const PAYLOAD_RAW: &str = "payloadRaw";

    // Inbound payload keys consulted when synthesizing error messages from
    // device responses. Kept here so a future protocol change has one site
    // to update.
    pub const IN_DATA: &str = "data";
    pub const IN_PAYLOAD: &str = "payload";
}

/// A sub-device (endpoint) of a gateway device.
#[derive(Clone)]
pub struct SubDevice {
    parent: Device,
    cid: String,
}

impl SubDevice {
    pub(crate) fn new(parent: Device, cid: &str) -> Self {
        Self {
            parent,
            cid: cid.to_string(),
        }
    }

    #[must_use]
    pub fn id(&self) -> &str {
        &self.cid
    }

    pub async fn status(&self) -> Result<Option<String>> {
        self.request(CommandType::DpQuery, None).await
    }

    pub async fn set_dps(&self, dps: Value) -> Result<Option<String>> {
        self.request(CommandType::Control, Some(dps)).await
    }

    /// Sets a single DP value.
    pub async fn set_value<I: ToString, T: Serialize>(
        &self,
        index: I,
        value: T,
    ) -> Result<Option<String>> {
        if let Ok(val) = serde_json::to_value(value) {
            self.set_dps(serde_json::json!({ index.to_string(): val }))
                .await
        } else {
            Err(TuyaError::InvalidPayload)
        }
    }

    pub async fn request(&self, cmd: CommandType, data: Option<Value>) -> Result<Option<String>> {
        self.parent.request(cmd, data, Some(self.cid.clone())).await
    }
}

enum DeviceCommand {
    // NOTE: `Disconnect` removed in M2.1. Use `close_notify` on DeviceInner for
    // graceful disconnect — going through the mpsc would queue behind pending
    // user requests, defeating the point of a fast close.
    Request {
        command: CommandType,
        data: Option<Value>,
        cid: Option<String>,
        resp_tx: oneshot::Sender<Result<Option<TuyaMessage>>>,
    },
    ConnectNow,
}

impl DeviceCommand {
    fn respond(self, result: Result<Option<TuyaMessage>>) {
        if let DeviceCommand::Request { resp_tx, .. } = self {
            let _ = resp_tx.send(result);
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ConnectionState {
    Disconnected,
    Connecting,
    Connected,
    Stopped,
}

struct DeviceState {
    config_address: String,
    real_ip: String,
    version: Version,
    port: u16,
    dev_type: DeviceType,
    state: ConnectionState,
    last_received: Instant,
    last_sent: Instant,
    persist: bool,
    session_key: Option<Vec<u8>>,
    failure_count: u32,
    success_count: u32,
    force_discovery: bool,
    timeout: Duration,
    cipher: Option<Arc<TuyaCipher>>,
    last_reported_discovered_ip: Option<String>,
    last_scanner_bypass_at: Option<Instant>,
    scanner_bypass_failures: u32,
}

pub struct DeviceBuilder {
    id: String,
    address: String,
    local_key: Vec<u8>,
    version: Version,
    dev_type: DeviceType,
    port: u16,
    persist: bool,
    timeout: Duration,
    nowait: bool,
}

impl DeviceBuilder {
    pub fn new<I, K>(id: I, local_key: K) -> Self
    where
        I: Into<String>,
        K: Into<Vec<u8>>,
    {
        Self {
            id: id.into(),
            address: ADDR_AUTO.to_string(),
            local_key: local_key.into(),
            version: Version::Auto,
            dev_type: DeviceType::Auto,
            port: 6668,
            persist: true,
            timeout: Duration::from_secs(10),
            nowait: false,
        }
    }

    pub fn address<A: Into<String>>(mut self, address: A) -> Self {
        self.address = address.into();
        self
    }

    pub fn version<V: Into<Version>>(mut self, version: V) -> Self {
        self.version = version.into();
        self
    }

    pub fn dev_type<DT: Into<DeviceType>>(mut self, dev_type: DT) -> Self {
        self.dev_type = dev_type.into();
        self
    }

    #[must_use]
    pub fn port(mut self, port: u16) -> Self {
        self.port = port;
        self
    }

    #[must_use]
    pub fn persist(mut self, persist: bool) -> Self {
        self.persist = persist;
        self
    }

    #[must_use]
    pub fn timeout(mut self, timeout: Duration) -> Self {
        self.timeout = timeout;
        self
    }

    #[must_use]
    pub fn nowait(mut self, nowait: bool) -> Self {
        self.nowait = nowait;
        self
    }

    /// Finalizes the builder and returns a connected [`Device`].
    ///
    /// This is the conventional builder terminal verb. The older
    /// [`run`](Self::run) name is retained as a deprecated alias and will be
    /// removed in a future minor release.
    #[must_use]
    pub fn build(self) -> Device {
        Device::with_builder(self)
    }

    /// Deprecated: prefer [`build`](Self::build).
    #[must_use]
    #[deprecated(since = "0.3.0", note = "use `DeviceBuilder::build` instead")]
    pub fn run(self) -> Device {
        self.build()
    }
}

struct DeviceInner {
    id: String,
    local_key: Vec<u8>,
    state: RwLock<DeviceState>,
    broadcast_tx: tokio::sync::broadcast::Sender<TuyaMessage>,
    cancel_token: CancellationToken,
    // Side-channel for graceful disconnect. `close()` notifies this; the
    // actor's `select!` short-circuits the current connection without
    // dragging shutdown behind any queued user requests. Using a Notify here
    // instead of an mpsc message means close() doesn't have to wait for the
    // actor to drain its command queue.
    close_notify: tokio::sync::Notify,
    nowait: AtomicBool,
}

impl Drop for DeviceInner {
    fn drop(&mut self) {
        self.cancel_token.cancel();
        debug!(
            "DeviceInner for {} dropped, cancelling connection task",
            self.id
        );
    }
}

#[derive(Clone)]
pub struct Device {
    inner: Arc<DeviceInner>,
    tx: Option<mpsc::Sender<DeviceCommand>>,
}

impl Device {
    /// Creates a new device with default settings and starts the background
    /// connection task.
    ///
    /// # Runtime behavior
    ///
    /// This is an **eager** constructor: it spawns a long-running tokio task
    /// on the library's dedicated runtime (see [`crate::runtime`]) before
    /// returning. The task stays alive — performing reconnects, heartbeats,
    /// and IP rediscovery — until the last `Device` clone is dropped or
    /// [`Device::stop`] is called.
    ///
    /// Consequences:
    ///
    /// - Calling `Device::new` outside any tokio runtime is fine; the
    ///   internal runtime takes care of execution.
    /// - You can `Device::new` from inside an async context as well; the
    ///   spawned task does *not* attach to the caller's runtime, so
    ///   consumer-side runtime shutdown won't kill it.
    /// - The constructor itself returns quickly — actual TCP work happens on
    ///   the spawned task.
    /// - Construct devices lazily if you may have hundreds-to-thousands of
    ///   them; each one starts a task on construction.
    pub fn new<I, K>(id: I, local_key: K) -> Self
    where
        I: Into<String>,
        K: Into<Vec<u8>>,
    {
        DeviceBuilder::new(id, local_key).build()
    }

    /// Returns a builder to configure device settings before running.
    pub fn builder<I, K>(id: I, local_key: K) -> DeviceBuilder
    where
        I: Into<String>,
        K: Into<Vec<u8>>,
    {
        DeviceBuilder::new(id, local_key)
    }

    pub(crate) fn with_builder(builder: DeviceBuilder) -> Self {
        let (addr, ip) = match builder.address.as_str() {
            "" | ADDR_AUTO => (ADDR_AUTO.to_string(), String::new()),
            _ => (builder.address.clone(), builder.address),
        };

        let (broadcast_tx, _) = tokio::sync::broadcast::channel(CHAN_BROADCAST_CAPACITY);
        let (tx, rx) = mpsc::channel(CHAN_MPSC_CAPACITY);
        let state = DeviceState {
            config_address: addr,
            real_ip: ip,
            version: builder.version,
            port: builder.port,
            dev_type: builder.dev_type,
            state: ConnectionState::Disconnected,
            last_received: Instant::now(),
            last_sent: Instant::now(),
            persist: builder.persist,
            session_key: None,
            failure_count: 0,
            success_count: 0,
            force_discovery: false,
            timeout: builder.timeout,
            cipher: TuyaCipher::new(&builder.local_key).ok().map(Arc::new),
            last_reported_discovered_ip: None,
            last_scanner_bypass_at: None,
            scanner_bypass_failures: 0,
        };

        let inner = Arc::new(DeviceInner {
            id: builder.id,
            local_key: builder.local_key,
            state: RwLock::new(state),
            broadcast_tx,
            cancel_token: CancellationToken::new(),
            close_notify: tokio::sync::Notify::new(),
            nowait: AtomicBool::new(builder.nowait),
        });

        let device = Self {
            inner: Arc::clone(&inner),
            tx: Some(tx),
        };

        let inner_weak = Arc::downgrade(&inner);
        let d_id = device.inner.id.clone();
        crate::runtime::spawn(async move {
            if let Some(inner) = inner_weak.upgrade() {
                let cancel_token = inner.cancel_token.clone();
                let d_task = Device { inner, tx: None };

                tokio::select! {
                    () = cancel_token.cancelled() => {
                        debug!("Device {d_id} connection task stopped via token");
                    }
                    () = d_task.run_connection_task(rx) => {
                        debug!("Device {d_id} connection task finished");
                    }
                }
            }
        });
        device
    }

    #[must_use]
    pub fn id(&self) -> &str {
        &self.inner.id
    }

    // ----------------------------------------------------------------
    // Getter implementation note
    //
    // These getters acquire `parking_lot::RwLock::read()` and return cheap
    // Copy values. We deliberately did not factor each field into a separate
    // atomic shadow: parking_lot's uncontended read is ~10ns, the getters
    // are not called in tight loops (the genuinely hot path,
    // `get_cipher`, was addressed separately with read-then-upgrade
    // locking — see M2.3), and dual storage between an atomic and the
    // canonical `DeviceState` would create a synchronization invariant
    // that's easy to forget on a setter. If a future profile shows one of
    // these is hot, the right move is to atomicize that single field.
    // ----------------------------------------------------------------

    #[must_use]
    pub fn dev_type(&self) -> DeviceType {
        self.with_state(|s| s.dev_type)
    }

    #[must_use]
    pub fn local_key(&self) -> &[u8] {
        &self.inner.local_key
    }

    #[must_use]
    pub fn address(&self) -> String {
        self.with_state(|s| {
            if s.real_ip.is_empty() {
                s.config_address.clone()
            } else {
                s.real_ip.clone()
            }
        })
    }

    /// Returns the user-configured address (e.g., "Auto" or a specific IP).
    #[must_use]
    pub fn config_address(&self) -> String {
        self.with_state(|s| s.config_address.clone())
    }

    #[must_use]
    pub fn version(&self) -> Version {
        self.with_state(|s| s.version)
    }

    #[must_use]
    pub fn is_connected(&self) -> bool {
        self.with_state(|s| s.state == ConnectionState::Connected)
    }

    #[must_use]
    pub fn is_stopped(&self) -> bool {
        self.with_state(|s| s.state == ConnectionState::Stopped)
    }

    /// Returns the timeout duration for network operations and responses.
    #[must_use]
    pub fn timeout(&self) -> Duration {
        self.with_state(|s| s.timeout)
    }

    #[must_use]
    pub fn port(&self) -> u16 {
        self.with_state(|s| s.port)
    }

    #[must_use]
    pub fn persist(&self) -> bool {
        self.with_state(|s| s.persist)
    }

    /// Returns whether the device is in nowait mode.
    #[must_use]
    pub fn nowait(&self) -> bool {
        self.inner.nowait.load(Ordering::Relaxed)
    }
}

impl Device {
    pub fn set_persist(&self, persist: bool) {
        self.with_state_mut(|s| s.persist = persist);
    }

    pub fn set_timeout(&self, timeout: Duration) {
        self.with_state_mut(|s| s.timeout = timeout);
    }

    pub fn set_port(&self, port: u16) {
        self.with_state_mut(|s| s.port = port);
    }

    /// Sets whether requests should wait for a response from the device.
    /// If true, methods like `status()` and `set_value()` will return immediately after
    /// dispatching the command, without waiting for the network response.
    pub fn set_nowait(&self, nowait: bool) {
        self.inner.nowait.store(nowait, Ordering::Relaxed);
    }

    pub fn set_version<V: Into<Version>>(&self, version: V) {
        let ver = version.into();

        self.with_state_mut(|s| {
            s.version = ver;
            // If dev_type is Auto, we can either leave it as Auto (to allow future detection)
            // or initialize it to Default. Given the user's requirement that only Auto
            // allows switching, we should keep it as Auto if the user hasn't specified Default.
        });
    }

    pub fn set_dev_type<DT: Into<DeviceType>>(&self, dev_type: DT) {
        self.with_state_mut(|s| s.dev_type = dev_type.into());
    }

    pub fn set_address<A: Into<String>>(&self, address: A) {
        let addr = address.into();
        self.with_state_mut(|s| {
            s.config_address = addr;
            s.force_discovery = true; // Force discovery to update real_ip if needed
        });
    }
}

impl Device {
    /// Returns a stream of broadcast messages from this device.
    ///
    /// The returned stream holds a strong reference to the device. As long as
    /// the stream is alive (e.g. owned by a spawned task or
    /// [`unified_listener`]), the device's background connection task stays
    /// alive too — even if the caller dropped their original `Device` handle.
    /// To force shutdown, call [`Device::stop`].
    pub fn listener(&self) -> impl Stream<Item = Result<TuyaMessage>> + Send + 'static {
        use tokio::sync::broadcast::error::RecvError;
        let mut rx = self.inner.broadcast_tx.subscribe();
        let device = self.clone();
        let cancel = self.inner.cancel_token.clone();
        async_stream::stream! {
            loop {
                tokio::select! {
                    // Honor explicit stop() while waiting on the broadcast.
                    // Without this branch the listener would only exit when
                    // broadcast_tx is dropped, which requires the very last
                    // Arc<DeviceInner> reference to disappear — and *we* are
                    // holding one of those refs ourselves.
                    () = cancel.cancelled() => break,
                    res = rx.recv() => match res {
                        Ok(msg) => {
                            if !msg.payload.is_empty() {
                                yield Ok(msg);
                            }
                        }
                        Err(RecvError::Lagged(skipped)) => {
                            warn!(
                                "Listener for device {} lagged behind broadcast, skipped {} messages",
                                device.inner.id, skipped
                            );
                            yield Ok(device.error_helper(
                                ERR_STATE,
                                Some(serde_json::json!({
                                    "reason": "listener_lagged",
                                    "skipped": skipped,
                                })),
                            ));
                        }
                        Err(RecvError::Closed) => break,
                    }
                }
            }
        }
    }

    /// Queries the device's current data points.
    ///
    /// # Note on request/response correlation
    ///
    /// The Tuya LAN protocol does not provide a reliable correlation token
    /// between an outbound request and its response — many firmwares echo
    /// `seqno=0` or a device-side counter unrelated to what was sent. The
    /// actor matches responses by command-id (+ optional CID), so an
    /// unsolicited `Status` push that arrives between subscribe-and-send may
    /// be returned as this call's response.
    ///
    /// This is by design (the protocol is asynchronous and fire-and-forget on
    /// the device side); callers that need strict correlation should
    /// serialize their own calls per `Device` handle, or use
    /// [`listener`](Self::listener) for asynchronous device-pushed events.
    pub async fn status(&self) -> Result<Option<String>> {
        self.request(CommandType::DpQuery, None, None).await
    }

    /// Sets multiple DP values at once.
    /// The `dps` argument should be a `serde_json::Value` object where keys are DP IDs.
    pub async fn set_dps(&self, dps: Value) -> Result<Option<String>> {
        self.request(CommandType::Control, Some(dps), None).await
    }

    /// Sets a single DP value by its ID.
    /// The `dp_id` can be provided as any type that can be converted to a String (e.g., u32, &str).
    /// The `value` can be any type that implements `Serialize` (e.g., bool, i32, String, `serde_json::Value`).
    pub async fn set_value<I: ToString, T: Serialize>(
        &self,
        dp_id: I,
        value: T,
    ) -> Result<Option<String>> {
        if let Ok(val) = serde_json::to_value(value) {
            self.set_dps(serde_json::json!({ dp_id.to_string(): val }))
                .await
        } else {
            Err(TuyaError::InvalidPayload)
        }
    }

    pub async fn sub_discover(&self) -> Result<Option<String>> {
        let data = serde_json::json!({
            "cids": [],
            keys::REQ_TYPE: "subdev_online_stat_query"
        });
        self.request(CommandType::LanExtStream, Some(data), None)
            .await
    }

    pub async fn receive(&self) -> Result<TuyaMessage> {
        use tokio::sync::broadcast::error::RecvError;
        let mut rx = self.inner.broadcast_tx.subscribe();
        loop {
            match rx.recv().await {
                Ok(msg) => {
                    if !msg.payload.is_empty() {
                        return Ok(msg);
                    }
                }
                Err(RecvError::Lagged(skipped)) => {
                    warn!(
                        "Device {} receive() lagged, skipped {} broadcast messages",
                        self.inner.id, skipped
                    );
                    continue;
                }
                Err(RecvError::Closed) => return Err(TuyaError::Offline),
            }
        }
    }

    #[must_use]
    pub fn sub(&self, cid: &str) -> SubDevice {
        SubDevice::new(self.clone(), cid)
    }

    /// Sends an arbitrary command and waits for the device's response.
    ///
    /// See [`status`](Self::status) for the same caveat about request/response
    /// correlation in the Tuya LAN protocol.
    pub async fn request(
        &self,
        command: CommandType,
        data: Option<Value>,
        cid: Option<String>,
    ) -> Result<Option<String>> {
        debug!("request: cmd={command:?}, data={data:?}");
        let resp = self
            .send_command_to_task(|resp_tx| DeviceCommand::Request {
                command,
                data,
                cid,
                resp_tx,
            })
            .await?;

        match resp {
            Some(msg) => {
                if let Some(s) = msg.payload_as_string() {
                    Ok(Some(s))
                } else {
                    Ok(Some(hex_encode(&msg.payload)))
                }
            }
            None => Ok(None),
        }
    }
}

impl Device {
    pub async fn close(&self) {
        self.fire_close();
    }

    pub async fn stop(&self) {
        self.fire_stop();
    }

    /// Synchronous variant of [`close`] — both Notify and the state lock are
    /// synchronously usable, so the `async` flavor is just a thin wrapper.
    pub fn fire_close(&self) {
        info!("Closing connection to device {}", self.inner.id);

        self.with_state_mut(|state| {
            if state.state != ConnectionState::Stopped {
                state.state = ConnectionState::Disconnected;
            }
        });

        // Signal the actor via the dedicated close channel. Going through the
        // user-command mpsc would queue the disconnect behind any pending
        // request; the Notify path interrupts the actor's `select!` directly.
        self.inner.close_notify.notify_waiters();
    }

    /// Synchronous variant of [`stop`].
    pub fn fire_stop(&self) {
        info!("Stopping device {} (explicit stop called)", self.inner.id);
        self.with_state_mut(|state| {
            state.state = ConnectionState::Stopped;
        });
        // Notify close FIRST so the actor drops its current connection
        // promptly, THEN cancel the token so the outer loop exits. Ordering
        // matters: if we cancelled first, the `select!` would race against
        // mid-flight reconnect logic.
        self.inner.close_notify.notify_waiters();
        self.inner.cancel_token.cancel();
    }

    /// Forces the device to attempt a connection immediately, bypassing any backoff.
    pub async fn connect_now(&self) {
        self.send_to_task(DeviceCommand::ConnectNow).await;
    }
}

impl Device {
    fn with_state<R>(&self, f: impl FnOnce(&DeviceState) -> R) -> R {
        f(&self.inner.state.read())
    }

    fn with_state_mut<R>(&self, f: impl FnOnce(&mut DeviceState) -> R) -> R {
        f(&mut self.inner.state.write())
    }

    fn broadcast_error(&self, code: u32, payload: Option<Value>) {
        let _ = self
            .inner
            .broadcast_tx
            .send(self.error_helper(code, payload));
    }

    fn update_last_received(&self) {
        self.inner.state.write().last_received = Instant::now();
    }

    fn update_last_sent(&self) {
        self.inner.state.write().last_sent = Instant::now();
    }

    fn reset_failure_count(&self) {
        let mut state = self.inner.state.write();
        state.success_count += 1;
        if state.failure_count > 0 && state.success_count >= 3 {
            debug!(
                "Resetting failure count for device {} (success_count: {})",
                self.inner.id, state.success_count
            );
            state.failure_count = 0;
            state.success_count = 0;
        }
    }

    async fn send_to_task(&self, cmd: DeviceCommand) {
        if let Some(tx) = &self.tx {
            if let Err(e) = tx.send(cmd).await {
                error!(
                    "Failed to queue command for device {}: {}",
                    self.inner.id, e
                );
            }
        } else {
            error!(
                "Cannot send command for device {}: task not running",
                self.inner.id
            );
        }
    }

    async fn send_command_to_task(
        &self,
        cmd_generator: impl FnOnce(oneshot::Sender<Result<Option<TuyaMessage>>>) -> DeviceCommand,
    ) -> Result<Option<TuyaMessage>> {
        let (resp_tx, resp_rx) = oneshot::channel();
        self.send_to_task(cmd_generator(resp_tx)).await;
        if !self.inner.nowait.load(Ordering::Relaxed) {
            resp_rx.await.map_err(|_| TuyaError::Offline)?
        } else {
            Ok(None)
        }
    }

    fn get_timestamp(&self) -> u64 {
        match SystemTime::now().duration_since(UNIX_EPOCH) {
            Ok(d) => d.as_secs(),
            Err(e) => {
                // System clock is before UNIX_EPOCH. Tuya devices typically
                // reject requests whose `t` field is too far off, so a 0
                // timestamp will mostly fail anyway — but the silent
                // `.unwrap_or_default()` it replaces made that failure mode
                // very confusing. Warn once per call so an operator at least
                // sees the cause in the logs.
                warn!(
                    "System clock is before UNIX_EPOCH on device {} ({:?}); \
                     sending t=0 which the device will likely reject",
                    self.inner.id, e
                );
                0
            }
        }
    }
}

/// Represents an event from a specific device.
#[derive(Debug, Clone, Serialize)]
pub struct DeviceEvent {
    /// The ID of the device that generated the event.
    pub device_id: String,
    /// The message received from the device.
    pub message: TuyaMessage,
}

/// Merges multiple device listeners into a single stream of events.
pub fn unified_listener(
    devices: Vec<Device>,
) -> impl Stream<Item = Result<DeviceEvent>> + Send + 'static {
    use futures_util::StreamExt;
    use futures_util::stream::select_all;

    let streams = devices.into_iter().map(|device| {
        let device_id = device.id().to_string();
        device
            .listener()
            .map(move |res| match res {
                Ok(message) => Ok(DeviceEvent {
                    device_id: device_id.clone(),
                    message,
                }),
                Err(e) => Err(e),
            })
            .boxed()
    });

    select_all(streams)
}

impl Device {
    async fn run_connection_task(&self, mut rx: mpsc::Receiver<DeviceCommand>) {
        let jitter = {
            let mut rng = rand::rng();
            Duration::from_millis(u64::from(rng.next_u32() % 5000))
        };

        debug!(
            "Starting background connection task for device {} with {:?} initial jitter",
            self.inner.id, jitter
        );

        // Stagger connection attempts
        tokio::select! {
            () = self.inner.cancel_token.cancelled() => return,
            () = sleep(jitter) => {}
        }

        let mut heartbeat_interval = interval(SLEEP_HEARTBEAT_CHECK);
        heartbeat_interval.set_missed_tick_behavior(MissedTickBehavior::Skip);

        loop {
            tokio::select! {
                () = self.inner.cancel_token.cancelled() => {
                    debug!("Background task for {} received stop signal", self.inner.id);
                    break;
                }
                res = async {
                    if self.is_stopped() {
                        return Some(());
                    }

                    // Reset seqno for each new connection attempt
                    let mut seqno = 1u32;

                    // 1. Connect and handshake
                    let (stream, initial_cmd) = match self
                        .try_connect_with_backoff(&mut rx, &mut seqno)
                        .await
                    {
                        Some(res) => res,
                        None => return Some(()),
                    };

                    // 2. Connection maintenance
                    let result = self
                        .maintain_connection(stream, &mut rx, &mut seqno, &mut heartbeat_interval, initial_cmd)
                        .await;

                    self.handle_disconnect(result.as_ref().err().cloned());

                    if let Err(e) = result {
                        self.with_state_mut(|s| {
                            s.failure_count += 1;
                            s.success_count = 0;
                        });
                        self.drain_rx(&mut rx, e, false);
                    } else {
                        return Some(());
                    }

                    if self.is_stopped() {
                        return Some(());
                    }

                    None
                } => {
                    if res.is_some() {
                        break;
                    }
                }
            }
        }

        // Ensure all associated tasks (like the Reader task) are stopped
        self.inner.cancel_token.cancel();
        debug!("Background connection task for {} exited", self.inner.id);
    }

    async fn maintain_connection(
        &self,
        stream: TcpStream,
        rx: &mut mpsc::Receiver<DeviceCommand>,
        seqno: &mut u32,
        heartbeat_interval: &mut Interval,
        initial_cmd: Option<DeviceCommand>,
    ) -> Result<()> {
        let (mut read_half, mut write_half) = stream.into_split();
        let (internal_tx, mut internal_rx) = mpsc::channel::<TuyaError>(1);

        // Process initial command if exists
        if let Some(cmd) = initial_cmd {
            self.process_command(&mut write_half, seqno, cmd)
                .await
                .map_err(|e| {
                    if !self.is_stopped() {
                        error!(
                            "Initial command processing failed for {}: {}",
                            self.inner.id, e
                        );
                    }
                    e
                })?;
        }

        let device_clone = self.clone();
        let parent_cancel_token = self.inner.cancel_token.clone();

        // Reader Task
        let reader_task = crate::runtime::spawn(async move {
            let mut packets_received = 0;
            loop {
                tokio::select! {
                    () = parent_cancel_token.cancelled() => break,
                    res = timeout(SLEEP_INACTIVITY_TIMEOUT, read_half.read_u8()) => {
                        match res {
                            Ok(Ok(byte)) => {
                                if let Err(e) = device_clone.process_socket_data(&mut read_half, byte).await {
                                    let _ = internal_tx.send(e).await;
                                    break;
                                }
                                packets_received += 1;
                            }
                            Ok(Err(e)) => {
                                let err = if e.kind() == std::io::ErrorKind::UnexpectedEof {
                                    if packets_received > 0 {
                                        TuyaError::io(std::io::ErrorKind::ConnectionReset, "Connection reset")
                                    } else {
                                        TuyaError::KeyOrVersionError
                                    }
                                } else {
                                    TuyaError::from(e)
                                };
                                let _ = internal_tx.send(err).await;
                                break;
                            }
                            Err(_) => {
                                if !device_clone.is_stopped() {
                                    warn!("Inactivity timeout for {}", device_clone.inner.id);
                                }
                                let _ = internal_tx.send(TuyaError::Timeout).await;
                                break;
                            }
                        }
                    }
                }
            }
            debug!("Reader task for {} stopped", device_clone.inner.id);
        });

        let result = async {
            loop {
                tokio::select! {
                    () = self.inner.cancel_token.cancelled() => {
                        return Ok(());
                    }
                    () = self.inner.close_notify.notified() => {
                        // close() was called: tear down the current connection
                        // (the outer loop will reconnect on the next user
                        // request unless persist=false).
                        debug!("close_notify fired for {}, dropping connection", self.inner.id);
                        return Ok(());
                    }
                    cmd_opt = rx.recv() => {
                        if let Some(cmd) = cmd_opt {
                            self.process_command(&mut write_half, seqno, cmd).await?;
                        } else {
                            self.inner.state.write().state = ConnectionState::Stopped;
                            return Ok(());
                        }
                    }
                    _ = heartbeat_interval.tick() => {
                        if self.with_state(|s| s.persist) {
                            self.process_heartbeat(&mut write_half, seqno)
                                .await
                                .map_err(|e| {
                                    error!("Heartbeat failed for {}: {}", self.inner.id, e);
                                    e
                                })?;
                        }
                    }
                    err_opt = internal_rx.recv() => {
                        if let Some(e) = err_opt {
                            error!("Connection closed due to reader task error for {}: {}", self.inner.id, e);
                            return Err(e);
                        }
                    }
                }
            }
        }.await;

        reader_task.abort();
        result
    }

    async fn try_connect_with_backoff(
        &self,
        rx: &mut mpsc::Receiver<DeviceCommand>,
        seqno: &mut u32,
    ) -> Option<(TcpStream, Option<DeviceCommand>)> {
        loop {
            if self.is_stopped() {
                self.drain_rx(rx, TuyaError::Offline, true);
                return None;
            }

            // Reset seqno for new connection
            *seqno = 1;

            // Wait before retry if failed
            let backoff = self.with_state(|s| {
                if s.failure_count > 0 {
                    Some(self.get_backoff_duration(s.failure_count - 1))
                } else {
                    None
                }
            });

            if let Some(b) = backoff {
                warn!(
                    "Waiting {}s before next connection attempt for {}",
                    b.as_secs(),
                    self.inner.id
                );
                self.wait_for_backoff(rx, b).await?;
            }

            let result = timeout(self.timeout() * 2, self.connect_and_handshake(seqno)).await;
            if let Ok(Ok(s)) = result {
                self.with_state_mut(|s| {
                    s.state = ConnectionState::Connected;
                    s.scanner_bypass_failures = 0;
                    s.last_scanner_bypass_at = None;
                });
                info!(
                    "Connected to device {} ({})",
                    self.inner.id,
                    self.with_state(|s| s.real_ip.clone())
                );
                self.broadcast_error(ERR_SUCCESS, None);
                return Some((s, None));
            } else {
                let e = match result {
                    Ok(Err(e)) => e,
                    _ => TuyaError::Offline,
                };

                self.handle_connection_error(&e).await;
                self.drain_rx(rx, e.clone(), false);

                if !self.with_state(|s| s.persist) {
                    // persist=false has its own backoff state so that rapid user
                    // requests against an unreachable device don't translate into
                    // a TCP SYN storm. failure_count is bumped on every failed
                    // retry; ConnectNow still bypasses the wait.
                    self.with_state_mut(|s| {
                        s.failure_count = s.failure_count.saturating_add(1);
                    });
                    warn!(
                        "Connection failed (persist: false) for {}: {}. Waiting for next command.",
                        self.inner.id, e
                    );

                    loop {
                        match rx.recv().await {
                            Some(DeviceCommand::ConnectNow) => break,
                            Some(cmd @ DeviceCommand::Request { .. }) => {
                                // Space user-triggered retries with the same
                                // exponential+jitter backoff used by the
                                // persist=true path. Cancellation aborts the
                                // sleep early; Disconnect during sleep is not
                                // observed here (we re-enter rx.recv() next
                                // iteration anyway).
                                let backoff = self.with_state(|s| {
                                    self.get_backoff_duration(s.failure_count.saturating_sub(1))
                                });
                                if !backoff.is_zero() {
                                    debug!(
                                        "persist=false: spacing {:?} before retry for {}",
                                        backoff, self.inner.id
                                    );
                                    tokio::select! {
                                        () = tokio::time::sleep(backoff) => {}
                                        () = self.inner.cancel_token.cancelled() => {
                                            cmd.respond(Err(TuyaError::Offline));
                                            return None;
                                        }
                                    }
                                }

                                let retry_result =
                                    timeout(self.timeout() * 2, self.connect_and_handshake(seqno))
                                        .await;

                                if let Ok(Ok(s)) = retry_result {
                                    self.with_state_mut(|s| {
                                        s.state = ConnectionState::Connected;
                                        s.scanner_bypass_failures = 0;
                                        s.last_scanner_bypass_at = None;
                                    });
                                    info!("Connected to {} on demand", self.inner.id);
                                    self.broadcast_error(ERR_SUCCESS, None);
                                    return Some((s, Some(cmd)));
                                } else {
                                    let err = match retry_result {
                                        Ok(Err(e)) => e,
                                        _ => TuyaError::Offline,
                                    };
                                    self.handle_connection_error(&err).await;
                                    self.with_state_mut(|s| {
                                        s.failure_count = s.failure_count.saturating_add(1);
                                    });
                                    cmd.respond(Err(err.clone()));
                                    self.broadcast_error(ERR_OFFLINE, None);
                                }
                            }
                            None => return None,
                        }
                    }
                    continue;
                }

                self.with_state_mut(|s| {
                    s.failure_count += 1;
                    s.success_count = 0;
                    if s.config_address == ADDR_AUTO {
                        match e {
                            TuyaError::KeyOrVersionError | TuyaError::Offline => {
                                s.force_discovery = true;
                                let _ = get_scanner().invalidate_cache(&self.inner.id);
                            }
                            _ => {}
                        }
                    }
                });
            }
        }
    }

    async fn wait_for_backoff(
        &self,
        rx: &mut mpsc::Receiver<DeviceCommand>,
        backoff: Duration,
    ) -> Option<()> {
        let sleep_fut = sleep(backoff);
        tokio::pin!(sleep_fut);

        // Subscribe ONCE: `watch::Receiver` retains the "last seen version",
        // so any publish that arrives between awaits is replayed on the next
        // `changed()` instead of being lost (the previous `Notify::notified()`
        // re-subscription pattern had a documented gap here).
        let mut discovery_rx = get_scanner().subscribe_discoveries();

        loop {
            tokio::select! {
                () = &mut sleep_fut => return Some(()),
                _ = discovery_rx.changed() => {
                    if let Some(res) = get_scanner().get_cached_result(&self.inner.id)
                        && res.discovered_at.elapsed() < Duration::from_secs(10)
                    {
                        let (current_ip, config_addr, last_reported, last_bypass_at, bypass_failures) =
                            self.with_state(|s| {
                                (
                                    s.real_ip.clone(),
                                    s.config_address.clone(),
                                    s.last_reported_discovered_ip.clone(),
                                    s.last_scanner_bypass_at,
                                    s.scanner_bypass_failures,
                                )
                            });
                        match decide_discovery_notify_action(
                            &current_ip,
                            &config_addr,
                            &res.ip,
                            last_reported.as_deref(),
                            last_bypass_at,
                            bypass_failures,
                            Instant::now(),
                        ) {
                            DiscoveryNotifyAction::BypassBackoff => {
                                debug!(
                                    "Bypassing backoff for {} on scanner notify ({} -> {})",
                                    self.inner.id, current_ip, res.ip
                                );
                                self.with_state_mut(|s| {
                                    s.last_scanner_bypass_at = Some(Instant::now());
                                    s.scanner_bypass_failures =
                                        s.scanner_bypass_failures.saturating_add(1);
                                });
                                return Some(());
                            }
                            DiscoveryNotifyAction::Report => {
                                warn!(
                                    "Device {} configured at {} but scanner discovered it at {}",
                                    self.inner.id, current_ip, res.ip
                                );
                                self.with_state_mut(|s| {
                                    s.last_reported_discovered_ip = Some(res.ip.clone());
                                });
                                self.broadcast_error(
                                    ERR_STATE,
                                    Some(serde_json::json!({
                                        "reason": "ip_mismatch",
                                        "configured": current_ip,
                                        "discovered": res.ip,
                                    })),
                                );
                            }
                            DiscoveryNotifyAction::Ignore => {}
                        }
                    }
                }
                () = self.inner.cancel_token.cancelled() => {
                    self.drain_rx(rx, TuyaError::Offline, true);
                    return None;
                }
                cmd_opt = rx.recv() => {
                    if let Some(cmd) = cmd_opt {
                        if let DeviceCommand::ConnectNow = cmd { return Some(()) }
                        debug!("Rejecting command during backoff for device {}", self.inner.id);
                        cmd.respond(Err(TuyaError::Offline));
                        self.broadcast_error(ERR_OFFLINE, None);
                    } else {
                        return None;
                    }
                }
            }
        }
    }

    fn handle_disconnect(&self, err: Option<TuyaError>) {
        self.with_state_mut(|s| {
            if s.state != ConnectionState::Stopped {
                s.state = ConnectionState::Disconnected;
            }
            s.session_key = None; // Clear session key on disconnect
        });

        if let Some(e) = err {
            if matches!(e, TuyaError::KeyOrVersionError) {
                warn!(
                    "Device {} possibly has key or version mismatch (Error 914)",
                    self.inner.id
                );
            } else if !self.is_stopped() {
                debug!(
                    "Connection lost for device {} due to error: {}",
                    self.inner.id, e
                );
            }

            if !self.is_stopped() {
                self.broadcast_error(e.code(), None);
            }
        } else if !self.is_stopped() {
            debug!("Connection closed normally for device {}", self.inner.id);
            self.broadcast_error(ERR_OFFLINE, None);
        }
    }

    async fn handle_connection_error(&self, e: &TuyaError) {
        self.with_state_mut(|s| {
            if s.state != ConnectionState::Stopped {
                s.state = ConnectionState::Disconnected;
            }
        });
        self.broadcast_error(e.code(), Some(serde_json::json!(format!("{e}"))));
    }

    fn drain_rx(&self, rx: &mut mpsc::Receiver<DeviceCommand>, err: TuyaError, close: bool) {
        if close {
            rx.close();
        }
        while let Ok(cmd) = rx.try_recv() {
            cmd.respond(Err(err.clone()));
        }
    }
}

impl Device {
    // -------------------------------------------------------------------------
    // Protocol Implementation & Handshake
    // -------------------------------------------------------------------------

    async fn connect_and_handshake(&self, seqno: &mut u32) -> Result<TcpStream> {
        let addr = self.resolve_address().await?;
        let port = self.with_state(|s| s.port);

        info!(
            "Connecting to device {} at {}:{}",
            self.inner.id, addr, port
        );
        let mut stream = timeout(self.timeout(), TcpStream::connect(format!("{addr}:{port}")))
            .await
            .map_err(|_| TuyaError::Timeout)?
            .map_err(|e| match e.kind() {
                std::io::ErrorKind::ConnectionRefused => TuyaError::ConnectionFailed,
                _ => TuyaError::from(e),
            })?;

        let protocol = get_protocol(self.version(), self.dev_type());
        if protocol.requires_session_key()
            && !self.negotiate_session_key(&mut stream, seqno).await?
        {
            return Err(TuyaError::KeyOrVersionError);
        }

        Ok(stream)
    }

    async fn negotiate_session_key(&self, stream: &mut TcpStream, seqno: &mut u32) -> Result<bool> {
        let protocol = get_protocol(self.version(), self.dev_type());
        debug!("Session negotiation (v{})", protocol.version());

        // 1. Send SessKeyNegStart
        let local_nonce = protocol.prepare_session_key_negotiation();
        self.send_raw_to_stream(
            stream,
            self.build_message(
                seqno,
                CommandType::SessKeyNegStart as u32,
                local_nonce.clone(),
            ),
        )
        .await?;

        // 2. Read response and verify
        let first_byte = timeout(self.timeout(), stream.read_u8())
            .await
            .map_err(|_| TuyaError::Timeout)?
            .map_err(|e| {
                if e.kind() == std::io::ErrorKind::UnexpectedEof {
                    TuyaError::KeyOrVersionError
                } else {
                    TuyaError::from(e)
                }
            })?;

        let resp = self
            .read_and_parse_from_stream(stream, first_byte)
            .await?
            .ok_or(TuyaError::HandshakeFailed)?;

        if resp.cmd != CommandType::SessKeyNegResp as u32 {
            return Err(TuyaError::KeyOrVersionError);
        }

        let remote_nonce = protocol.verify_session_key_response(
            &local_nonce,
            &resp.payload,
            &self.inner.local_key,
        )?;

        // 3. Finalize and send SessKeyNegFinish
        let (session_key, finish_hmac) =
            protocol.finalize_session_key(&local_nonce, &remote_nonce, &self.inner.local_key)?;

        self.send_raw_to_stream(
            stream,
            self.build_message(seqno, CommandType::SessKeyNegFinish as u32, finish_hmac),
        )
        .await?;

        // 4. Encrypt and store session key
        let cipher = TuyaCipher::new(&self.inner.local_key)?;
        let encrypted_key = protocol.encrypt_session_key(&session_key, &cipher, &local_nonce)?;

        self.with_state_mut(|s| s.session_key = Some(encrypted_key));
        Ok(true)
    }

    async fn resolve_address(&self) -> Result<String> {
        let (config_addr, force_discovery, version) =
            self.with_state(|s| (s.config_address.clone(), s.force_discovery, s.version));

        let ip_explicit =
            config_addr != ADDR_AUTO && config_addr != "0.0.0.0" && !config_addr.is_empty();
        let ver_explicit = version != Version::Auto;

        if ip_explicit && ver_explicit && !force_discovery {
            return Ok(config_addr);
        }

        if let Ok(Some(result)) = get_scanner()
            .discover_device_internal(
                &self.inner.id,
                force_discovery,
                Some(&self.inner.cancel_token),
            )
            .await
        {
            let mut state = self.inner.state.write();
            if let Some(v) = result.version
                && state.version == Version::Auto
            {
                state.version = v;
            }

            let target_ip = if ip_explicit { config_addr } else { result.ip };
            state.real_ip = target_ip.clone();
            state.force_discovery = false;
            Ok(target_ip)
        } else if ip_explicit {
            self.with_state_mut(|s| {
                s.real_ip = config_addr.clone();
                s.force_discovery = false;
            });
            Ok(config_addr)
        } else {
            Err(TuyaError::Offline)
        }
    }

    async fn generate_payload(
        &self,
        command: CommandType,
        data: Option<Value>,
        cid: Option<&str>,
    ) -> Result<(u32, Value)> {
        let (version, mut dev_type) = self.with_state(|s| (s.version, s.dev_type));
        // If dev_type is Auto, treat it as Default for protocol selection
        if dev_type == DeviceType::Auto {
            dev_type = DeviceType::Default;
        }
        let protocol = get_protocol(version, dev_type);
        let t = self.get_timestamp();
        protocol.generate_payload(&self.inner.id, command, data, cid, t)
    }

    async fn process_command<W: AsyncWriteExt + Unpin>(
        &self,
        stream: &mut W,
        seqno: &mut u32,
        cmd: DeviceCommand,
    ) -> Result<()> {
        match cmd {
            DeviceCommand::Request {
                command,
                data,
                cid,
                resp_tx,
            } => {
                let nowait = self.inner.nowait.load(Ordering::Relaxed);
                let cmd_code = command as u32;
                let response_rx = if !nowait && !NO_RESPONSE_CMDS.contains(&cmd_code) {
                    Some(self.inner.broadcast_tx.subscribe())
                } else {
                    None
                };

                let res = self
                    .generate_payload(command, data.clone(), cid.as_deref())
                    .await;
                let send_res = match res {
                    Ok((cmd_id, payload)) => {
                        debug!("Sending command: cmd=0x{:02X}, seqno={}", cmd_id, *seqno);
                        self.send_json_msg(stream, seqno, cmd_id, &payload).await
                    }
                    Err(e) => Err(e),
                };

                if let Err(e) = send_res {
                    let _ = resp_tx.send(Err(e));
                    return Ok(());
                }

                if let Some(mut rx) = response_rx {
                    let protocol = self.with_state(|s| get_protocol(s.version, s.dev_type));
                    let effective_cmd = protocol.get_effective_command(command);
                    let timeout_dur = self.timeout();

                    let id_for_logs = self.inner.id.clone();
                    let target_cid = cid.clone();
                    // NOTE: Tuya's LAN protocol is fundamentally asynchronous —
                    // the device treats inbound commands as fire-and-forget and
                    // independently emits unsolicited Status pushes whenever its
                    // DPs change. There is no request/response correlation token
                    // on the wire: many firmwares either echo seqno=0 or use a
                    // device-side counter unrelated to the request, so matching
                    // by request seqno is structurally impossible (not a bug to
                    // be fixed — it is the shape of the protocol).
                    //
                    // Consequently, an unsolicited Status push that arrives
                    // between subscribe() and the next user request may be
                    // surfaced as that request's response. This is by design;
                    // the "right" pattern for callers who care about strict
                    // correlation is to use the listener() stream for pushes
                    // and serialize their own per-Device request/response calls.
                    //
                    // DO NOT add seqno matching here in future reviews.
                    let wait_res = timeout(timeout_dur, async {
                        loop {
                            match rx.recv().await {
                                Ok(msg) => match match_response(
                                    &msg,
                                    effective_cmd,
                                    target_cid.as_deref(),
                                ) {
                                    MatchOutcome::Accept => return Ok(Some(msg)),
                                    MatchOutcome::Continue => continue,
                                },
                                Err(tokio::sync::broadcast::error::RecvError::Lagged(
                                    skipped,
                                )) => {
                                    warn!(
                                        "Response wait for device {} lagged, skipped {} broadcast messages",
                                        id_for_logs, skipped
                                    );
                                    continue;
                                }
                                Err(tokio::sync::broadcast::error::RecvError::Closed) => {
                                    return Err(TuyaError::Offline);
                                }
                            }
                        }
                    })
                    .await;

                    let wait_res = match wait_res {
                        Ok(inner) => inner,
                        Err(_) => Err(TuyaError::Timeout),
                    };

                    let _ = resp_tx.send(wait_res);
                } else {
                    let _ = resp_tx.send(Ok(None));
                }
            }
            DeviceCommand::ConnectNow => {
                debug!(
                    "Device {} is already connected, ignoring ConnectNow",
                    self.inner.id
                );
            }
        }
        Ok(())
    }

    async fn process_socket_data<R: AsyncReadExt + Unpin>(
        &self,
        stream: &mut R,
        first_byte: u8,
    ) -> Result<()> {
        if let Some(msg) = self.read_and_parse_from_stream(stream, first_byte).await? {
            self.update_last_received();
            self.reset_failure_count();
            debug!(
                "Received message: cmd=0x{:02X}, payload_len={}",
                msg.cmd,
                msg.payload.len()
            );
            if msg.payload.is_empty() {
                debug!(
                    "Received empty payload message (cmd 0x{:02X}), broadcasting as ACK",
                    msg.cmd
                );
                let _ = self.inner.broadcast_tx.send(msg);
            } else {
                // Check if payload is valid JSON
                if serde_json::from_slice::<Value>(&msg.payload).is_err() {
                    debug!("Non-JSON payload detected, broadcasting as ERR_JSON");
                    let payload_hex = hex_encode(&msg.payload);
                    self.broadcast_error(
                        ERR_JSON,
                        Some(serde_json::json!({
                            keys::PAYLOAD_RAW: payload_hex,
                            "cmd": msg.cmd
                        })),
                    );
                } else {
                    let _ = self.inner.broadcast_tx.send(msg);
                }
            }
        }
        Ok(())
    }

    async fn process_heartbeat<W: AsyncWriteExt + Unpin>(
        &self,
        stream: &mut W,
        seqno: &mut u32,
    ) -> Result<()> {
        let last = self.with_state(|s| s.last_sent);

        if last.elapsed() >= SLEEP_HEARTBEAT_DEFAULT {
            debug!("Auto-heartbeat for device {}", self.inner.id);
            let (cmd, payload) = self
                .generate_payload(CommandType::HeartBeat, None, None)
                .await?;
            self.send_json_msg(stream, seqno, cmd, &payload).await?;
        }
        Ok(())
    }
}

impl Device {
    // -------------------------------------------------------------------------
    // Low-level Message Framing & Encryption
    // -------------------------------------------------------------------------

    fn build_message<P: Into<Vec<u8>>>(
        &self,
        seqno: &mut u32,
        cmd: u32,
        payload: P,
    ) -> TuyaMessage {
        let payload = payload.into();
        let current_seq = *seqno;
        // `wrapping_add` so a long-lived connection that hits the u32 boundary
        // doesn't panic in debug builds. The seqno is informational on the
        // wire (Tuya doesn't reliably match request/response seqnos — see the
        // note in process_command) so wrapping is benign.
        *seqno = seqno.wrapping_add(1);
        debug!(
            "Building message: cmd=0x{:02X}, seqno={}, payload_len={}",
            cmd,
            current_seq,
            payload.len()
        );

        let protocol = get_protocol(self.version(), self.dev_type());

        TuyaMessage {
            seqno: current_seq,
            cmd,
            payload,
            prefix: protocol.get_prefix(),
            ..Default::default()
        }
    }

    fn pack_msg(&self, mut msg: TuyaMessage) -> Result<Vec<u8>> {
        let (version, dev_type) = self.with_state(|s| (s.version, s.dev_type));
        let cipher = self.get_cipher()?;
        let protocol = get_protocol(version, dev_type);

        msg.payload = protocol.pack_payload(&msg.payload, msg.cmd, &cipher)?;
        msg.prefix = protocol.get_prefix();

        let hmac_key = protocol.get_hmac_key(cipher.key());
        pack_message(&msg, hmac_key)
    }

    async fn send_json_msg<W: AsyncWriteExt + Unpin>(
        &self,
        stream: &mut W,
        seqno: &mut u32,
        cmd: u32,
        payload: &Value,
    ) -> Result<()> {
        let payload_bytes = serde_json::to_vec(payload)?;
        let msg = self.build_message(seqno, cmd, payload_bytes);
        self.send_raw_to_stream(stream, msg).await
    }

    async fn send_raw_to_stream<W: AsyncWriteExt + Unpin>(
        &self,
        stream: &mut W,
        msg: TuyaMessage,
    ) -> Result<()> {
        let packed = self.pack_msg(msg)?;
        timeout(self.timeout(), stream.write_all(&packed))
            .await
            .map_err(|_| TuyaError::Timeout)?
            .map_err(TuyaError::from)?;

        self.update_last_sent();
        Ok(())
    }

    async fn read_and_parse_from_stream<R: AsyncReadExt + Unpin>(
        &self,
        stream: &mut R,
        first_byte: u8,
    ) -> Result<Option<TuyaMessage>> {
        let prefix = self.scan_for_prefix(stream, first_byte).await?;

        // Read remaining 12 bytes of header (16 bytes total)
        let mut header_buf = [0u8; 16];
        header_buf[0..4].copy_from_slice(&prefix);
        timeout(self.timeout(), stream.read_exact(&mut header_buf[4..]))
            .await
            .map_err(|_| TuyaError::Timeout)?
            .map_err(TuyaError::from)?;

        // Parse and read body
        let dev_type_before = self.dev_type();
        match self.parse_and_read_body(stream, header_buf).await {
            Ok(Some(msg)) => {
                if dev_type_before != DeviceType::Device22
                    && self.dev_type() == DeviceType::Device22
                {
                    debug!("Device22 transition detected, reporting with original payload");
                    let original_payload = if msg.payload.is_empty() {
                        Value::Null
                    } else {
                        serde_json::from_slice(&msg.payload).unwrap_or_else(
                            |_| serde_json::json!({ keys::PAYLOAD_RAW: hex_encode(&msg.payload) }),
                        )
                    };
                    return Ok(Some(self.error_helper(ERR_DEVTYPE, Some(original_payload))));
                }
                Ok(Some(msg))
            }
            Ok(None) => Ok(None),
            Err(e) => {
                if matches!(e, TuyaError::Io { .. }) {
                    return Err(e);
                }
                warn!("Error parsing message from {}: {}", self.inner.id, e);
                Ok(Some(self.error_helper(
                    ERR_PAYLOAD,
                    Some(serde_json::json!(format!("{e}"))),
                )))
            }
        }
    }

    async fn scan_for_prefix<R: AsyncReadExt + Unpin>(
        &self,
        stream: &mut R,
        first_byte: u8,
    ) -> Result<[u8; 4]> {
        let mut current_prefix = first_byte as u32;

        for _ in 0..3 {
            let next_byte = timeout(self.timeout(), stream.read_u8())
                .await
                .map_err(|_| TuyaError::Timeout)?
                .map_err(TuyaError::from)?;
            current_prefix = (current_prefix << 8) | (next_byte as u32);
        }

        for _ in 0..1024 {
            if current_prefix == PREFIX_55AA || current_prefix == PREFIX_6699 {
                return Ok(current_prefix.to_be_bytes());
            }

            let next_byte = timeout(self.timeout(), stream.read_u8())
                .await
                .map_err(|_| TuyaError::Timeout)?
                .map_err(TuyaError::from)?;
            current_prefix = (current_prefix << 8) | (next_byte as u32);
        }

        // 1024 bytes without a recognizable frame prefix means the stream is
        // structurally out of sync — almost always wrong key/version, a
        // protocol mismatch, or a corrupted device. Returning `Ok(None)` would
        // leave the reader silently spinning forever; surface a hard error so
        // the actor tears down the connection and reconnects.
        warn!(
            "scan_for_prefix on device {} found no valid frame after 1024 bytes; \
             treating as protocol/key mismatch and forcing reconnect",
            self.inner.id
        );
        Err(TuyaError::KeyOrVersionError)
    }

    async fn parse_and_read_body<R: AsyncReadExt + Unpin>(
        &self,
        stream: &mut R,
        header_buf: [u8; 16],
    ) -> Result<Option<TuyaMessage>> {
        let (packet, header) = self.read_full_packet(stream, header_buf).await?;
        trace!("Received packet (hex): {:?}", hex_encode(&packet));

        let mut decoded = self.unpack_and_check_dev22(&packet, header).await?;

        if !decoded.payload.is_empty() {
            trace!("Raw payload (hex): {:?}", hex_encode(&decoded.payload));
            decoded.payload = self.decrypt_and_clean_payload(decoded.payload).await?;
        }

        Ok(Some(decoded))
    }

    async fn read_full_packet<R: AsyncReadExt + Unpin>(
        &self,
        stream: &mut R,
        header_buf: [u8; 16],
    ) -> Result<(Vec<u8>, TuyaHeader)> {
        let prefix =
            u32::from_be_bytes([header_buf[0], header_buf[1], header_buf[2], header_buf[3]]);

        let (header, mut packet) = if prefix == PREFIX_6699 {
            let mut extra = [0u8; 2];
            timeout(self.timeout(), stream.read_exact(&mut extra))
                .await
                .map_err(|_| TuyaError::Timeout)?
                .map_err(TuyaError::from)?;

            let mut fh = Vec::with_capacity(18);
            fh.extend_from_slice(&header_buf);
            fh.extend_from_slice(&extra);
            (parse_header(&fh)?, fh)
        } else {
            (parse_header(&header_buf)?, header_buf.to_vec())
        };

        let total_len = header.total_length as usize;
        let header_len = packet.len();

        packet.resize(total_len, 0);
        timeout(self.timeout(), stream.read_exact(&mut packet[header_len..]))
            .await
            .map_err(|_| TuyaError::Timeout)?
            .map_err(TuyaError::from)?;

        Ok((packet, header))
    }

    async fn unpack_and_check_dev22(
        &self,
        packet: &[u8],
        header: TuyaHeader,
    ) -> Result<TuyaMessage> {
        let (version, dev_type) = self.with_state(|s| (s.version, s.dev_type));
        let protocol = get_protocol(version, dev_type);
        let cipher = self.get_cipher()?;
        let hmac_key = protocol.get_hmac_key(cipher.key());

        unpack_message(packet, hmac_key, Some(header.clone()), Some(false)).or_else(|e| {
            // Only allow switching if dev_type is Auto and protocol allows it
            if protocol.should_check_dev22_fallback()
                && dev_type == DeviceType::Auto
                && let Ok(d) = unpack_message(packet, None, Some(header), Some(false))
            {
                info!("Device22 detected via CRC32 fallback. Switching mode.");
                self.set_dev_type(DeviceType::Device22);
                return Ok(d);
            }
            Err(e)
        })
    }

    async fn decrypt_and_clean_payload(&self, payload: Vec<u8>) -> Result<Vec<u8>> {
        let (version, mut dev_type) = self.with_state(|s| (s.version, s.dev_type));
        let original_dev_type = dev_type;
        if dev_type == DeviceType::Auto {
            dev_type = DeviceType::Default;
        }

        let cipher = self.get_cipher()?;
        let protocol = get_protocol(version, dev_type);

        let decrypted = protocol.decrypt_payload(payload, &cipher)?;

        if protocol.should_check_dev22_fallback()
            && original_dev_type == DeviceType::Auto
            && String::from_utf8_lossy(&decrypted).contains(DATA_UNVALID)
        {
            warn!("Device22 detected via '{DATA_UNVALID}' payload. Switching mode.");
            self.set_dev_type(DeviceType::Device22);
        }

        Ok(decrypted)
    }

    fn get_cipher(&self) -> Result<Arc<TuyaCipher>> {
        // Fast path: cache hit under a read lock so concurrent pack/unpack
        // calls don't serialize on a write lock. The cipher is keyed by the
        // session_key (if any) or local_key; cache invalidation happens by
        // updating `state.session_key` elsewhere and falling through to the
        // slow path below.
        {
            let state = self.inner.state.read();
            let key: &[u8] = state
                .session_key
                .as_deref()
                .unwrap_or(&self.inner.local_key);
            if let Some(ref cipher) = state.cipher
                && cipher.key() == key
            {
                return Ok(Arc::clone(cipher));
            }
        }

        // Slow path: build the cipher *outside* the write lock so we don't
        // hold an exclusive lock across a fallible operation. Then re-check
        // under the write lock in case another caller installed a matching
        // cipher between the read and write — if so, drop ours.
        // (Building a TuyaCipher from a 16-byte key is cheap and pure; the
        // duplicate work in a race is negligible compared to lock contention.)
        let key_owned: Vec<u8> = {
            let state = self.inner.state.read();
            state
                .session_key
                .clone()
                .unwrap_or_else(|| self.inner.local_key.clone())
        };
        let new_cipher = Arc::new(TuyaCipher::new(&key_owned)?);

        let mut state = self.inner.state.write();
        if let Some(ref cipher) = state.cipher
            && cipher.key() == key_owned.as_slice()
        {
            return Ok(Arc::clone(cipher));
        }
        state.cipher = Some(Arc::clone(&new_cipher));
        Ok(new_cipher)
    }

    fn get_backoff_duration(&self, failure_count: u32) -> Duration {
        let min_secs = SLEEP_RECONNECT_MIN.as_secs();
        let max_secs = SLEEP_RECONNECT_MAX.as_secs();
        // Base exponential backoff: 2^n * min_secs
        let base_secs = (2u64.pow(failure_count.min(10)) * min_secs).min(max_secs);

        if base_secs == 0 {
            return Duration::from_secs(0);
        }

        let base_ms = base_secs * 1000;
        let fixed_ms = (base_ms * 70) / 100; // 70% fixed
        let random_range_ms = base_ms - fixed_ms; // 30% random range

        // Apply Jitter: 70% fixed + random(0% to 30%)
        let mut rng = rand::rng();
        let jitter_ms = fixed_ms + (rng.next_u64() % random_range_ms.max(1));

        Duration::from_millis(jitter_ms)
    }

    fn error_helper(&self, code: u32, payload: Option<Value>) -> TuyaMessage {
        let mut response = serde_json::json!({
            keys::ERR_MSG: get_error_message(code),
            keys::ERR_CODE: code,
        });

        if let Some(p) = payload {
            match p {
                Value::String(s) => response[keys::PAYLOAD_STR] = Value::String(s),
                Value::Object(mut obj) => {
                    if let Some(raw) = obj
                        .remove(keys::IN_DATA)
                        .or_else(|| obj.remove(keys::IN_PAYLOAD))
                        .or_else(|| obj.remove(keys::PAYLOAD_RAW))
                    {
                        response[keys::PAYLOAD_RAW] = raw;
                    }
                    if let Some(res_obj) = response.as_object_mut() {
                        res_obj.extend(obj);
                    }
                }
                _ => response[keys::ERR_PAYLOAD_OBJ] = p,
            }
        }

        // Serializing a Value built from `serde_json::json!` cannot fail in
        // practice; if it ever does, fall back to an empty payload so we
        // still deliver an ACK rather than panicking inside the broadcast.
        let payload = serde_json::to_vec(&response).unwrap_or_else(|e| {
            warn!("error_helper: failed to serialize error payload: {e}");
            Vec::new()
        });
        TuyaMessage {
            payload,
            prefix: get_protocol(self.version(), self.dev_type()).get_prefix(),
            ..Default::default()
        }
    }
}

/// Verdict from `match_response` — either accept the broadcast message as
/// the current request's response, or skip it and keep waiting.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum MatchOutcome {
    Accept,
    Continue,
}

/// Decides whether a broadcast message is the response we're waiting for.
///
/// Matching is by command id + optional CID; the Tuya LAN protocol does not
/// carry a reliable request seqno (see the module-level note in
/// `process_command`), so this is the strongest correlation available.
///
/// Pure / synchronous on purpose: keeps the actor's `select!` body small
/// and lets us exhaustively unit-test the decision table.
fn match_response(msg: &TuyaMessage, effective_cmd: u32, target_cid: Option<&str>) -> MatchOutcome {
    // Device-reported error responses arrive as cmd=0; surface them as the
    // current request's outcome regardless of CID.
    if msg.cmd == 0 {
        debug!("Device returned error response (cmd 0), accepting");
        return MatchOutcome::Accept;
    }

    let cmd_matches = msg.cmd == effective_cmd || msg.cmd == CommandType::Status as u32;
    if !cmd_matches {
        return MatchOutcome::Continue;
    }

    let needs_data = MANDATORY_DATA_CMDS.contains(&msg.cmd);

    if let Some(target_cid) = target_cid {
        // Sub-device request: response must carry the same CID.
        if msg.payload.is_empty() {
            if needs_data {
                trace!(
                    "Received empty ACK for CID command requiring data (0x{:02X}), continuing wait",
                    msg.cmd
                );
                return MatchOutcome::Continue;
            }
            debug!("Received empty ACK for CID request ({target_cid}), accepting");
            return MatchOutcome::Accept;
        }
        if let Ok(val) = serde_json::from_slice::<Value>(&msg.payload) {
            let resp_cid = val.get("cid").and_then(|c| c.as_str());
            if resp_cid == Some(target_cid) {
                debug!("Received matching response for CID: {target_cid}");
                return MatchOutcome::Accept;
            }
            trace!("Ignoring response for CID: {resp_cid:?} (expected {target_cid})");
            return MatchOutcome::Continue;
        }
        // Non-JSON payload on a CID request — accept rather than spin.
        return MatchOutcome::Accept;
    }

    // Parent-device request: response must NOT carry a CID.
    if msg.payload.is_empty() {
        if needs_data {
            trace!(
                "Received empty ACK for parent command requiring data (0x{:02X}), continuing wait",
                msg.cmd
            );
            return MatchOutcome::Continue;
        }
        return MatchOutcome::Accept;
    }
    if let Ok(val) = serde_json::from_slice::<Value>(&msg.payload) {
        if val.get("cid").is_none() {
            return MatchOutcome::Accept;
        }
        trace!("Ignoring response with CID for parent request");
        return MatchOutcome::Continue;
    }
    // Non-JSON payload on a parent request — accept rather than spin.
    MatchOutcome::Accept
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum DiscoveryNotifyAction {
    /// Already-reported duplicate, cooldown not yet elapsed, or otherwise no-op.
    Ignore,
    /// Explicit-IP mismatch; emit `ERR_STATE` so the caller can decide.
    Report,
    /// Skip the remaining backoff and retry now. Fired either because the
    /// discovered IP changed (Auto mode) or because the device just proved
    /// it's reachable (same-IP case, throttled by the bypass cooldown).
    BypassBackoff,
}

/// Cooldown between successive scanner-triggered bypass attempts that target
/// the same IP we already know about.
///
/// Doubles on every failed bypass so a persistently broken setup (e.g. key
/// mismatch) self-throttles down to the regular reminder cadence; capped at
/// the main reconnect ceiling so the bypass cadence never outpaces backoff.
fn scanner_bypass_cooldown(failures: u32) -> Duration {
    let base = SCANNER_BYPASS_BASE_COOLDOWN.as_secs();
    let cap = SLEEP_RECONNECT_MAX.as_secs();
    let secs = base.saturating_mul(1u64 << failures.min(10)).min(cap);
    Duration::from_secs(secs)
}

fn decide_discovery_notify_action(
    current_ip: &str,
    config_addr: &str,
    discovered_ip: &str,
    last_reported_ip: Option<&str>,
    last_bypass_at: Option<Instant>,
    bypass_failures: u32,
    now: Instant,
) -> DiscoveryNotifyAction {
    let ip_match = !current_ip.is_empty() && current_ip == discovered_ip;
    let ip_explicit =
        config_addr != ADDR_AUTO && config_addr != "0.0.0.0" && !config_addr.is_empty();

    match (ip_explicit, ip_match) {
        // Auto + fresh IP: always bypass — the new IP itself is the news.
        (false, false) => DiscoveryNotifyAction::BypassBackoff,
        // Explicit + different IP: scanner news can't be acted on (resolve_address
        // would return the configured IP anyway), so elevate to the listener.
        (true, false) => {
            if last_reported_ip == Some(discovered_ip) {
                DiscoveryNotifyAction::Ignore
            } else {
                DiscoveryNotifyAction::Report
            }
        }
        // Same IP either way: device is alive — bypass if the cooldown allows.
        (_, true) => {
            let cooldown = scanner_bypass_cooldown(bypass_failures);
            let allowed = last_bypass_at.is_none_or(|t| now.duration_since(t) >= cooldown);
            if allowed {
                DiscoveryNotifyAction::BypassBackoff
            } else {
                DiscoveryNotifyAction::Ignore
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::{
        ADDR_AUTO, DiscoveryNotifyAction, MatchOutcome, SCANNER_BYPASS_BASE_COOLDOWN,
        SLEEP_RECONNECT_MAX, decide_discovery_notify_action, match_response,
        scanner_bypass_cooldown,
    };
    use crate::protocol::{CommandType, PREFIX_55AA, TuyaMessage};
    use std::time::{Duration, Instant};

    fn make_msg(cmd: u32, payload: &[u8]) -> TuyaMessage {
        TuyaMessage {
            seqno: 0,
            cmd,
            retcode: None,
            payload: payload.to_vec(),
            prefix: PREFIX_55AA,
            iv: None,
        }
    }

    #[test]
    fn match_cmd_zero_is_always_accept() {
        let m = make_msg(0, b"{}");
        assert_eq!(match_response(&m, 0x0d, None), MatchOutcome::Accept);
        assert_eq!(
            match_response(&m, 0x0d, Some("cid_a")),
            MatchOutcome::Accept
        );
    }

    #[test]
    fn match_wrong_cmd_continues() {
        let m = make_msg(0x09, b"{}"); // HeartBeat, not what we sent
        assert_eq!(
            match_response(&m, CommandType::DpQuery as u32, None),
            MatchOutcome::Continue
        );
    }

    #[test]
    fn match_status_cmd_is_accepted_as_response() {
        // Devices commonly answer with a Status push regardless of what was
        // sent — match_response accepts that as the response.
        let m = make_msg(CommandType::Status as u32, b"{\"dps\":{\"1\":true}}");
        assert_eq!(
            match_response(&m, CommandType::DpQueryNew as u32, None),
            MatchOutcome::Accept
        );
    }

    #[test]
    fn match_parent_request_rejects_cid_response() {
        let m = make_msg(
            CommandType::DpQueryNew as u32,
            b"{\"cid\":\"sub_a\",\"dps\":{}}",
        );
        assert_eq!(
            match_response(&m, CommandType::DpQueryNew as u32, None),
            MatchOutcome::Continue
        );
    }

    #[test]
    fn match_cid_request_accepts_matching_cid() {
        let m = make_msg(
            CommandType::DpQueryNew as u32,
            b"{\"cid\":\"sub_a\",\"dps\":{}}",
        );
        assert_eq!(
            match_response(&m, CommandType::DpQueryNew as u32, Some("sub_a")),
            MatchOutcome::Accept
        );
    }

    #[test]
    fn match_cid_request_rejects_other_cid() {
        let m = make_msg(
            CommandType::DpQueryNew as u32,
            b"{\"cid\":\"sub_b\",\"dps\":{}}",
        );
        assert_eq!(
            match_response(&m, CommandType::DpQueryNew as u32, Some("sub_a")),
            MatchOutcome::Continue
        );
    }

    #[test]
    fn match_cid_request_accepts_empty_ack() {
        // Many gateways send an empty ack for CID-targeted writes — accept
        // it rather than wait for data that may never come.
        let m = make_msg(CommandType::ControlNew as u32, b"");
        assert_eq!(
            match_response(&m, CommandType::ControlNew as u32, Some("sub_a")),
            MatchOutcome::Accept
        );
    }

    #[test]
    fn match_mandatory_data_cmd_rejects_empty_ack() {
        // LanExtStream is in MANDATORY_DATA_CMDS — empty payload must NOT be
        // taken as the response.
        let m = make_msg(CommandType::LanExtStream as u32, b"");
        assert_eq!(
            match_response(&m, CommandType::LanExtStream as u32, None),
            MatchOutcome::Continue
        );
    }

    /// Wrapper that fixes the cooldown-related inputs at "first call ever"
    /// (no prior bypass, no failures) so the legacy four-arg call sites stay
    /// readable.
    fn decide_fresh(
        current_ip: &str,
        config_addr: &str,
        discovered_ip: &str,
        last_reported_ip: Option<&str>,
    ) -> DiscoveryNotifyAction {
        decide_discovery_notify_action(
            current_ip,
            config_addr,
            discovered_ip,
            last_reported_ip,
            None,
            0,
            Instant::now(),
        )
    }

    #[test]
    fn auto_mode_with_different_ip_bypasses_backoff() {
        assert_eq!(
            decide_fresh("10.0.0.50", ADDR_AUTO, "10.0.0.73", None),
            DiscoveryNotifyAction::BypassBackoff
        );
    }

    #[test]
    fn auto_mode_with_empty_real_ip_bypasses_backoff() {
        assert_eq!(
            decide_fresh("", ADDR_AUTO, "10.0.0.73", None),
            DiscoveryNotifyAction::BypassBackoff
        );
    }

    #[test]
    fn auto_mode_with_matching_ip_bypasses_when_no_prior_attempt() {
        // Same-IP scanner news is now treated as a "device alive" signal and
        // bypasses backoff — throttled by the cooldown, but unrestricted on
        // the first encounter.
        assert_eq!(
            decide_fresh("10.0.0.73", ADDR_AUTO, "10.0.0.73", None),
            DiscoveryNotifyAction::BypassBackoff
        );
    }

    #[test]
    fn explicit_ip_with_different_discovery_reports() {
        assert_eq!(
            decide_fresh("10.0.0.50", "10.0.0.50", "10.0.0.73", None),
            DiscoveryNotifyAction::Report
        );
    }

    #[test]
    fn explicit_ip_with_matching_discovery_bypasses_when_no_prior_attempt() {
        assert_eq!(
            decide_fresh("10.0.0.50", "10.0.0.50", "10.0.0.50", None),
            DiscoveryNotifyAction::BypassBackoff
        );
    }

    #[test]
    fn explicit_ip_does_not_repeat_report_for_same_discovered_ip() {
        assert_eq!(
            decide_fresh("10.0.0.50", "10.0.0.50", "10.0.0.73", Some("10.0.0.73")),
            DiscoveryNotifyAction::Ignore
        );
    }

    #[test]
    fn explicit_ip_reports_again_when_discovered_ip_changes() {
        assert_eq!(
            decide_fresh("10.0.0.50", "10.0.0.50", "10.0.0.95", Some("10.0.0.73")),
            DiscoveryNotifyAction::Report
        );
    }

    #[test]
    fn zero_zero_zero_zero_treated_as_auto() {
        assert_eq!(
            decide_fresh("10.0.0.50", "0.0.0.0", "10.0.0.73", None),
            DiscoveryNotifyAction::BypassBackoff
        );
    }

    #[test]
    fn empty_config_addr_treated_as_auto() {
        assert_eq!(
            decide_fresh("10.0.0.50", "", "10.0.0.73", None),
            DiscoveryNotifyAction::BypassBackoff
        );
    }

    // --- Scanner-bypass cooldown behavior ---

    #[test]
    fn same_ip_bypass_is_throttled_within_cooldown() {
        let now = Instant::now();
        let just_bypassed = now - Duration::from_secs(5);
        assert_eq!(
            decide_discovery_notify_action(
                "10.0.0.50",
                "10.0.0.50",
                "10.0.0.50",
                None,
                Some(just_bypassed),
                1,
                now,
            ),
            DiscoveryNotifyAction::Ignore
        );
    }

    #[test]
    fn same_ip_bypass_allowed_once_cooldown_elapses() {
        let now = Instant::now();
        let past = now - SCANNER_BYPASS_BASE_COOLDOWN * 3;
        assert_eq!(
            decide_discovery_notify_action(
                "10.0.0.50",
                "10.0.0.50",
                "10.0.0.50",
                None,
                Some(past),
                1,
                now,
            ),
            DiscoveryNotifyAction::BypassBackoff
        );
    }

    #[test]
    fn cooldown_doubles_with_each_failed_bypass() {
        assert_eq!(scanner_bypass_cooldown(0), SCANNER_BYPASS_BASE_COOLDOWN);
        assert_eq!(scanner_bypass_cooldown(1), SCANNER_BYPASS_BASE_COOLDOWN * 2);
        assert_eq!(scanner_bypass_cooldown(2), SCANNER_BYPASS_BASE_COOLDOWN * 4);
    }

    #[test]
    fn cooldown_is_capped_at_main_reconnect_max() {
        // Many failures should still saturate at SLEEP_RECONNECT_MAX, never above.
        assert_eq!(scanner_bypass_cooldown(20), SLEEP_RECONNECT_MAX);
    }

    #[test]
    fn auto_mode_different_ip_ignores_cooldown() {
        // A fresh IP is brand-new information; cooldown must not gate it.
        let now = Instant::now();
        let just_bypassed = now - Duration::from_secs(1);
        assert_eq!(
            decide_discovery_notify_action(
                "10.0.0.50",
                ADDR_AUTO,
                "10.0.0.73",
                None,
                Some(just_bypassed),
                5,
                now,
            ),
            DiscoveryNotifyAction::BypassBackoff
        );
    }

    #[test]
    fn explicit_ip_different_ip_ignores_cooldown_and_reports() {
        // Cooldown only gates bypass; the Report path is independent.
        let now = Instant::now();
        let just_bypassed = now - Duration::from_secs(1);
        assert_eq!(
            decide_discovery_notify_action(
                "10.0.0.50",
                "10.0.0.50",
                "10.0.0.73",
                None,
                Some(just_bypassed),
                5,
                now,
            ),
            DiscoveryNotifyAction::Report
        );
    }

    // --- M5.2/M5.3 — lifecycle tests ---
    //
    // These exercise the publicly observable state transitions for
    // `close()` / `stop()` / drop, without needing a mock TCP server. They
    // pin the M2.1 / M2.2 invariants:
    //   * `close` leaves the device "Disconnected" but resumable.
    //   * `stop` is terminal — `is_stopped` returns true and `cancel_token`
    //     is fired.
    //   * `close_notify` is sent on both paths, so the actor's `select!` can
    //     short-circuit any in-flight command.

    use crate::Device;
    use std::sync::Arc;

    fn make_test_device() -> Device {
        // Use a clearly non-routable target so the background connection task
        // never actually connects. We're only exercising lifecycle state
        // machinery here, not network IO.
        Device::builder("test_lifecycle_id", b"0123456789abcdef".to_vec())
            .address("203.0.113.1") // TEST-NET-3, RFC 5737
            .persist(false)
            .build()
    }

    #[test]
    fn fire_close_marks_disconnected_but_not_stopped() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        rt.block_on(async {
            let device = make_test_device();
            assert!(!device.is_stopped());
            device.fire_close();
            assert!(!device.is_stopped(), "close must not move to Stopped");
        });
    }

    #[test]
    fn fire_stop_marks_stopped_and_cancels_token() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        rt.block_on(async {
            let device = make_test_device();
            let token = device.inner.cancel_token.clone();
            assert!(!device.is_stopped());
            assert!(!token.is_cancelled());

            device.fire_stop();

            assert!(device.is_stopped(), "stop must move to Stopped");
            assert!(token.is_cancelled(), "stop must fire cancel_token");
        });
    }

    #[test]
    fn close_notify_wakes_subscribers() {
        // The Notify-based close path (M2.1) must produce a wake-up for any
        // task that calls `notified().await` before close is fired.
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        rt.block_on(async {
            let device = make_test_device();
            let inner = Arc::clone(&device.inner);

            // Spawn a waiter that's parked on close_notify.
            let waiter = tokio::spawn(async move {
                inner.close_notify.notified().await;
            });

            // Give the waiter a moment to register its Notified future.
            tokio::time::sleep(std::time::Duration::from_millis(20)).await;

            device.fire_close();

            // The waiter must complete promptly.
            tokio::time::timeout(std::time::Duration::from_millis(200), waiter)
                .await
                .expect("close_notify did not wake waiter within 200ms")
                .expect("waiter task panicked");
        });
    }

    // M5.5: state-machine invariants for ConnectionState transitions driven
    // by fire_close / fire_stop. Hand-rolled exhaustive table rather than
    // pulling in proptest as a dev-dep for a 4-state machine.

    #[test]
    fn stopped_is_terminal_for_fire_close() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        rt.block_on(async {
            let device = make_test_device();
            device.fire_stop();
            assert!(device.is_stopped());
            // Calling close after stop must NOT resurrect the device into
            // any state other than Stopped.
            device.fire_close();
            assert!(device.is_stopped(), "fire_close on Stopped must be a no-op");
        });
    }

    #[test]
    fn fire_stop_is_idempotent() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        rt.block_on(async {
            let device = make_test_device();
            device.fire_stop();
            let token1_cancelled = device.inner.cancel_token.is_cancelled();
            device.fire_stop();
            let token2_cancelled = device.inner.cancel_token.is_cancelled();
            assert!(token1_cancelled && token2_cancelled);
            assert!(device.is_stopped());
        });
    }

    #[test]
    fn fire_close_does_not_set_stopped() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        rt.block_on(async {
            let device = make_test_device();
            // Initial state is Disconnected (test address can't connect).
            assert!(!device.is_stopped());
            device.fire_close();
            assert!(
                !device.is_stopped(),
                "fire_close should leave state available for restart, not move to Stopped"
            );
            assert!(
                !device.inner.cancel_token.is_cancelled(),
                "fire_close must not cancel the connection task"
            );
        });
    }

    #[test]
    fn dropping_last_device_clone_cancels_token() {
        // M2.2: when the last strong Arc<DeviceInner> goes away (i.e. user
        // dropped all Device clones AND the connection task exited), the
        // Drop impl on DeviceInner fires `cancel_token.cancel()`. We can
        // observe this by holding a clone of the token after dropping the
        // device.
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        rt.block_on(async {
            let device = make_test_device();
            let token = device.inner.cancel_token.clone();
            // Force the connection task to exit by calling fire_stop. After
            // stop, the actor drops its strong ref; dropping our Device
            // handle removes the last user-facing ref.
            device.fire_stop();
            drop(device);
            // The token is already cancelled by fire_stop, so DeviceInner's
            // Drop firing cancel again is a no-op — we just check the
            // observable invariant.
            assert!(token.is_cancelled());
        });
    }
}