pnio 0.1.0

//! `Ar`: the pure Application Relationship (AR) state machine (IEC 61158-6-10 §6).
//!
//! No I/O, no clock of its own: `on()` takes an `Event` and the current time and
//! returns the `Action`s to perform. `next_deadline()` is queried by the caller (the
//! `device` loop) instead of the state machine pushing timer-set actions — one less
//! action type, same behaviour: the loop sleeps until the deadline and then feeds a
//! `Tick`.
//!
//! State progression: `Idle` --Connect--> `Connected` --Write*--> `Connected`
//! --PrmEnd--> `AppReadySent` --AppReadyRsp(ok)--> `Data`. Any Release or abort
//! condition (activity timeout, ApplicationReady failure/rejection, or an external
//! reason) returns to `Idle`, dropping the AR context and clearing both timers.

use super::connect::{build_connect_res, validate, ArParams, ConnectReq};
use super::control::{app_ready_req, prm_end_done, release_done, ControlBlock};
use super::model::DeviceModel;
use super::status::PnioStatus;
use super::write::{build_write_res, Record, WriteReq};
use std::time::{Duration, Instant};

/// ApplicationReady request retry timeout: how long the device waits for the
/// controller's `IODControlRes(ApplicationReady)` before retrying.
pub const APP_READY_TIMEOUT: Duration = Duration::from_secs(1);
/// Maximum number of ApplicationReady requests sent (the initial one plus retries)
/// before the AR is aborted.
pub const APP_READY_MAX_ATTEMPTS: u8 = 3;
/// Unit of the AR's `activity_timeout_factor` (from the Connect request's
/// ARBlockReq), IEC 61158-6-10 §4.6.1.1: the activity deadline is
/// `now + activity_timeout_factor * ACTIVITY_TIMEOUT_UNIT`.
pub const ACTIVITY_TIMEOUT_UNIT: Duration = Duration::from_millis(100);
/// Maximum number of parameter-write records accumulated per AR.
const MAX_RECORDS: usize = 64;
/// Maximum summed `data` bytes across all parameter-write records accumulated per AR.
const MAX_RECORD_BYTES: usize = 64 * 1024;

/// The AR's lifecycle state.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ArState {
    /// No AR established.
    Idle,
    /// Connect accepted; accumulating parameter-write records until PrmEnd.
    Connected,
    /// PrmEnd received; waiting for the controller's answer to our ApplicationReady call.
    AppReadySent,
    /// ApplicationReady acknowledged: the AR is ready for cyclic (RTC1) data.
    Data,
}

/// Why an AR was aborted back to `Idle`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AbortReason {
    /// The controller sent a Release request.
    ControllerRelease,
    /// `APP_READY_MAX_ATTEMPTS` ApplicationReady requests all timed out.
    AppReadyFailed,
    /// The controller answered ApplicationReady with a non-OK status.
    AppReadyRejected(PnioStatus),
    /// No Write/PrmEnd activity within `activity_timeout_factor * ACTIVITY_TIMEOUT_UNIT`.
    ActivityTimeout,
    /// The controller re-sent Connect for this AR (same `ar_uuid`, new `session_key`)
    /// or for a different `ar_uuid` from the same initiator: it has abandoned the
    /// existing AR (typically after its own ERR-RTA abort, invisible to us on the
    /// alarm channel) and is establishing a new one. We drop the stale AR and process
    /// the Connect as if from `Idle`.
    ControllerReconnect,
    /// The RT runner's consumer watchdog expired: the controller stopped providing
    /// output data for the cyclic (RTC1) exchange within the negotiated window.
    RtWatchdog,
    /// Any other externally-triggered abort (e.g. link down), free-form reason.
    External(&'static str),
    /// The controller sent an ERR-RTA on the alarm channel.
    ControllerErrRta(PnioStatus),
    /// An alarm we sent was never acknowledged within `rta_retries` resends.
    AlarmSendFailed,
    /// The RT thread's socket failed (was reported as `RtWatchdog` before Plan 5).
    RtSocket,
    /// The application stopped the device (`IoDevice::stop`).
    Shutdown,
}

/// An input to the AR state machine. Carries the already-parsed request so `Ar`
/// itself does no codec work.
#[derive(Debug, Clone, PartialEq)]
pub enum Event {
    /// A Connect request arrived; validate it against the model and (re)establish the AR.
    ConnectReq(ConnectReq),
    /// A parameter-write request arrived; accumulate its records against the current AR.
    WriteReq(WriteReq),
    /// A PrmEnd control request arrived: parameterization is done, place the
    /// ApplicationReady call.
    PrmEndReq(ControlBlock),
    /// A Release control request arrived: tear the AR down.
    ReleaseReq(ControlBlock),
    /// The controller answered our ApplicationReady call.
    AppReadyRsp {
        /// The controller's reported result; `PnioStatus::OK` accepts the AR into `Data`.
        status: PnioStatus,
    },
    /// The caller's deadline (from [`Ar::next_deadline`]) elapsed; check timeouts.
    Tick,
    /// An external condition (RT watchdog, alarm failure, shutdown, ...) requires the AR to abort.
    Abort(AbortReason),
}

