mock-upcloud 0.1.3

//! **The state machine: what the account holds, and how it moves.**
//!
//! Everything the mock answers comes from here. There are no canned replies —
//! a canned reply cannot be *early*, and "early" is the whole difficulty with
//! this provider: a storage is `maintenance` before it is `online`, a server is
//! `maintenance` for a hundred seconds after it is created, and a delete of an
//! appliance with four member volumes is `maintenance` for five minutes. A test
//! fixture that answers `online` on the first read has tested nothing.
//!
//! # Transitions
//!
//! Every object may carry one scheduled [`Transition`]: a state to enter at a
//! virtual millisecond, and optionally a thing to do on arrival (mint the
//! `Resize Backup`, free the addresses, boot the guest). [`Estate::settle`] is
//! the only place a state changes on its own, and it is called at the top of
//! every request — so the state a caller sees is always the state as of the
//! moment it asked, and never a state nothing scheduled.

use crate::clock::{Clock, Timings};
use crate::faults::{Fault, Faults};
use crate::kvm::{GuestEngine, Machine, MachineDisk, VirtualGuest};
use crate::rng::SplitMix64;
use std::collections::BTreeMap;
use std::sync::Arc;

// ── the objects ──────────────────────────────────────────────────────────────

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Label {
    pub key: String,
    pub value: String,
}

/// UpCloud's storage kinds. `Normal` is a disk, `Backup` is what a resize (and a
/// backup rule) leaves behind, `Cdrom` is an installer medium, `Template` is a
/// public image.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum StorageKind {
    Normal,
    Backup,
    Cdrom,
    Template,
}

impl StorageKind {
    pub fn as_str(self) -> &'static str {
        match self {
            StorageKind::Normal => "normal",
            StorageKind::Backup => "backup",
            StorageKind::Cdrom => "cdrom",
            StorageKind::Template => "template",
        }
    }
}

#[derive(Clone, Debug)]
pub struct Storage {
    pub uuid: String,
    pub title: String,
    pub size_gib: u64,
    pub tier: String,
    pub zone: String,
    pub state: String,
    pub kind: StorageKind,
    pub labels: Vec<Label>,
    /// The volume this was made from. On a `Resize Backup` it points at the
    /// boot volume that was resized — which, once that server is deleted, names
    /// a uuid that no longer resolves (behaviour 9, and
    /// [`Fault::OrphanResizeBackup`] makes it so immediately).
    pub origin: Option<String>,
    /// Virtual ms, rendered as `created`. **It IS sent** — measured against the
    /// live account 2026-09-20 on both `GET /1.3/storage/private` and
    /// `GET /1.3/storage/{uuid}`. This crate used to withhold it and claim the
    /// provider did; [`Fault::WithholdCreatedField`] now takes it away only when
    /// armed by name, as a hypothesis.
    pub created_ms: u64,
    /// The direct-upload session, if one was ever opened on this volume.
    ///
    /// **It has its OWN clock, and that is the point.** MEASURED on the live
    /// estate: a 43 485 184-byte ISO was `created 09:36:16Z` and `completed
    /// 09:36:21Z` — five seconds — and then the STORAGE sat in `syncing` for
    /// roughly 100 to 130 seconds more before it turned `online`. The poller
    /// read `syncing` at 65 s and at 97 s while this object already said
    /// `completed`. A mock that turned the storage `online` when the upload
    /// finished would hide the entire cost, which is all of the cost.
    pub import: Option<Import>,
    pub transition: Option<Transition>,
}

/// **The direct-upload session.** `POST /1.3/storage/{uuid}/import` opens it,
/// a `PUT` to its `direct_upload_url` fills it, and its fields are what the
/// ladder verifies against — `sha256sum` in particular, which is compared with
/// the local file's. The digests here are the REAL digests of the REAL bytes
/// that were really PUT; see [`crate::digest`] for why they are not faked.
#[derive(Clone, Debug)]
pub struct Import {
    pub source: String,
    /// `prepared` | `uploading` | `completed` | `failed`.
    pub state: String,
    pub created_ms: u64,
    pub completed_ms: Option<u64>,
    pub client_content_length: u64,
    pub read_bytes: u64,
    pub written_bytes: u64,
    pub md5sum: Option<String>,
    pub sha256sum: Option<String>,
    pub error_code: Option<String>,
    pub error_message: Option<String>,
    pub direct_upload_url: String,
}

/// How a device rides on a server. UpCloud's `address` is `virtio:N` or `ide:B:D`,
/// and a detach names the ADDRESS, never the storage uuid.
#[derive(Clone, Debug)]
pub struct Device {
    pub address: String,
    pub storage: String,
    pub storage_title: String,
    pub storage_size: u64,
    pub kind: &'static str,
    pub boot_disk: bool,
}

#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum BootOrder {
    Cdrom,
    Disk,
    /// **`cdrom,disk` — the appliance's boot order, and it is not a synonym for
    /// `cdrom`.** holger's appliance and twin declare it so a re-image is
    /// "attach the medium and reboot" and an empty drive falls straight through
    /// to the disk. For behaviour 17 (an install killed mid-flight loops
    /// forever) it behaves exactly as `cdrom` does, because the CD is still
    /// FIRST — which is the whole of what makes the loop.
    CdromDisk,
}

impl BootOrder {
    pub fn as_str(self) -> &'static str {
        match self {
            BootOrder::Cdrom => "cdrom",
            BootOrder::Disk => "disk",
            BootOrder::CdromDisk => "cdrom,disk",
        }
    }

    pub fn parse(s: &str) -> Option<BootOrder> {
        match s {
            "cdrom" => Some(BootOrder::Cdrom),
            "disk" => Some(BootOrder::Disk),
            "cdrom,disk" => Some(BootOrder::CdromDisk),
            _ => None,
        }
    }

    /// Whether an attached medium is tried BEFORE the disk. The one question
    /// the boot path actually asks, so `cdrom,disk` cannot be forgotten at the
    /// place where forgetting it silently ends the installer loop.
    pub fn cdrom_first(self) -> bool {
        matches!(self, BootOrder::Cdrom | BootOrder::CdromDisk)
    }
}

/// What the guest is doing. The mock keeps this even without the `kvm` feature,
/// because the API's answers depend on it (an installer that loops never leaves
/// `Installing`) and because the KVM half must have somewhere to report to.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum Guest {
    /// No medium, nothing to run.
    Off,
    /// The installer is running. ~10 s, and **nothing narrates it**: PID 1
    /// brings no network up (behaviour 20) and there is no serial console
    /// (behaviour 13), so this window is silent by construction.
    Installing,
    /// The installer finished and the box booted its disk.
    Installed,
    /// The installer ran again, because the CD is still first in the boot order
    /// (behaviour 17). A mock whose guest could not do this could not reproduce
    /// the loop that cost an afternoon.
    Looping { rounds: u32 },
    /// The Ubuntu template's first boot: `dpkg lock-frontend` is held by
    /// `unattended-upgrade-shutdown --wait-for-signal` for the LIFE of the boot,
    /// so an `apt-get update` blocks rather than failing (behaviour 21).
    TemplateFirstBoot,
    /// PID 1 panicked. Invisible except on the framebuffer (behaviour 13).
    Panicked,
}

/// One network interface of a server, as the TERRAFORM door asks for them:
/// an index and a kind (`public` · `utility` · `private`). The addresses
/// themselves stay on [`Server`] — one public and one utility per machine, from
/// [`Addresses`] — because behaviour 12 is about the ADDRESS pool and not about
/// how many rows a caller declared.
///
/// It exists because `upcloud_server` declares its interfaces and READS THEM
/// BACK: holger's `outputs.tf` resolves `network_interface[0].ip_address` and
/// `network_interface[1].ip_address`, so a machine that declared one interface
/// and is answered with two has an output pointing at the wrong wire.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Iface {
    pub index: u32,
    pub kind: String,
    /// **Behaviour 61.** `IPv4` or `IPv6`, as the interface asked for it.
    /// UpCloud offers IPv6 (price 0); "no IPv6" is this estate's LAW, not the
    /// provider's limit. The mock used to answer IPv4 whatever was asked, so a
    /// declaration that broke the law would have passed here unseen.
    pub family: String,
}

impl Iface {
    /// What every caller in this crate had before the terraform door existed:
    /// one public, one utility. Kept as the default so the plugin's servers are
    /// unchanged by a door they never knock on.
    pub fn default_pair() -> Vec<Iface> {
        vec![
            Iface { index: 1, kind: "public".into(), family: "IPv4".into() },
            Iface { index: 2, kind: "utility".into(), family: "IPv4".into() },
        ]
    }
}

/// One firewall rule, every field a STRING, as the API sends them. See
/// [`crate::tf`] for the two spellings a write arrives in.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct Rule {
    pub position: String,
    pub direction: String,
    pub action: String,
    pub family: String,
    pub protocol: String,
    pub source_address_start: String,
    pub source_address_end: String,
    pub source_port_start: String,
    pub source_port_end: String,
    pub destination_address_start: String,
    pub destination_address_end: String,
    pub destination_port_start: String,
    pub destination_port_end: String,
    pub icmp_type: String,
    pub comment: String,
}

#[derive(Clone, Debug)]
pub struct Server {
    pub uuid: String,
    pub title: String,
    pub hostname: String,
    pub plan: String,
    pub zone: String,
    pub state: String,
    pub labels: Vec<Label>,
    pub devices: Vec<Device>,
    pub boot_order: BootOrder,
    pub remote_access_enabled: bool,
    pub remote_access_password: String,
    /// What the hypervisor is really listening on.
    pub vnc_port: u16,
    /// What the API SAYS it is listening on. These diverge on every stop/start
    /// (behaviour 14) and only a `remote_access_enabled` no→yes toggle
    /// reconciles them.
    pub reported_vnc_port: u16,
    /// **The console is a HOST and a port, and the toggle re-provisions both.**
    /// MEASURED: after two PUTs the console came back as
    /// `se-sto1.vnc.upcloud.com:60031` — a ZONE host, not the server's own
    /// address. A client that re-reads only the port after the cure keeps
    /// dialling the old place, which is the same defect one field further on.
    pub vnc_host: String,
    pub reported_vnc_host: String,
    pub public_ip: String,
    pub utility_ip: String,
    pub guest: Guest,
    /// **What this guest's DHCP client does with option 121.** The provider
    /// offers the routes; the guest does or does not install them. Same shape as
    /// [`crate::guest_clock::RtcInterpretation`], same reason.
    pub dhcp_client: crate::net::DhcpClient,
    /// **A new one on every re-image.** Which is expected, and which means every
    /// automation that pushes to the forge meets
    /// `REMOTE HOST IDENTIFICATION HAS CHANGED` on every bring-up.
    pub ssh_host_key: String,
    /// **How THIS guest's userland reads the RTC.** The hypervisor presents
    /// correct UTC to every server; the front gets it right and the appliance
    /// does not, and the difference is the guest's own software. Decided once,
    /// at create, from [`Fault::GuestReadsRtcAsLocalTime`] — see
    /// [`crate::guest_clock`] for why the clock itself is never skewed.
    pub rtc: crate::guest_clock::RtcInterpretation,
    pub created_ms: u64,
    pub transition: Option<Transition>,
    /// The interfaces this server was created with. See [`Iface`].
    pub ifaces: Vec<Iface>,
    /// The rule SET on this machine. Empty is not the same as absent: an empty
    /// set is a machine with the firewall ON and nothing allowed, and the API
    /// answers `[]` for it exactly as it does for a machine nobody ever wrote
    /// rules to. Only `firewall` tells the two apart.
    pub rules: Vec<Rule>,
    /// `firewall: "on" | "off"`, as asked for at create.
    pub firewall_on: bool,
    /// `metadata: "yes" | "no"`, as asked for at create.
    pub metadata: bool,
    /// **The timezone the server was CREATED with, which is not the guest's
    /// wall clock.** The detail renders this field because the provider reads
    /// it back and a machine that asked for `Europe/Stockholm` and is answered
    /// `UTC` is a permanent terraform diff. Behaviour 19 — the guest that reads
    /// the RTC as local time — lives in [`crate::guest_clock`] and is untouched
    /// by this: the API field says what was ASKED for, and the guest's clock is
    /// still wrong for its own reasons.
    pub timezone: String,
    /// **Behaviour 60: can this guest's kernel hot-plug PCI?** Decided at
    /// create from [`Fault::GuestKernelLacksHotplug`]; a property of the image.
    pub hotplug: bool,
    /// **Behaviour 50: the console has not moved yet.** Set by the "no"
    /// half of a toggle; a "yes" before the clock passes it gets the OLD
    /// endpoint back, which is why every working tool pauses between the two.
    pub console_settles_at: Option<u64>,
}

impl Server {
    /// What this guest's wall clock reads, minus the truth, in ms. Zero for a
    /// guest whose userland knows the RTC is UTC.
    pub fn clock_skew_ms(&self, unix_secs: i64) -> i64 {
        self.rtc.skew_ms(&self.zone, unix_secs)
    }

    /// The DHCP offer this server's utility NIC received. Note it is the OFFER,
    /// which is always complete: what varies is whether the guest took it.
    pub fn dhcp_offer(&self) -> crate::net::DhcpOffer {
        crate::net::DhcpOffer::for_address(&self.utility_ip)
    }

    /// The console endpoint the API reports, or `None` when remote access is
    /// off — in which case the API does not answer a stale one, it answers
    /// nothing, and a caller must be REFUSED BY NAME rather than handed the
    /// last known host and port.
    pub fn console(&self) -> Option<(String, u16)> {
        self.remote_access_enabled
            .then(|| (self.reported_vnc_host.clone(), self.reported_vnc_port))
    }
}

impl Server {
    pub fn label(&self, key: &str) -> Option<&str> {
        self.labels.iter().find(|l| l.key == key).map(|l| l.value.as_str())
    }
    pub fn boot_disk(&self) -> Option<&Device> {
        self.devices.iter().find(|d| d.boot_disk)
    }
}

/// How a caller reached the box it is checking the host key of.
///
/// Three, and they must agree. `Name` resolves through DNS to the front and
/// then through the DNAT; `Front` skips the DNS; `Direct` skips the DNAT too
/// and talks to the appliance's own public address. One key on three paths is
/// a re-imaged machine; one path disagreeing is a hijacked name.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum HostKeyPath {
    Name,
    Front,
    Direct,
}

impl HostKeyPath {
    pub fn parse(s: &str) -> Option<HostKeyPath> {
        match s {
            "name" => Some(HostKeyPath::Name),
            "front" => Some(HostKeyPath::Front),
            "direct" => Some(HostKeyPath::Direct),
            _ => None,
        }
    }
    pub const ALL: [HostKeyPath; 3] = [HostKeyPath::Name, HostKeyPath::Front, HostKeyPath::Direct];
    pub fn name(self) -> &'static str {
        match self {
            HostKeyPath::Name => "name",
            HostKeyPath::Front => "front",
            HostKeyPath::Direct => "direct",
        }
    }
}

