nucleic/Sources/NucleicCore/MacVM/MacVMEngine.swift

import Foundation
import os
#if arch(arm64)
import Virtualization
#endif

/// Engine diagnostics (`log stream --predicate 'subsystem == "com.nucleic" && category == "macvm"'`).
/// Suspend-to-disk save/restore failures MUST land here: both degrade silently by design (save falls
/// back to a RAM-pause, restore falls back to a cold boot), so without a log record a discarded
/// saved state is indistinguishable from a resume that never ran.
private let macvmLog = Logger(subsystem: "com.nucleic", category: "macvm")

/// In-process **macOS-guest VM runtime** built directly on Apple's `Virtualization` framework — the
/// Mac-native counterpart to ``ContainerEngine``. Where the container engine boots Linux guests via
/// the `containerization` framework, this actor boots real macOS guests: it owns a golden **base
/// bundle** (a provisioned macOS install with the dev toolchain + the native in-guest `agent`),
/// clones it copy-on-write per session, boots each clone on its own serial `VZVirtualMachine` queue,
/// and `exec`s commands inside **over vsock** (the native agent's streaming exec) so an agent can
/// build/test Mac targets in isolation — no SSH, no NAT, no network in the host↔guest control plane.
///
/// **Daemonless ⇒ ephemeral**, exactly like the container engine: no VM survives the app process, so
/// the `live` registry starts empty each launch and `reconcile` is on-disk GC. The per-session clone
/// (the CoW `Disk.img` + its aux storage) persists on disk, so an idle-stopped or app-restarted VM
/// reboots fast without re-installing macOS.
///
/// Lifecycle *policy* (busy ref-counting, idle timers, the concurrent-VM ceiling, per-session naming)
/// lives one layer up in ``MacVMManager``; this engine is the mechanism it drives. The only
/// entitlement required is `com.apple.security.virtualization` (already carried by the app and the
/// spike) — macOS guests use it too, and NAT networking needs no extra, restricted entitlement.
public actor MacVMEngine {
    /// One live macOS guest the engine is currently running.
    struct LiveVM {
        #if arch(arm64)
        let instance: MacVMInstance
        /// The native in-guest agent channel for this VM (docs/MACOS_VM_NATIVE_AGENT.md), probed
        /// lazily by ``MacVMEngine/agentClient(name:)`` and cached here for the VM's lifetime.
        var agent: MacVMAgentState = .unprobed
        /// The Linux **a11y (AT-SPI) semantic agent** channel (docs/LINUX_VM_SEMANTIC_AGENT.md) — a
        /// separate vsock port from the C `ping`/`exec` agent above, probed lazily by
        /// ``MacVMEngine/linuxA11yClient(name:)``. Only meaningful for Linux guests.
        var a11yAgent: MacVMAgentState = .unprobed
        #endif
        /// Which guest OS this VM is running — selects the exec shell and gates macOS-only computer-use
        /// paths (the semantic AX ops have no Linux equivalent).
        let os: GuestOS
        /// The clone bundle this VM booted from (its writable disk + aux storage live here).
        let bundle: MacVMBundle
        /// Locally-administered MAC assigned to this clone, used only to find its NAT DHCP lease for
        /// the settings-panel IP display (the control plane is vsock).
        let macAddress: String
        /// The guest account the native agent runs as (created by declarative provisioning).
        let sshUser: String
        /// The guest's NAT IP for display, once discovered — best-effort and not on any control path.
        var ipAddress: String?
        /// Configured memory ceiling in bytes, for the resource sample.
        let memoryCeilingBytes: UInt64
        /// When the VM was started (for a coarse CPU estimate / uptime).
        let startedAt: Date
        /// True once the boot has **fully completed** — the in-guest agent answered over vsock and the
        /// host paths were mounted (the tail of ``MacVMEngine/performBoot(_:)``). This is the real
        /// boot-readiness signal the Control panel keys "Booting…" vs. "running" on. ``ipAddress`` must
        /// NOT be used for it: the control plane is vsock-only now, so a fully-ready guest routinely has
        /// a `nil` ``ipAddress`` (no NAT DHCP lease) for its whole life — using the IP as the readiness
        /// signal is exactly what pinned VMs at "Booting…" forever after the vsock migration.
        var ready: Bool = false
        /// True when the agent is allowed to **drive** this VM (`mac_vm_computer`): the computer-use
        /// dispatch then routes HID injection through the host-side surface. Gates input only — capture
        /// is governed by ``hasSurface`` — so a VM with computer use *off* can still render in the
        /// monitor through a demand-activated surface but can't be driven.
        var computerUse: Bool = false
        /// True when this VM is registered with a host-side ``MacVMSurfaceHost``. Registration is
        /// available for every VM the app boots (independent of ``computerUse``), but the app-layer host
        /// materializes its `VZVirtualMachineView` only while capture/input/viewing is active. This keeps
        /// headless guests off the AppKit render path while preserving host-driver monitor capture.
        var hasSurface: Bool = false
        /// True while the guest is paused (frozen in RAM) via ``MacVMEngine/suspend(name:)`` — a
        /// `mac_vm_control`/`linux_vm_control` `suspend`. Reuse (``ensureRunning``) transparently thaws
        /// it, so a suspended VM is invisible to the exec/computer-use paths.
        var paused: Bool = false
        /// True only while a thaw is in flight — set as ``MacVMEngine/resume(name:)`` begins and cleared
        /// when it returns. `resume` awaits the VZ call (a suspension point on the engine actor), so a
        /// concurrent ``runningVMs()`` snapshot can observe this window and render "Resuming…".
        var resuming: Bool = false
        /// Whether this guest's configuration supports **suspend-to-disk**
        /// (`VZVirtualMachineConfiguration.validateSaveRestoreSupport()`). Computed once at boot. When
        /// `true`, ``MacVMEngine/suspend(name:)`` saves the guest to disk and frees its concurrency slot;
        /// when `false` it degrades to a plain RAM-pause (keeps the slot). Practically always `true` on
        /// the supported OS floor — the flag exists so an unexpected unsupported config never crashes.
        var canSaveState: Bool = false
    }

    /// A guest **suspended to disk**: its runtime state is saved in `bundle.savedStateURL`, its host RAM
    /// and its concurrency slot have been released (it is NOT in ``live``, so it doesn't count toward
    /// ``MacVMSettings/maxConcurrentVMs``), and the next ``resume(name:)`` / ``ensureRunning(_:)``
    /// reconstructs it from disk. This is the whole point of suspend-to-disk: a suspended VM frees a slot
    /// so another can boot in its place.
    struct DiskSuspendState {
        /// Guest OS, so the listing can label it and the restore rebuilds the right config.
        let os: GuestOS
        /// The spec needed to rebuild the config and restore the saved state on resume.
        let spec: MacVMSpec
        /// The clone bundle holding the writable disk + the saved-state file.
        let bundle: MacVMBundle
        /// Configured memory ceiling in bytes, carried for the resource listing.
        let memoryCeilingBytes: UInt64
    }

    /// On-disk root for engine artifacts: the cached restore image, the agent-account credential, the
    /// golden base bundle, and per-session clone bundles.
    let storageRoot: URL
    /// Live VMs keyed by logical name. Empty at launch (nothing survives the app process).
    var live: [String: LiveVM] = [:]
    /// Frozen screen (raw JPEG `Data`) grabbed the instant each VM was suspended, keyed by name. A paused
    /// guest's CPU is stopped, so a fresh capture would hang — the live monitor serves this still under
    /// its "Paused" overlay instead. Cleared when the VM resumes or stops. Stored as `Data` (the monitor's
    /// fast path wants raw bytes); the base64 ``captureScreen`` overload encodes on demand.
    private var pauseFrames: [String: Data] = [:]
    /// Guests **suspended to disk**, keyed by name (see ``DiskSuspendState``). A disk-suspended VM has
    /// released its host RAM and its concurrency slot — it is deliberately NOT in ``live`` — so this map
    /// is what keeps it visible to the UI (``runningVMs()``) and resumable (``resume(name:)`` /
    /// ``ensureRunning(_:)`` restore from `bundle.savedStateURL`). Its frozen monitor still lives in
    /// ``pauseFrames``. Empty at launch; the on-disk `savedStateURL` is the durable source of truth, so a
    /// clone whose save file survives an app restart is still restored on its next boot.
    private var diskSuspended: [String: DiskSuspendState] = [:]
    /// In-flight boots keyed by name, so concurrent `ensureRunning` for the SAME name join the one
    /// boot instead of racing into a second `VZVirtualMachine` on the same writable disk. Also counted
    /// toward the concurrency ceiling so a burst of distinct-name boots can't overshoot it.
    private var boots: [String: Task<(name: String, ipAddress: String), Error>] = [:]
    /// Last spec seen per name, so `restart` (which only has a name) can recreate from disk.
    var lastSpec: [String: MacVMSpec] = [:]

    /// A small rolling tail of each running guest's most recent `mac_vm_exec` output (stdout, stderr, and
    /// the `$ command` echoes), keyed by name. Fed line-by-line from ``run(name:command:workdir:env:)`` as
    /// a command streams, so the live monitor can show what a headless-busy guest is doing under its static
    /// screen. Bounded to the last ``maxBackgroundLogLines`` lines; cleared when the VM stops. This is a
    /// display convenience, not the command result — the full (capped) output still returns to the tool.
    private var backgroundLogLines: [String: [MacVMLogLine]] = [:]
    /// Per-VM monotonic line counter backing ``MacVMLogLine/id``, so ids stay stable as the window rolls
    /// (dropping the oldest lines never renumbers the survivors).
    private var backgroundLogSeq: [String: Int] = [:]
    /// How many lines of background output to retain per guest — enough for a useful scrolling tail
    /// without unbounded growth on a chatty build.
    private static let maxBackgroundLogLines = 300
    /// Hard cap on a single retained line's length, so one pathological multi-MB "line" (e.g. a base64
    /// blob echoed with no newlines) can't balloon the buffer. Overflow is clipped with an ellipsis.
    private static let maxBackgroundLogLineLength = 2000

    /// Admission/eviction policy for the concurrent-VM ceiling, injected by the app layer (`AppStore` —
    /// the only layer that knows chat state). Given the names of the VMs currently occupying slots, it
    /// returns those whose owning chat is *done* (no turn in flight — the brief idle-hold window a chat
    /// keeps its VM after completing), ordered stalest-interaction first. When a boot would exceed the
    /// ceiling the engine reclaims the first returned VM to grant the slot; an empty result means
    /// "nothing safe to evict — queue instead". `nil` in headless/spike contexts, where the old
    /// throw-on-full behavior stands (no eviction, no queueing). See ``ensureRunning(_:)``.
    private var evictionRanker: (@Sendable ([String]) async -> [String])?

    /// Boots parked on the admission queue because the ceiling is full and no done chat's VM could be
    /// reclaimed. Each is resumed `true` the moment a slot frees (a guest stopped, was reclaimed, or a
    /// boot failed), or `false` when its own wait deadline elapses. Keyed so a timeout resumes exactly
    /// its own waiter and a normal wake can drain them all. See ``awaitSlotOrTimeout(_:)``.
    private var slotWaiters: [UUID: CheckedContinuation<Bool, Never>] = [:]

    #if arch(arm64)
    /// Probe state of a live VM's native in-guest agent (the vsock channel). `unavailable` is
    /// sticky for the VM's lifetime — a base image without the agent will never grow one mid-boot,
    /// so re-probing every action would just pay the connect timeout repeatedly. A transport
    /// failure on a previously `available` channel resets to `unprobed` instead (the LaunchAgent's
    /// `KeepAlive` restarts a crashed agent, so one reconnect attempt is worth it).
    enum MacVMAgentState {
        case unprobed
        case unavailable
        case available(MacVMAgentClient)
    }
    #endif

    /// Held while at least one VM is live, to keep the host app from being App-Napped / auto-terminated
    /// out from under a running guest (mirrors ``ContainerEngine``'s assertion; the runtime is
    /// daemonless so the app process bounds every VM's life). `nil` when nothing is live.
    private var backgroundActivity: (any NSObjectProtocol)?

    /// Progress of the one-time base-image build (download / install / provision), surfaced to the VM
    /// settings panel. `nil` in the steady state (base already built, nothing installing). Mutated only
    /// on the actor; read via ``currentBaseProgress()``.
    var baseProgress: MacVMBaseProgress?
    /// True while a base-image build is in flight, so a second ``buildBaseImage`` call can't remove the
    /// shared `base.building` temp bundle out from under the running installer. Actor-guarded.
    var baseBuilding = false
    /// Which guest OS the in-flight base build is for (`nil` when no build is running). The settings
    /// panel reads this (via ``currentBaseBuildGuest()``) to route the shared ``baseProgress`` to the
    /// right section and to keep showing the "Starting…" state after the user navigates out of and back
    /// into Settings — which resets the view-local build flags but not this actor state. Set alongside
    /// ``baseBuilding`` at each build's start and cleared in its `defer`.
    var baseBuildGuest: GuestOS?
    /// True while the base is booted writable for a **Recovery** session (`csrutil disable`), set via
    /// ``beginBaseRecovery()``/``endBaseRecovery()``. Both flags block cutting a fresh session clone
    /// from the base, since cloning reads the base disk that would then be mid-write.
    var baseRecoveryActive = false
    /// True while the base is booted writable to **inject user apps** into its `/Applications`
    /// (``installAppsIntoBase(appPaths:)`` — the "add an app without a full rebuild" path). Like the
    /// build/Recovery flags it must block cutting a fresh clone while the base disk is mid-write.
    var baseAppInstalling = false
    /// Whether the operator asked to watch the base-build VM in a diagnostic monitor window. Applied
    /// when the provisioning surface attaches, and toggled live via ``setBaseProvisionObserver(visible:)``.
    var baseObserverRequested = false

    /// The system-managed base-image VM currently booted for maintenance (injecting user apps without
    /// a full rebuild), surfaced to the Control panel's VM section so the running base is visible
    /// alongside the per-session guests. `nil` in the steady state. Actor-guarded; read via
    /// ``baseMaintenanceVM()``.
    var baseMaintenance: MacVMMaintenanceInfo?

    /// The base disk is being written (a build/provisioning pass, a Recovery session, or an app
    /// injection), so it must not be read for a copy-on-write clone right now (a torn read would
    /// corrupt the clone).
    var baseIsBusy: Bool { baseBuilding || baseRecoveryActive || baseAppInstalling }

    /// Host-side computer-use surface (VZVirtualMachineView framebuffer capture + HID injection),
    /// injected by the app layer at startup. `nil` in headless/spike contexts. Bound to computer-use
    /// VMs (main-queue) for the agent, and to the base-build VM to drive the HID provisioning bootstrap.
    var surfaceHost: (any MacVMSurfaceHost)?

    /// Wire the app-layer host-IO surface (called once at startup, like the manager injection).
    public func setSurfaceHost(_ host: any MacVMSurfaceHost) { surfaceHost = host }

    /// Wire the admission/eviction policy (called once at startup, like ``setSurfaceHost(_:)``). Without
    /// it the ceiling stays a hard throw; with it, a boot at the ceiling reclaims the stalest *done*
    /// chat's VM, or queues when every slot is a busy chat. See ``evictionRanker`` and ``ensureRunning(_:)``.
    public func setEvictionRanker(_ ranker: @escaping @Sendable ([String]) async -> [String]) {
        evictionRanker = ranker
    }

    public init(storageRoot: URL? = nil) {
        self.storageRoot = storageRoot ?? Self.defaultStorageRoot()
    }

    // MARK: - Capability

    /// Whether the in-process macOS-guest runtime can run on this machine. `true` iff
    /// ``unsupportedReason`` is `nil`.
    public static var isSupported: Bool { unsupportedReason == nil }

    /// A specific, actionable reason the macOS-VM runtime is unavailable, or `nil` when supported.
    /// The hard gate is the CPU architecture: Apple's Mac-guest APIs (`VZMac*`) exist only on Apple
    /// silicon — an Intel host cannot create a macOS guest at all. The virtualization *entitlement* is
    /// undetectable here (a missing one surfaces as a `startFailed` when the first VM boots); the
    /// deployment floor (macOS 26) is enforced by the build itself.
    public static var unsupportedReason: String? {
        #if arch(arm64)
        return nil
        #else
        return "Running macOS guests requires Apple silicon."
        #endif
    }

    static func defaultStorageRoot() -> URL {
        let base = (try? FileManager.default.url(
            for: .applicationSupportDirectory, in: .userDomainMask,
            appropriateFor: nil, create: true))
            ?? URL(fileURLWithPath: NSTemporaryDirectory())
        return base.appendingPathComponent("Nucleic/macvms", isDirectory: true)
    }

    // MARK: - On-disk layout

    /// Directory holding per-session clone bundles.
    var instancesDir: URL { storageRoot.appendingPathComponent("instances", isDirectory: true) }
    /// The golden base bundle the engine builds when the user hasn't supplied a prebuilt one.
    var builtBaseDir: URL { storageRoot.appendingPathComponent("base", isDirectory: true) }
    /// The golden **Linux** base bundle (kernel + initrd + provisioned ext4 rootfs).
    var builtLinuxBaseDir: URL { storageRoot.appendingPathComponent("linux-base", isDirectory: true) }
    /// Cached Linux boot artifacts (downloaded kernel `Image`, initrd, Ubuntu Base rootfs tarball),
    /// version-keyed by source filename like the restore-image cache.
    var linuxArtifactsDir: URL { storageRoot.appendingPathComponent("linux-artifacts", isDirectory: true) }
    /// Cached restore images (`.ipsw`).
    var restoreDir: URL { storageRoot.appendingPathComponent("restore", isDirectory: true) }
    /// Small persisted credentials the base build needs (currently just the generated `agent`-account
    /// password used to prime sudo during provisioning). No SSH keys — the control plane is vsock.
    var credentialsDir: URL { storageRoot.appendingPathComponent("credentials", isDirectory: true) }

    /// Per-session clone bundle for `name`.
    func instanceBundle(for name: String) -> MacVMBundle {
        MacVMBundle(root: instancesDir.appendingPathComponent(name, isDirectory: true))
    }

    // MARK: - Base-image progress

    /// The one-time base-image build currently in flight, for the VM settings panel. `nil` once the
    /// golden base exists. A trivial property read (no I/O).
    public func currentBaseProgress() -> MacVMBaseProgress? { baseProgress }

    /// The guest OS of the base build currently in flight, or `nil` when none is. Lets the settings
    /// panel re-derive which section a shared ``baseProgress`` belongs to after its view-local state was
    /// torn down (navigating out of and back into Settings).
    public func currentBaseBuildGuest() -> GuestOS? { baseBuildGuest }

    // MARK: - Lifecycle

    /// Ensure the named macOS VM is running and return its name plus the guest's SSH-reachable IP.
    /// Idempotent: a VM already live is returned as-is; a boot already in flight for this name is
    /// *joined* (not duplicated); otherwise the golden base is cloned copy-on-write, booted on its own
    /// queue, and awaited until SSH answers.
    ///
    /// The whole dedup + concurrency-ceiling check runs synchronously on this actor before the first
    /// `await`, so it is atomic: two concurrent calls for the same name can't double-boot the same
    /// writable disk, and a burst of distinct-name calls can't blow past ``MacVMSettings/maxConcurrentVMs``.
    @discardableResult
    public func ensureRunning(_ spec: MacVMSpec) async throws -> (name: String, ipAddress: String) {
        lastSpec[spec.name] = spec
        // Reuse keys on `ready` — NOT on `ipAddress`. The control plane is vsock-only, so a fully
        // usable guest routinely has no NAT lease for its whole life; gating reuse on the IP would
        // send every follow-up call for a live VM into `performBoot`, double-booting its own disk.
        if let existing = live[spec.name], existing.ready {
            // Transparently thaw a VM the agent suspended (`*_vm_control suspend`) before handing it
            // back, so exec/computer-use never has to know it was frozen.
            if existing.paused { _ = await resume(name: spec.name) }
            return (spec.name, existing.ipAddress ?? "")
        }
        // Join an in-flight boot for the same name rather than starting a second one.
        if let inFlight = boots[spec.name] {
            return try await inFlight.value
        }
        guard Self.isSupported else {
            throw MacVMError.unavailable(Self.unsupportedReason ?? "unsupported")
        }
        // Admit the boot past the concurrent-guest ceiling. macOS caps simultaneous guests; overshooting
        // would fail an opaque `VZVirtualMachine.start`, so a boot at the ceiling must free a slot first.
        // Policy (injected via `evictionRanker`): reclaim the stalest *done* chat's VM — it's in its
        // post-completion idle-hold window anyway — and only if none can be reclaimed park in the
        // admission queue until a slot frees or the wait deadline elapses. The ceiling test below and the
        // slot reservation that follows it straddle no `await`, so they stay atomic per call: two callers
        // can't both read a free slot and both reserve it, and a parked waiter re-tests before reserving.
        //
        // With no admission policy wired (headless/spike embedders), keep the original behavior: a boot at
        // the ceiling throws immediately rather than evicting or queueing — there is no chat state to rank
        // by, and nothing to reclaim.
        let limit = MacVMSettings.maxConcurrentVMs
        if evictionRanker == nil, live.count + boots.count >= limit {
            throw MacVMError.concurrencyLimit(limit)
        }
        let deadline = Date().addingTimeInterval(TimeInterval(MacVMSettings.vmQueueTimeoutSeconds))
        while live.count + boots.count >= limit {
            // Another call may have booted or joined this exact VM while we were parked — re-honor the
            // dedup/reuse fast paths before consuming a slot for a duplicate of a VM that now exists.
            // Same rule as the fast path above: reuse keys on `ready`, never on the NAT IP.
            if let existing = live[spec.name], existing.ready {
                if existing.paused { _ = await resume(name: spec.name) }
                return (spec.name, existing.ipAddress ?? "")
            }
            if let inFlight = boots[spec.name] { return try await inFlight.value }
            // Reclaim the stalest done chat's VM if the policy offers one. Prefer **suspend-to-disk**:
            // the evicted chat's full live state is saved, so it resumes exactly where it left off rather
            // than cold-rebooting. That only frees the slot when the save succeeds, so if the victim's
            // config can't save (rare) or the save fails (it stays RAM-paused, still in `live`), cut its
            // power to guarantee the slot we're queued for actually frees — an evicted done chat's runtime
            // state isn't retained past this point, so there's nothing worth a grace period. Either way
            // the clone survives on disk, and a wedged victim can't stall the boot we're queued behind.
            if let victim = await pickEvictable(excluding: spec.name) {
                if live[victim]?.canSaveState == true { _ = await suspend(name: victim) }
                if live[victim] != nil { await stop(name: victim) }
                continue
            }
            // Nothing reclaimable — every occupied slot is an actively-working chat. Queue until one
            // frees, giving up with the ceiling error once the bounded wait is exhausted so a tool call
            // can't hang forever behind busy peers.
            if Date() >= deadline { throw MacVMError.concurrencyLimit(limit) }
            _ = await awaitSlotOrTimeout(deadline.timeIntervalSinceNow)
        }
        // Reserve the slot by publishing the boot task synchronously, then await it. The task inherits
        // this actor's isolation, so `performBoot` runs serialized on the actor.
        let task = Task { try await self.performBoot(spec) }
        boots[spec.name] = task
        defer { boots[spec.name] = nil }
        return try await task.value
    }

    /// The actual boot: clone the golden base, build the VZ config, start the guest on its own queue,
    /// wire the guest-stop hook, and wait until SSH answers. Only reached via ``ensureRunning``'s
    /// dedup/ceiling gate, so it never double-boots a name or overshoots the ceiling.
    private func performBoot(_ spec: MacVMSpec) async throws -> (name: String, ipAddress: String) {
        #if arch(arm64)
        // Resolve (building if necessary) the golden base for this guest OS, then clone it.
        let base = try await ensureBaseBundle(for: spec.os)
        let bundle = instanceBundle(for: spec.name)
        let mac: String
        if bundle.isComplete(for: spec.os) {
            // Reuse an existing clone (idle-restart / relaunch): its identity + MAC persist on disk.
            // This touches only the clone's own disk, so it's safe even while the base is busy.
            mac = (try? String(contentsOf: bundle.macAddressURL, encoding: .utf8))?
                .trimmingCharacters(in: .whitespacesAndNewlines) ?? Self.randomMAC()
        } else {
            // Cutting a fresh clone READS the base disk — refuse while it's booted writable (a build/
            // provisioning pass or a Recovery session), so the clone can't capture a torn state.
            guard !baseIsBusy else {
                throw MacVMError.baseBusy(
                    "the base image is being provisioned or edited — retry once it's done")
            }
            mac = try cloneBase(base, to: bundle, os: spec.os)
        }

        let config: VZVirtualMachineConfiguration
        do {
            config = try Self.makeConfiguration(
                bundle: bundle, os: spec.os, cpus: spec.cpus, memoryGiB: spec.memoryGiB, mac: mac,
                mounts: spec.mounts)
        } catch {
            throw MacVMError.startFailed("configuration: \(error)")
        }

        // Whether this guest can be suspended to disk (frees its RAM + slot). Probed once here from the
        // config; practically always true on the supported OS floor. Falsy only degrades suspend to a
        // RAM-pause — it never fails the boot.
        let canSaveState = (try? config.validateSaveRestoreSupport()) != nil

        // The VM always boots on its own dedicated background queue (see `MacVMInstance.init`); the
        // main thread is never used for guest I/O, even for displayed VMs. A host-side surface capability
        // is registered for every VM whenever a surface host is present, so capture/HID can materialize
        // a `VZVirtualMachineView` on demand. Registration itself creates no AppKit objects; the concrete
        // host keeps headless VMs off the main-thread render path.
        let willRegisterSurface = surfaceHost != nil
        let instance = MacVMInstance(configuration: config, label: spec.name)
        do {
            // Resume-from-disk vs cold boot: if a prior suspend-to-disk left a saved-state file in the
            // clone, reconstruct the guest from it (byte-for-byte RAM + device state) instead of booting
            // fresh. The save is consumed on success — the writable disk diverges the instant the guest
            // runs again, so the old RAM image would be inconsistent with it. Any restore error (a stale
            // file from a previous app/config, corruption) falls back to a clean cold boot from the
            // clone: exactly today's stop→reboot, just losing the in-RAM state we couldn't restore.
            if FileManager.default.fileExists(atPath: bundle.savedStateURL.path) {
                do {
                    try await instance.restore(from: bundle.savedStateURL)
                    try? FileManager.default.removeItem(at: bundle.savedStateURL)
                    // Restore leaves the guest paused; run it. A rare resume failure leaves it paused, so
                    // the `awaitAgentReady` gate below times out and tears it down — no special-casing.
                    _ = await instance.resume()
                    macvmLog.info("restored \(spec.name, privacy: .public) from saved state")
                } catch {
                    // Restore is config-exact: VZ rejects a saved state whose configuration differs at
                    // all from the restoring one (observed: VZErrorDomain code 12 "invalid argument"
                    // when the CPU-count setting changed between suspend and resume). The cold boot
                    // below is the right fallback, but it discards the guest's entire runtime state —
                    // never let that happen silently.
                    macvmLog.error(
                        "restore \(spec.name, privacy: .public) from saved state failed — discarding it and cold-booting: \(String(describing: error), privacy: .public)"
                    )
                    try? FileManager.default.removeItem(at: bundle.savedStateURL)
                    try await instance.start()
                }
            } else {
                try await instance.start()
            }
        } catch {
            throw MacVMError.startFailed(String(describing: error))
        }

        let memoryBytes = UInt64(spec.memoryGiB) * 1024 * 1024 * 1024
        live[spec.name] = LiveVM(
            instance: instance, os: spec.os, bundle: bundle, macAddress: mac, sshUser: spec.sshUser,
            ipAddress: nil, memoryCeilingBytes: memoryBytes, startedAt: Date(),
            computerUse: spec.computerUse, hasSurface: willRegisterSurface, canSaveState: canSaveState)
        // The guest is live again — it's no longer disk-suspended, and its frozen still is stale.
        diskSuspended[spec.name] = nil
        pauseFrames[spec.name] = nil
        syncBackgroundActivity()

        // Register the VM with the host-side surface capability. The app host retains the VM reference
        // without constructing a view/window until a monitor, computer-use action, or operator viewer
        // actually needs it; capture remains host-driver based with no in-guest software, TCC, or SIP.
        if let host = surfaceHost {
            await host.attach(
                name: spec.name, virtualMachine: UncheckedSendableBox(value: instance.vm as AnyObject))
        }
        // A guest that powers itself off (in-guest `shutdown`, panic, OOM) must not linger in `live`
        // — that would leak a concurrency slot and pin the app-activity assertion. Clear it when the
        // VZ delegate reports the stop. The hook hops back onto this actor.
        let name = spec.name
        instance.setOnStop { [weak self] in
            Task { await self?.handleGuestStopped(name) }
        }

        // Wait until the in-guest agent answers over vsock — the readiness gate. First boot of a fresh
        // clone can take a couple of minutes (login window + launchd + the agent's LaunchAgent); reuse
        // is quicker.
        do {
            try await awaitAgentReady(name: spec.name)
            // Re-create the repo's original host path inside the guest (a direct virtiofs mount on
            // macOS, bind-mount on Linux) so absolute build paths resolve. A macOS mount is required
            // readiness: without it there is intentionally no `/Volumes` fallback.
            try await recreateHostPaths(spec)
            // Best-effort NAT IP for the settings-panel display only; never gates readiness.
            let ip = Self.leaseIP(forMAC: mac) ?? ""
            live[spec.name]?.ipAddress = ip.isEmpty ? nil : ip
            // The guest is now fully usable (agent answered + host paths mounted). Flip the readiness
            // flag the UI keys "Booting…" → "running" on — decoupled from the (possibly nil) NAT IP.
            live[spec.name]?.ready = true
            return (spec.name, ip)
        } catch {
            // The VM came up but the agent never answered — tear it back down so a retry starts clean.
            await instance.stop()
            live[spec.name] = nil
            syncBackgroundActivity()
            signalSlotFreed()  // the reserved slot is freed by this failure — wake a queued boot
            throw error
        }
        #else
        throw MacVMError.unavailable(Self.unsupportedReason ?? "requires Apple silicon")
        #endif
    }

    /// Drop a VM the guest stopped on its own (VZ delegate → ``MacVMInstance/setOnStop(_:)``), so a
    /// crashed/powered-off guest doesn't keep consuming a concurrency slot or hold the app-activity
    /// assertion. Idempotent: a no-op if the engine already tore it down.
    private func handleGuestStopped(_ name: String) {
        guard let entry = live[name] else { return }
        live[name] = nil
        pauseFrames[name] = nil
        syncBackgroundActivity()
        signalSlotFreed()
        if entry.hasSurface { Task { await surfaceHost?.detach(name: name) } }
        #if arch(arm64)
        if case .available(let client) = entry.agent {
            Task { await client.close() }
        }
        #endif
    }

    // MARK: - Admission queue (concurrent-VM ceiling)

    /// Resume every boot parked on the admission queue: a slot just freed (a guest stopped, was
    /// reclaimed, or a boot failed), so each parked caller re-tests the ceiling and either reserves the
    /// slot or re-parks. Waking all rather than one is harmless at the small VM ceiling and avoids a
    /// lost wakeup when the freed slot is immediately re-taken — a needlessly-woken waiter just re-parks.
    private func signalSlotFreed() {
        guard !slotWaiters.isEmpty else { return }
        let waiters = slotWaiters
        slotWaiters.removeAll()
        for cont in waiters.values { cont.resume(returning: true) }
    }

    /// Park the caller on the admission queue until a slot frees (resumed `true` by ``signalSlotFreed()``)
    /// or `timeout` seconds elapse (resumed `false`). Actor-isolated: the continuation is stored and
    /// later resumed on this actor. A detached timer resumes the still-parked waiter at the deadline; a
    /// waiter already woken by a freed slot is a no-op for that timer (it was removed on wake).
    private func awaitSlotOrTimeout(_ timeout: TimeInterval) async -> Bool {
        let id = UUID()
        return await withCheckedContinuation { (cont: CheckedContinuation<Bool, Never>) in
            slotWaiters[id] = cont
            Task { [weak self] in
                try? await Task.sleep(nanoseconds: UInt64(max(0, timeout) * 1_000_000_000))
                await self?.expireSlotWaiter(id)
            }
        }
    }

    /// Deadline hook for ``awaitSlotOrTimeout(_:)``: resume the still-parked waiter `id` with `false`.
    /// A no-op if it was already woken (and removed) by a freed slot.
    private func expireSlotWaiter(_ id: UUID) {
        slotWaiters.removeValue(forKey: id)?.resume(returning: false)
    }

    /// Pick the stalest *done* chat's VM to reclaim for a boot that would otherwise exceed the ceiling,
    /// excluding the requesting session's own VM. Delegates the chat-state ranking to the injected
    /// ``evictionRanker`` (only the app layer knows which chats are done and how recently each was used);
    /// returns `nil` when no policy is wired or nothing is safe to evict (so the caller queues instead).
    private func pickEvictable(excluding requester: String) async -> String? {
        guard let evictionRanker else { return nil }
        let occupying = live.keys.filter { $0 != requester }
        guard !occupying.isEmpty else { return nil }
        return await evictionRanker(Array(occupying)).first
    }

    /// Run `argv` inside the named guest **over vsock** (the in-guest agent's streaming `exec` op),
    /// returning a ``ProcessHandle`` whose stdio is the remote command's stdio. The agent's mac-VM
    /// tool treats it identically to a host or containerized process — see ``MacVMExecChannel``.
    public func exec(
        name: String, workdir: String?, env: [String: String], argv: [String]
    ) async throws -> any ProcessHandle {
        let body = argv.map(Self.shQuote).joined(separator: " ")
        return try await execHandle(name: name, workdir: workdir, env: env, remoteBody: body)
    }

    /// Convenience one-shot used by the `mac_vm_exec` tool: run a shell `command` inside the guest and
    /// return its exit code plus captured stdout/stderr — the macOS analogue of `runOnHost`.
    ///
    /// `tap` controls whether the command and its output mirror into the guest's background-output tail
    /// (the live monitor's "Background output" console). Only the user-facing tool path opts in; internal
    /// housekeeping runs (the resource probe's `vm_stat`/`top`, computer-use cursor probes, MDM, host-path
    /// setup) leave it `false` so the console stays a faithful mirror of `mac_vm_exec` — not a dumping
    /// ground for the polling probes that would otherwise drown out the user's own output.
    public func run(
        name: String, command: String, workdir: String?, env: [String: String] = [:],
        tap: Bool = false
    ) async throws -> (exitCode: Int32, stdout: String, stderr: String) {
        let handle = try await execHandle(name: name, workdir: workdir, env: env, remoteBody: command)
        guard tap else {
            // Untapped (internal) run: just drain both streams concurrently, no background-log echo.
            async let out = Self.collect(handle.stdoutLines)
            async let err = Self.collect(handle.stderrLines)
            let stdout = await out
            let stderr = await err
            let code = await handle.wait()
            return (code, stdout, stderr)
        }
        // Echo the command into the guest's background-output tail, then drain both streams concurrently
        // — tapping each line into that same tail as it arrives so the live monitor can show the running
        // command's output under its static screen. Concurrent drain avoids deadlocking on a full stderr
        // pipe while we read stdout (or vice versa); the exit is awaited after.
        appendBackgroundLog(name: name, kind: .command, text: Self.commandMarker(command))
        async let out = collectTapping(handle.stdoutLines, name: name, kind: .stdout)
        async let err = collectTapping(handle.stderrLines, name: name, kind: .stderr)
        let stdout = await out
        let stderr = await err
        let code = await handle.wait()
        return (code, stdout, stderr)
    }

    /// Run a command in the guest over SSH and return its stdout up to `maxBytes`, plus whether it was
    /// truncated — for a base64 payload (a screenshot) that ``run``'s 64 KB cap would mangle but that
    /// still needs a ceiling so a pathological capture can't blow up host memory / the model context.
    /// A truncated base64 is unusable, so the caller discards it. Ignores exit code; stderr discarded.
    func rawStdout(
        name: String, command: String, maxBytes: Int = 8 * 1024 * 1024
    ) async throws -> (text: String, truncated: Bool) {
        let handle = try await execHandle(name: name, workdir: nil, env: [:], remoteBody: command)
        let result = await Self.collectBounded(handle.stdoutLines, cap: maxBytes)
        _ = await handle.wait()
        return result
    }

    /// Drain a line stream into one string up to `cap` bytes, reporting whether the cap was hit. Unlike
    /// ``collect`` this appends NO truncation marker (which would corrupt a base64 payload) — the caller
    /// decides what a truncated result means.
    private static func collectBounded(
        _ stream: AsyncThrowingStream<Data, Error>, cap: Int
    ) async -> (text: String, truncated: Bool) {
        var lines: [String] = []
        var bytes = 0
        var truncated = false
        do {
            for try await line in stream {
                if bytes < cap {
                    lines.append(String(decoding: line, as: UTF8.self))
                    bytes += line.count + 1
                } else {
                    truncated = true
                }
            }
        } catch { /* stream ended on error — return what we have */ }
        return (lines.joined(separator: "\n"), truncated)
    }

    /// Force-stop the guest (freeing its host RAM) but keep the on-disk clone bundle so the next
    /// `ensureRunning` reboots it without re-installing. Drops the registry entry; best-effort.
    ///
    /// Cuts virtual power without letting the guest quiesce. That's deliberate — session guests are
    /// disposable and nothing in their runtime state is retained past a stop, so no grace period is
    /// spent asking them to halt themselves. (Only base-image maintenance/provisioning still shuts a
    /// guest down cleanly, via `shutdownBaseGuest`, because the base disk it quiesces is re-cloned.)
    public func stop(name: String) async {
        // A disk-suspended VM holds no running guest and no slot — "stopping" it just discards its saved
        // state so the clone cold-boots next time (matching the stop→reboot contract). Its slot was
        // already freed when it suspended, so there's none to signal here.
        if let suspended = diskSuspended.removeValue(forKey: name) {
            try? FileManager.default.removeItem(at: suspended.bundle.savedStateURL)
            pauseFrames[name] = nil
            clearBackgroundLog(name: name)
            return
        }
        guard let entry = live[name] else { return }
        if entry.hasSurface { await surfaceHost?.detach(name: name) }
        #if arch(arm64)
        if case .available(let client) = entry.agent { await client.close() }
        await entry.instance.stop()
        #endif
        live[name] = nil
        pauseFrames[name] = nil
        clearBackgroundLog(name: name)
        syncBackgroundActivity()
        signalSlotFreed()  // a slot just freed — wake a boot parked on the admission queue
    }

    /// Suspend the named running guest, **freeing its concurrency slot** so another VM can boot in its
    /// place. It pauses the guest, saves its full runtime state to disk (`bundle.savedStateURL`), then
    /// tears the guest down to release its host RAM and its slot; ``resume(name:)`` (or the next
    /// ``ensureRunning(_:)``) reconstructs it from that saved state. When the guest's config can't be
    /// saved (rare on the supported OS floor) or the save fails, it degrades to a plain **RAM-pause**:
    /// state preserved and thaw is instant, but — like the old behavior — the slot stays occupied.
    /// Returns `true` once suspended (either mode); `false` when the VM isn't running or couldn't pause.
    ///
    /// The saved state has **no TTL**: it persists until the restore consumes it, an explicit
    /// `stop`/`remove` discards it, or launch reconcile GCs the dead session's whole clone. Suspend's
    /// contract to the agent is "resume picks up exactly where you left off", and a suspended guest
    /// already holds no RAM and no slot — expiring the file early (an earlier design reaped it after
    /// `idleTimeout`) silently turned every resume after that TTL into a cold boot that lost all
    /// in-guest state, indistinguishable from "resume is broken".
    @discardableResult
    public func suspend(name: String) async -> Bool {
        #if arch(arm64)
        guard let entry = live[name] else { return false }
        // Grab the guest's current screen *before* freezing it: once paused its CPU is stopped, so a
        // live capture would hang — the monitor needs this still to show under the "Suspended" overlay.
        // Best-effort; a miss just leaves the monitor on its last good frame. Skip if already paused
        // (keep the frame we froze on the way in the first time).
        if !entry.paused, let frame = await captureScreenData(name: name) {
            pauseFrames[name] = frame
        }
        // Pause first — required for BOTH suspend-to-disk (VZ demands `.paused` to save) and the
        // RAM-pause fallback. Skip if already paused (a prior RAM-pause we're now upgrading to disk).
        if !entry.paused {
            guard await entry.instance.pause() else { return false }
            live[name]?.paused = true
        }

        // No disk save possible (unsupported config, or we lost the spec needed to restore) → leave the
        // guest RAM-paused. State is preserved; the slot is not freed. This is the safety-net path.
        guard entry.canSaveState, let spec = lastSpec[name] else { return true }

        // Suspend-to-disk: persist the runtime state, then power the (paused) guest off to reclaim its
        // RAM and — crucially — its slot. A save failure keeps the guest RAM-paused instead.
        do {
            try await entry.instance.save(to: entry.bundle.savedStateURL)
        } catch {
            macvmLog.error(
                "suspend \(name, privacy: .public): saving state to disk failed — leaving the guest RAM-paused (slot NOT freed): \(String(describing: error), privacy: .public)"
            )
            try? FileManager.default.removeItem(at: entry.bundle.savedStateURL)
            return true
        }
        // Saved. This is `stop(name:)`'s teardown, minus clearing the frozen frame (the monitor keeps
        // showing it under the "Suspended" overlay), and moving the VM into `diskSuspended` (still
        // listed, still resumable) rather than dropping it entirely. Move it out of `live` and into
        // `diskSuspended` BEFORE powering off, so the guest-stop hook that fires on power-off sees it's
        // no longer live and no-ops (both guard on `live[name]`) instead of racing this teardown and
        // wiping the frozen frame.
        if entry.hasSurface { await surfaceHost?.detach(name: name) }
        if case .available(let client) = entry.agent { await client.close() }
        diskSuspended[name] = DiskSuspendState(
            os: entry.os, spec: spec, bundle: entry.bundle,
            memoryCeilingBytes: entry.memoryCeilingBytes)
        live[name] = nil
        await entry.instance.stop()
        macvmLog.info("suspended \(name, privacy: .public) to disk (slot freed)")
        syncBackgroundActivity()
        signalSlotFreed()  // the slot this VM held just freed — wake a boot parked on the admission queue
        return true
        #else
        return false
        #endif
    }

    /// Resume a suspended guest. A **disk-suspended** VM is reconstructed from its saved state by
    /// re-acquiring a concurrency slot through the normal admission gate (``ensureRunning`` restores via
    /// ``performBoot``'s restore branch) — heavier than a RAM thaw since it reads the whole guest RAM
    /// back from disk, but that's the cost of having freed the slot. A **RAM-paused** VM (the fallback
    /// mode) thaws in place, instantly. Returns `true` once resumed; `false` when the VM isn't suspended.
    @discardableResult
    public func resume(name: String) async -> Bool {
        #if arch(arm64)
        if let suspended = diskSuspended[name] {
            return (try? await ensureRunning(suspended.spec)) != nil
        }
        guard let entry = live[name], entry.paused else { return false }
        // Mark the thaw in flight so a snapshot taken during the (awaited) VZ resume shows "Resuming…"
        // rather than the frozen guest's stale meters. Cleared unconditionally once the call returns.
        live[name]?.resuming = true
        let ok = await entry.instance.resume()
        live[name]?.resuming = false
        if ok {
            live[name]?.paused = false
            pauseFrames[name] = nil  // guest is live again; drop the frozen still
        }
        return ok
        #else
        return false
        #endif
    }

    /// Whether the named VM is currently suspended — either RAM-paused (still in ``live``) or
    /// suspended-to-disk (freed its slot). `false` when it isn't suspended.
    public func isPaused(_ name: String) async -> Bool {
        if diskSuspended[name] != nil { return true }
        return live[name]?.paused ?? false
    }

    /// Stop then re-boot the guest in place (reuses the clone; reclaims the host RAM a long-running VM
    /// holds). No-op if the engine never saw this VM. Cuts power rather than asking the guest to halt —
    /// nothing in the guest's runtime state is retained, and the reboot fscks the clone anyway.
    public func restart(name: String) async {
        guard let spec = lastSpec[name] else { await stop(name: name); return }
        await stop(name: name)
        _ = try? await ensureRunning(spec)
    }

    /// Stop the guest and delete its on-disk clone bundle. Returns `true` once it's gone (including
    /// when it never existed). Best-effort.
    @discardableResult
    public func remove(name: String) async -> Bool {
        await stop(name: name)
        lastSpec[name] = nil
        let root = instanceBundle(for: name).root
        try? FileManager.default.removeItem(at: root)
        return !FileManager.default.fileExists(atPath: root.path)
    }

    /// True if the named VM is currently running in this process.
    public func isRunning(_ name: String) async -> Bool { live[name] != nil }

    /// True if the VM is running, or has a persisted clone bundle on disk (stopped but re-bootable).
    /// OS-agnostic: a clone directory holds either a macOS or a Linux guest, so accept either shape.
    public func vmExists(_ name: String) async -> Bool {
        if live[name] != nil { return true }
        let bundle = instanceBundle(for: name)
        return bundle.isComplete(for: .macOS) || bundle.isComplete(for: .linux)
    }

    /// Names of all macOS VMs currently running in this process.
    public func list() async -> [String] { Array(live.keys) }

    /// Running VMs with their discovered IPs, for the settings panel and the Control panel. Carries
    /// each guest's OS so the Control panel's Virtual Machines section can label it macOS vs. Linux.
    /// Includes **disk-suspended** guests: they hold no slot (deliberately not in ``live``) but must
    /// still be listed as suspended so the user sees them and can resume/remove them.
    public func runningVMs() async -> [MacVMEntry] {
        var entries = live.map { name, vm in
            MacVMEntry(
                name: name, ipAddress: vm.ipAddress, os: vm.os,
                paused: vm.paused, resuming: vm.resuming,
                // Still coming up until the agent answered + host paths mounted. Keyed on the real
                // readiness flag, NOT `ipAddress`, which the vsock control plane no longer populates.
                booting: !vm.ready)
        }
        for (name, s) in diskSuspended {
            entries.append(
                MacVMEntry(
                    name: name, ipAddress: nil, os: s.os,
                    paused: true, resuming: false, suspendedToDisk: true))
        }
        return entries
    }

    /// Number of macOS guests currently occupying a concurrency slot — the manager consults this against
    /// the configured ceiling before booting another (macOS caps simultaneous macOS guests).
    /// Disk-suspended guests are excluded (they freed their slot), which is the whole point.
    public func runningCount() async -> Int { live.count }

    /// The system-managed base-image VM booted for maintenance right now (app injection), or `nil`
    /// when the base isn't running for that. Kept separate from ``runningVMs()`` — that list drives
    /// the settings "Running now" / "Copy to running VMs" actions, which target per-session guests,
    /// not this transient base boot.
    public func baseMaintenanceVM() async -> MacVMMaintenanceInfo? { baseMaintenance }

    /// Best-effort resource sample for the VM panel. The Virtualization framework exposes no CPU or
    /// memory figures for a macOS guest, so we probe from *inside* it over SSH: `vm_stat` for memory
    /// (instantaneous and reliable) and a two-sample `top` for CPU (the first `top` sample is
    /// meaningless, hence `-l 2`). Both ride one round-trip. Memory total is the configured ceiling
    /// (= `hw.memsize` in the guest), so it needs no probe. Returns a ceiling-only sample (used/CPU
    /// zero) when the guest isn't reachable yet, and `nil` when the VM isn't running at all.
    ///
    /// This is called off the Control panel's critical path (a background refresh that caches the
    /// result), never inline — the `top -l 2` read alone sleeps ~1 s in the guest.
    public func sampleResourceUsage(name: String) async -> MacVMResourceSample? {
        guard let entry = live[name] else { return nil }
        let total = entry.memoryCeilingBytes
        // Still booting → the vsock probe below has nothing to talk to yet; report the ceiling only.
        // Gate on the real readiness flag, not `ipAddress`: the probe runs over vsock (`run`), not SSH,
        // so it never needs a NAT IP — which a vsock-ready guest routinely lacks.
        guard entry.ready else {
            return MacVMResourceSample(cpuPercent: 0, memoryUsedBytes: 0, memoryTotalBytes: total)
        }
        let probe = "vm_stat; echo __CPU__; top -l 2 -n 0 | awk '/CPU usage/{c=$0} END{print c}'"
        let stdout = (try? await run(name: name, command: probe, workdir: nil))?.stdout ?? ""
        return Self.parseResourceProbe(stdout, memoryTotal: total)
    }

    /// Parse the guest resource probe (`vm_stat` + a `top` "CPU usage" line) into a sample. Used
    /// memory is `(active + wired + compressed) × page-size`; CPU is `100 − idle%` from `top`. Pure
    /// and total-driven so it's unit-testable without a live guest — a blank/garbled probe yields a
    /// ceiling-only sample (zeros) rather than throwing.
    static func parseResourceProbe(_ output: String, memoryTotal: UInt64) -> MacVMResourceSample {
        var pageSize: UInt64 = 4096
        var active: UInt64 = 0, wired: UInt64 = 0, compressed: UInt64 = 0
        var cpuPercent = 0.0

        for raw in output.split(separator: "\n") {
            let line = raw.trimmingCharacters(in: .whitespaces)
            // `vm_stat` header carries the page size, e.g. "… (page size of 16384 bytes)".
            if line.contains("page size of"), let n = Self.trailingUInt(in: line) { pageSize = n }
            else if line.hasPrefix("Pages active:"), let n = Self.trailingUInt(in: line) { active = n }
            else if line.hasPrefix("Pages wired down:"), let n = Self.trailingUInt(in: line) { wired = n }
            else if line.hasPrefix("Pages occupied by compressor:"),
                let n = Self.trailingUInt(in: line) { compressed = n }
            else if line.contains("CPU usage"), let idle = Self.idlePercent(in: line) {
                cpuPercent = max(0, 100 - idle)
            }
        }

        let used = (active &+ wired &+ compressed) &* pageSize
        return MacVMResourceSample(
            cpuPercent: cpuPercent, memoryUsedBytes: used, memoryTotalBytes: memoryTotal)
    }

    /// The integer formed by the digits at the end of a `vm_stat` line (e.g. "Pages active: 123456."
    /// → 123456, "…page size of 16384 bytes)" picks up 16384 since it's the only run of digits).
    private static func trailingUInt(in line: String) -> UInt64? {
        let digits = String(line.unicodeScalars.filter { CharacterSet.decimalDigits.contains($0) })
        return UInt64(digits)
    }

    /// The idle percentage from a `top` "CPU usage: 4.76% user, 9.52% sys, 85.71% idle" line.
    private static func idlePercent(in line: String) -> Double? {
        guard let range = line.range(of: "% idle") else { return nil }
        let number = line[..<range.lowerBound]
            .reversed().prefix { $0.isNumber || $0 == "." }.reversed()
        return Double(String(number))
    }

    // MARK: - Host-side computer-use surface (SIP-free framebuffer capture + HID injection)

    /// Whether `name` is registered with the host-side surface capability — true for every VM the app
    /// boots when a host is installed. The concrete AppKit view remains lazy. Gates **capture** (the
    /// monitor's framebuffer grab); it does *not* imply the agent may
    /// drive the guest — that's ``hasComputerSurface``.
    func hasCaptureSurface(_ name: String) -> Bool {
        surfaceHost != nil && (live[name]?.hasSurface ?? false)
    }

    /// Whether the agent may **drive** `name` host-side — a bound surface *and* computer use allowed for
    /// this VM. When true, the computer-use dispatch injects HID host-side (no in-guest agent/TCC/SIP).
    func hasComputerSurface(_ name: String) -> Bool {
        hasCaptureSurface(name) && (live[name]?.computerUse ?? false)
    }

    /// Host-side framebuffer capture (JPEG base64, 1920×1200) from the guest's display driver, or `nil`.
    /// Keyed off ``hasCaptureSurface`` (not drive permission) so the monitor captures any running VM.
    func captureSurfaceBase64(name: String) async -> String? {
        guard hasCaptureSurface(name), let data = await surfaceHost?.capture(name: name) else {
            return nil
        }
        return data.base64EncodedString()
    }

    /// Inject one host-side input action into a computer-use VM (no-op without a surface).
    func sendSurfaceInput(name: String, _ input: MacVMSurfaceInput) async {
        guard hasComputerSurface(name) else { return }
        await surfaceHost?.send(name: name, input)
    }

    /// Host-tracked pointer position for a computer-use VM (we synthesize the moves).
    func surfaceCursor(name: String) async -> Point? {
        guard hasComputerSurface(name) else { return nil }
        return await surfaceHost?.cursorPosition(name: name)
    }

    /// Show/hide the operator-assist viewer for a computer-use VM: bring its live, interactive
    /// `VZVirtualMachineView` on-screen so the user can drive the guest by hand (`mac_vm_request_operator`),
    /// or send it back off-screen. No-op without a computer-use surface for `name`.
    func presentOperatorAssist(name: String, visible: Bool) async {
        guard hasComputerSurface(name) else { return }
        await surfaceHost?.presentOperatorAssist(name: name, visible: visible)
    }

    /// Current GUI framebuffer of the named guest as a base64 JPEG, for the live **VM monitor** panel.
    /// Captures **only** host-side, from the guest's display driver (the `VZVirtualMachineView`
    /// framebuffer — virtio-gpu on Linux, the Mac paravirtual display on macOS) — never via in-guest
    /// `screencapture`. No lifecycle effect; returns `nil` (rather than throwing) when the VM isn't
    /// running or the framebuffer grab is empty/unavailable, so a passive monitor can poll it cheaply and
    /// simply hold its last good frame on a miss (a macOS guest can transiently return a blank grab).
    public func captureScreen(name: String) async -> String? {
        await captureScreenData(name: name)?.base64EncodedString()
    }

    /// Raw JPEG `Data` of the named guest's current framebuffer — the live **VM monitor**'s fast path,
    /// which wants bytes it can decode straight into an image with no base64 round-trip. Same capture
    /// rules as ``captureScreen``: serves the frozen still for a suspended guest, else grabs host-side
    /// from the display driver; `nil` when the VM isn't running or the grab is empty.
    public func captureScreenData(name: String) async -> Data? {
        guard let entry = live[name] else { return nil }
        // A suspended guest's CPU is frozen, so a live capture would hang: serve the still grabbed the
        // moment it was paused — what the monitor shows under its "Paused" overlay.
        if entry.paused { return pauseFrames[name] }
        guard hasCaptureSurface(name) else { return nil }
        return await surfaceHost?.capture(name: name)
    }

    /// On app launch, reconcile on-disk clone bundles: keep only the active sessions' clones and drop
    /// the rest (daemonless — no live VM survives the process, so this is pure on-disk GC).
    public func reconcileDisk(keepNames: Set<String>) async {
        let fm = FileManager.default
        guard let entries = try? fm.contentsOfDirectory(
            at: instancesDir, includingPropertiesForKeys: nil) else { return }
        for dir in entries where !keepNames.contains(dir.lastPathComponent) {
            try? fm.removeItem(at: dir)
        }
    }

    // MARK: - Background activity

    /// Hold the anti-nap assertion while **any** guest is running — a per-session `live` VM *or* the
    /// transient base-image VM booted for a build/provision/Recovery/app-injection pass (`baseIsBusy`).
    /// The base-maintenance VM never enters `live`, so gating on `live` alone left the whole base
    /// build/provision pass unprotected: an auto-rebuild that fires at launch (e.g. right after an
    /// update, while the relaunched app is still in the background) would get App-Napped, throttling the
    /// main-queue provisioning VM and the HID that drives it — so the guest never reached a drivable
    /// desktop and the build wedged at "Waiting for the guest desktop". Callers must re-run this whenever
    /// `live` **or** a base-busy flag (`baseBuilding` / `baseAppInstalling` / `baseRecoveryActive`) changes.
    func syncBackgroundActivity() {
        let needed = !live.isEmpty || baseIsBusy
        if needed, backgroundActivity == nil {
            backgroundActivity = ProcessInfo.processInfo.beginActivity(
                options: [
                    .userInitiatedAllowingIdleSystemSleep, .suddenTerminationDisabled,
                    .automaticTerminationDisabled,
                ],
                reason: "Nucleic macOS VM running")
        } else if !needed, let activity = backgroundActivity {
            ProcessInfo.processInfo.endActivity(activity)
            backgroundActivity = nil
        }
    }

    // MARK: - vsock exec plumbing

    /// Build a `ProcessHandle` for a command run inside the guest **over vsock** — a dedicated
    /// connection to the in-guest agent's streaming `exec` op (docs/MACOS_VM_NATIVE_AGENT.md §4.4). This
    /// replaced the `ssh`-over-NAT handle: no IP discovery, no host-key churn, no Remote Login, no
    /// network permission surface. The composed script is identical to what sshd used to run, so the
    /// guest's toolchain PATH (via `/etc/zshenv`) resolves the same way.
    ///
    /// Each exec gets its OWN vsock connection (the agent listener is one thread per connection), so a
    /// multi-minute build never blocks the control channel and concurrent execs don't interfere.
    private func execHandle(
        name: String, workdir: String?, env: [String: String], remoteBody: String
    ) async throws -> any ProcessHandle {
        #if arch(arm64)
        return try await openExecChannel(name: name, workdir: workdir, env: env, remoteBody: remoteBody)
        #else
        throw MacVMError.unavailable(Self.unsupportedReason ?? "requires Apple silicon")
        #endif
    }

    #if arch(arm64)
    /// Open a dedicated vsock exec channel to the named guest and dispatch `remoteBody` (wrapped with
    /// env/workdir by ``remoteScript``). Returns the concrete ``MacVMExecChannel`` so bulk callers can
    /// stream binary stdin (``MacVMExecChannel/writeStdinRaw(_:deadline:)``); ``execHandle`` erases it
    /// to `any ProcessHandle` for the ordinary exec path.
    func openExecChannel(
        name: String, workdir: String?, env: [String: String], remoteBody: String
    ) async throws -> MacVMExecChannel {
        guard let entry = live[name] else { throw MacVMError.notRunning(name) }
        return try await openExecChannel(
            instance: entry.instance, label: name, workdir: workdir, env: env, remoteBody: remoteBody)
    }

    /// Open an exec channel directly on a ``MacVMInstance`` — for VMs the engine boots OUTSIDE the
    /// `live` registry (the base app-injection boot), which have no name to look up. `label` is used
    /// only in the error message.
    func openExecChannel(
        instance: MacVMInstance, label: String, workdir: String?, env: [String: String],
        remoteBody: String
    ) async throws -> MacVMExecChannel {
        let script = Self.remoteScript(workdir: workdir, env: env, body: remoteBody)
        let box: UncheckedSendableBox<VZVirtioSocketConnection>
        do {
            box = try await withAgentTimeout(seconds: 10) {
                try await instance.connectAgent(port: MacVMAgentWire.port)
            }
        } catch {
            throw MacVMError.agentUnavailable(
                "could not open an exec channel to \"\(label)\": \(error)")
        }
        return MacVMExecChannel(connection: box, command: script)
    }
    #endif

    /// Compose the remote shell script: export the extra env, `cd` into the working directory, then
    /// run the body. The guest's toolchain PATH is made available to non-login shells by provisioning
    /// (it appends to `/etc/zshenv`), so `xcodebuild`/`brew`/`node` resolve without a login shell.
    static func remoteScript(workdir: String?, env: [String: String], body: String) -> String {
        var s = ""
        for (k, v) in env.sorted(by: { $0.key < $1.key }) {
            s += "export \(k)=\(shQuote(v)); "
        }
        if let workdir, !workdir.isEmpty {
            s += "cd \(shQuote(workdir)) || exit 1; "
        }
        s += body
        return s
    }

    /// POSIX single-quote a string for safe interpolation into a shell command.
    static func shQuote(_ s: String) -> String {
        "'" + s.replacingOccurrences(of: "'", with: "'\\''") + "'"
    }

    // MARK: - Background-output log (live monitor tail)

    /// Newest-last snapshot of the named guest's recent `mac_vm_exec` output. Empty when the VM has run
    /// nothing (or has stopped). Read by the app layer for the monitor's under-screen console.
    func backgroundLog(name: String) -> [MacVMLogLine] {
        backgroundLogLines[name] ?? []
    }

    /// Reset the guest's background-output ring so a new headless work episode's console starts clean
    /// rather than trailing the previous one's output. The manager calls this when a `mac_vm_exec` begins
    /// with no other exec in flight (an idle→busy edge). Without it the ring is only cleared when the VM
    /// *stops*, so a long-lived guest's console would accumulate every command it ever ran for its whole
    /// life — and resurface that backlog (historically including the resource probe's `vm_stat` spam)
    /// whenever the monitor re-reads it, e.g. when the user clicks back into the session.
    func resetBackgroundLog(name: String) {
        clearBackgroundLog(name: name)
    }

    /// Append one line to a guest's rolling background-output tail, assigning it the next per-VM id and
    /// evicting the oldest lines past ``maxBackgroundLogLines``. Over-long lines are clipped so a single
    /// runaway line can't defeat the ring's bound.
    private func appendBackgroundLog(name: String, kind: MacVMLogLine.Kind, text: String) {
        let clipped = text.count > Self.maxBackgroundLogLineLength
            ? String(text.prefix(Self.maxBackgroundLogLineLength)) + "…"
            : text
        let id = backgroundLogSeq[name, default: 0]
        backgroundLogSeq[name] = id + 1
        var lines = backgroundLogLines[name] ?? []
        lines.append(MacVMLogLine(id: id, kind: kind, text: clipped))
        if lines.count > Self.maxBackgroundLogLines {
            lines.removeFirst(lines.count - Self.maxBackgroundLogLines)
        }
        backgroundLogLines[name] = lines
    }

    /// Drop a guest's retained background output — on teardown, so a rebooted clone starts clean.
    private func clearBackgroundLog(name: String) {
        backgroundLogLines[name] = nil
        backgroundLogSeq[name] = nil
    }

    /// The `$ <command>` echo written when an exec starts: the command's first line, trimmed, so a
    /// multi-line heredoc reads as one marker rather than flooding the tail.
    private static func commandMarker(_ command: String) -> String {
        let firstLine = command.split(
            separator: "\n", maxSplits: 1, omittingEmptySubsequences: false
        ).first.map(String.init) ?? command
        return "$ " + firstLine.trimmingCharacters(in: .whitespaces)
    }

    /// Drain a line stream exactly like ``collect`` (same cap + truncation marker) while *also* mirroring
    /// each line into the guest's rolling background-output tail as it arrives — so the live monitor shows
    /// a running command's output in near-real-time, not only once it finishes. Actor-isolated (unlike
    /// static ``collect``) so it can touch the ring; the per-line `await` on the stream releases the actor
    /// between lines, so the stdout and stderr drains still interleave and neither blocks the other.
    private func collectTapping(
        _ stream: AsyncThrowingStream<Data, Error>, name: String, kind: MacVMLogLine.Kind,
        cap: Int = 64 * 1024
    ) async -> String {
        var lines: [String] = []
        var bytes = 0
        var truncated = false
        do {
            for try await line in stream {
                let text = String(decoding: line, as: UTF8.self)
                appendBackgroundLog(name: name, kind: kind, text: text)
                if bytes < cap {
                    lines.append(text)
                    bytes += line.count + 1
                } else {
                    truncated = true
                }
            }
        } catch { /* stream ended on error — return what we have */ }
        var out = lines.joined(separator: "\n")
        if truncated { out += "\n…[output truncated at \(cap / 1024) KB]" }
        return out
    }

    /// Drain an async line stream into one newline-joined string, capped so a runaway build log
    /// (`xcodebuild` output is routinely multi-MB) can't blow up the tool result the agent reads back
    /// or balloon host memory — mirrors the host path's `runOnHost` capping. Keeps consuming past the
    /// cap (so the ssh pipe never blocks) but drops the overflow, appending a truncation marker.
    private static func collect(
        _ stream: AsyncThrowingStream<Data, Error>, cap: Int = 64 * 1024
    ) async -> String {
        var lines: [String] = []
        var bytes = 0
        var truncated = false
        do {
            for try await line in stream {
                if bytes < cap {
                    lines.append(String(decoding: line, as: UTF8.self))
                    bytes += line.count + 1
                } else {
                    truncated = true
                }
            }
        } catch { /* stream ended on error — return what we have */ }
        var out = lines.joined(separator: "\n")
        if truncated { out += "\n…[output truncated at \(cap / 1024) KB]" }
        return out
    }

    // MARK: - Guest readiness

    /// Wait until the in-guest agent answers a vsock `ping`, bounded. This IS the boot-readiness gate:
    /// the host↔guest control plane is vsock-only, so "the agent answers" is exactly "the guest is
    /// usable for exec + computer-use". No DHCP lease, no sshd, no IP on the critical path.
    ///
    /// On success the live `ping` channel is cached as the VM's control client, so the first
    /// computer-use op doesn't pay a second probe. Non-sticky: a failed probe just retries (unlike the
    /// lazy ``agentClient(name:)``, whose `.unavailable` is sticky for the VM's lifetime).
    func awaitAgentReady(name: String) async throws {
        #if arch(arm64)
        guard let instance = live[name]?.instance else { throw MacVMError.notRunning(name) }
        let deadline = Date().addingTimeInterval(300)  // first boot: login window + launchd + agent
        while Date() < deadline {
            if let box = try? await withAgentTimeout(seconds: 5, {
                try await instance.connectAgent(port: MacVMAgentWire.port)
            }) {
                let client = MacVMAgentClient(connection: box)
                if let pong = try? await client.request(op: "ping", timeout: 5),
                    pong["version"]?.intValue == MacVMAgentWire.version
                {
                    if live[name] != nil { live[name]?.agent = .available(client) }
                    return
                }
                await client.close()
            }
            try? await Task.sleep(nanoseconds: 2_000_000_000)
        }
        throw MacVMError.agentUnavailable(
            "the in-guest agent never answered a vsock ping for \"\(name)\" within the boot deadline")
        #else
        throw MacVMError.unavailable(Self.unsupportedReason ?? "requires Apple silicon")
        #endif
    }

    /// Look up the IPv4 address currently leased to `mac` in `/var/db/dhcpd_leases` (written by
    /// Apple's vmnet NAT DHCP server) — **display only** now (the NAT device exists purely for the
    /// guest's own outbound internet; the control plane is vsock). Reads the file, then delegates to
    /// the pure ``parseLeases``.
    static func leaseIP(forMAC mac: String) -> String? {
        guard let text = try? String(contentsOfFile: "/var/db/dhcpd_leases", encoding: .utf8)
        else { return nil }
        return parseLeases(text, forMAC: mac)
    }

    /// Pure parser for `dhcpd_leases` content: return the IPv4 leased to `mac`. MACs are compared
    /// octet-by-octet as integers, since the lease file drops leading zeros (`2:34:…`) while
    /// `VZMACAddress.string` keeps them (`02:34:…`). The last (newest) matching lease wins.
    static func parseLeases(_ text: String, forMAC mac: String) -> String? {
        let want = normalizeMAC(mac)
        var currentIP: String?
        var lastMatchIP: String?
        for rawLine in text.split(whereSeparator: \.isNewline) {
            let line = rawLine.trimmingCharacters(in: .whitespaces)
            if line == "{" { currentIP = nil }
            else if line.hasPrefix("ip_address=") {
                currentIP = String(line.dropFirst("ip_address=".count))
            } else if line.hasPrefix("hw_address=") {
                // Format: `hw_address=1,d6:a7:58:8e:78:d4` — strip the leading `<type>,`.
                var hw = String(line.dropFirst("hw_address=".count))
                if let comma = hw.firstIndex(of: ",") { hw = String(hw[hw.index(after: comma)...]) }
                if normalizeMAC(hw) == want, let ip = currentIP { lastMatchIP = ip }
            }
        }
        return lastMatchIP
    }

    /// Normalize a MAC to a colon-joined list of two-digit lowercase hex octets for comparison.
    static func normalizeMAC(_ mac: String) -> String {
        mac.split(separator: ":").compactMap { UInt8($0, radix: 16) }
            .map { String(format: "%02x", $0) }.joined(separator: ":")
    }

    /// A fresh locally-administered MAC (the U/L bit set, multicast bit clear) as a colon string.
    static func randomMAC() -> String {
        #if arch(arm64)
        return VZMACAddress.randomLocallyAdministered().string
        #else
        var octets = (0..<6).map { _ in UInt8.random(in: 0...255) }
        octets[0] = (octets[0] & 0xFE) | 0x02
        return octets.map { String(format: "%02x", $0) }.joined(separator: ":")
        #endif
    }
}