impl Event {
    fn name(&self) -> &'static str {
        match self {
            Event::ConnectReq(_) => "ConnectReq",
            Event::WriteReq(_) => "WriteReq",
            Event::PrmEndReq(_) => "PrmEndReq",
            Event::ReleaseReq(_) => "ReleaseReq",
            Event::AppReadyRsp { .. } => "AppReadyRsp",
            Event::Tick => "Tick",
            Event::Abort(_) => "Abort",
        }
    }
}

/// An effect the caller must perform: send an RPC response, call out to the
/// controller (ApplicationReady request), or notify observers of a state change.
#[derive(Debug, Clone, PartialEq)]
pub enum Action {
    /// Answer the request that triggered this event with these PNIO blocks (empty
    /// on rejection/wrong-state).
    Respond {
        /// Result to answer the triggering request with.
        status: PnioStatus,
        /// PNIO response blocks to append; empty on rejection/wrong-state.
        blocks: Vec<u8>,
    },
    /// Call the controller (ApplicationReady request PNIO blocks).
    CallController {
        /// The ApplicationReady request's PNIO blocks.
        blocks: Vec<u8>,
    },
    /// The AR changed state; `reason` is set only when the new state is `Idle` via
    /// an abort.
    Notify {
        /// The AR's new state.
        state: ArState,
        /// Why the AR aborted, when `state == ArState::Idle` via an abort (`None` on a
        /// normal forward transition).
        reason: Option<AbortReason>,
    },
}

/// The established AR's context: the negotiated parameters, the parameter-write
/// records accumulated so far, when the AR reached `Connected`, and the cached
/// Connect response bytes (so a duplicate Connect resend is byte-identical).
#[derive(Debug, Clone, PartialEq)]
pub struct ArContext {
    /// The AR parameters extracted from the validated Connect request.
    pub params: ArParams,
    /// Parameter-write records accumulated on this AR so far, in arrival order,
    /// capped at `MAX_RECORDS` records / `MAX_RECORD_BYTES` summed data bytes.
    pub records: Vec<Record>,
    /// When the AR transitioned into `Connected`.
    pub connected_at: Instant,
    connect_res: Vec<u8>,
}

/// The pure AR state machine: no I/O, no clock of its own. `on()` is the only way
/// to drive it; `next_deadline()` tells the caller when to feed a `Tick`.
pub struct Ar {
    model: DeviceModel,
    state: ArState,
    ctx: Option<ArContext>,
    app_ready_attempts: u8,
    app_ready_deadline: Option<Instant>,
    activity_deadline: Option<Instant>,
}

impl Ar {
    /// A fresh, idle AR state machine validating Connect requests against `model`.
    pub fn new(model: DeviceModel) -> Ar {
        Ar {
            model,
            state: ArState::Idle,
            ctx: None,
            app_ready_attempts: 0,
            app_ready_deadline: None,
            activity_deadline: None,
        }
    }

    /// The AR's current lifecycle state.
    pub fn state(&self) -> ArState {
        self.state
    }

    /// The established AR's context, if one exists (`state() != ArState::Idle`).
    pub fn context(&self) -> Option<&ArContext> {
        self.ctx.as_ref()
    }

    /// The earliest of the activity and ApplicationReady deadlines, if any is
    /// armed. The caller (`device` loop) sleeps until this instant and then feeds
    /// `Event::Tick`.
    pub fn next_deadline(&self) -> Option<Instant> {
        match (self.activity_deadline, self.app_ready_deadline) {
            (Some(a), Some(b)) => Some(a.min(b)),
            (Some(a), None) => Some(a),
            (None, Some(b)) => Some(b),
            (None, None) => None,
        }
    }

    /// Drive the state machine with one event, returning the actions to perform.
    pub fn on(&mut self, ev: Event, now: Instant) -> Vec<Action> {
        let prev = self.state;
        let event_name = ev.name();

        let actions = if let Event::Abort(reason) = ev {
            if self.state == ArState::Idle {
                Vec::new()
            } else {
                vec![self.abort(reason)]
            }
        } else {
            match ev {
                Event::ConnectReq(req) => self.handle_connect(req, now),
                Event::WriteReq(req) => self.handle_write(req, now),
                Event::PrmEndReq(req) => self.handle_prm_end(req, now),
                Event::ReleaseReq(req) => self.handle_release(req),
                Event::AppReadyRsp { status } => self.handle_app_ready_rsp(status),
                Event::Tick => self.handle_tick(now),
                Event::Abort(_) => unreachable!("handled above"),
            }
        };

        let aborted = actions.iter().any(|a| {
            matches!(
                a,
                Action::Notify {
                    reason: Some(_),
                    ..
                }
            )
        });
        // A no-op transition (nothing due on a Tick, an AppReadyRsp/Abort ignored
        // outside their state, or a request answered by resending an already-cached
        // response with the AR otherwise unchanged) is logged at `debug!` instead of
        // `info!` to avoid spamming the log with the CPU's periodic Tick-driven
        // polling and idempotent retransmissions.
        let plain_resend = prev != ArState::Idle
            && prev == self.state
            && matches!(event_name, "ConnectReq" | "PrmEndReq")
            && matches!(actions.as_slice(), [Action::Respond { status, .. }] if status.is_ok());
        if aborted {
            log::warn!("AR {prev:?} --{event_name}--> {:?} (abort)", self.state);
        } else if actions.is_empty() || plain_resend {
            log::debug!("AR {prev:?} --{event_name}--> {:?}", self.state);
        } else {
            log::info!("AR {prev:?} --{event_name}--> {:?}", self.state);
        }
        actions
    }

