import Foundation
import Containerization
import ContainerizationOCI
import ContainerizationExtras
import NIOPosix

/// Errors surfaced from the in-process container engine.
public enum ContainerError: Error, Sendable, CustomStringConvertible {
    /// Containerization isn't usable on this machine (not Apple silicon, missing the bundled
    /// kernel / vminitd initfs, or — surfaced later, at VM start — no virtualization entitlement).
    case unavailable(String)
    /// An operation referenced a container that the engine isn't currently running.
    case notRunning(String)
    /// Pulling the base image for the sandbox rootfs failed. Carries the underlying message.
    case imagePullFailed(String)
    /// Building the provisioned sandbox rootfs (unpack + in-VM provisioning) failed.
    case rootfsBuildFailed(String)
    /// The container VM failed to create/start. Carries the underlying message.
    case startFailed(String)
    /// Couldn't determine the host-gateway address for the container's network interface.
    case gatewayUnavailable

    public var description: String {
        switch self {
        case let .unavailable(why):
            return "Container runtime unavailable: \(why)"
        case let .notRunning(name):
            return "Container \"\(name)\" isn't running."
        case let .imagePullFailed(msg):
            return "Failed to pull the sandbox base image: \(msg)"
        case let .rootfsBuildFailed(msg):
            return "Failed to build the sandbox root filesystem: \(msg)"
        case let .startFailed(msg):
            return "Failed to start the container VM: \(msg)"
        case .gatewayUnavailable:
            return "Couldn't determine the host gateway address for the container."
        }
    }
}

/// A point-in-time CPU/memory reading for a running container, for the Control panel's resource
/// monitor. CPU is the container's aggregate utilization across its cores (0…100); memory is the
/// in-VM view — `total` is the configured memory ceiling the OOM-killer enforces, so `used / total`
/// is exactly "how close to the status-137 OOM kill." Sampled in-process via the framework's typed
/// `statistics()` (two reads around a short sleep for the CPU delta); absent when the container
/// isn't running.
public struct ContainerResourceSample: Sendable, Equatable {
    /// Aggregate CPU utilization over the sample window, clamped to 0…100.
    public let cpuPercent: Double
    /// Resident memory in use inside the container, bytes.
    public let memoryUsedBytes: UInt64
    /// Total memory visible inside the container (≈ the configured limit), bytes.
    public let memoryTotalBytes: UInt64

    public init(cpuPercent: Double, memoryUsedBytes: UInt64, memoryTotalBytes: UInt64) {
        self.cpuPercent = min(100, max(0, cpuPercent))
        self.memoryUsedBytes = memoryUsedBytes
        self.memoryTotalBytes = memoryTotalBytes
    }

    /// Memory utilization 0…100. Zero when the total is unknown (avoids a divide-by-zero).
    public var memoryPercent: Double {
        memoryTotalBytes == 0 ? 0 : min(100, Double(memoryUsedBytes) / Double(memoryTotalBytes) * 100)
    }
}

/// Post-mortem of an abnormal (137 / SIGKILL) agent exit, gathered by probing the container the
/// instant it happens. Distinguishes the three things that all surface as "status 137" but want
/// opposite responses: the whole container went down (→ every session in it dies at once, not an
/// OOM), a genuine kernel OOM kill (the persistent `oom_kill` counter proves it even after memory
/// reads low again), or a SIGKILL while the container is healthy and memory is low (something else
/// killed the process — not memory).
public struct ContainerKillDiagnosis: Sendable, Equatable {
    /// False when the engine no longer tracks the container — the app's own runtime stopped,
    /// restarted, or removed it (idle stop, recovery restart, teardown). Deliberate lifecycle, not
    /// a VM crash: a crashed VM stays registered (see ``probeUnanswered``).
    public let containerRunning: Bool
    /// True when the container is still registered but its status probe got no answer even after a
    /// retry — the VM itself crashed (e.g. a guest kernel panic) or is too wedged to respond.
    /// Distinguished from ``containerRunning`` `== false` so a transient probe failure is never
    /// misreported as "the app stopped the container" (or vice versa).
    public let probeUnanswered: Bool
    /// cgroup `oom_kill` counter (cumulative over the container's life). `> 0` is hard proof the
    /// kernel OOM-killed something here. `nil` when the counter is unavailable.
    public let oomKills: Int?
    /// Memory (and a zeroed CPU) at probe time, if the container answered. Memory can read low even
    /// for a real OOM, because the kill already freed it — hence `oomKills` is the OOM authority.
    public let sample: ContainerResourceSample?

    public init(
        containerRunning: Bool, probeUnanswered: Bool = false, oomKills: Int?,
        sample: ContainerResourceSample?
    ) {
        self.containerRunning = containerRunning
        self.probeUnanswered = probeUnanswered
        self.oomKills = oomKills
        self.sample = sample
    }
}

