godspeed_sdk/

service_context.rs

// SPDX-License-Identifier: Apache-2.0
//! ServiceContext - entry-point type handed to every service's `service_main`.
//!
//! Provides safe, named access to the capabilities the service declared in its
//! contract. Capability names match the contract field names exactly.
//! Requesting a cap not in the contract returns `Err(CapNotHeld)`.

use crate::capability::{CapError, CapHandle};
use crate::ipc::{IpcError, Message};
use crate::syscall::raw_syscall;

/// Why a `reacquire_cap_detail` failed. The two real cases want OPPOSITE responses, which is the
/// entire reason this type exists rather than a `bool`.
///
/// `NameNotRegistered` is **transient and expected**: peers are acquired before they are spawned at
/// boot, and again while one is mid-restart. Retry; it heals. `CapTableFull` is **permanent**: this
/// service cannot hold another capability, so it can never reacquire anything again and will not
/// recover until it is restarted - a failed recovery, which §26.7 says must stay exactly as visible as
/// the original failure.
///
/// Collapsed into one value (as `reacquire_cap` still does, for callers that only retry) a service can
/// only print both and commit to neither. That ambiguity cost a full debugging pass on the Pi 2: a
/// `block-driver` that was alive the whole time produced the same line as a reader out of cap slots.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AcquireFailure {
    /// The name does not resolve right now - no live instance is registered under it. Transient.
    NameNotRegistered,
    /// The name resolved, but this service's own capability table is full. Terminal until restart.
    CapTableFull,
    /// The request itself was refused (name empty, too long, or not a declared peer).
    Refused,
}

impl AcquireFailure {
    /// Decode the kernel's negative return. Codes are fixed at the syscall boundary
    /// (`kernel/src/syscall/dispatch.rs`); an UNRECOGNISED negative maps to `Refused` rather than
    /// being guessed at, so a future code cannot silently masquerade as one of the two named cases.
    fn from_code(ret: i64) -> Self {
        match ret {
            -20 => AcquireFailure::NameNotRegistered,
            -21 => AcquireFailure::CapTableFull,
            _   => AcquireFailure::Refused,
        }
    }

    /// One clause naming the cause, for a caller assembling a log line.
    pub fn reason(self) -> &'static str {
        match self {
            AcquireFailure::NameNotRegistered =>
                "the name does not resolve - no live instance is registered under it",
            AcquireFailure::CapTableFull =>
                "this service's capability table is FULL - it cannot hold another cap and will not recover until restarted",
            AcquireFailure::Refused =>
                "the kernel refused the request (name empty, too long, or not a declared peer)",
        }
    }
}

/// Outcome of [`ServiceContext::request_with_reply_deadline_outcome_into`] - the buffer-writing twin
/// of [`DeadlineOutcome`].
///
/// It exists because the `_into` form collapsed all four outcomes into `Option<usize>`, and the two
/// that matter are OPPOSITES. `SendFailed` means the request never left, so nothing can arrive late
/// and retrying after a reacquire is safe. `Timeout` means it DID leave, so a late reply is still
/// coming and re-sending desyncs the protocol - every later reply is matched to the wrong request.
/// A caller handed `None` cannot tell those apart, so it must either never retry (and stay broken
/// after a peer restarts) or always retry (and desync). `fs` had the first of those.
#[derive(Debug)]
pub enum DeadlineOutcomeInto {
    /// A reply arrived; the payload is in the caller's buffer, this many bytes of it.
    Reply(usize),
    /// The send never left: the peer's cap is stale or its name does not currently resolve. Nothing
    /// is in flight. Reacquire by name and retry ONCE - this is the only safe retry.
    SendFailed,
    /// The request WAS delivered, and the peer died holding it (`ReplyDead`, 8.6), so the work may
    /// have been done. Re-send only an operation that is safe to repeat (a block read, a rewrite of
    /// the same sector). It was folded into `SendFailed` until 2026-10-10 (`backlog/fixed/80` S3); `gs::call`
    /// has kept the two apart all along, as `Error::PeerDied` against `Error::Unreachable`.
    PeerDied,
    /// The peer's queue was full. The request did not leave either, but the peer is alive and busy.
    QueueFull,
    /// The request was delivered and no reply came in time. Do NOT re-send.
    Timeout,
}

/// Outcome of an abortable request/reply ([`ServiceContext::request_with_reply_abortable`]): the reply
/// arrived, the user pressed `q`/`Q`/ESC to abort the wait, or the deadline expired with no reply.
/// `Timeout` also covers a request that never left (no such peer, no reply cap, a failed send).
pub enum ReqOutcome {
    Reply(Message),
    Aborted,
    Timeout,
}

/// Outcome of [`ServiceContext::request_with_reply_deadline_outcome`], which distinguishes the two
/// ways a deadline request fails - so a caller can self-heal a *restarted* peer without penalising a
/// *silent* one. `SendFailed`: the send never left (the peer's cap is stale or its name is currently
/// unresolvable - it was killed and respawned, generation bumped), so reacquiring it by name and
/// retrying will reach the fresh instance. `Timeout`: the peer received the request but did not reply
/// within the deadline (a genuinely silent/absent host) - retrying only doubles the wait.
/// `request_with_reply_deadline` collapses both to `None`; use this when the difference matters.
pub enum DeadlineOutcome {
    Reply(Message),
    SendFailed,
    /// The peer is ALIVE and its queue is FULL. A different failure entirely, and it was folded into
    /// `SendFailed` until x86 showed the cost: `net-stack` answered a full queue by reacquiring a
    /// capability that was never stale, retried once and gave up - "5 of 6 REQUESTs never left the
    /// host", DHCP backing off for 34 seconds on a link that was up the whole time.
    ///
    /// Congestion is transient by definition. The answer is to pace and retry, not to go looking for
    /// a peer that never went anywhere.
    QueueFull,
    Timeout,
}

/// Where a wall-clock reading came from. Reported alongside the time so a displayed timestamp says what
/// it is standing on: a local hardware clock, a network correction, or nothing (§26.4 - a fallback chain
/// is mechanism while its choice is visible, and magic when it is not).
#[derive(Clone, Copy, PartialEq, Eq)]
pub enum ClockSource {
    /// No clock: no hardware RTC and nothing has set the time. `date` says so rather than inventing one.
    Unset,
    /// A local hardware real-time clock reading a plausible date.
    Rtc,
    /// Set from the network (SNTP) this boot.
    Ntp,
    /// Started from the persisted floor (`/system/services/time/floor`) - no RTC on this board and no network sync yet.
    ///
    /// A real reading and a distinct one: it is a LOWER BOUND carried over from the last boot, advancing
    /// correctly since, but it cannot know how long the machine was powered off. Reporting it as `Rtc`
    /// would claim hardware this board does not have, and as `Unset` would deny a time it is displaying.
    Floor,
}

/// The machine's one pending power action, as the kernel holds it (InspectKernel query 28, `docs/shutdown.md`).
/// See [`ServiceContext::power_pending`].
#[derive(Clone, Copy, PartialEq, Eq)]
pub struct PowerPending {
    /// Whole seconds until it falls due, rounded up.
    pub secs: u64,
    /// A reboot; otherwise a power-off.
    pub reboot: bool,
}

/// Wall-clock date/time read from the hardware RTC, fully decoded (binary,
/// 24-hour). See [`ServiceContext::datetime`].
#[derive(Clone, Copy, PartialEq, Eq)]
pub struct Datetime {
    pub year: u16,
    pub month: u8,
    pub day: u8,
    pub hour: u8,
    pub minute: u8,
    pub second: u8,
}

impl Datetime {
    /// Decode Unix epoch seconds into calendar fields (Howard Hinnant's `civil_from_days`) - the inverse
    /// of [`Datetime::epoch_secs`]. Used to render a stored epoch, such as the clock floor, as a date.
    pub fn from_epoch_secs(epoch: i64) -> Datetime {
        let days = epoch.div_euclid(86_400);
        let rem = epoch.rem_euclid(86_400);
        let z = days + 719_468;
        let era = if z >= 0 { z } else { z - 146_096 } / 146_097;
        let doe = z - era * 146_097;
        let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
        let y = yoe + era * 400;
        let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
        let mp = (5 * doy + 2) / 153;
        let d = doy - (153 * mp + 2) / 5 + 1;
        let m = if mp < 10 { mp + 3 } else { mp - 9 };
        Datetime {
            year:  (y + (m <= 2) as i64) as u16,
            month: m as u8,
            day:   d as u8,
            hour:   (rem / 3_600) as u8,
            minute: ((rem % 3_600) / 60) as u8,
            second: (rem % 60) as u8,
        }
    }

    /// Days since the epoch (1970-01-01), proleptic Gregorian and leap-year aware
    /// (Howard Hinnant's `days_from_civil`). The basis for both `weekday` and
    /// `epoch_secs`.
    fn days_since_epoch(&self) -> i64 {
        let mut y = self.year as i64;
        let m = self.month as i64;
        let d = self.day as i64;
        y -= (m <= 2) as i64;
        let era = (if y >= 0 { y } else { y - 399 }) / 400;
        let yoe = y - era * 400;
        let doy = (153 * (m + if m > 2 { -3 } else { 9 }) + 2) / 5 + d - 1;
        let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy;
        era * 146097 + doe - 719468
    }

    /// Day of week, 0 = Sunday .. 6 = Saturday (computed, not the RTC's
    /// often-unreliable weekday register).
    pub fn weekday(&self) -> u8 {
        (self.days_since_epoch() + 4).rem_euclid(7) as u8
    }

    /// Seconds since the epoch (1970-01-01). Assumes the RTC reads UTC; if the
    /// hardware clock is set to local time the value is offset by the timezone
    /// (v1 has no timezone database).
    pub fn epoch_secs(&self) -> i64 {
        self.days_since_epoch() * 86_400
            + self.hour as i64 * 3_600
            + self.minute as i64 * 60
            + self.second as i64
    }
}

#[cfg(test)]
mod datetime_tests {
    use super::Datetime;

    fn dt(year: u16, month: u8, day: u8, hour: u8, minute: u8, second: u8) -> Datetime {
        Datetime { year, month, day, hour, minute, second }
    }

    fn is_leap(y: i64) -> bool { (y % 4 == 0 && y % 100 != 0) || y % 400 == 0 }
    fn days_in_month(y: i64, m: i64) -> i64 {
        match m {
            1 => 31, 2 => if is_leap(y) { 29 } else { 28 }, 3 => 31, 4 => 30, 5 => 31, 6 => 30,
            7 => 31, 8 => 31, 9 => 30, 10 => 31, 11 => 30, 12 => 31, _ => 0,
        }
    }

    /// A deliberately naive, obviously-correct reference: count days from 1970 by iterating years +
    /// months. It cannot share a bug with Hinnant's closed form - which is what makes the cross-check valid.
    fn reference_epoch(y: i64, mo: i64, d: i64, h: i64, mi: i64, s: i64) -> i64 {
        let mut days: i64 = 0;
        for yy in 1970..y { days += if is_leap(yy) { 366 } else { 365 }; }
        for mm in 1..mo  { days += days_in_month(y, mm); }
        days += d - 1;
        days * 86_400 + h * 3_600 + mi * 60 + s
    }

    #[test]
    fn reference_matches_known_unix_anchors() {
        // Validate the REFERENCE itself against well-known Unix epochs first, so the cross-check is trustworthy.
        assert_eq!(reference_epoch(1970, 1, 1, 0, 0, 0), 0);
        assert_eq!(reference_epoch(2000, 1, 1, 0, 0, 0), 946_684_800);
        assert_eq!(reference_epoch(2038, 1, 19, 3, 14, 8), 2_147_483_648); // Y2038 (2^31)
        assert_eq!(reference_epoch(2000, 2, 29, 0, 0, 0), 951_782_400);    // leap day
    }

    #[test]
    fn epoch_secs_matches_known_unix_anchors() {
        assert_eq!(dt(1970, 1, 1, 0, 0, 0).epoch_secs(), 0);
        assert_eq!(dt(2000, 1, 1, 0, 0, 0).epoch_secs(), 946_684_800);
        assert_eq!(dt(2038, 1, 19, 3, 14, 8).epoch_secs(), 2_147_483_648);
    }

    #[test]
    fn epoch_secs_matches_reference_over_a_multi_century_sweep() {
        // Cross-check Hinnant (the SDK's epoch_secs - the twin every SERVICE uses) vs the naive reference for
        // every month of 1971..=2100: div-4 leaps, the 2100 century non-leap, the 2000 leap-400. This is the
        // drift guard - if a future edit to the SDK's days_since_epoch diverges from the kernel's (pinned in
        // kernel/src/clock.rs), this catches it.
        for y in 1971..=2100i64 {
            for mo in 1..=12i64 {
                let last = days_in_month(y, mo);
                for &d in &[1i64, 15, 28, last] {
                    for &(h, mi, s) in &[(0i64, 0i64, 0i64), (23, 59, 59), (12, 30, 15)] {
                        assert_eq!(
                            dt(y as u16, mo as u8, d as u8, h as u8, mi as u8, s as u8).epoch_secs(),
                            reference_epoch(y, mo, d, h, mi, s),
                            "SDK epoch_secs mismatch at {}-{:02}-{:02} {:02}:{:02}:{:02}", y, mo, d, h, mi, s);
                    }
                }
            }
        }
    }

    #[test]
    fn weekday_matches_known_anchors() {
        // weekday() shares days_since_epoch with epoch_secs. 0=Sun..6=Sat. Known: 1970-01-01 Thursday (4),
        // 2000-01-01 Saturday (6), 2026-06-06 Saturday (6) - the T630 `date` HW example.
        assert_eq!(dt(1970, 1, 1, 0, 0, 0).weekday(), 4);
        assert_eq!(dt(2000, 1, 1, 0, 0, 0).weekday(), 6);
        assert_eq!(dt(2026, 6, 6, 0, 0, 0).weekday(), 6);
    }

    #[test]
    fn from_epoch_secs_round_trips_over_the_sweep() {
        // The INVERSE (civil_from_days) must undo epoch_secs exactly, over the same multi-century sweep -
        // the drift guard its forward twin already has. Without this, the only thing pinning the decode
        // used to render a stored epoch (the clock floor) would be that it looked right once.
        for y in 1971..=2100i64 {
            for mo in 1..=12i64 {
                let last = days_in_month(y, mo);
                for &d in &[1i64, 15, 28, last] {
                    for &(h, mi, s) in &[(0i64, 0i64, 0i64), (23, 59, 59), (12, 30, 15)] {
                        let a = dt(y as u16, mo as u8, d as u8, h as u8, mi as u8, s as u8);
                        assert!(Datetime::from_epoch_secs(a.epoch_secs()) == a,
                            "from_epoch_secs round-trip mismatch at {}-{:02}-{:02} {:02}:{:02}:{:02}",
                            y, mo, d, h, mi, s);
                    }
                }
            }
        }
        assert!(Datetime::from_epoch_secs(0) == dt(1970, 1, 1, 0, 0, 0));
    }
}

// ---------------------------------------------------------------------------
// ServiceContextData page layout.
// MUST match `ServiceContextData` in `kernel/src/task/mod.rs`.
// ---------------------------------------------------------------------------

const SERVICE_CTX_ADDR:    u64   = 0x3ff000;
/// What a spawner tells the kernel about a task it wants started from an image IT supplies
/// (`SpawnImage`, syscall 52). Layout must match the kernel's `SpawnRequest` exactly.
///
/// Shaped for the end state, not for what the kernel honours today: the hardware and privilege
/// fields are here because step D needs them, and a request that sets one is REFUSED loudly rather
/// than silently ignored (`docs/service-ownership.md`).
#[repr(C)]
#[derive(Clone, Copy)]
pub struct SpawnRequest {
    pub version:      u32,
    pub flags:        u32,
    pub image_ptr:    u64,
    pub image_len:    u64,
    pub name_ptr:     u64,
    pub name_len:     u32,
    pub core:         u32,
    pub memory_limit: u64,
    pub privileges:   u32,
    pub hw_flags:     u32,
    pub mmio_base:    u64,
    pub mmio_len:     u64,
    pub dma_pages:    u32,
    pub bdf:          u32,
    pub irq_count:    u32,
    pub irqs:         [u8; 4],
    pub peers_ptr:      u64,
    pub peers_len:      u32,
    pub _pad:           u32,
    /// Caller-provided caps to install into the child: `[label_len][label][slot_lo][slot_hi]`
    /// repeated. The same encoding `SpawnWithCaps` uses - one wire format, not two.
    pub installs_ptr:   u64,
    pub installs_count: u32,
    pub _pad2:          u32,
    /// Mode selector. Named for probes, its first user, but general: `observe` reads it to choose
    /// one-shot / live / foreground, and takes the LIVE LOOP at 0.
    pub probe_mode:     u32,
    pub _pad3:          u32,
}

/// 2 since `installs` was added, 3 since `probe_mode` was. A spawner built against a different kernel
/// is refused loudly.
pub const SPAWN_REQUEST_VERSION:  u32 = 3;
pub const SPAWN_FLAG_REQ_RECV:    u32 = 1 << 0;
pub const SPAWN_FLAG_REQ_CONSOLE: u32 = 1 << 1;
/// `core` is a STRICT placement, not a preference.
///
/// The two are different rules (9.2) and the kernel has always had both: an explicitly-requested core
/// (a restart's `--core N`) is REJECTED with `PlacementInvalid` when that core is not ready, because
/// silently placing the service elsewhere would defeat the point of asking. A catalogue row's
/// PREFERRED core falls back to round-robin instead, so a machine with fewer cores still boots.
///
/// Without this bit a moved service's table preference read as an override, and every service naming
/// a core the machine does not have failed to spawn rather than degrading - `events` and `xhci` (both
/// core 2) simply vanished under `-smp 2`, which 11.3 requires to keep working.
pub const SPAWN_FLAG_CORE_STRICT: u32 = 1 << 2;
/// Mint the child's name-wired peer caps WITH `GRANT`, so it can re-delegate them.
///
/// AUTHORITY, not a setting: handing out a re-delegatable capability is itself a grant (§7.4), which
/// is why the kernel used to keep this keyed by name for the one probe that needs it (§22 Test 5A).
/// As a request bit it is checked the same way every privilege is - the spawner may ask for it only
/// because it could transfer such a cap itself.
pub const SPAWN_FLAG_PEERS_GRANT: u32 = 1 << 3;
/// Report this task's death to the supervisor's death-notification endpoint and count it as a restart. The
/// supervisor sets it for every service it manages; the kernel keeps no list of which services those are
/// (`docs/audio.md`, "No service names in the kernel").
pub const SPAWN_FLAG_WATCHED:     u32 = 1 << 4;
/// A live handover: start this beside the one task already running under its name, which keeps running until it
/// ends itself; the new one takes the name. The supervisor's alone, since only `SpawnImage` honours it (CLAUDE.md 2.5).
pub const SPAWN_FLAG_HANDOVER:    u32 = 1 << 5;

/// Bits for `SpawnRequest::privileges`. A spawner may only request what it HOLDS ITSELF - the kernel
/// checks, and refuses otherwise - so this passes authority on, it never mints it (3.1, 7.3).
pub mod privbits {
    pub const SPAWN:           u32 = 1 << 0;
    pub const CONSOLE_PUSH:    u32 = 1 << 1;
    pub const INTROSPECT:      u32 = 1 << 2;
    pub const SERVICE_CONTROL: u32 = 1 << 3;
    pub const FIRE_IRQ:        u32 = 1 << 4;
    pub const REBOOT:          u32 = 1 << 5;
    pub const ACQUIRE_ANY:     u32 = 1 << 6;
    pub const RESOURCE_MINT:   u32 = 1 << 7;
    /// ARM-only in practice; the bit is arch-neutral.
    pub const GPIO:            u32 = 1 << 8;
    /// SET_CLOCK with READ (raise the clock FLOOR), not WRITE (step the clock).
    pub const SET_CLOCK_FLOOR: u32 = 1 << 9;
    /// SET_CLOCK with WRITE (set the wall clock). Distinct from SET_CLOCK_FLOOR, the READ right.
    pub const SET_CLOCK:       u32 = 1 << 10;
    /// NET_DEVICE: move ethernet frames through an in-kernel network device (`NetFrame*`/`NetInfo`,
    /// syscalls 42-44). No port has one any more - every arch answers with a stub - and no spawn
    /// request in the tree sets this bit; the kernel still understands it.
    pub const NET_DEVICE:      u32 = 1 << 11;
    /// Read PCI configuration space through the legacy CF8/CFC ports (step D2). READ-ONLY and held
    /// by ONE service - see `docs/service-ownership.md` D2 for why the write side is not on offer.
    pub const PCI_CFG:         u32 = 1 << 12;
    /// Set the Arm cores to the platform's minimum or maximum clock (`CpuClock`, syscall 55). Held by
    /// ONE service, `power`, because the clock is one machine-wide setting (`docs/power.md`).
    pub const CPU_CLOCK:       u32 = 1 << 13;
    /// Turn the machine off, now or at a deadline the kernel holds, or halt it where it cannot cut its own
    /// power (`PowerOff`, syscall 56). Held by the shell, for the operator (`docs/shutdown.md`).
    pub const POWER_OFF:       u32 = 1 << 14;
}

/// Device classes a spawner can name in `SpawnRequest::hw_flags`. The kernel resolves the class to
/// what its own bus scan found - a CLASS rather than an address, because the kernel keeps a permanent
/// physical DMA reservation per device that a respawned driver must get back.
pub mod hwclass {
    /// `hw-enumerator`'s which-device question: `[BY_CLASS, class (3 bytes LE), PREFER_OWN]`, answered
    /// `[bdf u32, class u32]`. The supervisor asks it to choose a PCI driver's device, which the kernel
    /// then grants by that BDF; `hardware` asks it again to say which device that driver holds. One
    /// definition, so the two cannot disagree. `PREFER_OWN`: the first device of the class that is not
    /// a display's companion function, if there is one (the T630's HDMI audio, `docs/audio.md`).
    pub const BY_CLASS:    u8 = 3;
    pub const PREFER_OWN:  u8 = 1;
    pub const NONE:        u32 = 0;
    pub const NIC:         u32 = 2;
    pub const XHCI:        u32 = 3;
    pub const EHCI:        u32 = 4;
    pub const DWC2:        u32 = 5;
    pub const FRAMEBUFFER: u32 = 6;
    /// Not a device: the software-raised test interrupt (§22 IR1). A class, so that the probe which
    /// receives it names a CLASS like any driver and the kernel states the vector.
    pub const TEST_IRQ:    u32 = 7;
    /// An audio jack driven by PWM and fed by the SoC's DMA engine (the Pis, `docs/audio.md`). The kernel
    /// routes the jack's pins and starts the PWM clock as part of the grant, then maps the PWM block and
    /// the DMA engine side by side and grants a DMA arena.
    pub const AUDIO_PWM:   u32 = 8;
    /// A WiFi radio on an SDIO host at a fixed SoC address (the Pi 4's CYW43455 behind the Arasan, the
    /// VisionFive 2 Lite's AIC8800 behind a DesignWare `dw_mmc`). The
    /// kernel grants the host's window - and, where it can, the radio's power control - to the service
    /// whose request names this kind, never to a service because of its name.
    pub const WIFI_SDIO:   u32 = 9;

    /// Bit 31: the value is a PCI CLASS CODE, not one of the named kinds above.
    pub const PCI:         u32 = 1 << 31;
    /// Bit 28: place this device behind the IOMMU (§6.4).
    pub const PCI_CONFINE: u32 = 1 << 28;
    /// Bit 29: this driver wants an INTERRUPT. The kernel allocates a vector from its MSI pool and
    /// programs the device with it - the caller never names a vector, because routing one is
    /// authority (`task::hw_irqs_for`).
    pub const PCI_IRQ:     u32 = 1 << 29;

    /// Describe a PCI device by what the BUS says it is: the industry-standard 24-bit
    /// (class, subclass, prog-if) triple, plus which BAR holds its registers.
    ///
    /// This is what lets a driver be added with NO KERNEL CHANGE. `0x0C0330` is xHCI on every
    /// machine ever built, so the kernel needs no name for it - it looks the code up in the table
    /// its boot scan already filled. The BAR index is the one extra fact, because devices differ
    /// (xHCI 0, AHCI 5), and an INDEX is not an address, so it is safe to accept from a caller.
    pub const fn pci(class_code: u32, bar_ix: u32, confine: bool) -> u32 {
        PCI | (if confine { PCI_CONFINE } else { 0 }) | ((bar_ix & 0x7) << 24) | (class_code & 0x00FF_FFFF)
    }

    /// BAR index meaning "the first mapped MEMORY BAR" rather than a numbered one.
    ///
    /// Use this when one driver covers devices that put their registers in different BARs - the
    /// e1000 uses BAR0 where the RTL8168 puts I/O ports there and its registers in BAR2. Both are
    /// "the first memory BAR", and neither driver should have to know which it got.
    pub const BAR_AUTO: u32 = 7;

    /// `pci`, for a driver that needs an INTERRUPT as well.
    pub const fn pci_irq(class_code: u32, bar_ix: u32, confine: bool) -> u32 {
        pci(class_code, bar_ix, confine) | PCI_IRQ
    }
}

impl SpawnRequest {
    /// A request with everything the kernel does not yet honour left at zero.
    pub fn new(image: &[u8], name: &str) -> Self {
        Self {
            version: SPAWN_REQUEST_VERSION, flags: 0,
            image_ptr: image.as_ptr() as u64, image_len: image.len() as u64,
            name_ptr: name.as_ptr() as u64, name_len: name.len() as u32,
            core: u32::MAX, memory_limit: 0,
            privileges: 0, hw_flags: 0, mmio_base: 0, mmio_len: 0,
            dma_pages: 0, bdf: 0, irq_count: 0, irqs: [0; 4],
            peers_ptr: 0, peers_len: 0, _pad: 0,
            installs_ptr: 0, installs_count: 0, _pad2: 0,
            probe_mode: 0, _pad3: 0,
        }
    }

    /// Encode `(label, cap)` pairs into `buf` and point this request at them.
    ///
    /// The caller must hold each cap WITH GRANT: the kernel checks it, and that check is what makes
    /// installing them non-escalating (7.3) - a caller that can GRANT a cap could transfer it anyway.
    pub fn set_installs(&mut self, buf: &mut [u8], pairs: &[(&str, CapHandle)]) -> bool {
        let mut n = 0usize;
        for (label, cap) in pairs {
            let lb = label.as_bytes();
            // The kernel's limit, `PEER_NAME_BYTES`: this once accepted more than the kernel did, and the kernel refused
            // the spawn for a label it had been told was fine (`backlog/fixed/80` S5).
            if lb.is_empty() || lb.len() > PEER_NAME_BYTES { return false; }
            if n + 1 + lb.len() + 2 > buf.len() { return false; }
            buf[n] = lb.len() as u8; n += 1;
            buf[n..n + lb.len()].copy_from_slice(lb); n += lb.len();
            buf[n] = (cap.0 & 0xFF) as u8; buf[n + 1] = ((cap.0 >> 8) & 0xFF) as u8; n += 2;
        }
        self.installs_ptr   = if n > 0 { buf.as_ptr() as u64 } else { 0 };
        self.installs_count = pairs.len() as u32;
        true
    }
}

/// How a task ended, as the kernel says it in the death notice (`name`, NUL, then one of these). The
/// kernel's `task::death` holds the same values; `ask` asserts its copy against these.
pub mod death {
    /// Ended by the `kill` syscall; the name of the task that called it follows.
    pub const KILLED: u8 = b'K';
    /// Ended by a user-mode CPU fault; the kind below and the address (u64 LE) follow.
    pub const FAULT: u8 = b'F';
    /// Ended itself with `exit`; the status it gave (u32 LE) follows.
    pub const EXITED: u8 = b'E';
    /// A fault the port does not classify.
    pub const F_OTHER: u8 = 0;
    /// A load or store its page tables do not allow; the address is the data address.
    pub const F_MEMORY: u8 = 1;
    /// An instruction fetch its page tables do not allow; the address is the fetch address.
    pub const F_INSTRUCTION: u8 = 2;
    /// An instruction the CPU will not run; the address is the instruction's.
    pub const F_ILLEGAL: u8 = 3;
    /// x86's general protection fault; the address is the instruction's.
    pub const F_PROTECTION: u8 = 4;
    /// A misaligned access the CPU refuses.
    pub const F_ALIGNMENT: u8 = 5;
}

/// The supervisor's command channel wire format.
///
/// ONE definition. This lived in three services at once (`supervisor`, `control`, `shell`) - three
/// copies of a protocol, which is the second-truth problem this project rejects everywhere else
/// (26.4, Commandment III). A drifted copy would not fail loudly; it would send a byte the supervisor
/// reads as a different opcode.
///
/// The supervisor's endpoint receives two kinds of message, told apart by the first byte:
///
/// ```text
///   death notice   "pong" 0 how...               <- kernel-generated; a name, [a-z0-9-], then how it
///                                                   ended (`death`, CLAUDE.md 6.2 amendment 2026-10-11)
///   command        0x01 op <core u32 LE> name    <- 0x01 cannot begin a name
///                                    name may be followed by NUL-separated PEER names
/// ```
///
/// Choosing an impossible first byte means a death notice needs nothing to set it apart: it still
/// begins with the name, as it always has (what follows the NUL is `death`'s, added 2026-10-11), so
/// the two are unambiguous without the kernel learning anything about commands.
pub mod supcmd {
    /// First byte of a command. Not a legal first byte of a service name.
    pub const MARKER:  u8 = 0x01;
    /// Kill if alive, then spawn.
    pub const RESTART: u8 = b'R';
    /// Spawn a service that is not running.
    pub const SPAWN:   u8 = b'S';
    /// Which devices does the supervisor drive? Read only: `[MARKER, DEVICES]`, answered with
    /// `[OK, count, entries...]` - per entry `b'H'`, the spawn row's device word (u32 LE, `hwclass`),
    /// the service's name (length byte, bytes); or `b'U'`, a USB device it starts a driver for (vid
    /// u16 LE, pid u16 LE, attached 0/1, the reporting host's name, the driver's name, each a length
    /// byte and bytes). The `hardware` utility's DRIVER column (`docs/hardware-design.md`).
    pub const DEVICES: u8 = b'D';
    /// Why is this driver's device handled as it is? Read only: `[MARKER, WHY, name...]`, answered
    /// with `[OK, text...]` - the reason the supervisor keeps beside the spawn rows - or `[UNKNOWN]`
    /// for a service with none recorded. `hardware why` (`docs/hardware-design.md` 11).
    pub const WHY:     u8 = b'W';
    /// What has happened to the devices this supervisor drives since it started. Read only:
    /// `[MARKER, EVENTS]`, answered with `[OK, since u32 LE, recorded u32 LE, count, entries...]` - since:
    /// when this supervisor began its record (after its boot spawns), in seconds of uptime; recorded: every
    /// event noted, so `recorded - count`
    /// were overwritten; per entry the uptime second (u32 LE), an `EV_*` code, the USB vid and pid (u16
    /// LE each, 0 for a driver event), and the driver's name (length byte, bytes). `hardware events`.
    pub const EVENTS:  u8 = b'E';
    /// Which services does this supervisor keep running right now? Read only: `[MARKER, MANAGED]`,
    /// answered with `[OK, count, (len, name)*]` - the `MANAGED` services this build carries and wants
    /// now (a USB device's driver only while its device is attached). The question `ask`'s health
    /// answer turns on: a service on this list that is not running is down when it should be up.
    pub const MANAGED: u8 = b'M';
    /// How did this service last end? Read only: `[MARKER, DEATH, name...]`, answered with `[OK, since
    /// u32 LE, deaths u32 LE, secs u32 LE, how, kind, addr u64 LE, by_len, by...]` - since: when this
    /// supervisor began its record; deaths: the notices it has read for this name since then (0 = none,
    /// and the rest is zero); then the newest one: its uptime second, how it ended (`death::KILLED`,
    /// `death::FAULT`), the fault kind and address, and the name of the task that called `kill`. The
    /// kernel says how, in the death notice; the supervisor only keeps the newest per name. `ask`.
    pub const DEATH:   u8 = b'X';
    /// Start an installed package: `[MARKER, INSTALL, "publisher/name version"]`. The supervisor checks the fetched
    /// copy from the disk, grants the send peers its `grant` file names and its contract requests, and starts it
    /// under the task name `publisher/name` (`services/supervisor/src/installed.rs`). Answered `[OK]`, or
    /// `[FAILED, reason...]` - the reason is a sentence.
    pub const INSTALL: u8 = b'I';
    /// Stop an installed package and forget it: `[MARKER, UNINSTALL, "publisher/name"]`. `[OK]`, or `[UNKNOWN]` when
    /// it was not installed.
    pub const UNINSTALL: u8 = b'O';
    /// Which packages has this supervisor started from the disk? Read only: `[MARKER, INSTALLED]`, answered with
    /// `[OK, count, (len, name, running 0/1)*]`. `godspeed list installed`.
    pub const INSTALLED: u8 = b'L';
    /// Replace a running installed package with another version WITHOUT STOPPING IT: `[MARKER, HANDOVER, "publisher/name
    /// version"]`. The supervisor checks the new version, starts it beside the running one, and tells the running one to
    /// hand over (`gs::handover`). Where either version cannot - it is not running, or a contract does not declare
    /// `handover` - it stops the old and starts the new, as INSTALL after UNINSTALL would. Answered `[OK, how]`, `how`
    /// `b'H'` handed over live or `b'R'` restarted, or `[FAILED, reason...]`.
    pub const HANDOVER: u8 = b'H';
    /// A device's driver died and was restarted.
    pub const EV_RESTARTED: u8 = 1;
    /// A device's driver died and its restart failed.
    pub const EV_RESTART_FAILED: u8 = 2;
    /// A USB device was reported attached and its driver started.
    pub const EV_ATTACHED: u8 = 3;
    /// A USB device was reported gone and its driver stopped.
    pub const EV_DETACHED: u8 = 4;
    /// A device's driver was found dead by the sweep - its death notice never arrived - and restarted.
    pub const EV_SWEPT: u8 = 5;
    /// A USB device was reported attached and its driver could not be started.
    pub const EV_START_FAILED: u8 = 6;

