//! The instance-axis state machine (D9-2 REQ-INST-3, reworked by
//! INSTANCE-AXES releases#348) — the explicit, daemon-owned model behind
//! active / dormant (warm) / suspended (cold).
//!
//! CONTEXT §instance state + §dormant/suspended define the model this module
//! implements. **Two axes, not one** (REQ-INSTANCE-AXIS-SPLIT): whether a
//! controller is attached is the separate CONTROLLED axis (`info.json`
//! `controlled`, broker-stamped, gossiped) — attach and detach NEVER move the
//! instance axis, and this machine reads the controlled axis only through
//! auto-suspend. On the instance axis:
//!
//! - **→ active** only through an activation trigger (REQ-ACTIVATION-TRIGGERS):
//!   user input, a stealing message, an explicit wake ([`RestEvent::Wake`]), or
//!   coming warm with no active sibling ([`RestEvent::ComeWarm`]). Every edge
//!   into active takes a fresh ACTIVATION COUNTER — the highest seen for the id
//!   plus one (REQ-ACTIVATION-COUNTER) — which is how every other node orders it.
//! - **active → dormant** has exactly ONE cause: a sibling became active
//!   ([`RestEvent::SiblingActivated`]). Never a detach, never idleness.
//! - **→ suspended**: manual (`spt endpoint suspend` / shell
//!   `api owner-shutdown`), or auto-suspend under its `{mode, after}` setting
//!   (REQ-AUTO-SUSPEND-MODES). Suspending the ACTIVE instance VACATES active;
//!   nothing passes it on to a dormant sibling.
//!
//! The transition function is **pure** ([`transition`]): state + event +
//! the injected facts in, the new state out (`None` = no edge). Durability is
//! the perch's `info.json` (`rest_state`, `dormant_since_ms`, `activation`);
//! effects (the transition echo commune, the wake resurface + freshness pull —
//! REQ-INST-4) hang off the *edges* and are wired by the transition host,
//! not here. Keeping the table pure is what makes the CONTEXT state diagram
//! a unit test instead of a prose promise.
//!
//! ## What "suspended" does NOT do here
//!
//! Suspension's session mechanics (closing the harness seat, resume-on-wake)
//! ride the existing live-agent machinery; this module owns the **state
//! model** those mechanics consult. It answers WHICH INSTANCE a bare id
//! resolves to and whether a wake is owed first; it is never an input to a
//! delivery window. What a delivery window reads is the activity sentinel
//! (REQ-DELIVERY-WINDOW-IS-ACTIVITY-AXIS-ONLY, releases#341): this state
//! machine reads the sentinel only for auto-suspend's idle leg, and no
//! reading of this machine can decide a window.
// [impl->REQ-INST-3]
// [impl->REQ-INSTANCE-AXIS-SPLIT]

use std::path::Path;

use spt_store::autosuspend::{AutoSuspend, AutoSuspendMode};
use spt_store::info;
use spt_store::perch::{self, ParentHint};

/// The stable phrase that marks a rest event's "this id is not a locally hosted
/// perch" miss (REQ-REST-VERB-ROUTING). SINGLE-SOURCED: [`apply_event`] builds
/// its miss message FROM this const, and the CLI's A-3 bare-id remote-fallback
/// discriminates on it — so rewording the miss can never silently regress
/// bare-id routing to `WOKE_FAIL` (the drift class ADR-0034 Addendum 2 froze).
///
/// IN-PROCESS discriminant, not a wire surface: `spt` links the `spt-daemon`
/// lib and both the builder and the matcher compile from THIS const in one
/// binary — so single-source suffices; no frozen-compat freeze is required
/// (unlike [`crate::OP_NO_LONGER_HELD_MARKER`], which crosses the IPC boundary).
/// The drift-pin unit test asserts the built message contains it.
///
/// The phrase is OPERATOR language (F-1, REQ-PUBLIC-ERROR-SURFACES): it also
/// ships raw in the one line a user still sees — a qualified `id@node` rest
/// verb answered by a node that doesn't host the id (the D6 stale-row case) —
/// so it must name the situation, never store internals ("info.json").
pub const NOT_A_HOSTED_PERCH_MARKER: &str = "is not hosted on this node";

/// The resting states (CONTEXT §dormant/suspended): the registry's
/// `Status::Offline` is deliberately NOT here — offline means *node
/// unreachable*, a fact about the node observed by peers, never a state this
/// machine transitions an instance into.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RestState {
    /// The instance holding the id's attention. Active ⇒ warm and online.
    Active,
    /// Resting warm: the harness session stays running while a sibling holds
    /// attention.
    Dormant,
    /// Resting cold: the harness session is closed, resumed on wake.
    Suspended,
}

impl RestState {
    /// The durable `info.json` tag.
    pub fn as_tag(self) -> &'static str {
        match self {
            RestState::Active => "active",
            RestState::Dormant => "dormant",
            RestState::Suspended => "suspended",
        }
    }

    /// Parse a durable tag (`None` for an unknown/foreign value — a newer
    /// fleet's tag must degrade to "no resting record", never wedge a reader).
    pub fn from_tag(tag: &str) -> Option<RestState> {
        match tag {
            "active" => Some(RestState::Active),
            "dormant" => Some(RestState::Dormant),
            "suspended" => Some(RestState::Suspended),
            _ => None,
        }
    }
}

/// The events that can move an instance on the instance axis. Each names a
/// real observation point (the production feeds), not a poll. There is NO
/// detach event: the controlled axis never moves this one
/// (REQ-INSTANCE-AXIS-SPLIT).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RestEvent {
    /// A sibling of the same id became active — observed through the
    /// registry's activation order (REQ-ACTIVATION-COUNTER). The ONLY edge
    /// into dormant.
    SiblingActivated,
    /// Manual suspend (`spt endpoint suspend` / shell `api owner-shutdown`).
    Suspend,
    /// The pulse loop's auto-suspend check. Carries the clock and the
    /// unbroken-condition facts so the table stays pure.
    AutoSuspendTick { now_ms: u64, facts: SuspendFacts },
    /// An activation trigger that always lands ACTIVE: an explicit wake
    /// (`spt endpoint wake`, a shell wake-watcher), user input, or a stealing
    /// message (REQ-ACTIVATION-TRIGGERS 1, 2, 5).
    Wake,
    /// Coming warm — boot, or a controller attaching. Lands ACTIVE when no
    /// sibling is active (the vacancy rule, no time window), else DORMANT
    /// from suspended and no edge from dormant (REQ-ACTIVATION-TRIGGERS 4 +
    /// wake landing by cause).
    ComeWarm,
    /// A SAME-id message (a sibling's normal message) reaching this instance.
    /// It never steals; it only wakes a SUSPENDED instance, and wakes it
    /// DORMANT (wake landing by cause).
    SiblingMessage,
}

/// The auto-suspend condition's inputs, observed by the host at tick time
/// (REQ-AUTO-SUSPEND-MODES): the controlled axis and the activity sentinel,
/// each with the instant it last took its current value.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct SuspendFacts {
    /// A controller is attached (the controlled axis).
    pub controlled: bool,
    /// When `controlled` last changed. `None` (a pre-#348 record) ⇒ this leg
    /// contributes no anchor.
    pub controlled_since_ms: Option<u64>,
    /// The agent is idle (the activity sentinel, busy|idle).
    pub idle: bool,
    /// When the agent went idle. `None` ⇒ the instant is unknown and the tick
    /// never fires (fail toward warm, the cheap mistake).
    pub idle_since_ms: Option<u64>,
}

/// What the host knows about the id's OTHER instances when an event lands:
/// whether one is active, and the highest activation counter seen for the id
/// (the floor a new activation must exceed).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct SiblingView {
    pub active_sibling: bool,
    pub max_activation: u64,
}

/// The pure transition table. `dormant_since_ms` is the dormancy-onset anchor
/// (present iff `state` is dormant); `auto_suspend` is the RESOLVED setting
/// (`spt_store::autosuspend::effective`); `active_sibling` is whether another
/// instance of the id is active. Returns the new state on a real edge, `None`
/// when the event does not move this state (idempotent re-delivery of any
/// event is a no-op by construction — an event-feed replay can never
/// double-fire an edge's effects).
// [impl->REQ-INST-3]
// [impl->REQ-INSTANCE-AXIS-SPLIT]
// [impl->REQ-ACTIVATION-TRIGGERS]
pub fn transition(
    state: RestState,
    dormant_since_ms: Option<u64>,
    event: RestEvent,
    auto_suspend: AutoSuspend,
    active_sibling: bool,
) -> Option<RestState> {
    use RestState::{Active, Dormant, Suspended};
    match (state, event) {
        // The ONE edge into dormant: a sibling became active.
        (Active, RestEvent::SiblingActivated) => Some(Dormant),
        // Manual suspend from either warm state. From active it VACATES
        // active — nothing is passed on (the echo still fires, KH 3.3).
        (Active | Dormant, RestEvent::Suspend) => Some(Suspended),
        (Active | Dormant, RestEvent::AutoSuspendTick { now_ms, facts }) => {
            auto_suspend_due(state, dormant_since_ms, auto_suspend, now_ms, facts)
                .then_some(Suspended)
        }
        // Explicit wake / user input / stealing message: always active.
        (Dormant | Suspended, RestEvent::Wake) => Some(Active),
        // Coming warm: the vacancy rule decides.
        (Suspended, RestEvent::ComeWarm) => Some(if active_sibling { Dormant } else { Active }),
        (Dormant, RestEvent::ComeWarm) => (!active_sibling).then_some(Active),
        // A sibling's message wakes a suspended instance, never to active.
        (Suspended, RestEvent::SiblingMessage) => Some(Dormant),
        // Everything else: no edge. Notably — any trigger on an already
        // active instance, SiblingActivated on a resting one, Suspend on
        // suspended (idempotent), a tick on suspended.
        _ => None,
    }
}