    fn handle_connect(&mut self, req: ConnectReq, now: Instant) -> Vec<Action> {
        match self.state {
            ArState::Idle => match validate(&req, &self.model) {
                Ok(params) => {
                    let connect_res = build_connect_res(&params, &self.model);
                    let activity_deadline =
                        now + ACTIVITY_TIMEOUT_UNIT * params.activity_timeout_factor as u32;
                    self.ctx = Some(ArContext {
                        params,
                        records: Vec::new(),
                        connected_at: now,
                        connect_res: connect_res.clone(),
                    });
                    self.state = ArState::Connected;
                    self.activity_deadline = Some(activity_deadline);
                    vec![
                        Action::Respond {
                            status: PnioStatus::OK,
                            blocks: connect_res,
                        },
                        Action::Notify {
                            state: ArState::Connected,
                            reason: None,
                        },
                    ]
                }
                Err(status) => vec![Action::Respond {
                    status,
                    blocks: Vec::new(),
                }],
            },
            ArState::Connected | ArState::AppReadySent | ArState::Data => {
                let ctx = self.ctx.as_ref().expect("ctx present outside Idle");
                let same_ar = req.ar.ar_uuid == ctx.params.ar_uuid;
                let same_session = req.ar.session_key == ctx.params.session_key;
                let same_initiator =
                    req.ar.initiator_object_uuid == ctx.params.initiator_object_uuid;
                if same_ar && same_session {
                    // Exact retransmission: answer with the byte-identical cached
                    // response.
                    vec![Action::Respond {
                        status: PnioStatus::OK,
                        blocks: ctx.connect_res.clone(),
                    }]
                } else if same_ar || same_initiator {
                    // Same AR with a new session key, or a new AR from the same
                    // initiator: the controller has abandoned the old AR (we likely
                    // missed its ERR-RTA abort on the alarm channel) and is
                    // establishing a new one. Drop the stale context and process this
                    // Connect exactly as if it arrived in `Idle`.
                    let mut actions = vec![self.abort(AbortReason::ControllerReconnect)];
                    actions.extend(self.handle_connect(req, now));
                    actions
                } else {
                    vec![Action::Respond {
                        status: PnioStatus::connect_ar_already_exists(),
                        blocks: Vec::new(),
                    }]
                }
            }
        }
    }

    fn handle_write(&mut self, req: WriteReq, now: Instant) -> Vec<Action> {
        match self.state {
            ArState::Idle => vec![wrong_state()],
            // Activity timer is armed on Connect and re-armed on Write only while
            // still in Connected; once PrmEnd has started the ApplicationReady
            // phase (AppReadySent) or the AR is in Data, a Write stores records
            // and touches no timer.
            ArState::Connected | ArState::AppReadySent | ArState::Data => {
                let ctx = self.ctx.as_ref().expect("ctx present outside Idle");
                // A record addressed to a foreign AR must not be stored, whichever
                // state we're in: the request's `ar_uuid` is per-record, not
                // per-request, so every one of them must match.
                if req.records.iter().any(|r| r.ar_uuid != ctx.params.ar_uuid) {
                    return vec![reject(PnioStatus::write_wrong_ar())];
                }
                if exceeds_record_cap(ctx, &req.records) {
                    return vec![reject(PnioStatus::write_resource_unavailable())];
                }
                let blocks = build_write_res(&req);
                let rearm_activity = self.state == ArState::Connected;
                let factor = ctx.params.activity_timeout_factor;
                let ctx = self.ctx.as_mut().expect("ctx present outside Idle");
                ctx.records.extend(req.records);
                if rearm_activity {
                    self.activity_deadline = Some(now + ACTIVITY_TIMEOUT_UNIT * factor as u32);
                }
                vec![Action::Respond {
                    status: PnioStatus::OK,
                    blocks,
                }]
            }
        }
    }

    fn handle_prm_end(&mut self, req: ControlBlock, now: Instant) -> Vec<Action> {
        match self.state {
            ArState::Idle => vec![wrong_state()],
            ArState::Connected => {
                let ctx = self.ctx.as_ref().expect("ctx present outside Idle");
                if req.ar_uuid != ctx.params.ar_uuid {
                    return vec![wrong_state()];
                }
                let mut respond = Vec::new();
                prm_end_done(&req).write(&mut respond);
                let mut call = Vec::new();
                // Built from the AR's own context, not the request: a PrmEnd that
                // (once matched above) belongs to this AR still shouldn't let a
                // mismatched field in its body redirect the outgoing call.
                app_ready_req(ctx.params.ar_uuid, ctx.params.session_key).write(&mut call);
                self.state = ArState::AppReadySent;
                self.app_ready_attempts = 1;
                self.app_ready_deadline = Some(now + APP_READY_TIMEOUT);
                self.activity_deadline = None;
                vec![
                    Action::Respond {
                        status: PnioStatus::OK,
                        blocks: respond,
                    },
                    Action::CallController { blocks: call },
                ]
            }
            ArState::AppReadySent | ArState::Data => {
                let ctx = self.ctx.as_ref().expect("ctx present outside Idle");
                if req.ar_uuid != ctx.params.ar_uuid {
                    return vec![wrong_state()];
                }
                let mut respond = Vec::new();
                prm_end_done(&req).write(&mut respond);
                vec![Action::Respond {
                    status: PnioStatus::OK,
                    blocks: respond,
                }]
            }
        }
    }