/// A scheduled state change.
#[derive(Clone, Debug)]
pub struct Transition {
    pub to: String,
    pub at_ms: u64,
    pub then: After,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub enum After {
    Nothing,
    /// Free the object (the delete completes).
    Vanish,
    /// The guest starts running — installer, template first boot, or the disk.
    GuestBoots,
    /// The plan change completed; mint the `Resize Backup` off `origin`.
    MintResizeBackup { origin: String },
    /// **The upload finished and the SYNC begins.** The import object flips to
    /// `completed` here — with its `completed_ms` — while the storage enters
    /// `syncing` and stays there for [`Timings::storage_sync_lo_ms`] to
    /// [`Timings::storage_sync_hi_ms`]. Two clocks, one object, and the second
    /// one starts only when the first has stopped.
    BeginSync,
    /// Go `online` this many ms after arriving. The second leg of a clone's
    /// `maintenance` → `syncing` → `online`, which needs two hops and a
    /// transition carries one.
    OnlineIn { ms: u64 },
    /// **The installer powered the box off (behaviour 46).** The guest is
    /// `Installed`. The transition carrying this is HELD while a real machine
    /// behind the server still runs: the API reports UpCloud's time, but it
    /// never says `stopped` over a running VM.
    GuestPoweredOff,
    /// **The installer's brief `started` is over; the pass runs in
    /// `maintenance`** and ends `stopped` (via [`After::GuestPoweredOff`])
    /// after `left_ms` more.
    InstallPass { left_ms: u64 },
    /// **The rebooting medium's pass ended (behaviour 69): the guest reboots.**
    /// The server stays `started`. The CD still loaded and first → the
    /// installer runs again (17) and another pass is scheduled; otherwise the
    /// disk boots and the guest is `Installed`.
    MediumRebooted,
    /// **A real guest powered itself off while the server read `started`.**
    /// Noticed `poweroff_notice_ms` later; the guest is `Off`.
    GuestGone,
    /// **A size grow finished (behaviour 55).** Marks the `maintenance` of a
    /// grow, which a filesystem resize may follow at once: 5.44.1 sends
    /// `POST /storage/{uuid}/resize` straight after the PUT, without waiting,
    /// against the live account.
    GrowDone,
    /// **A fresh `POST /storage` settling (behaviour 66).** Nothing happens on
    /// arrival; it marks the create's `maintenance` as the one an import may be
    /// started in.
    Created,
}

// ── the address pools ────────────────────────────────────────────────────────

/// **RFC 5737's three documentation blocks** (`/24` each), the ONLY place a
/// mock public address may come from: not routed on the internet, so nothing
/// the mock hands out can ever be someone else's machine.
pub const TEST_NET: [&str; 3] = ["192.0.2", "198.51.100", "203.0.113"];

/// Is `ip` inside one of the [`TEST_NET`] blocks?
pub fn is_test_net(ip: &str) -> bool {
    match ip.parse::<std::net::Ipv4Addr>() {
        Ok(a) => {
            let o = a.octets();
            TEST_NET.contains(&format!("{}.{}.{}", o[0], o[1], o[2]).as_str())
        }
        Err(_) => false,
    }
}

/// Is `ip` one the utility pool hands out: `10.13.8.96–120` or `10.13.12.96–120`?
pub fn is_utility_pool(ip: &str) -> bool {
    match ip.parse::<std::net::Ipv4Addr>() {
        Ok(a) => {
            let o = a.octets();
            o[0] == 10 && o[1] == 13 && (o[2] == 8 || o[2] == 12) && (96..=120).contains(&o[3])
        }
        Err(_) => false,
    }
}

/// **Behaviour 12: addresses move.**
///
/// The appliance got `10.13.8.101` one re-lay and `10.13.8.99` the next, with
/// the twin holding the other; the public address changed every single time.
/// Code that held a literal went stale exactly that way, so the mock shuffles
/// both pools on every lay and hands them out in the shuffled order. A consumer
/// that pins an address fails here on the second lay instead of in production
/// on the second re-lay.
///
/// **The public pool is RFC 5737 TEST-NET, and nothing else — ever.** It was a
/// list of REAL UpCloud addresses (our estate's of 2026-09-15, and others), and
/// one of them had port 22 open on a machine that is not ours: every mock step
/// that connects to a handed-out address — verify, the ssh to the front, the
/// re-image's banner check — went out onto the internet, to a stranger. A
/// TEST-NET address is not routed, so a mock that connects to one fails
/// locally instead of reaching someone. [`TEST_NET`] is the check, and the
/// tests hold every address the estate can hand out to it.
///
/// The utility pool is RFC 1918 (`10.13.8.96–120`, `10.13.12.96–120`, across
/// both /22s for option 121, behaviour 29), disjoint from the networks of the
/// box the rig runs on.
pub struct Addresses {
    utility: Vec<String>,
    public: Vec<String>,
    next_utility: usize,
    next_public: usize,
    lays: u64,
}

impl Addresses {
    pub fn new(seed: u64) -> Addresses {
        let mut a = Addresses {
            // **Across BOTH utility prefixes**, because that is what makes
            // option 121 load-bearing: the appliance and the front land in
            // different /22s and the route between them arrives only as a
            // classless static route. A pool inside one prefix would let a
            // guest that ignores the option work perfectly, and the bug that
            // cost an afternoon would be unreachable here.
            utility: (96..=120)
                .map(|n| format!("10.13.8.{n}"))
                .chain((96..=120).map(|n| format!("10.13.12.{n}")))
                .collect(),
            // Eight from each TEST-NET block: enough that two lays almost
            // never hand the same server the same address (behaviour 12).
            public: TEST_NET
                .iter()
                .flat_map(|net| (10..18).map(move |h| format!("{net}.{h}")))
                .collect(),
            next_utility: 0,
            next_public: 0,
            lays: 0,
        };
        a.relay(seed);
        a
    }

    /// A new lay of the estate: reshuffle, hand out from the top again.
    pub fn relay(&mut self, seed: u64) {
        self.lays += 1;
        let mut r = SplitMix64::derive(seed, &format!("addresses/lay/{}", self.lays));
        r.shuffle(&mut self.utility);
        r.shuffle(&mut self.public);
        self.next_utility = 0;
        self.next_public = 0;
    }

    pub fn lays(&self) -> u64 {
        self.lays
    }

    pub fn take_utility(&mut self) -> String {
        let v = self.utility[self.next_utility % self.utility.len()].clone();
        self.next_utility += 1;
        v
    }

    pub fn take_public(&mut self) -> String {
        let v = self.public[self.next_public % self.public.len()].clone();
        self.next_public += 1;
        v
    }
}

// ── the refusals ─────────────────────────────────────────────────────────────

/// A refusal the mock can answer with. `code` is UpCloud's `error_code`.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Refusal {
    pub status: u16,
    pub code: &'static str,
    pub message: String,
}

impl Refusal {
    pub fn new(status: u16, code: &'static str, message: impl Into<String>) -> Refusal {
        Refusal { status, code, message: message.into() }
    }
}

pub type Answer<T> = Result<T, Refusal>;

// ── the estate ───────────────────────────────────────────────────────────────

pub struct Estate {
    pub clock: Clock,
    pub timings: Timings,
    pub faults: Faults,
    pub zone: String,
    /// What the mock says its own upload sessions live at. `serve` fills it in
    /// with the socket it bound, so a caller that FOLLOWS the
    /// `direct_upload_url` from the import reply reaches the mock instead of
    /// the internet. Empty renders the real provider's shape
    /// (`https://<zone>.img.upcloud.com/uploader/session/<uuid>`), which is what
    /// a parser should be tested against.
    pub upload_base: String,
    /// Every destination-NAT rule in the estate: the front's `:2222` to the
    /// appliance, and anything else a lay installs. Held here because the
    /// hairpin question ("can the box holding the rule use it") can only be
    /// answered by something that knows all of them.
    pub dnat: Vec<crate::net::Dnat>,
    /// The account's MaxIOPS quota in GiB (behaviour 44). MEASURED for yvra.
    pub maxiops_quota_gib: u64,
    /// **What is behind a server.** [`VirtualGuest`] by default, which runs
    /// nothing and leaves every answer to the state machine below. A real
    /// engine (the `draupnir-guest` crate's `DraupnirGuest`, injected by the
    /// `mock-upcloud-kvm` binary) boots a QEMU at every `started`, and the
    /// server's `state` then follows that machine: see [`Estate::reap`].
    pub engine: Arc<dyn GuestEngine>,
    servers: BTreeMap<String, Server>,
    storages: BTreeMap<String, Storage>,
    /// Servers and storages that ONCE existed. Behaviour 1 needs this: the
    /// firewall endpoint's 403 is what a DELETED server answers, and a mock that
    /// forgot its dead could not tell that from a uuid nobody ever minted —
    /// which, as it happens, is exactly the ambiguity that stopped Terraform.
    tombstones: BTreeMap<String, &'static str>,
    addresses: Addresses,
    ids: SplitMix64,
    seed: u64,
    correlations: u64,
}

impl Estate {
    pub fn new(clock: Clock, faults: Faults, seed: u64) -> Estate {
        let mut e = Estate {
            clock,
            timings: Timings::default(),
            faults,
            zone: "se-sto1".to_string(),
            upload_base: String::new(),
            dnat: Vec::new(),
            maxiops_quota_gib: 10_240,
            engine: Arc::new(VirtualGuest::default()),
            servers: BTreeMap::new(),
            storages: BTreeMap::new(),
            tombstones: BTreeMap::new(),
            addresses: Addresses::new(seed),
            ids: SplitMix64::derive(seed, "uuids"),
            seed,
            correlations: 0,
        };
        e.seed_public_templates();
        e
    }

    pub fn seed(&self) -> u64 {
        self.seed
    }

    /// The same estate with a real machine behind every server.
    pub fn with_engine(mut self, engine: Arc<dyn GuestEngine>) -> Estate {
        self.engine = engine;
        self
    }

    /// **The server's state follows its machine.** A server that reads
    /// `started` while its QEMU is gone is a health line that cannot fail —
    /// and the installer powers the box off ~2–3 s after INSTALL-OK, so this is
    /// the ordinary case, not a crash. Only a server the engine has a machine
    /// for is touched (`running` is `None` otherwise), so the default engine
    /// changes nothing.
    pub fn reap(&mut self) {
        let gone: Vec<String> = self
            .servers
            .values()
            .filter(|s| s.state == "started")
            .filter(|s| self.engine.running(&s.uuid) == Some(false))
            .map(|s| s.uuid.clone())
            .collect();
        // **The timing rule (lane T13, which owns this from here).** The API
        // reports UpCloud's time, not the VM's: a guest that powers itself off
        // is NOTICED `poweroff_notice_ms` later, so the server keeps reading
        // `started` for that long (the API may run behind the machine, never
        // ahead of it — the other direction is held in `settle_server`). A
        // server already on its way somewhere keeps its own transition.
        let at = self.clock.now_ms() + self.timings.poweroff_notice_ms;
        for uuid in gone {
            if let Some(s) = self.servers.get_mut(&uuid) {
                if s.transition.is_none() {
                    s.transition = Some(Transition { to: "stopped".into(), at_ms: at, then: After::GuestGone });
                }
            }
        }
    }

    /// The machine the engine is asked to power on for this server, from the
    /// records as they are NOW: sizes from the storage records (a resize
    /// changes those, not the device rows), the tray from the cdrom device.
    fn machine_for(&self, uuid: &str) -> Option<Machine> {
        let s = self.servers.get(uuid)?;
        let mut devs: Vec<&Device> = s.devices.iter().filter(|d| d.kind == "disk").collect();
        devs.sort_by_key(|d| !d.boot_disk);
        let disks = devs
            .iter()
            .map(|d| MachineDisk {
                storage: d.storage.clone(),
                size_gib: self.storages.get(&d.storage).map(|st| st.size_gib).unwrap_or(d.storage_size),
            })
            .collect();
        let medium = s
            .devices
            .iter()
            .find(|d| d.kind == "cdrom" && !d.storage.is_empty())
            .map(|d| d.storage.clone());
        Some(Machine {
            server: uuid.to_string(),
            mem_mb: crate::render::memory_of(&s.plan),
            cores: crate::render::cores_of(&s.plan),
            disks,
            cdrom_first: s.boot_order.cdrom_first() && medium.is_some(),
            medium,
            rtc_skew_seconds: crate::kvm::GuestSpec::new("", "").rtc_skew_seconds,
        })
    }

    /// The public images UpCloud publishes. They matter for one reason: a bare
    /// `GET /1.3/storage` lists them — thousands of rows, none of them the
    /// account's — which is why `gunnar/deploy/upcloud` reads
    /// `/1.3/storage/private` instead. The mock carries a handful so that
    /// difference is visible rather than theoretical.
    fn seed_public_templates(&mut self) {
        for title in [
            "Ubuntu Server 24.04 LTS (Noble Numbat)",
            "Ubuntu Server 22.04 LTS (Jammy Jellyfish)",
            "Debian GNU/Linux 12 (Bookworm)",
            "AlmaLinux 9",
        ] {
            let uuid = self.mint_uuid();
            self.storages.insert(
                uuid.clone(),
                Storage {
                    uuid,
                    title: title.to_string(),
                    size_gib: 10,
                    tier: "maxiops".into(),
                    zone: self.zone.clone(),
                    state: "online".into(),
                    kind: StorageKind::Template,
                    labels: vec![],
                    origin: None,
                    created_ms: 0,
                    import: None,
                    transition: None,
                },
            );
        }
    }

    fn mint_uuid(&mut self) -> String {
        // UpCloud's uuids are `01234567-89ab-cdef-0123-456789abcdef`, minted
        // here from the run's seed so the same seed replays the same ids and a
        // failure signature can name one.
        let a = self.ids.next_u64();
        let b = self.ids.next_u64();
        format!(
            "{:08x}-{:04x}-{:04x}-{:04x}-{:012x}",
            (a >> 32) as u32,
            (a >> 16) as u16,
            a as u16,
            (b >> 48) as u16,
            b & 0xffff_ffff_ffff
        )
    }

    pub fn next_correlation_id(&mut self) -> String {
        self.correlations += 1;
        format!("{:016x}{:08x}", self.ids.next_u64(), self.correlations as u32)
    }

    // ── time ────────────────────────────────────────────────────────────────

    /// Move every object whose transition is due, then — when the clock is
    /// virtual — push time part-way toward the next one.
    ///
    /// Part-way, not all the way, on purpose: a client that polls must poll
    /// several times, as it does against the real provider. Advancing straight
    /// to the next deadline would make every transition complete on the second
    /// read and silently excuse a client with no loop at all.
    pub fn tick(&mut self) {
        self.settle();
        if self.clock.speed_milli() == 0 {
            if let Some(next) = self.next_deadline() {
                let now = self.clock.now_ms();
                if next > now {
                    self.clock.advance_ms(((next - now) / 2).max(1));
                }
            }
        }
    }

