format

## Motivation

The RDMA setup instructions were missing a step: after booting to
Recovery mode, users need to open Terminal from the Utilities menu
before they can run the `rdma_ctl` command. Without this step, users
following the instructions wouldn't know how to access a terminal in
Recovery mode. This step was already in the README just not in the UI
notifications.

## Changes

Added a missing instruction step — "Open Terminal from the Utilities
menu" — to three instances of the RDMA setup flow in
`dashboard/src/routes/+page.svelte`.

## Why It Works

N/A copy change only. 

## Test Plan

### Manual Testing
Hardware: MacBook Pro M4 Max 48GB

### Automated Testing
No automated tests affected; this is a UI copy change only.

Co-authored-by: Sam Bradbury <sam@consultbradbury.com>

app/EXO/EXO/ContentView.swift

@@ -16,22 +16,13 @@ struct ContentView: View {
     @EnvironmentObject private var updater: SparkleUpdater
     @EnvironmentObject private var thunderboltBridgeService: ThunderboltBridgeService
     @EnvironmentObject private var settingsWindowController: SettingsWindowController
+    @EnvironmentObject private var bugReportWindowController: BugReportWindowController
     @State private var focusedNode: NodeViewModel?
     @State private var deletingInstanceIDs: Set<String> = []
     @State private var showAllNodes = false
     @State private var showAllInstances = false
     @State private var baseURLCopied = false
     @State private var showAdvanced = false
-    @State private var showDebugInfo = false
-    private enum BugReportPhase: Equatable {
-        case idle
-        case prompting
-        case sending(String)
-        case success(String)
-        case failure(String)
-    }
-    @State private var bugReportPhase: BugReportPhase = .idle
-    @State private var bugReportUserDescription: String = ""
     @State private var uninstallInProgress = false
     @State private var pendingNamespace: String = ""
     @State private var pendingHFToken: String = ""
@@ -294,6 +285,13 @@ struct ContentView: View {
             ) {
                 updater.checkForUpdates()
             }
+            HoverButton(
+                title: "Share Bug Report…",
+                tint: .primary,
+                trailingSystemImage: "ladybug"
+            ) {
+                bugReportWindowController.open()
+            }
             .padding(.bottom, 8)
             HoverButton(title: "Quit", tint: .secondary) {
                 controller.stop()
@@ -477,40 +475,6 @@ struct ContentView: View {
         }
     }
 
-    private var debugSection: some View {
-        VStack(alignment: .leading, spacing: 4) {
-            HoverButton(
-                title: "Debug Info",
-                tint: .primary,
-                trailingSystemImage: showDebugInfo ? "chevron.up" : "chevron.down",
-                small: true
-            ) {
-                showDebugInfo.toggle()
-            }
-            if showDebugInfo {
-                VStack(alignment: .leading, spacing: 4) {
-                    Text("Version: \(buildTag)")
-                        .font(.caption2)
-                        .foregroundColor(.secondary)
-                    Text("Commit: \(buildCommit)")
-                        .font(.caption2)
-                        .foregroundColor(.secondary)
-                    Text(thunderboltStatusText)
-                        .font(.caption2)
-                        .foregroundColor(thunderboltStatusColor)
-                    clusterThunderboltBridgeView
-                    interfaceIpList
-                    rdmaStatusView
-                    sendBugReportButton
-                        .padding(.top, 6)
-                }
-                .padding(.leading, 8)
-                .transition(.opacity)
-            }
-        }
-        .animation(.easeInOut(duration: 0.25), value: showDebugInfo)
-    }
-
     private var rdmaStatusView: some View {
         let rdmaStatuses = stateService.latestSnapshot?.nodeRdmaCtl ?? [:]
         let localNodeId = stateService.localNodeId
@@ -559,127 +523,6 @@ struct ContentView: View {
         }
     }
 
-    private var sendBugReportButton: some View {
-        VStack(alignment: .leading, spacing: 6) {
-            switch bugReportPhase {
-            case .idle:
-                Button {
-                    bugReportPhase = .prompting
-                    bugReportUserDescription = ""
-                } label: {
-                    HStack {
-                        Text("Send Bug Report")
-                            .font(.caption)
-                            .fontWeight(.semibold)
-                        Spacer()
-                    }
-                    .padding(.vertical, 6)
-                    .padding(.horizontal, 8)
-                    .background(
-                        RoundedRectangle(cornerRadius: 6)
-                            .fill(Color.accentColor.opacity(0.12))
-                    )
-                }
-                .buttonStyle(.plain)
-
-            case .prompting:
-                VStack(alignment: .leading, spacing: 6) {
-                    VStack(alignment: .leading, spacing: 2) {
-                        Text("Tell us what went wrong (optional)")
-                            .font(.caption2)
-                            .foregroundColor(.secondary)
-                        Text(
-                            "A quick description of what you were doing and what happened helps us track down the bug for you."
-                        )
-                        .font(.caption2)
-                        .foregroundColor(.secondary)
-                        .opacity(0.8)
-                        .fixedSize(horizontal: false, vertical: true)
-                    }
-                    TextEditor(text: $bugReportUserDescription)
-                        .font(.caption2)
-                        .frame(height: 60)
-                        .overlay(
-                            RoundedRectangle(cornerRadius: 4)
-                                .stroke(Color.secondary.opacity(0.3), lineWidth: 1)
-                        )
-                    HStack(spacing: 8) {
-                        Button("Send") {
-                            Task {
-                                await sendBugReport()
-                            }
-                        }
-                        .font(.caption2)
-                        .buttonStyle(.borderedProminent)
-                        .controlSize(.small)
-                        Button("Cancel") {
-                            bugReportPhase = .idle
-                        }
-                        .font(.caption2)
-                        .buttonStyle(.bordered)
-                        .controlSize(.small)
-                    }
-                }
-                .padding(8)
-                .background(
-                    RoundedRectangle(cornerRadius: 6)
-                        .fill(Color.accentColor.opacity(0.06))
-                )
-
-            case .sending(let message):
-                HStack(spacing: 6) {
-                    ProgressView()
-                        .scaleEffect(0.6)
-                    Text(message)
-                        .font(.caption2)
-                        .foregroundColor(.secondary)
-                }
-
-            case .success(let message):
-                VStack(alignment: .leading, spacing: 6) {
-                    Text(message)
-                        .font(.caption2)
-                        .foregroundColor(.secondary)
-                        .fixedSize(horizontal: false, vertical: true)
-                    Button {
-                        openGitHubIssue()
-                    } label: {
-                        HStack(spacing: 4) {
-                            Image(systemName: "arrow.up.right.square")
-                                .imageScale(.small)
-                            Text("Create GitHub Issue")
-                                .font(.caption2)
-                        }
-                    }
-                    .buttonStyle(.bordered)
-                    .controlSize(.small)
-                    Button("Done") {
-                        bugReportPhase = .idle
-                        bugReportUserDescription = ""
-                    }
-                    .font(.caption2)
-                    .buttonStyle(.plain)
-                    .foregroundColor(.secondary)
-                }
-
-            case .failure(let message):
-                VStack(alignment: .leading, spacing: 4) {
-                    Text(message)
-                        .font(.caption2)
-                        .foregroundColor(.red)
-                        .fixedSize(horizontal: false, vertical: true)
-                    Button("Dismiss") {
-                        bugReportPhase = .idle
-                    }
-                    .font(.caption2)
-                    .buttonStyle(.plain)
-                    .foregroundColor(.secondary)
-                }
-            }
-        }
-        .animation(.easeInOut(duration: 0.2), value: bugReportPhase)
-    }
-
     private var processToggleBinding: Binding<Bool> {
         Binding(
             get: {
@@ -720,61 +563,6 @@ struct ContentView: View {
         )
     }
 
-    private func sendBugReport() async {
-        bugReportPhase = .sending("Collecting logs...")
-        let service = BugReportService()
-        let description = bugReportUserDescription.trimmingCharacters(in: .whitespacesAndNewlines)
-        do {
-            let outcome = try await service.sendReport(
-                isManual: true,
-                userDescription: description.isEmpty ? nil : description
-            )
-            if outcome.success {
-                bugReportPhase = .success(outcome.message)
-            } else {
-                bugReportPhase = .failure(outcome.message)
-            }
-        } catch {
-            bugReportPhase = .failure(error.localizedDescription)
-        }
-    }
-
-    private func openGitHubIssue() {
-        let description = bugReportUserDescription.trimmingCharacters(in: .whitespacesAndNewlines)
-
-        var bodyParts: [String] = []
-        bodyParts.append("## Describe the bug")
-        bodyParts.append("")
-        if !description.isEmpty {
-            bodyParts.append(description)
-        } else {
-            bodyParts.append("A clear and concise description of what the bug is.")
-        }
-        bodyParts.append("")
-        bodyParts.append("## Environment")
-        bodyParts.append("")
-        bodyParts.append("- macOS Version: \(ProcessInfo.processInfo.operatingSystemVersionString)")
-        bodyParts.append("- EXO Version: \(buildTag) (\(buildCommit))")
-        bodyParts.append("")
-        bodyParts.append("## Additional context")
-        bodyParts.append("")
-        bodyParts.append("A bug report with diagnostic logs was submitted via the app.")
-
-        let body = bodyParts.joined(separator: "\n")
-
-        var components = URLComponents(string: "https://github.com/exo-explore/exo/issues/new")!
-        components.queryItems = [
-            URLQueryItem(name: "template", value: "bug_report.md"),
-            URLQueryItem(name: "title", value: "[BUG] "),
-            URLQueryItem(name: "body", value: body),
-            URLQueryItem(name: "labels", value: "bug"),
-        ]
-
-        if let url = components.url {
-            NSWorkspace.shared.open(url)
-        }
-    }
-
     private func showUninstallConfirmationAlert() {
         let alert = NSAlert()
         alert.messageText = "Uninstall EXO"
@@ -857,13 +645,6 @@ struct ContentView: View {
         }
     }
 
-    private var buildTag: String {
-        Bundle.main.infoDictionary?["EXOBuildTag"] as? String ?? "unknown"
-    }
-
-    private var buildCommit: String {
-        Bundle.main.infoDictionary?["EXOBuildCommit"] as? String ?? "unknown"
-    }
 }
 
 private struct HoverButton: View {

app/EXO/EXO/EXOApp.swift

@@ -22,6 +22,7 @@ struct EXOApp: App {
     @StateObject private var updater: SparkleUpdater
     @StateObject private var thunderboltBridgeService: ThunderboltBridgeService
     @StateObject private var settingsWindowController: SettingsWindowController
+    @StateObject private var bugReportWindowController: BugReportWindowController
     private let terminationObserver: TerminationObserver
     private let firstLaunchPopout = FirstLaunchPopout()
     private let ciContext = CIContext(options: nil)
@@ -46,6 +47,7 @@ struct EXOApp: App {
         let thunderboltBridge = ThunderboltBridgeService(clusterStateService: service)
         _thunderboltBridgeService = StateObject(wrappedValue: thunderboltBridge)
         _settingsWindowController = StateObject(wrappedValue: SettingsWindowController())
+        _bugReportWindowController = StateObject(wrappedValue: BugReportWindowController())
         enableLaunchAtLoginIfNeeded()
         // Install LaunchDaemon to disable Thunderbolt Bridge on startup (prevents network loops)
         NetworkSetupHelper.promptAndInstallIfNeeded()
@@ -66,6 +68,7 @@ struct EXOApp: App {
                 .environmentObject(updater)
                 .environmentObject(thunderboltBridgeService)
                 .environmentObject(settingsWindowController)
+                .environmentObject(bugReportWindowController)
         } label: {
             menuBarIcon
                 .onReceive(controller.$isFirstLaunchReady) { ready in

app/EXO/EXO/Services/ClusterStateService.swift

@@ -17,7 +17,7 @@ final class ClusterStateService: ObservableObject {
 
     init(
         baseURL: URL = URL(string: "http://127.0.0.1:52415")!,
-        session: URLSession = .shared
+        session: URLSession = ClusterStateService.makeNonCachingSession()
     ) {
         self.baseURL = baseURL
         self.endpoint = baseURL.appendingPathComponent("state")
@@ -27,6 +27,23 @@ final class ClusterStateService: ObservableObject {
         self.decoder = decoder
     }
 
+    /// `URLSession.shared` carries an on-disk `URLCache` that persists every
+    /// response body under `~/Library/Caches/exolabs.EXO/`. We poll `/state`
+    /// at 2 Hz from `startPolling`, so leaving the shared cache attached
+    /// dirties ~500–620 KB/sec of file-backed memory and trips macOS's
+    /// per-process `disk writes` resource limit (microstackshot reports
+    /// observed on M3 Ultra producing GBs of cached responses per hour).
+    /// Cluster-state polling responses are time-sensitive and small; they
+    /// gain nothing from being cached on disk. Use an ephemeral session
+    /// with `urlCache = nil` so neither response bodies nor metadata
+    /// touch disk.
+    private static func makeNonCachingSession() -> URLSession {
+        let config = URLSessionConfiguration.ephemeral
+        config.urlCache = nil
+        config.requestCachePolicy = .reloadIgnoringLocalCacheData
+        return URLSession(configuration: config)
+    }
+
     func startPolling(interval: TimeInterval = 0.5) {
         stopPolling()
         Task {

app/EXO/EXO/Views/BugReportWindowController.swift

@@ -0,0 +1,242 @@
+import AppKit
+import SwiftUI
+
+/// Manages a standalone window for the bug-report flow.
+/// Ensures only one instance exists and brings it to front on repeated opens.
+@MainActor
+final class BugReportWindowController: ObservableObject {
+    private var window: NSWindow?
+
+    func open() {
+        if let existing = window, existing.isVisible {
+            existing.makeKeyAndOrderFront(nil)
+            NSApp.activate()
+            return
+        }
+
+        let view = BugReportView(onDismiss: { [weak self] in
+            self?.window?.close()
+        })
+
+        let hostingController = NSHostingController(rootView: view)
+        hostingController.sizingOptions = [.preferredContentSize, .minSize]
+
+        let newWindow = NSWindow(contentViewController: hostingController)
+        newWindow.styleMask = [.titled, .closable, .resizable]
+        newWindow.title = "Send a Bug Report"
+        newWindow.center()
+        newWindow.setFrameAutosaveName("ExoBugReportWindow")
+        newWindow.isReleasedWhenClosed = false
+        newWindow.makeKeyAndOrderFront(nil)
+        NSApp.activate()
+
+        window = newWindow
+    }
+}
+
+private struct BugReportView: View {
+    fileprivate enum Phase: Equatable {
+        case prompting
+        case sending(String)
+        case success(String)
+        case failure(String)
+    }
+
+    let onDismiss: () -> Void
+
+    @State private var phase: Phase = .prompting
+    @State private var userDescription: String = ""
+    @FocusState private var descriptionFocused: Bool
+
+    var body: some View {
+        VStack(alignment: .leading, spacing: 12) {
+            switch phase {
+            case .prompting:
+                promptingView
+            case .sending(let message):
+                sendingView(message: message)
+            case .success(let message):
+                successView(message: message)
+            case .failure(let message):
+                failureView(message: message)
+            }
+        }
+        .padding(16)
+        .frame(minWidth: 380)
+        .animation(.easeInOut(duration: 0.2), value: phase)
+        .onAppear { descriptionFocused = true }
+    }
+
+    private var promptingView: some View {
+        VStack(alignment: .leading, spacing: 8) {
+            Text("Description (optional)")
+                .font(.subheadline)
+                .foregroundColor(.secondary)
+            ZStack(alignment: .topLeading) {
+                if userDescription.isEmpty {
+                    Text("What were you doing when it broke?")
+                        .font(.body)
+                        .foregroundColor(Color(nsColor: .placeholderTextColor))
+                        .padding(.horizontal, 10)
+                        .padding(.vertical, 8)
+                        .allowsHitTesting(false)
+                }
+                TextEditor(text: $userDescription)
+                    .font(.body)
+                    .scrollContentBackground(.hidden)
+                    .padding(4)
+                    .frame(height: 72)
+                    .focused($descriptionFocused)
+            }
+            .background(
+                RoundedRectangle(cornerRadius: 6)
+                    .fill(Color(nsColor: .textBackgroundColor))
+            )
+            .overlay(
+                RoundedRectangle(cornerRadius: 6)
+                    .strokeBorder(Color(nsColor: .separatorColor), lineWidth: 1)
+            )
+
+            Text("Diagnostic logs will be uploaded with your report.")
+                .font(.caption)
+                .foregroundColor(.secondary)
+
+            HStack {
+                Spacer()
+                Button("Cancel") { onDismiss() }
+                    .keyboardShortcut(.cancelAction)
+                Button("Send") {
+                    Task { await send() }
+                }
+                .keyboardShortcut(.defaultAction)
+            }
+            .padding(.top, 4)
+        }
+    }
+
+    private func sendingView(message: String) -> some View {
+        VStack(alignment: .leading, spacing: 12) {
+            HStack(spacing: 10) {
+                ProgressView().controlSize(.small)
+                Text(message)
+                    .foregroundColor(.secondary)
+            }
+            HStack {
+                Spacer()
+                Button("Cancel") { onDismiss() }
+                    .keyboardShortcut(.cancelAction)
+                    .disabled(true)
+                Button("Send") {}
+                    .disabled(true)
+            }
+        }
+    }
+
+    private func successView(message: String) -> some View {
+        VStack(alignment: .leading, spacing: 12) {
+            HStack(alignment: .top, spacing: 10) {
+                Image(systemName: "checkmark.circle.fill")
+                    .foregroundColor(.green)
+                    .font(.title2)
+                Text(message)
+                    .fixedSize(horizontal: false, vertical: true)
+            }
+            HStack {
+                Button {
+                    openGitHubIssue()
+                } label: {
+                    HStack(spacing: 4) {
+                        Image(systemName: "arrow.up.right.square")
+                        Text("Open GitHub Issue")
+                    }
+                }
+                Spacer()
+                Button("Done") { onDismiss() }
+                    .keyboardShortcut(.defaultAction)
+            }
+        }
+    }
+
+    private func failureView(message: String) -> some View {
+        VStack(alignment: .leading, spacing: 12) {
+            HStack(alignment: .top, spacing: 10) {
+                Image(systemName: "exclamationmark.triangle.fill")
+                    .foregroundColor(.orange)
+                    .font(.title2)
+                Text(message)
+                    .fixedSize(horizontal: false, vertical: true)
+            }
+            HStack {
+                Spacer()
+                Button("Try Again") {
+                    phase = .prompting
+                }
+                Button("Close") { onDismiss() }
+                    .keyboardShortcut(.defaultAction)
+            }
+        }
+    }
+
+    private func send() async {
+        phase = .sending("Collecting logs and uploading…")
+        let service = BugReportService()
+        let description = userDescription.trimmingCharacters(in: .whitespacesAndNewlines)
+        do {
+            let outcome = try await service.sendReport(
+                isManual: true,
+                userDescription: description.isEmpty ? nil : description
+            )
+            if outcome.success {
+                phase = .success(outcome.message)
+            } else {
+                phase = .failure(outcome.message)
+            }
+        } catch {
+            phase = .failure(error.localizedDescription)
+        }
+    }
+
+    private func openGitHubIssue() {
+        let description = userDescription.trimmingCharacters(in: .whitespacesAndNewlines)
+
+        var bodyParts: [String] = []
+        bodyParts.append("## Describe the bug")
+        bodyParts.append("")
+        if !description.isEmpty {
+            bodyParts.append(description)
+        } else {
+            bodyParts.append("A clear and concise description of what the bug is.")
+        }
+        bodyParts.append("")
+        bodyParts.append("## Environment")
+        bodyParts.append("")
+        bodyParts.append("- macOS Version: \(ProcessInfo.processInfo.operatingSystemVersionString)")
+        bodyParts.append("- EXO Version: \(buildTag) (\(buildCommit))")
+        bodyParts.append("")
+        bodyParts.append("## Additional context")
+        bodyParts.append("")
+        bodyParts.append("A bug report with diagnostic logs was submitted via the app.")
+
+        let body = bodyParts.joined(separator: "\n")
+
+        var components = URLComponents(string: "https://github.com/exo-explore/exo/issues/new")!
+        components.queryItems = [
+            URLQueryItem(name: "template", value: "bug_report.md"),
+            URLQueryItem(name: "title", value: "[BUG] "),
+            URLQueryItem(name: "body", value: body),
+            URLQueryItem(name: "labels", value: "bug"),
+        ]
+
+        if let url = components.url {
+            NSWorkspace.shared.open(url)
+        }
+    }
+
+    private var buildTag: String {
+        Bundle.main.infoDictionary?["EXOBuildTag"] as? String ?? "unknown"
+    }
+
+    private var buildCommit: String {
+        Bundle.main.infoDictionary?["EXOBuildCommit"] as? String ?? "unknown"
+    }
+}

app/EXO/EXO/Views/SettingsView.swift

@@ -21,8 +21,6 @@ struct SettingsView: View {
     @State private var pendingReadOnlyModelsDirs: String = ""
     @State private var pendingCustomEnvironmentVariables: [CustomEnvironmentVariable] = []
     @State private var needsRestart = false
-    @State private var bugReportInFlight = false
-    @State private var bugReportMessage: String?
     @State private var uninstallInProgress = false
 
     var body: some View {
@@ -202,8 +200,6 @@ struct SettingsView: View {
                 VStack(alignment: .leading, spacing: 2) {
                     rdmaStatusView
                 }
-
-                sendBugReportButton
             }
 
             Section("Danger Zone") {
@@ -504,50 +500,8 @@ struct SettingsView: View {
         }
     }
 
-    private var sendBugReportButton: some View {
-        VStack(alignment: .leading, spacing: 4) {
-            Button {
-                Task {
-                    await sendBugReport()
-                }
-            } label: {
-                HStack {
-                    if bugReportInFlight {
-                        ProgressView()
-                            .scaleEffect(0.6)
-                    }
-                    Text("Send Bug Report")
-                        .font(.caption)
-                        .fontWeight(.semibold)
-                    Spacer()
-                }
-            }
-            .disabled(bugReportInFlight)
-
-            if let message = bugReportMessage {
-                Text(message)
-                    .font(.caption2)
-                    .foregroundColor(.secondary)
-                    .fixedSize(horizontal: false, vertical: true)
-            }
-        }
-    }
-
     // MARK: - Actions
 
-    private func sendBugReport() async {
-        bugReportInFlight = true
-        bugReportMessage = "Collecting logs..."
-        let service = BugReportService()
-        do {
-            let outcome = try await service.sendReport(isManual: true)
-            bugReportMessage = outcome.message
-        } catch {
-            bugReportMessage = error.localizedDescription
-        }
-        bugReportInFlight = false
-    }
-
     private func showUninstallConfirmationAlert() {
         let alert = NSAlert()
         alert.messageText = "Uninstall EXO"

dashboard/src/routes/+page.svelte

@@ -3435,6 +3435,7 @@
             >
               <li>Connect nodes with TB5 cables</li>
               <li>Boot to Recovery (hold power 10s → Options)</li>
+              <li>Open Terminal from the Utilities menu</li>
               <li>
                 Run
                 <code class="text-yellow-300 bg-yellow-400/10 px-1 rounded"
@@ -4822,6 +4823,7 @@
                 >
                   <li>Connect nodes with TB5 cables</li>
                   <li>Boot to Recovery (hold power 10s → Options)</li>
+                  <li>Open Terminal from the Utilities menu</li>
                   <li>
                     Run
                     <code class="text-yellow-300 bg-yellow-400/10 px-1 rounded"
@@ -4968,6 +4970,7 @@
                   >
                     <li>Connect nodes with TB5 cables</li>
                     <li>Boot to Recovery (hold power 10s → Options)</li>
+                    <li>Open Terminal from the Utilities menu</li>
                     <li>
                       Run
                       <code

rust/exo_pyo3_bindings/exo_pyo3_bindings.pyi

@@ -92,3 +92,59 @@ class PyFromSwarm:
     
     ...
 
+@typing.final
+class UnixBlobChannel:
+    @staticmethod
+    def pair() -> tuple[UnixBlobChannel, UnixBlobChannel]:
+        r"""
+        Create a connected pair of unnamed Unix blob channels.
+        """
+    @staticmethod
+    def from_raw_fd(fd: builtins.int) -> UnixBlobChannel:
+        r"""
+        Wrap an inherited raw file descriptor.
+        
+        The returned object owns `fd`; do not close or reuse that descriptor
+        elsewhere after calling this method.
+        """
+    def raw_fd(self) -> builtins.int:
+        r"""
+        Return the underlying file descriptor without transferring ownership.
+        """
+    def fileno(self) -> builtins.int:
+        r"""
+        Alias for [`raw_fd`], matching Python file-like objects.
+        """
+    def into_raw_fd(self) -> builtins.int:
+        r"""
+        Consume this channel and return its file descriptor.
+        
+        After this method succeeds, the Python object is closed and the caller
+        owns the returned descriptor.
+        """
+    def close(self) -> None:
+        r"""
+        Close this channel.
+        """
+    def closed(self) -> builtins.bool:
+        r"""
+        Return whether this channel has been closed or consumed.
+        """
+    def send(self, bytes: bytes) -> None:
+        r"""
+        Send one binary blob.
+        """
+    def recv(self) -> bytes:
+        r"""
+        Receive one binary blob using the default maximum blob size.
+        """
+    def recv_with_max_blob_size(self, max_blob_size: builtins.int) -> bytes:
+        r"""
+        Receive one binary blob, allowing at most `max_blob_size` bytes.
+        """
+    @staticmethod
+    def default_max_blob_size() -> builtins.int:
+        r"""
+        Default maximum blob size accepted by `recv`.
+        """
+

rust/exo_pyo3_bindings/src/blob_channel.rs

@@ -0,0 +1,164 @@
+use std::os::fd::RawFd;
+use std::sync::{Mutex, MutexGuard};
+
+use pyo3::exceptions::{PyOSError, PyRuntimeError, PyValueError};
+use pyo3::prelude::{PyModule, PyModuleMethods as _};
+use pyo3::types::{PyBytes, PyBytesMethods as _};
+use pyo3::{Bound, PyResult, Python, pyclass, pymethods};
+use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pymethods};
+use util::blob_channel::{DEFAULT_MAX_BLOB_SIZE, UnixBlobChannel};
+
+#[gen_stub_pyclass]
+#[pyclass(name = "UnixBlobChannel")]
+#[derive(Debug)]
+pub struct PyUnixBlobChannel {
+    channel: Mutex<Option<UnixBlobChannel>>,
+}
+
+impl PyUnixBlobChannel {
+    const fn new(channel: UnixBlobChannel) -> Self {
+        Self {
+            channel: Mutex::new(Some(channel)),
+        }
+    }
+
+    fn lock_channel(&self) -> PyResult<MutexGuard<'_, Option<UnixBlobChannel>>> {
+        self.channel
+            .lock()
+            .map_err(|_| PyRuntimeError::new_err("UnixBlobChannel lock poisoned"))
+    }
+}
+
+#[allow(
+    clippy::multiple_inherent_impl,
+    clippy::significant_drop_tightening,
+    clippy::use_self,
+    clippy::wrong_self_convention
+)]
+#[gen_stub_pymethods]
+#[pymethods]
+impl PyUnixBlobChannel {
+    /// Create a connected pair of unnamed Unix blob channels.
+    #[staticmethod]
+    fn pair() -> PyResult<(PyUnixBlobChannel, PyUnixBlobChannel)> {
+        let (left, right) = UnixBlobChannel::pair().map_err(PyOSError::new_err)?;
+        Ok((Self::new(left), Self::new(right)))
+    }
+
+    /// Wrap an inherited raw file descriptor.
+    ///
+    /// The returned object owns `fd`; do not close or reuse that descriptor
+    /// elsewhere after calling this method.
+    #[staticmethod]
+    fn from_raw_fd(fd: RawFd) -> PyResult<Self> {
+        if fd < 0 {
+            return Err(PyValueError::new_err(
+                "file descriptor must be non-negative",
+            ));
+        }
+
+        // SAFETY: Python callers use this to adopt an inherited descriptor. The
+        // wrapper owns and closes the descriptor after this point.
+        Ok(Self::new(unsafe { UnixBlobChannel::from_raw_fd(fd) }))
+    }
+
+    /// Return the underlying file descriptor without transferring ownership.
+    fn raw_fd(&self) -> PyResult<RawFd> {
+        let raw_fd = self
+            .lock_channel()?
+            .as_ref()
+            .ok_or_else(|| PyValueError::new_err("UnixBlobChannel is closed"))?
+            .raw_fd();
+        Ok(raw_fd)
+    }
+
+    /// Alias for [`raw_fd`], matching Python file-like objects.
+    fn fileno(&self) -> PyResult<RawFd> {
+        self.raw_fd()
+    }
+
+    /// Consume this channel and return its file descriptor.
+    ///
+    /// After this method succeeds, the Python object is closed and the caller
+    /// owns the returned descriptor.
+    fn into_raw_fd(&self) -> PyResult<RawFd> {
+        let channel = self
+            .lock_channel()?
+            .take()
+            .ok_or_else(|| PyValueError::new_err("UnixBlobChannel is closed"))?;
+        Ok(channel.into_raw_fd())
+    }
+
+    /// Close this channel.
+    fn close(&self) -> PyResult<()> {
+        drop(self.lock_channel()?.take());
+        Ok(())
+    }
+
+    /// Return whether this channel has been closed or consumed.
+    fn closed(&self) -> PyResult<bool> {
+        Ok(self.lock_channel()?.is_none())
+    }
+
+    /// Send one binary blob.
+    fn send(&self, py: Python<'_>, bytes: &Bound<'_, PyBytes>) -> PyResult<()> {
+        let bytes = Vec::from(bytes.as_bytes());
+        py.detach(|| {
+            {
+                let mut guard = self.lock_channel()?;
+                let channel = guard
+                    .as_mut()
+                    .ok_or_else(|| PyValueError::new_err("UnixBlobChannel is closed"))?;
+                channel.send(&bytes)
+            }
+            .map_err(PyOSError::new_err)
+        })
+    }
+
+    /// Receive one binary blob using the default maximum blob size.
+    fn recv<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
+        let bytes = py.detach(|| {
+            {
+                let mut guard = self.lock_channel()?;
+                let channel = guard
+                    .as_mut()
+                    .ok_or_else(|| PyValueError::new_err("UnixBlobChannel is closed"))?;
+                channel.recv()
+            }
+            .map_err(PyOSError::new_err)
+        })?;
+        Ok(PyBytes::new(py, &bytes))
+    }
+
+    /// Receive one binary blob, allowing at most `max_blob_size` bytes.
+    fn recv_with_max_blob_size<'py>(
+        &self,
+        py: Python<'py>,
+        max_blob_size: usize,
+    ) -> PyResult<Bound<'py, PyBytes>> {
+        let bytes = py.detach(|| {
+            {
+                let mut guard = self.lock_channel()?;
+                let channel = guard
+                    .as_mut()
+                    .ok_or_else(|| PyValueError::new_err("UnixBlobChannel is closed"))?;
+                channel.recv_with_max_blob_size(max_blob_size)
+            }
+            .map_err(PyOSError::new_err)
+        })?;
+        Ok(PyBytes::new(py, &bytes))
+    }
+
+    /// Default maximum blob size accepted by `recv`.
+    #[staticmethod]
+    const fn default_max_blob_size() -> usize {
+        DEFAULT_MAX_BLOB_SIZE
+    }
+}
+
+pub fn blob_channel_submodule(m: &Bound<'_, PyModule>) -> PyResult<()> {
+    m.add_class::<PyUnixBlobChannel>()?;
+    m.add("DEFAULT_MAX_BLOB_SIZE", DEFAULT_MAX_BLOB_SIZE)?;
+
+    Ok(())
+}

rust/exo_pyo3_bindings/src/lib.rs

@@ -5,13 +5,15 @@
 //!
 
 mod allow_threading;
+mod blob_channel;
 mod ident;
 mod networking;
 
+use crate::blob_channel::blob_channel_submodule;
 use crate::ident::PyKeypair;
 use crate::networking::networking_submodule;
 use pyo3::prelude::PyModule;
-use pyo3::types::PyModuleMethods;
+use pyo3::types::PyModuleMethods as _;
 use pyo3::{Bound, PyResult, pyclass, pymodule};
 use pyo3_stub_gen::define_stub_info_gatherer;
 
@@ -163,6 +165,7 @@ fn main_module(m: &Bound<'_, PyModule>) -> PyResult<()> {
     //       work with maturin, where the types generate correctly, in the right folder, without
     //       too many importing issues...
     m.add_class::<PyKeypair>()?;
+    blob_channel_submodule(m)?;
     networking_submodule(m)?;
 
     // top-level constructs

rust/exo_pyo3_bindings/tests/test_python.py

@@ -6,6 +6,7 @@ from exo_pyo3_bindings import (
     NetworkingHandle,
     NoPeersSubscribedToTopicError,
     PyFromSwarm,
+    UnixBlobChannel,
 )
 
 
@@ -34,3 +35,22 @@ async def _await_recv(h: NetworkingHandle):
                 print(f"PYTHON: connection update: {c}")
             case PyFromSwarm.Message() as m:
                 print(f"PYTHON: message: {m}")
+
+
+def test_unix_blob_channel_roundtrip() -> None:
+    left, right = UnixBlobChannel.pair()
+
+    left.send(b"hello")
+
+    assert right.recv() == b"hello"
+
+
+def test_unix_blob_channel_raw_fd_handoff() -> None:
+    left, right = UnixBlobChannel.pair()
+    raw_fd = right.into_raw_fd()
+    adopted = UnixBlobChannel.from_raw_fd(raw_fd)
+
+    left.send(b"from raw fd")
+
+    assert adopted.recv() == b"from raw fd"
+    assert right.closed()

rust/util/src/blob_channel.rs

@@ -0,0 +1,239 @@
+use std::io::{self, ErrorKind, Read as _, Write as _};
+use std::mem::size_of;
+use std::os::fd::{AsFd, AsRawFd, BorrowedFd, FromRawFd as _, IntoRawFd as _, OwnedFd, RawFd};
+use std::os::unix::net::UnixStream;
+
+const LENGTH_PREFIX_SIZE: usize = size_of::<u64>();
+
+/// Default maximum blob size accepted by [`UnixBlobChannel::recv`].
+///
+/// This is a receiver-side allocation guard, not an expected message size.
+pub const DEFAULT_MAX_BLOB_SIZE: usize = 64 * 1024 * 1024;
+
+/// A connected Unix-domain channel for length-prefixed binary blobs.
+#[derive(Debug)]
+pub struct UnixBlobChannel {
+    stream: UnixStream,
+}
+
+impl UnixBlobChannel {
+    /// Create a connected pair of unnamed Unix blob channels.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if the socketpair cannot be created.
+    #[inline]
+    pub fn pair() -> io::Result<(Self, Self)> {
+        let (left, right) = UnixStream::pair()?;
+        Ok((Self { stream: left }, Self { stream: right }))
+    }
+
+    /// Wrap an owned file descriptor as a Unix blob channel.
+    #[must_use]
+    #[inline]
+    pub fn from_owned_fd(fd: OwnedFd) -> Self {
+        Self {
+            stream: UnixStream::from(fd),
+        }
+    }
+
+    /// Wrap an inherited raw file descriptor as a Unix blob channel.
+    ///
+    /// # Safety
+    ///
+    /// `raw_fd` must be open and uniquely owned by this call path. After this
+    /// function returns, the descriptor is owned by Rust and will be closed on
+    /// drop.
+    #[must_use]
+    #[inline]
+    pub unsafe fn from_raw_fd(raw_fd: RawFd) -> Self {
+        Self {
+            // SAFETY: The caller guarantees that `raw_fd` is open and uniquely
+            // owned by this call path.
+            stream: unsafe { UnixStream::from_raw_fd(raw_fd) },
+        }
+    }
+
+    /// Return the underlying raw file descriptor.
+    #[must_use]
+    #[inline]
+    pub fn raw_fd(&self) -> RawFd {
+        self.stream.as_raw_fd()
+    }
+
+    /// Consume this channel and return its owned file descriptor.
+    #[must_use]
+    #[inline]
+    pub fn into_owned_fd(self) -> OwnedFd {
+        self.stream.into()
+    }
+
+    /// Consume this channel and return its raw file descriptor.
+    #[must_use]
+    #[inline]
+    pub fn into_raw_fd(self) -> RawFd {
+        self.stream.into_raw_fd()
+    }
+
+    /// Send one binary blob.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if the length prefix or blob cannot be written.
+    #[inline]
+    pub fn send(&mut self, bytes: &[u8]) -> io::Result<()> {
+        let len = u64::try_from(bytes.len()).map_err(|_| {
+            io::Error::new(
+                ErrorKind::InvalidInput,
+                "blob length does not fit in the wire header",
+            )
+        })?;
+
+        self.stream.write_all(&len.to_be_bytes())?;
+        self.stream.write_all(bytes)
+    }
+
+    /// Receive one binary blob using [`DEFAULT_MAX_BLOB_SIZE`].
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if the length prefix or blob cannot be read, or if the
+    /// announced blob length exceeds [`DEFAULT_MAX_BLOB_SIZE`].
+    #[inline]
+    pub fn recv(&mut self) -> io::Result<Vec<u8>> {
+        self.recv_with_max_blob_size(DEFAULT_MAX_BLOB_SIZE)
+    }
+
+    /// Receive one binary blob, allowing at most `max_blob_size` bytes.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if the length prefix or blob cannot be read, or if the
+    /// announced blob length exceeds `max_blob_size`.
+    #[inline]
+    pub fn recv_with_max_blob_size(&mut self, max_blob_size: usize) -> io::Result<Vec<u8>> {
+        let mut len_bytes = [0; LENGTH_PREFIX_SIZE];
+        self.stream.read_exact(&mut len_bytes)?;
+
+        let len = u64::from_be_bytes(len_bytes);
+        let len = usize::try_from(len).map_err(|_| {
+            io::Error::new(
+                ErrorKind::InvalidData,
+                "blob length does not fit on this platform",
+            )
+        })?;
+
+        if len > max_blob_size {
+            return Err(io::Error::new(
+                ErrorKind::InvalidData,
+                "blob length exceeds maximum size",
+            ));
+        }
+
+        let mut bytes = vec![0; len];
+        self.stream.read_exact(&mut bytes)?;
+        Ok(bytes)
+    }
+}
+
+impl AsFd for UnixBlobChannel {
+    #[inline]
+    fn as_fd(&self) -> BorrowedFd<'_> {
+        self.stream.as_fd()
+    }
+}
+
+impl AsRawFd for UnixBlobChannel {
+    #[inline]
+    fn as_raw_fd(&self) -> RawFd {
+        self.stream.as_raw_fd()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::thread;
+
+    use super::*;
+
+    #[test]
+    fn sends_and_receives_bytes() -> io::Result<()> {
+        let (mut left, mut right) = UnixBlobChannel::pair()?;
+
+        left.send(b"hello")?;
+        assert_eq!(right.recv()?, b"hello");
+
+        Ok(())
+    }
+
+    #[test]
+    fn sends_and_receives_empty_blob() -> io::Result<()> {
+        let (mut left, mut right) = UnixBlobChannel::pair()?;
+
+        left.send(b"")?;
+        assert!(right.recv()?.is_empty());
+
+        Ok(())
+    }
+
+    #[test]
+    fn preserves_blob_boundaries() -> io::Result<()> {
+        let (mut left, mut right) = UnixBlobChannel::pair()?;
+
+        left.send(b"first")?;
+        left.send(b"second")?;
+
+        assert_eq!(right.recv()?, b"first");
+        assert_eq!(right.recv()?, b"second");
+
+        Ok(())
+    }
+
+    #[test]
+    fn sends_and_receives_large_blob() -> io::Result<()> {
+        let (mut left, right) = UnixBlobChannel::pair()?;
+        let payload = deterministic_blob(200 * 1024 * 1024);
+        let max_blob_size = payload.len();
+
+        let receiver_thread = thread::spawn(move || {
+            let mut receiver = right;
+            receiver.recv_with_max_blob_size(max_blob_size)
+        });
+        left.send(&payload)?;
+
+        let received = receiver_thread
+            .join()
+            .map_err(|_| io::Error::other("receiver thread panicked"))??;
+        assert_eq!(received, payload);
+
+        Ok(())
+    }
+
+    fn deterministic_blob(len: usize) -> Vec<u8> {
+        let mut state = 0x9e37_79b9_7f4a_7c15_u64;
+        let mut bytes = Vec::with_capacity(len);
+
+        while bytes.len() < len {
+            state = state
+                .wrapping_mul(0xbf58_476d_1ce4_e5b9)
+                .wrapping_add(0x94d0_49bb_1331_11eb);
+            bytes.push(state.to_le_bytes()[3]);
+        }
+
+        bytes
+    }
+
+    #[test]
+    fn rejects_oversized_blob() -> io::Result<()> {
+        let (mut left, mut right) = UnixBlobChannel::pair()?;
+
+        left.send(b"too large")?;
+
+        let Err(err) = right.recv_with_max_blob_size(3) else {
+            return Err(io::Error::other("blob should be too large"));
+        };
+        assert_eq!(err.kind(), ErrorKind::InvalidData);
+
+        Ok(())
+    }
+}

rust/util/src/lib.rs

@@ -1 +1,3 @@
+#[cfg(any(target_os = "linux", target_os = "macos"))]
+pub mod blob_channel;
 pub mod wakerdeque;

src/exo/download/download_utils.py

@@ -1,11 +1,12 @@
 import asyncio
 import hashlib
 import os
+import random
 import shutil
 import ssl
 import time
 import traceback
-from collections.abc import Awaitable
+from collections.abc import Awaitable, Mapping
 from datetime import timedelta
 from pathlib import Path
 from typing import Callable, Literal
@@ -55,6 +56,36 @@ class HuggingFaceAuthenticationError(Exception):
 class HuggingFaceRateLimitError(Exception):
     """429 Huggingface code"""
 
+    def __init__(self, msg: str, retry_after: float | None = None) -> None:
+        super().__init__(msg)
+        self.retry_after = retry_after
+
+
+def _parse_retry_after(headers: Mapping[str, str]) -> float | None:
+    """Parse seconds-to-reset from HF's RateLimit header.
+
+    HF sends e.g. ``ratelimit: "api";r=0;t=52`` on 429s; ``t`` is the wait.
+    Returns ``None`` if the header is missing or has no ``t`` field.
+    """
+    raw = headers.get("RateLimit") or headers.get("ratelimit")
+    if raw is None:
+        return None
+    for part in raw.split(";"):
+        key, _, val = part.strip().partition("=")
+        if key == "t":
+            try:
+                return float(val)
+            except ValueError:
+                return None
+    return None
+
+
+# reset window is 5 min
+_RATE_LIMIT_MAX_SLEEP_SECS = 300.0
+
+# 24h. Manually clear the cache (or `delete_model`) to force a refresh.
+_FILE_LIST_CACHE_TTL_SECS = 24 * 60 * 60
+
 
 async def _build_auth_error_message(status_code: int, model_id: ModelId) -> str:
     token = await get_hf_token()
@@ -348,9 +379,6 @@ async def _build_file_list_from_local_directory(
     return None
 
 
-_fetched_file_lists_this_session: set[str] = set()
-
-
 async def fetch_file_list_with_cache(
     model_id: ModelId,
     revision: str = "main",
@@ -360,13 +388,16 @@ async def fetch_file_list_with_cache(
 ) -> list[FileListEntry]:
     target_dir = await ensure_cache_dir(model_id)
     cache_file = target_dir / f"{model_id.normalize()}--{revision}--file_list.json"
-    cache_key = f"{model_id.normalize()}--{revision}"
 
-    if cache_key in _fetched_file_lists_this_session and await aios.path.exists(
-        cache_file
-    ):
-        async with aiofiles.open(cache_file, "r") as f:
-            return TypeAdapter(list[FileListEntry]).validate_json(await f.read())
+    # cache survives process restarts so cold starts don't re-burst HF
+    if await aios.path.exists(cache_file):
+        try:
+            cache_age = time.time() - (await aios.stat(cache_file)).st_mtime
+        except OSError:
+            cache_age = float("inf")
+        if cache_age < _FILE_LIST_CACHE_TTL_SECS:
+            async with aiofiles.open(cache_file, "r") as f:
+                return TypeAdapter(list[FileListEntry]).validate_json(await f.read())
 
     if skip_internet:
         if await aios.path.exists(cache_file):
@@ -395,7 +426,6 @@ async def fetch_file_list_with_cache(
             await f.write(
                 TypeAdapter(list[FileListEntry]).dump_json(file_list).decode()
             )
-        _fetched_file_lists_this_session.add(cache_key)
         return file_list
     except Exception as e:
         logger.opt(exception=e).warning(
@@ -426,17 +456,29 @@ async def fetch_file_list_with_retry(
     recursive: bool = False,
     on_connection_lost: Callable[[], None] = lambda: None,
 ) -> list[FileListEntry]:
-    n_attempts = 3
+    n_attempts = 5
     for attempt in range(n_attempts):
         try:
             return await _fetch_file_list(model_id, revision, path, recursive)
         except HuggingFaceAuthenticationError:
             raise
+        except HuggingFaceRateLimitError as e:
+            if attempt == n_attempts - 1:
+                raise
+            sleep_for = e.retry_after if e.retry_after is not None else 2.0**attempt
+            sleep_for = min(sleep_for, _RATE_LIMIT_MAX_SLEEP_SECS) + random.uniform(
+                0, 1
+            )
+            logger.warning(
+                f"Rate limited by HuggingFace fetching file list for {model_id}; "
+                f"sleeping {sleep_for:.1f}s before retry {attempt + 2}/{n_attempts}"
+            )
+            await asyncio.sleep(sleep_for)
         except Exception as e:
             on_connection_lost()
             if attempt == n_attempts - 1:
                 raise e
-            await asyncio.sleep(2.0**attempt)
+            await asyncio.sleep(2.0**attempt + random.uniform(0, 1))
     raise Exception(
         f"Failed to fetch file list for {model_id=} {revision=} {path=} {recursive=}"
     )
@@ -447,6 +489,9 @@ async def _fetch_file_list(
 ) -> list[FileListEntry]:
     api_url = f"{get_hf_endpoint()}/api/models/{model_id}/tree/{revision}"
     url = f"{api_url}/{path}" if path else api_url
+    # ?recursive=true returns the whole subtree in one request
+    if recursive:
+        url = f"{url}?recursive=true"
 
     headers = await get_download_headers()
     async with (
@@ -458,7 +503,8 @@ async def _fetch_file_list(
             raise HuggingFaceAuthenticationError(msg)
         elif response.status == 429:
             raise HuggingFaceRateLimitError(
-                f"Couldn't download {model_id} because of HuggingFace rate limit."
+                f"HuggingFace rate limit hit fetching file list for {model_id}",
+                retry_after=_parse_retry_after(response.headers),
             )
         elif response.status == 200:
             data_json = await response.text()
@@ -468,10 +514,14 @@ async def _fetch_file_list(
                 if item.type == "file":
                     files.append(FileListEntry.model_validate(item))
                 elif item.type == "directory" and recursive:
-                    subfiles = await _fetch_file_list(
-                        model_id, revision, item.path, recursive
-                    )
-                    files.extend(subfiles)
+                    # already inlined by ?recursive=true
+                    continue
+            if recursive and len(data) >= 1000:
+                # HF tree endpoint paginates at 1000; we don't follow cursors
+                logger.warning(
+                    f"File list for {model_id} hit the 1000-entry page cap "
+                    "and may be truncated; cursor pagination is not implemented"
+                )
             return files
         else:
             raise Exception(f"Failed to fetch file list: {response.status}")
@@ -552,6 +602,11 @@ async def file_meta(
         if r.status in [401, 403]:
             msg = await _build_auth_error_message(r.status, model_id)
             raise HuggingFaceAuthenticationError(msg)
+        if r.status == 429:
+            raise HuggingFaceRateLimitError(
+                f"HuggingFace rate limit hit fetching metadata for {model_id}/{path}",
+                retry_after=_parse_retry_after(r.headers),
+            )
         content_length = int(
             r.headers.get("x-linked-size") or r.headers.get("content-length") or 0
         )
@@ -571,7 +626,7 @@ async def download_file_with_retry(
     on_connection_lost: Callable[[], None] = lambda: None,
     skip_internet: bool = False,
 ) -> Path:
-    n_attempts = 3
+    n_attempts = 5
     for attempt in range(n_attempts):
         try:
             return await _download_file(
@@ -583,12 +638,16 @@ async def download_file_with_retry(
             raise
         except HuggingFaceRateLimitError as e:
             if attempt == n_attempts - 1:
-                raise e
-            logger.error(
-                f"Download error on attempt {attempt}/{n_attempts} for {model_id=} {revision=} {path=} {target_dir=}"
+                raise
+            sleep_for = e.retry_after if e.retry_after is not None else 2.0**attempt
+            sleep_for = min(sleep_for, _RATE_LIMIT_MAX_SLEEP_SECS) + random.uniform(
+                0, 1
             )
-            logger.error(traceback.format_exc())
-            await asyncio.sleep(2.0**attempt)
+            logger.warning(
+                f"Rate limited by HuggingFace downloading {model_id}/{path}; "
+                f"sleeping {sleep_for:.1f}s before retry {attempt + 2}/{n_attempts}"
+            )
+            await asyncio.sleep(sleep_for)
         except Exception as e:
             if attempt == n_attempts - 1:
                 on_connection_lost()
@@ -597,7 +656,7 @@ async def download_file_with_retry(
                 f"Download error on attempt {attempt + 1}/{n_attempts} for {model_id=} {revision=} {path=} {target_dir=}"
             )
             logger.error(traceback.format_exc())
-            await asyncio.sleep(2.0**attempt)
+            await asyncio.sleep(2.0**attempt + random.uniform(0, 1))
     raise Exception(
         f"Failed to download file {model_id=} {revision=} {path=} {target_dir=}"
     )
@@ -665,6 +724,11 @@ async def _download_file(
             if r.status in [401, 403]:
                 msg = await _build_auth_error_message(r.status, model_id)
                 raise HuggingFaceAuthenticationError(msg)
+            if r.status == 429:
+                raise HuggingFaceRateLimitError(
+                    f"HuggingFace rate limit hit downloading {model_id}/{path}",
+                    retry_after=_parse_retry_after(r.headers),
+                )
             assert r.status in [200, 206], (
                 f"Failed to download {path} from {url}: {r.status}"
             )

src/exo/download/tests/test_offline_mode.py

@@ -1,5 +1,7 @@
 """Tests for offline/air-gapped mode."""
 
+import os
+import time
 from collections.abc import AsyncIterator
 from pathlib import Path
 from unittest.mock import AsyncMock, patch
@@ -231,3 +233,64 @@ class TestFetchFileListOffline:
         raise FileNotFoundError."""
         with pytest.raises(FileNotFoundError, match="No internet"):
             await fetch_file_list_with_cache(model_id, "main", skip_internet=True)
+
+
+class TestFileListCacheTTL:
+    async def test_uses_fresh_cache_without_fetching(
+        self, model_id: ModelId, temp_models_dir: Path
+    ) -> None:
+        from pydantic import TypeAdapter
+
+        cache_dir = temp_models_dir / "caches" / model_id.normalize()
+        await aios.makedirs(cache_dir, exist_ok=True)
+
+        cached_list = [
+            FileListEntry(type="file", path="model.safetensors", size=1000),
+        ]
+        cache_file = cache_dir / f"{model_id.normalize()}--main--file_list.json"
+        async with aiofiles.open(cache_file, "w") as f:
+            await f.write(
+                TypeAdapter(list[FileListEntry]).dump_json(cached_list).decode()
+            )
+
+        with patch(
+            "exo.download.download_utils.fetch_file_list_with_retry",
+            new_callable=AsyncMock,
+        ) as mock_fetch:
+            result = await fetch_file_list_with_cache(model_id, "main")
+
+        assert result == cached_list
+        mock_fetch.assert_not_called()
+
+    async def test_refetches_when_cache_older_than_ttl(
+        self, model_id: ModelId, temp_models_dir: Path
+    ) -> None:
+        from pydantic import TypeAdapter
+
+        from exo.download.download_utils import (
+            _FILE_LIST_CACHE_TTL_SECS,  # pyright: ignore[reportPrivateUsage]
+        )
+
+        cache_dir = temp_models_dir / "caches" / model_id.normalize()
+        await aios.makedirs(cache_dir, exist_ok=True)
+
+        stale_list = [FileListEntry(type="file", path="stale.bin", size=1)]
+        cache_file = cache_dir / f"{model_id.normalize()}--main--file_list.json"
+        async with aiofiles.open(cache_file, "w") as f:
+            await f.write(
+                TypeAdapter(list[FileListEntry]).dump_json(stale_list).decode()
+            )
+
+        old_mtime = time.time() - _FILE_LIST_CACHE_TTL_SECS - 60
+        os.utime(cache_file, (old_mtime, old_mtime))
+
+        fresh_list = [FileListEntry(type="file", path="fresh.bin", size=2)]
+        with patch(
+            "exo.download.download_utils.fetch_file_list_with_retry",
+            new_callable=AsyncMock,
+            return_value=fresh_list,
+        ) as mock_fetch:
+            result = await fetch_file_list_with_cache(model_id, "main")
+
+        assert result == fresh_list
+        mock_fetch.assert_called_once()

src/exo/download/tests/test_rate_limit_handling.py

@@ -0,0 +1,355 @@
+"""Tests for HuggingFace 429 rate-limit handling in download_utils."""
+
+from collections.abc import AsyncIterator
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import aiofiles.os as aios
+import pytest
+
+from exo.download.download_utils import (
+    HuggingFaceRateLimitError,
+    _download_file,  # pyright: ignore[reportPrivateUsage]
+    _fetch_file_list,  # pyright: ignore[reportPrivateUsage]
+    _parse_retry_after,  # pyright: ignore[reportPrivateUsage]
+    download_file_with_retry,
+    fetch_file_list_with_retry,
+    file_meta,
+)
+from exo.shared.types.common import ModelId
+
+# captured from a real HF 429 on 2026-04-30 (header is lowercased by Cloudfront)
+REAL_HF_429_HEADERS_2026_04_30 = {
+    "ratelimit": '"api";r=0;t=52',
+    "ratelimit-policy": '"fixed window";"api";q=500;w=300',
+}
+
+
+class TestParseRetryAfter:
+    def test_parses_documented_format(self) -> None:
+        assert _parse_retry_after({"RateLimit": '"api";r=0;t=243'}) == 243.0
+
+    def test_parses_real_hf_response(self) -> None:
+        assert _parse_retry_after(REAL_HF_429_HEADERS_2026_04_30) == 52.0
+
+    def test_parses_resolvers_bucket(self) -> None:
+        assert _parse_retry_after({"ratelimit": '"resolvers";r=0;t=120'}) == 120.0
+
+    def test_parses_pages_bucket(self) -> None:
+        assert _parse_retry_after({"ratelimit": '"pages";r=0;t=10'}) == 10.0
+
+    def test_returns_none_when_header_missing(self) -> None:
+        assert _parse_retry_after({}) is None
+
+    def test_returns_none_when_only_retry_after_present(self) -> None:
+        assert _parse_retry_after({"Retry-After": "60"}) is None
+
+    def test_returns_none_when_format_unrecognised(self) -> None:
+        assert _parse_retry_after({"ratelimit": "garbage"}) is None
+
+    def test_handles_extra_whitespace(self) -> None:
+        assert _parse_retry_after({"ratelimit": '"api"; r=0; t=42'}) == 42.0
+
+
+class TestFetchFileListRetry:
+    async def test_uses_retry_after_from_error(self) -> None:
+        sleeps: list[float] = []
+
+        async def fake_sleep(seconds: float) -> None:
+            sleeps.append(seconds)
+
+        async def fake_fetch(*args: object, **kwargs: object) -> list[object]:
+            if not sleeps:
+                raise HuggingFaceRateLimitError("rate limited", retry_after=2.0)
+            return []
+
+        with (
+            patch(
+                "exo.download.download_utils._fetch_file_list", side_effect=fake_fetch
+            ),
+            patch("exo.download.download_utils.asyncio.sleep", side_effect=fake_sleep),
+        ):
+            result = await fetch_file_list_with_retry(ModelId("test/model"))
+
+        assert result == []
+        assert len(sleeps) == 1
+        assert 2.0 <= sleeps[0] < 3.0  # retry_after + jitter[0,1)
+
+    async def test_falls_back_to_exp_backoff_when_no_retry_after(self) -> None:
+        sleeps: list[float] = []
+
+        async def fake_sleep(seconds: float) -> None:
+            sleeps.append(seconds)
+
+        async def fake_fetch(*args: object, **kwargs: object) -> list[object]:
+            if not sleeps:
+                raise HuggingFaceRateLimitError("rate limited", retry_after=None)
+            return []
+
+        with (
+            patch(
+                "exo.download.download_utils._fetch_file_list", side_effect=fake_fetch
+            ),
+            patch("exo.download.download_utils.asyncio.sleep", side_effect=fake_sleep),
+        ):
+            await fetch_file_list_with_retry(ModelId("test/model"))
+
+        assert len(sleeps) == 1
+        assert 1.0 <= sleeps[0] < 2.0  # 2**0 + jitter[0,1)
+
+    async def test_caps_sleep_at_max_window(self) -> None:
+        sleeps: list[float] = []
+
+        async def fake_sleep(seconds: float) -> None:
+            sleeps.append(seconds)
+
+        async def fake_fetch(*args: object, **kwargs: object) -> list[object]:
+            if not sleeps:
+                raise HuggingFaceRateLimitError("rate limited", retry_after=10_000.0)
+            return []
+
+        with (
+            patch(
+                "exo.download.download_utils._fetch_file_list", side_effect=fake_fetch
+            ),
+            patch("exo.download.download_utils.asyncio.sleep", side_effect=fake_sleep),
+        ):
+            await fetch_file_list_with_retry(ModelId("test/model"))
+
+        assert len(sleeps) == 1
+        assert 300.0 <= sleeps[0] < 301.0  # cap + jitter[0,1)
+
+    async def test_retries_up_to_five_times(self) -> None:
+        sleeps: list[float] = []
+
+        async def fake_sleep(seconds: float) -> None:
+            sleeps.append(seconds)
+
+        async def fake_fetch(*args: object, **kwargs: object) -> list[object]:
+            raise HuggingFaceRateLimitError("rate limited", retry_after=1.0)
+
+        with (
+            patch(
+                "exo.download.download_utils._fetch_file_list", side_effect=fake_fetch
+            ),
+            patch("exo.download.download_utils.asyncio.sleep", side_effect=fake_sleep),
+            pytest.raises(HuggingFaceRateLimitError),
+        ):
+            await fetch_file_list_with_retry(ModelId("test/model"))
+
+        assert len(sleeps) == 4  # 5 attempts -> 4 sleeps before giving up
+
+
+class TestDownloadFileRetry:
+    @pytest.fixture
+    async def target_dir(self, tmp_path: Path) -> AsyncIterator[Path]:
+        target = tmp_path / "downloads"
+        await aios.makedirs(target, exist_ok=True)
+        yield target
+
+    async def test_uses_retry_after_from_error(self, target_dir: Path) -> None:
+        sleeps: list[float] = []
+        results: list[Path] = [target_dir / "file.bin"]
+
+        async def fake_sleep(seconds: float) -> None:
+            sleeps.append(seconds)
+
+        async def fake_download(*args: object, **kwargs: object) -> Path:
+            if not sleeps:
+                raise HuggingFaceRateLimitError("rate limited", retry_after=5.0)
+            return results[0]
+
+        with (
+            patch(
+                "exo.download.download_utils._download_file",
+                side_effect=fake_download,
+            ),
+            patch("exo.download.download_utils.asyncio.sleep", side_effect=fake_sleep),
+        ):
+            result = await download_file_with_retry(
+                ModelId("test/model"), "main", "file.bin", target_dir
+            )
+
+        assert result == results[0]
+        assert len(sleeps) == 1
+        assert 5.0 <= sleeps[0] < 6.0
+
+    async def test_caps_sleep_at_max_window(self, target_dir: Path) -> None:
+        sleeps: list[float] = []
+        results: list[Path] = [target_dir / "file.bin"]
+
+        async def fake_sleep(seconds: float) -> None:
+            sleeps.append(seconds)
+
+        async def fake_download(*args: object, **kwargs: object) -> Path:
+            if not sleeps:
+                raise HuggingFaceRateLimitError("rate limited", retry_after=99_999.0)
+            return results[0]
+
+        with (
+            patch(
+                "exo.download.download_utils._download_file",
+                side_effect=fake_download,
+            ),
+            patch("exo.download.download_utils.asyncio.sleep", side_effect=fake_sleep),
+        ):
+            await download_file_with_retry(
+                ModelId("test/model"), "main", "file.bin", target_dir
+            )
+
+        assert len(sleeps) == 1
+        assert 300.0 <= sleeps[0] < 301.0
+
+    async def test_retries_up_to_five_times(self, target_dir: Path) -> None:
+        sleeps: list[float] = []
+
+        async def fake_sleep(seconds: float) -> None:
+            sleeps.append(seconds)
+
+        with (
+            patch(
+                "exo.download.download_utils._download_file",
+                new_callable=AsyncMock,
+                side_effect=HuggingFaceRateLimitError("rate limited", retry_after=1.0),
+            ),
+            patch("exo.download.download_utils.asyncio.sleep", side_effect=fake_sleep),
+            pytest.raises(HuggingFaceRateLimitError),
+        ):
+            await download_file_with_retry(
+                ModelId("test/model"), "main", "file.bin", target_dir
+            )
+
+        assert len(sleeps) == 4
+
+
+def _make_mock_session_returning(
+    response_attrs: dict[str, object], method: str = "get"
+) -> MagicMock:
+    """Build a MagicMock that mimics ``create_http_session`` returning a
+    response whose ``status`` / ``headers`` are set from ``response_attrs``.
+
+    Mocks the chain ``create_http_session().__aenter__() -> session``, and
+    ``session.<method>().__aenter__() -> response``.
+    """
+    mock_response = MagicMock()
+    for k, v in response_attrs.items():
+        setattr(mock_response, k, v)
+
+    mock_session = MagicMock()
+    method_mock = getattr(mock_session, method)  # pyright: ignore[reportAny]
+    method_mock.return_value.__aenter__ = AsyncMock(  # pyright: ignore[reportAny]
+        return_value=mock_response
+    )
+    method_mock.return_value.__aexit__ = AsyncMock(  # pyright: ignore[reportAny]
+        return_value=None
+    )
+
+    mock_factory = MagicMock()
+    mock_factory.return_value.__aenter__ = AsyncMock(  # pyright: ignore[reportAny]
+        return_value=mock_session
+    )
+    mock_factory.return_value.__aexit__ = AsyncMock(  # pyright: ignore[reportAny]
+        return_value=None
+    )
+    return mock_factory
+
+
+REAL_HF_429_HEADER_DICT = {"ratelimit": '"api";r=0;t=52'}
+
+
+class TestRateLimitAtHttpCallSites:
+    """Verify each HF call site translates an HTTP 429 into a
+    ``HuggingFaceRateLimitError`` carrying the parsed ``retry_after``.
+
+    These tests would catch regressions where (a) the 429 branch is
+    deleted, (b) ``_parse_retry_after`` stops being called, or
+    (c) the wrong header object is passed to it.
+    """
+
+    async def test_fetch_file_list_maps_429_to_rate_limit_error(self) -> None:
+        mock_factory = _make_mock_session_returning(
+            {"status": 429, "headers": REAL_HF_429_HEADER_DICT}
+        )
+        with (
+            patch("exo.download.download_utils.create_http_session", mock_factory),
+            pytest.raises(HuggingFaceRateLimitError) as exc_info,
+        ):
+            await _fetch_file_list(ModelId("test/model"), "main")
+        assert exc_info.value.retry_after == 52.0
+
+    async def test_file_meta_maps_429_to_rate_limit_error(self) -> None:
+        mock_factory = _make_mock_session_returning(
+            {"status": 429, "headers": REAL_HF_429_HEADER_DICT}, method="head"
+        )
+        with (
+            patch("exo.download.download_utils.create_http_session", mock_factory),
+            pytest.raises(HuggingFaceRateLimitError) as exc_info,
+        ):
+            await file_meta(ModelId("test/model"), "main", "weights.safetensors")
+        assert exc_info.value.retry_after == 52.0
+
+    async def test_file_meta_maps_429_after_307_redirect(self) -> None:
+        """When the initial HEAD 307s and the redirected HEAD then 429s,
+        the 429 must still surface as ``HuggingFaceRateLimitError``."""
+        # First HEAD -> 307 with a Location header pointing somewhere new.
+        first_response = MagicMock()
+        first_response.status = 307
+        first_response.headers = {"location": "/redirected/url"}
+
+        # Second HEAD (the recursive call) -> 429 with the real-HF header.
+        second_response = MagicMock()
+        second_response.status = 429
+        second_response.headers = REAL_HF_429_HEADER_DICT
+
+        responses = iter([first_response, second_response])
+
+        mock_session = MagicMock()
+        mock_session.head.return_value.__aenter__ = AsyncMock(  # pyright: ignore[reportAny]
+            side_effect=lambda: next(responses)
+        )
+        mock_session.head.return_value.__aexit__ = AsyncMock(  # pyright: ignore[reportAny]
+            return_value=None
+        )
+
+        mock_factory = MagicMock()
+        mock_factory.return_value.__aenter__ = AsyncMock(  # pyright: ignore[reportAny]
+            return_value=mock_session
+        )
+        mock_factory.return_value.__aexit__ = AsyncMock(  # pyright: ignore[reportAny]
+            return_value=None
+        )
+
+        with (
+            patch("exo.download.download_utils.create_http_session", mock_factory),
+            pytest.raises(HuggingFaceRateLimitError) as exc_info,
+        ):
+            await file_meta(ModelId("test/model"), "main", "weights.safetensors")
+        assert exc_info.value.retry_after == 52.0
+
+    async def test_download_file_maps_429_to_rate_limit_error(
+        self, tmp_path: Path
+    ) -> None:
+        target_dir = tmp_path / "downloads"
+        await aios.makedirs(target_dir, exist_ok=True)
+        # No local file -> _download_file goes straight to file_meta then GET.
+        # We need both calls to succeed enough to reach the GET branch:
+        #   - file_meta returns a non-429 (size, etag) so we proceed.
+        #   - the GET then 429s.
+        with (
+            patch(
+                "exo.download.download_utils.file_meta",
+                new_callable=AsyncMock,
+                return_value=(100, "abc123"),
+            ),
+            patch(
+                "exo.download.download_utils.create_http_session",
+                _make_mock_session_returning(
+                    {"status": 429, "headers": REAL_HF_429_HEADER_DICT}
+                ),
+            ),
+            pytest.raises(HuggingFaceRateLimitError) as exc_info,
+        ):
+            await _download_file(
+                ModelId("test/model"), "main", "weights.safetensors", target_dir
+            )
+        assert exc_info.value.retry_after == 52.0

src/exo/utils/async_process.py

@@ -0,0 +1,352 @@
+from __future__ import annotations
+
+import contextlib
+import faulthandler
+import multiprocessing as mp
+import os
+import sys
+from collections.abc import Callable, Iterable, Mapping
+from multiprocessing.context import SpawnContext
+from multiprocessing.process import BaseProcess
+from multiprocessing.resource_sharer import DupFd
+from signal import Signals
+from typing import final
+
+from anyio import (
+    BrokenResourceError,
+    CancelScope,
+    ClosedResourceError,
+    Event,
+    Lock,
+    create_memory_object_stream,
+    move_on_after,
+    sleep,
+    wait_readable,
+)
+from anyio.abc import (
+    ByteReceiveStream,
+    ObjectReceiveStream,
+    ObjectSendStream,
+)
+
+from exo.utils.task_group import TaskGroup
+
+_STDOUT_FD = 1
+_STDERR_FD = 2
+_READ_CHUNK_SIZE = 64 * 1024
+_TERMINATE_GRACE_SECONDS = 5.0
+_KILL_GRACE_SECONDS = 5.0
+
+
+@final
+class MemoryByteReceiveStream(ByteReceiveStream):
+    def __init__(self, receive_stream: ObjectReceiveStream[bytes]) -> None:
+        self._receive_stream = receive_stream
+        self._buffer = bytearray()
+
+    async def receive(self, max_bytes: int = _READ_CHUNK_SIZE) -> bytes:
+        if max_bytes <= 0:
+            raise ValueError("max_bytes must be positive")
+
+        if self._buffer:
+            chunk = bytes(self._buffer[:max_bytes])
+            del self._buffer[:max_bytes]
+            return chunk
+
+        chunk = await self._receive_stream.receive()
+        if len(chunk) <= max_bytes:
+            return chunk
+
+        self._buffer.extend(chunk[max_bytes:])
+        return chunk[:max_bytes]
+
+    async def aclose(self) -> None:
+        self._buffer.clear()
+        await self._receive_stream.aclose()
+
+
+@final
+class AsyncSpawnProcess:
+    @staticmethod
+    def context() -> SpawnContext:
+        return mp.get_context("spawn")
+
+    def __init__(
+        self,
+        target: Callable[..., object] | None = None,
+        name: str | None = None,
+        args: Iterable[object] = (),
+        kwargs: Mapping[str, object] | None = None,
+        *,
+        daemon: bool | None = None,
+        stream_buffer_size: int = 16,
+    ) -> None:
+        if stream_buffer_size <= 0:
+            raise ValueError("stream_buffer_size must be positive")
+
+        # setup state
+        self._target = target
+        self._name = name
+        self._args = args
+        self._kwargs = kwargs
+        self._daemon = daemon
+        self._stream_buffer_size = stream_buffer_size
+
+        # lifecycle state
+        self._process: BaseProcess | None = None
+        self._pid: int | None = None
+        self._stdout: MemoryByteReceiveStream | None = None
+        self._stderr: MemoryByteReceiveStream | None = None
+        self._tg = TaskGroup()
+        self._started = Event()
+        self._stopped = Event()
+        self._wait_lock = Lock()
+        self._start_error: BaseException | None = None
+        self._has_stopped = False
+        self._closed = False
+        self._exitcode: int | None = None
+
+    async def run(self) -> None:
+        if self._closed:
+            raise RuntimeError("process has been closed")
+        if self._process is not None:
+            raise RuntimeError("process has already been started")
+
+        stdout_read_fd, stdout_write_fd = os.pipe()
+        stderr_read_fd, stderr_write_fd = os.pipe()
+        stdout_send, stdout_receive = create_memory_object_stream[bytes](
+            self._stream_buffer_size
+        )
+        stderr_send, stderr_receive = create_memory_object_stream[bytes](
+            self._stream_buffer_size
+        )
+
+        try:
+            process = self.context().Process(
+                target=_run_with_captured_stdio,
+                name=self._name,
+                args=(
+                    DupFd(stdout_write_fd),
+                    DupFd(stderr_write_fd),
+                    self._target,
+                    *self._args,
+                ),
+                kwargs={} if self._kwargs is None else self._kwargs,
+                daemon=self._daemon,
+            )
+            process.start()
+            pid = process.pid
+            if pid is None:
+                raise RuntimeError("started process has no pid")
+
+            # important to close parent write-side FD to prevent hangs
+            _close_fd(stdout_write_fd)
+            _close_fd(stderr_write_fd)
+
+            self._process = process
+            self._pid = pid
+            self._stdout = MemoryByteReceiveStream(stdout_receive)
+            self._stderr = MemoryByteReceiveStream(stderr_receive)
+            self._started.set()
+        except BaseException as exc:
+            self._start_error = exc
+            self._started.set()
+            self._has_stopped = True
+            self._stopped.set()
+            for stream in (stdout_send, stderr_send, stdout_receive, stderr_receive):
+                with contextlib.suppress(Exception):
+                    await stream.aclose()
+            for fd in (
+                stdout_read_fd,
+                stdout_write_fd,
+                stderr_read_fd,
+                stderr_write_fd,
+            ):
+                _close_fd(fd)
+            raise
+
+        try:
+            async with self._tg as tg:
+                tg.start_soon(_drain_fd, stdout_read_fd, stdout_send)
+                tg.start_soon(_drain_fd, stderr_read_fd, stderr_send)
+                await self.wait()
+        finally:
+            try:
+                with CancelScope(shield=True):
+                    await self._terminate_if_still_alive()
+            finally:
+                self._has_stopped = True
+                self._stopped.set()
+
+    async def wait_started(self) -> None:
+        await self._started.wait()
+        if self._start_error is not None:
+            raise self._start_error
+
+    async def wait_stopped(self) -> None:
+        await self._stopped.wait()
+
+    def shutdown(self) -> None:
+        if not self._has_stopped and self._tg.is_running():
+            self._tg.cancel_tasks()
+
+    async def aclose(self) -> None:
+        if self._closed:
+            return
+
+        self.shutdown()
+        if self._process is not None and not self._has_stopped:
+            await self.wait_stopped()
+        self.close()
+
+    async def wait(self) -> int:
+        if self._exitcode is not None:
+            return self._exitcode
+
+        async with self._wait_lock:
+            if self._exitcode is not None:
+                return self._exitcode
+
+            process = self.process
+            while True:
+                exitcode = process.exitcode
+                if exitcode is not None:
+                    process.join(0)
+                    self._exitcode = exitcode
+                    return exitcode
+
+                await sleep(0.01)
+
+    def terminate(self) -> None:
+        self.process.terminate()
+
+    def is_alive(self) -> bool:
+        if self._process is None:
+            return False
+        with contextlib.suppress(ValueError):
+            return self._process.is_alive()
+        return False
+
+    def join(self, timeout: float | None = None) -> None:
+        self.process.join(timeout)
+
+    def close(self) -> None:
+        self._closed = True
+        if self._process is None:
+            return
+        with contextlib.suppress(ValueError):
+            self._process.close()
+
+    def kill(self) -> None:
+        self.process.kill()
+
+    def send_signal(self, signal: Signals) -> None:
+        os.kill(self.pid, signal)
+
+    @property
+    def pid(self) -> int:
+        if self._pid is None:
+            raise RuntimeError("process has not been started")
+        return self._pid
+
+    @property
+    def exitcode(self) -> int | None:
+        if self._exitcode is not None:
+            return self._exitcode
+        if self._process is None:
+            return None
+
+        with contextlib.suppress(ValueError):
+            exitcode = self._process.exitcode
+            if exitcode is not None:
+                self._exitcode = exitcode
+            return exitcode
+        return None
+
+    @property
+    def stdout(self) -> ByteReceiveStream:
+        if self._stdout is None:
+            raise RuntimeError("process has not been started")
+        return self._stdout
+
+    @property
+    def stderr(self) -> ByteReceiveStream:
+        if self._stderr is None:
+            raise RuntimeError("process has not been started")
+        return self._stderr
+
+    @property
+    def process(self) -> BaseProcess:
+        if self._process is None:
+            raise RuntimeError("process has not been started")
+        return self._process
+
+    async def _terminate_if_still_alive(self) -> None:
+        process = self._process
+        if process is None:
+            return
+
+        if self.exitcode is not None:
+            return
+
+        with contextlib.suppress(ValueError):
+            if process.is_alive():
+                process.terminate()
+                with move_on_after(_TERMINATE_GRACE_SECONDS):
+                    await self.wait()
+
+            if self.exitcode is not None or not process.is_alive():
+                return
+
+            process.kill()
+            with move_on_after(_KILL_GRACE_SECONDS):
+                await self.wait()
+
+            if self.exitcode is not None or not process.is_alive():
+                return
+
+            raise RuntimeError(f"process {self.pid} is still alive after SIGKILL")
+
+
+# Spawn-mode multiprocessing requires a module-level target that can be pickled.
+def _run_with_captured_stdio(
+    stdout: DupFd,
+    stderr: DupFd,
+    target: Callable[..., object] | None,
+    *target_args: object,
+    **target_kwargs: object,
+) -> None:
+    stdout_fd = stdout.detach()
+    stderr_fd = stderr.detach()
+
+    try:
+        os.dup2(stdout_fd, _STDOUT_FD)
+        os.dup2(stderr_fd, _STDERR_FD)
+    finally:
+        for fd in (stdout_fd, stderr_fd):
+            if fd not in (_STDOUT_FD, _STDERR_FD):
+                _close_fd(fd)
+
+    faulthandler.enable(file=sys.stderr, all_threads=True)
+    if target is not None:
+        target(*target_args, **target_kwargs)
+
+
+async def _drain_fd(fd: int, send_stream: ObjectSendStream[bytes]) -> None:
+    try:
+        while True:
+            await wait_readable(fd)
+            chunk = os.read(fd, _READ_CHUNK_SIZE)
+            if not chunk:
+                return
+            await send_stream.send(chunk)
+    except (BrokenPipeError, BrokenResourceError, ClosedResourceError):
+        pass
+    finally:
+        _close_fd(fd)
+        await send_stream.aclose()
+
+
+def _close_fd(fd: int) -> None:
+    with contextlib.suppress(OSError):
+        os.close(fd)

src/exo/utils/channels.py

@@ -2,6 +2,7 @@ import contextlib
 import multiprocessing as mp
 from dataclasses import dataclass, field
 from math import inf
+from multiprocessing.context import BaseContext
 from multiprocessing.synchronize import Event
 from queue import Empty, Full
 from types import TracebackType
@@ -79,7 +80,7 @@ class _MpEndOfStream:
 
 
 class MpState[T]:
-    def __init__(self, max_buffer_size: float):
+    def __init__(self, max_buffer_size: float, mp_ctx: BaseContext):
         if max_buffer_size == inf:
             max_buffer_size = 0
         assert isinstance(max_buffer_size, int), (
@@ -87,8 +88,8 @@ class MpState[T]:
         )
 
         self.max_buffer_size: float = max_buffer_size
-        self.buffer: mp.Queue[T | _MpEndOfStream] = mp.Queue(max_buffer_size)
-        self.closed: Event = mp.Event()
+        self.buffer: mp.Queue[T | _MpEndOfStream] = mp_ctx.Queue(max_buffer_size)
+        self.closed: Event = mp_ctx.Event()
 
     def __getstate__(self):
         d = self.__dict__.copy()
@@ -296,7 +297,9 @@ class mp_channel[T]:  # noqa: N801
     """Create a pair of synchronous channels for interprocess communication"""
 
     # max buffer size uses math.inf to represent an unbounded queue, and 0 to represent a yet unimplemented "unbuffered" queue.
-    def __new__(cls, max_buffer_size: float = inf) -> tuple[MpSender[T], MpReceiver[T]]:
+    def __new__(
+        cls, max_buffer_size: float = inf, *, context: BaseContext | None = None
+    ) -> tuple[MpSender[T], MpReceiver[T]]:
         if (
             max_buffer_size == 0
             or max_buffer_size != inf
@@ -305,5 +308,7 @@ class mp_channel[T]:  # noqa: N801
             raise ValueError(
                 "max_buffer_size must be either an integer or math.inf. 0-sized buffers are not supported by multiprocessing"
             )
-        state = MpState[T](max_buffer_size)
+        state = MpState[T](
+            max_buffer_size, mp.get_context() if context is None else context
+        )
         return MpSender(_state=state), MpReceiver(_state=state)

src/exo/utils/tests/test_async_process.py

@@ -0,0 +1,459 @@
+import contextlib
+import os
+import signal
+import sys
+import time
+from collections.abc import Callable
+from types import FrameType
+
+import mlx.core as mx
+import pytest
+from _pytest.capture import CaptureFixture
+from anyio import EndOfStream, create_task_group, fail_after
+from anyio.abc import ByteReceiveStream
+from pytest import MonkeyPatch
+
+import exo.utils.async_process as async_process
+from exo.utils.async_process import (
+    AsyncSpawnProcess,
+)
+from exo.utils.channels import MpSender, mp_channel
+
+
+def _write_to_stdio(prefix: str, *, stderr_suffix: str) -> None:
+    print(f"{prefix}: python stdout")
+    print(f"{prefix}: python stderr {stderr_suffix}", file=sys.stderr)
+    os.write(1, f"{prefix}: fd stdout\n".encode())
+    os.write(2, f"{prefix}: fd stderr {stderr_suffix}\n".encode())
+
+
+def _write_large_output() -> None:
+    os.write(1, b"stdout-0123456789")
+    os.write(2, b"stderr-0123456789")
+
+
+def _write_all(fd: int, data: bytes) -> None:
+    remaining = memoryview(data)
+    while remaining:
+        written = os.write(fd, remaining)
+        remaining = remaining[written:]
+
+
+def _write_large_exact_output(size: int) -> None:
+    _write_all(1, b"stdout:" + (b"x" * size))
+    _write_all(2, b"stderr:" + (b"y" * size))
+
+
+def _raise_after_stderr_write() -> None:
+    os.write(2, b"stderr before exception\n")
+    raise RuntimeError("child boom")
+
+
+def _exit_after_stdio_write(prefix: str, exitcode: int) -> None:
+    os.write(1, f"{prefix}: stdout before _exit\n".encode())
+    os.write(2, f"{prefix}: stderr before _exit\n".encode())
+    os._exit(exitcode)
+
+
+def _abort_after_stdio_write(prefix: str) -> None:
+    os.write(1, f"{prefix}: stdout before abort\n".encode())
+    os.write(2, f"{prefix}: stderr before abort\n".encode())
+    os.abort()
+
+
+def _close_stdio_and_exit() -> None:
+    os.close(1)
+    os.close(2)
+    os._exit(0)
+
+
+def _sleep_without_output() -> None:
+    time.sleep(0.1)
+
+
+def _exit_on_sigterm(exitcode: int) -> None:
+    def handle_sigterm(_signum: int, _frame: FrameType | None) -> None:
+        os._exit(exitcode)
+
+    signal.signal(signal.SIGTERM, handle_sigterm)
+    os.write(1, b"sigterm-ready\n")
+    while True:
+        time.sleep(0.1)
+
+
+def _ignore_sigterm_forever() -> None:
+    signal.signal(signal.SIGTERM, signal.SIG_IGN)
+    os.write(1, b"sigterm-ready\n")
+    while True:
+        time.sleep(0.1)
+
+
+def _send_over_mp_channel(send: MpSender[str]) -> None:
+    send.send("hello from child")
+    send.close()
+
+
+def _mlx_force_oom(size: int = 40_000) -> None:
+    """
+    Force an Out-Of-Memory (OOM) error in MLX by performing large tensor operations.
+    """
+    print("CHILD: start")
+
+    mx.set_default_device(mx.gpu)
+    a = mx.random.uniform(shape=(size, size), dtype=mx.float32)
+    b = mx.random.uniform(shape=(size, size), dtype=mx.float32)
+    mx.eval(a, b)
+    c = mx.matmul(a, b)
+    d = mx.matmul(a, c)
+    e = mx.matmul(b, c)
+    f = mx.sigmoid(d + e)
+    mx.eval(f)
+
+    print("CHILD: end")
+
+
+async def _collect_stream(
+    stream: ByteReceiveStream,
+    output: bytearray,
+) -> None:
+    while True:
+        try:
+            output.extend(await stream.receive())
+        except EndOfStream:
+            return
+
+
+async def _collect_process_output(
+    process: AsyncSpawnProcess,
+) -> tuple[int, bytes, bytes]:
+    stdout = bytearray()
+    stderr = bytearray()
+
+    async with create_task_group() as task_group:
+        task_group.start_soon(_collect_stream, process.stdout, stdout)
+        task_group.start_soon(_collect_stream, process.stderr, stderr)
+        await process.wait()
+
+    if process.exitcode is None:
+        raise RuntimeError("process exited without a return code")
+    exitcode = process.exitcode
+    return exitcode, bytes(stdout), bytes(stderr)
+
+
+def _fd_identity(fd: int) -> tuple[int, int]:
+    fd_stat = os.fstat(fd)
+    return fd_stat.st_dev, fd_stat.st_ino
+
+
+def _fd_count() -> int | None:
+    for fd_dir in ("/proc/self/fd", "/dev/fd"):
+        with contextlib.suppress(OSError):
+            return len(os.listdir(fd_dir))
+    return None
+
+
+async def _run_and_collect(
+    target: Callable[..., object] | None,
+    *,
+    args: tuple[object, ...] = (),
+    kwargs: dict[str, object] | None = None,
+    stream_buffer_size: int = 16,
+) -> tuple[int, bytes, bytes]:
+    process = AsyncSpawnProcess(
+        target,
+        args=args,
+        kwargs=kwargs,
+        stream_buffer_size=stream_buffer_size,
+    )
+    result: tuple[int, bytes, bytes] | None = None
+    async with create_task_group() as task_group:
+        task_group.start_soon(process.run)
+        await process.wait_started()
+        result = await _collect_process_output(process)
+    if result is None:
+        raise RuntimeError("process collection did not run")
+    return result
+
+
+@pytest.mark.asyncio
+async def test_spawn_process_captures_stdout_and_stderr_separately(
+    capfd: CaptureFixture[str],
+) -> None:
+    process = AsyncSpawnProcess(
+        _write_to_stdio,
+        args=("child",),
+        kwargs={"stderr_suffix": "error"},
+    )
+    result: tuple[int, bytes, bytes] | None = None
+    async with create_task_group() as task_group:
+        task_group.start_soon(process.run)
+        await process.wait_started()
+        result = await _collect_process_output(process)
+    if result is None:
+        raise RuntimeError("process collection did not run")
+    exitcode, stdout_bytes, stderr_bytes = result
+
+    parent_output = capfd.readouterr()
+    stdout = stdout_bytes.decode("utf-8", errors="replace")
+    stderr = stderr_bytes.decode("utf-8", errors="replace")
+
+    assert exitcode == 0
+    assert "child: python stdout" in stdout
+    assert "child: fd stdout" in stdout
+    assert "child: python stderr error" in stderr
+    assert "child: fd stderr error" in stderr
+    assert "child:" not in parent_output.out
+    assert "child:" not in parent_output.err
+
+
+@pytest.mark.asyncio
+async def test_process_with_no_target_exits_successfully() -> None:
+    exitcode, stdout, stderr = await _run_and_collect(None)
+
+    assert exitcode == 0
+    assert stdout == b""
+    assert stderr == b""
+
+
+@pytest.mark.asyncio
+async def test_stdout_stream_honors_receive_size() -> None:
+    process = AsyncSpawnProcess(_write_large_output)
+    first_stdout: bytes | None = None
+    remaining_stdout = bytearray()
+    stderr = bytearray()
+
+    async with create_task_group() as task_group:
+        task_group.start_soon(process.run)
+        await process.wait_started()
+        first_stdout = await process.stdout.receive(6)
+
+        async with create_task_group() as collect_group:
+            collect_group.start_soon(_collect_stream, process.stdout, remaining_stdout)
+            collect_group.start_soon(_collect_stream, process.stderr, stderr)
+            await process.wait()
+
+    if first_stdout is None:
+        raise RuntimeError("process stdout was not read")
+    if process.exitcode is None:
+        raise RuntimeError("process exited without a return code")
+    exitcode = process.exitcode
+    assert exitcode == 0
+    assert first_stdout == b"stdout"
+    assert bytes(remaining_stdout) == b"-0123456789"
+    assert bytes(stderr) == b"stderr-0123456789"
+
+
+@pytest.mark.asyncio
+async def test_large_stdout_and_stderr_are_not_lost_with_bounded_buffers() -> None:
+    size = 1024 * 1024
+    exitcode, stdout, stderr = await _run_and_collect(
+        _write_large_exact_output,
+        args=(size,),
+        stream_buffer_size=1,
+    )
+
+    assert exitcode == 0
+    assert stdout == b"stdout:" + (b"x" * size)
+    assert stderr == b"stderr:" + (b"y" * size)
+
+
+@pytest.mark.asyncio
+async def test_child_exception_traceback_is_captured_from_stderr() -> None:
+    process = AsyncSpawnProcess(_raise_after_stderr_write)
+    result: tuple[int, bytes, bytes] | None = None
+
+    async with create_task_group() as task_group:
+        task_group.start_soon(process.run)
+        await process.wait_started()
+        result = await _collect_process_output(process)
+    if result is None:
+        raise RuntimeError("process collection did not run")
+    exitcode, _, stderr_bytes = result
+
+    assert exitcode == 1
+    stderr = stderr_bytes.decode("utf-8", errors="replace")
+    assert "stderr before exception" in stderr
+    assert "RuntimeError: child boom" in stderr
+
+
+@pytest.mark.asyncio
+async def test_repeated_bad_children_do_not_pollute_or_replace_parent_stdio(
+    capfd: CaptureFixture[str],
+) -> None:
+    stdout_object = sys.stdout
+    stderr_object = sys.stderr
+    stdout_identity = _fd_identity(1)
+    stderr_identity = _fd_identity(2)
+
+    cases: tuple[tuple[Callable[..., object], tuple[object, ...]], ...] = (
+        (_raise_after_stderr_write, ()),
+        (_exit_after_stdio_write, ("exit-child", 17)),
+        (_abort_after_stdio_write, ("abort-child",)),
+    )
+
+    for iteration in range(3):
+        for target, args in cases:
+            exitcode, stdout, stderr = await _run_and_collect(
+                target,
+                args=args,
+                stream_buffer_size=1,
+            )
+
+            assert exitcode != 0
+            if target is _exit_after_stdio_write:
+                assert stdout == b"exit-child: stdout before _exit\n"
+                assert stderr == b"exit-child: stderr before _exit\n"
+            elif target is _abort_after_stdio_write:
+                assert b"abort-child: stdout before abort\n" in stdout
+                assert b"abort-child: stderr before abort\n" in stderr
+                assert exitcode == -signal.SIGABRT
+            else:
+                assert stdout == b""
+                assert b"stderr before exception\n" in stderr
+                assert b"RuntimeError: child boom" in stderr
+
+        print(f"parent stdout still works {iteration}")
+        print(f"parent stderr still works {iteration}", file=sys.stderr)
+
+    parent_output = capfd.readouterr()
+
+    assert sys.stdout is stdout_object
+    assert sys.stderr is stderr_object
+    assert _fd_identity(1) == stdout_identity
+    assert _fd_identity(2) == stderr_identity
+    assert "parent stdout still works 0" in parent_output.out
+    assert "parent stdout still works 2" in parent_output.out
+    assert "parent stderr still works 0" in parent_output.err
+    assert "parent stderr still works 2" in parent_output.err
+    assert "exit-child:" not in parent_output.out
+    assert "exit-child:" not in parent_output.err
+    assert "abort-child:" not in parent_output.out
+    assert "abort-child:" not in parent_output.err
+    assert "child boom" not in parent_output.err
+
+
+@pytest.mark.asyncio
+async def test_child_can_close_stdio_without_corrupting_parent_stdio(
+    capfd: CaptureFixture[str],
+) -> None:
+    stdout_identity = _fd_identity(1)
+    stderr_identity = _fd_identity(2)
+
+    exitcode, stdout, stderr = await _run_and_collect(_close_stdio_and_exit)
+    os.write(1, b"parent stdout after child closed stdio\n")
+    os.write(2, b"parent stderr after child closed stdio\n")
+    parent_output = capfd.readouterr()
+
+    assert exitcode == 0
+    assert stdout == b""
+    assert stderr == b""
+    assert _fd_identity(1) == stdout_identity
+    assert _fd_identity(2) == stderr_identity
+    assert "parent stdout after child closed stdio" in parent_output.out
+    assert "parent stderr after child closed stdio" in parent_output.err
+
+
+@pytest.mark.asyncio
+async def test_repeated_crashing_children_do_not_grow_parent_fd_table() -> None:
+    await _run_and_collect(_exit_after_stdio_write, args=("warmup", 23))
+    before = _fd_count()
+    if before is None:
+        pytest.skip("fd table count is not available on this platform")
+
+    for iteration in range(20):
+        exitcode, stdout, stderr = await _run_and_collect(
+            _exit_after_stdio_write,
+            args=(f"fd-child-{iteration}", 31),
+            stream_buffer_size=1,
+        )
+
+        assert exitcode == 31
+        assert stdout == f"fd-child-{iteration}: stdout before _exit\n".encode()
+        assert stderr == f"fd-child-{iteration}: stderr before _exit\n".encode()
+
+    after = _fd_count()
+    assert after is not None
+    assert after <= before + 2
+
+
+@pytest.mark.asyncio
+async def test_shutdown_can_cancel_idle_drainers_before_child_exits() -> None:
+    process = AsyncSpawnProcess(_sleep_without_output)
+    async with create_task_group() as task_group:
+        task_group.start_soon(process.run)
+        await process.wait_started()
+
+        with fail_after(2):
+            process.shutdown()
+            await process.wait_stopped()
+
+    assert process.exitcode is not None
+
+
+@pytest.mark.asyncio
+async def test_shutdown_allows_child_to_exit_after_sigterm() -> None:
+    process = AsyncSpawnProcess(_exit_on_sigterm, args=(43,))
+
+    async with create_task_group() as task_group:
+        task_group.start_soon(process.run)
+        await process.wait_started()
+        assert await process.stdout.receive() == b"sigterm-ready\n"
+
+        with fail_after(2):
+            process.shutdown()
+            await process.wait_stopped()
+
+    assert process.exitcode == 43
+
+
+@pytest.mark.asyncio
+async def test_shutdown_escalates_to_sigkill_when_child_ignores_sigterm(
+    monkeypatch: MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(async_process, "_TERMINATE_GRACE_SECONDS", 0.1)
+    process = AsyncSpawnProcess(_ignore_sigterm_forever)
+
+    async with create_task_group() as task_group:
+        task_group.start_soon(process.run)
+        await process.wait_started()
+        assert await process.stdout.receive() == b"sigterm-ready\n"
+
+        with fail_after(3):
+            process.shutdown()
+            await process.wait_stopped()
+
+    assert process.exitcode == -signal.SIGKILL
+
+
+@pytest.mark.asyncio
+async def test_spawn_process_can_use_spawn_context_mp_channel() -> None:
+    send, recv = mp_channel[str](context=AsyncSpawnProcess.context())
+    process = AsyncSpawnProcess(_send_over_mp_channel, args=(send,))
+
+    async with create_task_group() as task_group:
+        task_group.start_soon(process.run)
+        await process.wait_started()
+        with fail_after(2):
+            assert await recv.receive_async() == "hello from child"
+            assert await process.wait() == 0
+
+    with contextlib.suppress(Exception):
+        recv.close()
+
+
+@pytest.mark.asyncio
+@pytest.mark.skip(reason="manual MLX OOM isolation check")
+async def test_death(capsys: CaptureFixture[str]) -> None:
+    with capsys.disabled():
+        process = AsyncSpawnProcess(_mlx_force_oom)
+        stdout = b""
+        stderr = b""
+        async with create_task_group() as task_group:
+            task_group.start_soon(process.run)
+            await process.wait_started()
+            _, stdout, stderr = await _collect_process_output(process)
+
+        print("PARENT: done")
+
+        print("CHILD out:", stdout.decode("utf-8", errors="replace"))
+        print("CHILD err:", stderr.decode("utf-8", errors="replace"), "hello :)")

src/exo/worker/runner/supervisor.py

@@ -1,5 +1,4 @@
 import contextlib
-import multiprocessing as mp
 import signal
 from dataclasses import dataclass, field
 from typing import Self
@@ -8,8 +7,10 @@ import anyio
 from anyio import (
     BrokenResourceError,
     ClosedResourceError,
+    EndOfStream,
     to_thread,
 )
+from anyio.abc import ByteReceiveStream
 from loguru import logger
 
 from exo.shared.types.chunks import ErrorChunk
@@ -41,6 +42,7 @@ from exo.shared.types.worker.runners import (
     RunnerWarmingUp,
 )
 from exo.shared.types.worker.shards import ShardMetadata
+from exo.utils.async_process import AsyncSpawnProcess
 from exo.utils.channels import MpReceiver, MpSender, Sender, mp_channel
 from exo.utils.task_group import TaskGroup
 from exo.worker.runner.bootstrap import entrypoint
@@ -53,7 +55,7 @@ DECODE_TIMEOUT_SECONDS = 5
 class RunnerSupervisor:
     shard_metadata: ShardMetadata
     bound_instance: BoundInstance
-    runner_process: mp.Process
+    runner_process: AsyncSpawnProcess
     initialize_timeout: float
     _ev_recv: MpReceiver[Event]
     _task_sender: MpSender[Task]
@@ -77,11 +79,12 @@ class RunnerSupervisor:
         event_sender: Sender[Event],
         initialize_timeout: float = 400,
     ) -> Self:
-        ev_send, ev_recv = mp_channel[Event]()
-        task_sender, task_recv = mp_channel[Task]()
-        cancel_sender, cancel_recv = mp_channel[TaskId]()
+        mp_ctx = AsyncSpawnProcess.context()
+        ev_send, ev_recv = mp_channel[Event](context=mp_ctx)
+        task_sender, task_recv = mp_channel[Task](context=mp_ctx)
+        cancel_sender, cancel_recv = mp_channel[TaskId](context=mp_ctx)
 
-        runner_process = mp.Process(
+        runner_process = AsyncSpawnProcess(
             target=entrypoint,
             args=(
                 bound_instance,
@@ -109,9 +112,18 @@ class RunnerSupervisor:
         return self
 
     async def run(self):
-        self.runner_process.start()
+        runner_started = False
         try:
             async with self._tg as tg:
+                tg.start_soon(self.runner_process.run)
+                await self.runner_process.wait_started()
+                runner_started = True
+                tg.start_soon(
+                    self._forward_runner_output, "stdout", self.runner_process.stdout
+                )
+                tg.start_soon(
+                    self._forward_runner_output, "stderr", self.runner_process.stderr
+                )
                 tg.start_soon(self._watch_runner)
                 tg.start_soon(self._forward_events)
         finally:
@@ -129,41 +141,13 @@ class RunnerSupervisor:
             with contextlib.suppress(ClosedResourceError):
                 self._cancel_sender.close()
 
-            await to_thread.run_sync(self.runner_process.join, 5)
-
-            if self.runner_process.is_alive():
-                logger.warning(
-                    "Runner process didn't shutdown succesfully, terminating"
-                )
-                self.runner_process.terminate()
-                self.runner_process.join(timeout=10)
-
-                if not self.runner_process.is_alive():
-                    logger.warning("Terminated nicely in the first attempt!")
-
-                else:
-                    # Try really hard to terminate
-                    for i in range(2, 11):
-                        self.runner_process.terminate()
-                        self.runner_process.join(timeout=2)
-                        if not self.runner_process.is_alive():
-                            logger.warning(f"That took {i} attempts :)")
-                            break
-                    # Try even harder to kill
-                    else:
-                        logger.critical(
-                            "Runner process didn't respond to SIGTERM, killing"
-                        )
-                        j = 0
-                        while self.runner_process.is_alive():
-                            j += 1
-                            self.runner_process.kill()
-                            self.runner_process.join(timeout=5)
-                            logger.warning(f"That took {j} attempts :(")
-            else:
-                logger.info("Runner process succesfully terminated")
-
-            self.runner_process.close()
+            if runner_started:
+                with anyio.CancelScope(shield=True):
+                    self.runner_process.shutdown()
+                    await self.runner_process.wait_stopped()
+                    if not self.runner_process.is_alive():
+                        logger.info("Runner process succesfully terminated")
+                    self.runner_process.close()
 
     def shutdown(self):
         self._tg.cancel_tasks()
@@ -249,6 +233,25 @@ class RunnerSupervisor:
                 if not self.runner_process.is_alive():
                     await self._check_runner(RuntimeError("Runner found to be dead"))
 
+    async def _forward_runner_output(
+        self,
+        stream_name: str,
+        stream: ByteReceiveStream,
+    ) -> None:
+        while True:
+            try:
+                chunk = await stream.receive()
+            except (EndOfStream, ClosedResourceError, BrokenResourceError):
+                return
+
+            message = chunk.decode("utf-8", errors="replace").rstrip()
+            if not message:
+                continue
+            if stream_name == "stderr":
+                logger.warning(f"Runner stderr: {message}")
+            else:
+                logger.debug(f"Runner stdout: {message}")
+
     async def _check_runner(self, e: Exception) -> None:
         if not self._cancel_watch_runner.cancel_called:
             self._cancel_watch_runner.cancel()

src/exo/worker/tests/unittests/test_runner/test_runner_supervisor.py

@@ -1,4 +1,3 @@
-import multiprocessing as mp
 from typing import cast
 
 import anyio
@@ -16,6 +15,7 @@ from exo.shared.types.text_generation import (
 )
 from exo.shared.types.worker.instances import BoundInstance, InstanceId
 from exo.shared.types.worker.runners import RunnerFailed, RunnerId
+from exo.utils.async_process import AsyncSpawnProcess
 from exo.utils.channels import channel, mp_channel
 from exo.worker.runner.supervisor import RunnerSupervisor
 from exo.worker.tests.unittests.conftest import get_bound_mlx_ring_instance
@@ -24,19 +24,11 @@ from exo.worker.tests.unittests.conftest import get_bound_mlx_ring_instance
 class _DeadProcess:
     exitcode = -6
 
-    def start(self) -> None:
-        return None
-
     def is_alive(self) -> bool:
         return False
 
-    def join(self, _timeout: float | None = None) -> None:
-        return None
-
-    def terminate(self) -> None:
-        return None
-
-    def kill(self) -> None:
+    def join(self, timeout: float | None = None) -> None:
+        _ = timeout
         return None
 
 
@@ -57,7 +49,7 @@ async def test_check_runner_emits_error_chunk_for_inflight_text_generation() ->
     supervisor = RunnerSupervisor(
         shard_metadata=bound_instance.bound_shard,
         bound_instance=bound_instance,
-        runner_process=cast("mp.Process", cast(object, _DeadProcess())),
+        runner_process=cast(AsyncSpawnProcess, cast(object, _DeadProcess())),
         initialize_timeout=400,
         _ev_recv=ev_recv,
         _task_sender=task_sender,