    fn handle_release(&mut self, req: ControlBlock) -> Vec<Action> {
        if self.state != ArState::Idle {
            let ctx = self.ctx.as_ref().expect("ctx present outside Idle");
            if req.ar_uuid != ctx.params.ar_uuid {
                return vec![wrong_state()];
            }
        }
        let mut respond = Vec::new();
        release_done(&req).write(&mut respond);
        let mut actions = vec![Action::Respond {
            status: PnioStatus::OK,
            blocks: respond,
        }];
        if self.state != ArState::Idle {
            actions.push(self.abort(AbortReason::ControllerRelease));
        }
        actions
    }

    fn handle_app_ready_rsp(&mut self, status: PnioStatus) -> Vec<Action> {
        if self.state != ArState::AppReadySent {
            return Vec::new(); // ignored outside AppReadySent
        }
        if status.is_ok() {
            self.state = ArState::Data;
            self.app_ready_deadline = None;
            self.activity_deadline = None;
            self.app_ready_attempts = 0;
            vec![Action::Notify {
                state: ArState::Data,
                reason: None,
            }]
        } else {
            vec![self.abort(AbortReason::AppReadyRejected(status))]
        }
    }

    fn handle_tick(&mut self, now: Instant) -> Vec<Action> {
        match self.state {
            ArState::Connected => match self.activity_deadline {
                Some(deadline) if now >= deadline => {
                    vec![self.abort(AbortReason::ActivityTimeout)]
                }
                _ => Vec::new(),
            },
            ArState::AppReadySent => match self.app_ready_deadline {
                Some(deadline) if now >= deadline => {
                    if self.app_ready_attempts < APP_READY_MAX_ATTEMPTS {
                        self.app_ready_attempts += 1;
                        self.app_ready_deadline = Some(now + APP_READY_TIMEOUT);
                        let ctx = self.ctx.as_ref().expect("ctx present in AppReadySent");
                        let mut call = Vec::new();
                        app_ready_req(ctx.params.ar_uuid, ctx.params.session_key).write(&mut call);
                        vec![Action::CallController { blocks: call }]
                    } else {
                        vec![self.abort(AbortReason::AppReadyFailed)]
                    }
                }
                _ => Vec::new(),
            },
            ArState::Idle | ArState::Data => Vec::new(),
        }
    }

    /// Clear the AR context and both timers, return to `Idle`, and produce the
    /// `Notify` action reporting the abort reason.
    fn abort(&mut self, reason: AbortReason) -> Action {
        self.ctx = None;
        self.app_ready_deadline = None;
        self.activity_deadline = None;
        self.app_ready_attempts = 0;
        self.state = ArState::Idle;
        Action::Notify {
            state: ArState::Idle,
            reason: Some(reason),
        }
    }
}

fn wrong_state() -> Action {
    reject(PnioStatus::control_wrong_state())
}

fn reject(status: PnioStatus) -> Action {
    Action::Respond {
        status,
        blocks: Vec::new(),
    }
}