/// In-process container runtime built directly on Apple's `containerization` framework — the same
/// framework Apple's `container` CLI is built on. Replaces the old `container`-CLI wrapper: instead
/// of shelling out to a daemon, this actor owns the runtime itself — one shared
/// `VZVirtualMachineManager` (bundled Linux kernel + vminitd initfs), an `ImageStore` (for pulling
/// the sandbox base image), a `VmnetNetwork` (NAT networking + the host gateway the guest calls
/// back on), and a registry of live `LinuxContainer`s keyed by name.
///
/// **Daemonless ⇒ ephemeral.** There is no external daemon, so every VM's lifetime is bounded by
/// the app process: the registry starts empty each launch and containers are created on demand.
/// `reconcile`/orphan-reaping therefore becomes on-disk GC (there are no surviving VMs to kill).
/// The per-container writable rootfs (a clone of the cached provisioned image) persists on disk, so
/// an idle-stopped or app-restarted container recreates fast and skips re-seeding.
///
/// Lifecycle *policy* (busy ref-counting, idle timers, shared-container naming) lives one layer up
/// in `ContainerManager`; this engine is the mechanism it drives.
///
/// > Note: `containerization` also exposes a type named `ContainerManager`; this is **not** it —
/// > Nucleic's `ContainerManager` is the policy actor, and this engine drives `LinuxContainer`
/// > directly (mirroring how `apple/container`'s `RuntimeService` consumes the framework).
public actor ContainerEngine {
    /// One live container the engine is currently running.
    struct LiveContainer {
        let container: LinuxContainer
        let gateway: String
        /// True when the rootfs clone was freshly created (so seeding must run); false when an
        /// existing clone was reused (idle-restart / relaunch — instrumentation already installed).
        let freshlyCloned: Bool
        /// The VM's memory balloon, when memory management is enabled at create time (else `nil`).
        /// Driven by the engine's autoballoon loop to return idle guest memory to the host.
        let balloon: MemoryBalloon?
        /// The VM's configured memory ceiling in bytes (`spec.memoryGiB`). Kept here so the autoballoon
        /// loop can fully deflate the balloon (target → ceiling) when a `statistics()` read fails,
        /// rather than leaving the guest pinned at a stale, inflated target it can never recover from.
        let memoryCeilingBytes: UInt64
    }

    /// On-disk root for engine artifacts (rootfs cache, per-container clones, image content store).
    let storageRoot: URL
    /// Live containers keyed by name. Empty at launch (nothing survives the app process).
    var live: [String: LiveContainer] = [:]
    /// Last spec seen per name, so `restart` (which only has a name) can recreate from disk.
    var lastSpec: [String: ContainerSpec] = [:]
    /// Cold starts currently in flight, keyed by container name. A start is a long, suspension-rich
    /// operation (image pull, unpack, rootfs clone, VM boot) and `live[name]` is only published at
    /// the very end — so without this, two turns racing to start the SAME container (the norm for a
    /// shared control container: every session's turn calls `ensureRunning`) each run the whole
    /// create path. The loser then failed on the rootfs clone with "an item with the same name
    /// already exists" (NSFileWriteFileExistsError), and even when it didn't, the two starts fought
    /// over one interface allocation and one `live` slot — leaking a VM the engine could no longer
    /// stop. Second and later callers join the first start's task instead.
    private var starting: [String: Task<(name: String, hostGateway: String), Error>] = [:]
    /// True while any cold start is in flight — the disk GC uses it to leave in-progress clone
    /// staging files alone.
    var isStarting: Bool { !starting.isEmpty }
    /// Image pull+unpacks currently in flight, keyed by image ref — the same coalescing one level
    /// down, for the case ``starting`` can't cover: two *differently named* containers coming up
    /// cold on the same image would otherwise each download and unpack it into the one shared cache.
    var pulling: [String: Task<URL, Error>] = [:]

    /// Held while at least one VM is live, to keep the host app from being App-Napped, automatically
    /// terminated, or suddenly killed while a container is running. The packaged app opts into
    /// `NSSupportsSuddenTermination`/`NSSupportsAutomaticTermination` (see `scripts/package-app.sh`),
    /// so a backgrounded, window-hidden app is a candidate for the OS to quietly quit — and because
    /// the runtime is daemonless (every VM's lifetime is bounded by this process), that takes every
    /// container down with it. This assertion suspends those two opt-ins plus App Nap for as long as a
    /// VM is up, so a container survives the app sitting in the background. `nil` when nothing is live.
    private var backgroundActivity: (any NSObjectProtocol)?

    /// Lazily-built shared VM manager (kernel + vminitd initfs) — one per app run.
    private var vmm: VirtualMachineManager?

    /// The autoballoon driver — one repeating task that retargets every live container's memory
    /// balloon from its working set. Started lazily when the first balloon-enabled container comes
    /// up; cancelled when none remain. `nil` when not running.
    private var balloonLoop: Task<Void, Never>?
    /// Lazily-built image store for pulling the sandbox base image.
    private var imageStore: ImageStore?
    /// Lazily-built NAT network; `var` because `createInterface` mutates it.
    private var network: VmnetNetwork?

    /// The artifact acquisition currently in flight (kernel / init / image pull + unpack), surfaced
    /// to the Control panel's progress bar and the chat's delay hint. `nil` whenever everything is
    /// cached and nothing is downloading — the steady state. Mutated only on the actor; read via
    /// ``currentDownloadProgress()``.
    private var downloadProgress: ContainerDownloadProgress?
    /// Nesting depth of in-flight acquisitions, so the *last* one to finish clears the indicator even
    /// when two containers provision concurrently against the shared on-disk cache.
    private var downloadDepth = 0
    /// Running byte accumulators for the active phase (reset at each ``beginDownload``); the registry
    /// progress handler reports increments, which we fold into these to drive a determinate bar.
    private var downloadReceived: Int64 = 0
    private var downloadTotal: Int64 = 0

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

    /// The vminitd initial-filesystem OCI image — the guest init (PID 1, gRPC over vsock). Pulled
    /// at runtime and materialized to an ext4, rather than bundled or cross-compiled.
    ///
    /// This is Nucleic's **custom** build (not Apple's stock `ghcr.io/apple/containerization/vminit`),
    /// because we carry guest-side patches in `third_party/containerization/vminitd` — see
    /// `third_party/containerization/PATCHES.md`. It is built locally with `make vminit-image` (root
    /// Makefile) FROM the vendored source, so the guest vminitd and the host `Containerization`
    /// framework compile from the same pinned commit and speak the same vsock protocol. (Built locally
    /// rather than in CI because the host framework needs the macOS 26+ Virtualization SDK that
    /// GitHub-hosted runners lack.)
    ///
    /// This is the **currently-deployed** tag. The workflow for a new guest patch: bump the
    /// Makefile's `NUCLEIC_VMINIT_REF` (the *next* tag to build), `make vminit-image` + `make
    /// vminit-image-push`, validate it in a real container, THEN bump this to match — so
    /// `NUCLEIC_VMINIT_REF` may be one revision ahead of this while a new image is being validated.
    ///
    /// `-nucleic4` = everything in `-nucleic3` (guest patches #1–#10, #12) plus patch #14 — the
    /// non-blocking, non-spinning guest I/O plane: `IOPair` gains real backpressure (non-blocking
    /// write fd + pending backlog + EPOLLOUT flush), `OSFile.splice` returns instead of busy-spinning
    /// when the destination is full, and `VsockProxy` non-blocks both connection fds before
    /// registration. Together these stop one slow peer from freezing the shared ProcessSupervisor
    /// poller thread — the root cause of the "control plane accepts but never answers / every
    /// session stalls at once" wedge. Deployed ahead of full runtime validation by explicit
    /// decision — if a container misbehaves on it, revert this constant to `…-nucleic3`;
    /// the reference-keyed initfs cache re-pulls on next launch either way.
    /// (`0.34.0` = the upstream vminit version this replaces; `-nucleicN` = our guest-patch revision.)
    ///
    /// `-nucleic5` (2026-07-29, PATCHES.md #15–#17): the fix for the wedge `-nucleic4`'s
    /// backpressure patch left reachable — the shared-offset splice conflation that spun the
    /// poller thread into the persistent "accepts connections but never answers" dead control
    /// plane (4 of the 5 stalls in the 36h before this deploy carried that exact probe verdict) —
    /// plus loss-free stdio teardown (no more truncated final output lines), epoll registration
    /// generations (recycled-fd stale events), and the TerminalIO shared-fd collision fix.
    public static let vminitReference = "ghcr.io/abkslm/vminit:0.34.0-nucleic5"

    /// In-guest path of the control bridge the init launches when a control socket is relayed in
    /// (loopback TCP → the relayed socket). Shipped by `containers/nucleic-sandbox/Dockerfile`; its
    /// port + socket defaults match `ContainerSpec.controlBridgePort` / `controlSocketGuestPath`.
    static let controlBridgeGuestPath = "/opt/nucleic/control-bridge.js"

    // MARK: - Capability

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

    /// A specific, actionable reason the runtime is unavailable, or `nil` when supported. The only
    /// thing this static check gates on is the CPU architecture (Apple silicon). Everything else is
    /// acquired automatically on first use — the kernel, the vminitd initfs, and the sandbox image
    /// all download + cache in the background — so a missing network surfaces as a pull error then,
    /// not as an "unsupported machine". It also can't detect the virtualization *entitlement*
    /// (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 "It requires Apple silicon."
        #endif
    }

    private 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/containers", isDirectory: true)
    }

    // MARK: - Download progress

    /// The artifact acquisition currently in flight, for the Control panel's progress bar and the
    /// chat's "why is this slow" hint. `nil` once every artifact is cached. A trivial property read
    /// (no I/O), so callers can poll it on the snapshot's critical path without risk of a stall.
    public func currentDownloadProgress() -> ContainerDownloadProgress? { downloadProgress }

    /// Mark the start of acquiring `phase`'s artifact and reset its byte accumulators. Paired with
    /// ``endDownload()`` (use `defer`) so the indicator clears even on a thrown pull. Internal (not
    /// private) so the rootfs builder in the extension file can drive the image-pull phase.
    func beginDownload(_ phase: ContainerDownloadProgress.Phase) {
        downloadDepth += 1
        downloadReceived = 0
        downloadTotal = 0
        downloadProgress = ContainerDownloadProgress(phase: phase)
    }

    /// Move the active acquisition to a later stage *under the same* begin/end (e.g. image-pull →
    /// unpack), resetting the byte accumulators so the new stage starts from zero.
    func setDownloadPhase(_ phase: ContainerDownloadProgress.Phase) {
        downloadReceived = 0
        downloadTotal = 0
        downloadProgress = ContainerDownloadProgress(phase: phase)
    }

    /// Fold incremental registry progress events into the current phase's running byte totals and
    /// republish. Called from the `ProgressHandler` the pull APIs invoke as layers stream in.
    func advanceDownload(_ events: [ProgressEvent], phase: ContainerDownloadProgress.Phase) {
        for event in events {
            switch event {
            case .addSize(let v): downloadReceived += v
            case .addTotalSize(let v): downloadTotal += v
            case .addItems, .addTotalItems: break  // item counts aren't surfaced; bytes drive the bar
            }
        }
        downloadProgress = ContainerDownloadProgress(
            phase: phase, receivedBytes: downloadReceived, totalBytes: downloadTotal)
    }

    /// Mark one acquisition done; clears the indicator once the last concurrent one finishes.
    func endDownload() {
        downloadDepth = max(0, downloadDepth - 1)
        if downloadDepth == 0 { downloadProgress = nil }
    }

    // MARK: - Shared runtime (lazy)

    /// Build (once) the shared VM manager from the kernel + vminitd initfs — both acquired
    /// automatically on first use (the kernel is downloaded/reused, the initfs pulled). Mirrors the
    /// wiring `apple/container`'s `RuntimeService` uses: a couple of hardening args appended to the
    /// kernel command line, then `VZVirtualMachineManager(kernel:initialFilesystem:)`. Internal (not
    /// private) so the extension file can reuse the same manager.
    func sharedVMM() async throws -> VirtualMachineManager {
        if let vmm { return vmm }
        guard Self.isSupported else {
            throw ContainerError.unavailable("requires Apple silicon")
        }
        let kernelURL = try await ensureKernel()
        let initfs = try await ensureInitfs()
        var kernel = Kernel(path: kernelURL, platform: .linuxArm)
        // Match the CLI's guest hardening (see apple/container RuntimeService).
        kernel.commandLine.kernelArgs.append("oops=panic")
        kernel.commandLine.kernelArgs.append("lsm=lockdown,capability,landlock,yama,apparmor")
        let vmm = VZVirtualMachineManager(
            kernel: kernel, initialFilesystem: initfs, group: Self.vmEventLoopGroup)
        self.vmm = vmm
        return vmm
    }

    /// One process-wide event-loop group shared by every container VM's vsock channels. Without an
    /// explicit group each `VZVirtualMachineInstance` builds its own `System.coreCount`-thread group
    /// and reclaims it only on a perfectly clean `stop()` — so every container boot (and every
    /// unclean teardown) grew the process by a dozen idle `NIO-ELT` threads: the 2026-07-28 hang
    /// reports show 4 such groups (48 threads) after one evening. Vsock agent RPC plus exec stdio
    /// streams are light, multiplexed work; a few shared loops carry any number of VMs.
    private static let vmEventLoopGroup = MultiThreadedEventLoopGroup(numberOfThreads: 4)

    /// Resolve the Linux kernel, acquiring it automatically the first time so users run no setup
    /// step. Resolution order (first hit wins, cached after):
    ///   1. `NUCLEIC_KERNEL_PATH` env (dev override).
    ///   2. The engine's on-disk cache.
    ///   3. A kernel bundled in the app's Resources (optional dev/offline fast-path).
    ///   4. Apple `container` CLI's installed kernel, if present — reused with no download.
    ///   5. Pull the kernel from the `ProjectSandbox.kernelImage` GHCR package — anonymously when the
    ///      package is public, else authenticated with the app's GitHub token (as the sandbox image).
    private func ensureKernel() async throws -> URL {
        if let p = ProcessInfo.processInfo.environment["NUCLEIC_KERNEL_PATH"],
            FileManager.default.fileExists(atPath: p)
        {
            return URL(fileURLWithPath: p)
        }
        let dir = storageRoot.appendingPathComponent("kernel", isDirectory: true)
        try FileManager.default.createDirectory(at: dir, withIntermediateDirectories: true)
        let cacheURL = dir.appendingPathComponent("vmlinux-arm64")
        if FileManager.default.fileExists(atPath: cacheURL.path) { return cacheURL }

        // `NUCLEIC_KERNEL_NO_REUSE` forces the download path, skipping the local fast-paths — for
        // testing the download, or when a local/CLI kernel is present but incompatible.
        if ProcessInfo.processInfo.environment["NUCLEIC_KERNEL_NO_REUSE"] == nil {
            // Bundled (optional) — copy into the cache so the path is stable.
            if let bundled = Bundle.main.url(forResource: "vmlinux-arm64", withExtension: nil),
                (try? FileManager.default.copyItem(at: bundled, to: cacheURL)) != nil
            {
                return cacheURL
            }
            // Opportunistic: reuse Apple `container` CLI's kernel if it's already on disk (no download).
            let cliKernel = FileManager.default.homeDirectoryForCurrentUser
                .appending(path: "Library/Application Support/com.apple.container/kernels/default.kernel-arm64")
                .resolvingSymlinksInPath()
            if FileManager.default.fileExists(atPath: cliKernel.path),
                (try? FileManager.default.copyItem(at: cliKernel, to: cacheURL)) != nil
            {
                return cacheURL
            }
        }
        // Download it (background, one-time, cached). The raw blob GET gives no incremental size,
        // so this phase shows as an indeterminate "Downloading runtime kernel" bar — fine, since the
        // kernel is small next to the sandbox image that follows.
        beginDownload(.kernel)
        defer { endDownload() }
        try await downloadKernel(to: cacheURL)
        return cacheURL
    }

    /// Download the kernel from the `ProjectSandbox.kernelImage` GHCR package to `dest`. The kernel
    /// ships as an OCI artifact (one raw `vmlinux-arm64` blob), so this speaks the registry's
    /// distribution API directly: exchange credentials for a pull token, fetch the manifest, then
    /// fetch the single layer blob. A token is **optional** — when the package is public the pull is
    /// anonymous (the point of shipping the kernel as a package: it can be public while the repo stays
    /// private). When the package is private it authenticates with the same GitHub token used for the
    /// sandbox image. Resolution mirrors `registryAuth`.
    private func downloadKernel(to dest: URL) async throws {
        do {
            try await OCIArtifact.fetchSingleBlob(imageRef: ProjectSandbox.kernelImage, to: dest)
        } catch let e as OCIArtifact.FetchError {
            // Re-wrap with the kernel-specific recovery hint (a local `NUCLEIC_KERNEL_PATH`).
            switch e {
            case let .authFailed(ref, hadToken):
                throw ContainerError.unavailable(hadToken
                    ? "could not authenticate to \(ref) for the sandbox kernel — check the token's "
                        + "`read:packages` scope, or set NUCLEIC_KERNEL_PATH to a local kernel."
                    : "the sandbox kernel package \(ref) appears private and no GitHub token is "
                        + "configured. Make the package public, add a token (Settings → Git access, "
                        + "`read:packages`), or set NUCLEIC_KERNEL_PATH to a local kernel.")
            default:
                throw ContainerError.imagePullFailed("sandbox kernel: \(e)")
            }
        }
    }

    /// Split a registry reference `host/path…/name[:tag]` into its host, repository (`path…/name`),
    /// and tag (defaulting to `latest`). The host is everything before the first `/`; once it's
    /// stripped, any remaining `:` is the tag separator (a `host:port` colon precedes the first `/`).
    static func parseImageRef(_ ref: String) throws -> (host: String, repo: String, tag: String) {
        guard let firstSlash = ref.firstIndex(of: "/") else {
            throw ContainerError.imagePullFailed("malformed image reference: \(ref)")
        }
        let host = String(ref[..<firstSlash])
        let remainder = String(ref[ref.index(after: firstSlash)...])
        if let colon = remainder.lastIndex(of: ":") {
            let tag = String(remainder[remainder.index(after: colon)...])
            return (host, String(remainder[..<colon]), tag.isEmpty ? "latest" : tag)
        }
        return (host, remainder, "latest")
    }

    /// The vminitd initial filesystem, pulled (once per `vminitReference`) and materialized to a
    /// cached ext4. Mirrors the framework `ContainerManager`'s own caching: materialize via
    /// `InitImage.initBlock` when the cache is absent, otherwise reconstruct the read-only block
    /// mount over the cached file. No bundling, no cross-compile — just a registry pull.
    ///
    /// The cache is keyed on the image reference via a sidecar file: a bare `vminit.ext4` used to be
    /// trusted forever, so repointing `vminitReference` at a new guest revision silently kept booting
    /// whatever initfs was materialized first — a guest-patch rollout that never actually deployed.
    /// A reference mismatch (or missing sidecar, the pre-sidecar installs) re-pulls; if that pull
    /// fails (offline, unpublished tag) an existing cache is still used — a stale guest beats no
    /// containers at all — and the sidecar stays unwritten so the next launch retries.
    private func ensureInitfs() async throws -> Containerization.Mount {
        let dir = storageRoot.appendingPathComponent("initfs", isDirectory: true)
        try FileManager.default.createDirectory(at: dir, withIntermediateDirectories: true)
        let cacheURL = dir.appendingPathComponent("vminit.ext4")
        let refURL = dir.appendingPathComponent("vminit.ext4.reference")
        let cachedMount = Containerization.Mount.block(
            format: "ext4", source: cacheURL.path, destination: "/", options: ["ro"])
        let haveCache = FileManager.default.fileExists(atPath: cacheURL.path)
        if haveCache {
            let cachedRef = (try? String(contentsOf: refURL, encoding: .utf8))?
                .trimmingCharacters(in: .whitespacesAndNewlines)
            if cachedRef == Self.vminitReference {
                return cachedMount
            }
        }
        beginDownload(.initfs)
        defer { endDownload() }
        let store = try sharedImageStore()
        let initImage: InitImage
        do {
            initImage = try await store.getInitImage(
                reference: Self.vminitReference,
                auth: registryAuth(for: Self.vminitReference),
                progress: { [self] events in await advanceDownload(events, phase: .initfs) })
        } catch {
            if haveCache { return cachedMount }
            throw ContainerError.imagePullFailed("vminit (\(Self.vminitReference)): \(error)")
        }
        try? FileManager.default.removeItem(at: cacheURL)
        let mount = try await initImage.initBlock(at: cacheURL, for: .linuxArm)
        try? Self.vminitReference.write(to: refURL, atomically: true, encoding: .utf8)
        return mount
    }

    /// Build (once) the image store, content-addressed under the engine's storage root.
    func sharedImageStore() throws -> ImageStore {
        if let imageStore { return imageStore }
        let dir = storageRoot.appendingPathComponent("imagestore", isDirectory: true)
        try FileManager.default.createDirectory(at: dir, withIntermediateDirectories: true)
        let store = try ImageStore(path: dir)
        self.imageStore = store
        return store
    }

    /// HTTP Basic credentials for pulling `reference`, or `nil` for an anonymous pull. Credentials
    /// are scoped to the **registry that hosts the sandbox image** (the host of
    /// `ProjectSandbox.defaultImage`) — so they're never attached to a *different* registry such as
    /// `docker.io`, which pulls public bases anonymously and rejects unrelated credentials (401).
    /// This lets the `nucleic-sandbox` package stay private without breaking public base-image pulls.
    ///
    /// Token precedence (for that host): the `NUCLEIC_REGISTRY_TOKEN` env, else — when the private
    /// host is `ghcr.io` — the app's GitHub token (`GitHubCredentialStore`, needs `read:packages`).
    /// Username: `NUCLEIC_REGISTRY_USER` env, else the configured `registryUsername`, else the
    /// reference's owner segment (GHCR ignores the username when the token is valid).
    func registryAuth(for reference: String) -> (any ContainerizationOCI.Authentication)? {
        let env = ProcessInfo.processInfo.environment
        let segments = reference.split(separator: "/").map(String.init)
        let host = segments.first ?? ""

        // Only authenticate against the registry our private sandbox image lives on.
        let privateHost = ProjectSandbox.defaultImage.split(separator: "/").first.map(String.init) ?? "ghcr.io"
        guard host == privateHost else { return nil }

        let token: String?
        if let t = env["NUCLEIC_REGISTRY_TOKEN"], !t.isEmpty {
            token = t
        } else if privateHost == "ghcr.io", let t = GitHubCredentialStore.loadToken(), !t.isEmpty {
            token = t
        } else {
            token = nil
        }
        guard let token else { return nil }

        let owner = segments.count >= 2 ? segments[1] : nil
        let username = env["NUCLEIC_REGISTRY_USER"].flatMap { $0.isEmpty ? nil : $0 }
            ?? ContainerServiceSettings.registryUsername
            ?? owner
            ?? "x-access-token"
        return BasicAuthentication(username: username, password: token)
    }

    /// Allocate a NAT-mode network interface for a container, returning the interface and the host
    /// gateway address the guest reaches the Mac on (the subnet `.1`). The shared `VmnetNetwork` is
    /// built on first use. Internal (not private) so the rootfs-builder in the extension can use it.
    func makeInterface(for name: String) throws -> (any Interface, String) {
        var net = try network ?? VmnetNetwork(mode: .VMNET_SHARED_MODE)
        guard let iface = try net.createInterface(name) else {
            throw ContainerError.gatewayUnavailable
        }
        self.network = net  // write back the mutation (createInterface is `mutating`)
        guard let gateway = iface.ipv4Gateway?.description else {
            throw ContainerError.gatewayUnavailable
        }
        return (iface, gateway)
    }

    // MARK: - Lifecycle

    /// Ensure the named container is running and return its name plus the host-gateway IP (the
    /// container's view of the Mac, for routing the approval MCP server). Idempotent: a container
    /// already live in the registry is returned as-is; otherwise it is created from a clone of the
    /// cached provisioned rootfs, started, and (when freshly cloned) seeded.
    ///
    /// Concurrency: the cold-start path is coalesced per name (see ``starting``) — callers racing to
    /// start the same container all await the one start rather than each running the create path.
    public func ensureRunning(_ spec: ContainerSpec) async throws
        -> (name: String, hostGateway: String)
    {
        lastSpec[spec.name] = spec
        if let existing = live[spec.name] {
            return (spec.name, existing.gateway)
        }
        // Join an already-running start for this name instead of racing it. A failed start is not
        // cached: the joiner sees the same error, and the next `ensureRunning` starts fresh.
        if let inFlight = starting[spec.name] {
            return try await inFlight.value
        }
        // The task clears its own slot as it finishes (rather than the caller doing it on return),
        // so a caller that's cancelled mid-join can't drop the entry while the start is still
        // running — which would let the next caller kick off a second, racing start.
        let task = Task<(name: String, hostGateway: String), Error> {
            defer { starting[spec.name] = nil }
            return try await performStart(spec)
        }
        starting[spec.name] = task
        return try await task.value
    }

    /// The actual cold-start path — only ever entered once per name at a time, via ``ensureRunning``.
    private func performStart(_ spec: ContainerSpec) async throws
        -> (name: String, hostGateway: String)
    {
        let vmm = try await sharedVMM()
        let (rootfs, freshlyCloned, needsSeed) = try await instanceRootfs(for: spec)
        let (iface, gateway) = try makeInterface(for: spec.name)

        let memoryBytes = UInt64(spec.memoryGiB) * 1024 * 1024 * 1024
        // Attach a memory balloon unless the policy is disabled, so the autoballoon loop can return
        // idle guest memory to the host (see MemoryBalloon / ContainerServiceSettings.memoryManagement).
        let balloon: MemoryBalloon? =
            ContainerServiceSettings.memoryManagement.policy.enabled ? MemoryBalloon() : nil
        let container: LinuxContainer
        do {
            container = try LinuxContainer(spec.name, rootfs: rootfs, vmm: vmm) { config in
                // Keep the VM alive for `exec`: a long-lived init process (the agent and all
                // tooling run as subsequent `exec`s, exactly like `container exec` did). In narOS
                // images that init is **naros-init** (docs/NAROS.md §5): real zombie reaping,
                // signal forwarding, and — when a control socket is relayed in (NAROS_BRIDGE=1) —
                // supervised restart-with-backoff of the in-container control bridge (loopback
                // TCP → the relayed socket). The root init can reach the root-owned relayed
                // socket while the non-root agent only ever touches loopback TCP. Non-narOS
                // images (legacy v7, custom) fall back to the old arrangement in the same probe:
                // a restart-looped bridge under a sleep keepalive. The bridge is SUPERVISED
                // either way — it is the single in-guest hop every session's MCP/control traffic
                // crosses, so a bridge that dies (guest OOM kill, a crash) without a restart
                // would sever the whole container's control plane until the container is
                // recreated.
                let narosInit = "/usr/sbin/naros-init"
                if spec.controlSocketHostPath != nil {
                    // When the token-proxy socket is relayed (opt-in), pass its bridge port + guest
                    // socket to the SAME `control-bridge.js` via env — it opens a second forwarder
                    // (loopback `proxyBridgePort` → the relayed proxy socket) alongside the control
                    // one. The env is set inline before both the `naros-init` exec (whose supervised
                    // `node` inherits it) and the non-narOS fallback `node`, so no init change is
                    // needed. Absent when the proxy isn't relayed → the bridge opens only the control
                    // forwarder, exactly as before.
                    let proxyBridgeEnv = spec.proxySocketHostPath != nil
                        ? "NUCLEIC_PROXY_BRIDGE_PORT=\(ContainerSpec.proxyBridgePort) "
                            + "NUCLEIC_PROXY_BRIDGE_SOCKET=\(ContainerSpec.proxySocketGuestPath) "
                        : ""
                    config.process.arguments = [
                        "/bin/sh", "-c",
                        "if [ -x \(narosInit) ]; then \(proxyBridgeEnv)NAROS_BRIDGE=1 exec \(narosInit); fi; "
                            + "(while true; do \(proxyBridgeEnv)node \(Self.controlBridgeGuestPath); sleep 1; done) & "
                            + "exec /bin/sleep infinity",
                    ]
                } else {
                    config.process.arguments = [
                        "/bin/sh", "-c",
                        "if [ -x \(narosInit) ]; then exec \(narosInit); fi; "
                            + "exec /bin/sleep infinity",
                    ]
                }
                config.cpus = spec.cpus
                config.memoryInBytes = memoryBytes
                config.interfaces = [iface]
                // NAT DNS lives at the gateway; the agent's tooling needs name resolution.
                config.dns = DNS(nameservers: [gateway])
                // Bind mounts: host dir → guest path at identical paths (virtiofs preserves the
                // host owner, so the agent — run as the host uid via `exec --user` — owns them).
                for mount in spec.mounts {
                    config.mounts.append(
                        .share(
                            source: mount.host, destination: mount.container,
                            options: mount.readOnly ? ["ro"] : []))
                }
                // Relay the per-container control socket in over vsock when set: the host approval
                // server + interceptor endpoint appears in the guest at `controlSocketGuestPath`,
                // so the agent reaches the host with no IP listener (no macOS network prompts). The
                // host side is dialed lazily per guest connection, so the server need only be
                // listening before the agent execs. `permissions` is left default here; the guest
                // socket's access for the non-root agent uid is set when the bridge lands.
                if let controlSocket = spec.controlSocketHostPath {
                    var sockets = [
                        UnixSocketConfiguration(
                            source: URL(fileURLWithPath: controlSocket),
                            destination: URL(fileURLWithPath: ContainerSpec.controlSocketGuestPath),
                            direction: .into)
                    ]
                    // Second relayed socket for the opt-in Claude token proxy (the in-guest bridge
                    // forwards `proxyBridgePort` to it). Only when the host started a proxy for this
                    // container; otherwise the session stays on the static token, unrelayed.
                    if let proxySocket = spec.proxySocketHostPath {
                        sockets.append(
                            UnixSocketConfiguration(
                                source: URL(fileURLWithPath: proxySocket),
                                destination: URL(fileURLWithPath: ContainerSpec.proxySocketGuestPath),
                                direction: .into))
                    }
                    config.sockets = sockets
                }
                if let balloon { config.vmExtensions = [balloon] }  // [vendored patch] attach balloon
            }
        } catch {
            // The interface was allocated above but the VM never came up, so it'll never be released
            // via `stop` (nothing landed in `live`). Release it here so a retry can re-allocate the
            // same name instead of failing with "allocation with id … already exists".
            try? network?.releaseInterface(spec.name)
            throw ContainerError.startFailed(String(describing: error))
        }
        do {
            try await container.create()
            try await container.start()
        } catch {
            // A failed create/start can leave a created-but-unstarted VM holding resources; tear it
            // down explicitly (best-effort) before releasing the interface, same as above.
            try? await container.stop()
            try? network?.releaseInterface(spec.name)
            throw ContainerError.startFailed(String(describing: error))
        }

        live[spec.name] = LiveContainer(
            container: container, gateway: gateway, freshlyCloned: freshlyCloned, balloon: balloon,
            memoryCeilingBytes: memoryBytes)
        syncBackgroundActivity()
        if balloon != nil { startBalloonLoopIfNeeded() }

        // Seed a fresh clone, and also re-seed a reused clone whose interceptor shims an app update has
        // superseded (`needsSeed` — see `instanceRootfs`). Seeding is idempotent (it reinstalls the
        // shims + git safe.directory), so re-running it in place safely refreshes the instrumentation
        // without discarding the container's accumulated working files.
        if needsSeed {
            await seed(spec, in: container)
            recordSeedRecipe(for: spec.name)
        }
        if freshlyCloned {
            // The vsock control plane is mandatory for agent containers, and its in-guest half —
            // `node` + the control bridge — rides in the image (sandbox image ≥ v4). A custom
            // image missing either would otherwise surface only as the agent CLI's opaque
            // "Available MCP tools: none" once the agent can't reach the host; probe once per
            // fresh clone and fail the start loudly and actionably instead.
            if spec.controlSocketHostPath != nil {
                try await verifyControlBridge(spec, in: container)
            }
        }
        return (spec.name, gateway)
    }

    /// Run `argv` inside the named container as a fresh `exec`, returning a ``ProcessHandle`` whose
    /// stdio is the guest process's stdio (over vsock). The agent process flows through this — the
    /// backend treats it identically to a host spawn. A unique exec id is generated per call.
    public func exec(
        name: String, workdir: String, env: [String: String], argv: [String],
        uid: Int? = nil, gid: Int? = nil, memoryLimitBytes: UInt64? = nil
    ) async throws -> any ProcessHandle {
        guard let live = live[name] else { throw ContainerError.notRunning(name) }

        let (outStream, outCont) = AsyncThrowingStream<Data, Error>.makeStream()
        let (errStream, errCont) = AsyncThrowingStream<Data, Error>.makeStream()
        let stdin = StreamReader()
        let outWriter = StreamWriter(outCont)
        let errWriter = StreamWriter(errCont)
        let execID = "exec-\(UUID().uuidString.prefix(8))"

        let process = try await live.container.exec(execID) { p in
            p.arguments = argv  // argv[0] is the executable
            p.environmentVariables = env.map { "\($0)=\($1)" }
            p.workingDirectory = workdir
            p.user = ContainerizationOCI.User(uid: UInt32(uid ?? 0), gid: UInt32(gid ?? 0))
            p.terminal = false  // pipe mode: keep stdout and stderr separable
            p.stdin = stdin
            p.stdout = outWriter
            p.stderr = errWriter
            // Hard per-exec memory ceiling → the guest applies it as this exec's own cgroup memory.max
            // (vendored containerization patch #9 + follow-up). nil = inherit the container's ceiling.
            p.memoryLimitInBytes = memoryLimitBytes
        }
        try await process.start()
        return ContainerizedProcessHandle(
            process: process, stdin: stdin, stdoutWriter: outWriter, stderrWriter: errWriter,
            stdoutLines: outStream, stderrLines: errStream)
    }

    /// Run `argv` to completion in the named **running** container (as `uid`/`gid`), draining and
    /// returning its combined stdout+stderr (line-joined, tail-capped) plus the exit code. Returns
    /// `nil` only when the container isn't live. Unlike ``exec`` — which streams a long-lived agent
    /// process — this is for short one-shot maintenance commands (the in-place CLI updater). Both
    /// streams are drained concurrently with the wait so neither back-pressures the guest.
    public func runCapturing(
        name: String, argv: [String], env: [String: String] = [:], uid: Int = 0, gid: Int = 0
    ) async -> (exitCode: Int32, output: String)? {
        guard live[name] != nil else { return nil }
        let handle: any ProcessHandle
        do {
            handle = try await exec(name: name, workdir: "/", env: env, argv: argv, uid: uid, gid: gid)
        } catch {
            return (exitCode: -1, output: "\(error)")
        }
        func drain(_ stream: AsyncThrowingStream<Data, Error>) async -> [String] {
            var acc: [String] = []
            do {
                for try await line in stream where acc.count < 500 {
                    if let s = String(data: line, encoding: .utf8) { acc.append(s) }
                }
            } catch { /* stream error → return what we collected */ }
            return acc
        }
        async let out = drain(handle.stdoutLines)
        async let err = drain(handle.stderrLines)
        let code = await handle.wait()
        let lines = await (out + err)
        // Cap the reported tail so a chatty install can't bloat the result surfaced in the UI.
        let tail = lines.suffix(60).joined(separator: "\n")
        return (exitCode: code, output: tail)
    }

    /// Stop the container's VM (freeing its memory) but keep its on-disk rootfs clone, so the next
    /// `ensureRunning` recreates it fast and skips re-seeding. Drops the registry entry; best-effort.
    public func stop(name: String) async {
        guard let entry = live[name] else { return }
        try? await entry.container.stop()
        live[name] = nil
        // Release the container's vmnet address allocation so the same name can get a fresh
        // interface on the next `ensureRunning`. The allocator keys reservations by container id;
        // without this release it still holds `name`, and the recreate's `createInterface(name)`
        // throws "allocation with id \(name) already exists" — turning every idle-stop / restart
        // into a container that can't be brought back up (e.g. sending a chat after an idle stop).
        try? network?.releaseInterface(name)
        syncBackgroundActivity()
    }

    /// Begin/end the process-activity assertion so it tracks `live`: hold it while any VM is running,
    /// release it once the last one stops. Idempotent — safe to call after every `live` mutation.
    /// Deliberately *allows* idle system sleep: a sleeping Mac suspends and later resumes the VM,
    /// which is fine, and blocking sleep would needlessly drain the battery. It only blocks App Nap
    /// and the app being terminated out from under a live container.
    private func syncBackgroundActivity() {
        if !live.isEmpty, backgroundActivity == nil {
            backgroundActivity = ProcessInfo.processInfo.beginActivity(
                options: [
                    .userInitiatedAllowingIdleSystemSleep, .suddenTerminationDisabled,
                    .automaticTerminationDisabled,
                ],
                reason: "Nucleic sandbox container running")
        } else if live.isEmpty, let activity = backgroundActivity {
            ProcessInfo.processInfo.endActivity(activity)
            backgroundActivity = nil
        }
    }

    /// Reclaim VM memory by stopping then recreating the container in place. The rootfs clone and
    /// its instrumentation persist on disk, so the recreate reuses them (no re-seed). No-op if the
    /// engine never saw this container (no spec cached). Returns `false` when the recreate failed
    /// (the container is now down until the next `ensureRunning` retries), so callers don't read a
    /// swallowed failure as a successful restart.
    @discardableResult
    public func restart(name: String) async -> Bool {
        guard let spec = lastSpec[name] else {
            await stop(name: name)
            return false
        }
        await stop(name: name)
        return (try? await ensureRunning(spec)) != nil
    }

    /// Rebuild the one vmnet NAT shared by every container in this app process. A normal container
    /// restart intentionally reuses `network`; that is why a wedged vmnet DNS proxy used to survive
    /// every agent retry and only quitting Nucleic fixed it. `ContainerManager` invokes this behind
    /// a global idle barrier, so stopping all live (but idle) VMs cannot interrupt a turn. Rootfs
    /// clones and `lastSpec` remain, making the next start a fast recreate on a new vmnet reference.
    public func resetNetwork() async {
        for name in Array(live.keys) {
            await stop(name: name)
        }
        network = nil
    }

    /// Stop the container and delete its on-disk rootfs clone. 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 clone = instanceRootfsURL(for: name)
        try? FileManager.default.removeItem(at: clone)
        return !FileManager.default.fileExists(atPath: clone.path)
    }

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

    /// True if the container is running, or has a persisted rootfs clone on disk (stopped but
    /// recreatable without a rebuild).
    public func containerExists(_ name: String) async -> Bool {
        live[name] != nil || FileManager.default.fileExists(atPath: instanceRootfsURL(for: name).path)
    }

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

    // MARK: - Resource usage

    /// Sample the container's live CPU and memory for the Control panel's resource monitor. CPU is a
    /// real delta: two `statistics()` reads around a ~200 ms sleep, normalized across the container's
    /// cores to 0…100. Best-effort: `nil` when the container isn't running or the probe fails.
    public func sampleResourceUsage(name: String) async -> ContainerResourceSample? {
        guard let entry = live[name] else { return nil }
        guard let s1 = try? await entry.container.statistics() else { return nil }
        let t1 = DispatchTime.now().uptimeNanoseconds
        try? await Task.sleep(nanoseconds: 200_000_000)
        guard let s2 = try? await entry.container.statistics() else { return nil }
        let t2 = DispatchTime.now().uptimeNanoseconds

        // `cpu.usageUsec` is cumulative CPU time across all cores. Busy fraction over the window,
        // normalized by the core count, gives an aggregate 0…100 utilization. Divide by the ACTUAL
        // elapsed wall-clock (the sleep plus scheduling/await latency), not the requested 200 ms —
        // under load the real interval is longer, and assuming 200 ms over-reads (a true 60% can
        // show as 100%). Guard against a zero/negative interval.
        let windowUsec = max(1.0, Double(t2 &- t1) / 1000.0)
        let cores = Double(max(1, entry.container.cpus))
        let deltaUsec = max(0, Double(s2.cpu?.usageUsec ?? 0) - Double(s1.cpu?.usageUsec ?? 0))
        let cpuPercent = deltaUsec / (windowUsec * cores) * 100
        return ContainerResourceSample(
            cpuPercent: cpuPercent,
            memoryUsedBytes: s2.memory?.usageBytes ?? 0,
            memoryTotalBytes: s2.memory?.limitBytes ?? 0)
    }

    // MARK: - Automatic memory reclamation (balloon)

    /// Start the autoballoon driver if it isn't already running. Idempotent.
    private func startBalloonLoopIfNeeded() {
        guard balloonLoop == nil else { return }
        balloonLoop = Task { [weak self] in
            while !Task.isCancelled {
                guard let self else { return }
                let nextDelaySeconds = await self.adjustBalloons()
                if nextDelaySeconds == 0 { return }  // nothing left to drive — stop
                try? await Task.sleep(nanoseconds: nextDelaySeconds * 1_000_000_000)
            }
        }
    }

    /// One autoballoon pass: under the current memory-management policy, retarget every live
    /// container's balloon from its live working set. Returns the seconds to wait before the next
    /// pass, or `0` to stop the loop (no live containers — it restarts on the next `ensureRunning`).
    /// The target always leaves headroom over the working set (never inflates below resident memory,
    /// which would force guest reclaim/OOM) and never drops below the policy floor.
    private func adjustBalloons() async -> UInt64 {
        if live.isEmpty {
            balloonLoop = nil
            return 0
        }
        let policy = ContainerServiceSettings.memoryManagement.policy
        // Policy is read each pass, so it can change live. A disabled policy idle-polls so a later
        // change resumes without needing a container restart.
        guard policy.enabled else { return 30 }
        for entry in live.values {
            guard let balloon = entry.balloon else { continue }
            // A failed/empty stats read must NEVER leave the balloon at a stale inflated target —
            // that pins the guest below its footprint with no way out (the non-recovering freeze).
            // Fully deflate (target → the configured ceiling) so the guest gets all its memory back,
            // then try again next pass.
            guard let stats = try? await entry.container.statistics(),
                let used = stats.memory?.usageBytes, used > 0,
                let ceiling = stats.memory?.limitBytes, ceiling > 0
            else {
                balloon.setTarget(entry.memoryCeilingBytes, hysteresis: 0)
                continue
            }
            let target = policy.target(workingSetBytes: used, ceilingBytes: ceiling)
            balloon.setTarget(target, hysteresis: policy.hysteresisBytes)
        }
        return max(1, policy.cadenceSeconds)
    }

    /// Force an immediate balloon pass (the "Reclaim memory now" action), so reclamation doesn't
    /// wait for the next scheduled tick. No-op when ballooning is off.
    public func reclaimMemoryNow() async {
        _ = await adjustBalloons()
    }

    /// End-to-end health of a container's control plane — the loopback-TCP → control-bridge →
    /// relayed-unix-socket → host-approval-server path every agent's MCP traffic crosses.
    public enum ControlPlaneProbe: Sendable, Equatable {
        /// A probe request got response bytes back: bridge, relay, and approval server all work.
        case responsive
        /// TCP connect to the bridge failed — the in-guest bridge is down (crashed / not started).
        case bridgeUnreachable(String)
        /// TCP connect succeeded but no bytes came back within the deadline: something past the
        /// bridge (the guest relay listener, the vsock hop, or the host server's accept path) is
        /// accepting connections it never services — the black-hole wedge.
        case unresponsive
        /// The probe itself couldn't run (exec failed / container gone); no verdict.
        case probeFailed(String)
    }

    /// Probe the named container's control plane from INSIDE the guest, exactly the way an agent
    /// reaches it: TCP to the loopback bridge port, one plain HTTP request, then wait briefly for
    /// ANY response bytes (an unauthenticated request still gets an HTTP error line — bytes back
    /// prove the whole path). Runs via `exec`, whose per-exec vsock stdio is independent of the
    /// control plane, so the probe works precisely when the control plane doesn't. Used by the
    /// spawn-stall watchdog to say WHICH hop is dead instead of guessing.
    public func probeControlPlane(name: String) async -> ControlPlaneProbe {
        let script = """
            const net = require('net');
            let done = false;
            const finish = (code, msg) => { if (done) return; done = true; console.log(msg); process.exit(code); };
            const s = net.connect(\(ContainerSpec.controlBridgePort), '127.0.0.1');
            s.on('connect', () => { s.write('GET /mcp HTTP/1.1\\r\\nHost: nucleic\\r\\nConnection: close\\r\\n\\r\\n'); });
            s.on('data', () => finish(0, 'RESPONSIVE'));
            s.on('error', (e) => finish(2, 'REFUSED:' + (e && e.code)));
            setTimeout(() => finish(3, 'SILENT'), 3000);
            """
        guard let result = await runCapturing(name: name, argv: ["node", "-e", script]) else {
            return .probeFailed("container isn't running")
        }
        switch result.exitCode {
        case 0: return .responsive
        case 2: return .bridgeUnreachable(result.output)
        case 3: return .unresponsive
        default: return .probeFailed("exit \(result.exitCode): \(result.output)")
        }
    }

    /// Post-mortem probe the moment an agent exits 137/SIGKILL, to tell apart the causes that all
    /// look like "status 137". One typed `statistics()` read: the cgroup `oom_kill` counter is the
    /// persistent, authoritative count of kernel OOM kills (it survives the kill). Three distinct
    /// findings: not registered (`containerRunning: false` — the app's own runtime stopped/restarted
    /// it), registered but silent even after a retried probe (`probeUnanswered` — the VM crashed or
    /// is wedged), or an answered probe with the OOM evidence. The retry matters: the balloon loop
    /// shows single `statistics()` reads DO fail transiently, and a one-off failure must not be
    /// reported as "the container went down".
    public func diagnoseKill(name: String) async -> ContainerKillDiagnosis {
        guard let entry = live[name] else {
            return ContainerKillDiagnosis(containerRunning: false, oomKills: nil, sample: nil)
        }
        var probed = try? await entry.container.statistics()
        if probed == nil {
            try? await Task.sleep(nanoseconds: 300_000_000)
            probed = try? await entry.container.statistics()
        }
        guard let stats = probed else {
            return ContainerKillDiagnosis(
                containerRunning: true, probeUnanswered: true, oomKills: nil, sample: nil)
        }
        var sample: ContainerResourceSample?
        if let mem = stats.memory {
            sample = ContainerResourceSample(
                cpuPercent: 0, memoryUsedBytes: mem.usageBytes, memoryTotalBytes: mem.limitBytes)
        }
        return ContainerKillDiagnosis(
            containerRunning: true,
            oomKills: stats.memoryEvents.map { Int($0.oomKill) },
            sample: sample)
    }
}

/// The Apple-containerization engine IS the macOS `SandboxEngine` (docs/WINDOWS_PORT.md §3.1):
/// every requirement is an existing method above (or in the Rootfs extension) with its
/// signature lifted verbatim into the protocol, so the conformance is empty by construction.
extension ContainerEngine: SandboxEngine {}