    /// Reply status, one byte, so a caller can log the truth rather than assume success.
    pub const OK:      u8 = 0;
    pub const FAILED:  u8 = 1;
    pub const UNKNOWN: u8 = 2;

    /// Longest command payload: opcode + core + a name and its peers, each at most `NAME_MAX` bytes.
    pub const MAX: usize = 6 + (super::MAX_SEND_PEERS + 1) * (super::NAME_MAX + 1);

    /// Build `MARKER op <core> name[\0peer]*` into `buf`, returning its length.
    pub fn encode(buf: &mut [u8; MAX], op: u8, core: u32, name: &str, peers: &[&str]) -> Option<usize> {
        if name.is_empty() { return None; }
        buf[0] = MARKER;
        buf[1] = op;
        buf[2..6].copy_from_slice(&core.to_le_bytes());
        let mut n = 6usize;
        let mut put = |b: &[u8], buf: &mut [u8; MAX], n: &mut usize| -> bool {
            if *n + b.len() > MAX { return false; }
            buf[*n..*n + b.len()].copy_from_slice(b);
            *n += b.len();
            true
        };
        if !put(name.as_bytes(), buf, &mut n) { return None; }
        for p in peers {
            if !put(&[0u8], buf, &mut n) || !put(p.as_bytes(), buf, &mut n) { return None; }
        }
        Some(n)
    }
}

/// A USB host's report to the supervisor about a device it does not drive itself, so the supervisor can
/// start that device's driver when it arrives and stop it when it leaves (`docs/usb-device-drivers.md`).
///
/// The host reports FACTS - present or not, VID:PID, how many times it has bound one - and never names a
/// driver: which driver a device gets is the supervisor's table. Sent with `try_send` and never answered,
/// so a host never waits on the supervisor (8.9). Each report is the host's whole state for that device,
/// not a change, so a duplicate is harmless and a lost one is corrected by the next.
pub mod usbdev {
    /// The report's opcode, after `supcmd::MARKER`: `[MARKER, REPORT, present, gen(4 LE), vid(2 LE), pid(2 LE), host]`.
    pub const REPORT: u8 = b'U';
    /// Supervisor to host, one byte and no reply capability: send your report again. How a respawned
    /// supervisor learns what is attached (6.2). In the host's request space beside `usbfn`'s ops (0x20 up),
    /// clear of them and of every block and net op.
    pub const ASK: u8 = 0x26;
    pub const LEN: usize = 12;

    /// Which host a report is from: a PC has two that bind a device for a driver of its own, `xhci` and `ehci`, and
    /// each reports only what IT has bound - so "nothing attached" from one says nothing of the other's.
    pub const HOST_DWC2: u8 = 0;
    pub const HOST_XHCI: u8 = 1;
    pub const HOST_EHCI: u8 = 2;
    /// How many host numbers there are: a table indexed by one is this long.
    pub const HOSTS: usize = 3;

    /// The service a host number names, or `None` for a number that is not one.
    pub fn host_name(host: u8) -> Option<&'static str> {
        match host {
            HOST_DWC2 => Some("dwc2"),
            HOST_XHCI => Some("xhci"),
            HOST_EHCI => Some("ehci"),
            _ => None,
        }
    }

    /// One report. `gen` counts every bind on that host, so the same dongle bound again - a replug, or the
    /// host re-enumerating it - reads differently from the same dongle still bound.
    #[derive(Clone, Copy, PartialEq, Eq, Debug)]
    pub struct Report {
        pub present: bool,
        pub gen: u32,
        pub vid: u16,
        pub pid: u16,
        /// The host it is from (`HOST_*`).
        pub host: u8,
    }

    pub fn encode(r: &Report) -> [u8; LEN] {
        let mut b = [0u8; LEN];
        b[0] = super::supcmd::MARKER;
        b[1] = REPORT;
        b[2] = r.present as u8;
        b[3..7].copy_from_slice(&r.gen.to_le_bytes());
        b[7..9].copy_from_slice(&r.vid.to_le_bytes());
        b[9..11].copy_from_slice(&r.pid.to_le_bytes());
        b[11] = r.host;
        b
    }

    /// `None` for anything that is not exactly a report.
    pub fn decode(p: &[u8]) -> Option<Report> {
        if p.len() != LEN || p[0] != super::supcmd::MARKER || p[1] != REPORT || p[2] > 1 || host_name(p[11]).is_none() {
            return None;
        }
        Some(Report {
            present: p[2] == 1,
            gen: u32::from_le_bytes([p[3], p[4], p[5], p[6]]),
            vid: u16::from_le_bytes([p[7], p[8]]),
            pid: u16::from_le_bytes([p[9], p[10]]),
            host: p[11],
        })
    }
}


// (Three probe-parameter flags that once rode in the upper bits of `Spawn`'s `arg0` sat here. Nothing
// set them after the kernel's probe path went, and `Spawn` itself was retired 2026-10-10; deleted the
// same day, `backlog/fixed/80` K14 and K20.)

/// Upper bound on a spawn name payload (`name` + NUL-separated peers). Matches the kernel's limit.
pub const SPAWN_PAYLOAD_MAX: usize = 128;

/// The most one boot-record read copies (InspectKernel query 27). Equal to the kernel's
/// `log::BOOT_READ_CHUNK`; a reader's buffer is exactly this size, so the kernel never writes past it.
pub const BOOT_READ_CHUNK: usize = 512;

/// MUST match `kernel::task::SERVICE_CTX_MAGIC`. It names the page's LAYOUT, and changes when the layout does.
const SERVICE_CTX_MAGIC:   u32   = 0xD0_5D_EA_D6;
/// MUST match `kernel::task::MAX_SEND_PEERS` - this indexes the kernel-written context page.
const MAX_SEND_PEERS:      usize = 6;
/// MUST match `kernel::task::PEER_NAME_BYTES`: a send-peer's name in the page the kernel writes.
const PEER_NAME_BYTES:     usize = NAME_MAX;
/// The longest name the kernel holds - a task's, a send-peer's, a directory entry's - and so the longest a
/// program can run under and be reached by. 128 since 2026-10-18: an installed package runs under its
/// `publisher/name`, which may be that long. It was 24 for a peer and for acquiring by name, and 32 for a
/// task, and a 26-byte package name was then unreachable by its own clients. Growing it changed the context
/// page's layout, so its magic changed with it, and a package built before that is refused at start
/// (`LAYOUT_RELEASE`).
pub const NAME_MAX:        usize = 128;
/// The first GodspeedOS release whose program layout this one runs: the context page and `TaskStat`, which
/// changed when names grew to 128 bytes. A package built for an earlier release would read them wrongly, so
/// the supervisor refuses to start it and says its publisher must rebuild it.
pub const LAYOUT_RELEASE:  &str  = "0.29.0";

#[repr(C)]
struct SendPeerEntry {
    slot:     u32,
    name_len: u32,
    name:     [u8; PEER_NAME_BYTES],
}

/// Layout of the kernel-written page at SERVICE_CTX_ADDR.
#[repr(C)]
struct ServiceContextData {
    magic:              u32,
    log_write_slot:     u32,
    recv_slot:          u32,
    spawn_slot:         u32,
    send_peer_count:    u32,
    core_id:            u32,
    probe_mode:         u32,
    console_read_slot:  u32, // u32::MAX = not present
    xhci_mmio_va:       u64, // 0 = not mapped; else VA of the mapped xHCI BAR
    xhci_mmio_len:      u64, // length of the mapped MMIO register window in bytes (SEC-4)
    xhci_dma_va:        u64, // 0 = none; else VA of the driver's DMA arena
    xhci_dma_phys:      u64, // physical base of the DMA arena
    xhci_dma_len:       u64, // length of the DMA arena in bytes
    console_push_slot:  u32, // u32::MAX = none; else CONSOLE_PUSH cap slot
    self_grant_slot:    u32, // u32::MAX = none; else SEND|GRANT cap to own endpoint (H11)
    // --- Framebuffer grant (the `console` service only) ---
    // The kernel maps the display's framebuffer into this service's address space Normal NON-cacheable
    // + USER, as a driver's MMIO BAR is mapped, and describes it here. Deliberately PIXEL geometry only:
    // no rows, no columns, no cell size. Character geometry belongs to the terminal, and the terminal is
    // the service (`docs/console-service.md` §9.7).
    fb_va:              u64, // 0 = no framebuffer grant; else VA of the mapped framebuffer
    fb_len:             u64, // length of the mapping in bytes (pitch * height)
    fb_pitch:           u32, // bytes per scanline
    fb_width:           u32, // visible width in pixels
    fb_height:          u32, // visible height in pixels
    fb_bpp:             u32, // bytes per pixel
    fb_shifts:          u32, // r_shift | g_shift << 8 | b_shift << 16
    send_peers:         [SendPeerEntry; MAX_SEND_PEERS],
    /// A SECOND endpoint, for REPLIES only. `u32::MAX` = none.
    ///
    /// A service that serves clients on the endpoint it also awaits replies on cannot drain that
    /// endpoint while it is blocked for a reply. Sixteen client requests arrive, the queue is full,
    /// and the reply it is waiting for is DROPPED by a peer that (correctly) uses `try_send` rather
    /// than deadlocking. The wait then runs to its full deadline - 30 s per block operation on x86,
    /// which is what made `write append` take 73 seconds.
    ///
    /// Correlation tags cannot reach this: a tag identifies a reply that ARRIVED, and this one never
    /// did. `docs/net-tags-design.md` rejected a second endpoint for lacking a `CreateEndpoint`
    /// syscall - true, and not needed: the first endpoint is minted at spawn and so is this one.
    reply_recv_slot:    u32,
    /// SEND|GRANT cap to `reply_recv_slot`'s endpoint, for handing out as a reply cap. `u32::MAX` = none.
    reply_grant_slot:   u32,
    /// The interrupt VECTOR(s) this service was granted, so a driver can learn which one it got.
    ///
    /// The kernel used to hand every driver a vector it could also hardcode: `XHCI_MSI_VECTOR` was a
    /// kernel constant, so both sides knew 0x28 statically and a driver comparing against its own copy
    /// was correct. The MSI pool (step D1b) makes the vector ALLOCATED AT SPAWN, and that quietly
    /// invalidated the hardcoded copy: the kernel began routing 0x30 while `services/xhci` still tested
    /// for 0x28, so it classified every real interrupt as an unknown message and reported `0 MSI, 7799
    /// msg` on hardware. Nothing broke - the interrupt still WOKE the driver - but the one instrument
    /// that says whether interrupts work had started reporting the opposite.
    ///
    /// The design gap was this field's absence, not the stale constant. A driver already learns its
    /// MMIO window and DMA arena from the kernel (`ctx.mmio()`, `ctx.dma_region()`) precisely because
    /// it may not name them; its vector is the same kind of fact and now travels the same way. A
    /// caller still cannot ASK for a particular vector - that is authority (§12.3) - it can only be
    /// told which one it was given.
    irq_count:          u32,      // 0 = no interrupt granted
    irqs:               [u8; 4],  // the granted vectors; only the first `irq_count` are meaningful
}

// The kernel writes this struct and the SDK reads it, from two crates, with no shared definition -
// they are kept in step BY HAND. There was no check on that, and adding a field to one and not the
// other silently misaligns every field after it: a service would read its neighbour's slot numbers.
//
// Pinned by SIZE in both crates. It does not prove field ORDER, but it catches the mistake that
// actually happens - an append on one side only - and it fails at compile time in the crate that
// drifted rather than at boot in a service that reads garbage.
const SERVICE_CONTEXT_DATA_SIZE: usize = 952;   // 320 + irq_count(4) + irqs[4] (D1b); then each of the 6 send-peer names 24 -> 128 bytes (2026-10-18), +624
const _: () = assert!(
    core::mem::size_of::<ServiceContextData>() == SERVICE_CONTEXT_DATA_SIZE,
    "ServiceContextData changed size: update BOTH kernel/src/task/mod.rs and      sdk/rust/src/service_context.rs, then update SERVICE_CONTEXT_DATA_SIZE in both"
);


// ---------------------------------------------------------------------------
// Dynamic send-cap cache - updated by `reacquire_cap` after EndpointDead.
// Safe: each service is a single-threaded process with its own BSS.
// ---------------------------------------------------------------------------

/// How many reacquired peers a service can hold at once.
///
/// **It must cover every peer a service names, because a peer that does not fit used to be dropped in
/// silence.** It was 8. The shell is wired at spawn with `fs` alone and reaches every other peer through
/// this cache, and it names more than eight - net-stack, wifi-driver, events, time, console, block-driver,
/// power, supervisor, nic-driver, recorder and more. After a chaos storm the respawned shell's `selfcheck`
/// filled the cache before reaching the network, so `reacquire_by_name("net-stack")` returned `true` while
/// storing nothing: every `ping` then found no send slot, reported `net-stack not responding` at once, and
/// leaked the capability it had just acquired - and killing net-stack could not help (Pi 4, 2026-10-04).
///
/// 32 is above the supervisor's whole managed roster. A full cache now EVICTS (see `reacquire_cap_detail`)
/// and says so, rather than refusing the newcomer quietly.
const CACHE_SIZE: usize = 32;

/// The next entry to evict when the cache is full - round robin, so no one peer is always the victim.
// SAFETY: single-threaded service process; no concurrent access.
static mut CACHE_NEXT_EVICT: usize = 0;

struct CacheEntry {
    slot:     u32,
    name_len: u8,
    name:     [u8; NAME_MAX],
}

impl CacheEntry {
    /// ALL ZEROS, and that is load-bearing: an empty entry is one with no name (`name_len == 0`), so the
    /// whole table is zero and lives in `.bss`, which costs a program's image nothing. It used to mark empty
    /// with `slot: u32::MAX`, which put the table in `.data` - stored in every program's file - and when names
    /// grew to 128 bytes that was 4 KiB more in each of the forty-odd services the Pi 2's kernel embeds, and
    /// pushed that kernel past the context page it must end below.
    const fn empty() -> Self {
        CacheEntry { slot: 0, name_len: 0, name: [0u8; NAME_MAX] }
    }
}

// SAFETY: single-threaded service process; no concurrent access.
static mut SEND_CAP_CACHE: [CacheEntry; CACHE_SIZE] =
    [const { CacheEntry::empty() }; CACHE_SIZE];

// ---------------------------------------------------------------------------
// TaskStat - returned by ServiceContext::task_stat.
// ---------------------------------------------------------------------------

/// Snapshot of kernel task state for a single scheduler slot.
#[derive(Clone, Copy)]
pub struct TaskStat {
    /// True if the slot holds a live task.
    pub valid:       bool,
    /// Task state: 0=Ready, 1=Running, 2=BlockedOnRecv, 3=BlockedOnSend, 4=Dead.
    pub state:       u8,
    /// Core the task is pinned to.
    pub core:        u8,
    /// Bytes dynamically allocated so far.
    pub mem_used:    u64,
    /// Maximum bytes the task may allocate.
    pub mem_limit:   u64,
    /// Byte length of the name stored in `name`.
    pub name_len:    usize,
    /// Task name bytes (zero-padded to `NAME_MAX` bytes).
    pub name:        [u8; NAME_MAX],
    /// Number of times this service has been restarted (0 on a fresh boot / first spawn, +1 per
    /// respawn). Saturating u64 - effectively unbounded.
    pub restart_count: u64,
    /// Current inbound IPC queue depth (0-16).
    pub queue_depth: u8,
    /// Timer ticks spent as the running task on its core (monotonic since boot).
    pub run_ticks:   u64,
    /// Seconds since this service last (re)started - resets on restart. Per-service uptime.
    pub uptime_secs: u64,
}

impl TaskStat {
    /// Return the task name as a `&str`.
    pub fn name_str(&self) -> &str {
        let len = self.name_len.min(NAME_MAX);
        core::str::from_utf8(&self.name[..len]).unwrap_or("?")
    }

    /// Return a short human-readable state label.
    pub fn state_str(&self) -> &'static str {
        match self.state {
            0 => "Ready",
            1 => "Running",
            2 => "BlockRecv",
            3 => "BlockSend",
            4 => "Dead",
            _ => "?",
        }
    }
}

/// One held capability, as reported by [`ServiceContext::task_caps`].
#[derive(Clone, Copy, Default)]
pub struct CapInfo {
    /// Resource the cap targets. Stable kernel resources: 1=log_write, 2=spawn,
    /// 3=console_read, 4=console_push, 5=introspect, 6..=18 the other kernel
    /// authorities (`kernel/src/capability/mod.rs`, e.g. 6=service_control,
    /// 8=reboot, 15=image_spawn); larger ids are IPC endpoints, and [4096, 6144)
    /// is the delegated-resource band (§7.10).
    pub resource_id: u64,
    /// Rights bitfield: READ=1, WRITE=2, SEND=4, RECV=8, GRANT=16, REVOKE=32.
    pub rights: u8,
}

/// THE KERNEL'S FIXED AUTHORITIES, by resource id: (id, the name a contract uses, what it lets a task do).
///
/// One table for everything that names them - the shell's `caps` and `ask "who can reboot?"` - so what a
/// service is observed to HOLD reads the same everywhere, and in the words its contract DECLARES
/// (`pci_cfg = true`). The ids are `kernel/src/capability/mod.rs`'s `*_RESOURCE`; anything above 18 is an
/// IPC endpoint. A resource added there and not here would be shown as an endpoint, which names the
/// wrong kind of thing (26.9), so the two move together.
pub const AUTHORITIES: &[(u64, &str, &str)] = &[
    (1, "log_write", "write to the kernel log and serial"),
    (2, "spawn", "start services"),
    (3, "console_read", "read the keyboard"),
    (4, "console_push", "type into the console as if at the keyboard"),
    (5, "introspect", "see every task, its capabilities and the kernel's counters"),
    (6, "service_control", "kill and restart services"),
    (7, "resource_mint", "mint capabilities for resources it owns, as fs does for files"),
    (8, "reboot", "reboot the machine"),
    (9, "acquire_any", "reach any service by name, not only its declared peers"),
    (10, "net_device", "move raw ethernet frames on an in-kernel network device (none exists; granted to nothing now)"),
    (11, "gpio_device", "drive the GPIO pins"),
    (12, "usb_disk", "read and write any sector of an in-kernel USB disk (none exists; granted to nothing now)"),
    (13, "set_clock", "set the wall clock (granted to nothing now)"),
    (14, "fire_irq", "inject a test interrupt"),
    (15, "image_spawn", "start a program from an image it supplies"),
    (16, "pci_cfg", "read and write PCI configuration space"),
    (17, "device_power", "cut and restore the power of the device it was granted"),
    (18, "cpu_clock", "set the Arm clock"),
];

// ---------------------------------------------------------------------------
// AllocError - returned by ServiceContext::alloc_mem.
// ---------------------------------------------------------------------------

/// Error from the AllocMem syscall (6).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AllocError {
    /// Allocation would exceed the task's memory limit (§10.3).
    Denied,
    /// Physical memory exhausted or other kernel-side failure.
    Failed,
}

// ---------------------------------------------------------------------------
// ServiceContext.
// ---------------------------------------------------------------------------

// (2026-10-09) HISTORY, kept as `//` so rustdoc does not hang it on `ServiceContext`. The wait
// helpers it describes BLOCK again now - each waits in `await_slice`
// (`RecvTimeout`), or in the kernel's `Call` / `CallDeadline` - so its first sentence is no longer true.
//
// These wait helpers POLL (`try_recv` + `yield_cpu`); they do not block. That is deliberate, and it is
// a REVERSAL - they were made to block earlier on this branch, and the change was wrong twice over.
//
// Why it was tried: a blocked task lets the core reach the scheduler's idle path, which saves power and
// (on ARM) lets idle-hook work run while a command waits. Both are real benefits.
//
// Why it is reverted:
// 1. **It broke x86 networking.** net-stack and nic-driver both sit on core 1, and every exchange
//    between them goes through `request_with_reply_deadline`. With blocking waits, net-stack degraded
//    to "no NIC MAC yet (driver absent/not ready)" and DHCP/ARP never completed; restoring the poll
//    made DHCP, ARP and a sustained 14 ms ping work immediately. Proven by an A/B on real hardware -
//    identical build, this one difference.
// 2. **It bought nothing that survived.** It was introduced to make USB hot-plug get noticed during a
//    `ping`. That symptom turned out to be the chaos harness pacing itself with a fixed yield count
//    (3000 yields x a full quantum = 30 s on the Pi), not these helpers. The hot-plug fixes that
//    actually worked were the hub change-latch and the catch-up sweep, neither of which needs this.
//
// It also exposed a genuine kernel bug on the way through - the BSP halting onto a consumed one-shot
// TSC deadline - which is FIXED and stays fixed independently of this revert.
//
// The lesson worth keeping: same-core request/reply through a blocking waiter is not exercised by the
// test suite (identity and fs-restart are all cross-core), so a change here cannot be validated by
// those suites. If blocking is attempted again, it needs a same-core request/reply test first.

/// Passed by the kernel to `service_main`. Non-Copy; one per service instance.
pub struct ServiceContext {
    _private: (),
}

/// The USB-disk syscalls' "device NAKed, re-ask" status.
///
/// The kernel claims no USB disk on any port any more (every arch's `usb_disk_sectors()` is 0, and the
/// USB stacks are services), so the four `usb_disk_*` syscalls have no caller in the tree; kept as the
/// record of their ABI.
///
/// Outside the capability-error range (-2..-7) ON PURPOSE, and this must stay in step with the kernel's
/// `USB_DISK_BUSY` (`kernel/src/syscall/dispatch.rs`). It was originally `-2`, which is `CapNotHeld`, so
/// a driver missing its `USB_DISK_RESOURCE` capability was indistinguishable from a busy device and got retried
/// thousands of times before being reported as a device that "stayed busy" - a cap failure wearing an
/// I/O failure's name.
pub const USB_DISK_BUSY: i64 = -20;

/// No USB disk is attached - the opposite instruction to [`USB_DISK_BUSY`], and it must stay in step with
/// the kernel's `USB_DISK_ABSENT`. BUSY means "come back, I am working"; ABSENT means "there is nothing
/// here, and asking again cannot change that - only a hot-plug can". A caller that retries on this waits
/// out its entire budget against an empty socket and then reports the device as busy, which is a false
/// statement about hardware sitting on the operator's desk.
pub const USB_DISK_ABSENT: i64 = -21;

/// Can this LBA survive the syscall ABI intact?
///
/// The ARM ABI gives each syscall argument exactly ONE 32-bit register (`arch/arm/CLAUDE.md`, hazard
/// A-U1), so a wider value is silently narrowed on the way in. That matters here more than anywhere:
/// the kernel guards `lba >= MSC_SECTORS` and then builds a READ(10)/WRITE(10) CDB, but it guards the
/// value it RECEIVED. An LBA of 0x1_0000_0000 arrives as 0, passes the range check, and overwrites the
/// GSFS superblock while the device, `block-driver` and `fs` all report success - a range check
/// defeated one level above itself, and silent corruption of exactly the block the check exists to
/// protect. On x86 the same request is rejected loudly, so the two ports would disagree about what is
/// a valid write.
///
/// Rejecting it in the wrapper is the fix the ABI hazard prescribes (clamp before the syscall, or split
/// into a register pair). Nothing legitimate is refused: capacity comes from SCSI READ CAPACITY(10),
/// whose last-LBA field is 32 bits wide, so an LBA above `u32::MAX` is a caller bug - and it is
/// reported as a failure rather than aliased onto a real block (Invariant 12, failures are loud).
#[inline]
fn lba_fits_syscall_abi(lba: u64) -> bool { lba <= u32::MAX as u64 }

impl ServiceContext {
    /// A handle for the PANIC HANDLER, which has no `ctx` to borrow.
    ///
    /// This type is a zero-sized token - every method reads the kernel-written context block rather
    /// than any field - so making one costs nothing and grants nothing that was not already granted.
    /// It is `pub(crate)` because the panic path is the only caller that legitimately lacks a context:
    /// a service is handed one and must thread it, and an escape hatch for anyone else would let a
    /// caller pretend to authority it was never given.
    pub(crate) fn for_panic() -> Self {
        ServiceContext { _private: () }
    }