/// Whether the auto-suspend condition has held UNBROKEN for the setting's
/// `after` (REQ-AUTO-SUSPEND-MODES): uncontrolled AND idle AND (dormant, under
/// `dormant-enable`). The timer anchor is the LATEST instant any leg took its
/// current value, so a break in any leg restarts it.
// [impl->REQ-AUTO-SUSPEND-MODES]
fn auto_suspend_due(
    state: RestState,
    dormant_since_ms: Option<u64>,
    setting: AutoSuspend,
    now_ms: u64,
    facts: SuspendFacts,
) -> bool {
    let dormant_leg = match setting.mode {
        AutoSuspendMode::Disable => return false,
        AutoSuspendMode::Enable => None,
        AutoSuspendMode::DormantEnable => {
            if state != RestState::Dormant {
                return false;
            }
            // A dormant record without its anchor (torn legacy state) never
            // auto-suspends — fail toward warm.
            let Some(since) = dormant_since_ms else {
                return false;
            };
            Some(since)
        }
    };
    if facts.controlled || !facts.idle {
        return false;
    }
    let Some(idle_since) = facts.idle_since_ms else {
        return false;
    };
    let anchor = idle_since
        .max(facts.controlled_since_ms.unwrap_or(0))
        .max(dormant_leg.unwrap_or(0));
    now_ms.saturating_sub(anchor) >= setting.after_ms
}

/// The pure CLASSIFIER for a message reaching a locally hosted instance at
/// ADMISSION (REQ-ACTIVATION-TRIGGERS 2 + wake landing by cause) — the ONE
/// place the stealing rule is read. Both admission points (local `spt send`,
/// WAN receive) call it and feed its answer to the transition host; neither
/// decides anything itself.
///
/// - busy-only (`active_only`) window ⇒ never a trigger;
/// - empty sender (shell → owner frames, anonymous re-spools) ⇒ never;
/// - a DIFFERENT id ⇒ [`RestEvent::Wake`] (a stealing message lands active);
/// - the SAME id (a sibling's normal message, a self-send) ⇒ never steals;
///   [`RestEvent::SiblingMessage`] only wakes a suspended instance DORMANT.
// [impl->REQ-ACTIVATION-TRIGGERS]
pub fn message_trigger(target: &str, sender: &str, window: &str) -> Option<RestEvent> {
    if window == spt_store::spool::WINDOW_ACTIVE_ONLY || sender.is_empty() {
        return None;
    }
    if sender == target {
        Some(RestEvent::SiblingMessage)
    } else {
        Some(RestEvent::Wake)
    }
}

/// The wake-edge freshness-pull marker file (under `identity/`). A
/// dormant→active transition drops it ([`request_freshness_pull`]); the peer
/// pump consumes it ([`take_freshness_pull`]) and forces a sync round NOW
/// instead of waiting out the anti-entropy cadence — the D9-1 seam the wake
/// edge feeds (a freshly-woken instance pulls the latest mind immediately,
/// CONTEXT §catch-up-on-activation). File-based because the transition can
/// happen in another process (`spt wake`) than the pump's daemon.
pub const PULL_MARKER_FILE: &str = "sync-pull-now";

/// Drop the freshness-pull marker (idempotent — a second wake before the
/// pump's next tick just re-requests the same pull).
// [impl->REQ-INST-3]
pub fn request_freshness_pull(marker_path: &Path) -> std::io::Result<()> {
    if let Some(parent) = marker_path.parent() {
        std::fs::create_dir_all(parent)?;
    }
    std::fs::write(marker_path, b"")
}

/// Consume the freshness-pull marker: `true` exactly once per drop (the
/// remove is the take — two pump ticks can never both claim one wake).
// [impl->REQ-INST-3]
pub fn take_freshness_pull(marker_path: &Path) -> bool {
    std::fs::remove_file(marker_path).is_ok()
}

/// What one applied event did: the edge taken and whether the transition
/// echo ran (it fires only on active → resting edges).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct EdgeReport {
    pub from: RestState,
    pub to: RestState,
    pub echo_fired: bool,
}

/// THE effective resting state of an instance (REQ-EFFECTIVE-INSTANCE-STATE,
/// ADR-0033): liveness discriminates warm from cold, the stored intent refines
/// only *within* warm, and absent intent NEVER defaults active. Every
/// rest-state reader routes through this one derivation — `advertised_status`
/// (via a RestState→Status map) and `apply_event`'s `from` both — so they can
/// never disagree. The NO_EDGE-on-a-suspended-endpoint bug was exactly that
/// disagreement: `apply_event` read the intent field alone (`unwrap_or(Active)`),
/// so a cold void perch looked Active, a `Wake` found it "already there", and
/// the endpoint could never be woken.
///
/// - **alive** (bound + online): stored intent refines — a void or `Active`
///   intent is Active, else the intent verbatim.
/// - **unbound** (warm skeleton): Dormant, regardless of stored intent.
/// - **cold** (offline / dead pid / corrupt): Suspended, regardless of intent.
// [impl->REQ-EFFECTIVE-INSTANCE-STATE]
pub fn effective_rest_state(alive: bool, unbound: bool, intent: Option<RestState>) -> RestState {
    if alive {
        match intent {
            Some(RestState::Active) | None => RestState::Active,
            Some(RestState::Dormant) => RestState::Dormant,
            Some(RestState::Suspended) => RestState::Suspended,
        }
    } else if unbound {
        RestState::Dormant
    } else {
        RestState::Suspended
    }
}

/// Apply one event against a perch's durable resting record — the transition
/// host. Reads the record (a recordless live perch starts `Active` — the
/// machine's production feeds only run for hosted live endpoints), resolves
/// the auto-suspend setting, runs the pure table against the injected
/// [`SiblingView`], and on a real edge:
///
/// 1. **active → (dormant|suspended)**: runs `on_rest_edge` — the transition
///    echo commune capturing the outgoing instance's final context delta,
///    BEFORE the state flip persists (the KH 3.3 ordering: echo before any
///    teardown the new state implies). An echo failure is loud but does NOT
///    wedge the instance active — the edge is a fact; the state persists and
///    the error surfaces after (the `signoff_with` posture). [impl->REQ-INST-4]
/// 2. persists the landing state ([`write_rest`] — the dormancy anchor moves
///    with it). An edge INTO active takes the next activation counter —
///    `max(own, highest seen for the id) + 1` — in the same write
///    (REQ-ACTIVATION-COUNTER);
/// 3. **→ active**: runs `on_wake` — the resurface-at-boundary +
///    freshness-pull effects (re-activation in place catches up immediately).
///
/// `Ok(None)` = no edge (the pure table's idempotence: replaying any event
/// against a state it does not move can never double-fire effects).
// [impl->REQ-INST-3]
// [impl->REQ-ACTIVATION-COUNTER]
pub fn apply_event(
    perch_path: &Path,
    event: RestEvent,
    node_auto_suspend: Option<AutoSuspend>,
    siblings: SiblingView,
    now_ms: u64,
    on_rest_edge: impl FnOnce() -> Result<(), String>,
    on_wake: impl FnOnce(),
) -> Result<Option<EdgeReport>, String> {
    apply_event_with_liveness(
        perch_path,
        event,
        node_auto_suspend,
        siblings,
        now_ms,
        on_rest_edge,
        on_wake,
        None,
    )
}