    /// Jump the virtual clock to the moment nothing is pending any more.
    ///
    /// Only two callers, and both are honest about why: a `restart` is a stop
    /// and a start and the stop must COMPLETE in between, and a test that is
    /// not testing the poll loop says so by calling this. It is a no-op at real
    /// speed, where waiting is the only way through — which is the right
    /// asymmetry: virtual time may skip, real time may not.
    pub fn run_to_quiet(&mut self) {
        if self.clock.speed_milli() != 0 {
            self.settle();
            return;
        }
        for _ in 0..1000 {
            self.settle();
            match self.next_deadline() {
                None => return,
                Some(at) => {
                    let now = self.clock.now_ms();
                    self.clock.advance_ms(at.saturating_sub(now).max(1));
                }
            }
        }
    }

    fn next_deadline(&self) -> Option<u64> {
        let a = self.servers.values().filter_map(|s| s.transition.as_ref().map(|t| t.at_ms));
        let b = self.storages.values().filter_map(|s| s.transition.as_ref().map(|t| t.at_ms));
        // The console settle (behaviour 50) is a deadline a poller should be
        // able to wait out at virtual speed; only a FUTURE one, or
        // `run_to_quiet` would spin on it.
        let now = self.clock.now_ms();
        let c = self.servers.values().filter_map(|s| s.console_settles_at).filter(|t| *t > now);
        a.chain(b).chain(c).min()
    }

    /// Apply every due transition. Idempotent, and the only writer of a state
    /// that nobody asked for.
    pub fn settle(&mut self) {
        self.reap();
        loop {
            let now = self.clock.now_ms();
            let due_server = self
                .servers
                .values()
                .find(|s| s.transition.as_ref().is_some_and(|t| t.at_ms <= now))
                .map(|s| s.uuid.clone());
            if let Some(uuid) = due_server {
                self.settle_server(&uuid);
                continue;
            }
            let due_storage = self
                .storages
                .values()
                .find(|s| s.transition.as_ref().is_some_and(|t| t.at_ms <= now))
                .map(|s| s.uuid.clone());
            if let Some(uuid) = due_storage {
                self.settle_storage(&uuid);
                continue;
            }
            return;
        }
    }

    fn settle_server(&mut self, uuid: &str) {
        let Some(s) = self.servers.get_mut(uuid) else { return };
        let Some(t) = s.transition.take() else { return };
        // **Chain from the moment the transition was DUE, not the moment it is
        // settled.** A caller whose clock jumped (a 10 s poll, a virtual sleep)
        // settles late; a follow-on scheduled from "now" would open a short
        // window — the installer's ~83 ms `started` — AT the jump, and the next
        // read would always see it however coarse the poll. From `due`, a
        // window that closed during the jump is settled in the same pass and
        // never read: virtual time replays like real time (T14's flow test
        // a_medium_that_powers_off_may_end_its_pass_without_ever_being_read_started).
        let due = t.at_ms;
        // **The timing rule, one direction of it.** A modelled `stopped` that
        // a real machine contradicts (it is still running) waits for the
        // machine; the API may run behind the VM, never ahead of it.
        if t.then == After::GuestPoweredOff && self.engine.running(uuid) == Some(true) {
            let at = self.clock.now_ms() + self.timings.poweroff_notice_ms;
            if let Some(s) = self.servers.get_mut(uuid) {
                s.transition = Some(Transition { at_ms: at, ..t });
            }
            return;
        }
        let Some(s) = self.servers.get_mut(uuid) else { return };
        s.state = t.to.clone();
        // `stopped` means no machine. A soft stop pressed the power button when
        // it was asked; a guest that has not acted on it by now is killed, which
        // is what the provider's stop timeout does too.
        if t.to == "stopped" {
            self.engine.power_off(uuid, true);
        }
        match t.then {
            After::Nothing => {}
            After::Vanish => {
                let s = self.servers.remove(uuid).expect("just had it");
                self.tombstones.insert(uuid.to_string(), "server");
                // `?storages=1` took the attached disks with it. A BACKUP is
                // never attached and so never goes this way — which is exactly
                // how a `Resize Backup` outlives the server it came from and
                // keeps billing (behaviour 9).
                let dead: Vec<String> = s
                    .devices
                    .iter()
                    .filter(|d| {
                        self.storages
                            .get(&d.storage)
                            .is_some_and(|st| st.kind != StorageKind::Backup && st.kind != StorageKind::Template)
                    })
                    .map(|d| d.storage.clone())
                    .collect();
                for u in dead {
                    self.storages.remove(&u);
                    self.engine.forget_storage(&u);
                    self.tombstones.insert(u, "storage");
                }
                return;
            }
            After::GuestBoots => {
                let looping = self.faults.fires(Fault::InstallerLoop);
                let s = self.servers.get_mut(uuid).expect("just had it");
                // A MEDIUM in the tray, not a cdrom DEVICE: an eject leaves the
                // device with an empty `storage` (L52), and `cdrom,disk` with
                // an empty tray falls through to the disk (L54). Asking for
                // the device made every disk boot after an eject an installer
                // pass, held in `maintenance` while the appliance served gRPC.
                let has_cdrom = s.devices.iter().any(|d| d.kind == "cdrom" && !d.storage.is_empty());
                s.guest = match (s.boot_order.cdrom_first(), has_cdrom) {
                    // Behaviour 17: the CD is still first, so the installer runs
                    // again. Forever, because nothing in the guest changes the
                    // boot order — only `cdrom/eject` from outside ends it.
                    (true, true) if looping => Guest::Looping { rounds: 1 },
                    (true, true) => Guest::Installing,
                    (_, _) if s.boot_disk().is_some() => Guest::TemplateFirstBoot,
                    _ => Guest::Off,
                };
                // **A re-image mints a new SSH host key. Every time.** Expected,
                // and the reason every automation that pushes to the forge meets
                // `REMOTE HOST IDENTIFICATION HAS CHANGED` on every bring-up.
                let now = self.clock.now_ms();
                let mut kr = SplitMix64::derive(self.seed, &format!("hostkey/{uuid}/{now}"));
                let key = format!("SHA256:{:016x}{:016x}", kr.next_u64(), kr.next_u64());
                let lo = self.timings.guest_install_uart_lo_ms;
                let hi = self.timings.guest_install_uart_hi_ms;
                let install_ms = kr.range(lo, hi);
                let at = self.clock.now_ms() + install_ms;
                let s = self.servers.get_mut(uuid).expect("just had it");
                // An install mints them, and so does an ordinary first boot —
                // cloud-init generates host keys on a template's first boot for
                // exactly the same reason. Either way the box that comes up is
                // not the box that went down, as far as `known_hosts` is
                // concerned.
                if matches!(s.guest, Guest::Installing | Guest::Looping { .. } | Guest::TemplateFirstBoot) {
                    s.ssh_host_key = key;
                }
                // **Behaviour 46, corrected: an installer that powers the box off
                // reads `started` BRIEFLY, then `maintenance` for the pass, then
                // `stopped`.** MEASURED 2026-09-21 on the live re-image (receipt:
                // `started` read 3.67 s after the start, `maintenance` 83 ms
                // later, `stopped` 153.5 s after the start). The 2026-09-19
                // "never `started`" was a 10 s poll that began at t+10 s. The
                // guest's own install is 2–5 s (`at`, the UART clock); the API
                // reports the PROVIDER's pass, and that is the floor even when a
                // real guest behind the mock finished early. A real guest that
                // runs longer holds the `stopped` back (`After::GuestPoweredOff`).
                // **Behaviour 69, the REBOOTING medium** (fault
                // `installer-reboots`): no power-off, so the API reads
                // `started` for the whole 900–1100 s pass, and the pass ends
                // in a REBOOT (see `After::MediumRebooted`).
                let reboots = matches!(s.guest, Guest::Installing) && self.faults.fires(Fault::InstallerReboots);
                if reboots {
                    let pass = kr.range(self.timings.reboot_pass_lo_ms, self.timings.reboot_pass_hi_ms);
                    let s = self.servers.get_mut(uuid).expect("just had it");
                    s.transition = Some(Transition {
                        to: "started".into(),
                        at_ms: due + pass,
                        then: After::MediumRebooted,
                    });
                } else if matches!(s.guest, Guest::Installing) {
                    let _ = at;
                    let pass = kr.range(self.timings.install_pass_lo_ms, self.timings.install_pass_hi_ms);
                    let window = self.timings.installer_started_window_ms;
                    let left = pass.saturating_sub(self.timings.installer_started_ms + window).max(1);
                    let s = self.servers.get_mut(uuid).expect("just had it");
                    // `started` stays, for the window only.
                    s.transition = Some(Transition {
                        to: "maintenance".into(),
                        at_ms: due + window,
                        then: After::InstallPass { left_ms: left },
                    });
                }
                // **The machine.** `started` is when the provider's VM really
                // starts, so this is where a real engine boots one. A refusal
                // (a clause the engine will not boot without, a host with no
                // KVM) leaves the server `stopped`, never `started` over nothing.
                if let Some(m) = self.machine_for(uuid) {
                    if let Err(why) = self.engine.power_on(&m) {
                        eprintln!("mock-upcloud  server {uuid}: the machine did not start: {why}");
                        let s = self.servers.get_mut(uuid).expect("just had it");
                        s.state = "stopped".into();
                        s.guest = Guest::Off;
                        s.transition = None;
                    } else if let Some((host, port)) = self.engine.console(uuid) {
                        // **The console is the guest's real VNC.** The machine
                        // is where the hypervisor really listens (behaviour 14's
                        // `vnc_*`); the REPORTED pair follows it only if this
                        // start did not leave it stale — `start_server` set the
                        // two equal exactly when the stale-port fault did not
                        // fire, so the defect stays on top of a real console.
                        let s = self.servers.get_mut(uuid).expect("just had it");
                        let follows = s.reported_vnc_port == s.vnc_port && s.reported_vnc_host == s.vnc_host;
                        s.vnc_host = host;
                        s.vnc_port = port;
                        if follows {
                            s.reported_vnc_host = s.vnc_host.clone();
                            s.reported_vnc_port = s.vnc_port;
                        }
                    }
                }
            }
            After::MintResizeBackup { origin } => {
                let _ = self.mint_resize_backup(&origin);
            }
            // A server has no import and no sync: those are a STORAGE's two
            // clocks. Named rather than swallowed, so a new `After` that a
            // server should honour fails to compile here instead of silently
            // doing nothing.
            After::MediumRebooted => {
                let now = self.clock.now_ms();
                let mut kr = SplitMix64::derive(self.seed, &format!("reboot/{uuid}/{now}"));
                let pass = kr.range(self.timings.reboot_pass_lo_ms, self.timings.reboot_pass_hi_ms);
                let key = format!("SHA256:{:016x}{:016x}", kr.next_u64(), kr.next_u64());
                if let Some(s) = self.servers.get_mut(uuid) {
                    let cd_again = s.boot_order.cdrom_first()
                        && s.devices.iter().any(|d| d.kind == "cdrom" && !d.storage.is_empty());
                    if cd_again {
                        // Behaviour 17: the firmware finds the CD first again
                        // and the installer runs again — a new pass, a new key.
                        let rounds = match s.guest {
                            Guest::Looping { rounds } => rounds + 1,
                            _ => 1,
                        };
                        s.guest = Guest::Looping { rounds };
                        s.ssh_host_key = key;
                        s.transition = Some(Transition { to: "started".into(), at_ms: due + pass, then: After::MediumRebooted });
                    } else {
                        // No medium (ejected) or the disk first: the installed
                        // system boots, and the server simply stays `started`.
                        s.guest = Guest::Installed;
                    }
                }
            }
            After::InstallPass { left_ms } => {
                let at = due + left_ms;
                if let Some(s) = self.servers.get_mut(uuid) {
                    s.transition = Some(Transition { to: "stopped".into(), at_ms: at, then: After::GuestPoweredOff });
                }
            }
            After::BeginSync | After::OnlineIn { .. } | After::GrowDone | After::Created => {}
            After::GuestPoweredOff => {
                if let Some(s) = self.servers.get_mut(uuid) {
                    s.guest = Guest::Installed;
                }
            }
            After::GuestGone => {
                if let Some(s) = self.servers.get_mut(uuid) {
                    s.guest = Guest::Off;
                }
            }
        }
    }

    fn settle_storage(&mut self, uuid: &str) {
        let Some(s) = self.storages.get_mut(uuid) else { return };
        let Some(t) = s.transition.take() else { return };
        s.state = t.to.clone();
        match t.then {
            After::Vanish => {
                self.storages.remove(uuid);
                self.engine.forget_storage(uuid);
                self.tombstones.insert(uuid.to_string(), "storage");
            }
            // **The two clocks part company here.** Every byte has arrived and
            // been checksummed — the import says `completed`, with a
            // `completed_ms` five seconds after its `created` — and the storage
            // now enters the hundred-and-something seconds of `syncing` during
            // which neither side is doing anything the caller can see.
            After::BeginSync => {
                let now = self.clock.now_ms();
                let over_budget = self.faults.fires(Fault::SyncExceedsBudget);
                let failed = self.faults.fires(Fault::ImportFailed);
                let (lo, hi) = (self.timings.storage_sync_lo_ms, self.timings.storage_sync_hi_ms);
                let tail_ms = self.timings.sync_tail_maintenance_ms;
                let mut r = SplitMix64::derive(self.seed, &format!("sync/{uuid}"));
                // The caller polls with a 1200 s budget. This fault steps over
                // it — deliberately, because a timeout that has never fired is
                // a timeout nobody has read the handler of.
                let ms = if over_budget { 1_300_000 } else { r.range(lo, hi) };
                let s = self.storages.get_mut(uuid).expect("just settled");
                if let Some(im) = s.import.as_mut() {
                    if failed {
                        im.state = "failed".into();
                        im.completed_ms = Some(now);
                        im.error_code = Some("IMPORT_FAILED".into());
                        im.error_message =
                            Some("the uploaded image could not be written to the storage".into());
                    } else {
                        im.state = "completed".into();
                        im.completed_ms = Some(now);
                    }
                }
                if failed {
                    s.state = "error".into();
                } else {
                    // **Behaviour 52.** syncing → maintenance → online, MEASURED as
                    // a sequence (clone_probe: "maintenance, syncing, maintenance,
                    // online"). The split is not measured, so the tail is carved
                    // out of the sync range and the total stays the measured one.
                    let tail = tail_ms.min(ms.saturating_sub(1));
                    s.transition = Some(Transition {
                        to: "maintenance".into(),
                        at_ms: now + ms - tail,
                        then: After::OnlineIn { ms: tail },
                    });
                }
            }
            After::OnlineIn { ms } => {
                let at = self.clock.now_ms() + ms;
                if let Some(s) = self.storages.get_mut(uuid) {
                    s.transition = Some(Transition { to: "online".into(), at_ms: at, then: After::Nothing });
                }
            }
            _ => {}
        }
    }

    // ── behaviour 9: the Resize Backup ──────────────────────────────────────