    #[inline]
    fn ctx() -> &'static ServiceContextData {
        // SAFETY: kernel maps a valid ServiceContextData at SERVICE_CTX_ADDR
        // before SYSRETQ into the service; page is read-only and lifetime-stable.
        unsafe { &*(SERVICE_CTX_ADDR as *const ServiceContextData) }
    }

    /// Look up a named capability from this service's cap table.
    pub fn capability(&self, name: &str) -> Result<CapHandle, CapError> {
        let data = Self::ctx();
        if data.magic != SERVICE_CTX_MAGIC { return Err(CapError::CapNotHeld); }
        match name {
            "log_write" if data.log_write_slot != u32::MAX =>
                Ok(CapHandle(data.log_write_slot)),
            "spawn" if data.spawn_slot != u32::MAX =>
                Ok(CapHandle(data.spawn_slot)),
            "recv" if data.recv_slot != u32::MAX =>
                Ok(CapHandle(data.recv_slot)),
            _ => Err(CapError::CapNotHeld),
        }
    }

    /// Publish `msgs.received = 0` the first time this service WAITS for a message.
    ///
    /// The third case of the same bug the comment inside `count_recv` records twice. Publishing on the first
    /// message covers a service that is quiet; republishing on a sink change covers a sink that
    /// restarted. Neither covers a SERVICE that restarted and has not been spoken to since - and that
    /// is the ordinary state of `block-driver` after a chaos storm, because nothing sends to it until
    /// somebody does disk I/O. Observed on the Wyse: 1917 log lines between
    /// `supervisor: block-driver restarted` and a metrics query that found no row for it, with zero
    /// messages received in between. The selfcheck caught it as
    /// `where owner contains block-driver | assert contains msgs.received`.
    ///
    /// A row that is absent reads as dead, which is the exact question this metric exists to answer -
    /// so an idle live service must have one.
    ///
    /// THE ZERO IS EARNED, which is the only kind worth printing: the service is running, it has
    /// entered its receive loop, and it has genuinely received nothing. That is a fact about the
    /// system, not a placeholder standing in for one. Emitted once per instance, from the wait rather
    /// than from startup, because reaching the wait is what proves the service got as far as serving.
    #[inline]
    fn note_listening(&self) {
        static ANNOUNCED: core::sync::atomic::AtomicBool =
            core::sync::atomic::AtomicBool::new(false);
        if !ANNOUNCED.swap(true, core::sync::atomic::Ordering::Relaxed) {
            self.metric("msgs.received", 0);
        }
    }

    /// Count one RECEIVED message, and publish the running total every 64.
    ///
    /// In the SDK rather than in each service, for the same reason trace emission is: it is the same
    /// counter in every one of them, and ten hand-placed copies is ten chances to place it wrong. Every
    /// receive path funnels here, and only a message that actually ARRIVED is counted - a polling
    /// driver calling `try_recv` in a tight loop counts nothing until something comes.
    ///
    /// WHY THIS ONE IS WORTH HAVING EVERYWHERE. Paired with the `age_s` column, it answers the
    /// question an operator actually has when something is wrong: not "how busy is this service" but
    /// "is it doing ANYTHING". A `block-driver` whose count last moved forty seconds ago is a
    /// different fault from one that is still receiving and failing, and nothing else in the system
    /// distinguishes them.
    ///
    /// Costs a service with no `events` cap one relaxed add and a not-taken branch; `events` itself
    /// holds no send cap to itself, so its own publish resolves to `u32::MAX` and returns - the same
    /// cut that stops it tracing its own sends. No recursion is possible.
    #[inline]
    fn count_recv(&self) {
        static SERVED: core::sync::atomic::AtomicU64 = core::sync::atomic::AtomicU64::new(0);
        const PUBLISH_EVERY: u64 = 64;
        let n = SERVED.fetch_add(1, core::sync::atomic::Ordering::Relaxed) + 1;
        // THE FIRST MESSAGE PUBLISHES TOO, not just every 64th. Publishing only on the interval meant a
        // service that had received fewer than 64 messages had NO ROW AT ALL - and a service with no row
        // is indistinguishable from one that is dead, which is precisely the question this metric exists
        // to answer. Observed: `block-driver`, `time`, `xhci`, `ehci` and `dwc2` all absent from a
        // healthy machine. One extra send per service for the life of the machine.
        // ...OR THE SINK JUST CAME BACK. A restarted sink has never heard of this service, and
        // waiting out the interval leaves it looking dead for up to 63 more messages - on a service
        // that only receives when someone else does I/O, that is however long the machine stays
        // quiet. Republishing on the first message after a reacquire is the metric half of "reacquire
        // and re-establish" (14.3). Observed: `block-driver` had no row for the whole window after a
        // chaos storm restarted the sink, purely because its next 64th message had not arrived.
        if n == 1 || n % PUBLISH_EVERY == 0 || crate::trace::take_sink_changed() {
            self.metric("msgs.received", n);
        }
    }

    /// Block until a message arrives on this service's primary recv endpoint.
    ///
    /// **Every failure here now PANICS rather than spinning.** All three exits used to be `loop {}`:
    /// a silent, logless, non-yielding tight spin that pegged the core. That is the worst possible
    /// response to `EndpointDead`, which is not corruption but an ordinary runtime truth (§8.6) - the
    /// service was told its endpoint died and answered by burning a core forever, telling nobody.
    ///
    /// Panicking is now the RIGHT answer because the panic handler was fixed in the same pass: it
    /// logs and faults, so the kernel kills the task, bumps its endpoint generation, wakes any peer
    /// blocked in `call` with `ReplyDead`, and the supervisor restarts it. Loud, and recovered.
    ///
    /// A service that wants to HANDLE the error instead of dying should call `recv_result`.
    pub fn recv(&self) -> Message {
        self.note_listening();
        let data = Self::ctx();
        if data.magic != SERVICE_CTX_MAGIC {
            panic!("recv: corrupt ServiceContext (bad magic) - the kernel handed us an unusable context");
        }
        let slot = data.recv_slot;
        if slot == u32::MAX {
            panic!("recv: no receive endpoint - this service has no ipc_receive in its contract");
        }
        match crate::ipc::recv(CapHandle(slot)) {
            Ok(msg) => { self.count_recv(); msg }
            Err(e)  => panic!("recv failed: {:?} - endpoint is gone; dying so the supervisor restarts us", e),
        }
    }

    /// Non-blocking receive on this service's primary recv endpoint: `Some(msg)` if a
    /// message was waiting, `None` if the queue is empty. A busy-polling driver uses this
    /// to drain interrupt events (§12) each loop iteration without blocking.
    ///
    /// A receive ERROR on this service's OWN endpoint (it is dead, or the cap was refused) is not an
    /// empty queue: the service dies loudly, exactly as [`Self::recv`] does, so the supervisor
    /// restarts it. It read as `None` until 2026-10-10, and a loop polling a dead endpoint spun on it
    /// forever (`backlog/fixed/80` S5).
    pub fn try_recv(&self) -> Option<Message> {
        self.note_listening();
        let data = Self::ctx();
        if data.magic != SERVICE_CTX_MAGIC { return None; }
        let slot = data.recv_slot;
        if slot == u32::MAX { return None; }
        let m = match crate::ipc::try_recv(CapHandle(slot)) {
            Ok(m) => m,
            Err(e) => panic!("try_recv failed: {:?} - endpoint is gone; dying so the supervisor restarts us", e),
        };
        if m.is_some() { self.count_recv(); }
        m
    }

    /// Block on this service's recv endpoint until a message arrives or `timeout_cycles`
    /// (cycle-counter ticks, the unit of `read_tsc` - convert with `duration_cycles`) elapse:
    /// `Some(msg)` = message, `None` = timed out. A receive ERROR on this service's own endpoint dies
    /// loudly, as [`Self::recv`] and [`Self::try_recv`] do (`backlog/fixed/80` S5).
    /// `timeout_cycles == 0` blocks forever. The kernel turns the count into whole scheduler ticks
    /// (at least one), so the bound is never finer than one quantum. A driver uses this to idle on
    /// its hardware interrupt while still waking on a timer for auto-repeat (§12 timed-wait).
    ///
    /// It returns a 4 KiB `Message` by value. It once carried `#[inline(always)]` to save that frame;
    /// 8f32a5e9 removed it, because inlining a body that holds a 4 KiB local into a caller that holds
    /// one puts both in the same frame.
    pub fn recv_timeout(&self, timeout_cycles: u64) -> Option<Message> {
        self.note_listening();
        let data = Self::ctx();
        if data.magic != SERVICE_CTX_MAGIC { return None; }
        let slot = data.recv_slot;
        if slot == u32::MAX { return None; }
        let m = match crate::ipc::recv_timeout(CapHandle(slot), timeout_cycles) {
            Ok(m) => m,
            Err(e) => panic!("recv_timeout failed: {:?} - endpoint is gone; dying so the supervisor restarts us", e),
        };
        if m.is_some() { self.count_recv(); }
        m
    }

    /// Re-open the kernel's IOAPIC gate for a level-triggered IRQ `vector` after this driver
    /// has cleared its device's interrupt source (§12). The kernel masks a level INTx while the
    /// driver handles it (so it can't storm); call this to let it fire again. Only the driver
    /// registered for `vector` (via its `hw_interrupt` route) may unmask it. No-op for MSI.
    pub fn irq_unmask(&self, vector: u8) {
        // SAFETY: syscall(36) = IrqUnmask; gated kernel-side by the route registration.
        let _ = unsafe { raw_syscall(36, vector as u64, 0, 0) };
    }

    /// Block this task for roughly `cycles` TSC cycles, then return (syscall 37). A real sleep:
    /// the core can halt while parked, so a poll/wait loop does not busy-`yield` (which pegs the
    /// core at ~100% and makes every task on it read as fully busy in `observe`). Like `yield`,
    /// needs no capability. A sleep under one quantum (10 ms) uses the arch's sub-tick timer where it
    /// has one (`hires_arm`), with a tick backstop; otherwise granularity is one scheduler quantum.
    /// Use for UI repaint pacing and "wait for child" loops - not for precise timing. On a 32-bit
    /// target `cycles` is clamped to what one register carries, so a long sleep is shortened to the
    /// longest the ABI can express rather than wrapping to a short one (`backlog/fixed/80` S5).
    pub fn sleep(&self, cycles: u64) {
        let cycles = cycles.min(usize::MAX as u64);
        // SAFETY: syscall(37) = Sleep; sleeping your own task is unprivileged (like yield).
        let _ = unsafe { raw_syscall(37, cycles, 0, 0) };
    }

    // `net_frame_tx`, `net_frame_rx` and `net_info` (syscalls 42-44, NET_DEVICE) were here until
    // 2026-10-11: no port has an in-kernel network device, nothing is granted NET_DEVICE, and nothing
    // called them (`backlog/21` step 2). The syscalls themselves are the last step, a kernel change.

    /// Block until a message arrives; returns the error instead of looping silently.
    pub fn recv_result(&self) -> Result<Message, crate::ipc::IpcError> {
        let data = Self::ctx();
        if data.magic != SERVICE_CTX_MAGIC { return Err(crate::ipc::IpcError::EndpointDead); }
        let slot = data.recv_slot;
        if slot == u32::MAX { return Err(crate::ipc::IpcError::EndpointDead); }
        crate::ipc::recv(CapHandle(slot))
    }

    /// Send to a named peer declared in `ipc_send`. Blocking.
    pub fn send(&self, peer: &str, msg: &Message) -> Result<(), IpcError> {
        let slot = self.find_send_slot(peer).ok_or(IpcError::CapError(CapError::CapNotHeld))?;
        crate::ipc::send(CapHandle(slot), msg)
    }

    /// Non-blocking send; returns `QueueFull` immediately if the queue is full.
    pub fn try_send(&self, peer: &str, msg: &Message) -> Result<(), IpcError> {
        let slot = self.find_send_slot(peer).ok_or(IpcError::CapError(CapError::CapNotHeld))?;
        crate::ipc::try_send(CapHandle(slot), msg)
    }

    /// Acquire a fresh SEND cap to `peer` via the kernel name directory.
    ///
    /// Called after `try_send` returns `EndpointDead` (§14.2). Updates the
    /// per-service dynamic cap cache so subsequent `try_send` calls use the
    /// new slot without going to the kernel again. Every failure collapses to
    /// `CapNotHeld`; `reacquire_cap_detail` says which.
    pub fn reacquire_cap(&self, peer: &str) -> Result<CapHandle, CapError> {
        self.reacquire_cap_detail(peer).map_err(|_| CapError::CapNotHeld)
    }

    /// `reacquire_cap`, but saying WHICH failure - see [`AcquireFailure`].
    ///
    /// Use this wherever the answer changes what the caller does or prints. The plain
    /// `reacquire_cap`/`reacquire_by_name` collapse every failure to one value, which is fine for a
    /// caller that only retries, and wrong for one that reports: a message naming two possible causes
    /// and committing to neither is a message that sends the reader to the wrong service (§26.7).
    pub fn reacquire_cap_detail(&self, peer: &str) -> Result<CapHandle, AcquireFailure> {
        let bytes = peer.as_bytes();
        let len   = bytes.len();
        if len == 0 || len > NAME_MAX { return Err(AcquireFailure::Refused); }

        // SAFETY: syscall(10) = AcquireSendCap; peer bytes are in user space.
        let ret = unsafe {
            raw_syscall(10, bytes.as_ptr() as u64, len as u64, 0)
        };
        if ret < 0 { return Err(AcquireFailure::from_code(ret)); }
        let new_slot = ret as u32;

        // Update the dynamic cache. Reuse this peer's EXISTING entry if it has one (and reclaim its
        // now-stale cap), otherwise take a free slot. Reclaiming the old cap is essential: without it
        // every restart-reacquire orphans the previous cap, and a storm (e.g. `chaos max-carnage`)
        // fills the cap table until `derive_cap`/`acquire_send_cap` start returning None - which shows
        // up as "storage unavailable" (fs) and "never registered" (pipe filters). Searching the peer's
        // entry first also prevents creating a duplicate entry when a free slot precedes it.
        // SAFETY: single-threaded service; no concurrent cache writes. addr_of_mut! avoids a direct
        // &mut to the static (static_mut_refs lint).
        //
        // NEVER DROP THE NEW CAP. If the peer has no entry and no entry is free, an existing one is
        // evicted - its cap reclaimed, its peer reacquired on next use - and the eviction is logged. This
        // used to fall through doing nothing and still return Ok: the caller believed the peer was
        // reacquired, `find_send_slot` could not find it, and the cap just acquired leaked (CACHE_SIZE).
        let mut stale: Option<u32> = None;
        let mut evicted: Option<([u8; NAME_MAX], usize)> = None;
        let mut placed = false;
        unsafe {
            let cache = &mut *core::ptr::addr_of_mut!(SEND_CAP_CACHE);
            for entry in cache.iter_mut() {
                if entry.name_len as usize == len && &entry.name[..len] == bytes {
                    if entry.slot != new_slot { stale = Some(entry.slot); }
                    entry.slot = new_slot;
                    placed = true;
                    break;
                }
            }
            if !placed {
                let free = cache.iter().position(|e| e.name_len == 0);
                let i = match free {
                    Some(i) => i,
                    None => {
                        let next = &mut *core::ptr::addr_of_mut!(CACHE_NEXT_EVICT);
                        let i = *next % CACHE_SIZE;
                        *next = next.wrapping_add(1);
                        stale = Some(cache[i].slot);
                        evicted = Some((cache[i].name, cache[i].name_len as usize));
                        i
                    }
                };
                let entry = &mut cache[i];
                entry.slot     = new_slot;
                entry.name_len = len as u8;
                entry.name     = [0u8; NAME_MAX];
                entry.name[..len].copy_from_slice(bytes);
            }
        }
        if let Some(old) = stale { self.remove_cap(CapHandle(old)); }
        if let Some((name, nlen)) = evicted {
            let gone = core::str::from_utf8(&name[..nlen.min(NAME_MAX)]).unwrap_or("?");
            self.log_fmt(format_args!(
                "sdk: the send-cap cache is full ({} peers) - dropped the cap to '{}' to hold '{}'; '{}' is reacquired when next used",
                CACHE_SIZE, gone, peer, gone));
        }

        Ok(CapHandle(new_slot))
    }

    /// Derive a duplicate of a capability this service holds **with the GRANT right**
    /// into a fresh slot (syscall 29 = `DeriveCap`). The copy carries the same
    /// resource, generation, and (non-widened) rights.
    ///
    /// Used to hand out many copies of a held endpoint cap: derive a copy per
    /// recipient and grant that copy away (via `send_with_cap_by_handle`) while
    /// keeping the original. Returns `None` if the cap lacks GRANT, is stale, or the
    /// cap table is full.
    pub fn derive_cap(&self, held: CapHandle) -> Option<CapHandle> {
        // SAFETY: syscall(29) = DeriveCap; `held.0` is a slot index into this task's
        // own cap table. The kernel validates GRANT + generation before duplicating.
        let ret = unsafe { raw_syscall(29, held.0 as u64, 0, 0) };
        if ret < 0 { None } else { Some(CapHandle(ret as u32)) }
    }

    /// Return the probe mode written by the kernel at spawn (0 for all production services).
    pub fn probe_mode(&self) -> u32 { Self::ctx().probe_mode }

    /// The REPLY mailbox: `(recv, grant)` for the endpoint that exists only to receive replies, or
    /// `None` on a task that has none (then the caller uses the shared endpoint, as before).
    ///
    /// Awaiting a reply on the endpoint you also SERVE means you cannot drain client traffic while
    /// you wait; the queue fills and your own reply is dropped. Replies come here instead, where
    /// nothing else is ever sent.
    fn reply_mailbox(&self) -> Option<(crate::capability::CapHandle, crate::capability::CapHandle)> {
        let d = Self::ctx();
        if d.reply_recv_slot == u32::MAX || d.reply_grant_slot == u32::MAX { return None; }
        Some((crate::capability::CapHandle(d.reply_recv_slot),
              crate::capability::CapHandle(d.reply_grant_slot)))
    }

    /// Discard replies left over from requests this service already gave up on. Returns how many.
    ///
    /// **A DETECTED DESYNC HAS TO BE REPAIRABLE, and this is the repair.** `fs` tags every block
    /// request and checks the tag on the reply, so a stale reply is always DETECTED - and detection
    /// was all there was. The mismatched reply was discarded and the request failed, which leaves this
    /// service's own reply still queued for the NEXT request to find. Every subsequent request then
    /// receives its predecessor's reply, discards it, fails, and queues another: permanently one reply
    /// out of phase, alternating forever, with the volume mounted and every operation failing.
    ///
    /// Observed 133 times in one `osdev test peer-storm` run, still alternating after the storm had
    /// stopped. The code that discarded expected the queue to drain on its own - "the caller retries
    /// at its own level, by which point the queue has drained" - and it never does, because each
    /// retry consumes exactly one stale reply and contributes exactly one more.
    ///
    /// WHY DRAINING IS SAFE HERE, which is the whole reason this is in the SDK rather than hand-rolled
    /// at a call site: it drains the dedicated REPLY MAILBOX, an endpoint that carries nothing but
    /// replies to calls this service made. A caller that is synchronous - one outstanding request at a
    /// time, which is what `request_with_reply*` enforces - can have nothing legitimate waiting there
    /// at the moment it is about to issue a new request. Anything present is a reply it stopped
    /// waiting for.
    ///
    /// **It refuses to drain the SHARED endpoint**, and that refusal is load-bearing. A service
    /// without a reply mailbox awaits replies on the endpoint its CLIENTS send to, so draining there
    /// would silently eat client requests - the precise hazard the `CallDeadline` amendment (§8.2)
    /// exists to describe. No mailbox, no drain, and the caller keeps the behaviour it had.
    ///
    /// SO THIS REPAIR IS CONDITIONAL, and the condition is not theoretical. A reply mailbox is an
    /// OPTIONAL endpoint (`routing::try_register_optional`): it is refused when the routing table is
    /// near full, and a bare-metal T630 refuses three times during `selfcheck`. A service refused one
    /// falls back to its shared endpoint and silently loses this repair - which is correct, because
    /// the alternative is eating client traffic, but it means "the desync is fixed" holds only where
    /// the mailbox was granted.
    ///
    /// Measured rather than assumed: an arm32 Pi 2 run logged ZERO refusals, so `fs` had its mailbox
    /// and the repair was live on that port. `spawn[ipc]: '<name>' gets no reply mailbox` in a log
    /// names each service in the other case (it read `routing: reply endpoint refused`, unnamed, until
    /// `backlog/74`).
    pub fn drain_stale_replies(&self) -> usize {
        let Some((recv, _)) = self.reply_mailbox() else { return 0 };
        let mut n = 0usize;
        // Bounded by the endpoint's own depth (§8.5): a queue that cannot hold more than 16 cannot
        // yield more than 16, so this terminates without needing to trust the peer.
        for _ in 0..16 {
            match crate::ipc::try_recv(recv) {
                Ok(Some(_)) => { self.release_discarded_cap(); n += 1 }
                _ => break,
            }
        }
        n
    }

    /// [`Self::drain_stale_replies`] at the top of a request, and SAID when it found anything - the one
    /// thing every request path that waits on the reply mailbox must do before it issues.
    ///
    /// It was done by ONE of the three such paths. `request_with_reply_deadline_into` drained (it is
    /// where `backlog/fixed/37` was fixed); `request_with_reply_call_err`, which `gs::call` and so every `gs::fs`
    /// call goes through, and `request_with_reply_deadline_outcome_into` did not. So the exact failure
    /// [`Self::drain_stale_replies`] describes was still live for every service on the standard library:
    /// in the jobs suite the shell's history save timed out while a background `drives check` held `fs`
    /// for five seconds, its reply arrived late, and from then on every shell request to `fs` received
    /// its predecessor's reply and rejected it as malformed - `churn verify` failing as "storage
    /// unavailable" and every later history save lost (2026-10-11). One helper now, so a path cannot
    /// have the drain without the log or the log without the drain.
    fn drain_before_request(&self, peer: &str) {
        let dropped = self.drain_stale_replies();
        if dropped > 0 {
            // LOUD, because it means a previous request was abandoned and this one would have
            // inherited its reply. Silent self-repair is how the fs desync stayed invisible for so
            // long (§26.7 - a recovery that hides what it recovered from is a silent fallback).
            self.log_fmt(format_args!(
                "sdk: discarded {} abandoned repl(y/ies) before a request to `{}` - an earlier request timed out",
                dropped, peer));
        }
    }

    /// Give back a capability a DISCARDED message may have carried. A received `Message` does not
    /// record its caps; the kernel queues them for `take_pending_cap`. Every caller claims a cap
    /// straight after its own receive, so what is pending while a drain runs came with the message it
    /// is discarding - and was leaked one table slot at a time until 2026-10-10 (`backlog/fixed/80` S5).
    fn release_discarded_cap(&self) {
        if let Some(c) = self.take_pending_cap() {
            self.remove_cap(c);
        }
    }

    /// Wait, bounded by `max_ms`, for up to `owed` replies to arrive in the reply MAILBOX, and discard
    /// them. Returns how many were discarded, or `None` when this service has no mailbox - in which case
    /// owed-reply accounting cannot work (replies land on the shared endpoint) and the caller must fall
    /// back. For a synchronous caller that gave up on `owed` requests whose answers are still in flight:
    /// `drain_stale_replies` clears what has ARRIVED; this also waits for what has not.
    pub fn drain_owed_replies(&self, owed: usize, max_ms: u64) -> Option<usize> {
        let (recv, _) = self.reply_mailbox()?;
        let t0 = self.read_tsc();
        let limit = self.duration_cycles(max_ms);
        let mut n = 0usize;
        while n < owed {
            match crate::ipc::try_recv(recv) {
                Ok(Some(_)) => { self.release_discarded_cap(); n += 1 }
                _ => {
                    if self.read_tsc().wrapping_sub(t0) >= limit {
                        break;
                    }
                    self.sleep_ms(10);
                }
            }
        }
        Some(n)
    }

    /// Return the recv cap handle for direct-handle use (e.g. wrong-right test probing).
    pub fn recv_handle(&self) -> Option<crate::capability::CapHandle> {
        let slot = Self::ctx().recv_slot;
        if slot == u32::MAX { None } else { Some(crate::capability::CapHandle(slot)) }
    }

    // (A stray doc block for `request_with_reply` sat here, attached by rustdoc to `trace_as` below. It
    // is a plain comment now; the function's own doc is at `request_with_reply`.)
    //
    // `request_with_reply` embeds a per-request reply cap - a `SEND|GRANT` copy of this service's own
    // endpoint cap - so the server can reply via `take_pending_cap()` + `send_by_handle()` (§8). Its
    // wait is a synchronous kernel CALL (syscall 41): if `peer` dies after receiving the request but
    // before replying, the kernel wakes the caller with `ReplyDead` (§8.6) instead of hanging it. It
    // returns `None` for that and for a failed send alike, which is exactly the distinction a caller
    // needs: `ReplyDead` means the request was DELIVERED and may have been acted on, so re-sending is
    // safe only for an operation that is safe to repeat. A service uses `gs::call::request_within`,
    // which keeps the two apart (`Error::PeerDied`, never re-sent, against `Error::Unreachable`).
    #[inline]
    /// Declare this service's own name for the trace ring, once, at startup.
    ///
    /// A service cannot ask what it is called - identity is not ambient - so a traced service says.
    /// Costs nothing in trust: the whole event is already the emitter's testimony (see `crate::trace`).
    /// A service that never calls this still traces; its events read `?` in the caller column, which
    /// is the honest answer rather than a guess.
    pub fn trace_as(&self, name: &str) {
        // A TRUNCATED SERVICE NAME IS REPORTED, not swallowed. `PEER_LEN` is 12 - sized for
        // `block-driver` - and `hw-enumerator` is 13, so it appears everywhere as `hw-enumerato`.
        // Readable, but it is still a name silently becoming a different name, and two services
        // sharing a 12-byte prefix would merge into ONE metric row with their counters interleaving
        // (the exact failure an unnamed service caused before every service declared itself).
        //
        // Widening `PEER_LEN` was the other option and was rejected on cost: the trace DUMP is bounded
        // by one 4 KiB message, so a 16-byte field drops it from 110 events per screen to 95. Trading
        // real scrollback in the tool for four characters of a name is the wrong way round. Say it
        // once instead, so the fix - a shorter name - is the developer's to make.
        if name.len() > crate::trace::PEER_LEN {
            self.log_fmt(format_args!(
                "sdk: service name '{}' exceeds {} bytes and is TRUNCATED in traces and metrics - shorten it, or two services could share one row",
                name, crate::trace::PEER_LEN));
        }
        crate::trace::set_caller(name);
    }

    /// Declare that requests to `peer` carry their opcode at byte `at` (default 0).
    ///
    /// For protocols that prepend a correlation tag - `shell -> fs` and `fs -> block-driver` both do -
    /// byte 0 is a request id, so a trace that recorded it was showing noise in a column labelled
    /// "op". The SDK is generic and cannot know; the service that owns the protocol can. See
    /// `crate::trace::set_op_offset`.
    pub fn trace_op_at(&self, peer: &str, at: u8) { crate::trace::set_op_offset(peer, at); }

    /// Emit one trace event, or do nothing at all.
    ///
    /// The whole cost when this service is not tracing is the `trace::should_resolve` and
    /// `trace::sink_slot` loads and a not-taken branch. When it IS tracing, the send is `try_send` and its result is DISCARDED on purpose: an
    /// observer must never be able to slow, block, or fail the thing it observes. A full events queue
    /// loses one event, which the ring counts and `events status` reports - a visible loss, not a
    /// silent one (invariant 12).
    ///
    /// Recursion is cut at the source: `events` itself holds no `events` send cap, so the sink cannot trace
    /// its own sends, and no event can beget another.
    fn trace_emit(&self, peer: &str, op: u8, kind: u8) {
        // Arm lazily, ONCE. A service is handed a context with no init hook to run in, so there is
        // nowhere else to resolve from. A service whose contract does not grant `ipc_send =
        // ["events"]` records `u32::MAX` here and never looks again: one relaxed load per call for the
        // rest of its life, which is the cost the untraced majority pays.
        if crate::trace::should_resolve() {
            let mut slot = self.find_send_slot(crate::trace::SINK_NAME).unwrap_or(u32::MAX);
            if slot == u32::MAX {
                // NO CAP WIRED AT SPAWN - ASK THE KERNEL FOR ONE. `find_send_slot` only inspects caps
                // this service ALREADY HOLDS, so a service spawned during the milliseconds `events`
                // was restarting has nothing to find, resolves to MAX, and returns early on every
                // emission for the rest of its life. The reacquire that would fix it sits behind a
                // send FAILURE - and a send that never happens never fails, so it was unreachable.
                //
                // Measured on the T630: after a storm that restarted the sink ~48 times, `fs` and
                // `block-driver` published NOTHING for the remaining 9.4 s of the run, while the
                // sink's own LOCALLY-written rows were all present. That split - local rows yes, every
                // IPC-delivered row no - is the signature of a publisher whose message never left.
                //
                // `AcquireSendCap` is authorised by the DECLARATION, which the kernel keeps now even
                // when the peer was down at spawn. This is the call that spends it (14.3).
                if self.reacquire_by_name(crate::trace::SINK_NAME) {
                    slot = self.find_send_slot(crate::trace::SINK_NAME).unwrap_or(u32::MAX);
                }
            }
            crate::trace::set_sink_slot(slot);
            // AND SAY IT, IF IT TOOK MORE THAN ONE TRY. This report was written once already and
            // silently lost: the edit that added it failed to apply and only its trace.rs half landed,
            // so `take_late_resolve` sat with no caller and the hardware run reported "0 late
            // resolves" - a zero that measured nothing. A missing instrument reads exactly like a
            // healthy system, which is the worst way for one to fail.
            if slot != u32::MAX {
                if let Some(missed) = crate::trace::take_late_resolve() {
                    if missed > 1 {
                        self.log_fmt(format_args!(
                            "sdk: events sink resolved LATE - {} emission(s) went nowhere first",
                            missed - 1));
                    }
                }
            }
        }
        let slot = crate::trace::sink_slot();
        if slot == u32::MAX {
            return;
        }
        // Do NOT record a call to the SINK ITSELF. The `trace` utility reads the ring by asking
        // `events` for it, so tracing that call would fill the ring with the reader's own questions -
        // each dump adding two more events, pushing out the traffic the reader came to see. An
        // instrument that mostly measures itself is not measuring anything. This hides nothing: every
        // OTHER peer is still recorded, and `events status` still counts every event accepted.
        if peer == crate::trace::SINK_NAME {
            return;
        }
        // NO CLOCK READ HERE. `epoch_secs_monotonic` is a full CMOS RTC read: `wait_update_clear`
        // spins until the update-in-progress flag clears (up to ~1 ms), then seven port-I/O reads,
        // repeated until two agree. Two of those per request/reply turned `dir`, `move`, `find` and tab
        // completion into TIMEOUTS, and cost the shell so much time inside the kernel that it stopped
        // draining the console and lost Enter keystrokes. Measured, not guessed: `osdev test files`
        // was 222/0 before, 213/9 with the clock read, 222/0 again without it.
        //
        // The SINK stamps the event instead. `events` has to wake to receive it either way, so the
        // cost lands on the service whose job this is rather than on every service that talks to
        // anyone - and the caller already used `try_send` and did not wait. An observer must not be
        // able to slow the thing it observes, and a millisecond of port I/O per IPC is exactly that.
        let ev = crate::trace::encode(peer, op, kind);
        if self.try_send_by_handle(CapHandle(slot), &crate::ipc::Message::from_bytes(&ev)).is_err() {
            // THE SINK RESTARTED - REACQUIRE IT. The slot is resolved once and cached, so a `events`
            // respawn left every emitter holding a stale generation FOREVER: tracing silently stopped,
            // `events ipc` could not reach the ring, and each emission logged a kernel gen-mismatch.
            // Seen on hardware after a chaos storm restarted `events` forty times - cap generation 985
            // against a live record of 1025.
            //
            // This is the ordinary reacquire-by-name recovery every client owes a restartable peer
            // (14.3); the trace path just never did it. Only on FAILURE, so the healthy path is
            // unchanged, and if the sink is genuinely down the next emission tries again.
            //
            // AND RETRY ON THE FRESH SLOT. Reacquiring but dropping the event that prompted it is a
            // recovery that discards its own result (26.7) - the peer is reachable again and the
            // sample is thrown away anyway. 14.3 says reacquire AND retry; this used to do half.
            if self.reacquire_by_name(crate::trace::SINK_NAME) {
                if let Some(fresh) = self.find_send_slot(crate::trace::SINK_NAME) {
                    crate::trace::set_sink_slot(fresh);
                    let _ = self.try_send_by_handle(
                        CapHandle(fresh),
                        &crate::ipc::Message::from_bytes(&ev),
                    );
                }
            }
        }
    }

    /// Publish a metric sample to `events`. Fire-and-forget, and never able to slow this service down.
    ///
    /// The value is a SET, not an increment: this service owns the counter (it is this service's own
    /// state, §3.8) and publishes what it currently reads; `events` remembers the last sample. So a
    /// service that dies leaves its final value behind rather than taking it along, which is the one
    /// useful thing an observer can still learn from a service that is gone.
    ///
    /// Shaped exactly like `trace_emit`, and the shape is the point:
    /// - `try_send`, result DISCARDED. An observer must never slow, block or fail the thing it
    ///   observes. A full sink loses one sample, which the sink counts and reports (invariant 12).
    /// - No clock read here. The sink stamps it; a CMOS RTC read on every publish once cost the shell
    ///   its keystrokes.
    /// - Reacquire on failure only, so the healthy path is unchanged and a restarted sink does not
    ///   leave every emitter holding a stale generation forever (14.3).
    ///
    /// SELF-OBSERVATION NEEDS NO CHECK HERE, and that is deliberate. `events` holds no send cap to
    /// `events`, so this resolves to `u32::MAX` inside its own address space and returns - the same cut
    /// that stops the sink tracing its own sends. A service reporting on itself must never take an IPC
    /// hop, because the hop is itself a reportable event and the ring would fill with itself. `events`
    /// counts its own work locally instead (`docs/observability.md` 9).
    pub fn metric(&self, name: &str, value: u64) {
        // FUNCTION-LOCAL, so the latch is owned by the only code that reads it (§3.8, invariant 9).
        // A one-shot: it never resets, because the point is to say it once for the life of the service.
        static TRUNCATED_SAID: core::sync::atomic::AtomicBool =
            core::sync::atomic::AtomicBool::new(false);
        if crate::trace::should_resolve() {
            let mut slot = self.find_send_slot(crate::trace::SINK_NAME).unwrap_or(u32::MAX);
            if slot == u32::MAX {
                // NO CAP WIRED AT SPAWN - ASK THE KERNEL FOR ONE. `find_send_slot` only inspects caps
                // this service ALREADY HOLDS, so a service spawned during the milliseconds `events`
                // was restarting has nothing to find, resolves to MAX, and returns early on every
                // emission for the rest of its life. The reacquire that would fix it sits behind a
                // send FAILURE - and a send that never happens never fails, so it was unreachable.
                //
                // Measured on the T630: after a storm that restarted the sink ~48 times, `fs` and
                // `block-driver` published NOTHING for the remaining 9.4 s of the run, while the
                // sink's own LOCALLY-written rows were all present. That split - local rows yes, every
                // IPC-delivered row no - is the signature of a publisher whose message never left.
                //
                // `AcquireSendCap` is authorised by the DECLARATION, which the kernel keeps now even
                // when the peer was down at spawn. This is the call that spends it (14.3).
                if self.reacquire_by_name(crate::trace::SINK_NAME) {
                    slot = self.find_send_slot(crate::trace::SINK_NAME).unwrap_or(u32::MAX);
                }
            }
            crate::trace::set_sink_slot(slot);
            // AND SAY IT, IF IT TOOK MORE THAN ONE TRY. This report was written once already and
            // silently lost: the edit that added it failed to apply and only its trace.rs half landed,
            // so `take_late_resolve` sat with no caller and the hardware run reported "0 late
            // resolves" - a zero that measured nothing. A missing instrument reads exactly like a
            // healthy system, which is the worst way for one to fail.
            if slot != u32::MAX {
                if let Some(missed) = crate::trace::take_late_resolve() {
                    if missed > 1 {
                        self.log_fmt(format_args!(
                            "sdk: events sink resolved LATE - {} emission(s) went nowhere first",
                            missed - 1));
                    }
                }
            }
        }
        let slot = crate::trace::sink_slot();
        if slot == u32::MAX {
            return;
        }
        // A TRUNCATED NAME IS A DIFFERENT NAME. Said once per service, not per sample: repeating it
        // would flood the channel the warning has to be read on, and one line names the offender well
        // enough to fix it. Silence here would mean two metrics quietly merging into one row under a
        // shared prefix, with the numbers interleaving and nothing to say why (invariant 12).
        if name.len() > crate::trace::MET_NAME_LEN
            && !TRUNCATED_SAID.swap(true, core::sync::atomic::Ordering::Relaxed)
        {
            self.log_fmt(format_args!(
                "sdk: metric name '{}' is longer than {} bytes and was TRUNCATED - shorten it, or two metrics will share one row",
                name, crate::trace::MET_NAME_LEN));
        }
        let m = crate::trace::encode_metric(name, value);
        if self
            .try_send_by_handle(CapHandle(slot), &crate::ipc::Message::from_bytes(&m))
            .is_err()
        {
            // AND RETRY, for the reason given on the trace path above. It matters MORE here: a
            // publisher samples on an INTERVAL (`fs` every 32 requests), so a dropped sample is not
            // retried for another whole interval - and after a sink restart that meant TWO intervals
            // with no row for that service at all. A service with no row is indistinguishable from a
            // dead one, which is the single question this metric exists to answer.
            // THE FAILING BRANCH IS NAMED, NOT COLLAPSED TO A BOOL. This used to be
            // `if reacquire_by_name(..)`, so a reacquire that FAILED did nothing and said nothing -
            // the sample was dropped and the service simply stopped publishing, with no symptom but a
            // row that never appeared. The three causes are not alike and the difference is the whole
            // diagnosis: NameNotRegistered is transient (the sink is mid-restart, try again next
            // interval), while CapTableFull is TERMINAL until this service restarts - it will never
            // publish again, and that deserves to be said out loud once rather than discovered as an
            // absence. Invariant 12, and 26.7's "a recovery that itself fails is still a failure".
            // ONE LATCH PER SEVERITY, and the wording matches the severity. A single latch was
            // wrong twice over: it said "metrics stop here" for the TRANSIENT case (the sink is
            // simply mid-restart - the next interval reconnects, and a kill-storm produces this by
            // the dozen), and having latched on that it would then stay silent about a genuinely
            // TERMINAL cause arriving later. Observed on the T630: 17 of these during a 100-round
            // storm, every one of them transient and every one of them overstating the damage.
            static ACQ_TRANSIENT_SAID: core::sync::atomic::AtomicBool =
                core::sync::atomic::AtomicBool::new(false);
            static ACQ_TERMINAL_SAID: core::sync::atomic::AtomicBool =
                core::sync::atomic::AtomicBool::new(false);
            let acq = self.reacquire_cap_detail(crate::trace::SINK_NAME);
            if let Err(why) = acq {
                let transient = matches!(why, AcquireFailure::NameNotRegistered);
                let said = if transient { &ACQ_TRANSIENT_SAID } else { &ACQ_TERMINAL_SAID };
                if !said.swap(true, core::sync::atomic::Ordering::Relaxed) {
                    if transient {
                        self.log("sdk: events sink not registered right now - samples dropped until it is back");
                    } else {
                        self.log_fmt(format_args!(
                            "sdk: cannot reacquire the events sink ({}) - metrics from this service stop here",
                            why.reason()));
                    }
                }
            }
            if acq.is_ok() {
                if let Some(fresh) = self.find_send_slot(crate::trace::SINK_NAME) {
                    crate::trace::set_sink_slot(fresh);
                    // A RETRY THAT STILL FAILS IS ITS OWN OUTCOME (26.7): the reacquire worked, so
                    // this is not "the sink is down" - it is a fresh cap that still would not take the
                    // sample. Said once per service; silence here would look identical to success.
                    static RETRY_FAILED_SAID: core::sync::atomic::AtomicBool =
                        core::sync::atomic::AtomicBool::new(false);
                    if self
                        .try_send_by_handle(CapHandle(fresh), &crate::ipc::Message::from_bytes(&m))
                        .is_err()
                        && !RETRY_FAILED_SAID.swap(true, core::sync::atomic::Ordering::Relaxed)
                    {
                        self.log("sdk: events sink reacquired but the retried sample still failed");
                    }
                }
            }
        }
    }

    // ---------------------------------------------------------------------------
    // The request/reply family: each wraps its `_inner` implementation with one
    // trace event on the way in and one on the way out.
    // ---------------------------------------------------------------------------
    //
    // THESE ARE WRAPPERS AND NOT EDITS TO EACH BODY, deliberately. There are a dozen of these now (it
    // was eight), most an independent implementation with several early returns inside a wait loop;
    // instrumenting the bodies means finding every exit in each function and getting all of them right, forever. A
    // wrapper cannot miss one - the value returned IS the outcome.
    //
    // Instrumenting them ALL is also deliberate. The first cut traced only `request_with_reply`, and
    // on hardware that produced an EMPTY ring while the shell was busily talking to `fs` - because the
    // shell uses `_abortable` and `_deadline`, not the plain one. Partial instrumentation is worse
    // than none: `events ipc` then shows SOME traffic and silently omits the rest, with nothing on
    // screen saying which. That is a silent gap in the instrument, which is the exact failure this
    // ring exists to prevent (26.4, invariant 12).
    //
    // `op` is byte 0 of OUR OWN message. A service is entitled to interpret its own protocol; the
    // kernel is not, and that asymmetry is the whole reason this lives here rather than at the
    // routing point (4.4, 26.10).

    // `#[inline]` on every wrapper below is LOAD-BEARING, not decoration. `Message` is 4 KiB by
    // value, so an un-inlined wrapper adds a whole extra payload move to each request - and the shell
    // paths that do many round trips (tab completion listing a directory, `move`, `find` descending)
    // went from passing to timing out on it. Measured, not assumed: `osdev test files` was 222/0
    // before the wrappers, 213/9 after, and 213/9 again with tracing DISABLED - so the cost was the
    // restructuring, never the emission.
    #[inline]
    fn trace_out(&self, peer: &str, op: u8, kind: u8) { self.trace_emit(peer, op, kind); }

    /// The op byte of an outgoing request. NO EVENT IS EMITTED HERE.
    ///
    /// One event per exchange, recording how it ENDED, not two recording that it started and how it
    /// ended. Two doubled the traffic through the sink's single 16-deep endpoint, and the sink drains
    /// on its own core's schedule - so the flood crowded out the READER's own request and `trace
    /// status` reported a service that was alive and busy as "unavailable". Every exchange still
    /// produces exactly one event carrying its fate (REPLY / TIMEOUT / PEER_LOST / QUEUE_FULL /
    /// ABORTED), so nothing is lost but the duplicate.
    ///
    /// A request still IN FLIGHT is therefore not in the ring - and that is the one question the ring
    /// was never the right instrument for: `events blocked` reads it from the kernel, live, which is
    /// what a hang needs (`utilities/46_trace.md` mechanism A).
    #[inline]
    fn trace_in(&self, peer: &str, msg: &crate::ipc::Message) -> u8 {
        // The opcode's byte is the PEER'S protocol's business, not a fixed 0 - see `trace_op_at`.
        let at = crate::trace::op_offset(peer);
        msg.payload_bytes().get(at).copied().unwrap_or(0)
    }

    /// Synchronous request/reply on the caller's own endpoint (kernel `Call`, syscall 41). Blocks until
    /// the reply arrives, or until the kernel wakes it with `ReplyDead` because the replier died (8.6).
    /// It has NO deadline: a replier that stays alive and never answers holds the caller forever.
    #[inline]
    pub fn request_with_reply(&self, peer: &str, msg: &crate::ipc::Message)
        -> Option<crate::ipc::Message>
    {
        let op = self.trace_in(peer, msg);
        let out = self.request_with_reply_inner(peer, msg);
        // No deadline on this variant, so `None` is always a lost peer, never a timeout.
        self.trace_out(peer, op, if out.is_some() { crate::trace::KIND_REPLY }
                                 else { crate::trace::KIND_PEER_LOST });
        out
    }

    /// `request_with_reply_call`, saying WHY it failed instead of collapsing everything to `None`.
    #[inline]
    pub fn request_with_reply_call_err(&self, peer: &str, msg: &crate::ipc::Message, max_secs: i64)
        -> Result<Option<crate::ipc::Message>, crate::ipc::IpcError>
    {
        let op = self.trace_in(peer, msg);
        let out = self.request_with_reply_call_err_inner(peer, msg, max_secs);
        // A full queue is a busy peer, not a lost one - the ring said PEER_LOST for both until
        // 2026-10-10, so `gs::trace` showed `gs::call`'s `Busy` as a lost peer (`backlog/fixed/80` S2).
        self.trace_out(peer, op, match &out {
            Ok(Some(_)) => crate::trace::KIND_REPLY,
            Ok(None)    => crate::trace::KIND_TIMEOUT,
            Err(crate::ipc::IpcError::QueueFull) => crate::trace::KIND_QUEUE_FULL,
            Err(_)      => crate::trace::KIND_PEER_LOST,
        });
        out
    }

    /// Bounded request/reply that decodes straight into the caller buffer.
    ///
    /// This variant collapses "the deadline passed" and "the send never left" into one `None`, so the
    /// ring records both as TIMEOUT. Where that difference matters the caller should be using
    /// `_call_err` or `_deadline_outcome`, which keep it - and so does the ring.
    #[inline]
    pub fn request_with_reply_deadline_into(
        &self, peer: &str, req: &[u8], buf: &mut [u8], max_secs: i64,
    ) -> Option<usize> {
        let op = req.get(crate::trace::op_offset(peer)).copied().unwrap_or(0);
        let out = self.request_with_reply_deadline_into_inner(peer, req, buf, max_secs);
        self.trace_out(peer, op, if out.is_some() { crate::trace::KIND_REPLY }
                                 else { crate::trace::KIND_TIMEOUT });
        out
    }

    /// Bounded request/reply. `None` is "no answer within the deadline" (see `_into` on the collapse).
    #[inline]
    pub fn request_with_reply_deadline(
        &self, peer: &str, msg: &crate::ipc::Message, max_secs: i64,
    ) -> Option<crate::ipc::Message> {
        let op = self.trace_in(peer, msg);
        let out = self.request_with_reply_deadline_inner(peer, msg, max_secs);
        self.trace_out(peer, op, if out.is_some() { crate::trace::KIND_REPLY }
                                 else { crate::trace::KIND_TIMEOUT });
        out
    }

    /// `request_with_reply_deadline` with a millisecond budget.
    #[inline]
    pub fn request_with_reply_ms(
        &self, peer: &str, msg: &crate::ipc::Message, max_ms: u64,
    ) -> Option<crate::ipc::Message> {
        let op = self.trace_in(peer, msg);
        let out = self.request_with_reply_ms_inner(peer, msg, max_ms);
        self.trace_out(peer, op, if out.is_some() { crate::trace::KIND_REPLY }
                                 else { crate::trace::KIND_TIMEOUT });
        out
    }

    /// Bounded request/reply that distinguishes every way it can fail.
    #[inline]
    pub fn request_with_reply_deadline_outcome(
        &self, peer: &str, msg: &crate::ipc::Message, max_secs: i64,
    ) -> DeadlineOutcome {
        let op = self.trace_in(peer, msg);
        let out = self.request_with_reply_deadline_outcome_inner(peer, msg, max_secs);
        self.trace_out(peer, op, match &out {
            DeadlineOutcome::Reply(_)   => crate::trace::KIND_REPLY,
            DeadlineOutcome::SendFailed => crate::trace::KIND_PEER_LOST,
            DeadlineOutcome::QueueFull  => crate::trace::KIND_QUEUE_FULL,
            DeadlineOutcome::Timeout    => crate::trace::KIND_TIMEOUT,
        });
        out
    }

    /// Bounded request/reply that SIFTS what arrives instead of believing the first thing that lands.
    ///
    /// Every other variant on this page returns whatever message turns up next, which is correct for a
    /// caller whose endpoint carries nothing but its own replies and WRONG for a caller that also
    /// SERVES on it. `net-stack` is the second kind: it receives client requests and `nic-driver`
    /// replies in one mailbox, so a client that spoke while it was waiting for the driver was consumed
    /// as the driver's answer, misparsed, and never served - the failure `docs/net-tags-design.md`
    /// opens with, and the reason that service still cannot do any background work.
    ///
    /// `mine` is asked about each message as it arrives, BEFORE anything else is done with it, and
    /// answers one question: is this the reply I sent for? Say `true` and the wait ends with it. Say
    /// `false` and the wait continues - the message is the caller's to deal with, and the caller is
    /// the only thing that can, because only it knows whether that means stashing a client request or
    /// discarding a reply that arrived too late. This is deliberately not an enum of dispositions: the
    /// SDK would then be deciding what to keep, which is policy, and the thing being kept usually has
    /// a reply capability that must be reclaimed or answered rather than merely stored (26.10, 8.5).
    ///
    /// The kernel's per-task pending cap and receive badge are both readable from inside `mine`
    /// (`take_pending_cap`, `last_recv_badge`), which is the point of calling it at the moment of
    /// arrival rather than afterwards: they describe THE MESSAGE JUST RECEIVED and are overwritten by
    /// the next one. A discriminator built on the pending cap is what `net-stack` uses, and it is free
    /// - a client request carries a reply cap, a driver reply does not.
    ///
    /// Everything else matches [`Self::request_with_reply_deadline_outcome`] exactly, including the
    /// reply-cap reclaim on a failed send (a delivered request's cap is the peer's - backlog/67) and the warning attached to `Timeout`: the request was
    /// SENT, so a late reply is still coming and re-sending desyncs the protocol.
    #[inline]
    pub fn request_with_reply_deadline_sifted(
        &self, peer: &str, msg: &crate::ipc::Message, max_secs: i64,
        mine: impl FnMut(&crate::ipc::Message) -> bool,
    ) -> DeadlineOutcome {
        let op = self.trace_in(peer, msg);
        let out = self.request_with_reply_deadline_sifted_inner(peer, msg, max_secs, mine);
        self.trace_out(peer, op, match &out {
            DeadlineOutcome::Reply(_)   => crate::trace::KIND_REPLY,
            DeadlineOutcome::SendFailed => crate::trace::KIND_PEER_LOST,
            DeadlineOutcome::QueueFull  => crate::trace::KIND_QUEUE_FULL,
            DeadlineOutcome::Timeout    => crate::trace::KIND_TIMEOUT,
        });
        out
    }

    /// [`Self::request_with_reply_deadline_sifted`], bounded in MILLISECONDS.
    ///
    /// The same reason the unsifted `_ms` form exists: a sub-second window cannot be built out of a
    /// whole-second bound, and the drain inside `net-stack`'s ~900 ms ping window is exactly such a
    /// window. Bounded by the CYCLE counter for the same reason too - it is the only clock here with
    /// sub-second resolution, and it is the one the caller's own budget is measured in.
    #[inline]
    pub fn request_with_reply_ms_sifted(
        &self, peer: &str, msg: &crate::ipc::Message, max_ms: u64,
        mut mine: impl FnMut(&crate::ipc::Message) -> bool,
    ) -> Option<crate::ipc::Message> {
        let op = self.trace_in(peer, msg);
        let target = CapHandle(self.find_send_slot(peer)?);
        let self_grant = self.self_grant_handle()?;
        let reply_cap = self.derive_cap(self_grant)?;
        if self.offer_request(target, reply_cap, msg).is_err() {
            self.remove_cap(reply_cap);
            self.trace_out(peer, op, crate::trace::KIND_PEER_LOST);
            return None;
        }
        let t0 = self.read_tsc();
        let budget = self.duration_cycles(max_ms);
        loop {
            if let Some(r) = self.await_slice(Self::AWAIT_SLICE_MS.min(max_ms.max(1))) {
                if mine(&r) {
                    self.trace_out(peer, op, crate::trace::KIND_REPLY);
                    return Some(r);
                }
            }
            // wrapping_sub, so a counter that wraps mid-wait reads as a small elapsed rather than as
            // an enormous one that expires the deadline instantly.
            if self.read_tsc().wrapping_sub(t0) >= budget {
                // NOT RECLAIMED: the send succeeded, so the kernel moved the reply cap to the peer and emptied
                // this slot - and the first cap this task RECEIVED since went into it. Removing "the reply
                // cap" here destroyed that one: a client's reply cap stashed mid-wait, so its answer went
                // into a dead cap (backlog/67). The peer removes the cap after answering.
                self.trace_out(peer, op, crate::trace::KIND_TIMEOUT);
                return None;
            }
        }
    }

    /// Bounded request/reply the user can abandon with `q`.
    #[inline]
    pub fn request_with_reply_abortable(
        &self, peer: &str, msg: &crate::ipc::Message, max_secs: i64,
    ) -> ReqOutcome {
        let op = self.trace_in(peer, msg);
        let out = self.request_with_reply_abortable_inner(peer, msg, max_secs);
        self.trace_out(peer, op, match &out {
            ReqOutcome::Reply(_) => crate::trace::KIND_REPLY,
            ReqOutcome::Aborted  => crate::trace::KIND_ABORTED,
            ReqOutcome::Timeout  => crate::trace::KIND_TIMEOUT,
        });
        out
    }

    /// `request_with_reply_abortable`, plus a callback when the wait starts to linger.
    #[inline]
    pub fn request_with_reply_qhint(
        &self, peer: &str, msg: &crate::ipc::Message, hint_after_secs: i64, max_secs: i64,
        on_linger: impl FnOnce(),
    ) -> ReqOutcome {
        self.request_with_reply_keyhint(peer, msg, hint_after_secs, max_secs, Self::QUIT_KEYS, on_linger)
    }

    /// [`Self::request_with_reply_qhint`] that HANDS BACK every key it reads that is not a quit key.
    ///
    /// A wait that watches the console for `q` has to read what is typed, and a key read is a key
    /// taken: the plain form discards the rest, so whatever the operator typed ahead while it waited
    /// was lost (`backlog/fixed/80` H16). This passes each one to `keep`, in order, for the caller to put
    /// where its own input reader will find it first.
    #[inline]
    pub fn request_with_reply_qhint_keeping(
        &self, peer: &str, msg: &crate::ipc::Message, hint_after_secs: i64, max_secs: i64,
        on_linger: impl FnOnce(), keep: &dyn Fn(u8),
    ) -> ReqOutcome {
        let op = self.trace_in(peer, msg);
        let out = self.request_with_reply_qhint_inner(
            peer, msg, hint_after_secs, max_secs, Self::QUIT_KEYS, on_linger, |_| true, Some(keep));
        self.trace_out(peer, op, match &out {
            ReqOutcome::Reply(_) => crate::trace::KIND_REPLY,
            ReqOutcome::Aborted  => crate::trace::KIND_ABORTED,
            ReqOutcome::Timeout  => crate::trace::KIND_TIMEOUT,
        });
        out
    }

    /// The keys [`Self::request_with_reply_qhint`] leaves on: `q`, `Q` and Escape.
    pub const QUIT_KEYS: &'static [u8] = &[b'q', b'Q', 0x1b];

    /// [`Self::request_with_reply_qhint`] with the LEAVE KEYS chosen by the caller. The convention
    /// (`utilities/0_conventions.md`) gives `q` to a task that STOPS when the operator leaves and `b` to
    /// one that keeps running; a request whose peer finishes the work whether or not this caller waits -
    /// `wifi radio off hard`, where the driver cuts the chip's power either way - is the second kind, and
    /// offering `q` for it would promise a stop that cannot happen. `Aborted` means one of `leave_keys`
    /// was pressed; what that means for the task is the caller's to say.
    pub fn request_with_reply_keyhint(
        &self, peer: &str, msg: &crate::ipc::Message, hint_after_secs: i64, max_secs: i64,
        leave_keys: &[u8], on_linger: impl FnOnce(),
    ) -> ReqOutcome {
        let op = self.trace_in(peer, msg);
        let out = self.request_with_reply_qhint_inner(peer, msg, hint_after_secs, max_secs, leave_keys, on_linger, |_| true, None);
        self.trace_out(peer, op, match &out {
            ReqOutcome::Reply(_) => crate::trace::KIND_REPLY,
            ReqOutcome::Aborted  => crate::trace::KIND_ABORTED,
            ReqOutcome::Timeout  => crate::trace::KIND_TIMEOUT,
        });
        out
    }

    /// [`Self::request_with_reply_keyhint`] that SIFTS what arrives, as
    /// [`Self::request_with_reply_deadline_sifted`] does for the unabortable wait. `mine` is asked about
    /// each message as it lands; `true` ends the wait with it, `false` hands it to the caller - who must
    /// release any cap it carries - and the wait goes on, its deadline unchanged by how many arrive.
    /// Messages ALREADY queued when it is called are drained first without being asked (the stale-reply
    /// drain every abortable wait opens with), so a caller that counts late answers drains before calling.
    ///
    /// For a peer whose replies the caller can recognise (a correlation tag) on an endpoint that also
    /// takes other traffic: the unsifted form returns the first message whatever it is, so a late
    /// answer to a request this caller already gave up on is read as the answer to this one.
    pub fn request_with_reply_keyhint_sifted(
        &self, peer: &str, msg: &crate::ipc::Message, hint_after_secs: i64, max_secs: i64,
        leave_keys: &[u8], on_linger: impl FnOnce(),
        mine: impl FnMut(&crate::ipc::Message) -> bool,
    ) -> ReqOutcome {
        let op = self.trace_in(peer, msg);
        let out = self.request_with_reply_qhint_inner(peer, msg, hint_after_secs, max_secs, leave_keys, on_linger, mine, None);
        self.trace_out(peer, op, match &out {
            ReqOutcome::Reply(_) => crate::trace::KIND_REPLY,
            ReqOutcome::Aborted  => crate::trace::KIND_ABORTED,
            ReqOutcome::Timeout  => crate::trace::KIND_TIMEOUT,
        });
        out
    }

    /// [`Self::request_with_reply_keyhint_sifted`] that HANDS BACK every key it reads that is not a leave
    /// key, in order, as [`Self::request_with_reply_qhint_keeping`] does for the unsifted request. For a
    /// caller whose waits must say `[q] quit` and must not lose what the operator types meanwhile.
    pub fn request_with_reply_keyhint_sifted_keeping(
        &self, peer: &str, msg: &crate::ipc::Message, hint_after_secs: i64, max_secs: i64,
        leave_keys: &[u8], on_linger: impl FnOnce(),
        mine: impl FnMut(&crate::ipc::Message) -> bool, keep: &dyn Fn(u8),
    ) -> ReqOutcome {
        let op = self.trace_in(peer, msg);
        let out = self.request_with_reply_qhint_inner(peer, msg, hint_after_secs, max_secs, leave_keys, on_linger, mine, Some(keep));
        self.trace_out(peer, op, match &out {
            ReqOutcome::Reply(_) => crate::trace::KIND_REPLY,
            ReqOutcome::Aborted  => crate::trace::KIND_ABORTED,
            ReqOutcome::Timeout  => crate::trace::KIND_TIMEOUT,
        });
        out
    }



    fn request_with_reply_inner(
        &self,
        peer: &str,
        msg:  &crate::ipc::Message,
    ) -> Option<crate::ipc::Message> {
        let target = CapHandle(self.find_send_slot(peer)?);
        let self_grant = self.self_grant_handle()?;
        let reply_cap = self.derive_cap(self_grant)?;
        let recv = self.recv_handle()?;
        match crate::ipc::call(target, reply_cap, recv, msg) {
            Ok(reply) => Some(reply),
            Err(_) => {
                // Send failed (dead endpoint) or the peer died before replying (ReplyDead): the
                // embedded reply cap may not have been transferred, so reclaim it. Safe when the kernel
                // DID move it out: a Call receives nothing but its own reply, so no other cap can have
                // been inserted into the emptied slot, and removing an empty slot does nothing. (The
                // send-then-recv paths cannot say that, and do not reclaim - backlog/67.) Without this, a
                // storm of failed calls would leak reply caps until the table fills and every request
                // returns None.
                self.remove_cap(reply_cap);
                None
            }
        }
    }

    #[inline(never)]
    /// A mark of how deep the stack currently is, in bytes below the first mark taken.
    ///
    /// THE INSTRUMENT THAT WAS MISSING. Five attempts were made to fit correlation tags into `fs`
    /// without ever measuring the frame they had to fit in; each removed one 4 KiB item, the overflow
    /// returned at a different pc, and QEMU - which has no disk, so `fs` never mounts and never makes
    /// the deep calls - reported every build clean. A change to stack usage that cannot be measured
    /// can only be guessed at, and five guesses is the evidence.
    ///
    /// Safe Rust and no kernel involvement: the address of a local IS the stack pointer, near enough.
    /// Relative to a mark taken at service start, so it needs no knowledge of where the stack lives.
    pub fn stack_mark(&self) -> usize {
        let probe = 0u8;
        &probe as *const u8 as usize
    }

    /// Bytes of stack used at this point, without needing a reference mark.
    ///
    /// A mark taken at service start only works if the report is taken at the DEEPEST point, and the
    /// first attempt took it after `mount` had returned - so it measured a popped frame and printed
    /// zero. Absolute is harder to misuse: it can be called from anywhere and still means the same
    /// thing.
    ///
    /// The stack is one region, 256 KiB, and top-aligned, so rounding the current pointer UP to the
    /// next 256 KiB boundary lands on its top and the difference is what has been used. That derives
    /// the top from the alignment the kernel already guarantees rather than restating the address here,
    /// which would be a second copy of a layout decision (Commandment III).
    pub fn stack_used(&self) -> usize {
        const REGION: usize = 256 * 1024;
        let sp = self.stack_mark();
        let top = (sp + (REGION - 1)) & !(REGION - 1);
        top - sp
    }

    /// A bounded request/reply through the KERNEL's `CallDeadline`, returning the reply as a message.
    ///
    /// The same primitive `request_with_reply_deadline_into` already uses, in the shape the other
    /// request helpers have, so a caller can move to it without restructuring its buffers.
    ///
    /// Use this rather than `request_with_reply_deadline` / `_deadline_outcome` when the caller SERVES
    /// clients on the same endpoint it awaits replies on. Those helpers wait with a plain timed `recv`,
    /// which takes whatever is next - so under client load the peer's reply can be crowded out of a 16-deep queue and the
    /// caller reports a timeout that blames the peer for something it did correctly. `net-stack` hit
    /// exactly that: the wire showed 7 DHCP REQUESTs sent and 7 ACKs returned while the log said
    /// "6 of 6 REQUESTs never left the host - the driver refused them".
    ///
    /// `CallDeadline` does not have that failure: `call_dequeue` matches the reply BY ITS SENDER and
    /// leaves every other message queued, which is the entire reason §8.2 added it - hand-rolling the
    /// bounded wait out of send + recv is what lost messages. No new kernel surface: syscall 50 and
    /// its semantics are existing, ratified law. `max_secs <= 0` is a one-second
    /// bound, never none: the kernel reads a 0 as NO deadline, and a bounded call turning into an
    /// unbounded one by a unit slip is the hang it exists to prevent (`backlog/fixed/80` S5).
    pub fn request_with_reply_call(&self, peer: &str, msg: &crate::ipc::Message, max_secs: i64)
        -> Option<crate::ipc::Message>
    {
        self.request_with_reply_call_err(peer, msg, max_secs).ok().flatten()
    }

    /// `request_with_reply_call`, but saying WHY it failed instead of collapsing everything to None.
    ///
    /// Exists because "no answer" is not a diagnosis. A caller that turns every failure into one word
    /// then reports it as the PEER's fault - `net-stack` blamed the driver for refusing frames the wire
    /// showed it had sent - and there is no way to tell a send that never left from a reply that never
    /// came. `Ok(None)` is the deadline passing; `Err(e)` is the send failing, and `e` says how.
    fn request_with_reply_call_err_inner(&self, peer: &str, msg: &crate::ipc::Message, max_secs: i64)
        -> Result<Option<crate::ipc::Message>, crate::ipc::IpcError>
    {
        let target = match self.find_send_slot(peer) {
            Some(s) => CapHandle(s),
            None    => return Err(crate::ipc::IpcError::CapError(crate::capability::CapError::CapNotHeld)),
        };
        // Abandoned replies out first, or this request takes one of them (`drain_before_request`).
        self.drain_before_request(peer);
        let (recv, grant) = match self.reply_mailbox() {
            Some((r, g)) => (r, g),
            None => match (self.recv_handle(), self.self_grant_handle()) {
                (Some(r), Some(g)) => (r, g),
                _ => return Err(crate::ipc::IpcError::CapError(crate::capability::CapError::CapNotHeld)),
            },
        };
        let reply_cap = match self.derive_cap(grant) {
            Some(c) => c,
            None    => return Err(crate::ipc::IpcError::CapError(crate::capability::CapError::CapNotHeld)),
        };
        let secs = deadline_secs(max_secs);
        let mut buf = [0u8; crate::ipc::MAX_PAYLOAD];
        let n = msg.payload_bytes().len().min(buf.len());
        buf[..n].copy_from_slice(&msg.payload_bytes()[..n]);
        match crate::ipc::call_deadline_into(target, reply_cap, recv, &msg.payload_bytes()[..n],
                                             &mut buf, secs) {
            Ok(Some(len)) => Ok(Some(crate::ipc::Message::from_bytes(&buf[..len]))),
            // Delivered, deadline passed: the reply cap is the peer's now (backlog/67).
            Ok(None)      => Ok(None),
            Err(e)        => { self.remove_cap(reply_cap); Err(e) }
        }
    }

    /// Bounded request/reply into a caller buffer, reporting WHICH failure happened.
    ///
    /// Same work as [`Self::request_with_reply_deadline_into`], same stack shape - the caller supplies
    /// the buffer so no reply-sized frame appears here - but it does not throw the reason away. See
    /// [`DeadlineOutcomeInto`] for why the distinction is not a nicety: the two common failures want
    /// opposite responses, and a caller given `None` has to guess wrong in one direction or the other.
    pub fn request_with_reply_deadline_outcome_into(
        &self, peer: &str, req: &[u8], buf: &mut [u8], max_secs: i64,
    ) -> DeadlineOutcomeInto {
        let op = req.get(crate::trace::op_offset(peer)).copied().unwrap_or(0);
        let out = self.request_with_reply_deadline_outcome_into_inner(peer, req, buf, max_secs);
        self.trace_out(peer, op, match &out {
            DeadlineOutcomeInto::Reply(_)   => crate::trace::KIND_REPLY,
            DeadlineOutcomeInto::SendFailed => crate::trace::KIND_PEER_LOST,
            DeadlineOutcomeInto::PeerDied   => crate::trace::KIND_PEER_LOST,
            DeadlineOutcomeInto::QueueFull  => crate::trace::KIND_QUEUE_FULL,
            DeadlineOutcomeInto::Timeout    => crate::trace::KIND_TIMEOUT,
        });
        out
    }

    fn request_with_reply_deadline_outcome_into_inner(
        &self, peer: &str, req: &[u8], buf: &mut [u8], max_secs: i64,
    ) -> DeadlineOutcomeInto {
        // Each of these is a DISTINCT reason the request never left, and every one of them used to be
        // an anonymous `None`: no such peer wired or resolvable, no endpoint of our own to be replied
        // to, and - the one that made this permanent rather than transient - NO FREE CAP SLOT to derive
        // a reply cap into. A service whose table has filled reports "peer did not answer", which sends
        // the reader looking at the peer.
        let target = match self.find_send_slot(peer) {
            Some(s) => CapHandle(s),
            None => return DeadlineOutcomeInto::SendFailed,
        };
        // Abandoned replies out first, or this request takes one of them (`drain_before_request`).
        self.drain_before_request(peer);
        let (recv, grant) = match self.reply_mailbox() {
            Some((r, g)) => (r, g),
            None => match (self.recv_handle(), self.self_grant_handle()) {
                (Some(r), Some(g)) => (r, g),
                _ => return DeadlineOutcomeInto::SendFailed,
            },
        };
        let reply_cap = match self.derive_cap(grant) {
            Some(c) => c,
            None => return DeadlineOutcomeInto::SendFailed,
        };
        let secs = deadline_secs(max_secs);
        let out = crate::ipc::call_deadline_into(target, reply_cap, recv, req, buf, secs);
        // The kernel consumes the reply cap on a delivered call - and `Ok(None)` is a delivered call
        // whose deadline passed, so it is the peer's then too (backlog/67). On a failed send it is ours
        // to reclaim, or the slot leaks one per failed request (§8.5, the three checks).
        match out {
            Ok(Some(n)) => DeadlineOutcomeInto::Reply(n),
            Ok(None) => DeadlineOutcomeInto::Timeout,
            Err(crate::ipc::IpcError::QueueFull) => {
                self.remove_cap(reply_cap);
                DeadlineOutcomeInto::QueueFull
            }
            // Delivered: the reply cap went with the request, as on `Ok(None)` - nothing to reclaim.
            Err(crate::ipc::IpcError::ReplyDead) => DeadlineOutcomeInto::PeerDied,
            Err(_) => { self.remove_cap(reply_cap); DeadlineOutcomeInto::SendFailed }
        }
    }

    /// Send `req` to `peer` and receive the reply into `buf`, bounded by `max_secs`; returns the
    /// reply's length. `max_secs <= 0` is a one-second bound, never an unbounded wait (see
    /// `request_with_reply_call`).
    ///
    /// The by-value path costs 4 KiB for the request `Message`, 4 KiB for the reply `Message`, and
    /// 4 KiB again at every wrapper that forwards it. This costs none of that: the caller owns both
    /// buffers, and only a length is returned. Drains abandoned replies from the reply mailbox first
    /// (`drain_stale_replies`), and reclaims the reply cap only when the send failed (§8.5, backlog/67).
    ///
    /// **This used to be send + `recv_timeout_into`, and that was wrong in a way that cost days.**
    /// `recv_timeout_into` takes whatever is next on the endpoint, and a service that SERVES clients on
    /// the same endpoint it awaits replies on would receive an unrelated client request, fail to match
    /// it, and drop it - losing that request outright while its own reply arrived later as an orphan
    /// that desynced the next exchange. `fs` does exactly that, which is why its clients hung and its
    /// block protocol "lost step" on both boards.
    ///
    /// It now uses `CallDeadline`, which dequeues the REPLY specifically (the kernel's `call_dequeue`
    /// takes only a message stamped as sent by the target's owner) and leaves everything else queued.
    /// The correct primitive, bounded.
    fn request_with_reply_deadline_into_inner(
        &self, peer: &str, req: &[u8], buf: &mut [u8], max_secs: i64,
    ) -> Option<usize> {
        let target = CapHandle(self.find_send_slot(peer)?);
        // DRAIN ABANDONED REPLIES BEFORE ISSUING, FOR EVERY CALLER.
        //
        // `drain_stale_replies` documents this failure precisely and had exactly one caller - `fs` -
        // so every OTHER service was one timeout away from being permanently broken. The shell was:
        // a single scroll request that timed out left its reply queued, and from that moment every
        // request waited out its full deadline and failed, for the rest of the boot. Observed on a
        // Dell Wyse as `scrollback: the console did not answer` once a second, indefinitely, with the
        // console sitting idle at `BlockRecv, queue 0` the whole time (`backlog/fixed/37`).
        //
        // **It is safe here for the reason that function already gives**, and that reason is exactly
        // why this belongs in the SDK rather than at each call site: `request_with_reply*` enforces
        // ONE OUTSTANDING REQUEST AT A TIME, so at the instant a new request is issued nothing
        // legitimate can be waiting in the reply mailbox. Anything present is a reply this service
        // stopped waiting for. Leaving it to each caller to remember means a caller that forgets is
        // not slightly worse - it is permanently broken after its first timeout, silently.
        //
        // Cheap: a non-blocking `try_recv` that finds nothing in the healthy case, bounded by the
        // endpoint depth.
        self.drain_before_request(peer);
        // Reply mailbox when the task has one, shared endpoint when it does not. The reply cap and the
        // endpoint waited on must name the SAME endpoint, or the kernel's reply-matched dequeue waits
        // for something that will never be delivered there.
        let (recv, grant) = match self.reply_mailbox() {
            Some((r, g)) => (r, g),
            None => (self.recv_handle()?, self.self_grant_handle()?),
        };
        let reply_cap = self.derive_cap(grant)?;
        let secs = deadline_secs(max_secs);
        let out = crate::ipc::call_deadline_into(target, reply_cap, recv, req, buf, secs);
        // The kernel consumes the reply cap on a delivered call - and `Ok(None)` is a delivered call
        // whose deadline passed, so it is the peer's then too (backlog/67). On a failed send it is ours
        // to reclaim, or the slot leaks one per failed request (§8.5, the three checks).
        match out {
            Ok(Some(n)) => Some(n),
            Ok(None) => None,
            Err(_)   => { self.remove_cap(reply_cap); None }
        }
    }

    /// How long one blocking wait slice lasts.
    ///
    /// 20 ms: two scheduler quanta, so a waiter never holds its core, and short enough that a caller
    /// checking for a keypress between slices still feels instant to a human. Nothing depends on the
    /// value for correctness - a message arriving mid-slice wakes the task immediately; the slice only
    /// bounds how often a caller gets to look at anything OTHER than its endpoint.
    const AWAIT_SLICE_MS: u64 = 20;
    /// The slice a noticing wait blocks for before its notice is due, when it is not reading the
    /// console and has only the reply and the clock to watch.
    const PRE_NOTICE_SLICE_MS: u64 = 250;

    /// Wait for a message on our own endpoint, BLOCKING, for at most one slice.
    ///
    /// The whole point of this helper, and the reason it exists rather than a `try_recv` loop: a task
    /// that spins while waiting is not idle, it is COMPETING with whatever it is waiting for. On this
    /// system that is not an abstract cost - `net-stack` waiting for a DHCP reply spun through the SDK
    /// and through its own `drain_scan`, two nested loops, and saturated `nic-driver` and the USB
    /// driver that had to fetch the very reply it was waiting for. The waiting starved the answering.
    ///
    /// Pinning the services to different cores hid it and fixed nothing: on a single-core machine the
    /// spinner simply consumes the whole system. A blocked task consumes nothing and is woken the
    /// instant a message lands (`RecvTimeout`, §12), so this is correct at any core count - which is
    /// the only kind of correct worth having.
    ///
    /// Sliced rather than one long block so callers that must also watch something else - a keypress,
    /// an abort flag - get the chance, without any of them going back to spinning.
    ///
    /// (2026-10-09) The `#[inline(always)]` the next paragraphs describe is NOT on this function any
    /// more: 8f32a5e9 removed it, because inlining a body that holds a 4 KiB local into a caller that
    /// already holds one puts both in the same frame. The cost they describe is real either way.
    ///
    /// `#[inline(always)]`, and that attribute is load-bearing rather than an optimisation hint. A
    /// `Message` is 4096 bytes BY VALUE, so a wrapper that returns one is not free: it is a whole
    /// extra 4 KiB stack frame on every request in every service. Introducing this helper cost `fs`
    /// exactly that, and `fs` was already the service closest to its 256 KiB limit - it began taking a
    /// data abort at `mount`, with SP below the bottom of its stack region.
    ///
    /// It only showed on hardware. QEMU's raspi2b has no disk, so `fs` never mounts and never makes
    /// the deep block calls, and thirty-second QEMU boots kept coming back clean while the board
    /// restart-looped. A one-line wrapper around a large return value has to be inlined or it must not
    /// exist.
    fn await_slice(&self, ms: u64) -> Option<crate::ipc::Message> {
        self.recv_timeout(self.duration_cycles(ms))
    }

    /// Offer a request to `target`, riding out a FULL peer queue instead of calling it a failure.
    ///
    /// A full queue and an unreachable peer are different things, and reporting them the same way is
    /// what desynced the shell against `fs` on x86. The caller's answer to "send failed" is to
    /// reacquire the peer by name and RE-SEND with a fresh tag - correct when a restart made the cap
    /// stale, and actively harmful when the queue was merely full: the original request is still in
    /// flight, so its reply arrives as an orphan and every request afterwards collects the PREVIOUS
    /// one's reply. The log showed exactly that, permanently one behind:
    ///
    ///     shell: discarded an fs reply for tag 68 while awaiting 69
    ///     shell: discarded an fs reply for tag 72 while awaiting 73
    ///
    /// Congestion clears as soon as the peer is scheduled, so the fix is to wait a moment and offer
    /// the SAME request again - never a second one. Bounded, and a queue still full after that is
    /// reported honestly to the caller.
    fn offer_request(
        &self, target: CapHandle, reply_cap: CapHandle, msg: &crate::ipc::Message,
    ) -> Result<(), crate::ipc::IpcError> {
        const BUSY_MS: u64 = 2;
        const BUSY_TRIES: u32 = 8;
        let mut last = crate::ipc::IpcError::QueueFull;
        for _ in 0..BUSY_TRIES {
            match self.send_with_cap_by_handle(target, reply_cap, msg) {
                Ok(()) => return Ok(()),
                Err(crate::ipc::IpcError::QueueFull) => {
                    last = crate::ipc::IpcError::QueueFull;
                    self.sleep(self.duration_cycles(BUSY_MS));
                }
                Err(e) => return Err(e),
            }
        }
        Err(last)
    }

    /// Like `request_with_reply`, but the wait for the reply is **bounded** by `max_secs` of
    /// **wall-clock** time (`epoch_secs_monotonic`, InspectKernel query 17). Returns `None` on timeout,
    /// and on a send that never left - so a peer that receives the request and then stays silent cannot
    /// hold the caller forever (`request_with_reply`, which has no deadline, would). Use it when the
    /// peer may be unstable - e.g. writing a report to `fs` right after a chaos storm hammered `fs` +
    /// its `block-driver`.
    ///
    /// Uses the RTC (not a TSC-cycle deadline) deliberately: a cycle bound is not portable - under
    /// QEMU's TCG the guest TSC races ahead and expires the deadline before the reply arrives, while
    /// the RTC is real wall-clock on both TCG and hardware. Waits in blocking slices (`await_slice`),
    /// so it does not spin, and its resolution is one second.
    /// **Costs one message-sized stack frame, not two.**
    ///
    /// This used to be a thin wrapper over `request_with_reply_deadline_outcome`, and that made it
    /// unusable on a tight stack: `DeadlineOutcome` CARRIES a `Message`, so the enum is a second
    /// 4 KiB temporary on top of the `Option<Message>` returned. `fs` already sits near its 256 KiB
    /// user stack, and switching its block RPC to this wrapper produced an instant data abort at
    /// startup and an 826-deep restart loop on real hardware - the machine unusable, from a change
    /// whose whole purpose was to make a hang impossible.
    ///
    /// So the wait is written out here instead of borrowed. It is the same loop, minus the enum: a
    /// caller that does not need to distinguish "the send never left" from "the peer was silent"
    /// should not pay a message-sized frame for the distinction. Callers that DO need it still have
    /// `..._outcome`, and should keep an eye on their own stack.
    ///
    /// **Shares the flaw `request_with_reply_deadline_into` was fixed for, and is NOT fixed here.**
    /// It sends with a reply cap and then receives generically, so a caller that also SERVES on its
    /// endpoint can consume an unrelated message and drop it. `_into` moved to `CallDeadline`, which
    /// dequeues the reply specifically; this one has not, because nothing has yet been shown to be
    /// harmed by it and a blind sweep of four helpers is how a fix becomes a regression.
    ///
    /// Note that `_abortable` and `_qhint` CANNOT simply follow: they interleave on purpose, to
    /// notice a `q` keypress while waiting. Making them dequeue only the reply would delete that. If
    /// they need this too, the answer is a bounded stash, not a substitution.
    fn request_with_reply_deadline_inner(
        &self,
        peer: &str,
        msg:  &crate::ipc::Message,
        max_secs: i64,
    ) -> Option<crate::ipc::Message> {
        let target = CapHandle(self.find_send_slot(peer)?);
        let self_grant = self.self_grant_handle()?;
        let reply_cap = self.derive_cap(self_grant)?;
        let recv = self.recv_handle()?;
        if self.offer_request(target, reply_cap, msg).is_err() {
            // The send never left, so the embedded cap was not transferred - reclaim it or a storm of
            // failures leaks the table (§8.5).
            self.remove_cap(reply_cap);
            return None;
        }
        let t0 = self.epoch_secs_monotonic();
        loop {
            // BLOCK, do not spin. See `await_slice`.
            if let Some(r) = self.await_slice(Self::AWAIT_SLICE_MS) {
                return Some(r);
            }
            let now = self.epoch_secs_monotonic();
            // Guard the clock going BACKWARDS as well as forwards: on hardware whose counter is
            // unreliable, `now - t0` can read huge and expire the deadline on the first pass.
            if now >= t0 && now - t0 >= max_secs {
                // Abandoned: the reply may still arrive later and sit in our queue, which is the
                // hazard `..._outcome` documents. The caller reports the failure.
                // NOT RECLAIMED: the send succeeded, so the kernel moved the reply cap to the peer and emptied
                // this slot - and the first cap this task RECEIVED since went into it. Removing "the reply
                // cap" here destroyed that one: a client's reply cap stashed mid-wait, so its answer went
                // into a dead cap (backlog/67). The peer removes the cap after answering.
                let _ = recv;
                return None;
            }
        }
    }

    /// `request_with_reply_deadline`, bounded in MILLISECONDS instead of whole seconds.
    ///
    /// Exists because a sub-second window cannot be built out of a whole-second bound, and one was
    /// being attempted. `net-stack`'s ping window is ~900 ms and cycle-based, but the drain inside it
    /// was bounded by `LINK_SECS = 1` on `epoch_secs_monotonic`, whose resolution IS one second - so a
    /// single drain could legitimately outlast the entire window. Measured on a Pi 4: the window
    /// reported `closed after 1017663 us [budget 900000 us]` with `0 drains`, meaning it completed
    /// exactly ONE call and that call overran the whole budget. Ping sends once a second, so a window
    /// that runs 1.018 s misses roughly every third reply - which read as "the network is flaky" and
    /// was arithmetic.
    ///
    /// Bounded by the CYCLE counter, deliberately, because that is the only clock here with sub-second
    /// resolution and it is the same clock the caller's own window uses - a bound and the budget it
    /// must fit inside should not be measured by different clocks. `duration_cycles` already floors an
    /// uncalibrated counter to one quantum, so a board with no calibration degrades to "return almost
    /// at once" rather than to "wait forever".
    ///
    /// No new kernel surface: this is the same userspace poll the seconds variants use, with a
    /// different clock read in the deadline test.
    fn request_with_reply_ms_inner(
        &self,
        peer: &str,
        msg:  &crate::ipc::Message,
        max_ms: u64,
    ) -> Option<crate::ipc::Message> {
        let target = CapHandle(self.find_send_slot(peer)?);
        let self_grant = self.self_grant_handle()?;
        let reply_cap = self.derive_cap(self_grant)?;
        let recv = self.recv_handle()?;
        if self.offer_request(target, reply_cap, msg).is_err() {
            self.remove_cap(reply_cap);
            return None;
        }
        let t0 = self.read_tsc();
        let budget = self.duration_cycles(max_ms);
        loop {
            if let Some(r) = self.await_slice(Self::AWAIT_SLICE_MS.min(max_ms.max(1))) {
                return Some(r);
            }
            // wrapping_sub, so a counter that wraps mid-wait reads as a small elapsed rather than as
            // an enormous one that expires the deadline instantly.
            if self.read_tsc().wrapping_sub(t0) >= budget {
                // NOT RECLAIMED: the send succeeded, so the kernel moved the reply cap to the peer and emptied
                // this slot - and the first cap this task RECEIVED since went into it. Removing "the reply
                // cap" here destroyed that one: a client's reply cap stashed mid-wait, so its answer went
                // into a dead cap (backlog/67). The peer removes the cap after answering.
                let _ = recv;
                return None;
            }
        }
    }

    /// Like [`Self::request_with_reply_deadline`] but returns the DISTINCTION between a send that never
    /// left (stale/unresolvable peer cap) and a peer that received the request but stayed silent past
    /// the deadline (see [`DeadlineOutcome`]). Same bounded, RTC-deadline wait; use this when a caller
    /// must reacquire+retry a *restarted* peer (`SendFailed`) but must NOT double the wait on a merely
    /// *silent* one (`Timeout`).
    fn request_with_reply_deadline_outcome_inner(
        &self,
        peer: &str,
        msg:  &crate::ipc::Message,
        max_secs: i64,
    ) -> DeadlineOutcome {
        let target = match self.find_send_slot(peer) { Some(s) => CapHandle(s), None => return DeadlineOutcome::SendFailed };
        let self_grant = match self.self_grant_handle() { Some(g) => g, None => return DeadlineOutcome::SendFailed };
        let reply_cap = match self.derive_cap(self_grant) { Some(c) => c, None => return DeadlineOutcome::SendFailed };
        if let Err(e) = self.send_with_cap_by_handle(target, reply_cap, msg) {
            self.remove_cap(reply_cap);   // send failed: reclaim the untransferred reply cap (no leak)
            return match e {
                crate::ipc::IpcError::QueueFull => DeadlineOutcome::QueueFull,
                _ => DeadlineOutcome::SendFailed,
            };
        }
        // Deglitched monotonic clock, not the raw RTC: a single CMOS misread (the "4383d" glitch on the
        // T630) would otherwise make `now - t0` read huge and expire the deadline instantly.
        let t0 = self.epoch_secs_monotonic();
        loop {
            // Block, do not spin - and now it actually does. This comment said so while the line
            // beneath it polled, which is how the claim survived so long unexamined.
            if let Some(r) = self.await_slice(Self::AWAIT_SLICE_MS) { return DeadlineOutcome::Reply(r); }
            if self.epoch_secs_monotonic() - t0 >= max_secs {
                // NOT RECLAIMED: the send succeeded, so the kernel moved the reply cap to the peer and emptied
                // this slot - and the first cap this task RECEIVED since went into it. Removing "the reply
                // cap" here destroyed that one: a client's reply cap stashed mid-wait, so its answer went
                // into a dead cap (backlog/67). The peer removes the cap after answering.
                // CALLER BEWARE: the request was already SENT, so the peer will reply into our endpoint
                // whether we are listening or not. Leaving the reply cap alone does not remove that message
                // from the queue - the NEXT `try_recv` on this endpoint may return the ABANDONED reply
                // instead of the answer it expects. That has bitten for real: a timed-out fs read left a
                // 1-byte `[FS_NOTFOUND]` behind, and the next command consumed it and reported a healthy
                // mounted disk as absent. `request_with_reply_abortable` avoids this with a DRAIN at its
                // own top; this variant cannot drain blindly, because a service that also SERVES on this
                // endpoint (net-stack) would discard live client requests. So a caller that can time out
                // must deal with the late reply itself - see the shell's `drain_stale_fs_replies`, which
                // discards leftovers BEFORE sending rather than chasing one afterwards (a race that
                // cannot be won: if the late reply has not arrived yet, the NEXT command eats it).
                return DeadlineOutcome::Timeout;
            }
            self.yield_cpu();
        }
    }

    /// The sifting wait. A deliberate near-copy of `request_with_reply_deadline_outcome_inner`
    /// rather than a shared body that one of them passes a closure into.
    ///
    /// Folding the two together would put a generic parameter on the hot path of every service that
    /// makes a bounded request, for the benefit of the one that sifts. The SDK has been bitten by
    /// exactly that kind of tidying before: a one-line wrapper returning a 4 KiB `Message` by value
    /// cost `fs` a whole extra stack frame per request and took it over its stack limit on hardware
    /// while every QEMU run stayed green (see `await_slice`). The duplication is four lines of loop,
    /// it is visible, and both copies are right here.
    fn request_with_reply_deadline_sifted_inner(
        &self,
        peer: &str,
        msg:  &crate::ipc::Message,
        max_secs: i64,
        mut mine: impl FnMut(&crate::ipc::Message) -> bool,
    ) -> DeadlineOutcome {
        let target = match self.find_send_slot(peer) { Some(s) => CapHandle(s), None => return DeadlineOutcome::SendFailed };
        let self_grant = match self.self_grant_handle() { Some(g) => g, None => return DeadlineOutcome::SendFailed };
        let reply_cap = match self.derive_cap(self_grant) { Some(c) => c, None => return DeadlineOutcome::SendFailed };
        if let Err(e) = self.send_with_cap_by_handle(target, reply_cap, msg) {
            self.remove_cap(reply_cap);   // send failed: reclaim the untransferred reply cap (no leak)
            return match e {
                crate::ipc::IpcError::QueueFull => DeadlineOutcome::QueueFull,
                _ => DeadlineOutcome::SendFailed,
            };
        }
        let t0 = self.epoch_secs_monotonic();
        loop {
            if let Some(r) = self.await_slice(Self::AWAIT_SLICE_MS) {
                // Asked at the moment of arrival, so `take_pending_cap` and `last_recv_badge` inside
                // it still describe THIS message. A `false` hands ownership to the caller and the
                // wait goes on; the deadline below is unchanged by how many arrive, so a flood of
                // other traffic cannot extend it.
                if mine(&r) { return DeadlineOutcome::Reply(r); }
            }
            if self.epoch_secs_monotonic() - t0 >= max_secs {
                // NOT RECLAIMED: the send succeeded, so the kernel moved the reply cap to the peer and emptied
                // this slot - and the first cap this task RECEIVED since went into it. Removing "the reply
                // cap" here destroyed that one: a client's reply cap stashed mid-wait, so its answer went
                // into a dead cap (backlog/67). The peer removes the cap after answering.
                // Same caveat as the unsifted twin: the request WAS sent, so the peer answers whether
                // or not anyone is still listening, and that answer will arrive later. A caller that
                // times out must expect to meet it - here, `mine` will simply be asked about it and
                // can say no.
                return DeadlineOutcome::Timeout;
            }
            self.yield_cpu();
        }
    }

    /// Like [`Self::request_with_reply_deadline`] but ABORTABLE: while waiting it also drains the console
    /// and, on q/Q/ESC, returns [`ReqOutcome::Aborted`] IMMEDIATELY - it does NOT wait for the in-flight
    /// reply (that wait felt like "it pauses instead of quitting"). The request is already sent, so the
    /// peer replies into our endpoint whether we listen or not; that late reply is cleared by the DRAIN at
    /// the top of the *next* abortable request, so it never pollutes a later command (the `net scan` ->
    /// `net` "0.0.0.0 / 00:00 MAC" bug that drain closes). Sends the request ONCE (no re-trigger). Use for
    /// any interactive command that blocks on a peer (the "q to quit" rule, `utilities/0_conventions.md`).
    /// A service with no console foreground never sees input, so this degrades to the plain deadline wait.
    fn request_with_reply_abortable_inner(
        &self,
        peer: &str,
        msg:  &crate::ipc::Message,
        max_secs: i64,
    ) -> ReqOutcome {
        // Drain any stale reply a prior INSTANT-abort left in our endpoint (the peer replied after we
        // stopped listening), so this request cannot read it as its own. Safe for the shell - a client
        // whose endpoint only holds replies; between commands it is otherwise empty.
        while self.try_recv().is_some() {}
        let target = match self.find_send_slot(peer) { Some(s) => CapHandle(s), None => return ReqOutcome::Timeout };
        let self_grant = match self.self_grant_handle() { Some(g) => g, None => return ReqOutcome::Timeout };
        let reply_cap = match self.derive_cap(self_grant) { Some(c) => c, None => return ReqOutcome::Timeout };
        if self.offer_request(target, reply_cap, msg).is_err() {
            self.remove_cap(reply_cap);
            return ReqOutcome::Timeout;
        }
        let t0 = self.epoch_secs_monotonic();
        loop {
            // BLOCK for the reply, with a short timeout - do not spin. `try_recv` + `yield_cpu` kept this
            // task permanently RUNNABLE, so a core running a waiting command never reached the scheduler's
            // idle path at all. On ARM that path is where USB hot-plug is watched, so plugging or
            // unplugging anything during a `ping` was noticed only once the ping ended: `events` queued
            // up and all arrived at the prompt. It also pegged the core at 100% for a task that is, in
            // truth, doing nothing (the same busy-wait the `observe` and muted loops were already fixed
            // for - see the shell's MUTED_POLL_MS). Blocking parks the task, the core halts, idle work runs.
            if let Some(r) = self.await_slice(Self::AWAIT_SLICE_MS) {
                // DO NOT remove the reply cap on a REPLY. The send already removed it.
                //
                // §8.5: a cap embedded in a message "is transferred and REMOVED from sender's table".
                // The instant the request went out this slot stopped being ours, and by the time the
                // reply lands the kernel has REUSED it - `CapTable::insert` hands out the first EMPTY
                // slot, and this one is empty. When the reply carries an embedded cap (a file cap from
                // `fs`), that cap lands in exactly this slot, so removing "the reply cap" DELETES IT.
                //
                // Measured on the Pi 4: `fcap` printed `file=12 reply=12` - the file handle and the
                // next derived cap were the same slot, which is only possible if the slot was empty.
                // Every file-cap read/write then failed before reaching `fs`, and two negative
                // sub-checks "passed" vacuously because no invoke could get that far.
                //
                // A remove-by-stale-index can bite ANY request whose reply carries a cap, not just
                // fcap - and the abort and timeout paths below are no different: the send DID deliver
                // there too, so the slot is not ours either, and they leave it alone (backlog/67).
                return ReqOutcome::Reply(r);
            }
            while let Some(b) = self.try_console_read() {
                // q/Q/ESC aborts IMMEDIATELY - never wait for the in-flight reply (that wait was the
                // "it pauses instead of quitting" complaint). The peer's late reply lands in our endpoint
                // and the drain atop the NEXT abortable request clears it, so it pollutes nothing.
                // "Immediately" still holds: the abort does not wait on the peer. What the block adds is
                // up to one poll interval before the keypress is LOOKED at - tens of milliseconds, under
                // the threshold at which a person can tell, and the same trade the observe loop makes.
                if b == b'q' || b == b'Q' || b == 0x1b { return ReqOutcome::Aborted; }
            }
            if self.epoch_secs_monotonic() - t0 >= max_secs {
                return ReqOutcome::Timeout;
            }
        }
    }

    /// Like [`Self::request_with_reply_abortable`], but if no reply has arrived after
    /// `hint_after_secs` it invokes `on_linger` ONCE (e.g. to print a "[q] quit" hint) and keeps
    /// waiting/aborting. A snappy reply never fires the hint, so a fast request stays silent and only
    /// a genuinely lingering wait tells the user they can bail. Abort semantics are identical to
    /// `request_with_reply_abortable` (q/Q/ESC -> `Aborted` immediately; the request is sent once and
    /// a late reply is drained atop the next abortable/qhint request). The hint text lives in the
    /// caller's closure, so this stays mechanism - the SDK provides the *timing*, the caller the UX.
    fn request_with_reply_qhint_inner(
        &self,
        peer: &str,
        msg:  &crate::ipc::Message,
        hint_after_secs: i64,
        max_secs: i64,
        leave_keys: &[u8],
        on_linger: impl FnOnce(),
        mut mine: impl FnMut(&crate::ipc::Message) -> bool,
        keep: Option<&dyn Fn(u8)>,
    ) -> ReqOutcome {
        // Drain any stale reply a prior INSTANT-abort left in our endpoint (see the abortable variant).
        while self.try_recv().is_some() {}
        let target = match self.find_send_slot(peer) { Some(s) => CapHandle(s), None => return ReqOutcome::Timeout };
        let self_grant = match self.self_grant_handle() { Some(g) => g, None => return ReqOutcome::Timeout };
        let reply_cap = match self.derive_cap(self_grant) { Some(c) => c, None => return ReqOutcome::Timeout };
        if self.offer_request(target, reply_cap, msg).is_err() {
            self.remove_cap(reply_cap);
            return ReqOutcome::Timeout;
        }
        let t0 = self.epoch_secs_monotonic();
        let mut on_linger = Some(on_linger);   // FnOnce, fired at most once when the wait lingers
        loop {
            // Block, do not spin - see `request_with_reply_abortable`. This is the variant `net`/`ping`
            // actually use (the "[q] quit" hint), so it is the one that kept core 0 permanently
            // busy during a continuous ping and starved the idle-path USB hot-plug watch.
            //
            // THE CONSOLE IS NOT TOUCHED UNTIL `[q] quit` HAS BEEN SHOWN. Before the notice the
            // operator has been offered nothing to press, so a key typed then is typing ahead, not an
            // answer: reading it would either lose it or - a `q` meant for the NEXT command, a pager
            // say - end this wait. And most requests are answered before the notice is due, so they
            // now block in the kernel for the whole wait instead of polling the console every slice
            // (`backlog/fixed/80` H16: both of those made the shell keep its noticing waits out of scripts
            // and pipelines).
            let noticed = on_linger.is_none();
            let slice = if noticed { Self::AWAIT_SLICE_MS } else { Self::PRE_NOTICE_SLICE_MS };
            if let Some(r) = self.await_slice(slice) {
                // DO NOT remove the reply cap on a REPLY. The send already removed it.
                //
                // §8.5: a cap embedded in a message "is transferred and REMOVED from sender's table".
                // The instant the request went out this slot stopped being ours, and by the time the
                // reply lands the kernel has REUSED it - `CapTable::insert` hands out the first EMPTY
                // slot, and this one is empty. When the reply carries an embedded cap (a file cap from
                // `fs`), that cap lands in exactly this slot, so removing "the reply cap" DELETES IT.
                //
                // Measured on the Pi 4: `fcap` printed `file=12 reply=12` - the file handle and the
                // next derived cap were the same slot, which is only possible if the slot was empty.
                // Every file-cap read/write then failed before reaching `fs`, and two negative
                // sub-checks "passed" vacuously because no invoke could get that far.
                //
                // A remove-by-stale-index can bite ANY request whose reply carries a cap, not just
                // fcap - and the abort and timeout paths below are no different: the send DID deliver
                // there too, so the slot is not ours either, and they leave it alone (backlog/67).
                //
                // Asked at the moment of arrival, so a cap the message carries is still pending for
                // `mine` to take; `false` hands the message to the caller and the wait goes on.
                if mine(&r) {
                    return ReqOutcome::Reply(r);
                }
            }
            if noticed {
                while let Some(b) = self.try_console_read() {
                    if leave_keys.contains(&b) { return ReqOutcome::Aborted; }
                    // Not a leave key: the operator's typing, handed back when the caller can keep it.
                    if let Some(k) = keep { k(b); }
                }
            }
            let elapsed = self.epoch_secs_monotonic() - t0;
            if elapsed >= hint_after_secs {
                if let Some(f) = on_linger.take() { f(); }
            }
            if elapsed >= max_secs {
                return ReqOutcome::Timeout;
            }
        }
    }

    /// Wait for the next message on our own recv endpoint, ABORTABLE (q/Q/ESC) and bounded by an RTC
    /// deadline, WITHOUT sending anything. The failure-aware twin of [`Self::recv`]: where `recv`
    /// blocks forever (and panics on error), this returns [`ReqOutcome::Timeout`] if no message
    /// arrives within `max_secs` and [`ReqOutcome::Aborted`] the instant the user presses q. A receive
    /// ERROR is not reported: it reads as no message, so the wait runs to its deadline. Use it to
    /// await a reply we already sent by some path OTHER than a named-peer request - a badged
    /// `resource_invoke` (a file/socket capability), or draining a pipe filter's stream - where
    /// [`Self::request_with_reply_abortable`] (which does its own send) does not fit. This is the wait
    /// half of the abortable request, factored out: a peer that received our invocation but died before
    /// replying, or a filter that wedges mid-stream, can no longer hang us (Commandment VIII - wait on
    /// truth *including failure*). The caller owns any reply cap it derived and reclaims it only when the send failed (backlog/67), not on every
    /// outcome. A service with no console foreground never sees input, so this degrades to a plain
    /// deadline wait. Does NOT drain a stale reply first - a caller that can be re-entered after an
    /// abort should `while self.try_recv().is_some() {}` before it sends (as the request variants do).
    pub fn recv_abortable_deadline(&self, max_secs: i64) -> ReqOutcome {
        self.recv_abortable_deadline_keeping(max_secs, &|_| {})
    }

    /// [`Self::recv_abortable_deadline`] that HANDS BACK every key it reads that is not a quit key, in
    /// order, as [`Self::request_with_reply_qhint_keeping`] does for a request. The plain form reads the
    /// console to see a `q` and discards everything else, so what the operator typed while a pipe stage
    /// ran was lost: `ask`'s answer takes seconds, and the next command typed during it vanished in QEMU
    /// (the receive-side half of `backlog/fixed/80` H16).
    pub fn recv_abortable_deadline_keeping(&self, max_secs: i64, keep: &dyn Fn(u8)) -> ReqOutcome {
        self.recv_abortable_deadline_noticing(max_secs, i64::MAX, &|| {}, keep)
    }

    /// [`Self::recv_abortable_deadline_keeping`] that calls `on_linger` ONCE when no message has come
    /// after `hint_after_secs` - the receive-side twin of [`Self::request_with_reply_qhint`]. The SDK
    /// supplies the timing and the caller the words, as there: a task with no prompt has nothing to
    /// print, so this never writes to the console itself.
    pub fn recv_abortable_deadline_noticing(
        &self, max_secs: i64, hint_after_secs: i64, on_linger: &dyn Fn(), keep: &dyn Fn(u8),
    ) -> ReqOutcome {
        let t0 = self.epoch_secs_monotonic();
        let mut lingered = false;
        loop {
            // Block, do not spin - see `await_slice`.
            if let Some(r) = self.await_slice(Self::AWAIT_SLICE_MS) { return ReqOutcome::Reply(r); }
            while let Some(b) = self.try_console_read() {
                if b == b'q' || b == b'Q' || b == 0x1b { return ReqOutcome::Aborted; }
                keep(b);
            }
            let elapsed = self.epoch_secs_monotonic() - t0;
            if !lingered && elapsed >= hint_after_secs {
                lingered = true;
                on_linger();
            }
            if elapsed >= max_secs {
                return ReqOutcome::Timeout;
            }
        }
    }

    /// Reacquire a fresh SEND cap to `peer` and point the named-peer cache at it, so subsequent
    /// `try_send(peer)` / `send(peer)` use the new cap. Returns `false` if `peer` cannot currently
    /// be resolved (e.g. it has not finished respawning) - the caller should retry on a later tick.
    ///
    /// A thin shim over `reacquire_cap` (syscall 10): name resolution is the **kernel name
    /// directory**, not a service. The directory is populated synchronously at each service's spawn,
    /// so there is no round-trip and no bootstrap chicken-and-egg (the directory lives in the kernel,
    /// always reachable). `reacquire_cap` also updates the send-cap cache.
    #[must_use = "the peer is NOT reacquired if this is false"]
    pub fn reacquire_by_name(&self, peer: &str) -> bool {
        self.reacquire_cap(peer).is_ok()
    }

    /// Handle to this service's `SEND|GRANT` cap to its **own** endpoint, minted at
    /// spawn. A service hands a copy of its endpoint to a peer by deriving one
    /// (`derive_cap`) and granting it across - keeping this original so it can derive
    /// again later. `None` if the service has no endpoint.
    pub fn self_grant_handle(&self) -> Option<crate::capability::CapHandle> {
        let slot = Self::ctx().self_grant_slot;
        if slot == u32::MAX { None } else { Some(crate::capability::CapHandle(slot)) }
    }

    /// The send cap for a named peer, or `None` if this service has no such peer.
    ///
    /// The public form of what `request_with_reply` resolves internally. Exposed for the case that
    /// helper cannot serve: a request whose reply must NOT be waited for. A single-threaded service that
    /// owns an in-memory answer (the `time` service and its wall clock) has to be able to send a
    /// best-effort request to a slow peer without becoming unable to answer its own clients meanwhile -
    /// so it pairs this with `derive_cap` + `send_with_cap_by_handle` and picks the reply up later, out
    /// of its ordinary receive loop, matched by the protocol's correlation tag.
    pub fn send_peer_handle(&self, peer: &str) -> Option<crate::capability::CapHandle> {
        self.find_send_slot(peer).map(crate::capability::CapHandle)
    }

    /// Return the cap handle for the Nth send-peer entry (0-indexed).
    ///
    /// Used by property-test probes (P9) to access multiple cap slots wired to
    /// the same endpoint, verifying all are invalidated on endpoint death (§7.5).
    pub fn send_peer_at(&self, idx: usize) -> Option<crate::capability::CapHandle> {
        let data  = Self::ctx();
        let count = (data.send_peer_count as usize).min(MAX_SEND_PEERS);
        if idx >= count { return None; }
        let slot = data.send_peers[idx].slot;
        if slot == u32::MAX { None } else { Some(crate::capability::CapHandle(slot)) }
    }

    /// Send to a specific cap handle directly, bypassing peer-name lookup.
    ///
    /// Used by the probe service to test kernel cap enforcement (§22 Tests 3B, 9B).
    pub fn try_send_by_handle(
        &self,
        handle: crate::capability::CapHandle,
        msg:    &crate::ipc::Message,
    ) -> Result<(), crate::ipc::IpcError> {
        crate::ipc::try_send(handle, msg)
    }

    /// Send a message to a named peer WITH an embedded capability grant.
    ///
    /// The send-peer cap (which must carry the `GRANT` right) is transferred to
    /// the receiver. On success the calling service loses that cap (§7.6).
    /// Returns `CapNotGrantable` if the cap lacks `GRANT` - the cap is kept.
    pub fn send_with_cap(&self, peer: &str, msg: &crate::ipc::Message) -> Result<(), crate::ipc::IpcError> {
        let slot = self.find_send_slot(peer)
            .ok_or(crate::ipc::IpcError::CapError(crate::capability::CapError::CapNotHeld))?;
        // syscall 11 = SendWithCap
        // arg0 = (grant_slot << 16) | endpoint_slot - same slot holds both SEND and GRANT.
        let packed  = ((slot as u64) << 16) | (slot as u64);
        let payload = msg.payload_bytes();
        // SAFETY: syscall(11) = SendWithCap; packed and payload are from user space.
        let ret = unsafe {
            raw_syscall(11, packed, payload.as_ptr() as u64, payload.len() as u64)
        };
        if ret == 0 { Ok(()) } else { Err(crate::ipc::i64_to_ipc_error(ret)) }
    }

    /// Take the next pending received capability, if any.
    ///
    /// After `recv()` delivers a message containing an embedded cap, the kernel
    /// installs the cap into this task's table and queues the slot index. Call
    /// this once per embedded cap to retrieve each one.
    pub fn take_pending_cap(&self) -> Option<CapHandle> {
        // SAFETY: syscall(12) = TakePendingCap; no args.
        let ret = unsafe { raw_syscall(12, 0, 0, 0) };
        if ret >= 0 { Some(CapHandle(ret as u32)) } else { None }
    }

    /// Acquire a fresh SEND|GRANT cap to `peer` via the kernel name directory.
    ///
    /// Used by property-test probes that need to transfer capabilities (P3).
    /// Returns `None` if the service is not registered or the cap table is full.
    pub fn acquire_send_grant_cap(&self, peer: &str) -> Option<CapHandle> {
        let bytes = peer.as_bytes();
        let len   = bytes.len();
        if len == 0 || len > NAME_MAX { return None; }
        // SAFETY: syscall(10) = AcquireSendCap; arg2=1 requests SEND|GRANT.
        let ret = unsafe { raw_syscall(10, bytes.as_ptr() as u64, len as u64, 1) };
        if ret < 0 { None } else { Some(CapHandle(ret as u32)) }
    }

    /// Acquire a fresh SEND cap to `peer` via the kernel name directory.
    ///
    /// Returns the new cap handle, or `None` if the name is not registered.
    pub fn acquire_send_cap(&self, peer: &str) -> Option<CapHandle> {
        let bytes = peer.as_bytes();
        let len   = bytes.len();
        if len == 0 || len > NAME_MAX { return None; }
        // SAFETY: syscall(10) = AcquireSendCap; arg2=0 = SEND only.
        let ret = unsafe { raw_syscall(10, bytes.as_ptr() as u64, len as u64, 0) };
        if ret < 0 { None } else { Some(CapHandle(ret as u32)) }
    }

    /// Query the current generation of the named endpoint.
    ///
    /// Returns the generation counter as a u64, or 0 if the name is not
    /// registered OR the caller lacks INTROSPECT (query 2 is gated). Used by
    /// property tests P2 and P8 (§7.5, §14.2).
    pub fn inspect_endpoint_generation(&self, name: &str) -> u64 {
        let bytes = name.as_bytes();
        let len   = bytes.len();
        if len == 0 || len > NAME_MAX { return 0; }
        // SAFETY: syscall(13) = InspectKernel; query_id=2 = endpoint generation by name.
        let ret = unsafe {
            raw_syscall(13, 2, bytes.as_ptr() as u64, len as u64)
        };
        if ret < 0 { 0 } else { ret as u64 }
    }

    /// One byte from the COM2 operator channel, or `None` when the port is empty (kernel query 21).
    ///
    /// The kernel owns the UART - hardware, and §11.4 already sanctions it owning a serial console -
    /// and hands bytes out; the `control` service owns what they MEAN. Transport in the kernel,
    /// interpretation in a service (C1-6).
    pub fn com2_byte(&self) -> Option<u8> {
        // SAFETY: syscall(13) = InspectKernel; query 21 = pop one COM2 byte, -1 when empty.
        let v = unsafe { raw_syscall(13, 21, 0, 0) };
        if v < 0 { None } else { Some(v as u8) }
    }

    /// The board's own MAC address (kernel query 23), or `None` where the board cannot say.
    ///
    /// A NIC driver runs in userspace and cannot reach the board identity: on the Pi this address lives
    /// in the VideoCore mailbox, nowhere near the controller's register window. Without it a driver has
    /// to invent an address, and an invented address that is hardcoded is the same on every board -
    /// which stops being harmless the moment two of them share a network.
    pub fn board_mac(&self) -> Option<[u8; 6]> {
        // SAFETY: syscall(13) = InspectKernel; query 23 = the board MAC packed little-endian, -1 if none.
        let v = unsafe { raw_syscall(13, 23, 0, 0) };
        if v < 0 { return None; }
        let v = v as u64;
        Some([v as u8, (v >> 8) as u8, (v >> 16) as u8,
              (v >> 24) as u8, (v >> 32) as u8, (v >> 40) as u8])
    }

    /// Inject a test interrupt. Requires the FIRE_IRQ capability, held only by `control`; a non-holder
    /// gets `false` rather than a silent no-op.
    pub fn fire_irq(&self, irq: u8) -> bool {
        // SAFETY: syscall(51) = FireIrq; the kernel validates FIRE_IRQ by holdings.
        unsafe { raw_syscall(51, irq as u64, 0, 0) == 0 }
    }

    /// Return the bytes dynamically allocated by this task so far.
    ///
    /// Wraps InspectKernel query 0. Used by property test P4 (§10.3).
    pub fn inspect_kernel_alloc_bytes(&self) -> u64 {
        // SAFETY: syscall(13) = InspectKernel; query_id=0 = task alloc bytes.
        let ret = unsafe { raw_syscall(13, 0, 0, 0) };
        if ret < 0 { 0 } else { ret as u64 }
    }

    /// Return the count of live endpoints in the kernel routing table.
    ///
    /// Wraps InspectKernel query 1 (INTROSPECT-gated; 0 when refused). Used by property test P5 (§8.3).
    pub fn inspect_kernel_endpoint_count(&self) -> Option<u32> {
        // SAFETY: syscall(13) = InspectKernel; query_id=1 = live endpoint count.
        let ret = unsafe { raw_syscall(13, 1, 0, 0) };
        // A refusal (no INTROSPECT) or an error is `None`, never a value: it read as 0 until
        // 2026-10-10, a believable count of nothing (`backlog/fixed/80` K18).
        if ret < 0 { None } else { Some(ret as u32) }
    }

    /// Capacity of the in-kernel USB mass-storage device in 512-byte sectors, 0 if none is attached.
    /// Requires the `USB_DISK_RESOURCE` capability. Syscall 46. The kernel drives no USB disk on any
    /// port now (see [`USB_DISK_BUSY`]), so this is always 0 and the block calls below always fail.
    pub fn usb_disk_sectors(&self) -> u64 {
        // SAFETY: syscall(46) = UsbDiskInfo; no arguments, gated by the USB_DISK capability.
        let ret = unsafe { raw_syscall(46, 0, 0, 0) };
        if ret < 0 { 0 } else { ret as u64 }
    }

    /// Read the 512-byte block at `lba` from the USB mass-storage device into `dst`. Returns false if
    /// there is no device, the LBA is past the end, or the transfer failed. Requires `USB_DISK_RESOURCE`.
    /// Syscall 47.
    #[must_use = "the destination buffer is NOT valid data if this is false"]
    pub fn usb_disk_read(&self, lba: u64, dst: &mut [u8; 512]) -> bool {
        // SAFETY: syscall(47) = UsbDiskRead; the kernel writes exactly 512 bytes to `dst` on success,
        // through its checked user-pointer path.
        if !lba_fits_syscall_abi(lba) { return false; }
        let ret = unsafe { raw_syscall(47, lba, dst.as_mut_ptr() as u64, 0) };
        ret == 0
    }

    /// Write `src` as the 512-byte block at `lba` on the USB mass-storage device. Requires `USB_DISK_RESOURCE`.
    /// Syscall 48.
    #[must_use = "the block did NOT reach the medium if this is false"]
    pub fn usb_disk_write(&self, lba: u64, src: &[u8; 512]) -> bool {
        // SAFETY: syscall(48) = UsbDiskWrite; the kernel reads exactly 512 bytes from `src` through its
        // checked user-pointer path.
        if !lba_fits_syscall_abi(lba) { return false; }
        let ret = unsafe { raw_syscall(48, lba, src.as_ptr() as u64, 0) };
        ret == 0
    }

    /// Read a block, returning the raw status so BUSY is distinguishable from FAILED: `0` ok, the
    /// device is busy (`USB_DISK_BUSY` - re-ask, nothing is wrong), anything else a real error. The `bool` variants
    /// above collapse those two, which is exactly the conflation that made a busy stick look broken.
    pub fn usb_disk_read_status(&self, lba: u64, dst: &mut [u8; 512]) -> i64 {
        // SAFETY: syscall(47) = UsbDiskRead; the kernel writes exactly 512 bytes through its checked
        // user-pointer path.
        if !lba_fits_syscall_abi(lba) { return -1; }
        unsafe { raw_syscall(47, lba, dst.as_mut_ptr() as u64, 0) }
    }

    /// Write a block, returning the raw status (see `usb_disk_read_status`).
    pub fn usb_disk_write_status(&self, lba: u64, src: &[u8; 512]) -> i64 {
        // SAFETY: syscall(48) = UsbDiskWrite; the kernel reads exactly 512 bytes from `src`.
        if !lba_fits_syscall_abi(lba) { return -1; }
        unsafe { raw_syscall(48, lba, src.as_ptr() as u64, 0) }
    }

    /// Make previously written blocks durable on the USB mass-storage device (SCSI SYNCHRONIZE CACHE).
    /// Requires `USB_DISK_RESOURCE` WRITE. Syscall 49.
    ///
    /// A write is only ACKNOWLEDGED when `usb_disk_write` returns - the device may still be holding the
    /// bytes in a volatile buffer. Anything that promises durability (a format, a journal commit) has to
    /// ask for it, and check the answer: `false` means the data is NOT known to be on the medium.
    #[must_use = "prior writes are NOT durable if this is false"]
    pub fn usb_disk_flush(&self) -> bool {
        // SAFETY: syscall(49) = UsbDiskFlush; takes no arguments and touches no user memory.
        let ret = unsafe { raw_syscall(49, 0, 0, 0) };
        ret == 0
    }

    /// The SD/EMMC controller's base clock in Hz (0 if the platform does not report one), via
    /// InspectKernel query 20. The block driver needs it to compute its clock divider: on the BCM283x
    /// the controller's own capability register reports the base clock wrongly, and the driver holds
    /// only its controller's registers, so it cannot ask the platform firmware itself. A driver that
    /// gets 0 should report that rather than guess - a wrong divider runs the card's identification
    /// clock at the wrong speed, which fails on hardware and not under emulation. Ungated - task-neutral
    /// hardware info, like the console geometry and the RTC.
    pub fn emmc_base_clock_hz(&self) -> u32 {
        // SAFETY: syscall(13) = InspectKernel; query_id=20 = EMMC base clock (Hz).
        let ret = unsafe { raw_syscall(13, 20, 0, 0) };
        if ret < 0 { 0 } else { ret as u32 }
    }

    /// The discovered NIC's PCI identity as `vendor | device<<16` (0 if no NIC), via InspectKernel
    /// query 14. A NIC driver reads it to know which chip it is driving (e.g. Intel e1000 =
    /// 0x100E_8086 vs Realtek RTL8168 = 0x8168_10EC). Ungated - task-neutral hardware info.
    pub fn nic_vendor_device(&self) -> u32 {
        // SAFETY: syscall(13) = InspectKernel; query_id=14 = NIC vendor|device.
        let ret = unsafe { raw_syscall(13, 14, 0, 0) };
        if ret < 0 { 0 } else { ret as u32 }
    }

    /// Whether the PCI scan found an xHCI USB host controller (InspectKernel query 18, ungated
    /// task-neutral hardware fact). The supervisor reads it to skip spawning the `xhci` driver on a
    /// machine that has no xHCI controller (an idle driver would busy-hold a core). Falls back to
    /// `true` if the query is unavailable, preserving the always-spawn behaviour.
    pub fn xhci_present(&self) -> bool {
        // SAFETY: syscall(13) = InspectKernel; query_id=18 = USB controller presence bitmask.
        let ret = unsafe { raw_syscall(13, 18, 0, 0) };
        if ret < 0 { return true; } // query unavailable - spawn as before
        (ret & 0b01) != 0
    }

    /// Whether the PCI scan found an EHCI (USB 2.0) host controller (InspectKernel query 18, ungated).
    /// The supervisor reads it to skip spawning the `ehci` driver on a machine with no EHCI at all
    /// (e.g. the Wyse 5070), so an idle driver does not busy-hold a core. Falls back to `true` if the
    /// query is unavailable, preserving the always-spawn behaviour.
    pub fn ehci_present(&self) -> bool {
        // SAFETY: syscall(13) = InspectKernel; query_id=18 = USB controller presence bitmask.
        let ret = unsafe { raw_syscall(13, 18, 0, 0) };
        if ret < 0 { return true; } // query unavailable - spawn as before
        (ret & 0b10) != 0
    }

    /// Whether the PCI scan found a NIC this build can actually drive (present AND an e1000 or RTL8168),
    /// via InspectKernel query 18 (ungated). The supervisor reads it to skip spawning `nic-driver` (and
    /// its dependent `net-stack`) on a machine with no usable NIC, so they do not busy-hold cores.
    /// Falls back to `true` if the query is unavailable, preserving the always-spawn behaviour.
    pub fn nic_present(&self) -> bool {
        // SAFETY: syscall(13) = InspectKernel; query_id=18 = hardware-driver presence bitmask.
        let ret = unsafe { raw_syscall(13, 18, 0, 0) };
        if ret < 0 { return true; } // query unavailable - spawn as before
        (ret & 0b100) != 0
    }

    /// The NIC's register-space MMIO base (the BAR the PCI scan chose), 0 if none. InspectKernel query
    /// 15, ungated. A diagnostic - a driver reads it to confirm which BAR it was handed.
    pub fn nic_mmio_base(&self) -> u64 {
        // SAFETY: syscall(13) = InspectKernel; query_id=15 = NIC MMIO base.
        let ret = unsafe { raw_syscall(13, 15, 0, 0) };
        if ret < 0 { 0 } else { ret as u64 }
    }

    /// Return the number of free physical frames.
    ///
    /// Wraps InspectKernel query 4. Queries 4-8 are INTROSPECT-gated, and these wrappers read a
    /// refusal as 0 (or, for the core count, 1) - a value the caller cannot tell from a real one.
    pub fn inspect_kernel_free_frames(&self) -> Option<u64> {
        // SAFETY: syscall(13) = InspectKernel; query_id=4 = free frame count.
        let ret = unsafe { raw_syscall(13, 4, 0, 0) };
        // A refusal (no INTROSPECT) or an error is `None`, never a value: it read as 0 until
        // 2026-10-10, a believable count of nothing (`backlog/fixed/80` K18).
        if ret < 0 { None } else { Some(ret as u64) }
    }

    /// Return the total usable physical frames at boot time.
    ///
    /// Wraps InspectKernel query 5.
    pub fn inspect_kernel_total_frames(&self) -> Option<u64> {
        // SAFETY: syscall(13) = InspectKernel; query_id=5 = total frame count.
        let ret = unsafe { raw_syscall(13, 5, 0, 0) };
        // A refusal (no INTROSPECT) or an error is `None`, never a value: it read as 0 until
        // 2026-10-10, a believable count of nothing (`backlog/fixed/80` K18).
        if ret < 0 { None } else { Some(ret as u64) }
    }

    /// The endpoint task `slot` OWNS, or 0 if it owns none (InspectKernel query 24, INTROSPECT).
    ///
    /// With [`Self::task_awaits_endpoint`] this is the whole of the blocked-chain walk: map the
    /// endpoint a stuck task awaits back to the task that owns it, then ask what THAT one awaits.
    /// See `utilities/47_events.md`.
    pub fn task_own_endpoint(&self, slot: u32) -> Option<u64> {
        // SAFETY: syscall(13) = InspectKernel; query_id=24 = the task's own endpoint.
        let ret = unsafe { raw_syscall(13, 24, slot as u64, 0) };
        // A refusal (no INTROSPECT) or an error is `None`, never a value: it read as 0 until
        // 2026-10-10, a believable count of nothing (`backlog/fixed/80` K18).
        if ret < 0 { None } else { Some(ret as u64) }
    }

    /// The endpoint task `slot` is blocked-in-CALL awaiting a reply from, or 0 if no call is in
    /// flight (InspectKernel query 25, INTROSPECT).
    ///
    /// A best-effort snapshot, on the same contract as [`Self::task_stat`]: the kernel reads it
    /// without taking the routing lock, so an observer cannot stall the thing it observes.
    pub fn task_awaits_endpoint(&self, slot: u32) -> Option<u64> {
        // SAFETY: syscall(13) = InspectKernel; query_id=25 = the endpoint awaited in a CALL.
        let ret = unsafe { raw_syscall(13, 25, slot as u64, 0) };
        // A refusal (no INTROSPECT) or an error is `None`, never a value: it read as 0 until
        // 2026-10-10, a believable count of nothing (`backlog/fixed/80` K18).
        if ret < 0 { None } else { Some(ret as u64) }
    }

    /// The kernel's BOOT RECORD size: `(held, capacity)` in bytes, or `None` without INTROSPECT
    /// (InspectKernel query 27). The record is a fixed copy of the first bytes ever logged, which
    /// never wraps; `held == capacity` means it filled, and later lines are only in the kernel's ring
    /// (until it wraps) and on serial.
    pub fn boot_record_size(&self) -> Option<(usize, usize)> {
        let held = self.boot_record_query(0, 0);
        let cap = self.boot_record_query(1, 0);
        if held < 0 || cap < 0 { None } else { Some((held as usize, cap as usize)) }
    }

    /// Copy the boot record from `offset` into `buf`: `Some(n)` bytes, `Some(0)` at the end, `None`
    /// refused (InspectKernel query 27, INTROSPECT). The buffer is the kernel's chunk size exactly, so
    /// no read can ask it to write past the end of one.
    pub fn boot_record_read(&self, offset: usize, buf: &mut [u8; BOOT_READ_CHUNK]) -> Option<usize> {
        let r = self.boot_record_query(offset as u64, buf.as_mut_ptr() as u64);
        if r < 0 { None } else { Some(r as usize) }
    }

    fn boot_record_query(&self, a1: u64, a2: u64) -> i64 {
        // SAFETY: syscall(13) = InspectKernel; query_id=27 = the boot record. `a2` is 0 or a buffer of
        // BOOT_READ_CHUNK bytes the caller holds mutably; the kernel validates the range before writing.
        unsafe { raw_syscall(13, 27, a1, a2) }
    }

    /// The wall-clock datetime captured by the kernel at **boot** (InspectKernel query 12, ungated).
    /// Same packed layout as `datetime`. Pairs with `datetime` to compute uptime as a wall-clock
    /// delta - portable across timer modes (a tick counter's rate is not: periodic-mode QEMU ticks
    /// at ~10 Hz, TSC-deadline HW at 100 Hz). Returns the epoch (all-zero fields) if not captured.
    pub fn boot_datetime(&self) -> Datetime {
        // SAFETY: syscall(13) = InspectKernel; query_id=12 = packed boot datetime.
        let p = unsafe { raw_syscall(13, 12, 0, 0) } as u64;
        Self::unpack_datetime(p)
    }

    /// System uptime in **seconds** = now − boot, both from the hardware RTC. Never negative
    /// (saturates at 0). The `uptime` shell command renders this. Wall-clock based, so it is
    /// correct regardless of the APIC timer mode (unlike a raw tick counter).
    pub fn uptime_secs(&self) -> i64 {
        // "now" is the DEGLITCHED MONOTONIC clock (query 17), not the raw RTC datetime (which is frozen
        // at 0 on a board with no RTC, e.g. the Pi 2 - the reason the old `datetime - boot` read 0 there).
        let now = self.epoch_secs_monotonic();
        let boot_epoch = self.boot_datetime().epoch_secs();
        // A board with no RTC reports a zero/garbage boot datetime whose epoch is far before 2001 (a
        // packed all-zeros datetime unpacks to ~year 0 = ~-6.2e10 s - subtracting THAT would ADD ~1970
        // years). There the monotonic clock is already seconds-since-boot (its CNTPCT baseline, ARM), so
        // it IS the uptime. With a real RTC, uptime is the elapsed wall-clock since boot.
        if boot_epoch < 1_000_000_000 { now.max(0) } else { (now - boot_epoch).max(0) }
    }

    /// Timer ticks the given core spent running a user task (not idle) since boot.
    ///
    /// Wraps InspectKernel query 6 (arg1 = core index).
    pub fn inspect_core_active_ticks(&self, core: u32) -> Option<u64> {
        // SAFETY: syscall(13) = InspectKernel; query_id=6, arg1=core.
        let ret = unsafe { raw_syscall(13, 6, core as u64, 0) };
        // A refusal (no INTROSPECT) or an error is `None`, never a value: it read as 0 until
        // 2026-10-10, a believable count of nothing (`backlog/fixed/80` K18).
        if ret < 0 { None } else { Some(ret as u64) }
    }

    /// Total timer ticks seen on the given core since boot.
    ///
    /// Wraps InspectKernel query 7 (arg1 = core index).
    pub fn inspect_core_total_ticks(&self, core: u32) -> Option<u64> {
        // SAFETY: syscall(13) = InspectKernel; query_id=7, arg1=core.
        let ret = unsafe { raw_syscall(13, 7, core as u64, 0) };
        // A refusal (no INTROSPECT) or an error is `None`, never a value: it read as 0 until
        // 2026-10-10, a believable count of nothing (`backlog/fixed/80` K18).
        if ret < 0 { None } else { Some(ret as u64) }
    }

    /// Number of CPU cores ready since boot.
    ///
    /// Wraps InspectKernel query 8.
    pub fn inspect_core_count(&self) -> u32 {
        // SAFETY: syscall(13) = InspectKernel; query_id=8.
        let ret = unsafe { raw_syscall(13, 8, 0, 0) };
        // A refusal or an error is NOT "one core": it said 1 until 2026-10-10, so a caller without
        // INTROSPECT reported a one-core machine (`backlog/fixed/80` H20). 0 is "the kernel did not say", and
        // a caller must not present it as a count.
        if ret <= 0 { 0 } else { ret as u32 }
    }

    /// Terminal geometry as `(rows, cols)` text cells, or `(0, 0)` if it cannot be determined.
    ///
    /// **Asked of the `console` service, not the kernel.** It used to be `InspectKernel` query 9, which
    /// is now deleted: rows and columns are derived from the safe-area inset, the cell size and the
    /// font-scale rule, all of which live in the terminal - so the terminal is the only party that can
    /// answer, and asking anyone else would be a second source of truth (Commandment III).
    ///
    /// `(0, 0)` means **unknown**, and callers already treat it that way explicitly rather than
    /// substituting a size behind the user's back (the pager prints unpaged; `edit` falls back to 24x80
    /// and says so). It is returned when there is no console service, when it is mid-restart, or when it
    /// holds no framebuffer - all cases where a guessed geometry would be a silent fallback.
    ///
    /// Costs one IPC round trip, so callers that lay out a screen should ask once and keep the answer
    /// for that screen rather than per line.
    pub fn console_dims(&self) -> (u16, u16) {
        let mut buf = [0u8; 8];
        // Opcode 1 = REQ_DIMS (`services/console/src/main.rs`). The reply is rows then cols, u16 LE.
        // ACQUIRE THE PEER BY NAME IF WE WERE NEVER WIRED TO IT.
        //
        // The shell's contract declares `console` as a send peer, but the supervisor cannot wire it:
        // the shell is spawned BEFORE the console service exists, so there is nothing to hand it. The
        // kernel name directory is exactly the answer to that (§14.3 - reacquire by name), and
        // `time_rpc` in the shell already does this. This did not, so `find_send_slot` missed, the
        // request returned None, and the caller read (0, 0).
        //
        // Zero rows is not a small error here: `cmd_help` takes it to mean "no terminal" and prints
        // its whole table instead of paging it. A feature disappeared because a lookup missed, which
        // is the quiet kind of failure §26.7 is about - nothing was reported, it simply stopped
        // behaving.
        let mut n = self.request_with_reply_deadline_into("console", &[1u8], &mut buf, 2);
        if n.is_none() && self.reacquire_by_name("console") {
            n = self.request_with_reply_deadline_into("console", &[1u8], &mut buf, 2);
        }
        match n {
            Some(k) if k >= 4 => (
                u16::from_le_bytes([buf[0], buf[1]]),
                u16::from_le_bytes([buf[2], buf[3]]),
            ),
            _ => (0, 0),
        }
    }

    // RETAINED REASONING FROM THE DELETED `console_scroll`, and it is `//` rather than `///` on
    // purpose. A `///` run attaches to the item BELOW it, and the item below this one is
    // `console_history` - so while these lines were doc comments, rustdoc documented
    // `console_history` as "Move the console's scrolled-back view" and handed the reader a
    // deleted method's contract as if it were that method's. The words are kept because two of
    // the lessons in them were expensive and are still true of anything that asks the console a
    // question: a failure must not wear a valid value, and a KEYSTROKE may not wait seconds.
    // The last sentence below stops mid-clause; its tail went with the code, and it is left
    // truncated rather than completed by guess.
    //
    // Move the console's scrolled-back view, returning `(lines back, most it can go back)`.
    //
    // `action` is one of the `SCROLL_*` values in `services/console/src/term.rs`: 0 live, 1 line
    // up, 2 line down, 3 page up, 4 page down, 5 oldest kept.
    //
    // The framebuffer console has no scrollback of its own - reaching the bottom of the screen used
    // to mean the top was gone permanently - and this is how a holder of the keyboard asks to look
    // back. The console does the arithmetic, because only it knows how many lines it is holding and
    // how tall the screen is; a second copy of either number here would be a second thing to drift.
    //
    // **THE SCROLL IS A REQUEST, NOT AN ESCAPE SEQUENCE.** Console output is untrusted content: a
    // file being `read` can hold any bytes, so a scroll expressed in the byte stream would let a
    // file scroll the view of the terminal showing it. A request carries a reply cap and output
    // does not, so the two channels are separated by construction.
    //
    // **`None` MEANS THE CONSOLE DID NOT ANSWER, and that is not the same as `(0, 0)`.**
    //
    // It returned `(0, 0)` for both "the view is at live" and "I could not reach the console",
    // and the caller could not tell them apart. On a Dell Wyse, 2026-09-18, that turned a timed-out
    // request into a plausible success: the shell believed it had scrolled back to live, left the
    // scrollback view, and returned to the prompt **while the screen was still showing history** -
    // the two disagreeing about what was on the display, with nothing said. A failure wearing a
    // valid value is the exact shape invariant 12 and 26.7 exist to forbid, and it is worth the
    // `Option` to make it unrepresentable.
    //
    // **ONE SECOND, AND NO RETRY - because this is a KEYSTROKE.** It used to be the same two
    // seconds plus a reacquire-and-retry that `console_dims` uses, which is right for a
    // once-per-command lookup and wrong here: it let a single press of an arrow key block the
    // shell for four seconds, which the operator experiences as the machine locking up. Nothing
    // above the kernel may hang on a peer (Commandment V); a scroll that cannot be served quickly
    // must FAIL quickly and say so.
    //
    // Dropping the retry costs little. It exists so a client survives the peer RESTARTING, and a
    // console that has just restarted is rebuilding the screen anyway; the next `console_dims`
        // `console_scroll` IS GONE, and it is the mechanism `backlog/fixed/37` was about.
    //
    // It asked the console to move its own view, and the console could not answer until it had
    // repainted the framebuffer - `paint_view` + `present`, synchronously, inside the request. On a
    // 3840x2160 panel that is the most expensive thing it does, and the caller gave it one second
    // from the same core. That is why a scroll took "over two seconds to answer": nothing was lost
    // or stuck, the work simply did not fit the deadline.
    //
    // `console_history` below replaced it by moving the WORK rather than tuning the number: the
    // console hands over bytes, and the caller paints its own screen through ordinary output, which
    // is a send and carries no deadline at all. Deleting this leaves no way to reintroduce the
    // shape by accident.

    /// Read a page of the console's scrollback AS DATA, starting at line `from` (0 = oldest KEPT).
    ///
    /// Returns `(lines, total retained, any line has aged out, bytes written)`. The payload after
    /// the 4-byte header is `lines` records of `[len, bytes...]`; `bytes` is the whole reply so a
    /// caller can bound its own walk.
    ///
    /// **The replacement for the deleted `console_scroll`, which could not be made cheap.**
    /// Moving the terminal's own view forces a full repaint before the console can reply - on a
    /// 3840x2160 panel the most expensive thing it does, paid inside the caller's blocking request,
    /// from the same core (`backlog/fixed/37`). This asks for BYTES: the console copies out of its ring
    /// and replies, and the caller paints its own screen through ordinary output, which is a send
    /// rather than a call and carries no deadline at all.
    ///
    /// The history is BOUNDED (32 KiB / 512 lines). `aged` is how a view says "this is the oldest
    /// line kept" rather than implying it is the start of the session (§26.7).
    pub fn console_history(&self, from: u16, buf: &mut [u8]) -> Option<(usize, u16, bool, usize)> {
        // Opcode 3 = REQ_HISTORY (`services/console/src/main.rs`).
        let req = [3u8, from as u8, (from >> 8) as u8];
        let k = self.request_with_reply_deadline_into("console", &req, buf, 1)?;
        if k < 4 { return None; }
        Some((buf[0] as usize, u16::from_le_bytes([buf[1], buf[2]]), buf[3] != 0, k))
    }

    /// Whether the input driver has reported setup complete (syscall 13, query 10).
    /// The deterministic end-of-boot signal: the shell watches it to auto-clear the
    /// boot screen the moment the keyboard subsystem is up. Ambient.
    pub fn input_ready(&self) -> bool {
        // SAFETY: syscall(13) = InspectKernel; query_id=10 = input-ready flag.
        unsafe { raw_syscall(13, 10, 0, 0) > 0 }
    }

    /// Report that input-subsystem setup is complete (syscall 27). Called by the
    /// USB keyboard driver (xHCI) in every terminal path once it has finished - the
    /// end-of-boot signal. Requires the CONSOLE_PUSH cap (the input driver only).
    pub fn signal_input_ready(&self) {
        let slot = Self::ctx().console_push_slot;
        if slot == u32::MAX { return; }
        // SAFETY: syscall(27) = SignalInputReady; slot is the kernel-written cap index.
        let _ = unsafe { raw_syscall(27, slot as u64, 0, 0) };
    }

    /// Read the hardware TSC (Time Stamp Counter) via the kernel.
    ///
    /// Returns the arch's free-running counter (InspectKernel query 3): RDTSC on x86, CNTPCT on
    /// ARM; its rate is `tsc_ticks_per_10ms`. Useful for measuring kernel operation latencies
    /// in benchmark probes (§22 Perf B1-B10). Not comparable across hosts.
    pub fn read_tsc(&self) -> u64 {
        // SAFETY: syscall(13) = InspectKernel; query_id=3 = read TSC.
        let ret = unsafe { raw_syscall(13, 3, 0, 0) };
        if ret < 0 { 0 } else { ret as u64 }
    }

    /// Counter ticks per 10 ms, from the kernel's boot-time calibration (CPUID on x86, the measured
    /// timer rate / 100 on ARM) (InspectKernel query 16).
    /// Convert a TSC delta to milliseconds with `delta_cycles * 10 / tsc_ticks_per_10ms()`. Returns 0
    /// if the TSC was not calibrated (callers should then skip the millisecond conversion). `ping` uses
    /// it to report round-trip time.
    pub fn tsc_ticks_per_10ms(&self) -> u64 {
        // SAFETY: syscall(13) = InspectKernel; query_id=16 = TSC ticks per 10 ms quantum.
        let ret = unsafe { raw_syscall(13, 16, 0, 0) };
        if ret < 0 { 0 } else { ret as u64 }
    }

    /// Sleep for approximately `ms` milliseconds - the PORTABLE way to pace a loop.
    ///
    /// Prefer this over `sleep(cycles)` for anything expressing a DURATION. A "cycle" is not a portable
    /// unit: on x86 it is a ~2 GHz CPU cycle, on the Pi 2 it is a tick of a ~1 MHz timer, so the same
    /// literal means two wildly different lengths of time. Services were written with x86 in mind -
    /// `60_000_000` meaning "~30 ms at 2 GHz" - and on ARM that same constant asks for ~60 SECONDS once
    /// the quantum is calibrated. Three loops (the shell's muted poll and observe q-poll, and `observe`'s
    /// repaint pace) were exactly that, and each would have become a minute-long stall.
    ///
    /// Converts through the kernel's own calibration, so it is right on any machine. Floors at one
    /// scheduler quantum: the underlying `sleep(0)` returns immediately, which would turn a paced loop
    /// into a busy spin - the opposite of what every caller wants.
    pub fn sleep_ms(&self, ms: u64) {
        self.sleep(self.duration_cycles(ms));
    }

    /// Cycles for approximately `ms` milliseconds, for the two syscalls that take a cycle count
    /// (`sleep`, `recv_timeout`). See [`Self::sleep_ms`] for why a raw cycle literal is not portable.
    ///
    /// Never returns 0. That matters beyond pacing: `recv_timeout(0)` means **block forever**, so a
    /// zero here would convert a bounded wait into an unbounded one - a hang produced by a unit
    /// conversion. An uncalibrated machine (query 16 reads 0) also lands on the floor, which is exactly
    /// the one-quantum behaviour those platforms had before any of this existed.
    pub fn duration_cycles(&self, ms: u64) -> u64 {
        let per_10ms = self.tsc_ticks_per_10ms();
        if per_10ms == 0 { return 1; }              // uncalibrated: floor to one quantum
        let c = per_10ms.saturating_mul(ms) / 10;
        if c == 0 { 1 } else { c }
    }

    /// Read the hardware real-time clock (wall-clock date/time) via the kernel.
    ///
    /// Ambient - the time of day is task-neutral hardware info, like the TSC.
    /// Wraps InspectKernel query 11; the kernel returns the fields packed into a
    /// `u64` (see `kernel/src/arch/x86_64/rtc.rs`), which this unpacks.
    pub fn datetime(&self) -> Datetime {
        // SAFETY: syscall(13) = InspectKernel; query_id=11 = packed RTC datetime.
        let p = unsafe { raw_syscall(13, 11, 0, 0) } as u64;
        Self::unpack_datetime(p)
    }

    /// Deglitched monotonic "now" in epoch seconds (kernel query 17). Unlike `datetime().epoch_secs()`
    /// (the raw RTC, query 11 - a CMOS misread on an in-range year slips through and reads years off), this
    /// drops backward / huge-forward glitches. Use it for time-DELTA deadlines and pacing, NOT for display.
    pub fn epoch_secs_monotonic(&self) -> i64 {
        // SAFETY: syscall(13) = InspectKernel; query 17 = deglitched monotonic epoch seconds.
        unsafe { raw_syscall(13, 17, 0, 0) }
    }

    // `set_wall_clock`, `set_clock_floor` and `clock_synced_secs_ago` were REMOVED with the kernel's
    // wall clock (clock slice 3). They called syscall 50 and query 21, which no longer mean what they
    // meant: 50 is deleted, and 21 was REUSED for `com2_byte`, so the stale reader did not merely fail -
    // it popped a byte off the operator channel. Ask the `time` service; it owns the clock now.
    // (2026-10-09) 50 has since been REUSED too: it is `CallDeadline`. Neither number is free.




    /// A hardware-random u32 from the SoC RNG (the BCM2835 RNG on the Pi 2), or None if this build exposes
    /// no hardware RNG. Ungated (entropy confers no authority). The `random` shell utility consumes it.
    pub fn hw_random(&self) -> Option<u32> {
        // SAFETY: syscall(13) = InspectKernel; query 19 = a hardware-random u32 (-1 if unavailable).
        let r = unsafe { raw_syscall(13, 19, 0, 0) };
        if r < 0 { None } else { Some(r as u32) }
    }

    /// Drive a SoC GPIO pin (the shell `gpio` command; the Pi 2's BCM2835). `op`: 0 input / 1 output /
    /// 2 high / 3 low / 4 read. Returns the level (0/1) for a read, 0 on success, -1 on a bad pin /
    /// unsupported arch. Gated by the GPIO_DEVICE cap (only the shell holds it).
    pub fn gpio(&self, op: u32, pin: u32) -> i64 {
        // SAFETY: syscall(45) = Gpio; the kernel validates the GPIO_DEVICE cap and bounds op/pin.
        unsafe { raw_syscall(45, op as u64, pin as u64, 0) }
    }

    /// Decode the packed RTC `u64` (the layout shared by query 11 / 12) into a `Datetime`.
    fn unpack_datetime(p: u64) -> Datetime {
        Datetime {
            second: (p & 0x3F) as u8,
            minute: ((p >> 6) & 0x3F) as u8,
            hour: ((p >> 12) & 0x1F) as u8,
            day: ((p >> 17) & 0x1F) as u8,
            month: ((p >> 22) & 0x0F) as u8,
            year: ((p >> 26) & 0xFFF) as u16,
        }
    }

    /// Query the kernel task stat for scheduler slot `slot` (syscall 16).
    ///
    /// Returns a best-effort snapshot. If `slot` is out of range or the task
    /// is dead, `valid` will be false - and also when the caller lacks the
    /// INTROSPECT cap the kernel requires, which this cannot tell apart.
    pub fn task_stat(&self, slot: u32) -> TaskStat {
        // The kernel's `STAT_SIZE`: 176 bytes since names grew to 128 (`handle_task_stat`).
        let mut buf = [0u8; 176];
        // SAFETY: syscall(16) = TaskStat; buf is a local array on the user stack.
        let ret = unsafe {
            raw_syscall(16, slot as u64, buf.as_mut_ptr() as u64, 176)
        };
        if ret != 0 {
            return TaskStat {
                valid: false, state: 0, core: 0,
                mem_used: 0, mem_limit: 0, name_len: 0, name: [0u8; NAME_MAX],
                restart_count: 0, queue_depth: 0, run_ticks: 0, uptime_secs: 0,
            };
        }
        let valid       = buf[0] != 0;
        let state       = buf[1];
        let core        = buf[2];
        let name_len    = u32::from_le_bytes([buf[4], buf[5], buf[6], buf[7]]) as usize;
        let mem_used    = u64::from_le_bytes([buf[8],  buf[9],  buf[10], buf[11],
                                              buf[12], buf[13], buf[14], buf[15]]);
        let mem_limit   = u64::from_le_bytes([buf[16], buf[17], buf[18], buf[19],
                                              buf[20], buf[21], buf[22], buf[23]]);
        let mut name = [0u8; NAME_MAX];
        let copy_len = name_len.min(NAME_MAX);
        name[..copy_len].copy_from_slice(&buf[24..24 + copy_len]);
        let restart_count = u64::from_le_bytes([buf[152], buf[153], buf[154], buf[155],
                                                buf[156], buf[157], buf[158], buf[159]]);
        let queue_depth = buf[3];
        let run_ticks   = u64::from_le_bytes([buf[160], buf[161], buf[162], buf[163],
                                              buf[164], buf[165], buf[166], buf[167]]);
        let uptime_secs = u64::from_le_bytes([buf[168], buf[169], buf[170], buf[171],
                                              buf[172], buf[173], buf[174], buf[175]]);
        TaskStat { valid, state, core, mem_used, mem_limit, name_len: copy_len, name,
                   restart_count, queue_depth, run_ticks, uptime_secs }
    }

    /// List the capabilities held by the task in `slot`, into `out`. Returns the
    /// number of entries written (capped at `out.len()` and 64). Requires the
    /// INTROSPECT cap; a refusal returns 0, the same as a task holding nothing.
    /// Best-effort snapshot - see [`task_stat`](Self::task_stat).
    pub fn task_caps(&self, slot: u32, out: &mut [CapInfo]) -> usize {
        const ENTRY: usize = 16;
        const MAX: usize = 64;
        let want = out.len().min(MAX);
        if want == 0 { return 0; }
        let mut buf = [0u8; ENTRY * MAX];
        // SAFETY: syscall(28) = TaskCaps; buf is a local array on the user stack.
        let ret = unsafe {
            raw_syscall(28, slot as u64, buf.as_mut_ptr() as u64, (want * ENTRY) as u64)
        };
        if ret <= 0 { return 0; }
        let count = (ret as usize).min(want);
        for i in 0..count {
            let o = i * ENTRY;
            out[i].resource_id = u64::from_le_bytes([
                buf[o], buf[o + 1], buf[o + 2], buf[o + 3],
                buf[o + 4], buf[o + 5], buf[o + 6], buf[o + 7],
            ]);
            out[i].rights = buf[o + 8];
        }
        count
    }

    /// Send a message via an explicit cap handle (blocking).
    ///
    /// Used by benchmark probes that dynamically acquire send caps rather than
    /// using named peer slots, avoiding repeated name-lookup overhead.
    pub fn send_by_handle(&self, handle: CapHandle, msg: &Message) -> Result<(), IpcError> {
        crate::ipc::send(handle, msg)
    }

    /// Query the rights bitfield of the cap at `handle`.
    ///
    /// Returns the rights byte as a u64, or `None` if the slot is empty.
    /// Used by property test P3 to verify rights do not widen on transfer (§7.3).
    pub fn query_cap_rights(&self, handle: CapHandle) -> Option<u64> {
        // SAFETY: syscall(14) = QueryCapRights; slot is a cap table index.
        let ret = unsafe { raw_syscall(14, handle.0 as u64, 0, 0) };
        if ret < 0 { None } else { Some(ret as u64) }
    }

    /// Remove the cap at `handle` from this task's cap table.
    ///
    /// Idempotent: removing an already-empty slot is a no-op.
    pub fn remove_cap(&self, handle: CapHandle) {
        // SAFETY: syscall(15) = RemoveCap; slot is a cap table index.
        unsafe { raw_syscall(15, handle.0 as u64, 0, 0); }
    }

    /// Send a message to `endpoint` with cap `grant` embedded.
    ///
    /// Unlike `send_with_cap` (which looks up a peer by name), this takes
    /// explicit handles - used by P3 where the endpoint and grant slot are
    /// the same (self-referential cap transfer).  On success the grant cap is
    /// removed from the caller's table (§7.6).
    pub fn send_with_cap_by_handle(
        &self,
        endpoint: CapHandle,
        grant:    CapHandle,
        msg:      &crate::ipc::Message,
    ) -> Result<(), crate::ipc::IpcError> {
        let packed  = ((grant.0 as u64) << 16) | (endpoint.0 as u64);
        let payload = msg.payload_bytes();
        // SAFETY: syscall(11) = SendWithCap; packed and payload are from user space.
        let ret = unsafe {
            raw_syscall(11, packed, payload.as_ptr() as u64, payload.len() as u64)
        };
        if ret == 0 { Ok(()) } else { Err(crate::ipc::i64_to_ipc_error(ret)) }
    }

    /// Mint a **delegated resource** (§7.10, P2 file-as-capability): the kernel allocates a
    /// fresh resource owned by this service and a cap for it carrying `rights` (use the
    /// `RIGHT_*` bits), returning `(resource_id, cap)`. Requires the `RESOURCE_MINT` authority
    /// (held by `fs`). The `resource_id` is what the kernel badges into a later
    /// `resource_invoke`, so this service knows which resource a client is acting on (e.g. `fs`
    /// maps it → file). `None` if the authority is missing, the band is full, or the cap table
    /// is full. Syscall 30 = `ResourceMint`.
    pub fn resource_mint(&self, rights: u8) -> Option<(u64, CapHandle)> {
        let mut id: u64 = 0;
        // SAFETY: syscall(30) = ResourceMint; arg1 points at our own `id` for the kernel to
        // fill via write_user_bytes (validated kernel-side). Return is the cap slot or <0.
        let ret = unsafe { raw_syscall(30, rights as u64, &mut id as *mut u64 as u64, 0) };
        if ret < 0 { None } else { Some((id, CapHandle(ret as u32))) }
    }

    /// Use a delegated resource cap (§7.10) - the "send" of file-as-capability. The kernel
    /// validates the cap holds `right` (a read needs `RIGHT_READ`, a write needs `RIGHT_WRITE`;
    /// a cap lacking it fails `CapInsufficientRights` - the non-escalation check), then routes
    /// `msg` to the owning service badged with the resource id + the validated right, embedding
    /// `reply` (a SEND|GRANT cap) so the owner can reply. `Ok(())` on delivery. Syscall 31.
    pub fn resource_invoke(&self, file: CapHandle, right: u8, reply: CapHandle, msg: &Message)
        -> Result<(), IpcError> {
        // Packed into ONE 32-BIT word: file[0..12] | reply[12..24] | right[24..32].
        //
        // A syscall argument is a single register, and on a 32-bit target (arm32's r1/r2/r3) that
        // register is 32 bits - so anything placed above bit 31 is silently truncated on the way in.
        // The original layout put `right` at bit 32, which arrived as 0 on ARM: the kernel then had
        // no right to validate, `fs` received a badge of 0, and every real operation failed `op <=
        // right` while a read-only cap invoked declaring WRITE sailed past the kernel check that was
        // supposed to stop it. Three symptoms, one truncated field.
        //
        // 12 bits per slot is 4095 against a MAX_CAPS_PER_TASK of 64 - room to grow by 60x - and
        // `right` is a u8, so the whole thing lands in 24 bits with 8 to spare. This is the A-U1
        // rule from arch/arm/CLAUDE.md: on a 32-bit ABI, a syscall argument that does not fit in one
        // register must be narrowed at the wrapper, never assumed to survive.
        // REJECT a handle that does not fit its field instead of letting it corrupt the next one.
        //
        // Each slot gets 12 bits. A handle above 4095 does not simply get truncated - its high bits
        // land in the NEXT field. The `fcap` forged-handle check passes 60000 (0xEA60) and the kernel
        // logged `file_slot=2656 reply_slot=14`: the 0xE spilled out of `file` and rewrote `reply`, so
        // a bad file handle silently redirected the REPLY to whatever cap happened to sit in slot 14.
        // The invocation was rejected for the right reason by luck, not by design.
        //
        // A slot above 4095 cannot be valid anyway (MAX_CAPS_PER_TASK is 64), so this rejects with the
        // same error the kernel gives for a slot it does not hold - the caller sees no difference,
        // and no neighbouring field is ever silently rewritten.
        const SLOT_MAX: u32 = 0xFFF;
        if file.0 > SLOT_MAX || reply.0 > SLOT_MAX {
            return Err(crate::ipc::i64_to_ipc_error(-2)); // -2 = capability not held
        }
        let packed = ((right as u64) << 24) | ((reply.0 as u64) << 12) | (file.0 as u64);

        let payload = msg.payload_bytes();
        // SAFETY: syscall(31) = ResourceInvoke; packed + payload are user values the kernel
        // validates (cap slots, rights, generation, and the message bounds) before acting.
        let ret = unsafe {
            raw_syscall(31, packed, payload.as_ptr() as u64, payload.len() as u64)
        };
        if ret == 0 { Ok(()) } else { Err(crate::ipc::i64_to_ipc_error(ret)) }
    }

    /// Read (and clear) the delegated-resource badge of the message just `recv`'d (§7.10). A
    /// service that owns delegated resources (e.g. `fs`) calls this right after `recv`: `Some((
    /// resource_id, right))` means the message was a **kernel-validated** invocation of a real
    /// cap on `resource_id` with `right` already checked (the owner enforces op ≤ `right`); `None`
    /// means an ordinary message (no badge - handle it on the name-addressed path). The badge
    /// cannot be forged over a plain `send`, so its presence is trustworthy. Syscall 33.
    pub fn last_recv_badge(&self) -> Option<(u64, u8)> {
        // SAFETY: syscall(33) = LastRecvBadge; reads+clears this task's stored badge.
        let packed = unsafe { raw_syscall(33, 0, 0, 0) } as u64;
        if packed == 0 {
            None
        } else {
            Some((packed & 0xFFFF_FFFF, ((packed >> 32) & 0xFF) as u8))
        }
    }

    /// Revoke a delegated resource this service owns (§7.10): bumps its generation so every
    /// outstanding cap to it goes stale (clients see `CapRevoked`/`EndpointDead` on next use).
    /// Owner-gated by the kernel (ownership is the check). `true` on success. Syscall 32.
    #[must_use = "the capability is STILL VALID if this is false"]
    pub fn resource_revoke(&self, resource_id: u64) -> bool {
        // SAFETY: syscall(32) = ResourceRevoke; the kernel checks this task owns the resource.
        unsafe { raw_syscall(32, resource_id, 0, 0) == 0 }
    }

    /// Inject one byte into the console input ring (syscall 20). Only effective
    /// for an input-driver service holding a CONSOLE_PUSH cap (the USB keyboard
    /// driver, §12); the byte reaches the shell exactly like a serial keystroke.
    /// No-op for services without the cap.
    pub fn console_push(&self, byte: u8) {
        let slot = Self::ctx().console_push_slot;
        if slot == u32::MAX {
            return;
        }
        // SAFETY: syscall(20) = ConsolePush; slot is the kernel-written cap index.
        let _ = unsafe { raw_syscall(20, slot as u64, byte as u64, 0) };
    }

    /// Block until one byte is available on COM1 console input (syscall 17).
    ///
    /// Returns the byte value. Only usable by a service spawned with a CONSOLE_READ cap
    /// (`SPAWN_FLAG_REQ_CONSOLE` in its spawn request: the shell, `chaos`, `observe-live`). A kernel error
    /// is returned as byte 0, and a missing cap or corrupt context parks the caller in a busy loop.
    pub fn console_read(&self) -> u8 {
        let data = Self::ctx();
        // A wrong magic means the kernel handed us a corrupt ServiceContext - `self.log()` reads the
        // same corrupt ctx, so it cannot be trusted to report; parking is the service-level analog of
        // the kernel's halt-on-corrupt-state (§6.2). The slot guard below CAN speak (audit L8).
        if data.magic != SERVICE_CTX_MAGIC { loop {} }
        let slot = data.console_read_slot;
        if slot == u32::MAX {
            // No CONSOLE_READ cap: a caller that isn't the shell reached a shell-only syscall. Say so
            // LOUDLY (inv12) rather than wedge silently, then park - the caller's contract is wrong.
            self.log("sdk: console_read called without a CONSOLE_READ cap - parking (contract error)");
            loop {}
        }
        // SAFETY: syscall(17) = ConsoleRead; slot is kernel-written cap index.
        let ret = unsafe { raw_syscall(17, slot as u64, 0, 0) };
        if ret >= 0 { ret as u8 } else { 0 }
    }

    /// Non-blocking console read (syscall 24). Returns `Some(byte)` if a keystroke
    /// is waiting, `None` if the ring is empty. A foreground full-screen app polls
    /// this for `q`-to-quit between repaints instead of blocking in `console_read`.
    /// Requires the CONSOLE_READ cap (`SPAWN_FLAG_REQ_CONSOLE` in the spawn request).
    pub fn try_console_read(&self) -> Option<u8> {
        let data = Self::ctx();
        if data.magic != SERVICE_CTX_MAGIC { return None; }
        let slot = data.console_read_slot;
        if slot == u32::MAX { return None; }
        // SAFETY: syscall(24) = TryConsoleRead; slot is kernel-written cap index.
        // Returns 0..=255 (byte), 256 (empty), or negative (cap error).
        let ret = unsafe { raw_syscall(24, slot as u64, 0, 0) };
        if (0..=255).contains(&ret) { Some(ret as u8) } else { None }
    }

    /// Enable (`true`) or disable (`false`) console keystroke echo (syscall 25).
    /// A foreground full-screen app disables echo while it owns the screen - so
    /// its raw key polls do not smear its frame - and re-enables it on exit.
    /// Requires the CONSOLE_READ cap.
    pub fn console_echo(&self, on: bool) {
        let data = Self::ctx();
        if data.magic != SERVICE_CTX_MAGIC { return; }
        let slot = data.console_read_slot;
        if slot == u32::MAX { return; }
        // SAFETY: syscall(25) = ConsoleEcho; slot is kernel-written cap index.
        let _ = unsafe { raw_syscall(25, slot as u64, on as u64, 0) };
    }

    /// End boot-log mirroring to the framebuffer and clear the TV (syscall 26).
    /// The shell calls this once, on the first keystroke, so the user sees the
    /// boot sequence on the display and then gets a clean interactive console.
    /// Requires the CONSOLE_READ cap.
    pub fn console_boot_complete(&self) {
        let data = Self::ctx();
        if data.magic != SERVICE_CTX_MAGIC { return; }
        let slot = data.console_read_slot;
        if slot == u32::MAX { return; }
        // SAFETY: syscall(26) = ConsoleBootComplete; slot is kernel-written cap index.
        let _ = unsafe { raw_syscall(26, slot as u64, 0, 0) };
    }

    /// Whether THIS task currently owns (or shares, when unclaimed) console input - i.e. its console
    /// reads return bytes. False when another task holds the foreground (syscall 40, e.g. `chaos`): a
    /// backgrounded task should then stay quiet (not draw, not read) and redraw its prompt only when
    /// this returns true again. InspectKernel query 13 (ungated, caller-specific).
    pub fn is_console_foreground(&self) -> bool {
        // SAFETY: syscall(13) = InspectKernel; query 13 = is-foreground for the caller.
        unsafe { raw_syscall(13, 13, 0, 0) != 0 }
    }

    /// Claim exclusive console input (syscall 40, op = 1): after this, only THIS task's
    /// `try_console_read` returns bytes; every other task reads empty. The `chaos` service
    /// claims it for the duration of a run so a resurrected shell cannot swallow its
    /// `q`-to-quit. Pair with `release_console_foreground` on exit, after ensuring a live
    /// shell exists to hand the keyboard back to. Requires the CONSOLE_READ cap.
    pub fn claim_console_foreground(&self) {
        let data = Self::ctx();
        if data.magic != SERVICE_CTX_MAGIC { return; }
        let slot = data.console_read_slot;
        if slot == u32::MAX { return; }
        // SAFETY: syscall(40) = ConsoleForeground; op 1 = claim; slot is kernel-written cap index.
        let _ = unsafe { raw_syscall(40, slot as u64, 1, 0) };
    }

    /// Release exclusive console input (syscall 40, op = 0) back to the unclaimed state, so
    /// any CONSOLE_READ holder (the shell) reads normally again. Idempotent.
    pub fn release_console_foreground(&self) {
        let data = Self::ctx();
        if data.magic != SERVICE_CTX_MAGIC { return; }
        let slot = data.console_read_slot;
        if slot == u32::MAX { return; }
        // SAFETY: syscall(40) = ConsoleForeground; op 0 = release; slot is kernel-written cap index.
        let _ = unsafe { raw_syscall(40, slot as u64, 0, 0) };
    }

    /// Return the core this service was spawned on.
    pub fn core_id(&self) -> u32 { Self::ctx().core_id }

    /// Safe MMIO handle for this service's xHCI controller, if one was granted
    /// (§12). The kernel mapped the controller's BAR into this driver's address
    /// space; the returned [`crate::Mmio`] reads/writes the uncached device
    /// registers directly. `None` for non-driver services.
    pub fn xhci_mmio(&self) -> Option<crate::mmio::Mmio> {
        let va = Self::ctx().xhci_mmio_va;
        if va == 0 {
            None
        } else {
            Some(crate::mmio::Mmio::new(va as *mut u8, Self::ctx().xhci_mmio_len as usize))
        }
    }

    /// Safe MMIO handle for this service's EHCI controller, if one was granted
    /// (§12). Reads the same kernel-mapped controller-BAR window as
    /// [`xhci_mmio`](Self::xhci_mmio) - a driver service holds exactly one
    /// controller, so the field is shared and unambiguous. `None` for non-drivers.
    pub fn ehci_mmio(&self) -> Option<crate::mmio::Mmio> {
        let va = Self::ctx().xhci_mmio_va;
        if va == 0 {
            None
        } else {
            Some(crate::mmio::Mmio::new(va as *mut u8, Self::ctx().xhci_mmio_len as usize))
        }
    }

    /// The framebuffer the kernel granted this service, or `None` if it holds no grant.
    ///
    /// Held only by `console`, which renders the terminal into it (`docs/console-service.md` §9). The
    /// grant carries PIXEL geometry only - character rows and columns are the terminal's own business,
    /// and the kernel deliberately does not compute or publish them (Commandment III).
    pub fn framebuffer(&self) -> Option<crate::mmio::Framebuffer> {
        let d = Self::ctx();
        if d.fb_va == 0 || d.fb_len == 0 {
            return None;
        }
        let sh = d.fb_shifts;
        Some(crate::mmio::Framebuffer::new(
            d.fb_va as *mut u8,
            d.fb_len as usize,
            d.fb_pitch as usize,
            d.fb_bpp as usize,
            d.fb_width as usize,
            d.fb_height as usize,
            sh & 0xFF,
            (sh >> 8) & 0xFF,
            (sh >> 16) & 0xFF,
        ))
    }

    /// Safe MMIO handle to this service's device register window, if one was
    /// granted (§12) - the neutrally-named accessor for non-USB drivers (e.g. the
    /// AHCI `block-driver`, which maps its HBA ABAR here). Same kernel-mapped
    /// window as [`xhci_mmio`](Self::xhci_mmio). `None` for non-driver services.
    pub fn mmio(&self) -> Option<crate::mmio::Mmio> {
        let va = Self::ctx().xhci_mmio_va;
        if va == 0 {
            None
        } else {
            Some(crate::mmio::Mmio::new(va as *mut u8, Self::ctx().xhci_mmio_len as usize))
        }
    }

    /// Safe handle to this service's DMA arena, if one was granted (§12). The
    /// kernel mapped a physically-contiguous region into this driver; the
    /// returned [`crate::Dma`] gives the CPU view (read/write) and the physical
    /// base to program into the controller. `None` for non-driver services.
    pub fn dma_region(&self) -> Option<crate::dma::Dma> {
        let d = Self::ctx();
        if d.xhci_dma_va == 0 {
            None
        } else {
            Some(crate::dma::Dma::new(
                d.xhci_dma_va as *mut u8,
                d.xhci_dma_phys,
                d.xhci_dma_len as usize,
            ))
        }
    }


    /// The interrupt vector the kernel granted this service, if any.
    ///
    /// A driver must NOT hardcode this. It used to be safe to: every vector was a kernel constant, so
    /// both sides knew the number statically. The MSI pool allocates one at spawn instead, and a
    /// driver holding a stale copy does not fail loudly - it keeps working (the interrupt still wakes
    /// it) while reporting every interrupt as an unknown message. That is exactly what `services/xhci`
    /// did on the T630: `0 MSI, 7799 msg`.
    ///
    /// Use it to recognise an interrupt notification: the kernel's wake payload is the vector number.
    ///
    /// ```ignore
    /// let irq = ctx.irq_vector();
    /// let is_irq = irq.is_some_and(|v| msg.payload_bytes() == [v]);
    /// ```
    ///
    /// `None` means no interrupt was granted - a non-driver service, or a driver whose device offered
    /// neither MSI nor MSI-X. A driver that needs interrupts should say so rather than poll silently.
    pub fn irq_vector(&self) -> Option<u8> {
        let d = Self::ctx();
        if d.irq_count == 0 { None } else { Some(d.irqs[0]) }
    }

    /// Every interrupt vector granted to this service (at most four).
    ///
    /// [`Self::irq_vector`] is the common case; this is for a driver routed more than one line.
    pub fn irq_vectors(&self) -> &'static [u8] {
        let d = Self::ctx();
        let n = core::cmp::min(d.irq_count as usize, d.irqs.len());
        &d.irqs[..n]
    }

    /// Read one 32-bit PCI configuration register. Requires the `PCI_CFG` privilege.
    ///
    /// `sel` is a bus/device/function selector in the platform's own encoding, which is the CALLER'S
    /// knowledge - the kernel performs the access and enforces which registers may be reached, but
    /// never interprets what the selector names. `offset` is the register within that function.
    ///
    /// ONE call, not a select-then-read pair, because the underlying hardware register pair is
    /// stateful: split across two calls, two callers would each read whichever register the other
    /// selected, and the KERNEL uses the same pair on its spawn and kill paths. Atomic here, that
    /// window does not exist - and this interface cannot write anything at all.
    ///
    /// `None` if the capability is missing, or the access is one this machine will not admit (a bus
    /// its bridge does not forward, say - reading past it is an abort on some hardware rather than a
    /// harmless all-ones). Both are reported by the kernel too (invariant 12), so a caller need not
    /// guess which it was. Note the distinction from `Some(0xFFFF_FFFF)`, which is a SUCCESSFUL read
    /// of an absent device - the bus floats high, and that is data, not an error.
    pub fn pci_cfg_read(&self, sel: u32, offset: u16) -> Option<u32> {
        // SAFETY: a plain syscall; the kernel validates the capability and the access before any I/O.
        let r = unsafe { crate::syscall::raw_syscall(53, sel as u64, offset as u64, 0) };
        if r < 0 { None } else { Some(r as u32) }
    }

    /// Cut (`false`) or restore (`true`) the power of the device this service was granted a fixed
    /// peripheral window to (`DevicePower`, syscall 54). Needs `DEVICE_POWER`, which the kernel mints
    /// with the window wherever the machine can power the device; everywhere else, and for every
    /// service without such a window, this returns `false` and the kernel says why on the serial log.
    /// How long to hold the device off and how long to wait after restoring it are the caller's to know
    /// - they are properties of the device, not of the kernel (`docs/wifi.md` 47).
    pub fn device_power(&self, on: bool) -> bool {
        // SAFETY: a plain syscall; the kernel validates the capability before touching any pin.
        let r = unsafe { crate::syscall::raw_syscall(54, if on { 1 } else { 0 }, 0, 0) };
        r == 0
    }

    /// Set the Arm cores to the platform's maximum (`true`) or minimum (`false`) clock, and return the rate
    /// they read back in Hz (`CpuClock`, syscall 55). Needs `CPU_CLOCK`, which only the `power` service is
    /// spawned with; everyone else asks `power` for a lease over IPC. `None` is a refusal or a machine with
    /// no control over its clock, and the kernel says which on the serial log.
    pub fn cpu_clock(&self, max: bool) -> Option<u32> {
        // SAFETY: a plain syscall; the kernel validates the capability before asking the firmware.
        let r = unsafe { crate::syscall::raw_syscall(55, if max { 1 } else { 0 }, 0, 0) };
        if r > 0 { Some(r as u32) } else { None }
    }

    /// Turn the machine off NOW (`PowerOff`, syscall 56, op 0). Needs `POWER_OFF`, which the shell holds for the
    /// operator. With it, this does not return: the machine is off, or - where the hardware cannot cut its own
    /// power - every core is halted and the kernel's last line says it is safe to switch off. It returns only
    /// when refused, with the kernel's code; the kernel says why on the serial log.
    pub fn power_off(&self) -> i64 {
        self.power_off_call(0, 0)
    }

    /// Set the machine's one pending power-off `secs` from now (1 ..= 24 h), held by the KERNEL so it survives
    /// every service (`docs/shutdown.md` 6). `0`: this is now the pending one. `1`: one due sooner was kept
    /// (earliest wins). `2`: a reboot is pending, and is kept. `-1`: out of range. `-2`: refused (no `POWER_OFF`).
    pub fn power_off_in(&self, secs: u64) -> i64 {
        self.power_off_call(1, secs)
    }

    /// Cancel the pending power-off. `0`: there was one. `1`: none is pending. `2`: the pending one is a reboot,
    /// and is kept. `-2`: refused.
    pub fn power_off_cancel(&self) -> i64 {
        self.power_off_call(2, 0)
    }

    /// Set the machine to reboot `secs` from now (1 ..= 24 h), on the same one kernel deadline as a power-off
    /// (`PowerOff` op 3). Needs `REBOOT`. The answers are `power_off_in`'s, with `2` meaning a power-off is
    /// pending. A reboot NOW is `reboot`.
    pub fn reboot_in(&self, secs: u64) -> i64 {
        self.power_off_call(3, secs)
    }

    /// Cancel the pending reboot (`PowerOff` op 4). The answers are `power_off_cancel`'s, with `2` meaning the
    /// pending one is a power-off.
    pub fn reboot_cancel(&self) -> i64 {
        self.power_off_call(4, 0)
    }

    fn power_off_call(&self, op: u64, secs: u64) -> i64 {
        // SAFETY: a plain syscall with two integers; the kernel validates the capability before acting.
        unsafe { crate::syscall::raw_syscall(56, op, secs, 0) }
    }

    /// The machine's pending power action (InspectKernel query 28): whole seconds until it, rounded up, and
    /// whether it is a reboot - or `None` when none is pending. Readable by every task, deliberately: a pending
    /// shutdown or reboot must never be hidden.
    pub fn power_pending(&self) -> Option<PowerPending> {
        // SAFETY: syscall(13) = InspectKernel; query 28 takes no pointer and is ungated.
        let r = unsafe { raw_syscall(13, 28, 0, 0) };
        if r >= 0 { Some(PowerPending { secs: r as u64 / 2, reboot: r & 1 == 1 }) } else { None }
    }

    /// Whole seconds until the pending power-off, or `None` when none is pending - a pending reboot is not one.
    pub fn power_off_pending(&self) -> Option<u64> {
        self.power_pending().filter(|p| !p.reboot).map(|p| p.secs)
    }

    /// Whole seconds until the pending reboot, or `None` when none is pending - a pending power-off is not one.
    pub fn reboot_pending(&self) -> Option<u64> {
        self.power_pending().filter(|p| p.reboot).map(|p| p.secs)
    }

    /// Allocate `size` bytes of read/write memory within this task's budget.
    ///
    /// Returns the virtual address of the mapping on success, or `AllocError`
    /// if the allocation would exceed the contract memory limit (AllocDenied)
    /// or physical memory is exhausted.
    pub fn alloc_mem(&self, size: usize) -> Result<u64, AllocError> {
        // SAFETY: syscall(6) = AllocMem; no user pointers passed.
        let ret = unsafe { raw_syscall(6, size as u64, 0, 0) };
        if ret >= 0 {
            Ok(ret as u64)
        } else if ret == -11 {
            Err(AllocError::Denied)
        } else {
            Err(AllocError::Failed)
        }
    }

    // `abort()` (the kernel `Abort`/9 syscall) was removed: it let any task panic the kernel, an
    // ungated §3.1 hole, and its only caller (`init`) is gone (Phase 5). A service that hits a fatal
    // error now simply dies (and is restarted by the supervisor) rather than aborting the kernel.

    /// Trigger a hardware reset via the kernel reboot syscall (18). Does not return.
    ///
    /// The kernel prints `reboot: hardware reset` to serial before resetting, so the operator sees
    /// confirmation before the line goes silent. Requires REBOOT (held by the shell); a refusal is
    /// logged and the caller then yields forever.
    pub fn reboot(&self) -> ! {
        // SAFETY: syscall(18) = Reboot; no arguments.
        let rc = unsafe { raw_syscall(18, 0, 0, 0) };
        // A REFUSED reboot must not look like a successful one.
        //
        // The kernel gates this on the REBOOT capability (§3.1) and returns `CapNotHeld` to anyone
        // without it. This used to fall into a bare `loop {}`, so a caller that lacked the cap simply
        // stopped - no message, no reset, a frozen prompt. "Reboot does nothing" is exactly how that
        // reads from the outside, and it is indistinguishable from a reset that failed in hardware.
        //
        // Say which it was. The kernel's own reset path is bounded and reports if the SoC does not come
        // down; this covers the other half, where the syscall never got that far. Invariant 12: failures
        // are loud, never silent.
        self.log_fmt(format_args!(
            "reboot REFUSED by the kernel (rc {}) - this service does not hold the REBOOT capability", rc));
        loop {
            self.yield_cpu();
        }
    }

    /// Attempt a reboot but RETURN the syscall result instead of assuming it never comes back.
    ///
    /// A successful reset does not return; a denial returns a negative error code (CapNotHeld = -2
    /// when the caller lacks the REBOOT capability, §3.1). For tests/probes that must *observe* the
    /// denial without resetting the machine - ordinary rebooters use `reboot()`.
    pub fn try_reboot(&self) -> i64 {
        // SAFETY: syscall(18) = Reboot; no arguments. On success it never returns; on denial it
        // returns the error code, which we hand back to the caller.
        unsafe { raw_syscall(18, 0, 0, 0) }
    }

    /// End this program, with `status`: 0 for success, anything else for a failure it has already said.
    ///
    /// The kernel ends the caller and no one else - it takes no capability - and tells the supervisor how
    /// it ended (`death::EXITED` and the status), which is how a program that FINISHED is told apart from one
    /// that crashed. A service the supervisor keeps running is restarted all the same: exiting is for a
    /// program that does one job (CLAUDE.md 14.5 amendment).
    pub fn exit(&self, status: u32) -> ! {
        // SAFETY: syscall(57) = Exit; one integer, no pointer. It does not return.
        unsafe { raw_syscall(57, status as u64, 0, 0); }
        // Unreachable unless the kernel refused, which it never does for the caller itself: park rather than
        // return into code that believes it has ended.
        self.park()
    }

    /// Advisory yield (§9.3).
    pub fn yield_cpu(&self) {
        // SAFETY: syscall(4) = Yield; always valid from ring-3.
        unsafe { raw_syscall(4, 0, 0, 0); }
    }

    /// Park this task forever: block with no waker. For idle services that have
    /// no further work - far better than `loop { yield_cpu() }`,
    /// which keeps the core busy and prevents it from halting (so it never runs
    /// cool). Nothing wakes a parked task in v1; the loop re-parks defensively.
    pub fn park(&self) -> ! {
        loop {
            // SAFETY: syscall(21) = Park; blocks this task indefinitely.
            unsafe { raw_syscall(21, 0, 0, 0); }
        }
    }

    /// Offer a queryable COPY of one log line to `events`. Never the log itself.
    ///
    /// Called only AFTER `log`'s syscall has already written the kernel ring and serial, so the
    /// authoritative record exists before this runs and does not depend on it in any way. A service
    /// with no `events` cap does nothing here; `events` itself holds no cap to itself, so its own
    /// logging cannot recurse into the sink.
    ///
    /// NO REACQUIRE ON FAILURE, deliberately, unlike `metric` and `trace_emit`. The cap slot is shared
    /// between all three, and those two already refresh it after a sink restart - `count_recv`
    /// publishes every 64 messages, so the slot heals on its own. Reacquiring from inside the LOGGING
    /// path is the one place it could bite: a reacquire that itself wanted to report something would
    /// re-enter here. Dropping a copy costs a line of queryable scrollback and nothing else, because
    /// serial already has it.
    #[inline]
    fn log_copy(&self, text: &[u8]) {
        let slot = crate::trace::sink_slot();
        if slot == u32::MAX {
            return;
        }
        let mut buf = [0u8; 1 + crate::trace::PEER_LEN + crate::trace::LOG_TEXT_MAX];
        let n = crate::trace::encode_log(text, &mut buf);
        let _ = self.try_send_by_handle(CapHandle(slot), &crate::ipc::Message::from_bytes(&buf[..n]));
    }

    /// Log a string via the kernel ring buffer and serial (syscall 5, requires the log_write cap).
    /// Over 256 bytes it is cut on a character boundary and followed by ` [TRUNCATED]`. Without the
    /// cap it does nothing, and a kernel refusal is not reported.
    pub fn log(&self, msg: &str) {
        let data = Self::ctx();
        if data.magic != SERVICE_CTX_MAGIC { return; }
        let slot = data.log_write_slot;
        if slot == u32::MAX { return; }

        let bytes = msg.as_bytes();
        let len   = bytes.len();
        if len == 0 { return; }

        // TOO LONG IS TRUNCATED, NEVER DROPPED.
        //
        // This used to be `if len == 0 || len > 256 { return; }` - a message over the limit was
        // discarded whole, with no truncation, no marker and no error. The reporting channel itself
        // failed silently, which makes every carefully-worded warning in the system conditional on
        // its own length.
        //
        // It was not hypothetical. The two longest literal log lines in the tree are both in `fs`,
        // and both are messages the constitution leans on: the durability-not-attested warning (280
        // bytes), which CLAUDE.md 6.1's backend-conditional TCB claim describes as "`fs` says so once
        // per mount", and the journal-recovery refusal (356 bytes), whose own comment argues it must
        // stay as visible as the fault it reports. Neither has ever been printed on this board, and
        // the durability one is latched before the call, so it never would be.
        //
        // Truncating loses the tail of a sentence. Dropping loses the fact. The marker says which
        // happened, so a reader is never left believing they saw the whole message.
        const LOG_MAX: usize = 256;
        if len > LOG_MAX {
            const MARK: &str = " [TRUNCATED]";
            let keep = LOG_MAX - MARK.len();
            // Cut on a CHARACTER boundary: `as_bytes` is UTF-8 and slicing mid-codepoint would hand
            // the kernel an invalid string. Walk back at most 3 bytes to the start of a codepoint.
            let mut end = keep;
            while end > 0 && (bytes[end] & 0xC0) == 0x80 {
                end -= 1;
            }
            // SAFETY: syscall(5) = Log; both slices are valid, in-bounds and within user space. Sent
            // as two calls rather than staged through a buffer, because a fixed staging buffer here
            // is exactly the kind of hidden limit this change exists to remove.
            unsafe {
                raw_syscall(5, slot as u64, bytes.as_ptr() as u64, end as u64);
                raw_syscall(5, slot as u64, MARK.as_ptr() as u64, MARK.len() as u64);
            }
            self.log_copy(&bytes[..end]);
            return;
        }

        // SAFETY: syscall(5) = Log; bytes is a valid slice within user space.
        unsafe {
            raw_syscall(5, slot as u64, bytes.as_ptr() as u64, len as u64);
        }
        self.log_copy(bytes);
    }

    /// Write a string to serial WITHOUT a trailing newline (syscall 22, requires the
    /// log_write cap). For inline output such as the shell prompt, where `log`'s newline
    /// would push the user's typed echo to the next line. A string longer than the kernel's 256-byte
    /// limit is written in pieces, each ending on a character boundary - it was dropped whole, with
    /// no marker, until 2026-10-10 (`backlog/fixed/80` S5).
    pub fn print(&self, msg: &str) {
        let data = Self::ctx();
        if data.magic != SERVICE_CTX_MAGIC { return; }
        let slot = data.log_write_slot;
        if slot == u32::MAX { return; }
        for piece in Pieces::new(msg, CONSOLE_WRITE_MAX) {
            // SAFETY: syscall(22) = Print; `piece` is a valid slice within user space, at most
            // CONSOLE_WRITE_MAX bytes.
            unsafe {
                raw_syscall(22, slot as u64, piece.as_ptr() as u64, piece.len() as u64);
            }
        }
    }

    /// Log a formatted message, rendered into a 256-byte stack buffer. A message longer than that is
    /// cut on a character boundary and ends `...`, so the cut is visible (`backlog/fixed/80` S5: it was cut
    /// with no marker, and a cut through a character logged `(fmt error)` instead of the message).
    pub fn log_fmt(&self, args: core::fmt::Arguments) {
        let mut buf = [0u8; FMT_MAX];
        let s = render(&mut buf, args);
        if !s.is_empty() {
            self.log(s);
        }
    }

    /// Write a string to the **interactive console** (serial + framebuffer),
    /// WITHOUT a trailing newline (syscall 23, requires log_write cap in Stage 1).
    /// This is the user-facing path: the shell prompt, command results, and
    /// `observe` frames. Unlike `log`/`print` (now serial-only), this also reaches
    /// the framebuffer/TV - the interactive surface (see docs/console-service.md).
    pub fn console_write(&self, msg: &str) {
        let data = Self::ctx();
        if data.magic != SERVICE_CTX_MAGIC { return; }
        let slot = data.log_write_slot;
        if slot == u32::MAX { return; }

        // Longer than the kernel takes in one call: written in pieces on character boundaries, never
        // dropped (`backlog/fixed/80` S5).
        for piece in Pieces::new(msg, CONSOLE_WRITE_MAX) {
            // SAFETY: syscall(23) = ConsoleWrite; `piece` is a valid slice within user space, at
            // most CONSOLE_WRITE_MAX bytes.
            unsafe {
                raw_syscall(23, slot as u64, piece.as_ptr() as u64, piece.len() as u64);
            }
        }
    }

    /// Write a string to the interactive console followed by a newline.
    pub fn console_writeln(&self, msg: &str) {
        self.console_write(msg);
        self.console_write("\n");
    }

    /// Write a formatted message to the interactive console, with **no** trailing
    /// newline (e.g. a pager status line the cursor should park on).
    pub fn console_write_fmt(&self, args: core::fmt::Arguments) {
        let mut buf = [0u8; FMT_MAX];
        self.console_write(render(&mut buf, args));
    }

    /// Write a formatted message to the interactive console, followed by a newline.
    pub fn console_writeln_fmt(&self, args: core::fmt::Arguments) {
        let mut buf = [0u8; FMT_MAX];
        self.console_write(render(&mut buf, args));
        self.console_write("\n");
    }

    /// Write one console line. When `clear_eol` is true the line ends with
    /// `ESC[K` (erase to end of line) before the newline - so a full-screen app
    /// repainting in place (cursor homed each frame) overwrites a previous,
    /// longer line without leaving stale characters, and without a full-screen
    /// clear (no flicker). When false, behaves exactly like `console_writeln`.
    pub fn console_line(&self, clear_eol: bool, msg: &str) {
        self.console_write(msg);
        self.console_write(if clear_eol { "\x1b[K\n" } else { "\n" });
    }

    /// Formatted variant of [`Self::console_line`].
    pub fn console_line_fmt(&self, clear_eol: bool, args: core::fmt::Arguments) {
        let mut buf = [0u8; FMT_MAX];
        self.console_write(render(&mut buf, args));
        self.console_write(if clear_eol { "\x1b[K\n" } else { "\n" });
    }

    /// Spawn a service by name on the kernel-selected core.
    pub fn spawn(&self, name: &str) -> Result<(), crate::Error> {
        self.spawn_on(name, 0xFFFF)
    }

    /// Spawn a service by name on `core` (0xFFFF = kernel round-robin).
    ///
    /// **Asks the SUPERVISOR when this service can reach it**, and the kernel otherwise.
    ///
    /// Step C moves service images out of the kernel (`docs/service-ownership.md`), and a service the
    /// kernel does not hold cannot be spawned by name through the kernel. Seven separate callers did
    /// exactly that - control's RESTART, the supervisor's own respawn, the shell's `spawn`, the
    /// shell's PIPES, `spawncap`, `spawnwired`, and `chaos` - and each was found by a regression
    /// rather than by reading. Routing HERE fixes all of them at once, including the ones not yet
    /// found.
    ///
    /// With no supervisor peer there is nothing to ask: the kernel's by-name spawn (syscall 7) was
    /// retired, because its catalogue holds `supervisor` alone and it could only refuse
    /// (`backlog/fixed/80` K14). So that case is refused here, plainly, rather than sent to a syscall that
    /// would refuse it less clearly.
    pub fn spawn_on(&self, name: &str, core: u32) -> Result<(), crate::Error> {
        if self.find_send_slot("supervisor").is_some() {
            // The supervisor hands back a cap to the new service's endpoint. This path does not keep
            // it, so it is released here - `map(|_| ())` dropped the handle and leaked one table slot per
            // spawn (`backlog/fixed/80`, found during H20).
            return self.spawn_via_supervisor(name, core, &[]).map(|cap| {
                if let Some(c) = cap { self.remove_cap(c); }
            });
        }
        let _ = (name, core);
        Err(crate::Error::NotFound)
    }

    /// Ask the supervisor to spawn `name`, returning the `SEND|GRANT` cap it hands back (`None` if
    /// the service has no recv endpoint, or the cap could not be taken).
    ///
    /// The supervisor is RESTARTABLE (6.2), so a cached cap to it goes stale on every respawn. This
    /// reacquires by name and retries ONCE when the send NEVER LEFT - a stale cap - and never when it
    /// was delivered: not on `Ok(None)`, where the deadline passed, and not on `ReplyDead`, where the
    /// supervisor died holding it. Either may have started the service, and retrying would start it
    /// twice. (`ReplyDead` was retried until 2026-10-10, `backlog/fixed/80` S1.) A full queue is not retried
    /// either: the supervisor is alive and busy, and the caller decides whether to ask again.
    pub fn spawn_via_supervisor(&self, name: &str, core: u32, peers: &[&str])
        -> Result<Option<CapHandle>, crate::Error>
    {
        let mut buf = [0u8; supcmd::MAX];
        let n = supcmd::encode(&mut buf, supcmd::SPAWN, core, name, peers)
            .ok_or(crate::Error::InvalidArgument)?;
        let msg = crate::ipc::Message::from_bytes(&buf[..n]);
        for attempt in 0..2 {
            match self.request_with_reply_call_err("supervisor", &msg, 2) {
                Ok(Some(reply)) => {
                    let ok = reply.payload_bytes().first() == Some(&supcmd::OK);
                    let cap = self.take_pending_cap();
                    if ok { return Ok(cap); }
                    // Answered "no". Reclaim any cap that rode along, or it leaks a table slot.
                    if let Some(c) = cap { self.remove_cap(c); }
                    return Err(crate::Error::InvalidArgument);
                }
                Ok(None) => return Err(crate::Error::InvalidArgument),
                Err(crate::ipc::IpcError::ReplyDead) | Err(crate::ipc::IpcError::QueueFull) =>
                    return Err(crate::Error::InvalidArgument),
                Err(_) if attempt == 0 => { let _ = self.reacquire_by_name("supervisor"); }
                Err(_) => return Err(crate::Error::InvalidArgument),
            }
        }
        Err(crate::Error::InvalidArgument)
    }

    /// Spawn a task from an image THIS service supplies (`SpawnImage`, syscall 52).
    ///
    /// The kernel loads what it is handed instead of looking a name up in its own catalogue - the
    /// step that ends that catalogue (`docs/service-ownership.md`). Requires the SPAWN capability,
    /// exactly as `spawn` does: supplying the image is not authority to start it - AND the caller
    /// must hold `IMAGE_SPAWN`, which only the supervisor has and cannot delegate (CLAUDE.md 14.1).
    /// A name already running comes back as `AlreadyRunning`; every other refusal as `InvalidArgument`,
    /// and the kernel's serial log says which it was.
    ///
    /// `peers` are NUL-joined into a caller-owned buffer; each name still resolves through the
    /// kernel name directory, so this grants nothing a contract's `send_peers` would not.
    /// Returns a `SEND|GRANT` cap to the new service's recv endpoint, or `None` if it has none -
    /// the spawner needs it to record `name -> cap` and re-wire dependents after a restart.
    pub fn spawn_image(&self, req: &mut SpawnRequest, peers_buf: &mut [u8], peers: &[&str])
        -> Result<Option<CapHandle>, crate::Error>
    {
        let data = Self::ctx();
        if data.magic != SERVICE_CTX_MAGIC { return Err(crate::Error::InvalidArgument); }
        let slot = data.spawn_slot;
        if slot == u32::MAX { return Err(crate::Error::Cap(CapError::CapNotHeld)); }

        let mut n = 0usize;
        for (i, p) in peers.iter().enumerate() {
            if i > 0 {
                if n >= peers_buf.len() { return Err(crate::Error::InvalidArgument); }
                peers_buf[n] = 0; n += 1;
            }
            let b = p.as_bytes();
            if n + b.len() > peers_buf.len() { return Err(crate::Error::InvalidArgument); }
            peers_buf[n..n + b.len()].copy_from_slice(b);
            n += b.len();
        }
        req.peers_ptr = if n > 0 { peers_buf.as_ptr() as u64 } else { 0 };
        req.peers_len = n as u32;

        // SAFETY: syscall(52) = SpawnImage; `req` is a live, fully initialised SpawnRequest whose
        // layout matches the kernel's, and the kernel copies it once before reading any field.
        let ret = unsafe {
            raw_syscall(52, req as *const SpawnRequest as u64,
                        core::mem::size_of::<SpawnRequest>() as u64, slot as u64)
        };
        // slot + 1, so 0 means "spawned, no recv endpoint" without colliding with slot 0.
        // -22 is the kernel's SPAWN_ALREADY_RUNNING.
        if ret == -22 { Err(crate::Error::AlreadyRunning) }
        else if ret < 0 { Err(crate::Error::InvalidArgument) }
        else if ret == 0 { Ok(None) }
        else { Ok(Some(CapHandle(ret as u32 - 1))) }
    }


    /// Kill a named service. Requires SERVICE_CONTROL (the kernel refuses with `CapNotHeld`
    /// otherwise); every failure is returned as `InvalidArgument`.
    pub fn kill(&self, name: &str) -> Result<(), crate::Error> {
        let bytes = name.as_bytes();
        // SAFETY: syscall(8) = Kill; bytes is a valid slice within user space.
        let ret = unsafe {
            raw_syscall(8, bytes.as_ptr() as u64, bytes.len() as u64, 0)
        };
        // The kernel says which failure it was, and so does this (`backlog/fixed/80` S4): -2 is the
        // `SERVICE_CONTROL` refusal, anything else is no running service of that name (or a name
        // the kernel could not read). Every one used to come back as `InvalidArgument`.
        match ret {
            0 => Ok(()),
            -2 => Err(crate::Error::Cap(CapError::CapNotHeld)),
            _ => Err(crate::Error::NotFound),
        }
    }

    /// Kill then respawn a service with optional core override (§14.4).
    ///
    /// A service that is not running is simply started - that is `NotFound` from the kill, and the
    /// one kill failure that is passed over. A refusal (no `SERVICE_CONTROL`) is returned, and
    /// nothing is spawned: until 2026-10-10 the kill's result was discarded, so a caller without the
    /// authority to kill went on to spawn a second copy (`backlog/fixed/80` S4, CLAUDE.md 13.6).
    pub fn restart(&self, name: &str, core_override: Option<u32>) -> Result<(), crate::Error> {
        match self.kill(name) {
            Ok(()) | Err(crate::Error::NotFound) => {}
            Err(e) => return Err(e),
        }
        let core = core_override.unwrap_or(0xFFFF);
        self.spawn_on(name, core)
    }

    // ---------------------------------------------------------------------------
    // Private helpers.
    // ---------------------------------------------------------------------------

    /// Find the cap slot for a named send peer.
    ///
    /// Search order: dynamic cache (post-restart reacquisitions), then the
    /// kernel-written ServiceContextData send_peers array.
    fn find_send_slot(&self, peer: &str) -> Option<u32> {
        let bytes = peer.as_bytes();
        let len   = bytes.len();

        // 1. Dynamic cache (updated after EndpointDead + reacquire).
        // SAFETY: single-threaded service process.
        unsafe {
            for entry in SEND_CAP_CACHE.iter() {
                if entry.name_len != 0
                    && entry.name_len as usize == len
                    && &entry.name[..len] == bytes
                {
                    return Some(entry.slot);
                }
            }
        }

        // 2. ServiceContextData send_peers (wired at spawn).
        let data  = Self::ctx();
        let count = (data.send_peer_count as usize).min(MAX_SEND_PEERS);
        for i in 0..count {
            let entry = &data.send_peers[i];
            if entry.slot == u32::MAX { continue; }
            let nlen = entry.name_len as usize;
            if nlen == len && &entry.name[..len] == bytes {
                return Some(entry.slot);
            }
        }

        None
    }
}