/// [`apply_event`] with an explicit liveness override (ADR-0041 decision 6,
/// REQ-ENDPOINT-CYCLE-HONEST): `alive_hint = Some(true)` makes the `from`
/// derivation treat the endpoint as ALIVE even when its perch status reads
/// offline — the SHUTDOWN verb passes broker-session truth here so the cycle
/// verbs share ONE liveness answer. The deployah wedge shape: a zombie hosted
/// session made `run` say ALREADY_LIVE while the offline perch made shutdown's
/// Suspend read `from=Suspended` → `NO_EDGE` (claiming it was already down) —
/// three answers about one endpoint. `None` keeps the perch-derived read.
// [impl->REQ-ENDPOINT-CYCLE-HONEST]
#[allow(clippy::too_many_arguments)] // apply_event's seams + the one override
pub fn apply_event_with_liveness(
    perch_path: &Path,
    event: RestEvent,
    node_auto_suspend: Option<AutoSuspend>,
    siblings: SiblingView,
    now_ms: u64,
    on_rest_edge: impl FnOnce() -> Result<(), String>,
    on_wake: impl FnOnce(),
    alive_hint: Option<bool>,
) -> Result<Option<EdgeReport>, String> {
    // The miss line ships to operators (F-1, REQ-PUBLIC-ERROR-SURFACES):
    // situation + next action, built FROM the marker const (the A-3 drift pin).
    // [impl->REQ-PUBLIC-ERROR-SURFACES]
    let rec = info::read_info(perch_path).ok_or_else(|| {
        format!(
            "this endpoint {NOT_A_HOSTED_PERCH_MARKER} — \
             `spt endpoint list` shows where it lives"
        )
    })?;
    // Derive `from` through THE shared effective-state fn, not the stored
    // intent alone — a liveness-blind read lies about cold perches (the field
    // NO_EDGE-on-suspended bug: a cold void perch read Active, so Wake no-edged
    // and the endpoint could never wake). REQ-EFFECTIVE-INSTANCE-STATE.
    // [impl->REQ-EFFECTIVE-INSTANCE-STATE]
    let alive = alive_hint.unwrap_or_else(|| spt_store::liveness::is_perch_alive(perch_path));
    let from = effective_rest_state(
        alive,
        spt_store::liveness::is_perch_unbound(perch_path),
        rec.rest_state.as_deref().and_then(RestState::from_tag),
    );
    let (setting, _) = spt_store::autosuspend::effective(node_auto_suspend, rec.auto_suspend);
    let Some(to) = transition(from, rec.dormant_since_ms, event, setting, siblings.active_sibling)
    else {
        return Ok(None);
    };

    // The transition echo (REQ-INST-4, KH 3.3): at the active → resting edge,
    // before the flip persists. Failure is captured, not short-circuited —
    // the state change is a fact about the instance, not about the echo.
    // [impl->REQ-INST-4]
    let mut echo_err: Option<String> = None;
    let echo_fired = from == RestState::Active && to != RestState::Active;
    if echo_fired {
        if let Err(e) = on_rest_edge() {
            echo_err = Some(e);
        }
    }

    if to == RestState::Active {
        let activation = next_activation(rec.activation, siblings.max_activation);
        info::set_rest_state_activating(perch_path, activation)
            .map_err(|e| format!("persist rest state: {e}"))?;
        on_wake();
    } else {
        write_rest(perch_path, to, now_ms).map_err(|e| format!("persist rest state: {e}"))?;
    }

    if let Some(e) = echo_err {
        return Err(format!("transition echo failed (state persisted): {e}"));
    }
    Ok(Some(EdgeReport {
        from,
        to,
        echo_fired,
    }))
}

/// Where a perch coming warm from nothing (a boot: a fresh life or a new
/// perch) lands, and the activation counter it carries — the table's
/// `ComeWarm` from cold (REQ-ACTIVATION-TRIGGERS 4): ACTIVE with the next
/// counter when no sibling is active, DORMANT (counter unchanged) otherwise.
// [impl->REQ-ACTIVATION-TRIGGERS]
pub fn come_warm_landing(siblings: SiblingView, own_activation: u64) -> (RestState, u64) {
    match transition(
        RestState::Suspended,
        None,
        RestEvent::ComeWarm,
        spt_store::autosuspend::DEFAULT_AUTO_SUSPEND,
        siblings.active_sibling,
    ) {
        Some(RestState::Active) => {
            (RestState::Active, next_activation(own_activation, siblings.max_activation))
        }
        _ => (RestState::Dormant, own_activation),
    }
}

/// Wall-clock epoch ms (the dormancy anchor's clock).
pub fn wall_now_ms() -> u64 {
    now_ms()
}

/// The counter an activation takes: the highest this instance has seen for
/// the id — its own last value or any sibling's — plus one
/// (REQ-ACTIVATION-COUNTER). Saturates rather than wrapping.
// [impl->REQ-ACTIVATION-COUNTER]
pub fn next_activation(own: u64, max_seen: u64) -> u64 {
    own.max(max_seen).saturating_add(1)
}

/// Observe the auto-suspend condition's facts for a perch
/// (REQ-AUTO-SUSPEND-MODES): the controlled axis off `info.json` (the broker's
/// stamp + its flip instant) and the activity sentinel off the perch
/// (`.idle` + the `.activity` stamp's instant).
// [impl->REQ-AUTO-SUSPEND-MODES]
pub fn suspend_facts(perch_path: &Path) -> SuspendFacts {
    let rec = info::read_info(perch_path);
    let (idle, idle_since_ms) = perch::read_activity_at(perch_path);
    SuspendFacts {
        controlled: rec.as_ref().is_some_and(|r| r.controlled),
        controlled_since_ms: rec.and_then(|r| r.controlled_changed_ms),
        idle,
        idle_since_ms,
    }
}

/// Whether a sibling's active claim outranks the local instance — the
/// activation ORDER (REQ-ACTIVATION-COUNTER): counter first, node id breaks
/// a tie. Strictly greater: a claim equal to our own rank is our own row.
// [impl->REQ-ACTIVATION-COUNTER]
pub fn sibling_outranks(claim: (u64, &str), local: (u64, &str)) -> bool {
    claim > local
}

/// The [`SiblingView`] for `id` from the registry snapshots: every OTHER
/// node's row for the id, across every visible subnet. Reads the snapshot
/// files (not the in-memory registry) because the transition host runs in
/// the CLI process too (`spt endpoint wake`, `api state busy`, `spt send`).
/// Stale-tolerant by contract: a missing snapshot reads as no siblings.
// [impl->REQ-ACTIVATION-COUNTER]
pub fn sibling_view(id: &str) -> SiblingView {
    sibling_view_from(
        &crate::presence::load_registry_snapshots(&crate::presence::registry_snapshot_dir()),
        id,
        &crate::presence::local_node_hex(),
    )
}

/// Pure core of [`sibling_view`].
// [impl->REQ-ACTIVATION-COUNTER]
pub fn sibling_view_from(
    regs: &std::collections::BTreeMap<String, spt_net::net::registry::SubnetRegistry>,
    id: &str,
    own_node: &str,
) -> SiblingView {
    let mut view = SiblingView::default();
    for reg in regs.values() {
        for inst in reg.instances(id).iter().filter(|i| i.node != own_node) {
            view.max_activation = view.max_activation.max(inst.activation);
            if inst.status == spt_net::net::registry::Status::Active {
                view.active_sibling = true;
            }
        }
    }
    view
}

/// Arm the endpoint's echo-gate sentinel — the **cross-process transition
/// echo trigger** (REQ-INST-4). The daemon's feed points (the attach arm's
/// detach, the registry arm's attention shift, `spt suspend`) hold no
/// manifest runtime, so they cannot run the bounded summarizer call
/// themselves; arming the gate hands the echo to the endpoint's **own**
/// pulse loop ([`spt_live::pulse::take_echo_gate`] — read-and-clear, fires
/// exactly once), which owns the runtime, history, and commune dir.
///
/// This is also the KH 7.4 posture **by construction**: the daemon never
/// hosts another agent's bounded LLM call from a shared loop — N agents'
/// transition echoes are N gate files consumed by N per-agent pulse loops,
/// so one slow summarizer cannot stall another agent's edge
/// (REQ-HAZARD-DAEMON-SCHED-NONBLOCKING stays unbound until the daemon
/// hosts per-agent runtimes — the same 7.4 fan-out seam the in-pump
/// reconcile turn waits on).
// [impl->REQ-INST-4]
pub fn arm_transition_echo(id: &str) {
    let gate = perch::resolve_edge_echo_file(id, ParentHint::Infer);
    if let Some(parent) = gate.parent() {
        let _ = std::fs::create_dir_all(parent);
    }
    let _ = std::fs::write(&gate, b"");
}

/// The wake-edge catch-up effects (best-effort, loud-on-surface like every
/// notif path): resurface undismissed notifs at the Wake boundary (the D8
/// deferral — real now that the edge exists) and drop the freshness-pull
/// marker the peer pump consumes.
// [impl->REQ-INST-3]
pub fn fire_wake_effects(id: &str) {
    if let Ok(store) = spt_store::notif::NotifStore::open() {
        let policy = crate::notif::NotifSurfacePolicy::load();
        let _ = crate::notif::resurface_at_boundary(
            &store,
            id,
            &policy,
            &perch::owlery_dir(),
            now_ms(),
            crate::notif::SUPPRESSION_WINDOW_MS,
        );
    }
    let marker = perch::identity_dir().join(PULL_MARKER_FILE);
    let _ = request_freshness_pull(&marker);
}

/// The daemon-side transition host for feed points that own no manifest
/// runtime (the attach arm, the registry arm's attention shift, the
/// `spt suspend`/`spt wake` CLI): [`apply_event`] with the echo seam bound
/// to the gate ([`arm_transition_echo`] — the endpoint's own pulse loop
/// fires the real commune) and the wake seam to [`fire_wake_effects`].
/// Resolves the canonical perch for `id`. A lifecycle host that *does* own
/// the runtime uses `BrainLifecycle::rest_event` instead (inline echo).
// [impl->REQ-INST-3]
pub fn daemon_rest_event(
    id: &str,
    event: RestEvent,
    node_auto_suspend: Option<AutoSuspend>,
) -> Result<Option<EdgeReport>, String> {
    daemon_rest_event_with_liveness(id, event, node_auto_suspend, None)
}