#if arch(arm64)
/// Owns one live `VZVirtualMachine` and the serial dispatch queue every VZ call must run on. VZ
/// objects are not `Sendable` and are only safe to touch on their own queue, so this box confines all
/// of that: it is `@unchecked Sendable`, never hands a VZ-typed value across its boundary, and bridges
/// the queue's completion-handler API to `async` via continuations. The engine actor holds these
/// boxes but performs every VM operation through this narrow, queue-correct surface.
final class MacVMInstance: NSObject, VZVirtualMachineDelegate, @unchecked Sendable {
    let queue: DispatchQueue
    /// The wrapped VM. Internal (not private) so the install extension in `MacVMEngine+Base` can drive
    /// `VZMacOSInstaller` against it — but ONLY ever touched on `queue`.
    let vm: VZVirtualMachine
    /// Fired (once) when the guest powers itself off or stops with an error.
    private var onStop: (@Sendable () -> Void)? = nil

    init(configuration: VZVirtualMachineConfiguration, label: String) {
        // EVERY VM runs on its own dedicated serial queue — never the main queue — so guest virtio
        // device I/O never contends with the app's UI on the main thread. This includes VMs whose
        // screen is displayed: a `VZVirtualMachineView` requires only that the *view* object be
        // touched on the main thread (AppKit), NOT that the VM run on the main queue. The view is
        // bound on `@MainActor` in `MacVMComputerSurface.attach` while the VM lives here on `queue`,
        // exactly as UTM does. `qos: .userInitiated` keeps guest responsiveness high without the
        // priority-inversion warnings that `.userInteractive` triggers against the main thread.
        // `init` runs on the engine actor (never on this fresh queue), so the `q.sync` calls below —
        // which create the VM and set its delegate on its own queue, as VZ requires — can't deadlock.
        let q = DispatchQueue(label: "xyz.blakeslee.nucleic.macvm.\(label)", qos: .userInitiated)
        self.queue = q
        self.vm = q.sync { VZVirtualMachine(configuration: configuration, queue: q) }
        super.init()
        q.sync { self.vm.delegate = self }
    }