    /// **Behaviour 9.** A `stop-resize-start` leaves this behind: type `backup`,
    /// detached, `origin` naming the volume that was resized — which may already
    /// be gone — and carrying the estate's own labels (`site`, `role`, `repo`,
    /// `volume`) but NOT the word "gunnar" anywhere in its title.
    ///
    /// That last clause is the defect: the cleanup filtered by title prefix, and
    /// a title-prefix filter is BLIND to this row. It billed maxiops for it at
    /// about €0.20 per GB-month, forever, and nothing listed it.
    fn mint_resize_backup(&mut self, origin: &str) -> String {
        let (size, labels, zone) = match self.storages.get(origin) {
            Some(o) => (o.size_gib, o.labels.clone(), o.zone.clone()),
            // The origin is already gone. The backup is minted anyway — this is
            // the orphan, and it is the normal case after the server that owned
            // the volume has been destroyed.
            None => (20, vec![], self.zone.clone()),
        };
        // The brief's hypothesis, by name: a backup the provider titled ITSELF
        // and labelled with nothing. The default is the measurement (labels
        // copied); see [`Fault::ResizeBackupUnlabelled`].
        let labels = if self.faults.fires(Fault::ResizeBackupUnlabelled) { vec![] } else { labels };
        let now = self.clock.now_ms();
        let uuid = self.mint_uuid();
        let orphan = self.faults.fires(Fault::OrphanResizeBackup);
        self.storages.insert(
            uuid.clone(),
            Storage {
                uuid: uuid.clone(),
                // The auto-title. No product name, no site name, no "gunnar".
                title: format!("Resize Backup {}", now / 1000),
                size_gib: size,
                tier: "maxiops".into(),
                zone,
                state: "online".into(),
                kind: StorageKind::Backup,
                labels,
                origin: Some(if orphan {
                    // A uuid that resolves to nothing: the origin volume was
                    // deleted with its server before anyone looked.
                    format!("{}-gone", &origin[..origin.len().min(30)])
                } else {
                    origin.to_string()
                }),
                created_ms: now,
                import: None,
                transition: None,
            },
        );
        uuid
    }

    /// **Behaviour 34 — `POST /1.3/storage/{uuid}/resize`, the door that
    /// actually left the 44 GB behind.**
    ///
    /// Behaviour 9 modelled the `Resize Backup` on a PLAN change, which is a
    /// door this estate never opens. The one it does open is this: `cargo xtask
    /// grow` grows the twin's volume with `PUT /storage/{uuid}` (behaviour 22)
    /// and then asks the provider to grow the LAST PARTITION and the xfs inside
    /// it with `POST /storage/{uuid}/resize`. The provider takes a backup FIRST
    /// and hands it back in the reply as `resize_backup` — the whole storage
    /// object, uuid and all — and nothing in the estate deletes it afterwards.
    /// MEASURED 2026-09-20: a 44 GB `Resize Backup` on the live account whose
    /// `origin` was the twin's data volume, billed at maxiops since 2026-09-19
    /// 21:54:06Z, and `upcloud-orphans` said "no orphans" over it.
    ///
    /// The backup is the resized volume's size (that is what 44 GB was: the
    /// twin's volume after its growth), carries the volume's labels unless
    /// [`Fault::ResizeBackupUnlabelled`] is armed, and its `origin` is the
    /// volume — or a gone uuid under [`Fault::OrphanResizeBackup`]. The volume
    /// itself goes `maintenance` for the resize and comes back `online`, so a
    /// caller that does not poll gets the same lesson every other write here
    /// teaches.
    ///
    /// Refused on a volume attached to a server that is not `stopped`
    /// (`SERVER_STATE_ILLEGAL`; the estate stops the twin first and says so in
    /// its journal) and on a volume that is not `online`.
    pub fn resize_filesystem(&mut self, uuid: &str) -> Answer<String> {
        self.refuse_if_write_unavailable()?;
        let Some(s) = self.storages.get(uuid) else {
            return Err(Refusal::new(404, "STORAGE_NOT_FOUND", format!("storage {uuid} not found")));
        };
        // Online, or still in the `maintenance` of a size grow: 5.44.1 sends
        // this straight after the PUT, without waiting, and it works against
        // the live account (behaviour 55's other half).
        let after_grow = s.state == "maintenance" && s.transition.as_ref().is_some_and(|t| t.then == After::GrowDone);
        if s.state != "online" && !after_grow {
            return Err(Refusal::new(
                409,
                "STORAGE_STATE_ILLEGAL",
                format!("storage {uuid} is {} — wait for online before resizing its filesystem", s.state),
            ));
        }
        for sv in self.attached_servers(uuid) {
            let state = self.servers.get(&sv).map(|x| x.state.clone()).unwrap_or_default();
            if state != "stopped" {
                return Err(Refusal::new(
                    409,
                    "SERVER_STATE_ILLEGAL",
                    format!("storage {uuid} is attached to server {sv}, which is {state}; the filesystem is resized only on a stopped server"),
                ));
            }
        }
        let backup = self.mint_resize_backup(uuid);
        if let Some(gib) = self.storages.get(uuid).map(|s| s.size_gib) {
            // Only reachable on a stopped server (refused above otherwise), so
            // the disk is never grown under a running machine.
            if let Err(why) = self.engine.resize_disk(uuid, gib) {
                eprintln!("mock-upcloud  storage {uuid}: disk resize refused: {why}");
            }
        }
        let now = self.clock.now_ms();
        let ms = self.timings.resize_ms;
        if let Some(s) = self.storages.get_mut(uuid) {
            s.state = "maintenance".into();
            s.transition = Some(Transition { to: "online".into(), at_ms: now + ms, then: After::Nothing });
        }
        Ok(backup)
    }

    // ── reads ───────────────────────────────────────────────────────────────

    pub fn server(&self, uuid: &str) -> Option<&Server> {
        self.servers.get(uuid)
    }

    pub fn storage(&self, uuid: &str) -> Option<&Storage> {
        self.storages.get(uuid)
    }