// ---------------------------------------------------------------------------
// Stack-based fmt::Write helper for log_fmt.
// ---------------------------------------------------------------------------

struct StackWriter<'a> {
    buf: &'a mut [u8],
    pos: &'a mut usize,
    /// Set when something did not fit, so the result can say it was cut.
    cut: bool,
}

impl<'a> core::fmt::Write for StackWriter<'a> {
    fn write_str(&mut self, s: &str) -> core::fmt::Result {
        let space = self.buf.len().saturating_sub(*self.pos);
        // Never split a character: what is kept is always valid UTF-8.
        let mut n = s.len().min(space);
        while n > 0 && !s.is_char_boundary(n) {
            n -= 1;
        }
        if n < s.len() {
            self.cut = true;
        }
        self.buf[*self.pos .. *self.pos + n].copy_from_slice(&s.as_bytes()[..n]);
        *self.pos += n;
        Ok(())
    }
}

/// The most one formatted line renders to, on the stack.
const FMT_MAX: usize = 256;
/// The most the kernel takes in one `Print` or `ConsoleWrite` call.
const CONSOLE_WRITE_MAX: usize = 256;

/// Render `args` into `buf`. A rendering that did not fit ends `...` in place of its last bytes, on a
/// character boundary, so a reader can see it was cut.
fn render<'b>(buf: &'b mut [u8; FMT_MAX], args: core::fmt::Arguments) -> &'b str {
    let mut pos = 0usize;
    let cut = {
        let mut w = StackWriter { buf: &mut buf[..], pos: &mut pos, cut: false };
        let _ = core::fmt::write(&mut w, args);
        w.cut
    };
    if cut {
        let mut end = FMT_MAX - 3;
        while end > 0 && core::str::from_utf8(&buf[..end]).is_err() {
            end -= 1;
        }
        buf[end..end + 3].copy_from_slice(b"...");
        pos = end + 3;
    }
    core::str::from_utf8(&buf[..pos]).unwrap_or("")
}