/// [`daemon_rest_event`] with the ONE-liveness-authority override (ADR-0041
/// decision 6): the SHUTDOWN verb passes `Some(true)` when the broker holds a
/// genuinely-live (non-zombie) session for `id`, so its Suspend edge answers
/// from the same truth `endpoint run`'s dup-guard consults — never a NO_EDGE
/// about an endpoint the run verb calls ALREADY_LIVE.
// [impl->REQ-ENDPOINT-CYCLE-HONEST]
pub fn daemon_rest_event_with_liveness(
    id: &str,
    event: RestEvent,
    node_auto_suspend: Option<AutoSuspend>,
    alive_hint: Option<bool>,
) -> Result<Option<EdgeReport>, String> {
    let perch_path = perch::resolve_perch_path(id, ParentHint::Infer);
    let report = apply_event_with_liveness(
        &perch_path,
        event,
        node_auto_suspend,
        sibling_view(id),
        now_ms(),
        || {
            arm_transition_echo(id);
            Ok(())
        },
        || fire_wake_effects(id),
        alive_hint,
    )?;
    if let Some(r) = report {
        cascade_shells_on_edge(id, &r);
        // A real rest edge changes the advertised state — wake the pump's
        // registry leg so peers render it within seconds, not a cadence
        // (M8 decision 15, REQ-CONV-2).
        // [impl->REQ-CONV-2]
        crate::registryhost::request_advertise_now();
    }
    Ok(report)
}

/// The shell leg of a rest edge (M5-D4a, REQ-SHELL-2): owner → suspended
/// closes its online shells (per-shell ephemeral/persistent divergence);
/// owner → active relaunches its `persistent` offline shells. Shared by both
/// transition hosts ([`daemon_rest_event`] + `BrainLifecycle::rest_event`),
/// best-effort beside the echo gate + wake effects — a shell failure is loud
/// on stderr but never wedges the (already persisted) rest state. An edge
/// whose echo errs skips the cascade with the early return; the D4b
/// reconciler is the catch-up for that rare gap.
// [impl->REQ-SHELL-2]
pub(crate) fn cascade_shells_on_edge(owner: &str, report: &EdgeReport) {
    let to_suspended = report.to == RestState::Suspended;
    let to_active = report.to == RestState::Active && report.from != RestState::Active;
    if !to_suspended && !to_active {
        return;
    }
    let adapters_dir = spt_store::perch::adapters_dir();
    let registered = spt_runtime::registry::registered(&adapters_dir);
    for failure in crate::shellhost::cascade_owner_edge(
        &perch::owlery_dir(),
        owner,
        &registered,
        &adapters_dir,
        to_suspended,
        to_active,
    ) {
        spt_proto::emit_line_err!("SHELL_CASCADE_WARN:{owner}: {failure}");
    }
}

fn now_ms() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_millis() as u64)
        .unwrap_or(0)
}

/// One instance's durable resting record.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RestRecord {
    pub state: RestState,
    /// Dormancy-onset anchor (present iff `state` is dormant).
    pub dormant_since_ms: Option<u64>,
    /// The endpoint leg of the auto-suspend knob chain.
    pub auto_suspend: Option<AutoSuspend>,
    /// The activation counter this instance last took.
    pub activation: u64,
}

/// Read a perch's resting record. `None` ⇒ no record (a pre-D9-2 / interim
/// perch, or an unreadable `info.json`) — callers fall back to the legacy
/// liveness-derived status, exactly what they did before this slice.
// [impl->REQ-INST-3]
pub fn read_rest(perch_path: &Path) -> Option<RestRecord> {
    let rec = info::read_info(perch_path)?;
    let state = RestState::from_tag(rec.rest_state.as_deref()?)?;
    Some(RestRecord {
        state,
        dormant_since_ms: rec.dormant_since_ms,
        auto_suspend: rec.auto_suspend,
        activation: rec.activation,
    })
}

/// Persist a transition's landing state. Stamps the dormancy anchor when the
/// new state is dormant (the auto-suspend clock starts at the edge) and
/// clears it otherwise — one atomic write, so no reader ever sees a dormant
/// record without its anchor.
// [impl->REQ-INST-3]
pub fn write_rest(perch_path: &Path, state: RestState, now_ms: u64) -> std::io::Result<()> {
    let anchor = (state == RestState::Dormant).then_some(now_ms);
    info::set_rest_state(perch_path, state.as_tag(), anchor)
}

/// The routing decision for a BARE-id rest verb (A-3, REQ-REST-VERB-ROUTING):
/// where the CLI's bare-id assembly lands after the local try and (on the
/// hosted-perch miss) the remote goal-satisfaction selection. RENDERING — the
/// F-1 operator strings, node qualifiers, exit codes — stays at the caller;
/// this is the DECISION only.
#[derive(Debug)]
pub enum RestRoute {
    /// The local daemon handled it (an edge, or the idempotent no-edge `None`).
    Local(Option<EdgeReport>),
    /// A genuine local failure — not the hosted-perch miss, or the caller
    /// disallowed remote fallback (e.g. shutdown's own-endpoint verb).
    LocalErr(String),
    /// Exactly one actionable instance — route to it via the qualified WAN arm.
    WanAct { node: String },
    /// Goal already satisfied at these node(s) — advisory no-op (gossiped
    /// status may be stale; the qualified `id@node` path stays the override).
    NoOp { nodes: Vec<String> },
    /// Several actionable instances — refuse; the caller renders the
    /// copy-pasteable `id@node` candidates.
    Ambiguous { nodes: Vec<String> },
    /// No instance in any visible subnet.
    NotFound,
    /// The event has no remote arm (a daemon-internal event).
    NoRemoteArm,
}

