Merge claude/nucleic-stdio-stall-391480 into dev

Sources/NucleicCore/Claude/MCPApprovalServer.swift

@@ -3292,6 +3292,18 @@ private final class NWByteConn: ByteConn, @unchecked Sendable {
 /// suspended on an approval holds no thread (no read is in flight between request and response). The
 /// control plane's live connection count is small (one MCP connection plus short-lived interceptor
 /// posts per container), so a thread parked on `read` per live connection is acceptable.
+///
+/// **fd lifetime (load-bearing).** `close()` only `shutdown(2)`s the socket; the descriptor itself
+/// is released exactly once, in `deinit`, and every offloaded read/write block captures `self` — so
+/// the fd *number* cannot be recycled while any block that could still pass it to the kernel is in
+/// flight. A straggler `receive`/`send` after `close()` sees the shut-down socket (instant EOF /
+/// EPIPE), never a stranger's descriptor. The previous revision `close(2)`d in `close()`: the serve
+/// loop's next `receive` — dispatched concurrently with an SSE handler's mid-loop `close()` (see
+/// `respondStreamingToolCall`) — could then issue a blocking `read(2)` on a number the kernel had
+/// already handed to a NEW fd (another session's MCP socket, a container's stdio channel…), parking
+/// a global-queue thread forever on — and stealing bytes from — an unrelated stream. Those zombie
+/// threads/reads accumulated until every container session's stdio stalled at once (the "produced
+/// no output within 60s / stdio transport stalled" lockup).
 private final class UnixSocketByteConn: ByteConn, @unchecked Sendable {
     private let fd: Int32
     private let lock = NSLock()
@@ -3308,37 +3320,61 @@ private final class UnixSocketByteConn: ByteConn, @unchecked Sendable {
         #endif
     }
 
+    deinit {
+        // The sole close(2) of the descriptor. Every offloaded block strongly captures `self`, so
+        // deinit is ordered after the last kernel call that could reference this fd number.
+        posixCloseFD(fd)
+    }
+
+    private var isClosed: Bool {
+        lock.lock()
+        defer { lock.unlock() }
+        return closed
+    }
+
     func receive(maxLength: Int) async throws -> Data? {
-        let fd = self.fd
-        return try await withCheckedThrowingContinuation { cont in
-            DispatchQueue.global(qos: .userInitiated).async {
-                var buffer = [UInt8](repeating: 0, count: maxLength)
-                while true {
-                    let n = buffer.withUnsafeMutableBytes { read(fd, $0.baseAddress, maxLength) }
-                    if n > 0 {
-                        cont.resume(returning: Data(buffer[0..<n]))
-                    } else if n == 0 {
-                        cont.resume(returning: nil)
-                    } else if errno == EINTR {
-                        continue
-                    } else {
-                        cont.resume(throwing: POSIXError(POSIXErrorCode(rawValue: errno) ?? .EIO))
+        // Fast EOF once closed (the serve loop re-enters receive after an SSE handler's mid-loop
+        // close). Racing closes are safe regardless: the fd stays valid until deinit, and shutdown
+        // makes a late read return 0.
+        guard !isClosed else { return nil }
+        return try await withTaskCancellationHandler {
+            try await withCheckedThrowingContinuation { cont in
+                DispatchQueue.global(qos: .userInitiated).async {
+                    var buffer = [UInt8](repeating: 0, count: maxLength)
+                    while true {
+                        let n = buffer.withUnsafeMutableBytes {
+                            read(self.fd, $0.baseAddress, maxLength)
+                        }
+                        if n > 0 {
+                            cont.resume(returning: Data(buffer[0..<n]))
+                        } else if n == 0 {
+                            cont.resume(returning: nil)
+                        } else if errno == EINTR {
+                            continue
+                        } else {
+                            cont.resume(throwing: POSIXError(POSIXErrorCode(rawValue: errno) ?? .EIO))
+                        }
+                        return
                     }
-                    return
                 }
             }
+        } onCancel: {
+            // Task cancellation (server stop / connection teardown) can't interrupt a blocking
+            // read(2); shutting the socket down wakes the parked read with EOF so the serve loop
+            // actually exits instead of stranding the thread until the peer hangs up.
+            self.close()
         }
     }
 
     func send(_ data: Data) async throws {
-        let fd = self.fd
+        guard !isClosed else { throw POSIXError(.EPIPE) }
         try await withCheckedThrowingContinuation { (cont: CheckedContinuation<Void, Error>) in
             DispatchQueue.global(qos: .userInitiated).async {
                 let result: Result<Void, Error> = data.withUnsafeBytes { raw in
                     guard let base = raw.baseAddress else { return .success(()) }
                     var offset = 0
                     while offset < raw.count {
-                        let n = writeNoSigpipe(fd, base + offset, raw.count - offset)
+                        let n = writeNoSigpipe(self.fd, base + offset, raw.count - offset)
                         if n > 0 {
                             offset += n
                         } else if n < 0 && errno == EINTR {
@@ -3359,8 +3395,9 @@ private final class UnixSocketByteConn: ByteConn, @unchecked Sendable {
         defer { lock.unlock() }
         guard !closed else { return }
         closed = true
+        // Wakes any parked read with EOF and fails later writes with EPIPE/ENOTCONN. Deliberately
+        // NO close(2) here — deinit releases the descriptor (see the class comment).
         shutdown(fd, Int32(SHUT_RDWR))  // Glibc spells the constant as Int; Int32 on both platforms
-        posixCloseFD(fd)
     }
 }
 

Sources/NucleicCore/Container/ContainerEngine.swift

@@ -378,16 +378,31 @@ public actor ContainerEngine {
         return (host, remainder, "latest")
     }
 
-    /// The vminitd initial filesystem, pulled (once, from `vminitReference`) and materialized to a
+    /// 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")
-        if FileManager.default.fileExists(atPath: cacheURL.path) {
-            return .block(format: "ext4", source: cacheURL.path, destination: "/", options: ["ro"])
+        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() }
@@ -396,11 +411,16 @@ public actor ContainerEngine {
         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)")
         }
-        return try await initImage.initBlock(at: cacheURL, for: .linuxArm)
+        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.

Tests/NucleicCoreTests/MCPApprovalServerTests.swift

@@ -739,6 +739,113 @@ import Testing
         #expect(await captured.argv == ["merge", "feature/x"])
     }
 
+    /// Like ``unixRequest`` but an SSE-eliciting `tools/call`: keep-alive request (no
+    /// `Connection: close`) with `Accept: text/event-stream`, so the server answers as a stream and
+    /// closes the connection itself (`respondStreamingToolCall`). Reads to EOF; returns the raw text.
+    func unixSSEToolCall(
+        socketPath: String, token: String, id: Int
+    ) async throws -> String {
+        let body: JSONValue = [
+            "jsonrpc": "2.0", "id": .number(Double(id)), "method": "tools/call",
+            "params": [
+                "name": "approve",
+                "arguments": [
+                    "tool_name": "Bash", "input": ["command": "true"],
+                    "tool_use_id": .string("t\(id)"),
+                ],
+            ],
+        ]
+        let bodyData = try body.encodedData()
+        var head = "POST /mcp HTTP/1.1\r\nHost: nucleic\r\n"
+        head += "Content-Type: application/json\r\n"
+        head += "Accept: application/json, text/event-stream\r\n"
+        head += "Authorization: Bearer \(token)\r\n"
+        head += "Content-Length: \(bodyData.count)\r\n\r\n"
+        var payload = Data(head.utf8)
+        payload.append(bodyData)
+
+        return try await withCheckedThrowingContinuation { cont in
+            DispatchQueue.global(qos: .userInitiated).async {
+                let fd = socket(AF_UNIX, SOCK_STREAM, 0)
+                guard fd >= 0 else {
+                    cont.resume(throwing: POSIXError(.EIO))
+                    return
+                }
+                defer { close(fd) }
+                var addr = sockaddr_un()
+                addr.sun_family = sa_family_t(AF_UNIX)
+                let pathBytes = Array(socketPath.utf8)
+                let capacity = MemoryLayout.size(ofValue: addr.sun_path)
+                withUnsafeMutablePointer(to: &addr.sun_path) { raw in
+                    raw.withMemoryRebound(to: CChar.self, capacity: capacity) { dst in
+                        for (i, b) in pathBytes.enumerated() { dst[i] = CChar(bitPattern: b) }
+                        dst[pathBytes.count] = 0
+                    }
+                }
+                let rc = withUnsafePointer(to: &addr) { ptr in
+                    ptr.withMemoryRebound(to: sockaddr.self, capacity: 1) {
+                        connect(fd, $0, socklen_t(MemoryLayout<sockaddr_un>.size))
+                    }
+                }
+                guard rc == 0 else {
+                    cont.resume(throwing: POSIXError(POSIXErrorCode(rawValue: errno) ?? .EIO))
+                    return
+                }
+                payload.withUnsafeBytes { raw in
+                    guard let base = raw.baseAddress else { return }
+                    var off = 0
+                    while off < raw.count {
+                        let n = write(fd, base + off, raw.count - off)
+                        if n > 0 { off += n } else { break }
+                    }
+                }
+                var received = Data()
+                var buf = [UInt8](repeating: 0, count: 1 << 16)
+                while true {
+                    let n = buf.withUnsafeMutableBytes { read(fd, $0.baseAddress, 1 << 16) }
+                    if n > 0 { received.append(contentsOf: buf[0..<n]) } else { break }
+                }
+                cont.resume(returning: String(decoding: received, as: UTF8.self))
+            }
+        }
+    }
+
+    @Test func unixSocketSurvivesSSECloseReceiveRace() async throws {
+        // Regression for the "container stdio transport stalled" lockup. An SSE-streamed
+        // `tools/call` closes its connection server-side mid-serve-loop
+        // (`respondStreamingToolCall` → `conn.close()`), racing the loop's next `receive`. The old
+        // `UnixSocketByteConn.close()` released the fd NUMBER immediately, so that racing receive
+        // could issue a blocking read(2) on a *recycled* descriptor — parking a global-queue thread
+        // forever on (and stealing bytes from) whatever unrelated stream inherited the number:
+        // another session's MCP socket, a container's stdout vsock channel. Zombie threads/reads
+        // accumulated until every container session stalled at once. Now `close()` only shuts the
+        // socket down (the fd is released at deinit, pinned by any in-flight block), so a straggler
+        // read gets EOF and fresh connections keep serving. Each round does an SSE call and then
+        // IMMEDIATELY dials a new connection — the window where fd-number reuse was likeliest;
+        // under the old code this loop wedged (stolen request bytes → the client never gets EOF).
+        let socket = "/tmp/nuc-sse-\(UUID().uuidString.prefix(8)).sock"
+        let server = MCPApprovalServer()
+        _ = try await server.start(unixSocketPath: socket)
+        defer { Task { await server.stop() } }
+        // An instant decision: the faster the handler, the wider the old close-vs-receive race.
+        await server.register(token: "sse") { _ in .deny(message: "n/a") }
+
+        for round in 0..<10 {
+            let sse = try await unixSSEToolCall(socketPath: socket, token: "sse", id: round)
+            #expect(sse.contains("text/event-stream"), "round \(round): \(sse)")
+            // The decision rides as a JSON-stringified object inside the MCP text content block,
+            // so in the raw SSE bytes its quotes arrive escaped.
+            #expect(sse.contains(#"\"behavior\":\"deny\""#), "round \(round): \(sse)")
+            let follow = try await unixRequest(
+                socketPath: socket, route: "/mcp", token: "sse",
+                body: ["jsonrpc": "2.0", "id": 1, "method": "initialize"])
+            #expect(follow.status == 200, "round \(round)")
+            #expect(
+                follow.body?["result"]?["serverInfo"]?["name"]?.stringValue == "nucleic-approval",
+                "round \(round)")
+        }
+    }
+
     @Test func unixSocketStartSelfHealsAfterFileRemoved() async throws {
         // The core self-heal: the shared control server is long-lived, so if its socket file is
         // unlinked out from under it (a `/tmp`-style reaper, a container teardown), the next session's

docs/CONTAINER_ISOLATION.md

@@ -44,6 +44,13 @@ container" concern applies only to the container's init/lifecycle agent, not to
 - **Guest-side** patches live in `third_party/containerization/vminitd/` (the guest agent, PID 1). They
   ride the **vminit initfs OCI image** and are **inert until that image is rebuilt and published** and
   `ContainerEngine.vminitReference` points at it. This is the offload (#8) and the cgroups work (#9).
+  The materialized initfs (`…/Nucleic/containers/initfs/vminit.ext4`) is cached **keyed on the image
+  reference** (a `vminit.ext4.reference` sidecar): repointing `vminitReference` re-pulls on next launch.
+  (Before the sidecar existed the first-ever materialized initfs was trusted forever, so a repoint
+  silently kept booting the old guest — verify a rollout actually landed by checking the sidecar's
+  contents.) The pull authenticates with the app's GitHub token when the image is on ghcr.io, so the
+  package can stay private; if the pull fails and a cached initfs exists, the cache is used and the
+  pull retries next launch.
 
 ## 3. The per-exec cgroup layout (#9)