    /// Start the guest; resumes once `VZVirtualMachine.start` reports success or throws its error.
    func start() async throws {
        try await withCheckedThrowingContinuation { (cont: CheckedContinuation<Void, Error>) in
            queue.async { [self] in
                vm.start { result in
                    switch result {
                    case .success: cont.resume()
                    case .failure(let error): cont.resume(throwing: error)
                    }
                }
            }
        }
    }

    /// Force-stop (power-off) the guest. Best-effort and idempotent — a clone is disposable, so this
    /// doesn't bother with a graceful guest shutdown request.
    func stop() async {
        await withCheckedContinuation { (cont: CheckedContinuation<Void, Never>) in
            queue.async { [self] in
                guard vm.canStop else { cont.resume(); return }
                vm.stop { _ in cont.resume() }
            }
        }
    }

    /// Pause (freeze) the running guest in place: it stays resident in host RAM but consumes no CPU,
    /// and ``resume()`` thaws it instantly (much cheaper than a stop→reboot, but — unlike stop — it does
    /// NOT give the RAM back). Best-effort/idempotent: a no-op returning `false` when the guest can't
    /// pause (already paused, or not in a pausable state). Runs on `queue`, like every other VZ call.
    @discardableResult
    func pause() async -> Bool {
        await withCheckedContinuation { (cont: CheckedContinuation<Bool, Never>) in
            queue.async { [self] in
                guard vm.canPause else { cont.resume(returning: false); return }
                vm.pause { result in
                    if case .success = result { cont.resume(returning: true) }
                    else { cont.resume(returning: false) }
                }
            }
        }
    }