/// Drive a bare-id rest verb's PRODUCTION routing assembly: try the local
/// daemon edge; on the SINGLE-SOURCED hosted-perch miss (and only when the
/// caller allows fallback) select across the injected registry candidates by
/// the verb's goal-satisfaction rule (wake = ∃-Active, suspend = ∀-Suspended —
/// `select_rest_target`). I/O is INJECTED (`try_local`, `load_candidates`) so
/// the assembly itself is lib-testable and the two-host ladder drives THIS fn
/// — the production glue — not a test-side reimplementation. `load_candidates`
/// runs ONLY on the fallback path (the hot local verb never touches the
/// snapshot dir), preserving the pre-lift I/O profile exactly.
// [impl->REQ-REST-VERB-ROUTING]
pub fn route_rest_event(
    event: RestEvent,
    allow_remote_fallback: bool,
    try_local: impl FnOnce() -> Result<Option<EdgeReport>, String>,
    load_candidates: impl FnOnce() -> Vec<(String, spt_net::net::registry::Status)>,
) -> RestRoute {
    use spt_net::net::registry::{select_rest_target, GoalKind, RestGoal, RestTarget, Status};
    match try_local() {
        Ok(report) => RestRoute::Local(report),
        Err(e) if allow_remote_fallback && e.contains(NOT_A_HOSTED_PERCH_MARKER) => {
            let goal = match event {
                RestEvent::Wake => RestGoal { target: Status::Active, kind: GoalKind::Exists },
                RestEvent::Suspend => {
                    RestGoal { target: Status::Suspended, kind: GoalKind::Forall }
                }
                _ => return RestRoute::NoRemoteArm,
            };
            match select_rest_target(&load_candidates(), goal) {
                RestTarget::Act(node) => RestRoute::WanAct { node },
                RestTarget::NoOp(nodes) => RestRoute::NoOp { nodes },
                RestTarget::Ambiguous(nodes) => RestRoute::Ambiguous { nodes },
                RestTarget::NotFound => RestRoute::NotFound,
            }
        }
        Err(e) => RestRoute::LocalErr(e),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use spt_store::info::{write_info, InfoJson};

    // [unit->REQ-REST-VERB-ROUTING] the lifted routing assembly's contract table
    // (seam #2 — the ladder drives THIS fn, so its arms are pinned here):
    // local-ok passes through; ONLY the marker-miss (with fallback allowed)
    // enters remote selection; a non-marker error and a fallback-disallowed miss
    // are LocalErr; a daemon-internal event on the fallback path is NoRemoteArm;
    // the selector's four verdicts map 1:1. `load_candidates` must NOT run on
    // the local paths (the hot-verb I/O profile is part of the contract).
    #[test]
    fn route_rest_event_contract_table() {
        use spt_net::net::registry::Status;
        let miss = || Err(format!("info.json absent — {NOT_A_HOSTED_PERCH_MARKER}"));
        let no_candidates_expected = || -> Vec<(String, Status)> {
            panic!("load_candidates must not run on a local path")
        };

        // Local edge + local no-edge pass through, candidates never loaded.
        assert!(matches!(
            route_rest_event(RestEvent::Wake, true, || Ok(None), no_candidates_expected),
            RestRoute::Local(None)
        ));
        // Non-marker local error → LocalErr (no fallback), candidates never loaded.
        assert!(matches!(
            route_rest_event(
                RestEvent::Wake,
                true,
                || Err("registry locked".into()),
                no_candidates_expected
            ),
            RestRoute::LocalErr(e) if e == "registry locked"
        ));
        // Marker-miss with fallback DISALLOWED → LocalErr (shutdown's own-verb rule).
        assert!(matches!(
            route_rest_event(RestEvent::Wake, false, miss, no_candidates_expected),
            RestRoute::LocalErr(_)
        ));
        // Daemon-internal event on the fallback path → NoRemoteArm (before load).
        assert!(matches!(
            route_rest_event(
                RestEvent::AutoSuspendTick { now_ms: 0, facts: SuspendFacts::default() },
                true,
                miss,
                no_candidates_expected
            ),
            RestRoute::NoRemoteArm
        ));

        // Selector verdicts map 1:1 on the marker-miss fallback path.
        let one_suspended = vec![("nodeb".to_string(), Status::Suspended)];
        assert!(matches!(
            route_rest_event(RestEvent::Wake, true, miss, || one_suspended.clone()),
            RestRoute::WanAct { node } if node == "nodeb"
        ));
        let one_active = vec![("nodeb".to_string(), Status::Active)];
        assert!(matches!(
            route_rest_event(RestEvent::Wake, true, miss, || one_active.clone()),
            RestRoute::NoOp { nodes } if nodes == vec!["nodeb".to_string()]
        ));
        let two_suspended = vec![
            ("nodeb".to_string(), Status::Suspended),
            ("nodec".to_string(), Status::Suspended),
        ];
        assert!(matches!(
            route_rest_event(RestEvent::Wake, true, miss, || two_suspended.clone()),
            RestRoute::Ambiguous { nodes } if nodes.len() == 2
        ));
        assert!(matches!(
            route_rest_event(RestEvent::Wake, true, miss, Vec::new),
            RestRoute::NotFound
        ));
        // Suspend's ∀-goal: mixed one-active → Act on it.
        let mixed = vec![
            ("nodeb".to_string(), Status::Suspended),
            ("nodec".to_string(), Status::Active),
        ];
        assert!(matches!(
            route_rest_event(RestEvent::Suspend, true, miss, || mixed.clone()),
            RestRoute::WanAct { node } if node == "nodec"
        ));
    }

    // [unit->REQ-REST-VERB-ROUTING] DRIFT-PIN: apply_event builds its
    // not-a-hosted-perch miss message FROM the shared marker const — the same
    // const the CLI's A-3 bare-id remote-fallback discriminates on. Rewording
    // the miss without this pin would silently regress bare-id routing to
    // WOKE_FAIL with every other test still green (ADR-0034 Addendum 2 drift
    // class). The matcher-uses-const half is enforced at compile time in cli.rs.
    #[test]
    fn apply_event_miss_message_is_single_sourced() {
        let err = apply_event(
            std::path::Path::new("no-such-perch-dir-drift-pin"),
            RestEvent::Wake,
            None,
            SiblingView::default(),
            0,
            || Ok(()),
            || {},
        )
        .expect_err("a missing perch must miss read_info");
        assert!(
            err.contains(NOT_A_HOSTED_PERCH_MARKER),
            "miss message must be built from NOT_A_HOSTED_PERCH_MARKER, got: {err}"
        );
    }

    // [unit->REQ-INST-3] [unit->REQ-INSTANCE-AXIS-SPLIT] THE transition table —
    // the INSTANCE-AXES model as assertions (releases#348). REPINNED from the
    // D9-2 table: detach no longer exists as an event (the controlled axis
    // never moves this one), and the ONLY edge into dormant is a sibling
    // becoming active. Suspend from either warm state (from active it
    // vacates); explicit wake from either resting state; everything else a
    // no-edge, so a replayed feed can never double-fire an edge.
    #[test]
    fn transition_table_matches_the_context_model() {
        use RestEvent::*;
        use RestState::*;
        let d = spt_store::autosuspend::DEFAULT_AUTO_SUSPEND;

        // The ONE edge into dormant.
        assert_eq!(transition(Active, None, SiblingActivated, d, true), Some(Dormant));
        // Manual suspend from active (vacates) AND from dormant.
        assert_eq!(transition(Active, None, Suspend, d, false), Some(Suspended));
        assert_eq!(transition(Dormant, Some(5), Suspend, d, true), Some(Suspended));
        // Explicit wake / user input / stealing message: active from either
        // resting state, whether or not a sibling is active.
        assert_eq!(transition(Dormant, Some(5), Wake, d, true), Some(Active));
        assert_eq!(transition(Suspended, None, Wake, d, true), Some(Active));
        assert_eq!(transition(Suspended, None, Wake, d, false), Some(Active));

        // No-edges.
        assert_eq!(transition(Active, None, Wake, d, false), None, "already active");
        assert_eq!(transition(Dormant, Some(5), SiblingActivated, d, true), None);
        assert_eq!(transition(Suspended, None, SiblingActivated, d, true), None);
        assert_eq!(transition(Suspended, None, Suspend, d, false), None, "idempotent");
        // Idleness is not an edge into dormant: an idle, uncontrolled ACTIVE
        // instance under the default mode stays active however long it waits.
        let idle = SuspendFacts {
            controlled: false,
            controlled_since_ms: Some(0),
            idle: true,
            idle_since_ms: Some(0),
        };
        assert_eq!(
            transition(Active, None, AutoSuspendTick { now_ms: u64::MAX, facts: idle }, d, true),
            None,
            "dormant-enable never suspends the active instance, and nothing makes it dormant"
        );
    }

    // [unit->REQ-ACTIVATION-TRIGGERS] coming warm follows the VACANCY rule,
    // with no time window: from suspended it lands active when no sibling is
    // active and dormant when one is (attach to a suspended instance); from
    // dormant it takes active only during a vacancy. A sibling's (same-id)
    // message only ever wakes a SUSPENDED instance, and wakes it dormant.
    #[test]
    fn come_warm_and_sibling_message_follow_the_vacancy_rule() {
        use RestEvent::*;
        use RestState::*;
        let d = spt_store::autosuspend::DEFAULT_AUTO_SUSPEND;

        assert_eq!(transition(Suspended, None, ComeWarm, d, false), Some(Active));
        assert_eq!(transition(Suspended, None, ComeWarm, d, true), Some(Dormant));
        assert_eq!(transition(Dormant, Some(1), ComeWarm, d, false), Some(Active));
        assert_eq!(transition(Dormant, Some(1), ComeWarm, d, true), None);
        assert_eq!(transition(Active, None, ComeWarm, d, false), None);

        assert_eq!(transition(Suspended, None, SiblingMessage, d, true), Some(Dormant));
        assert_eq!(transition(Suspended, None, SiblingMessage, d, false), Some(Dormant));
        assert_eq!(transition(Dormant, Some(1), SiblingMessage, d, false), None);
        assert_eq!(transition(Active, None, SiblingMessage, d, true), None);

        // The boot landing is the same rule, carrying the next counter.
        let vacant = SiblingView { active_sibling: false, max_activation: 7 };
        assert_eq!(come_warm_landing(vacant, 3), (Active, 8));
        let taken = SiblingView { active_sibling: true, max_activation: 7 };
        assert_eq!(come_warm_landing(taken, 3), (Dormant, 3));
    }

    // [unit->REQ-ACTIVATION-TRIGGERS] the ONE stealing-message classifier:
    // busy-only never; empty sender (shell → owner) never; a different id
    // steals (Wake); the same id never steals (SiblingMessage). The
    // admission points only feed its answer.
    #[test]
    fn message_trigger_classifies_by_window_and_sender_id() {
        use spt_store::spool::{WINDOW_ACTIVE_ONLY, WINDOW_DEFAULT, WINDOW_IDLE_ONLY};
        assert_eq!(message_trigger("ling", "doyle", WINDOW_DEFAULT), Some(RestEvent::Wake));
        assert_eq!(message_trigger("ling", "doyle", WINDOW_IDLE_ONLY), Some(RestEvent::Wake));
        assert_eq!(message_trigger("ling", "doyle", WINDOW_ACTIVE_ONLY), None, "busy-only");
        assert_eq!(message_trigger("ling", "", WINDOW_DEFAULT), None, "shell frames");
        assert_eq!(
            message_trigger("ling", "ling", WINDOW_DEFAULT),
            Some(RestEvent::SiblingMessage),
            "same id never steals"
        );
        assert_eq!(message_trigger("ling", "ling", WINDOW_ACTIVE_ONLY), None);
    }

    // [unit->REQ-ACTIVATION-COUNTER] the counter: an activation takes the
    // highest seen (own or sibling) + 1; the order is (counter, node) with the
    // node id breaking a tie; strictly greater outranks (an equal rank is our
    // own row). The sibling view reads only OTHER nodes' rows, and an N-1 row
    // (no field) reads as 0.
    #[test]
    fn activation_counter_orders_by_counter_then_node() {
        use spt_net::net::registry::{Status, SubnetRegistry};
        assert_eq!(next_activation(0, 0), 1);
        assert_eq!(next_activation(4, 9), 10);
        assert_eq!(next_activation(9, 4), 10);
        assert_eq!(next_activation(u64::MAX, 0), u64::MAX, "saturates");

        assert!(sibling_outranks((5, "aa"), (4, "zz")), "counter first");
        assert!(sibling_outranks((5, "bb"), (5, "aa")), "tie → node id");
        assert!(!sibling_outranks((5, "aa"), (5, "bb")));
        assert!(!sibling_outranks((5, "aa"), (5, "aa")), "own rank never outranks");

        let row = |node: &str, status: Status, activation: u64| {
            let mut inst: spt_net::net::registry::Instance = serde_json::from_str(&format!(
                r#"{{"node":"{node}","status":"Active","epoch":1}}"#
            ))
            .expect("an N-1 row without the field parses");
            assert_eq!(inst.activation, 0, "absent field reads as 0");
            inst.status = status;
            inst.activation = activation;
            inst
        };
        let mut reg = SubnetRegistry::default();
        reg.merge_instance("ling", row("own", Status::Active, 50));
        reg.merge_instance("ling", row("n2", Status::Dormant, 7));
        reg.merge_instance("ling", row("n3", Status::Active, 3));
        let regs = std::collections::BTreeMap::from([("home".to_string(), reg)]);
        assert_eq!(
            sibling_view_from(&regs, "ling", "own"),
            SiblingView { active_sibling: true, max_activation: 7 },
            "own row excluded; max over siblings; any active sibling counts"
        );
        assert_eq!(sibling_view_from(&regs, "other", "own"), SiblingView::default());
    }

    // [unit->REQ-AUTO-SUSPEND-MODES] the three modes and the UNBROKEN
    // condition. disable: never. dormant-enable: dormant + uncontrolled +
    // idle, timed from the LATEST of dormancy onset, uncontrolled-since and
    // idle-since. enable: the active instance too (vacating active). Any leg
    // broken ⇒ no suspend; an unknown idle instant fails warm.
    #[test]
    fn auto_suspend_modes_and_the_unbroken_condition() {
        use spt_store::autosuspend::{AutoSuspend, AutoSuspendMode::*};
        use RestState::*;
        let tick = |now_ms, facts| RestEvent::AutoSuspendTick { now_ms, facts };
        let facts = |controlled, csince, idle, isince| SuspendFacts {
            controlled,
            controlled_since_ms: csince,
            idle,
            idle_since_ms: isince,
        };
        let de = AutoSuspend { mode: DormantEnable, after_ms: 500 };
        let en = AutoSuspend { mode: Enable, after_ms: 500 };
        let off = AutoSuspend { mode: Disable, after_ms: 500 };
        let quiet = facts(false, Some(100), true, Some(200));

        // dormant-enable: anchor = max(dormant 1000, uncontrolled 100, idle 200).
        assert_eq!(transition(Dormant, Some(1_000), tick(1_499, quiet), de, true), None);
        assert_eq!(transition(Dormant, Some(1_000), tick(1_500, quiet), de, true), Some(Suspended));
        // The latest leg sets the anchor: idle only since 1200.
        let late_idle = facts(false, Some(100), true, Some(1_200));
        assert_eq!(transition(Dormant, Some(1_000), tick(1_699, late_idle), de, true), None);
        assert_eq!(
            transition(Dormant, Some(1_000), tick(1_700, late_idle), de, true),
            Some(Suspended)
        );
        // Uncontrolled only since 1300 (a detach) restarts the timer.
        let late_detach = facts(false, Some(1_300), true, Some(200));
        assert_eq!(transition(Dormant, Some(1_000), tick(1_799, late_detach), de, true), None);
        // Any leg broken ⇒ never.
        let driven = facts(true, Some(100), true, Some(200));
        let busy = facts(false, Some(100), false, None);
        assert_eq!(transition(Dormant, Some(0), tick(u64::MAX, driven), de, true), None);
        assert_eq!(transition(Dormant, Some(0), tick(u64::MAX, busy), de, true), None);
        // Unknown idle instant / missing dormancy anchor ⇒ fail warm.
        let unknown = facts(false, None, true, None);
        assert_eq!(transition(Dormant, Some(0), tick(u64::MAX, unknown), de, true), None);
        assert_eq!(transition(Dormant, None, tick(u64::MAX, quiet), de, true), None);
        // A pre-#348 record (no controlled instant) contributes no anchor.
        let legacy = facts(false, None, true, Some(200));
        assert_eq!(transition(Dormant, Some(0), tick(700, legacy), de, true), Some(Suspended));
        // dormant-enable never touches the active instance.
        assert_eq!(transition(Active, None, tick(u64::MAX, quiet), de, false), None);

        // enable: the active instance too, from the same unbroken condition.
        assert_eq!(transition(Active, None, tick(699, quiet), en, false), None);
        assert_eq!(transition(Active, None, tick(700, quiet), en, false), Some(Suspended));
        assert_eq!(transition(Dormant, Some(0), tick(700, quiet), en, true), Some(Suspended));
        // disable: never.
        assert_eq!(transition(Dormant, Some(0), tick(u64::MAX, quiet), off, true), None);
        assert_eq!(transition(Active, None, tick(u64::MAX, quiet), off, false), None);
        // Suspended: the tick moves nothing.
        assert_eq!(transition(Suspended, None, tick(u64::MAX, quiet), en, false), None);
    }

    // [unit->REQ-INST-4] the transition echo fires EXACTLY ONCE per
    // active→resting edge: the first detach runs the echo before the flip
    // persists; replaying the same event is a no-edge and can never re-fire
    // it; the dormant→suspended edge (already resting) fires no echo; and
    // the wake edge runs the wake effects instead.
    #[test]
    fn apply_event_fires_echo_once_per_rest_edge_and_wake_on_wake() {
        use spt_store::info::set_status;
        use spt_store::liveness::STATUS_ONLINE;

        let d = tempfile::tempdir().unwrap();
        write_info(
            d.path(),
            &InfoJson::new("ling", "t", 4242, "sid", "live_agent"),
        )
        .unwrap();
        // A recordless LIVE perch: online, so the liveness-aware `from`
        // derivation reads Active (REQ-EFFECTIVE-INSTANCE-STATE).
        set_status(d.path(), STATUS_ONLINE).unwrap();

        let mut echoes = 0;
        let mut wakes = 0;

        // Recordless live perch starts Active: detach = the rest edge.
        let report = apply_event(
            d.path(),
            RestEvent::SiblingActivated,
            None,
            SiblingView::default(),
            1_000,
            || {
                echoes += 1;
                Ok(())
            },
            || wakes += 1,
        )
        .expect("apply ok")
        .expect("a real edge");
        assert_eq!(
            (report.from, report.to),
            (RestState::Active, RestState::Dormant)
        );
        assert!(
            report.echo_fired,
            "active→dormant fires the transition echo"
        );
        assert_eq!((echoes, wakes), (1, 0));
        assert_eq!(
            read_rest(d.path()).unwrap().state,
            RestState::Dormant,
            "flip persisted"
        );

        // Replay: no edge, no second echo — idempotent by the pure table.
        let replay = apply_event(
            d.path(),
            RestEvent::SiblingActivated,
            None,
            SiblingView::default(),
            1_100,
            || {
                echoes += 1;
                Ok(())
            },
            || wakes += 1,
        )
        .expect("apply ok");
        assert_eq!(replay, None, "replayed event is a no-edge");
        assert_eq!(echoes, 1, "echo fired exactly once per edge");

        // dormant → suspended: already resting — NO echo (the echo captures
        // the *outgoing active* instance's delta; it already ran).
        let report = apply_event(
            d.path(),
            RestEvent::Suspend,
            None,
            SiblingView::default(),
            1_200,
            || {
                echoes += 1;
                Ok(())
            },
            || wakes += 1,
        )
        .expect("apply ok")
        .expect("edge");
        assert_eq!(report.to, RestState::Suspended);
        assert!(!report.echo_fired, "resting→resting fires no echo");
        assert_eq!(echoes, 1);

        // wake: the wake effects fire, no echo.
        let report = apply_event(
            d.path(),
            RestEvent::Wake,
            None,
            SiblingView::default(),
            1_300,
            || {
                echoes += 1;
                Ok(())
            },
            || wakes += 1,
        )
        .expect("apply ok")
        .expect("edge");
        assert_eq!(
            (report.from, report.to),
            (RestState::Suspended, RestState::Active)
        );
        assert_eq!(
            (echoes, wakes),
            (1, 1),
            "wake runs wake effects, not the echo"
        );
        assert_eq!(read_rest(d.path()).unwrap().state, RestState::Active);
    }

    // [unit->REQ-EFFECTIVE-INSTANCE-STATE] the ONE shared derivation as a
    // table (ADR-0033): liveness discriminates warm/cold, stored intent refines
    // only within warm, absent intent NEVER defaults active. This is the exact
    // mapping `advertised_status` and `apply_event`'s `from` both route through.
    #[test]
    fn effective_rest_state_table() {
        use RestState::*;

        // alive: intent refines — void/Active ⇒ Active, else verbatim.
        assert_eq!(effective_rest_state(true, false, None), Active);
        assert_eq!(effective_rest_state(true, false, Some(Active)), Active);
        assert_eq!(effective_rest_state(true, false, Some(Dormant)), Dormant);
        assert_eq!(effective_rest_state(true, false, Some(Suspended)), Suspended);

        // unbound (warm skeleton): Dormant for EVERY intent, incl. void.
        for intent in [None, Some(Active), Some(Dormant), Some(Suspended)] {
            assert_eq!(
                effective_rest_state(false, true, intent),
                Dormant,
                "unbound is warm ⇒ Dormant regardless of stored intent"
            );
        }

        // cold (not alive, not unbound): Suspended for EVERY intent — a void
        // cold perch is Suspended, NOT the old liveness-blind Active.
        for intent in [None, Some(Active), Some(Dormant), Some(Suspended)] {
            assert_eq!(
                effective_rest_state(false, false, intent),
                Suspended,
                "cold ⇒ Suspended regardless of stored intent"
            );
        }
    }

    // [unit->REQ-EFFECTIVE-INSTANCE-STATE] the field bug, in the wire: a cold
    // (offline) perch with NO resting intent must derive `from = Suspended`, so
    // a Wake is a real Suspended→Active edge — NOT the old `unwrap_or(Active)`
    // that made a definitely-suspended endpoint look Active, no-edged the Wake
    // (Ok(None)/NO_EDGE), and stranded the endpoint unwakeable.
    #[test]
    fn apply_event_wakes_a_cold_void_perch() {
        use spt_store::info::set_status;
        use spt_store::liveness::STATUS_OFFLINE;

        let d = tempfile::tempdir().unwrap();
        write_info(
            d.path(),
            &InfoJson::new("ling", "t", 4242, "sid", "live_agent"),
        )
        .unwrap();
        // Cold: offline status, no resting record written.
        set_status(d.path(), STATUS_OFFLINE).unwrap();

        let mut woke = false;
        let report = apply_event(
            d.path(),
            RestEvent::Wake,
            None,
            SiblingView::default(),
            1_000,
            || Ok(()),
            || woke = true,
        )
        .expect("apply ok")
        .expect("a real edge — a cold perch is Suspended, so Wake wakes it");
        assert_eq!(
            (report.from, report.to),
            (RestState::Suspended, RestState::Active),
            "cold void ⇒ from=Suspended, Wake is a real edge"
        );
        assert!(!report.echo_fired, "wake fires no transition echo");
        assert!(woke, "the wake seam ran");
        assert_eq!(read_rest(d.path()).unwrap().state, RestState::Active);
    }

    // [unit->REQ-ENDPOINT-CYCLE-HONEST] ADR-0041 decision 6 — ONE liveness
    // authority: the deployah wedge's shutdown lie. An OFFLINE-stamped perch
    // whose endpoint still holds a live broker session used to make Suspend
    // read `from=Suspended` → NO_EDGE ("already down") while `run` said
    // ALREADY_LIVE about the same endpoint. With the broker-truth liveness
    // hint the Suspend is a REAL Active→Suspended edge; without a live
    // session (hint None) the perch-derived idempotence is unchanged.
    #[test]
    fn suspend_with_broker_liveness_hint_edges_instead_of_no_edge() {
        use spt_store::info::set_status;
        use spt_store::liveness::STATUS_OFFLINE;

        let d = tempfile::tempdir().unwrap();
        write_info(
            d.path(),
            &InfoJson::new("ling", "t", 4242, "sid", "live_agent"),
        )
        .unwrap();
        // The wedge shape: perch stamped OFFLINE (endpoint stop) while the
        // broker still hosts a live session for the id.
        set_status(d.path(), STATUS_OFFLINE).unwrap();

        // WITHOUT the hint: perch-derived cold ⇒ Suspended ⇒ Suspend NO_EDGEs
        // (the old three-way lie's shutdown leg — and the still-correct
        // idempotent answer when no live session exists).
        let report = apply_event(d.path(), RestEvent::Suspend, None, SiblingView::default(), 1_000, || Ok(()), || {})
            .expect("apply ok");
        assert!(report.is_none(), "no hint + cold perch ⇒ idempotent NO_EDGE");

        // WITH the broker-truth hint (a live non-zombie session exists): the
        // Suspend is a REAL edge — shutdown answers from the same authority
        // as the run verb's dup-guard.
        let report = apply_event_with_liveness(
            d.path(),
            RestEvent::Suspend,
            None,
            SiblingView::default(),
            1_000,
            || Ok(()),
            || {},
            Some(true),
        )
        .expect("apply ok")
        .expect("a live-session hint makes Suspend a real edge, never NO_EDGE");
        assert_eq!(
            (report.from, report.to),
            (RestState::Active, RestState::Suspended),
            "broker-truth alive ⇒ from=Active, Suspend edges"
        );
        assert_eq!(read_rest(d.path()).unwrap().state, RestState::Suspended);
    }

    // [unit->REQ-EFFECTIVE-INSTANCE-STATE] the mirror-image of the field bug: a
    // cold void perch is already Suspended, so a Suspend event is a no-edge
    // (Ok(None)) and fires NO echo. The old `unwrap_or(Active)` faked an
    // Active→Suspended edge on a dead driver and fired a spurious transition
    // echo (a bounded LLM call) for an instance that was never active.
    #[test]
    fn apply_event_no_spurious_echo_on_cold_void_suspend() {
        use spt_store::info::set_status;
        use spt_store::liveness::STATUS_OFFLINE;

        let d = tempfile::tempdir().unwrap();
        write_info(
            d.path(),
            &InfoJson::new("ling", "t", 4242, "sid", "live_agent"),
        )
        .unwrap();
        set_status(d.path(), STATUS_OFFLINE).unwrap();

        let mut echoed = false;
        let out = apply_event(
            d.path(),
            RestEvent::Suspend,
            None,
            SiblingView::default(),
            1_000,
            || {
                echoed = true;
                Ok(())
            },
            || {},
        )
        .expect("apply ok");
        assert_eq!(
            out, None,
            "cold void is already Suspended — Suspend is a no-edge"
        );
        assert!(
            !echoed,
            "no fake Active→Suspended edge ⇒ no spurious echo on a dead driver"
        );
    }

    // [unit->REQ-INST-4] a failing echo is LOUD but does not wedge the
    // instance active: the state persists (the driver is gone — that is a
    // fact), the error surfaces after (the signoff_with posture).
    #[test]
    fn apply_event_echo_failure_is_loud_but_state_persists() {
        use spt_store::info::set_status;
        use spt_store::liveness::STATUS_ONLINE;

        let d = tempfile::tempdir().unwrap();
        write_info(
            d.path(),
            &InfoJson::new("ling", "t", 4242, "sid", "live_agent"),
        )
        .unwrap();
        // Live perch (online) ⇒ recordless `from` derives Active.
        set_status(d.path(), STATUS_ONLINE).unwrap();

        let err = apply_event(
            d.path(),
            RestEvent::SiblingActivated,
            None,
            SiblingView::default(),
            1_000,
            || Err("summarizer exploded".to_string()),
            || {},
        )
        .expect_err("echo failure surfaces");
        assert!(err.contains("state persisted"), "loud and honest: {err}");
        assert_eq!(
            read_rest(d.path()).unwrap().state,
            RestState::Dormant,
            "the rest edge is a fact about the driver, not the echo"
        );
    }

    // [unit->REQ-AUTO-SUSPEND-MODES] apply_event resolves the chain
    // end-to-end: a node default auto-suspends a quiet dormant perch from the
    // tick, and an endpoint `disable` override holds it warm against the same
    // node default. (Repinned from the D9-2 `Some(500)` / endpoint-0 knob.)
    #[test]
    fn apply_event_auto_suspends_under_the_resolved_chain() {
        use spt_store::info::set_status;
        use spt_store::liveness::STATUS_ONLINE;

        let d = tempfile::tempdir().unwrap();
        write_info(
            d.path(),
            &InfoJson::new("ling", "t", 4242, "sid", "live_agent"),
        )
        .unwrap();
        // Online + a Dormant record ⇒ liveness-aware `from` reads Dormant.
        set_status(d.path(), STATUS_ONLINE).unwrap();
        write_rest(d.path(), RestState::Dormant, 1_000).unwrap();

        // Node default 500ms; dormant since 1000, idle + uncontrolled from
        // before; tick at 2000 ⇒ suspended.
        let quiet = SuspendFacts {
            controlled: false,
            controlled_since_ms: Some(0),
            idle: true,
            idle_since_ms: Some(0),
        };
        let node = Some(spt_store::autosuspend::AutoSuspend {
            mode: spt_store::autosuspend::AutoSuspendMode::DormantEnable,
            after_ms: 500,
        });
        let report = apply_event(
            d.path(),
            RestEvent::AutoSuspendTick { now_ms: 2_000, facts: quiet },
            node,
            SiblingView::default(),
            2_000,
            || Ok(()),
            || {},
        )
        .expect("apply ok")
        .expect("edge");
        assert_eq!(report.to, RestState::Suspended);
        assert!(
            !report.echo_fired,
            "dormant→suspended: the echo already ran at the rest edge"
        );

        // Endpoint 0-override: same node knob, but the endpoint opted out.
        let d2 = tempfile::tempdir().unwrap();
        write_info(
            d2.path(),
            &InfoJson::new("oak", "t", 4242, "sid", "live_agent"),
        )
        .unwrap();
        let mut rec = spt_store::info::read_info(d2.path()).unwrap();
        rec.auto_suspend = Some(spt_store::autosuspend::AutoSuspend {
            mode: spt_store::autosuspend::AutoSuspendMode::Disable,
            after_ms: 0,
        });
        spt_store::info::write_info(d2.path(), &rec).unwrap();
        set_status(d2.path(), STATUS_ONLINE).unwrap();
        write_rest(d2.path(), RestState::Dormant, 1_000).unwrap();
        let out = apply_event(
            d2.path(),
            RestEvent::AutoSuspendTick { now_ms: u64::MAX, facts: quiet },
            node,
            SiblingView::default(),
            2_000,
            || Ok(()),
            || {},
        )
        .expect("apply ok");
        assert_eq!(out, None, "endpoint disable beats the node default");
    }

    // [unit->REQ-INST-4] the daemon-side host arms the endpoint's echo gate
    // at the rest edge (the cross-process trigger — the endpoint's OWN pulse
    // loop runs the bounded summarizer call, never the daemon's feed worker:
    // the KH 7.4 non-blocking posture by construction) and fires no gate on
    // the wake edge.
    #[test]
    fn daemon_rest_event_arms_gate_at_rest_edge_only() {
        use spt_store::info::set_status;
        use spt_store::liveness::STATUS_ONLINE;

        crate::test_home::with_home(|_| {
            let perch_path = perch::resolve_perch_path("ling", ParentHint::Infer);
            std::fs::create_dir_all(&perch_path).unwrap();
            write_info(
                &perch_path,
                &InfoJson::new("ling", "t", 4242, "sid", "live_agent"),
            )
            .unwrap();
            // Live perch (online) ⇒ recordless `from` derives Active.
            set_status(&perch_path, STATUS_ONLINE).unwrap();
            let gate = perch::resolve_edge_echo_file("ling", ParentHint::Infer);
            assert!(!gate.exists());

            // active → dormant: the gate is armed for ling's pulse loop.
            let report = daemon_rest_event("ling", RestEvent::SiblingActivated, None)
                .expect("apply ok")
                .expect("edge");
            assert!(
                report.echo_fired,
                "the rest edge fired the (gate-armed) echo seam"
            );
            assert!(
                gate.exists(),
                "gate armed — ling's own pulse fires the real commune"
            );
            assert_eq!(read_rest(&perch_path).unwrap().state, RestState::Dormant);

            // wake: no gate re-arm; the freshness-pull marker drops instead.
            std::fs::remove_file(&gate).unwrap();
            let report = daemon_rest_event("ling", RestEvent::Wake, None)
                .expect("apply ok")
                .expect("edge");
            assert_eq!(report.to, RestState::Active);
            assert!(!gate.exists(), "wake arms no echo");
            assert!(
                perch::identity_dir().join(PULL_MARKER_FILE).exists(),
                "wake dropped the freshness-pull marker"
            );
        });
    }

    // [unit->REQ-SHELL-2] the owner-edge shell cascade (CONTEXT §Shell
    // sleep/wake): "offline" for shells means the owner SUSPENDED — a detach
    // (dormant, resting-warm) closes nothing; the suspend edge closes every
    // online shell with its per-shell divergence (ephemeral ⇒ erased,
    // non-ephemeral ⇒ offline + intact); the wake edge relaunches exactly the
    // `persistent` offline shells (binary up, perch still offline until its
    // bind — the D3b contract).
    #[test]
    fn rest_edges_cascade_shells_with_divergence() {
        crate::test_home::with_home(|home| {
            use spt_store::shellinfo::{self, SHELL_STATUS_OFFLINE, SHELL_STATUS_ONLINE};
            let owlery = perch::owlery_dir();
            let perch_path = perch::resolve_perch_path("ling", ParentHint::Infer);
            std::fs::create_dir_all(&perch_path).unwrap();
            write_info(
                &perch_path,
                &InfoJson::new("ling", "t", 4242, "sid", "live_agent"),
            )
            .unwrap();
            // Live perch (online) ⇒ recordless `from` derives Active.
            spt_store::info::set_status(&perch_path, spt_store::liveness::STATUS_ONLINE).unwrap();

            // Two registered shell adapters: a persistent one and an
            // ephemeral one, both spawning a cross-platform no-op.
            #[cfg(windows)]
            let noop = "cmd /c exit 0";
            #[cfg(unix)]
            let noop = "true";
            let adapters = spt_store::perch::adapters_dir();
            for (name, extra) in [
                ("mock-persist", "persistent = true"),
                ("mock-ephem", "ephemeral = true"),
            ] {
                let src = home.join("srcs").join(name);
                std::fs::create_dir_all(&src).unwrap();
                std::fs::write(
                    src.join("manifest.toml"),
                    format!(
                        "[adapter]\nname = \"{name}\"\nkind = \"shell\"\nversion = \"1\"\n\
                         min_spt_core_version = \"0\"\n\n[shell]\nspawn = '{noop}'\n{extra}\n"
                    ),
                )
                .unwrap();
                spt_runtime::registry::register(&adapters, &src, 1).unwrap();
            }

            // One instance of each, manually onlined (the bind's job — not
            // under test here).
            let persist = shellinfo::spawn_record(&owlery, "ling", "mock-persist", None).unwrap();
            let ephem = shellinfo::spawn_record(&owlery, "ling", "mock-ephem", None).unwrap();
            for id in [&persist, &ephem] {
                let p = perch::resolve_shell_perch_path_in(&owlery, "ling", id);
                let mut info = shellinfo::read_shell_info(&p).unwrap();
                info.status = SHELL_STATUS_ONLINE.to_string();
                shellinfo::write_shell_info(&p, &info).unwrap();
            }

            // active → dormant (detach): resting-WARM — shells ride through.
            daemon_rest_event("ling", RestEvent::SiblingActivated, None)
                .expect("ok")
                .expect("edge");
            let pp = perch::resolve_shell_perch_path_in(&owlery, "ling", &persist);
            assert_eq!(
                shellinfo::read_shell_info(&pp).unwrap().status,
                SHELL_STATUS_ONLINE,
                "dormant closes nothing"
            );

            // dormant → suspended: the cascade, with per-shell divergence.
            daemon_rest_event("ling", RestEvent::Suspend, None)
                .expect("ok")
                .expect("edge");
            assert_eq!(
                shellinfo::read_shell_info(&pp).unwrap().status,
                SHELL_STATUS_OFFLINE,
                "persistent shell cascaded offline"
            );
            assert!(
                !perch::resolve_shell_perch_path_in(&owlery, "ling", &ephem).exists(),
                "ephemeral shell torn down + history erased on the same edge"
            );

            // suspended → active (wake): exactly the persistent one
            // relaunches — binary up (pid recorded), perch offline until its
            // own bind onlines it.
            daemon_rest_event("ling", RestEvent::Wake, None)
                .expect("ok")
                .expect("edge");
            assert!(
                pp.join(crate::shellhost::SHELL_PID_FILE).exists(),
                "persistent shell relaunched on the wake edge"
            );
            assert_eq!(
                shellinfo::read_shell_info(&pp).unwrap().status,
                SHELL_STATUS_OFFLINE,
                "launch is not the online switch — the bind is"
            );
            assert_eq!(
                shellinfo::list_shells(&owlery, "ling").len(),
                1,
                "the ephemeral instance stayed gone"
            );
        });
    }

    // [unit->REQ-INST-3] the freshness-pull marker: dropped idempotently,
    // consumed exactly once — two pump ticks can never both claim one wake.
    #[test]
    fn freshness_pull_marker_is_taken_exactly_once() {
        let d = tempfile::tempdir().unwrap();
        let marker = d.path().join("identity").join(PULL_MARKER_FILE);

        assert!(!take_freshness_pull(&marker), "nothing to take yet");
        request_freshness_pull(&marker).unwrap();
        request_freshness_pull(&marker).unwrap(); // re-request is idempotent
        assert!(take_freshness_pull(&marker), "first take claims the wake");
        assert!(!take_freshness_pull(&marker), "second take gets nothing");
    }

    // [unit->REQ-INST-3] durability round-trip: write_rest stamps the
    // dormancy anchor exactly on the dormant state (cleared otherwise),
    // read_rest reads it back, a recordless perch reads None (legacy
    // fallback), and a foreign future tag degrades to None instead of
    // wedging the reader.
    #[test]
    fn rest_record_round_trips_with_anchor_discipline() {
        let d = tempfile::tempdir().unwrap();
        write_info(
            d.path(),
            &InfoJson::new("ling", "t", 4242, "sid", "live_agent"),
        )
        .unwrap();

        assert_eq!(
            read_rest(d.path()),
            None,
            "pre-D9-2 record has no resting state"
        );

        write_rest(d.path(), RestState::Dormant, 7_000).unwrap();
        let rec = read_rest(d.path()).expect("record present");
        assert_eq!(rec.state, RestState::Dormant);
        assert_eq!(
            rec.dormant_since_ms,
            Some(7_000),
            "anchor stamped at the edge"
        );

        write_rest(d.path(), RestState::Active, 8_000).unwrap();
        let rec = read_rest(d.path()).expect("record present");
        assert_eq!(rec.state, RestState::Active);
        assert_eq!(
            rec.dormant_since_ms, None,
            "anchor cleared on leaving dormancy"
        );

        write_rest(d.path(), RestState::Suspended, 9_000).unwrap();
        let rec = read_rest(d.path()).expect("record present");
        assert_eq!(rec.state, RestState::Suspended);
        assert_eq!(
            rec.dormant_since_ms, None,
            "suspended carries no dormancy anchor"
        );

        // A foreign (newer-fleet) tag degrades to "no record", never an error.
        spt_store::info::set_rest_state(d.path(), "hibernating", None).unwrap();
        assert_eq!(
            read_rest(d.path()),
            None,
            "unknown tag reads as no resting record"
        );

        // Tags round-trip through the parser.
        for s in [RestState::Active, RestState::Dormant, RestState::Suspended] {
            assert_eq!(RestState::from_tag(s.as_tag()), Some(s));
        }
        assert_eq!(RestState::from_tag(""), None);
    }
}