    pub fn was(&self, uuid: &str) -> Option<&'static str> {
        self.tombstones.get(uuid).copied()
    }

    /// `GET /1.3/server`, filtered by `?label=key=value` (all must match).
    ///
    /// **Behaviour 5**: a revoked credential answers this with 200 and ZERO
    /// rows. Not 401. A caller that treats "no rows" as "nothing to clean up"
    /// deletes nothing and reports success, and a caller that treats it as
    /// "nothing exists" creates a second copy of everything.
    pub fn servers_matching(&self, labels: &[(String, String)]) -> Vec<&Server> {
        if self.faults.fires(Fault::RevokedCredential) {
            return vec![];
        }
        self.servers
            .values()
            .filter(|s| labels.iter().all(|(k, v)| s.label(k) == Some(v.as_str())))
            .collect()
    }

    pub fn storages_matching(&self, labels: &[(String, String)], private_only: bool) -> Vec<&Storage> {
        if self.faults.fires(Fault::RevokedCredential) {
            return vec![];
        }
        self.storages
            .values()
            .filter(|s| !private_only || s.kind != StorageKind::Template)
            .filter(|s| {
                labels
                    .iter()
                    .all(|(k, v)| s.labels.iter().any(|l| l.key == *k && l.value == *v))
            })
            .collect()
    }

    // ── writes ──────────────────────────────────────────────────────────────

    pub fn create_storage(
        &mut self,
        title: &str,
        size_gib: u64,
        tier: &str,
        zone: &str,
        labels: Vec<Label>,
    ) -> Answer<String> {
        self.refuse_if_write_unavailable()?;
        if size_gib == 0 {
            return Err(Refusal::new(400, "STORAGE_INVALID_SIZE", "storage size must be at least 1 GiB"));
        }
        self.refuse_over_quota_for(tier, size_gib)?;
        let uuid = self.mint_uuid();
        let now = self.clock.now_ms();
        self.storages.insert(
            uuid.clone(),
            Storage {
                uuid: uuid.clone(),
                title: title.to_string(),
                size_gib,
                tier: tier.to_string(),
                zone: zone.to_string(),
                // Every storage is born in `maintenance`. It is `online` only
                // after `storage_create_ms`, and an attach before that is
                // refused — which is the poll loop's reason to exist.
                state: "maintenance".into(),
                kind: StorageKind::Normal,
                labels,
                origin: None,
                created_ms: now,
                import: None,
                transition: Some(Transition {
                    to: "online".into(),
                    at_ms: now + self.timings.storage_create_ms,
                    then: After::Created,
                }),
            },
        );
        Ok(uuid)
    }

    /// **Behaviour 22.** A volume never shrinks. A growth is permanent and
    /// billed forever, so the refusal is by NAME and not a silent clamp: a
    /// clamp would let a caller believe it had shrunk something.
    pub fn modify_storage(&mut self, uuid: &str, size_gib: Option<u64>, title: Option<&str>) -> Answer<()> {
        self.refuse_if_write_unavailable()?;
        let Some(cur) = self.storages.get(uuid).map(|s| s.size_gib) else {
            return Err(Refusal::new(404, "STORAGE_NOT_FOUND", format!("storage {uuid} not found")));
        };
        if let Some(n) = size_gib {
            // **Behaviour 54.** MEASURED 2026-09-14 on a throwaway volume:
            // smaller → `400 SIZE_INVALID` "The new size must be greater than
            // the old size." — and a SAME-size retry gets the same answer. The
            // mock said `STORAGE_INVALID_SIZE` and accepted the same size.
            if n <= cur {
                return Err(Refusal::new(
                    400,
                    "SIZE_INVALID",
                    format!("The new size must be greater than the old size. ({uuid} is {cur} GiB, {n} GiB was asked for)"),
                ));
            }
            // **Behaviour 56.** A grow under a RUNNING server is refused by
            // both of the docs' readings ("the server must be stopped" /
            // `STORAGE_ATTACHED` "must first be detached"); the stricter one —
            // any attachment at all — is `ResizeRequiresDetach`. The mock let
            // it through, and so did terraform through it: 5.44.1 sends this
            // PUT without stopping the server (BEHAVIOURS-LEDGER L101).
            let strict = self.faults.fires(Fault::ResizeRequiresDetach);
            for sv in self.attached_servers(uuid) {
                let state = self.servers.get(&sv).map(|x| x.state.clone()).unwrap_or_default();
                if strict || state != "stopped" {
                    return Err(Refusal::new(
                        409,
                        "STORAGE_ATTACHED",
                        format!("storage {uuid} is attached to server {sv} ({state}); it must first be detached (or the server stopped)"),
                    ));
                }
            }
            // **Behaviour 44.** The account's quota for the tier.
            self.refuse_over_quota(uuid, n - cur)?;
        }
        let now = self.clock.now_ms();
        let grow_ms = self.timings.storage_grow_ms;
        let s = self.storages.get_mut(uuid).expect("looked up above");
        if let Some(n) = size_gib {
            s.size_gib = n;
            // **Behaviour 55.** `maintenance` for 37 s, then `online` —
            // MEASURED on a detached 1→2 GiB grow. It was instant here.
            s.state = "maintenance".into();
            s.transition = Some(Transition { to: "online".into(), at_ms: now + grow_ms, then: After::GrowDone });
            // The record is the truth; the disk follows it now if no running
            // machine has it open, and at the next power-on otherwise.
            if let Err(why) = self.engine.resize_disk(uuid, n) {
                eprintln!("mock-upcloud  storage {uuid}: disk resize deferred: {why}");
            }
        }
        if let Some(t) = title {
            if let Some(s) = self.storages.get_mut(uuid) {
                s.title = t.to_string();
            }
        }
        // ★ **A device carries a COPY of its storage's size and title, and the
        // copy must move with the original.**
        //
        // MEASURED 2026-09-21: `terraform apply` grew holger-web's system disk
        // from 20 to 25 GB, the PUT succeeded, and the provider then refused its
        // own apply —
        //
        //   Provider produced inconsistent result after apply … .template[0].size:
        //   was cty.NumberIntVal(25), but now cty.NumberIntVal(20)
        //
        // — because it reads a machine's template size off the SERVER's
        // `storage_devices`, where this mock was still reporting the size the
        // device had when it was attached. The account cannot answer that way:
        // there is one volume and one size, and the device row is a view of it.
        // A grow that no read reports is a write that reports its own success
        // and did not happen, which is the class of defect this crate exists to
        // reproduce — not to have.
        let (size, title) = {
            let s = self.storages.get(uuid).expect("just modified");
            (s.size_gib, s.title.clone())
        };
        for sv in self.servers.values_mut() {
            for d in sv.devices.iter_mut().filter(|d| d.storage == uuid) {
                d.storage_size = size;
                d.storage_title = title.clone();
            }
        }
        Ok(())
    }

    pub fn delete_storage(&mut self, uuid: &str) -> Answer<()> {
        self.refuse_if_write_unavailable()?;
        let attached: Vec<String> = self
            .servers
            .values()
            .filter(|s| s.devices.iter().any(|d| d.storage == uuid))
            .map(|s| s.uuid.clone())
            .collect();
        if !attached.is_empty() {
            return Err(Refusal::new(
                409,
                "STORAGE_DEVICE_ATTACHED",
                format!("storage {uuid} is attached to {}", attached.join(", ")),
            ));
        }
        // **Behaviour 59.** A storage mid-operation, or the SOURCE of a clone
        // that is still being made, is refused `409 STORAGE_STATE_ILLEGAL` —
        // REPORTED (monetize-impl `tests.rs`; gunnar `api.rs`: "or a clone's
        // source while the clone exists"). The mock deleted both.
        if let Some(s) = self.storages.get(uuid) {
            if s.state == "maintenance" || s.state == "syncing" {
                return Err(Refusal::new(
                    409,
                    "STORAGE_STATE_ILLEGAL",
                    format!("storage {uuid} is {}; it cannot be deleted now", s.state),
                ));
            }
        }
        if let Some(c) = self.storages.values().find(|c| {
            c.kind == StorageKind::Normal && c.origin.as_deref() == Some(uuid) && c.transition.is_some()
        }) {
            return Err(Refusal::new(
                409,
                "STORAGE_STATE_ILLEGAL",
                format!("storage {uuid} is the source of clone {}, which is still {}", c.uuid, c.state),
            ));
        }
        let now = self.clock.now_ms();
        let ms = self.timings.storage_delete_ms;
        let Some(s) = self.storages.get_mut(uuid) else {
            return Err(Refusal::new(404, "STORAGE_NOT_FOUND", format!("storage {uuid} not found")));
        };
        s.state = "maintenance".into();
        s.transition = Some(Transition { to: "gone".into(), at_ms: now + ms, then: After::Vanish });
        Ok(())
    }

    pub fn attached_servers(&self, storage: &str) -> Vec<String> {
        self.servers
            .values()
            .filter(|s| s.devices.iter().any(|d| d.storage == storage))
            .map(|s| s.uuid.clone())
            .collect()
    }

    #[allow(clippy::too_many_arguments)]
    pub fn create_server(
        &mut self,
        title: &str,
        hostname: &str,
        plan: &str,
        zone: &str,
        labels: Vec<Label>,
        boot_disk_title: &str,
        boot_disk_gib: u64,
    ) -> Answer<String> {
        self.refuse_if_write_unavailable()?;
        // **Behaviour 37 — a sold-out zone refuses the CREATE, not only the
        // poweron.** Scaleway's `fr-par-1` (verified against Scaleway, not UpCloud;
        // kept as a generic cloud-provider fault) answered `412 out_of_stock` to `poweron` from
        // 2026-09-09 and had not cleared by 2026-09-20, and what is out of
        // stock there is CAPACITY FOR A PLAN IN A ZONE — which a create needs
        // exactly as much as a start does, and asks for first. The mock served
        // only the poweron door, so a caller that meets the outage at the
        // create could never have been driven against it: it would have got a
        // `201`, polled a server the provider never had, and reported the
        // timeout rather than the refusal.
        //
        // The SAME sticky [`Fault::OutOfStock`], deliberately, and not a second
        // variant. One zone is out of stock or it is not; a mock that could be
        // sold out at `poweron` and in stock at `create` would model a provider
        // nobody has measured, and it would let a caller pass by discovering
        // the shortage at whichever door it happened to knock on. If the two
        // ever ARE measured apart, that measurement is what earns the second
        // variant.
        //
        // Nothing is minted before this: the refusal leaves the account
        // untouched, so a caller that retries does not find half a server.
        if self.faults.fires(Fault::OutOfStock) {
            return Err(Refusal::new(
                412,
                "out_of_stock",
                "the zone has no capacity for this plan at the moment",
            ));
        }
        self.refuse_over_quota_for("maxiops", boot_disk_gib)?;
        let uuid = self.mint_uuid();
        let now = self.clock.now_ms();
        let disk_uuid = self.mint_uuid();
        self.storages.insert(
            disk_uuid.clone(),
            Storage {
                uuid: disk_uuid.clone(),
                title: boot_disk_title.to_string(),
                size_gib: boot_disk_gib,
                tier: "maxiops".into(),
                zone: zone.to_string(),
                state: "maintenance".into(),
                kind: StorageKind::Normal,
                labels: labels.clone(),
                origin: None,
                created_ms: now,
                import: None,
                transition: Some(Transition {
                    to: "online".into(),
                    at_ms: now + self.timings.storage_create_ms,
                    then: After::Nothing,
                }),
            },
        );
        // Behaviour 6: 98–105 s, drawn from the run's seed.
        let mut r = SplitMix64::derive(self.seed, &format!("create/{uuid}"));
        let ms = r.range(self.timings.server_create_lo_ms, self.timings.server_create_hi_ms);
        // Five figures, like `se-sto1.vnc.upcloud.com:60031`. It was 5900+ here
        // and 60000+ in `start_server`, which is the kind of disagreement that
        // lets a client key off the WRONG shape and pass.
        let port = 60_000 + (r.below(1000) as u16);
        // Decided ONCE, here, and never again: an appliance boots with a
        // userland that cannot say what the RTC holds, and a front boots with
        // one that can. Which of the two this server is, is a fact about the
        // image it was made from, not a coin flipped on every read.
        let rtc_local = self.faults.fires(Fault::GuestReadsRtcAsLocalTime);
        // Decided once, like the RTC interpretation and for the same reason: it
        // is a fact about the image this server was made from, not a coin
        // flipped per packet.
        let ignores_121 = self.faults.fires(Fault::GuestIgnoresDhcpOption121);
        let hotplug = !self.faults.fires(Fault::GuestKernelLacksHotplug);
        let public_ip = self.addresses.take_public();
        let utility_ip = self.addresses.take_utility();
        self.servers.insert(
            uuid.clone(),
            Server {
                uuid: uuid.clone(),
                title: title.to_string(),
                hostname: hostname.to_string(),
                plan: plan.to_string(),
                zone: zone.to_string(),
                state: "maintenance".into(),
                labels,
                devices: vec![Device {
                    address: "virtio:0".into(),
                    storage: disk_uuid,
                    storage_title: boot_disk_title.to_string(),
                    storage_size: boot_disk_gib,
                    kind: "disk",
                    boot_disk: true,
                }],
                boot_order: BootOrder::Disk,
                remote_access_enabled: false,
                remote_access_password: String::new(),
                vnc_port: port,
                reported_vnc_port: port,
                // `<zone>.vnc.upcloud.com` at the provider; RFC 2606 here, so
                // a mock never reports a host that resolves to anything real.
                // A real guest behind the server replaces it with its own
                // loopback VNC at boot.
                vnc_host: format!("{zone}.vnc.mock.invalid"),
                reported_vnc_host: format!("{zone}.vnc.mock.invalid"),
                public_ip,
                utility_ip,
                guest: Guest::Off,
                dhcp_client: if ignores_121 {
                    crate::net::DhcpClient::IgnoresOption121
                } else {
                    crate::net::DhcpClient::ReadsOption121
                },
                ssh_host_key: String::new(),
                rtc: if rtc_local {
                    crate::guest_clock::RtcInterpretation::LocalTime
                } else {
                    crate::guest_clock::RtcInterpretation::Utc
                },
                created_ms: now,
                transition: Some(Transition {
                    to: "started".into(),
                    at_ms: now + ms,
                    then: After::GuestBoots,
                }),
                ifaces: Iface::default_pair(),
                rules: vec![],
                firewall_on: false,
                metadata: true,
                timezone: "UTC".into(),
                hotplug,
                console_settles_at: None,
            },
        );
        Ok(uuid)
    }

    /// ★ **The terraform create's second half.** `create_server` mints the
    /// machine every caller in this crate shares; this says the things only the
    /// terraform door asks for, in ONE place, so the create path is not forked
    /// into two machines that drift.
    ///
    /// It is deliberately separate from `modify_server`: that one is the
    /// `PUT /1.3/server/{uuid}` a caller makes later and carries the
    /// stop/start behaviours. This is part of the create and makes no
    /// transition at all.
    pub fn configure_server(
        &mut self,
        uuid: &str,
        ifaces: Vec<Iface>,
        boot_order: Option<BootOrder>,
        firewall_on: bool,
        metadata: bool,
        timezone: &str,
    ) -> Answer<()> {
        let s = self
            .servers
            .get_mut(uuid)
            .ok_or_else(|| Refusal::new(404, "SERVER_NOT_FOUND", format!("server {uuid} not found")))?;
        if !ifaces.is_empty() {
            s.ifaces = ifaces;
        }
        if let Some(b) = boot_order {
            s.boot_order = b;
        }
        s.firewall_on = firewall_on;
        s.metadata = metadata;
        s.timezone = timezone.to_string();
        Ok(())
    }

    /// The rule set of a machine, or the refusal the firewall door owes. The
    /// 403s (behaviours 1 and 35) stay in [`crate::http`] where they are the
    /// ANSWER rather than the state.
    pub fn rules(&self, uuid: &str) -> Option<&[Rule]> {
        self.servers.get(uuid).map(|s| s.rules.as_slice())
    }

    /// Write a machine's whole rule set. UpCloud's firewall is a SET and not a
    /// list of independently addressable objects — a write replaces it — which
    /// is why `upcloud_firewall_rules` is one terraform resource per machine
    /// and not one per rule, and why this takes the whole vector.
    pub fn set_rules(&mut self, uuid: &str, rules: Vec<Rule>) -> Answer<()> {
        self.refuse_if_write_unavailable()?;
        let s = self
            .servers
            .get_mut(uuid)
            .ok_or_else(|| Refusal::new(404, "SERVER_NOT_FOUND", format!("server {uuid} not found")))?;
        s.rules = rules;
        // The positions the API hands back are its own, renumbered from 1 in
        // the order the set was written. A caller that sent none gets them
        // anyway, and a caller that sent its own does not get to keep gaps.
        for (i, r) in s.rules.iter_mut().enumerate() {
            r.position = (i + 1).to_string();
        }
        Ok(())
    }

    /// **Behaviour 11.** A server must be STOPPED before it can be deleted with
    /// its storages. A started one is refused, not queued.
    ///
    /// **Behaviour 7.** The delete itself sits in `maintenance` for 60 s plus
    /// 65 s per attached member volume — an appliance carrying four of them is
    /// over five minutes, which is what was measured and what every caller's
    /// timeout was too short for.
    pub fn delete_server(&mut self, uuid: &str, with_storages: bool) -> Answer<()> {
        self.refuse_if_write_unavailable()?;
        let Some(s) = self.servers.get(uuid) else {
            return Err(Refusal::new(404, "SERVER_NOT_FOUND", format!("server {uuid} not found")));
        };
        if s.state != "stopped" {
            return Err(Refusal::new(
                409,
                "SERVER_STATE_ILLEGAL",
                format!("server {uuid} is {} — stop it before deleting it", s.state),
            ));
        }
        let members = s.devices.iter().filter(|d| !d.boot_disk && d.kind == "disk").count() as u64;
        let ms = self.timings.server_delete_ms + members * self.timings.server_delete_per_volume_ms;
        let now = self.clock.now_ms();
        // The machine goes now; its disks go when their storages do.
        self.engine.forget_server(uuid);
        let s = self.servers.get_mut(uuid).expect("just had it");
        s.state = "maintenance".into();
        if !with_storages {
            s.devices.clear();
        }
        s.transition = Some(Transition { to: "gone".into(), at_ms: now + ms, then: After::Vanish });
        Ok(())
    }

    /// **Behaviour 3.** Out of stock, at poweron, with a `412`. It was measured
    /// on `fr-par-1` from 2026-09-09 and it did not clear for days, so the fault
    /// is sticky: a retry loop must not be able to outwait it.
    pub fn start_server(&mut self, uuid: &str) -> Answer<()> {
        self.refuse_if_write_unavailable()?;
        if self.faults.fires(Fault::OutOfStock) {
            return Err(Refusal::new(
                412,
                "out_of_stock",
                "the zone has no capacity for this plan at the moment",
            ));
        }
        let now = self.clock.now_ms();
        // An INSTALLER start (CD first, a medium in the tray) reads its brief
        // `started` at 3.6 s (receipt 2026-09-21); a disk boot at ~10 s.
        let installer = self.servers.get(uuid).is_some_and(|s| {
            s.boot_order.cdrom_first() && s.devices.iter().any(|d| d.kind == "cdrom" && !d.storage.is_empty())
        });
        let ms = if installer { self.timings.installer_started_ms } else { self.timings.start_ms };
        let stale = self.faults.fires(Fault::StaleVncPort);
        let mut r = SplitMix64::derive(self.seed, &format!("vnc/{uuid}/{now}"));
        // MEASURED: `se-sto1.vnc.upcloud.com:60031`. The port is five figures
        // and the host is the ZONE's console, not the server's own address, so
        // both halves are re-provisioned and neither is derivable from the
        // server.
        let fresh_port = 60_000 + (r.below(1000) as u16);
        let fresh_host = format!("{}.vnc.mock.invalid", self.zone);
        let Some(s) = self.servers.get_mut(uuid) else {
            return Err(Refusal::new(404, "SERVER_NOT_FOUND", format!("server {uuid} not found")));
        };
        if s.state == "started" {
            return Err(Refusal::new(409, "SERVER_STATE_ILLEGAL", "server is already started"));
        }
        // A start while an operation is in flight (a create, a stop, an
        // install pass, a delete) is refused, as every other illegal state is
        // (docs). It used to be accepted and to OVERWRITE the pending
        // transition — which is how a plan change lost its Resize Backup.
        if s.state == "maintenance" {
            return Err(Refusal::new(
                409,
                "SERVER_STATE_ILLEGAL",
                format!("server {uuid} is in maintenance; wait for it to settle before starting it"),
            ));
        }
        // Behaviour 14: the hypervisor binds a NEW port on every start, and the
        // API keeps reporting the old one until `remote_access_enabled` is
        // toggled. The divergence is the defect; the toggle is the cure, and
        // both are here so the cure can be tested.
        s.vnc_port = fresh_port;
        s.vnc_host = fresh_host.clone();
        if !stale {
            s.reported_vnc_port = fresh_port;
            s.reported_vnc_host = fresh_host;
        }
        s.state = "maintenance".into();
        s.transition = Some(Transition { to: "started".into(), at_ms: now + ms, then: After::GuestBoots });
        Ok(())
    }

    pub fn stop_server(&mut self, uuid: &str, hard: bool) -> Answer<()> {
        self.refuse_if_write_unavailable()?;
        let now = self.clock.now_ms();
        let ms = if hard { self.timings.stop_hard_ms } else { self.timings.stop_soft_ms };
        let Some(s) = self.servers.get_mut(uuid) else {
            return Err(Refusal::new(404, "SERVER_NOT_FOUND", format!("server {uuid} not found")));
        };
        if s.state == "stopped" {
            return Err(Refusal::new(409, "SERVER_STATE_ILLEGAL", "server is already stopped"));
        }
        // A wedged installer answers no ACPI event at all, so a SOFT stop of a
        // looping guest never completes — which is why the re-image uses a hard
        // stop once the media are on.
        if !hard && matches!(s.guest, Guest::Looping { .. } | Guest::Panicked) {
            return Err(Refusal::new(
                409,
                "SERVER_STATE_ILLEGAL",
                "the guest did not answer the shutdown request within the timeout",
            ));
        }
        s.state = "maintenance".into();
        s.guest = Guest::Off;
        s.transition = Some(Transition { to: "stopped".into(), at_ms: now + ms, then: After::Nothing });
        // Soft: the ACPI button, now. Hard: the kill, now. Either way the
        // transition to `stopped` above kills whatever is left.
        self.engine.power_off(uuid, hard);
        Ok(())
    }

    /// `PUT /1.3/server/{uuid}` — the plan change, the boot order, the labels
    /// and the `remote_access_enabled` toggle, which is the whole of what this
    /// estate uses it for.
    pub fn modify_server(
        &mut self,
        uuid: &str,
        plan: Option<&str>,
        boot_order: Option<BootOrder>,
        labels: Option<Vec<Label>>,
        remote_access: Option<bool>,
        remote_access_password: Option<&str>,
    ) -> Answer<()> {
        self.refuse_if_write_unavailable()?;
        let now = self.clock.now_ms();
        let settle_ms = self.timings.vnc_settle_ms;
        let Some(s) = self.servers.get_mut(uuid) else {
            return Err(Refusal::new(404, "SERVER_NOT_FOUND", format!("server {uuid} not found")));
        };
        // **Behaviour 51.** The console toggle answers 409 while the server is
        // in `maintenance` — MEASURED (memory: upcloud-vnc-port-toggle, "409
        // while state=maintenance (retry)"). Checked before anything changes.
        if remote_access.is_some() && s.state == "maintenance" {
            return Err(Refusal::new(
                409,
                "SERVER_STATE_ILLEGAL",
                format!("server {uuid} is in maintenance; remote access cannot be changed now"),
            ));
        }
        if let Some(bo) = boot_order {
            s.boot_order = bo;
        }
        if let Some(l) = labels {
            s.labels = l;
        }
        if let Some(on) = remote_access {
            // **The cure for behaviour 14.** Turning remote access off and on
            // again is what makes the API report the port the hypervisor is
            // really on. Nothing else does — not a stop/start, not a read.
            s.remote_access_enabled = on;
            if !on {
                // **Behaviour 50.** The "no" starts moving the console, and the
                // move takes ~2 s. The PUT has already answered by then.
                s.console_settles_at = Some(now + settle_ms);
            } else if s.console_settles_at.is_some_and(|t| now < t) {
                // A "yes" inside the window: the OLD endpoint comes back, and
                // stays until the next toggle — MEASURED, the reason every
                // working tool pauses between the two PUTs.
            } else {
                // BOTH halves. A cure that reconciled only the port would leave
                // a client dialling the right port at the wrong host.
                s.reported_vnc_port = s.vnc_port;
                s.reported_vnc_host = s.vnc_host.clone();
                s.console_settles_at = None;
            }
        }
        if let Some(p) = remote_access_password {
            s.remote_access_password = p.to_string();
        }
        // ★ **A PUT that names the SAME plan is not a plan change.**
        //
        // MEASURED 2026-09-21 with `--log`: `UpCloudLtd/upcloud` 5.44.1 grows a
        // system disk by sending the WHOLE server object, plan included and
        // unchanged. This mock read "a plan is present" as "the plan changed",
        // so an unrelated apply —
        //
        //   POST /server/{u}/stop        → maintenance
        //   GET  /server/{u}             → stopped        (the provider waited)
        //   PUT  /server/{u}             → maintenance    ← for a no-op
        //   PUT  /storage/{disk}         (the growth)
        //   POST /storage/{disk}/resize  → 409 SERVER_STATE_ILLEGAL
        //
        // — put the machine back into `maintenance` for a plan it already had,
        // refused the filesystem resize that followed, and minted a spurious
        // `Resize Backup` for a resize that never happened. Behaviour 9 is real
        // and stays; what was wrong was firing it for a change that is not one.
        // A mock stricter than the provider invents verdicts, and this one
        // invented a broken growth path AND the very leak it was written to
        // reproduce.
        let plan = plan.filter(|p| *p != s.plan);
        if let Some(p) = plan {
            if s.state != "stopped" {
                return Err(Refusal::new(
                    409,
                    "SERVER_STATE_ILLEGAL",
                    format!("the plan can only be changed while the server is stopped; it is {}", s.state),
                ));
            }
            let origin = s.boot_disk().map(|d| d.storage.clone()).unwrap_or_default();
            s.plan = p.to_string();
            // **Behaviour 48.** The plan change is done when the PUT answers:
            // the server stays `stopped`. MEASURED 2026-09-21 through the real
            // provider: 5.44.1 sends `start` straight after this PUT without
            // waiting, and it is applied against the live account, so the live
            // PUT leaves nothing to wait for. The mock used to hold the server
            // in `maintenance` for `resize_ms`; the provider's `start` then
            // overwrote that transition and the Resize Backup it carried was
            // never minted. Behaviour 9 (a plan change mints one) stays, but is
            // UNCONFIRMED: the measured 44 GB came from a FILESYSTEM resize.
            let _ = self.mint_resize_backup(&origin);
        }
        Ok(())
    }

    /// A server's `hostname` and `title`, changed in place.
    pub fn rename_server(&mut self, uuid: &str, hostname: Option<&str>, title: Option<&str>) {
        if let Some(s) = self.servers.get_mut(uuid) {
            if let Some(h) = hostname {
                s.hostname = h.to_string();
            }
            if let Some(t) = title {
                s.title = t.to_string();
            }
        }
    }

    pub fn attach(&mut self, server: &str, storage: &str, kind: &str) -> Answer<String> {
        self.attach_at(server, storage, kind, None)
    }

    /// **An attach, at the address the caller asked for (behaviour 58).**
    /// `virtio:3` is honoured; `virtio` (or nothing) lets UpCloud pick the
    /// first FREE slot — REPORTED (monetize-impl `grow.rs`: `address:"virtio"`
    /// lets UpCloud pick). The mock used to ignore the request and take
    /// `virtio:<count>`, which collides after a middle detach and never puts a
    /// disk where a caller asked. Where it lands decides the guest's name for
    /// it ([`Estate::guest_disk_names`]), and that decides the installer's
    /// disk election.
    pub fn attach_at(&mut self, server: &str, storage: &str, kind: &str, want: Option<&str>) -> Answer<String> {
        self.refuse_if_write_unavailable()?;
        let st = self
            .storages
            .get(storage)
            .ok_or_else(|| Refusal::new(404, "STORAGE_NOT_FOUND", format!("storage {storage} not found")))?;
        if st.state != "online" {
            return Err(Refusal::new(
                409,
                "STORAGE_STATE_ILLEGAL",
                format!("storage {storage} is {} — wait for online", st.state),
            ));
        }
        let (title, size) = (st.title.clone(), st.size_gib);
        // **Behaviour 57.** A storage another device already holds is refused
        // `409 STORAGE_ATTACHED` — REPORTED (monetize-impl treats it as a
        // retry; private-holger-ops `front.rs`: "UpCloud refuses to attach a
        // storage another server holds"). The mock attached it twice. A
        // cdrom with an empty tray holds nothing and does not count.
        if let Some(holder) = self
            .servers
            .values()
            .find(|s| s.devices.iter().any(|d| d.storage == storage && !storage.is_empty()))
        {
            return Err(Refusal::new(
                409,
                "STORAGE_ATTACHED",
                format!("storage {storage} is already attached to server {}", holder.uuid),
            ));
        }
        let s = self
            .servers
            .get_mut(server)
            .ok_or_else(|| Refusal::new(404, "SERVER_NOT_FOUND", format!("server {server} not found")))?;
        // **An attach is not the mirror of a detach, and the asymmetry is the
        // provider's.** A VIRTIO disk hot-plugs onto a RUNNING server and the
        // call succeeds — that is how a growth adds a member volume without a
        // reboot. An IDE cdrom does not, and is refused with the same
        // `IDE_HOTPLUG_UNSUPPORTED` a detach gets.
        //
        // This was wrong here first: the mock refused BOTH on a started server,
        // and the storm's very first run reported 86.67% of a hundred thousand
        // purchases failing with `409 SERVER_STATE_ILLEGAL` on a perfectly legal
        // hot-plug. A mock that is stricter than the provider manufactures
        // defects, which is worse than one that is laxer — a lax mock misses a
        // bug, a strict one invents eighty-six thousand.
        if s.state != "stopped" && kind == "cdrom" {
            return Err(Refusal::new(
                409,
                "IDE_HOTPLUG_UNSUPPORTED",
                format!("an ide cdrom cannot be attached while {server} is {}", s.state),
            ));
        }
        // **Behaviour 60.** …but only if the GUEST can take it. A kernel with
        // no PCI hot-plug never acks, and the call answers `511
        // HOTPLUG_FAILED` (MEASURED 2026-09-14, tunnr 6.12.104).
        if s.state == "started" && kind != "cdrom" && !s.hotplug {
            return Err(hotplug_failed(server));
        }
        let address = match (kind, want) {
            ("cdrom", Some(a)) if a.starts_with("ide:") => a.to_string(),
            ("cdrom", _) => "ide:0:0".to_string(),
            (_, Some(a)) if a.starts_with("virtio:") => a.to_string(),
            _ => {
                let used: Vec<u32> = s
                    .devices
                    .iter()
                    .filter_map(|d| d.address.strip_prefix("virtio:").and_then(|n| n.parse().ok()))
                    .collect();
                let n = (0..).find(|n| !used.contains(n)).expect("an unbounded range has a free slot");
                format!("virtio:{n}")
            }
        };
        if s.devices.iter().any(|d| d.address == address) {
            return Err(Refusal::new(409, "STORAGE_DEVICE_ADDRESS_IN_USE", format!("{address} is taken")));
        }
        s.devices.push(Device {
            address: address.clone(),
            storage: storage.to_string(),
            storage_title: title,
            storage_size: size,
            kind: if kind == "cdrom" { "cdrom" } else { "disk" },
            boot_disk: false,
        });
        Ok(address)
    }

    /// **The names the GUEST gives the disks (behaviour 58).** virtio-blk
    /// devices are named in PCI slot order and the names are CONTIGUOUS:
    /// `virtio:0, virtio:2, virtio:5` are `vda, vdb, vdc`. MEASURED as a rule
    /// the ladder depends on (gunnar `machine.rs`, holger HOLGER-PLAN: the ISO's
    /// virtio copy is attached AFTER the data volume "so that it becomes vdc,
    /// not vdb"), and it is what `korp-installer`'s disk election reads. The
    /// IDE cdrom is not a Linux block device here: the kernel has no ATA.
    pub fn guest_disk_names(&self, server: &str) -> Vec<(String, String)> {
        let Some(s) = self.servers.get(server) else { return vec![] };
        let mut slots: Vec<(u32, String)> = s
            .devices
            .iter()
            .filter_map(|d| d.address.strip_prefix("virtio:").and_then(|n| n.parse().ok()).map(|n| (n, d.address.clone())))
            .collect();
        slots.sort();
        slots
            .into_iter()
            .enumerate()
            .map(|(i, (_, a))| (a, format!("vd{}", (b'a' + i as u8) as char)))
            .collect()
    }

    /// **A device named in the CREATE body** (`storage_devices` entries with
    /// `"action": "attach"`). The server is in `maintenance` and has never run,
    /// so the cdrom's hot-plug refusal in [`Estate::attach`] does not apply:
    /// this is how the provider takes an installer medium at create. The storage
    /// must exist and be `online`, exactly as for a later attach.
    pub fn attach_at_create(&mut self, server: &str, storage: &str, kind: &str, want: Option<&str>) -> Answer<String> {
        let state = self.servers.get(server).map(|s| s.state.clone());
        if let Some(s) = self.servers.get_mut(server) {
            s.state = "stopped".into();
        }
        let r = self.attach_at(server, storage, kind, want);
        if let (Some(st), Some(s)) = (state, self.servers.get_mut(server)) {
            s.state = st;
        }
        r
    }

    /// **Behaviour 16, the half that refuses.** A detach names an ADDRESS, and
    /// on a STARTED server an ide address is refused `IDE_HOTPLUG_UNSUPPORTED`,
    /// and a virtio one is the guest's to allow (behaviour 60: a kernel without
    /// PCI hot-plug answers `511 HOTPLUG_FAILED`). The only thing that takes a medium off a running box
    /// is [`Estate::eject`].
    pub fn detach(&mut self, server: &str, address: &str) -> Answer<()> {
        self.refuse_if_write_unavailable()?;
        // Read before the server is borrowed mutably, as `start_server` reads
        // the stale-VNC decision: `fires` wants `&self`.
        let lies = self.faults.fires(Fault::DetachSaysSuccessButStaysAttached);
        let s = self
            .servers
            .get_mut(server)
            .ok_or_else(|| Refusal::new(404, "SERVER_NOT_FOUND", format!("server {server} not found")))?;
        // **A detach names an ADDRESS, and only an address.** MEASURED against
        // the live account on 2026-09-07 and written down in
        // `gunnar/deploy/upcloud/src/api.rs`: `{"storage_device": {"address":
        // "virtio:1"}}`. A body that carries a storage uuid instead is missing
        // the one attribute the endpoint takes.
        //
        // The refusal is explicit because a silent `404 no device at ""` reads
        // like "that disk is already gone" — which is exactly the wrong
        // conclusion for a caller trying to stop paying for it. (The status and
        // code here are this mock's reading of UpCloud's 1.3 error table for a
        // missing attribute; what is MEASURED is that a uuid-only detach does
        // not work, not which of its error codes comes back.)
        if address.is_empty() {
            return Err(Refusal::new(
                400,
                "MISSING_ATTRIBUTE",
                "storage_device.address is required: a detach names an address (virtio:1, ide:0:0), never a storage uuid",
            ));
        }
        let Some(i) = s.devices.iter().position(|d| d.address == address) else {
            return Err(Refusal::new(404, "STORAGE_DEVICE_NOT_FOUND", format!("no device at {address}")));
        };
        // **Behaviour 60 (was 16, and 16 had the wrong code).** On a STARTED
        // server an IDE detach is `409 IDE_HOTPLUG_UNSUPPORTED`, and a virtio
        // detach is the GUEST's to allow: a hot-plug kernel acks the eject
        // and it succeeds; one without PCI hot-plug never acks and the call is
        // `511 HOTPLUG_FAILED` — MEASURED 2026-09-14 on the live appliance.
        // This mock answered `409 VIRTIO_HOTPLUG_UNSUPPORTED`, a code nobody
        // measured, for every guest; monetize-impl catches only that 409, so
        // a real 511 skips its stop/detach/start fallback (ledger X4).
        if s.state == "started" {
            if address.starts_with("ide") {
                return Err(Refusal::new(
                    409,
                    "IDE_HOTPLUG_UNSUPPORTED",
                    format!("{address} cannot be detached while the server is running"),
                ));
            }
            if !s.hotplug {
                return Err(hotplug_failed(server));
            }
        }
        // **Behaviour 38 — the write that reports its own success and did not
        // happen.** Every refusal above still holds: the address must exist and
        // the server must not be running, because this is not "detach is
        // broken", it is "the detach that WOULD have worked answered 200 and
        // left the device on". A caller that reads the status and not the
        // server afterwards cannot tell this from the real thing, which is the
        // entire point — and until this path existed the mock had no way to be
        // wrong here at all, so every green a sweep printed against it was a
        // green about an instrument that could only say yes truthfully.
        if lies {
            return Ok(());
        }
        s.devices.remove(i);
        Ok(())
    }

    /// **Behaviour 16, the half that works.** `cdrom/eject` takes the MEDIUM out
    /// and leaves the device. MEASURED `200` on a STARTED server on 2026-09-14,
    /// and it ended an install loop: the next boot found no medium and fell
    /// through to the disk. It is the only never-loop primitive there is.
    pub fn eject(&mut self, server: &str) -> Answer<()> {
        self.refuse_if_write_unavailable()?;
        let s = self
            .servers
            .get_mut(server)
            .ok_or_else(|| Refusal::new(404, "SERVER_NOT_FOUND", format!("server {server} not found")))?;
        let Some(d) = s.devices.iter_mut().find(|d| d.kind == "cdrom") else {
            return Err(Refusal::new(404, "STORAGE_DEVICE_NOT_FOUND", "no cdrom device"));
        };
        d.storage = String::new();
        d.storage_title = String::new();
        d.storage_size = 0;
        if matches!(s.guest, Guest::Looping { .. }) {
            s.guest = Guest::Installed;
        }
        // The tray empties on the running machine too; the next power-on has
        // no medium and so no CD in its boot order.
        self.engine.eject(server);
        Ok(())
    }

    /// **`cdrom/load` — the other half of eject.** Puts a storage into the
    /// existing cdrom device of an existing server, so a re-image with a
    /// DIFFERENT medium needs no detach and no re-create. With the CD first in
    /// the boot order, the next hypervisor start boots it (L50).
    ///
    /// NOT MEASURED, and not in the ledger: the codes are UpCloud 1.3's
    /// documented ones for this endpoint. What is modelled: it works on a
    /// started server like eject does (L52); a tray that already holds a
    /// medium is refused (`CDROM_DEVICE_IN_USE`, eject first); a server with no
    /// cdrom device is refused (`CDROM_DEVICE_NOT_FOUND`); the storage must
    /// exist and be `online`. The provider also requires the storage to be of
    /// type `cdrom`; the mock does not, because its uploaded media are `normal`
    /// storages that the ladder attaches as `type: cdrom`, which is MEASURED to
    /// work (L49).
    pub fn load_cdrom(&mut self, server: &str, storage: &str) -> Answer<()> {
        self.refuse_if_write_unavailable()?;
        if storage.is_empty() {
            return Err(Refusal::new(400, "STORAGE_MISSING", "storage_device.storage is required: name the storage to load"));
        }
        if !self.servers.contains_key(server) {
            return Err(Refusal::new(404, "SERVER_NOT_FOUND", format!("server {server} not found")));
        }
        let st = self
            .storages
            .get(storage)
            .ok_or_else(|| Refusal::new(404, "STORAGE_NOT_FOUND", format!("storage {storage} not found")))?;
        if st.state != "online" {
            return Err(Refusal::new(
                409,
                "STORAGE_STATE_ILLEGAL",
                format!("storage {storage} is {} — wait for online", st.state),
            ));
        }
        let (title, size) = (st.title.clone(), st.size_gib);
        let s = self.servers.get_mut(server).expect("checked above");
        let Some(d) = s.devices.iter_mut().find(|d| d.kind == "cdrom") else {
            return Err(Refusal::new(404, "CDROM_DEVICE_NOT_FOUND", format!("server {server} has no cdrom device")));
        };
        if !d.storage.is_empty() {
            return Err(Refusal::new(
                409,
                "CDROM_DEVICE_IN_USE",
                format!("the cdrom of {server} already holds {}; eject it first", d.storage),
            ));
        }
        d.storage = storage.to_string();
        d.storage_title = title;
        d.storage_size = size;
        self.engine.load(server, storage);
        Ok(())
    }

    // ── the network ─────────────────────────────────────────────────────────

    /// **Can `from` reach `dest:port`, from where it is standing?**
    ///
    /// There is no such thing here as "is that address up" — only "is it up
    /// from here". Both of the asymmetries this models (a guest with no route
    /// off its prefix, and a box that cannot use its own DNAT) make the answer
    /// depend on the asker, and a reachability call that did not name one would
    /// be answering a question nobody has.
    pub fn reach(&self, from: &str, dest: &str, port: u16) -> Answer<crate::net::Reach> {
        let s = self
            .servers
            .get(from)
            .ok_or_else(|| Refusal::new(404, "SERVER_NOT_FOUND", format!("server {from} not found")))?;
        let listening = |addr: &str, _p: u16| {
            self.servers
                .values()
                .any(|x| (x.public_ip == addr || x.utility_ip == addr) && x.state == "started")
        };
        Ok(crate::net::outbound_reach(
            &s.utility_ip,
            &s.public_ip,
            &s.uuid,
            s.dhcp_client,
            dest,
            port,
            &self.dnat,
            listening,
        ))
    }

    /// **The three paths a host key must answer identically**, which is what
    /// tells a re-imaged machine from a hijacked name.
    ///
    /// A re-image mints a new key, so `REMOTE HOST IDENTIFICATION HAS CHANGED`
    /// is expected and must not be trusted blindly. The check that makes it
    /// safe is that the SAME key answers on the name, on the front's public
    /// address, and on the appliance's OWN public address bypassing the DNAT.
    /// [`Fault::HijackedName`] makes the name path answer a different key,
    /// which is the case the whole check exists for — and a verifier that only
    /// looked at one path would accept it.
    /// **Behaviour 62: can `from_ip` reach `to`:`port` INBOUND, through the
    /// provider's firewall?** `Dropped` is silence (the firewall), `Refused`
    /// is an RST (admitted, nothing listening — a server that is not
    /// `started`), `Ok` is a listener.
    pub fn inbound(&self, from_ip: &str, to: &str, proto: &str, port: u16) -> Answer<crate::net::Reach> {
        let s = self
            .servers
            .get(to)
            .ok_or_else(|| Refusal::new(404, "SERVER_NOT_FOUND", format!("server {to} not found")))?;
        let dest = s.public_ip.clone();
        if crate::net::firewall_admits(s.firewall_on, &s.rules, proto, from_ip, 0, port) == crate::net::Admit::Drop {
            return Ok(crate::net::Reach::Dropped { dest, port });
        }
        if s.state != "started" {
            return Ok(crate::net::Reach::Refused { dest, port });
        }
        Ok(crate::net::Reach::Ok)
    }

    /// **Behaviour 62, the UDP half: does the reply to this server's own
    /// outbound query come back?** The reply is an inbound packet FROM
    /// `from_ip:from_port` (53 for DNS, 123 for NTP) to an ephemeral port, and
    /// the firewall is stateless for UDP — so a rule set ending in a catch-all
    /// drop eats it, and a server with its firewall off gets it. This is the
    /// MECHANISM behind [`Fault::UdpInboundDropped`], which stays the
    /// estate-wide default-ON approximation for callers that ask no server.
    pub fn udp_reply_arrives(&self, server: &str, from_ip: &str, from_port: u16) -> Answer<bool> {
        let s = self
            .servers
            .get(server)
            .ok_or_else(|| Refusal::new(404, "SERVER_NOT_FOUND", format!("server {server} not found")))?;
        Ok(crate::net::firewall_admits(s.firewall_on, &s.rules, "udp", from_ip, from_port, 40_000) == crate::net::Admit::Accept)
    }

    pub fn host_key_via(&self, uuid: &str, path: HostKeyPath) -> Answer<String> {
        let s = self
            .servers
            .get(uuid)
            .ok_or_else(|| Refusal::new(404, "SERVER_NOT_FOUND", format!("server {uuid} not found")))?;
        if s.ssh_host_key.is_empty() {
            return Err(Refusal::new(409, "NO_HOST_KEY", format!("server {uuid} has never been installed")));
        }
        if path == HostKeyPath::Name && self.faults.fires(Fault::HijackedName) {
            // A different machine, answering on the name. The only thing that
            // catches it is another path disagreeing.
            return Ok(format!("SHA256:{:016x}{:016x}", 0xdeadbeefdeadbeefu64, 0xfeedfacefeedfaceu64));
        }
        Ok(s.ssh_host_key.clone())
    }

    /// A lay of the estate: every address is reshuffled. Nothing else changes —
    /// this is what a `xtask estate rebuild` does to the ADDRESSES, and the
    /// point is that the next lay's appliance is not on the last lay's IP.
    pub fn relay(&mut self) {
        let seed = self.seed;
        self.addresses.relay(seed);
    }

    pub fn lays(&self) -> u64 {
        self.addresses.lays()
    }

    // ── the direct-upload import, and the wait that follows it ──────────────

    /// `POST /1.3/storage/{uuid}/import` — open a direct-upload session.
    ///
    /// The session is `prepared` and nothing has moved yet. The URL it hands
    /// back is the mock's own, so a caller that follows the reply (rather than
    /// building the URL itself) reaches the mock without being told to.
    pub fn start_import(&mut self, uuid: &str, source: &str) -> Answer<Import> {
        self.refuse_if_write_unavailable()?;
        let now = self.clock.now_ms();
        let base = self.upload_base.clone();
        let zone = self.zone.clone();
        let Some(s) = self.storages.get_mut(uuid) else {
            return Err(Refusal::new(404, "STORAGE_NOT_FOUND", format!("storage {uuid} not found")));
        };
        // **Behaviour 66.** An import straight after `POST /storage`, before the
        // new storage has settled, is ACCEPTED: gunnar `deploy/upcloud`
        // `start_medium` (plan.rs:550-553) imports with no wait and no retry,
        // and the live re-image of 2026-09-21 08:42 (private-gunnar-ops
        // `.reimage/reinstall-receipt.json`: media-cdrom at 24.5 s,
        // media-virtio at 47.4 s, `done: true`) did it twice without a 409 —
        // as did every re-image since 2026-09-07 and the 2026-09-20 clone probe.
        // Whether the storage READ `maintenance` at that moment was never
        // recorded; that it was accepted was. Any other non-online state
        // (syncing, a grow, a delete) is still refused.
        // (T12, faithful-guest: and only while no import is open — a storage
        // busy with an import already is still refused.)
        let fresh = s.state == "maintenance" && s.import.is_none() && s.transition.as_ref().is_some_and(|t| t.then == After::Created);
        if s.state != "online" && !fresh {
            return Err(Refusal::new(
                409,
                "STORAGE_STATE_ILLEGAL",
                format!("storage {uuid} is {} — an import needs it online", s.state),
            ));
        }
        // The real one is `https://<zone>.img.upcloud.com/uploader/session/<uuid>`.
        // The mock's is its own socket with the same path, so the shape a caller
        // parses is the shape it will parse in production.
        let url = if base.is_empty() {
            // The provider's shape with an RFC 2606 host: the path a parser
            // must handle, and nothing a follower could reach.
            format!("https://{zone}.img.mock.invalid/uploader/session/{uuid}")
        } else {
            format!("{base}/uploader/session/{uuid}")
        };
        let im = Import {
            source: source.to_string(),
            state: "prepared".into(),
            created_ms: now,
            completed_ms: None,
            client_content_length: 0,
            read_bytes: 0,
            written_bytes: 0,
            md5sum: None,
            sha256sum: None,
            error_code: None,
            error_message: None,
            direct_upload_url: url,
        };
        s.import = Some(im.clone());
        Ok(im)
    }

    /// The `PUT` of the bytes to the session URL.
    ///
    /// `read_bytes` and `written_bytes` are the REAL count and the digests are
    /// the REAL digests — the ladder compares `sha256sum` with the local file's,
    /// so a made-up value here would make the verification pass without ever
    /// having run. Five seconds for a 43 MB ISO, and then the storage enters
    /// `syncing` for a hundred and something more.
    pub fn upload(&mut self, uuid: &str, bytes: &[u8]) -> Answer<Import> {
        self.refuse_if_write_unavailable()?;
        let now = self.clock.now_ms();
        let per_mib = self.timings.import_upload_ms_per_mib;
        let Some(s) = self.storages.get_mut(uuid) else {
            return Err(Refusal::new(404, "STORAGE_NOT_FOUND", format!("storage {uuid} not found")));
        };
        let Some(im) = s.import.as_mut() else {
            return Err(Refusal::new(404, "STORAGE_IMPORT_NOT_FOUND", format!("no import session on {uuid}")));
        };
        if im.state == "completed" || im.state == "failed" {
            return Err(Refusal::new(409, "STORAGE_IMPORT_ALREADY_DONE", "this session has already finished"));
        }
        let n = bytes.len() as u64;
        im.state = "uploading".into();
        im.client_content_length = n;
        im.read_bytes = n;
        im.written_bytes = n;
        im.md5sum = Some(crate::digest::md5_hex(bytes));
        im.sha256sum = Some(crate::digest::sha256_hex(bytes));
        let out = im.clone();
        // The bytes themselves, for an engine that will put them in a tray.
        if let Err(why) = self.engine.store_medium(uuid, bytes) {
            eprintln!("mock-upcloud  storage {uuid}: the engine could not keep the upload: {why}");
        }
        // The import grows the storage to fit the image it received — which is
        // why the seed volume only has to be LEGAL, not large.
        let gib = n.div_ceil(1024 * 1024 * 1024).max(1);
        s.size_gib = s.size_gib.max(gib);
        let upload_ms = (n / (1024 * 1024)).max(1) * per_mib;
        s.state = "maintenance".into();
        s.transition = Some(Transition { to: "syncing".into(), at_ms: now + upload_ms, then: After::BeginSync });
        Ok(out)
    }

    /// **`POST /1.3/storage/{uuid}/clone` — the candidate fix.**
    ///
    /// The ladder uploads the same 43 MB medium twice per re-image, once as a
    /// CD-ROM and once as a virtio disk, because `korp-installer` probes virtio
    /// and nothing else. The second one could be a clone of the first: the same
    /// bytes, the same sha256, no second upload.
    ///
    /// **Behaviour 53, MEASURED 2026-09-20** (gunnar `clone_probe.rs`): the call
    /// is quick, then `maintenance` → `online` in 47 s with NO `syncing`. This
    /// doc used to say the answer was not known; it was, one repository over.
    /// And the same probe says why it is still not the fix: a clone cannot
    /// start until its source is `online`, so import→clone is serial (161 s)
    /// where two parallel imports are ~115 s. [`Fault::CloneSyncsLikeImport`]
    /// keeps the old pessimistic guess, by name.
    pub fn clone_storage(&mut self, uuid: &str, title: &str) -> Answer<String> {
        self.refuse_if_write_unavailable()?;
        let now = self.clock.now_ms();
        let prepare = self.timings.clone_prepare_ms;
        let (lo, hi) = (self.timings.clone_sync_lo_ms, self.timings.clone_sync_hi_ms);
        let old_guess = self.faults.fires(Fault::CloneSyncsLikeImport);
        let online_ms = self.timings.clone_online_ms;
        let src = self
            .storages
            .get(uuid)
            .ok_or_else(|| Refusal::new(404, "STORAGE_NOT_FOUND", format!("storage {uuid} not found")))?;
        if src.state != "online" {
            return Err(Refusal::new(
                409,
                "STORAGE_STATE_ILLEGAL",
                format!("storage {uuid} is {} — a clone needs it online", src.state),
            ));
        }
        let (size, tier, zone, labels) = (src.size_gib, src.tier.clone(), src.zone.clone(), src.labels.clone());
        // A clone carries the SAME BYTES, so it carries the same digests. That
        // is the whole appeal: the ladder can verify the second medium against
        // the same local sha256 without uploading it again.
        let digests = src.import.as_ref().map(|i| (i.md5sum.clone(), i.sha256sum.clone(), i.read_bytes));
        let new = self.mint_uuid();
        let mut r = SplitMix64::derive(self.seed, &format!("clone/{new}"));
        let sync_ms = r.range(lo, hi);
        let import = digests.map(|(md5, sha, n)| Import {
            source: "storage".into(),
            state: "completed".into(),
            created_ms: now,
            completed_ms: Some(now),
            client_content_length: n,
            read_bytes: n,
            written_bytes: n,
            md5sum: md5,
            sha256sum: sha,
            error_code: None,
            error_message: None,
            direct_upload_url: String::new(),
        });
        let transition = if old_guess {
            Transition { to: "syncing".into(), at_ms: now + prepare, then: After::OnlineIn { ms: sync_ms } }
        } else {
            Transition { to: "online".into(), at_ms: now + online_ms, then: After::Nothing }
        };
        self.storages.insert(
            new.clone(),
            Storage {
                uuid: new.clone(),
                title: title.to_string(),
                size_gib: size,
                tier,
                zone,
                state: "maintenance".into(),
                kind: StorageKind::Normal,
                labels,
                origin: Some(uuid.to_string()),
                created_ms: now,
                import,
                transition: Some(transition),
            },
        );
        Ok(new)
    }

    /// **Behaviour 44: the account's storage quota.** MEASURED: yvra's
    /// `storage_maxiops` limit is 10 240 GiB (`/1.3/account`). The refusal's
    /// shape, `403 MAXIOPS_STORAGE_LIMIT_REACHED`, is REPORTED (monetize-impl
    /// `api.rs`), not measured. Only MaxIOPS is counted: the other tiers'
    /// limits were never read.
    fn refuse_over_quota(&self, uuid: &str, extra_gib: u64) -> Answer<()> {
        let tier = self.storages.get(uuid).map(|s| s.tier.clone()).unwrap_or_else(|| "maxiops".into());
        self.refuse_over_quota_for(&tier, extra_gib)
    }

    fn refuse_over_quota_for(&self, tier: &str, extra_gib: u64) -> Answer<()> {
        if tier != "maxiops" {
            return Ok(());
        }
        let used: u64 = self
            .storages
            .values()
            .filter(|s| s.tier == "maxiops" && s.kind != StorageKind::Template)
            .map(|s| s.size_gib)
            .sum();
        if used + extra_gib > self.maxiops_quota_gib {
            return Err(Refusal::new(
                403,
                "MAXIOPS_STORAGE_LIMIT_REACHED",
                format!(
                    "the account's MaxIOPS quota is {} GiB; {used} GiB is used and {extra_gib} GiB more was asked for",
                    self.maxiops_quota_gib
                ),
            ));
        }
        Ok(())
    }

    fn refuse_if_write_unavailable(&self) -> Answer<()> {
        if self.faults.fires(Fault::WriteUnavailable) {
            return Err(Refusal::new(503, "SERVICE_UNAVAILABLE", "the service is temporarily unavailable"));
        }
        Ok(())
    }

    /// Everything, for the mock's own `/mock/estate` view and for the tests.
    pub fn all_servers(&self) -> impl Iterator<Item = &Server> {
        self.servers.values()
    }
    pub fn all_storages(&self) -> impl Iterator<Item = &Storage> {
        self.storages.values()
    }
}