/// Whether accumulating `incoming` records onto `ctx.records` would exceed the
/// per-AR limits ([`MAX_RECORDS`] records or [`MAX_RECORD_BYTES`] summed `data`).
fn exceeds_record_cap(ctx: &ArContext, incoming: &[Record]) -> bool {
    let count = ctx.records.len() + incoming.len();
    let bytes: usize = ctx
        .records
        .iter()
        .chain(incoming)
        .map(|r| r.data.len())
        .sum();
    count > MAX_RECORDS || bytes > MAX_RECORD_BYTES
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::cm::block::ty;
    use crate::cm::connect::ConnectReq;
    use crate::cm::control::{cmd, ControlBlock};
    use crate::cm::model::DeviceModel;
    use crate::cm::write::WriteReq;
    use crate::eth::MacAddr;
    use crate::testutil::golden;
    use std::time::{Duration, Instant};

    const BLOCKS: usize = 142;
    const MAC: MacAddr = MacAddr([0x8c, 0xf3, 0x19, 0xcd, 0x19, 0xf8]);

    fn ar() -> Ar {
        Ar::new(DeviceModel::pnet_sample(MAC))
    }
    fn connect() -> Event {
        Event::ConnectReq(ConnectReq::parse(&golden("connect_req")[BLOCKS..]).unwrap())
    }
    fn write() -> Event {
        Event::WriteReq(WriteReq::parse(&golden("write_req")[BLOCKS..]).unwrap())
    }
    fn prm_end() -> Event {
        Event::PrmEndReq(ControlBlock::parse(&golden("prmend_req")[BLOCKS..]).unwrap())
    }
    fn t0() -> Instant {
        Instant::now()
    }

    fn respond_ok(actions: &[Action]) -> &Vec<u8> {
        match &actions[0] {
            Action::Respond { status, blocks } if status.is_ok() => blocks,
            other => panic!("{other:?}"),
        }
    }

    /// A Release request block (built by mutating a parsed PrmEnd request, same as
    /// the original `release_aborts_and_answers` test does inline).
    fn release_block() -> ControlBlock {
        let mut rel = ControlBlock::parse(&golden("prmend_req")[BLOCKS..]).unwrap();
        rel.block_type = ty::RELEASE_BLOCK_REQ;
        rel.command = cmd::RELEASE;
        rel
    }

    /// Drive a fresh `Ar` all the way to `Data` via the nominal path (Connect,
    /// Write, PrmEnd, AppReadyRsp ok), for tests of the Data row of the transition
    /// table. Returns the machine and the `now` used throughout.
    fn to_data() -> (Ar, Instant) {
        let mut ar = ar();
        let now = t0();
        ar.on(connect(), now);
        ar.on(write(), now);
        ar.on(prm_end(), now);
        ar.on(
            Event::AppReadyRsp {
                status: PnioStatus::OK,
            },
            now,
        );
        assert_eq!(ar.state(), ArState::Data);
        (ar, now)
    }

    #[test]
    fn nominal_idle_to_data() {
        let mut ar = ar();
        let now = t0();
        let a = ar.on(connect(), now);
        assert_eq!(respond_ok(&a), &golden("connect_res")[BLOCKS..]);
        assert!(matches!(
            a[1],
            Action::Notify {
                state: ArState::Connected,
                reason: None
            }
        ));
        assert_eq!(ar.state(), ArState::Connected);
        let a = ar.on(write(), now);
        assert_eq!(respond_ok(&a), &golden("write_res")[BLOCKS..]);
        assert_eq!(ar.context().unwrap().records.len(), 5);
        let a = ar.on(prm_end(), now);
        assert_eq!(respond_ok(&a), &golden("prmend_res")[BLOCKS..]);
        assert!(
            matches!(&a[1], Action::CallController { blocks } if blocks == &golden("appready_req")[BLOCKS..])
        );
        assert_eq!(ar.state(), ArState::AppReadySent);
        assert_eq!(ar.next_deadline(), Some(now + APP_READY_TIMEOUT));
        let a = ar.on(
            Event::AppReadyRsp {
                status: PnioStatus::OK,
            },
            now,
        );
        assert!(matches!(
            a[0],
            Action::Notify {
                state: ArState::Data,
                reason: None
            }
        ));
        assert_eq!(ar.state(), ArState::Data);
        assert_eq!(ar.next_deadline(), None);
    }

    #[test]
    fn rejected_connect_stays_idle_with_status() {
        let mut ar = Ar::new({
            let mut m = DeviceModel::pnet_sample(MAC);
            m.slots.pop();
            m
        });
        let a = ar.on(connect(), t0());
        assert!(
            matches!(&a[0], Action::Respond { status, blocks } if !status.is_ok() && blocks.is_empty())
        );
        assert_eq!(ar.state(), ArState::Idle);
        assert!(ar.context().is_none());
    }

    #[test]
    fn duplicate_connect_is_idempotent_and_other_ar_is_rejected() {
        let mut ar = ar();
        let now = t0();
        let first = respond_ok(&ar.on(connect(), now)).clone();
        assert_eq!(respond_ok(&ar.on(connect(), now)), &first);
        let mut other = match connect() {
            Event::ConnectReq(c) => c,
            _ => unreachable!(),
        };
        other.ar.ar_uuid = crate::rpc::Uuid([7; 16]);
        other.ar.initiator_object_uuid = crate::rpc::Uuid([8; 16]);
        let a = ar.on(Event::ConnectReq(other), now);
        assert!(
            matches!(&a[0], Action::Respond { status, .. } if *status == PnioStatus::connect_ar_already_exists())
        );
        assert_eq!(ar.state(), ArState::Connected);
    }

    #[test]
    fn write_or_prm_end_in_idle_is_wrong_state() {
        let mut ar = ar();
        let a = ar.on(write(), t0());
        assert!(
            matches!(&a[0], Action::Respond { status, .. } if *status == PnioStatus::control_wrong_state())
        );
        let a = ar.on(prm_end(), t0());
        assert!(
            matches!(&a[0], Action::Respond { status, .. } if *status == PnioStatus::control_wrong_state())
        );
    }

    #[test]
    fn app_ready_retries_three_times_then_aborts() {
        let mut ar = ar();
        let now = t0();
        ar.on(connect(), now);
        let a = ar.on(prm_end(), now);
        assert!(matches!(a[1], Action::CallController { .. }));
        let t1 = now + APP_READY_TIMEOUT + Duration::from_millis(1);
        let a = ar.on(Event::Tick, t1);
        assert!(matches!(a[0], Action::CallController { .. }));
        let t2 = t1 + APP_READY_TIMEOUT + Duration::from_millis(1);
        let a = ar.on(Event::Tick, t2);
        assert!(matches!(a[0], Action::CallController { .. }));
        let t3 = t2 + APP_READY_TIMEOUT + Duration::from_millis(1);
        let a = ar.on(Event::Tick, t3);
        assert!(matches!(
            a[0],
            Action::Notify {
                state: ArState::Idle,
                reason: Some(AbortReason::AppReadyFailed)
            }
        ));
        assert_eq!(ar.state(), ArState::Idle);
    }

    #[test]
    fn app_ready_bad_status_aborts() {
        let mut ar = ar();
        let now = t0();
        ar.on(connect(), now);
        ar.on(prm_end(), now);
        let bad = PnioStatus::new(0xdd, 0x81, 1, 1);
        let a = ar.on(Event::AppReadyRsp { status: bad }, now);
        assert!(matches!(
            a[0],
            Action::Notify {
                state: ArState::Idle,
                reason: Some(AbortReason::AppReadyRejected(s))
            } if s == bad
        ));
    }

    #[test]
    fn activity_timeout_before_data_aborts() {
        let mut ar = ar();
        let now = t0();
        ar.on(connect(), now); // factor 200 -> 20 s
        assert_eq!(
            ar.next_deadline(),
            Some(now + Duration::from_millis(200 * 100))
        );
        let a = ar.on(Event::Tick, now + Duration::from_secs(21));
        assert!(matches!(
            a[0],
            Action::Notify {
                state: ArState::Idle,
                reason: Some(AbortReason::ActivityTimeout)
            }
        ));
    }

    #[test]
    fn release_aborts_and_answers() {
        let mut ar = ar();
        let now = t0();
        ar.on(connect(), now);
        let mut rel = ControlBlock::parse(&golden("prmend_req")[BLOCKS..]).unwrap();
        rel.block_type = ty::RELEASE_BLOCK_REQ;
        rel.command = cmd::RELEASE;
        let a = ar.on(Event::ReleaseReq(rel), now);
        assert!(
            matches!(&a[0], Action::Respond { status, blocks } if status.is_ok() && blocks[0..2] == [0x81, 0x14])
        );
        assert!(matches!(
            a[1],
            Action::Notify {
                state: ArState::Idle,
                reason: Some(AbortReason::ControllerRelease)
            }
        ));
    }

    #[test]
    fn connect_after_abort_succeeds_with_fresh_context() {
        let mut ar = ar();
        let now = t0();
        ar.on(connect(), now);
        ar.on(Event::Abort(AbortReason::External("test")), now);
        assert!(ar.context().is_none());
        let a = ar.on(connect(), now);
        assert_eq!(respond_ok(&a), &golden("connect_res")[BLOCKS..]);
    }

    // -----------------------------------------------------------------------
    // Fix round 1: records accumulate; the activity timer is armed only while
    // Connected; and one test per remaining, previously-untested transition
    // table cell.
    // -----------------------------------------------------------------------

    #[test]
    fn records_accumulate_across_writes() {
        let mut ar = ar();
        let now = t0();
        ar.on(connect(), now);
        ar.on(write(), now);
        assert_eq!(ar.context().unwrap().records.len(), 5);
        ar.on(write(), now);
        assert_eq!(ar.context().unwrap().records.len(), 10);
    }

    #[test]
    fn write_in_app_ready_sent_does_not_touch_timers() {
        let mut ar = ar();
        let now = t0();
        ar.on(connect(), now);
        ar.on(prm_end(), now);
        assert_eq!(ar.next_deadline(), Some(now + APP_READY_TIMEOUT));
        let a = ar.on(write(), now);
        assert_eq!(respond_ok(&a), &golden("write_res")[BLOCKS..]);
        assert_eq!(ar.context().unwrap().records.len(), 5);
        assert_eq!(ar.next_deadline(), Some(now + APP_READY_TIMEOUT));
    }

    // --- Idle ---

    #[test]
    fn idle_release_answers_without_abort() {
        let mut ar = ar();
        let a = ar.on(Event::ReleaseReq(release_block()), t0());
        assert_eq!(a.len(), 1);
        assert!(
            matches!(&a[0], Action::Respond { status, blocks } if status.is_ok() && blocks[0..2] == [0x81, 0x14])
        );
        assert_eq!(ar.state(), ArState::Idle);
        assert!(ar.context().is_none());
    }

    #[test]
    fn idle_app_ready_rsp_is_ignored() {
        let mut ar = ar();
        let a = ar.on(
            Event::AppReadyRsp {
                status: PnioStatus::OK,
            },
            t0(),
        );
        assert!(a.is_empty());
        assert_eq!(ar.state(), ArState::Idle);
    }

    #[test]
    fn idle_tick_is_noop() {
        let mut ar = ar();
        let a = ar.on(Event::Tick, t0());
        assert!(a.is_empty());
        assert_eq!(ar.next_deadline(), None);
    }

    // --- Connected ---

    #[test]
    fn connected_app_ready_rsp_is_ignored() {
        let mut ar = ar();
        let now = t0();
        ar.on(connect(), now);
        let a = ar.on(
            Event::AppReadyRsp {
                status: PnioStatus::OK,
            },
            now,
        );
        assert!(a.is_empty());
        assert_eq!(ar.state(), ArState::Connected);
    }

    #[test]
    fn connected_tick_before_deadline_is_noop() {
        let mut ar = ar();
        let now = t0();
        ar.on(connect(), now); // factor 200 -> 20 s deadline
        let a = ar.on(Event::Tick, now + Duration::from_secs(1));
        assert!(a.is_empty());
        assert_eq!(ar.state(), ArState::Connected);
    }

    #[test]
    fn connected_write_rearms_activity_deadline_from_now() {
        let mut ar = ar();
        let now = t0();
        ar.on(connect(), now);
        let now2 = now + Duration::from_secs(5);
        ar.on(write(), now2);
        assert_eq!(ar.next_deadline(), Some(now2 + Duration::from_secs(20)));
    }

    // --- AppReadySent ---

    #[test]
    fn app_ready_sent_connect_same_uuid_is_idempotent() {
        let mut ar = ar();
        let now = t0();
        ar.on(connect(), now);
        ar.on(prm_end(), now);
        let a = ar.on(connect(), now);
        assert_eq!(respond_ok(&a), &golden("connect_res")[BLOCKS..]);
        assert_eq!(ar.state(), ArState::AppReadySent);
    }

    #[test]
    fn app_ready_sent_prm_end_again_is_idempotent_without_new_call() {
        let mut ar = ar();
        let now = t0();
        ar.on(connect(), now);
        ar.on(prm_end(), now);
        let a = ar.on(prm_end(), now);
        assert_eq!(a.len(), 1);
        assert_eq!(respond_ok(&a), &golden("prmend_res")[BLOCKS..]);
        // No re-arm/retry happened: the deadline is unchanged, so attempts is too.
        assert_eq!(ar.next_deadline(), Some(now + APP_READY_TIMEOUT));
    }

    #[test]
    fn app_ready_sent_release_aborts() {
        let mut ar = ar();
        let now = t0();
        ar.on(connect(), now);
        ar.on(prm_end(), now);
        let a = ar.on(Event::ReleaseReq(release_block()), now);
        assert!(
            matches!(&a[0], Action::Respond { status, blocks } if status.is_ok() && blocks[0..2] == [0x81, 0x14])
        );
        assert!(matches!(
            a[1],
            Action::Notify {
                state: ArState::Idle,
                reason: Some(AbortReason::ControllerRelease)
            }
        ));
        assert_eq!(ar.state(), ArState::Idle);
    }

    // --- Data ---

    #[test]
    fn data_connect_other_uuid_is_rejected() {
        let (mut ar, now) = to_data();
        let mut other = match connect() {
            Event::ConnectReq(c) => c,
            _ => unreachable!(),
        };
        other.ar.ar_uuid = crate::rpc::Uuid([7; 16]);
        other.ar.initiator_object_uuid = crate::rpc::Uuid([8; 16]);
        let a = ar.on(Event::ConnectReq(other), now);
        assert!(
            matches!(&a[0], Action::Respond { status, .. } if *status == PnioStatus::connect_ar_already_exists())
        );
        assert_eq!(ar.state(), ArState::Data);
    }

    // -----------------------------------------------------------------------
    // Fix round 2 (HIL bring-up): a controller reconnect (same `ar_uuid` with a
    // new `session_key`, or a new `ar_uuid` from the same initiator) takes over
    // the AR instead of being treated as either an idempotent resend or a
    // conflicting other AR.
    // -----------------------------------------------------------------------

    #[test]
    fn reconnect_with_new_session_key_takes_over() {
        let (mut ar, now) = to_data();
        let first_res = ar.context().unwrap().connect_res.clone();

        let mut reconnect = match connect() {
            Event::ConnectReq(c) => c,
            _ => unreachable!(),
        };
        reconnect.ar.session_key = 5;
        let a = ar.on(Event::ConnectReq(reconnect), now);

        assert_eq!(a.len(), 3);
        assert!(matches!(
            a[0],
            Action::Notify {
                state: ArState::Idle,
                reason: Some(AbortReason::ControllerReconnect)
            }
        ));
        let new_res = match &a[1] {
            Action::Respond { status, blocks } if status.is_ok() => blocks.clone(),
            other => panic!("{other:?}"),
        };
        assert!(matches!(
            a[2],
            Action::Notify {
                state: ArState::Connected,
                reason: None
            }
        ));

        assert_eq!(ar.context().unwrap().params.session_key, 5);
        assert_eq!(ar.state(), ArState::Connected);

        // The response is byte-identical to the first one except for the
        // ARBlockRes session-key field (BlockHeader (6) + ar_type (2) + ar_uuid
        // (16) = offset 24, 2 bytes).
        assert_eq!(new_res[..24], first_res[..24]);
        assert_eq!(new_res[26..], first_res[26..]);
        assert_eq!(new_res[24..26], 5u16.to_be_bytes());
        assert_ne!(new_res[24..26], first_res[24..26]);
    }

    #[test]
    fn reconnect_with_new_ar_uuid_same_initiator_takes_over() {
        let (mut ar, now) = to_data();
        let new_ar_uuid = crate::rpc::Uuid([7; 16]);

        let mut reconnect = match connect() {
            Event::ConnectReq(c) => c,
            _ => unreachable!(),
        };
        reconnect.ar.ar_uuid = new_ar_uuid;
        let a = ar.on(Event::ConnectReq(reconnect), now);

        assert_eq!(a.len(), 3);
        assert!(matches!(
            a[0],
            Action::Notify {
                state: ArState::Idle,
                reason: Some(AbortReason::ControllerReconnect)
            }
        ));
        assert!(matches!(&a[1], Action::Respond { status, .. } if status.is_ok()));
        assert!(matches!(
            a[2],
            Action::Notify {
                state: ArState::Connected,
                reason: None
            }
        ));

        assert_eq!(ar.context().unwrap().params.ar_uuid, new_ar_uuid);
        assert_eq!(ar.state(), ArState::Connected);
    }

    #[test]
    fn data_connect_same_uuid_resends_cached_bytes() {
        let (mut ar, now) = to_data();
        let a = ar.on(connect(), now);
        assert_eq!(respond_ok(&a), &golden("connect_res")[BLOCKS..]);
        assert_eq!(ar.state(), ArState::Data);
    }

    #[test]
    fn data_write_stores_records_and_touches_no_timer() {
        let (mut ar, now) = to_data();
        let a = ar.on(write(), now);
        assert_eq!(respond_ok(&a), &golden("write_res")[BLOCKS..]);
        assert_eq!(ar.next_deadline(), None);
    }

    #[test]
    fn data_prm_end_is_idempotent() {
        let (mut ar, now) = to_data();
        let a = ar.on(prm_end(), now);
        assert_eq!(respond_ok(&a), &golden("prmend_res")[BLOCKS..]);
        assert_eq!(ar.state(), ArState::Data);
    }

    #[test]
    fn data_release_aborts() {
        let (mut ar, now) = to_data();
        let a = ar.on(Event::ReleaseReq(release_block()), now);
        assert!(
            matches!(&a[0], Action::Respond { status, blocks } if status.is_ok() && blocks[0..2] == [0x81, 0x14])
        );
        assert!(matches!(
            a[1],
            Action::Notify {
                state: ArState::Idle,
                reason: Some(AbortReason::ControllerRelease)
            }
        ));
        assert_eq!(ar.state(), ArState::Idle);
    }

    #[test]
    fn data_app_ready_rsp_is_ignored() {
        let (mut ar, now) = to_data();
        let a = ar.on(
            Event::AppReadyRsp {
                status: PnioStatus::OK,
            },
            now,
        );
        assert!(a.is_empty());
        assert_eq!(ar.state(), ArState::Data);
    }

    #[test]
    fn data_tick_is_noop() {
        let (mut ar, now) = to_data();
        let a = ar.on(Event::Tick, now + Duration::from_secs(100));
        assert!(a.is_empty());
        assert_eq!(ar.state(), ArState::Data);
    }

    // -----------------------------------------------------------------------
    // Fix round 3 (final review): Write/PrmEnd/Release must belong to the
    // established AR, records are capped, and the ApplicationReady call is built
    // from the context rather than the request.
    // -----------------------------------------------------------------------

    #[test]
    fn write_with_foreign_ar_uuid_is_refused() {
        let mut ar = ar();
        let now = t0();
        ar.on(connect(), now);
        let mut w = match write() {
            Event::WriteReq(w) => w,
            _ => unreachable!(),
        };
        for r in &mut w.records {
            r.ar_uuid = crate::rpc::Uuid([9; 16]);
        }
        let a = ar.on(Event::WriteReq(w), now);
        assert_eq!(a.len(), 1);
        assert!(
            matches!(&a[0], Action::Respond { status, blocks } if *status == PnioStatus::write_wrong_ar() && blocks.is_empty())
        );
        assert!(ar.context().unwrap().records.is_empty());
    }

    #[test]
    fn prm_end_with_foreign_ar_uuid_is_refused() {
        let mut ar = ar();
        let now = t0();
        ar.on(connect(), now);
        let mut p = match prm_end() {
            Event::PrmEndReq(p) => p,
            _ => unreachable!(),
        };
        p.ar_uuid = crate::rpc::Uuid([9; 16]);
        let a = ar.on(Event::PrmEndReq(p), now);
        assert_eq!(a.len(), 1);
        assert!(
            matches!(&a[0], Action::Respond { status, blocks } if *status == PnioStatus::control_wrong_state() && blocks.is_empty())
        );
        assert_eq!(ar.state(), ArState::Connected);
    }

    #[test]
    fn release_with_foreign_ar_uuid_is_refused() {
        let mut ar = ar();
        let now = t0();
        ar.on(connect(), now);
        let mut rel = release_block();
        rel.ar_uuid = crate::rpc::Uuid([9; 16]);
        let a = ar.on(Event::ReleaseReq(rel), now);
        assert_eq!(a.len(), 1);
        assert!(
            matches!(&a[0], Action::Respond { status, blocks } if *status == PnioStatus::control_wrong_state() && blocks.is_empty())
        );
        assert_eq!(ar.state(), ArState::Connected);
    }

    #[test]
    fn records_are_capped() {
        let mut ar = ar();
        let now = t0();
        ar.on(connect(), now);
        // 12 golden Writes x 5 records = 60 stored, under the 64 cap.
        for _ in 0..12 {
            let a = ar.on(write(), now);
            assert!(matches!(&a[0], Action::Respond { status, .. } if status.is_ok()));
        }
        assert_eq!(ar.context().unwrap().records.len(), 60);
        // The 13th Write would bring the total to 65, over the cap: refused, and
        // nothing from it is stored.
        let a = ar.on(write(), now);
        assert_eq!(a.len(), 1);
        assert!(
            matches!(&a[0], Action::Respond { status, blocks } if *status == PnioStatus::write_resource_unavailable() && blocks.is_empty())
        );
        assert_eq!(ar.context().unwrap().records.len(), 60);
    }

    #[test]
    fn app_ready_call_uses_context_not_request() {
        let mut ar = ar();
        let now = t0();
        ar.on(connect(), now);
        let mut p = match prm_end() {
            Event::PrmEndReq(p) => p,
            _ => unreachable!(),
        };
        p.session_key = 9; // right ar_uuid, wrong session_key
        let a = ar.on(Event::PrmEndReq(p), now);
        assert!(
            matches!(&a[1], Action::CallController { blocks } if blocks == &golden("appready_req")[BLOCKS..])
        );
    }
}