    /// Resume (thaw) a paused guest. Best-effort/idempotent: a no-op returning `false` when the guest
    /// isn't in a resumable (paused) state.
    @discardableResult
    func resume() async -> Bool {
        await withCheckedContinuation { (cont: CheckedContinuation<Bool, Never>) in
            queue.async { [self] in
                guard vm.canResume else { cont.resume(returning: false); return }
                vm.resume { result in
                    if case .success = result { cont.resume(returning: true) }
                    else { cont.resume(returning: false) }
                }
            }
        }
    }

    /// **Suspend to disk:** write the entire runtime state (guest RAM + device state) to `url`, so the
    /// guest can be torn down (freeing its host RAM and its concurrency slot) and later reconstructed
    /// byte-for-byte by ``restore(from:)`` — the state-preserving counterpart to a stop→cold-reboot.
    /// The VM MUST be paused first (VZ requires `.paused`); the engine pauses before calling this.
    /// Throws on any framework error (the caller then falls back to a plain RAM-pause).
    func save(to url: URL) async throws {
        try await withCheckedThrowingContinuation { (cont: CheckedContinuation<Void, Error>) in
            queue.async { [self] in
                vm.saveMachineStateTo(url: url) { error in
                    if let error { cont.resume(throwing: error) } else { cont.resume() }
                }
            }
        }
    }