/// **`511 HOTPLUG_FAILED`** — the guest never acked the ACPI eject/insert.
/// MEASURED 2026-09-14 (memory: upcloud-reimage-install-loop,
/// tunnr-kernel-hotplug). Behaviour 60.
fn hotplug_failed(server: &str) -> Refusal {
    Refusal::new(
        511,
        "HOTPLUG_FAILED",
        format!("server {server}: the guest did not acknowledge the hot-plug request"),
    )
}

/// **The timing rule, proven against a machine the test controls.**
///
/// The API reports UpCloud's time; the VM runs at its own speed. So a modelled
/// state is HELD even when a real guest finishes early, and the API never says
/// `stopped` while the guest still runs. Behaviours 45 and 46 are the modelled
/// half (tests/ledger.rs); these are the reconciliation with a real guest.
#[cfg(test)]
mod timeline_tests {
    use super::*;
    use crate::kvm::{GuestEngine, GuestOutcome, GuestSpec, Machine};
    use std::sync::Mutex;

    /// A machine whose power the TEST holds: it runs from `power_on` until the
    /// test (the "guest") or the estate (a stop) turns it off.
    #[derive(Default)]
    struct Hand(Mutex<BTreeMap<String, bool>>);
    impl Hand {
        fn guest_powers_off(&self, s: &str) {
            self.0.lock().unwrap().insert(s.into(), false);
        }
    }
    impl GuestEngine for Hand {
        fn name(&self) -> &'static str {
            "hand"
        }
        fn boot(&self, _: &GuestSpec) -> GuestOutcome {
            unreachable!("the estate drives power_on, not boot")
        }
        fn power_on(&self, m: &Machine) -> Result<(), String> {
            self.0.lock().unwrap().insert(m.server.clone(), true);
            Ok(())
        }
        fn power_off(&self, s: &str, _hard: bool) {
            self.0.lock().unwrap().insert(s.into(), false);
        }
        fn running(&self, s: &str) -> Option<bool> {
            self.0.lock().unwrap().get(s).copied()
        }
    }

    fn rig() -> (Estate, Arc<Hand>) {
        let hand = Arc::new(Hand::default());
        let e = Estate::new(Clock::virtual_only(), Faults::quiet(), 7).with_engine(hand.clone());
        (e, hand)
    }

    /// A stopped server with an installer medium first in the boot order.
    fn an_installer(e: &mut Estate) -> String {
        let u = e.create_server("app", "app", "2xCPU-4GB", "se-sto1", vec![], "boot", 20).unwrap();
        e.run_to_quiet();
        e.stop_server(&u, true).unwrap();
        e.run_to_quiet();
        let iso = e.create_storage("iso", 1, "maxiops", "se-sto1", vec![]).unwrap();
        e.run_to_quiet();
        e.attach(&u, &iso, "cdrom").unwrap();
        e.modify_server(&u, None, Some(BootOrder::CdromDisk), None, None, None).unwrap();
        u
    }

    fn advance_to(e: &mut Estate, t: u64) {
        let now = e.clock.now_ms();
        if t > now {
            e.clock.advance_ms(t - now);
        }
        e.settle();
    }

    /// The VM installs in "2 s" and powers off; the API still reads
    /// `maintenance` until the measured 130–146 s pass is over.
    #[test]
    fn a_guest_that_finishes_early_does_not_make_the_api_early() {
        let (mut e, hand) = rig();
        let u = an_installer(&mut e);
        let t0 = e.clock.now_ms();
        e.start_server(&u).unwrap();
        { let t = e.timings; advance_to(&mut e, t0 + t.start_ms); }
        assert_eq!(hand.running(&u), Some(true), "the machine is up");
        { let t = e.timings; advance_to(&mut e, t0 + t.start_ms + 2_000); }
        hand.guest_powers_off(&u);
        { let t = e.timings; advance_to(&mut e, t0 + t.install_pass_lo_ms - 1); }
        assert_eq!(e.server(&u).unwrap().state, "maintenance", "the API reports UpCloud's pass, not the VM's 2 s");
        { let t = e.timings; advance_to(&mut e, t0 + t.install_pass_hi_ms); }
        assert_eq!(e.server(&u).unwrap().state, "stopped");
        assert_eq!(e.server(&u).unwrap().guest, Guest::Installed);
    }

    /// The VM outlives the modelled pass: the API does NOT say `stopped` over
    /// it, and says it within the notice once the machine is gone.
    #[test]
    fn the_api_never_says_stopped_while_the_vm_runs() {
        let (mut e, hand) = rig();
        let u = an_installer(&mut e);
        let t0 = e.clock.now_ms();
        e.start_server(&u).unwrap();
        // Step to the hypervisor start and through the brief `started` first:
        // a transition chains from the moment it is SETTLED, so one jump would
        // start the pass late.
        { let t = e.timings; advance_to(&mut e, t0 + t.installer_started_ms); }
        { let t = e.timings; advance_to(&mut e, t0 + t.installer_started_ms + t.installer_started_window_ms); }
        { let t = e.timings; advance_to(&mut e, t0 + t.install_pass_hi_ms + 60_000); }
        assert_eq!(hand.running(&u), Some(true));
        assert_eq!(e.server(&u).unwrap().state, "maintenance", "held: the VM still runs");
        let off = e.clock.now_ms();
        hand.guest_powers_off(&u);
        { let t = e.timings; advance_to(&mut e, off + t.poweroff_notice_ms); }
        assert_eq!(e.server(&u).unwrap().state, "stopped", "noticed within poweroff_notice_ms");
    }

    /// A disk-booted guest that powers itself off is noticed, not mirrored:
    /// `started` for the notice window, then `stopped` — never before the VM.
    #[test]
    fn a_self_powered_off_guest_is_noticed_after_the_notice_window() {
        let (mut e, hand) = rig();
        let u = e.create_server("front", "front", "2xCPU-4GB", "se-sto1", vec![], "boot", 20).unwrap();
        e.run_to_quiet();
        assert_eq!(e.server(&u).unwrap().state, "started");
        let off = e.clock.now_ms();
        hand.guest_powers_off(&u);
        e.settle();
        assert_eq!(e.server(&u).unwrap().state, "started", "the API runs behind the machine");
        { let t = e.timings; advance_to(&mut e, off + t.poweroff_notice_ms); }
        assert_eq!(e.server(&u).unwrap().state, "stopped");
        assert_eq!(e.server(&u).unwrap().guest, Guest::Off);
    }

    /// **An ejected tray is a disk boot, whatever the boot order says.** With
    /// `cdrom,disk` and a medium in the tray a start is an installer pass
    /// (`maintenance` until `stopped`, never `started`, 46); eject the medium,
    /// leave the order alone, and the same start reads `started` at ~10 s
    /// (L52, L54). The device row survives the eject, which is what fooled
    /// the old check: T3's TK run timed out on exactly this, with gRPC up.
    #[test]
    fn an_ejected_tray_boots_the_disk_even_with_cdrom_first() {
        let (mut e, hand) = rig();
        let iso = e.create_storage("medium", 1, "maxiops", "se-sto1", vec![]).unwrap();
        let u = e.create_server("appliance", "appliance", "2xCPU-4GB", "se-sto1", vec![], "boot", 20).unwrap();
        e.run_to_quiet();
        e.stop_server(&u, true).unwrap();
        e.run_to_quiet();
        e.attach(&u, &iso, "cdrom").unwrap();
        e.modify_server(&u, None, Some(BootOrder::Cdrom), None, None, None).unwrap();

        // Medium in, CD first: the installer's timeline — `started` only for
        // the brief window at 3.6 s (receipt 2026-09-21), then `maintenance`.
        let t0 = e.clock.now_ms();
        e.start_server(&u).unwrap();
        { let t = e.timings; advance_to(&mut e, t0 + t.installer_started_ms - 1); }
        assert_eq!(e.server(&u).unwrap().state, "maintenance");
        { let t = e.timings; advance_to(&mut e, t0 + t.installer_started_ms); }
        assert_eq!(e.server(&u).unwrap().guest, Guest::Installing);
        assert_eq!(e.server(&u).unwrap().state, "started", "the installer's brief started");
        { let t = e.timings; advance_to(&mut e, t0 + t.installer_started_ms + t.installer_started_window_ms); }
        assert_eq!(e.server(&u).unwrap().state, "maintenance", "the pass runs in maintenance");
        { let t = e.timings; advance_to(&mut e, t0 + t.start_ms + 1); }
        assert_eq!(e.server(&u).unwrap().state, "maintenance", "and does not read started again");
        hand.guest_powers_off(&u);
        e.run_to_quiet();
        assert_eq!(e.server(&u).unwrap().state, "stopped");

        // Eject; the order still says cdrom,disk and the device row stays.
        e.eject(&u).unwrap();
        assert!(e.server(&u).unwrap().devices.iter().any(|d| d.kind == "cdrom"), "the device survives an eject");
        assert!(e.server(&u).unwrap().boot_order.cdrom_first());
        let t1 = e.clock.now_ms();
        e.start_server(&u).unwrap();
        { let t = e.timings; advance_to(&mut e, t1 + t.start_ms - 1); }
        assert_eq!(e.server(&u).unwrap().state, "maintenance");
        { let t = e.timings; advance_to(&mut e, t1 + t.start_ms); }
        assert_eq!(e.server(&u).unwrap().state, "started", "an empty tray falls through to the disk");
        assert_ne!(e.server(&u).unwrap().guest, Guest::Installing);
    }

    /// The modelled floor for a disk boot is ~10 s, whatever the VM does.
    #[test]
    fn a_disk_boot_reads_started_at_the_modelled_ten_seconds() {
        let (mut e, _hand) = rig();
        let u = e.create_server("front", "front", "2xCPU-4GB", "se-sto1", vec![], "boot", 20).unwrap();
        e.run_to_quiet();
        e.stop_server(&u, true).unwrap();
        e.run_to_quiet();
        let t0 = e.clock.now_ms();
        e.start_server(&u).unwrap();
        { let t = e.timings; advance_to(&mut e, t0 + t.start_ms - 1); }
        assert_eq!(e.server(&u).unwrap().state, "maintenance");
        { let t = e.timings; advance_to(&mut e, t0 + t.start_ms); }
        assert_eq!(e.server(&u).unwrap().state, "started");
    }
}