/// A string in pieces of at most `max` bytes, each ending on a character boundary.
struct Pieces<'s> {
    rest: &'s str,
    max: usize,
}

impl<'s> Pieces<'s> {
    fn new(s: &'s str, max: usize) -> Self {
        Pieces { rest: s, max }
    }
}

impl<'s> Iterator for Pieces<'s> {
    type Item = &'s [u8];
    fn next(&mut self) -> Option<&'s [u8]> {
        if self.rest.is_empty() {
            return None;
        }
        let mut n = self.rest.len().min(self.max);
        while n > 0 && !self.rest.is_char_boundary(n) {
            n -= 1;
        }
        if n == 0 {
            return None;
        }
        let (head, tail) = self.rest.split_at(n);
        self.rest = tail;
        Some(head.as_bytes())
    }
}

/// The seconds a `CallDeadline` is given for a caller's `max_secs`. Never 0, which the kernel reads
/// as NO deadline: a non-positive bound is the shortest real one, not the absence of one.
fn deadline_secs(max_secs: i64) -> u64 {
    if max_secs <= 0 { 1 } else { max_secs as u64 }
}

// ---------------------------------------------------------------------------
// Placeholder types retained for compatibility.
// ---------------------------------------------------------------------------

pub struct ServiceDescriptor;
impl ServiceDescriptor {
    pub fn name(&self) -> &str { todo!() }
}

pub struct BootManifest;
impl BootManifest {
    pub fn services(&self) -> &[ServiceDescriptor] { todo!() }
}