    /// **Resume from disk:** reconstruct the runtime state previously written by ``save(to:)`` into
    /// this freshly-created VM (VZ requires state `.stopped`). Leaves the VM **paused** — the caller
    /// then calls ``resume()`` to run it. Throws when the saved state is incompatible/corrupt (the
    /// caller then discards it and cold-boots the clone instead).
    func restore(from url: URL) async throws {
        try await withCheckedThrowingContinuation { (cont: CheckedContinuation<Void, Error>) in
            queue.async { [self] in
                vm.restoreMachineStateFrom(url: url) { error in
                    if let error { cont.resume(throwing: error) } else { cont.resume() }
                }
            }
        }
    }

    /// Register a one-shot hook fired when the guest stops on its own (power-off / panic / error).
    /// Assigned on `queue` so `onStop` stays confined to the VM queue — the same queue the delegate
    /// callbacks (hence `fireStop`) run on, so there's no cross-thread access to the reference.
    func setOnStop(_ hook: @escaping @Sendable () -> Void) {
        queue.async { [self] in onStop = hook }
    }

    // VZVirtualMachineDelegate — the guest stopped on its own or errored. Fire the hook (once).
    // Delivered by VZ on `queue`, so `onStop` is read/cleared here on its owning queue.
    func guestDidStop(_ virtualMachine: VZVirtualMachine) { fireStop() }
    func virtualMachine(_ virtualMachine: VZVirtualMachine, didStopWithError error: Error) {
        fireStop()
    }

    private func fireStop() {
        let hook = onStop
        onStop = nil
        hook?()
    }
}
#endif