#[cfg(test)]
mod address_tests {
    use super::*;
    use crate::{Clock, Faults};

    /// **The pool holds no literal outside TEST-NET.** A real address here
    /// once pointed the mock's ssh at a stranger's port 22.
    #[test]
    fn the_public_pool_is_test_net_and_nothing_else() {
        let a = Addresses::new(7);
        assert!(!a.public.is_empty());
        for ip in &a.public {
            assert!(is_test_net(ip), "public pool holds {ip}, which is not RFC 5737 TEST-NET");
        }
        for ip in &a.utility {
            assert!(is_utility_pool(ip), "utility pool holds {ip}");
            assert!(ip.starts_with("10."), "utility must be RFC 1918: {ip}");
        }
        // The checks themselves can say no.
        assert!(!is_test_net("94.237.30.222") && !is_test_net("81.27.106.241") && !is_test_net("192.0.3.1"));
        assert!(is_test_net("192.0.2.10") && is_test_net("198.51.100.17") && is_test_net("203.0.113.1"));
    }

    /// **The estate can NEVER hand out an address outside TEST-NET or its
    /// utility range** — over many servers, many lays and several seeds, and
    /// behaviour 12 still holds: the addresses move between lays.
    #[test]
    fn the_estate_never_hands_out_a_routable_public_address() {
        for seed in [0u64, 1, 41, 1234, 0xdead_beef] {
            let mut e = Estate::new(Clock::virtual_only(), Faults::none(), seed);
            let mut firsts = Vec::new();
            for lay in 0..6 {
                if lay > 0 {
                    e.relay();
                }
                for i in 0..30 {
                    let u = e.create_server(&format!("s{lay}-{i}"), "h", "1xCPU-1GB", "se-sto1", vec![], "boot", 1).unwrap();
                    let s = e.server(&u).unwrap();
                    assert!(is_test_net(&s.public_ip), "seed {seed}: public {} is routable", s.public_ip);
                    assert!(is_utility_pool(&s.utility_ip), "seed {seed}: utility {} is outside the pool", s.utility_ip);
                    if i == 0 {
                        firsts.push(s.public_ip.clone());
                    }
                }
            }
            firsts.dedup();
            assert!(firsts.len() > 1, "seed {seed}: the first address never moved across lays (behaviour 12)");
        }
    }
}