feat: add transient event transport

## 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

src/exo/api/main.py

@@ -121,6 +121,8 @@ from exo.api.types.openai_responses import (
 )
 from exo.master.image_store import ImageStore
 from exo.master.placement import place_instance as get_instance_placements
+from exo.routing.event_router import EventRouter
+from exo.routing.snapshot_receiver import SnapshotReceiver
 from exo.shared.apply import apply
 from exo.shared.constants import (
     DASHBOARD_DIR,
@@ -164,6 +166,7 @@ from exo.shared.types.commands import (
     ImageEdits,
     ImageGeneration,
     PlaceInstance,
+    RequestSnapshot,
     SendInputChunk,
     SetInstanceLink,
     StartDownload,
@@ -171,16 +174,16 @@ from exo.shared.types.commands import (
     TaskFinished,
     TextGeneration,
 )
-from exo.shared.types.common import CommandId, Id, NodeId, SystemId
+from exo.shared.types.common import CommandId, Id, NodeId, SessionId, SystemId
 from exo.shared.types.events import (
     ChunkGenerated,
     Event,
     IndexedEvent,
-    InstanceDeleted,
     TracesMerged,
 )
 from exo.shared.types.instance_link import InstanceLink, InstanceLinkId
 from exo.shared.types.memory import Memory
+from exo.shared.types.snapshots import SnapshotChunk
 from exo.shared.types.state import State
 from exo.shared.types.tasks import (
     ImageEdits as ImageEditsTask,
@@ -207,6 +210,8 @@ from exo.utils.task_group import TaskGroup
 _API_EVENT_LOG_DIR = EXO_EVENT_LOG_DIR / "api"
 ONBOARDING_COMPLETE_FILE = EXO_CACHE_HOME / "onboarding_complete"
 
+_SNAPSHOT_FETCH_TIMEOUT_SECONDS = 30
+
 
 def _format_to_content_type(image_format: Literal["png", "jpeg", "webp"] | None) -> str:
     return f"image/{image_format or 'png'}"
@@ -236,9 +241,12 @@ class API:
     def __init__(
         self,
         node_id: NodeId,
+        session_id: SessionId,
         *,
         port: int,
+        event_router: EventRouter,
         event_receiver: Receiver[IndexedEvent],
+        snapshot_chunk_receiver: Receiver[SnapshotChunk],
         command_sender: Sender[ForwarderCommand],
         download_command_sender: Sender[ForwarderDownloadCommand],
         # This lets us pause the API if an election is running
@@ -247,14 +255,16 @@ class API:
         self.state = State()
         self._event_log = DiskEventLog(_API_EVENT_LOG_DIR)
         self._system_id = SystemId()
+        self.session_id = session_id
+        self.event_router = event_router
         self.command_sender = command_sender
         self.download_command_sender = download_command_sender
         self.event_receiver = event_receiver
+        self.snapshot_chunk_receiver = snapshot_chunk_receiver
         self.election_receiver = election_receiver
         self.node_id: NodeId = node_id
         self.last_completed_election: int = 0
         self.port = port
-        self._sent_image_hashes: set[str] = set()
 
         self.paused: bool = False
         self.paused_ev: anyio.Event = anyio.Event()
@@ -289,22 +299,34 @@ class API:
         self._image_generation_queues: dict[
             CommandId, Sender[ImageChunk | ErrorChunk]
         ] = {}
+        self._observed_generation_commands: set[CommandId] = set()
         self._image_store = ImageStore(EXO_IMAGE_CACHE_DIR)
         self._tg: TaskGroup = TaskGroup()
 
-    def reset(self, result_clock: int, event_receiver: Receiver[IndexedEvent]):
+    def reset(
+        self,
+        result_clock: int,
+        session_id: SessionId,
+        event_router: EventRouter,
+        event_receiver: Receiver[IndexedEvent],
+        snapshot_chunk_receiver: Receiver[SnapshotChunk],
+    ):
         logger.info("Resetting API State")
         self._event_log.close()
         self._event_log = DiskEventLog(_API_EVENT_LOG_DIR)
         self.state = State()
         self._system_id = SystemId()
+        self.session_id = session_id
+        self.event_router = event_router
         self._text_generation_queues = {}
         self._image_generation_queues = {}
+        self._observed_generation_commands = set()
         self.unpause(result_clock)
         self.event_receiver.close()
         self.event_receiver = event_receiver
-        self._tg.start_soon(self._apply_state)
-        self._sent_image_hashes = set()
+        self.snapshot_chunk_receiver.close()
+        self.snapshot_chunk_receiver = snapshot_chunk_receiver
+        self._tg.start_soon(self._bootstrap_then_apply_state)
 
     def unpause(self, result_clock: int):
         logger.info("Unpausing API")
@@ -826,18 +848,8 @@ class API:
         )
         command = TextGeneration(task_params=task_params)
 
-        new_images: list[tuple[int, str]] = []
-        for idx, (img, h) in enumerate(zip(images, hashes, strict=True)):
-            if h not in self._sent_image_hashes:
-                self._sent_image_hashes.add(h)
-                new_images.append((idx, img))
-
-        if not new_images:
-            await self._send(command)
-            return command
-
         all_chunks: list[tuple[int, str]] = []
-        for img_idx, img_data in new_images:
+        for img_idx, img_data in enumerate(images):
             for i in range(0, len(img_data), EXO_MAX_CHUNK_SIZE):
                 all_chunks.append((img_idx, img_data[i : i + EXO_MAX_CHUNK_SIZE]))
 
@@ -1848,7 +1860,8 @@ class API:
         try:
             async with self._tg as tg:
                 logger.info("Starting API")
-                tg.start_soon(self._apply_state)
+                tg.start_soon(self._bootstrap_then_apply_state)
+                tg.start_soon(self._reconcile_streams)
                 tg.start_soon(self._pause_on_new_election)
                 tg.start_soon(self._cleanup_expired_images)
                 print_startup_banner(self.port)
@@ -1862,6 +1875,7 @@ class API:
             self._event_log.close()
             self.command_sender.close()
             self.event_receiver.close()
+            self.snapshot_chunk_receiver.close()
 
     async def run_api(self, ev: anyio.Event):
         cfg = Config()
@@ -1877,9 +1891,43 @@ class API:
                 shutdown_trigger=ev.wait,
             )
 
+    async def _bootstrap_then_apply_state(self):
+        await self._fetch_snapshot()
+        await self._apply_state()
+
+    async def _fetch_snapshot(self) -> None:
+        receiver = SnapshotReceiver(self.node_id, self.session_id)
+        await self.command_sender.send(
+            ForwarderCommand(
+                origin=self._system_id,
+                command=RequestSnapshot(requester_node_id=self.node_id),
+            )
+        )
+
+        with anyio.move_on_after(_SNAPSHOT_FETCH_TIMEOUT_SECONDS):
+            with self.snapshot_chunk_receiver as chunks:
+                async for chunk in chunks:
+                    received = receiver.ingest(chunk)
+                    if received is None:
+                        continue
+                    self.state = received.state
+                    self.event_router.set_buffer_start(
+                        received.last_event_applied_idx + 1
+                    )
+                    logger.info(
+                        f"API bootstrapped from snapshot at idx "
+                        f"{received.last_event_applied_idx}"
+                    )
+                    return
+        logger.info(
+            "API: no snapshot received before timeout; falling back to full event-log replay"
+        )
+
     async def _apply_state(self):
         with self.event_receiver as events:
             async for i_event in events:
+                if i_event.idx <= self.state.last_event_applied_idx:
+                    continue
                 self._event_log.append(i_event.event)
                 self.state = apply(self.state, i_event)
                 event = i_event.event
@@ -1901,23 +1949,47 @@ class API:
                             await queue.send(event.chunk)
                         except (BrokenResourceError, ClosedResourceError):
                             self._text_generation_queues.pop(event.command_id, None)
-                if isinstance(event, InstanceDeleted):
-                    self._close_streams_for_instance(event.instance_id)
                 if isinstance(event, TracesMerged):
                     self._save_merged_trace(event)
 
-    def _close_streams_for_instance(self, instance_id: InstanceId) -> None:
-        """Close any active generation streams for commands running on the given instance."""
-        for task in self.state.tasks.values():
-            if task.instance_id != instance_id:
-                continue
-            if not isinstance(
+    async def _reconcile_streams(self) -> None:
+        while True:
+            await anyio.sleep(1)
+            self._reconcile_streams_once()
+
+    def _reconcile_streams_once(self) -> None:
+        generation_tasks = [
+            task
+            for task in self.state.tasks.values()
+            if isinstance(
                 task, (TextGenerationTask, ImageGenerationTask, ImageEditsTask)
-            ):
-                continue
-            if sender := self._text_generation_queues.pop(task.command_id, None):
+            )
+        ]
+        state_command_ids = {task.command_id for task in generation_tasks}
+        self._observed_generation_commands.update(state_command_ids)
+
+        live_command_ids = {
+            task.command_id
+            for task in generation_tasks
+            if task.instance_id in self.state.instances
+        }
+        queued_command_ids = set(self._text_generation_queues) | set(
+            self._image_generation_queues
+        )
+        stale_command_ids = (
+            self._observed_generation_commands - live_command_ids
+        ) & queued_command_ids
+        self._close_streams_for_commands(stale_command_ids)
+
+        self._observed_generation_commands = (
+            self._observed_generation_commands & queued_command_ids
+        ) | state_command_ids
+
+    def _close_streams_for_commands(self, command_ids: set[CommandId]) -> None:
+        for command_id in command_ids:
+            if sender := self._text_generation_queues.pop(command_id, None):
                 sender.close()
-            if sender := self._image_generation_queues.pop(task.command_id, None):
+            if sender := self._image_generation_queues.pop(command_id, None):
                 sender.close()
 
     def _save_merged_trace(self, event: TracesMerged) -> None:

src/exo/api/tests/test_api_snapshot_bootstrap.py

@@ -0,0 +1,136 @@
+# pyright: reportPrivateUsage=false
+
+import hashlib
+
+import anyio
+import pytest
+import zstandard
+
+from exo.api.main import API
+from exo.routing.event_router import EventRouter
+from exo.shared.types.commands import ForwarderCommand, RequestSnapshot
+from exo.shared.types.common import NodeId, SessionId, SystemId
+from exo.shared.types.events import (
+    Event,
+    GlobalForwarderEvent,
+    IndexedEvent,
+    LocalForwarderEvent,
+    TestEvent,
+)
+from exo.shared.types.snapshots import SnapshotChunk, SnapshotTransferId
+from exo.shared.types.state import State
+from exo.utils.channels import Receiver, Sender, channel
+
+
+class _FakeEventLog:
+    def __init__(self) -> None:
+        self.appended: list[Event] = []
+
+    def append(self, event: Event) -> None:
+        self.appended.append(event)
+
+
+def _snapshot_chunk(
+    state: State, *, requester_node_id: NodeId, session_id: SessionId
+) -> SnapshotChunk:
+    body = zstandard.ZstdCompressor().compress(state.model_dump_json().encode("utf-8"))
+    return SnapshotChunk.from_data(
+        data=body,
+        transfer_id=SnapshotTransferId("transfer-1"),
+        requester_node_id=requester_node_id,
+        session_id=session_id,
+        schema_version=state.schema_version,
+        last_event_applied_idx=state.last_event_applied_idx,
+        chunk_index=0,
+        total_chunks=1,
+        sha256_hex=hashlib.sha256(body).hexdigest(),
+    )
+
+
+def _api(
+    node_id: NodeId, session_id: SessionId
+) -> tuple[
+    API,
+    EventRouter,
+    Receiver[ForwarderCommand],
+    Sender[SnapshotChunk],
+    Sender[IndexedEvent],
+    _FakeEventLog,
+]:
+    router_command_sender, _router_command_receiver = channel[ForwarderCommand]()
+    _global_event_sender, global_event_receiver = channel[GlobalForwarderEvent]()
+    local_event_sender, _local_event_receiver = channel[LocalForwarderEvent]()
+    event_router = EventRouter(
+        session_id=session_id,
+        command_sender=router_command_sender,
+        external_inbound=global_event_receiver,
+        external_outbound=local_event_sender,
+    )
+
+    event_sender, event_receiver = channel[IndexedEvent]()
+    command_sender, command_receiver = channel[ForwarderCommand]()
+    snapshot_sender, snapshot_receiver = channel[SnapshotChunk]()
+
+    api = object.__new__(API)
+    api.node_id = node_id
+    api.session_id = session_id
+    api.event_router = event_router
+    api.event_receiver = event_receiver
+    api.snapshot_chunk_receiver = snapshot_receiver
+    api.command_sender = command_sender
+    api._system_id = SystemId("api-system")
+    api.state = State()
+    event_log = _FakeEventLog()
+    api._event_log = event_log  # pyright: ignore[reportAttributeAccessIssue]
+    api._image_generation_queues = {}
+    api._text_generation_queues = {}
+    return api, event_router, command_receiver, snapshot_sender, event_sender, event_log
+
+
+@pytest.mark.asyncio
+async def test_api_fetch_snapshot_applies_state_and_fast_forwards_router() -> None:
+    node_id = NodeId("api")
+    session_id = SessionId(master_node_id=NodeId("master"), election_clock=1)
+    api, event_router, command_receiver, snapshot_sender, _event_sender, _event_log = (
+        _api(node_id, session_id)
+    )
+    state = State(last_event_applied_idx=7)
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(api._fetch_snapshot)
+        command = await command_receiver.receive()
+        assert isinstance(command.command, RequestSnapshot)
+        assert command.command.requester_node_id == node_id
+
+        await snapshot_sender.send(
+            _snapshot_chunk(state, requester_node_id=node_id, session_id=session_id)
+        )
+
+    assert api.state.last_event_applied_idx == 7
+    assert event_router.event_buffer.next_idx_to_release == 8
+
+
+@pytest.mark.asyncio
+async def test_api_apply_state_ignores_events_covered_by_snapshot() -> None:
+    node_id = NodeId("api")
+    session_id = SessionId(master_node_id=NodeId("master"), election_clock=1)
+    (
+        api,
+        _event_router,
+        _command_receiver,
+        _snapshot_sender,
+        event_sender,
+        event_log,
+    ) = _api(node_id, session_id)
+    api.state = State(last_event_applied_idx=7)
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(api._apply_state)
+        await event_sender.send(IndexedEvent(idx=7, event=TestEvent()))
+        await event_sender.send(IndexedEvent(idx=8, event=TestEvent()))
+
+        while api.state.last_event_applied_idx != 8:
+            await anyio.sleep(0.001)
+        tg.cancel_scope.cancel()
+
+    assert len(event_log.appended) == 1

src/exo/api/tests/test_instance_deleted_stream_cleanup.py

@@ -1,11 +1,11 @@
 # pyright: reportUnusedFunction=false, reportAny=false
-"""Tests that InstanceDeleted events close active generation streams."""
+"""Tests that streaming queues reconcile against durable State."""
 
 from unittest.mock import MagicMock
 
 from exo.api.main import API
 from exo.api.types import ImageGenerationTaskParams
-from exo.shared.types.common import CommandId, ModelId
+from exo.shared.types.common import CommandId, ModelId, NodeId
 from exo.shared.types.state import State
 from exo.shared.types.tasks import ImageGeneration, TextGeneration
 from exo.shared.types.text_generation import (
@@ -13,15 +13,16 @@ from exo.shared.types.text_generation import (
     InputMessageContent,
     TextGenerationTaskParams,
 )
-from exo.shared.types.worker.instances import InstanceId
+from exo.shared.types.worker.instances import InstanceId, MlxRingInstance
+from exo.shared.types.worker.runners import ShardAssignments
 
 
 def _make_api_with_state(state: State) -> API:
-    """Create a minimal API instance with pre-set state."""
     api = object.__new__(API)
     api.state = state
     api._text_generation_queues = {}  # pyright: ignore[reportPrivateUsage]
     api._image_generation_queues = {}  # pyright: ignore[reportPrivateUsage]
+    api._observed_generation_commands = set()  # pyright: ignore[reportPrivateUsage]
     return api
 
 
@@ -38,45 +39,90 @@ def _make_text_gen_task(
     )
 
 
-def test_close_streams_for_deleted_instance() -> None:
-    """Deleting an instance closes the text generation sender for commands on that instance."""
+def _make_instance(instance_id: InstanceId) -> MlxRingInstance:
+    return MlxRingInstance(
+        instance_id=instance_id,
+        shard_assignments=ShardAssignments(
+            model_id=ModelId("test-model"),
+            node_to_runner={},
+            runner_to_shard={},
+        ),
+        hosts_by_node={NodeId("node-1"): []},
+        ephemeral_port=1,
+    )
+
+
+def test_reconcile_closes_stream_when_task_instance_is_missing() -> None:
     instance_id = InstanceId("inst-1")
     command_id = CommandId("cmd-1")
     task = _make_text_gen_task(instance_id, command_id)
-
-    state = State(tasks={task.task_id: task})
-    api = _make_api_with_state(state)
+    api = _make_api_with_state(State(tasks={task.task_id: task}, instances={}))
 
     sender = MagicMock()
     api._text_generation_queues[command_id] = sender  # pyright: ignore[reportPrivateUsage]
 
-    api._close_streams_for_instance(instance_id)  # pyright: ignore[reportPrivateUsage]
+    api._reconcile_streams_once()  # pyright: ignore[reportPrivateUsage]
 
     sender.close.assert_called_once()
     assert command_id not in api._text_generation_queues  # pyright: ignore[reportPrivateUsage]
 
 
-def test_close_streams_ignores_unrelated_instances() -> None:
-    """Deleting an instance does NOT close streams for commands on other instances."""
-    target_id = InstanceId("inst-delete")
-    other_id = InstanceId("inst-keep")
-    other_cmd = CommandId("cmd-keep")
-    other_task = _make_text_gen_task(other_id, other_cmd)
-
-    state = State(tasks={other_task.task_id: other_task})
-    api = _make_api_with_state(state)
+def test_reconcile_keeps_stream_for_live_task_instance() -> None:
+    instance_id = InstanceId("inst-live")
+    command_id = CommandId("cmd-live")
+    task = _make_text_gen_task(instance_id, command_id)
+    api = _make_api_with_state(
+        State(
+            tasks={task.task_id: task},
+            instances={instance_id: _make_instance(instance_id)},
+        )
+    )
 
     sender = MagicMock()
-    api._text_generation_queues[other_cmd] = sender  # pyright: ignore[reportPrivateUsage]
+    api._text_generation_queues[command_id] = sender  # pyright: ignore[reportPrivateUsage]
 
-    api._close_streams_for_instance(target_id)  # pyright: ignore[reportPrivateUsage]
+    api._reconcile_streams_once()  # pyright: ignore[reportPrivateUsage]
 
     sender.close.assert_not_called()
-    assert other_cmd in api._text_generation_queues  # pyright: ignore[reportPrivateUsage]
+    assert command_id in api._text_generation_queues  # pyright: ignore[reportPrivateUsage]
 
 
-def test_close_streams_for_deleted_instance_image_generation() -> None:
-    """Deleting an instance closes the image generation sender for commands on that instance."""
+def test_reconcile_does_not_close_command_before_state_observes_it() -> None:
+    command_id = CommandId("cmd-not-created-yet")
+    api = _make_api_with_state(State())
+
+    sender = MagicMock()
+    api._text_generation_queues[command_id] = sender  # pyright: ignore[reportPrivateUsage]
+
+    api._reconcile_streams_once()  # pyright: ignore[reportPrivateUsage]
+
+    sender.close.assert_not_called()
+    assert command_id in api._text_generation_queues  # pyright: ignore[reportPrivateUsage]
+
+
+def test_reconcile_closes_stream_after_observed_task_leaves_state() -> None:
+    instance_id = InstanceId("inst-live")
+    command_id = CommandId("cmd-deleted")
+    task = _make_text_gen_task(instance_id, command_id)
+    api = _make_api_with_state(
+        State(
+            tasks={task.task_id: task},
+            instances={instance_id: _make_instance(instance_id)},
+        )
+    )
+
+    sender = MagicMock()
+    api._text_generation_queues[command_id] = sender  # pyright: ignore[reportPrivateUsage]
+    api._reconcile_streams_once()  # pyright: ignore[reportPrivateUsage]
+
+    api.state = State(instances={instance_id: _make_instance(instance_id)})
+    api._reconcile_streams_once()  # pyright: ignore[reportPrivateUsage]
+
+    sender.close.assert_called_once()
+    assert command_id not in api._text_generation_queues  # pyright: ignore[reportPrivateUsage]
+
+
+def test_reconcile_closes_image_stream_when_task_instance_is_missing() -> None:
     instance_id = InstanceId("inst-img")
     command_id = CommandId("cmd-img")
     task = ImageGeneration(
@@ -84,14 +130,12 @@ def test_close_streams_for_deleted_instance_image_generation() -> None:
         command_id=command_id,
         task_params=ImageGenerationTaskParams(prompt="a cat", model="test-model"),
     )
-
-    state = State(tasks={task.task_id: task})
-    api = _make_api_with_state(state)
+    api = _make_api_with_state(State(tasks={task.task_id: task}, instances={}))
 
     sender = MagicMock()
     api._image_generation_queues[command_id] = sender  # pyright: ignore[reportPrivateUsage]
 
-    api._close_streams_for_instance(instance_id)  # pyright: ignore[reportPrivateUsage]
+    api._reconcile_streams_once()  # pyright: ignore[reportPrivateUsage]
 
     sender.close.assert_called_once()
     assert command_id not in api._image_generation_queues  # pyright: ignore[reportPrivateUsage]

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/main.py

@@ -59,6 +59,7 @@ class Node:
         await router.register_topic(topics.ELECTION_MESSAGES)
         await router.register_topic(topics.CONNECTION_MESSAGES)
         await router.register_topic(topics.DOWNLOAD_COMMANDS)
+        await router.register_topic(topics.SNAPSHOT_RESPONSES)
         event_router = EventRouter(
             session_id,
             command_sender=router.sender(topics.COMMANDS),
@@ -83,8 +84,11 @@ class Node:
         if args.spawn_api:
             api = API(
                 node_id,
+                session_id,
                 port=args.api_port,
+                event_router=event_router,
                 event_receiver=event_router.receiver(),
+                snapshot_chunk_receiver=router.receiver(topics.SNAPSHOT_RESPONSES),
                 command_sender=router.sender(topics.COMMANDS),
                 download_command_sender=router.sender(topics.DOWNLOAD_COMMANDS),
                 election_receiver=router.receiver(topics.ELECTION_MESSAGES),
@@ -95,8 +99,11 @@ class Node:
         if not args.no_worker:
             worker = Worker(
                 node_id,
+                session_id,
+                event_router=event_router,
                 event_receiver=event_router.receiver(),
                 event_sender=event_router.sender(),
+                snapshot_chunk_receiver=router.receiver(topics.SNAPSHOT_RESPONSES),
                 command_sender=router.sender(topics.COMMANDS),
                 download_command_sender=router.sender(topics.DOWNLOAD_COMMANDS),
                 api_port=args.api_port,
@@ -112,6 +119,7 @@ class Node:
             global_event_sender=router.sender(topics.GLOBAL_EVENTS),
             local_event_receiver=router.receiver(topics.LOCAL_EVENTS),
             command_receiver=router.receiver(topics.COMMANDS),
+            snapshot_chunk_sender=router.sender(topics.SNAPSHOT_RESPONSES),
             download_command_sender=router.sender(topics.DOWNLOAD_COMMANDS),
         )
 
@@ -210,6 +218,9 @@ class Node:
                         global_event_sender=self.router.sender(topics.GLOBAL_EVENTS),
                         local_event_receiver=self.router.receiver(topics.LOCAL_EVENTS),
                         command_receiver=self.router.receiver(topics.COMMANDS),
+                        snapshot_chunk_sender=self.router.sender(
+                            topics.SNAPSHOT_RESPONSES
+                        ),
                         download_command_sender=self.router.sender(
                             topics.DOWNLOAD_COMMANDS
                         ),
@@ -246,8 +257,13 @@ class Node:
                         # TODO: add profiling etc to resource monitor
                         self.worker = Worker(
                             self.node_id,
+                            result.session_id,
+                            event_router=self.event_router,
                             event_receiver=self.event_router.receiver(),
                             event_sender=self.event_router.sender(),
+                            snapshot_chunk_receiver=self.router.receiver(
+                                topics.SNAPSHOT_RESPONSES
+                            ),
                             command_sender=self.router.sender(topics.COMMANDS),
                             download_command_sender=self.router.sender(
                                 topics.DOWNLOAD_COMMANDS
@@ -256,7 +272,13 @@ class Node:
                         )
                         self._tg.start_soon(self.worker.run)
                     if self.api:
-                        self.api.reset(result.won_clock, self.event_router.receiver())
+                        self.api.reset(
+                            result.won_clock,
+                            result.session_id,
+                            self.event_router,
+                            self.event_router.receiver(),
+                            self.router.receiver(topics.SNAPSHOT_RESPONSES),
+                        )
                     self._tg.start_soon(self.event_router.run)
                 else:
                     if self.api:

src/exo/master/main.py

@@ -1,6 +1,8 @@
+import hashlib
 from datetime import datetime, timedelta, timezone
 
 import anyio
+from anyio import to_thread
 from loguru import logger
 
 from exo.master.placement import (
@@ -25,6 +27,7 @@ from exo.shared.types.commands import (
     ImageGeneration,
     PlaceInstance,
     RequestEventLog,
+    RequestSnapshot,
     SendInputChunk,
     SetInstanceLink,
     TaskCancelled,
@@ -54,6 +57,7 @@ from exo.shared.types.events import (
     TracesMerged,
 )
 from exo.shared.types.instance_link import InstanceLink
+from exo.shared.types.snapshots import SnapshotChunk, SnapshotTransferId
 from exo.shared.types.state import State
 from exo.shared.types.tasks import (
     ImageEdits as ImageEditsTask,
@@ -74,6 +78,15 @@ from exo.utils.disk_event_log import DiskEventLog
 from exo.utils.event_buffer import MultiSourceBuffer
 from exo.utils.task_group import TaskGroup
 
+_SNAPSHOT_CHUNK_BYTES = 512 * 1024
+_MAX_EVENT_LOG_REPLAY_BATCH = 1000
+
+
+def _encode_state_for_transfer(state: State) -> bytes:
+    import zstandard
+
+    return zstandard.ZstdCompressor().compress(state.model_dump_json().encode("utf-8"))
+
 
 def _prefill_endpoint_for(state: State, decode_instance_id: InstanceId) -> str | None:
     decode = state.instances.get(decode_instance_id)
@@ -125,6 +138,7 @@ class Master:
         event_sender: Sender[Event],
         local_event_receiver: Receiver[LocalForwarderEvent],
         global_event_sender: Sender[GlobalForwarderEvent],
+        snapshot_chunk_sender: Sender[SnapshotChunk],
         download_command_sender: Sender[ForwarderDownloadCommand],
     ):
         self.node_id = node_id
@@ -135,6 +149,7 @@ class Master:
         self.command_receiver = command_receiver
         self.local_event_receiver = local_event_receiver
         self.global_event_sender = global_event_sender
+        self.snapshot_chunk_sender = snapshot_chunk_sender
         self.download_command_sender = download_command_sender
         self.event_sender = event_sender
         self._system_id = SystemId()
@@ -155,6 +170,7 @@ class Master:
             self._event_log.close()
             self.global_event_sender.close()
             self.local_event_receiver.close()
+            self.snapshot_chunk_sender.close()
             self.command_receiver.close()
 
     async def shutdown(self):
@@ -441,12 +457,19 @@ class Master:
                         case RequestEventLog():
                             # We should just be able to send everything, since other buffers will ignore old messages
                             # rate limit to 1000 at a time
-                            end = min(command.since_idx + 1000, len(self._event_log))
+                            end = min(
+                                command.since_idx + _MAX_EVENT_LOG_REPLAY_BATCH,
+                                len(self._event_log),
+                            )
                             for i, event in enumerate(
                                 self._event_log.read_range(command.since_idx, end),
                                 start=command.since_idx,
                             ):
                                 await self._send_event(IndexedEvent(idx=i, event=event))
+                        case RequestSnapshot():
+                            self._tg.start_soon(
+                                self._serve_snapshot, command.requester_node_id
+                            )
                     for event in generated_events:
                         await self.event_sender.send(event)
                 except ValueError as e:
@@ -506,6 +529,42 @@ class Master:
                     self._event_log.append(event)
                     await self._send_event(indexed)
 
+    async def _serve_snapshot(self, requester_node_id: NodeId) -> None:
+        state = self.state
+        if state.last_event_applied_idx < 0:
+            logger.info(
+                f"RequestSnapshot from {requester_node_id} but master has no events yet"
+            )
+            return
+
+        body = await to_thread.run_sync(_encode_state_for_transfer, state)
+        sha256 = hashlib.sha256(body).hexdigest()
+        chunks = [
+            body[i : i + _SNAPSHOT_CHUNK_BYTES]
+            for i in range(0, len(body), _SNAPSHOT_CHUNK_BYTES)
+        ] or [b""]
+        transfer_id = SnapshotTransferId()
+
+        logger.info(
+            f"Serving snapshot to {requester_node_id}: "
+            f"idx={state.last_event_applied_idx}, "
+            f"{len(chunks)} chunk(s), {len(body)} bytes total"
+        )
+        for index, chunk in enumerate(chunks):
+            await self.snapshot_chunk_sender.send(
+                SnapshotChunk.from_data(
+                    data=chunk,
+                    transfer_id=transfer_id,
+                    requester_node_id=requester_node_id,
+                    session_id=self.session_id,
+                    schema_version=state.schema_version,
+                    last_event_applied_idx=state.last_event_applied_idx,
+                    chunk_index=index,
+                    total_chunks=len(chunks),
+                    sha256_hex=sha256,
+                )
+            )
+
     # This function is re-entrant, take care!
     async def _send_event(self, event: IndexedEvent):
         # Convenience method since this line is ugly

src/exo/master/tests/test_master.py

@@ -7,12 +7,14 @@ from loguru import logger
 
 from exo.master.main import Master
 from exo.routing.router import get_node_id_keypair
+from exo.routing.snapshot_receiver import SnapshotReceiver
 from exo.shared.models.model_cards import ModelCard, ModelTask
 from exo.shared.types.commands import (
     CommandId,
     ForwarderCommand,
     ForwarderDownloadCommand,
     PlaceInstance,
+    RequestSnapshot,
     TextGeneration,
 )
 from exo.shared.types.common import ModelId, NodeId, SessionId, SystemId
@@ -29,6 +31,7 @@ from exo.shared.types.memory import Memory
 from exo.shared.types.profiling import (
     MemoryUsage,
 )
+from exo.shared.types.snapshots import SnapshotChunk
 from exo.shared.types.tasks import TaskStatus
 from exo.shared.types.tasks import TextGeneration as TextGenerationTask
 from exo.shared.types.text_generation import (
@@ -56,6 +59,7 @@ async def test_master():
     local_event_sender, le_receiver = channel[LocalForwarderEvent]()
     fcds, _fcdr = channel[ForwarderDownloadCommand]()
     ev_send, ev_recv = channel[Event]()
+    snapshot_chunk_send, _snapshot_chunk_recv = channel[SnapshotChunk]()
 
     async def mock_event_router():
         idx = 0
@@ -92,6 +96,7 @@ async def test_master():
         global_event_sender=ge_sender,
         local_event_receiver=le_receiver,
         command_receiver=co_receiver,
+        snapshot_chunk_sender=snapshot_chunk_send,
         download_command_sender=fcds,
     )
     logger.info("run the master")
@@ -229,3 +234,52 @@ async def test_master():
 
         ev_send.close()
         await master.shutdown()
+
+
+@pytest.mark.asyncio
+async def test_master_serves_snapshot_for_current_state():
+    node_id = NodeId("master")
+    requester_node_id = NodeId("worker")
+    session_id = SessionId(master_node_id=node_id, election_clock=0)
+
+    ge_sender, _global_event_receiver = channel[GlobalForwarderEvent]()
+    command_sender, command_receiver = channel[ForwarderCommand]()
+    _local_event_sender, local_event_receiver = channel[LocalForwarderEvent]()
+    download_command_sender, _download_command_receiver = channel[
+        ForwarderDownloadCommand
+    ]()
+    event_sender, _event_receiver = channel[Event]()
+    snapshot_chunk_sender, snapshot_chunk_receiver = channel[SnapshotChunk]()
+
+    master = Master(
+        node_id,
+        session_id,
+        event_sender=event_sender,
+        global_event_sender=ge_sender,
+        local_event_receiver=local_event_receiver,
+        command_receiver=command_receiver,
+        snapshot_chunk_sender=snapshot_chunk_sender,
+        download_command_sender=download_command_sender,
+    )
+    master.state = master.state.model_copy(update={"last_event_applied_idx": 12})
+
+    receiver = SnapshotReceiver(requester_node_id, session_id)
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(master.run)
+        await command_sender.send(
+            ForwarderCommand(
+                origin=SystemId("api"),
+                command=RequestSnapshot(requester_node_id=requester_node_id),
+            )
+        )
+
+        received = None
+        while received is None:
+            chunk = await snapshot_chunk_receiver.receive()
+            received = receiver.ingest(chunk)
+
+        assert received.last_event_applied_idx == 12
+        assert received.state.last_event_applied_idx == 12
+
+        await master.shutdown()
+        tg.cancel_scope.cancel()

src/exo/routing/event_router.py

@@ -80,6 +80,9 @@ class EventRouter:
     def shutdown(self) -> None:
         self._tg.cancel_tasks()
 
+    def set_buffer_start(self, idx: int) -> None:
+        self.event_buffer.fast_forward_to(idx)
+
     async def _ingest(self, system_id: SystemId, recv: Receiver[Event]):
         idx = 0
         with recv as events:
@@ -95,7 +98,6 @@ class EventRouter:
                 self.out_for_delivery[event.event_id] = (anyio.current_time(), f_ev)
 
     async def _run_ext_in(self):
-        buf = OrderedBuffer[Event]()
         with self.external_inbound as events:
             async for event in events:
                 if event.session != self.session_id:
@@ -103,12 +105,12 @@ class EventRouter:
                 if event.origin != self.session_id.master_node_id:
                     continue
 
-                buf.ingest(event.origin_idx, event.event)
+                self.event_buffer.ingest(event.origin_idx, event.event)
                 event_id = event.event.event_id
                 if event_id in self.out_for_delivery:
                     self.out_for_delivery.pop(event_id)
 
-                drained = buf.drain_indexed()
+                drained = self.event_buffer.drain_indexed()
                 if drained:
                     self._nack_attempts = 0
                     if self._nack_cancel_scope:
@@ -119,7 +121,9 @@ class EventRouter:
                     or self._nack_cancel_scope.cancel_called
                 ):
                     # Request the next index.
-                    self._tg.start_soon(self._nack_request, buf.next_idx_to_release)
+                    self._tg.start_soon(
+                        self._nack_request, self.event_buffer.next_idx_to_release
+                    )
                     continue
 
                 for idx, event in drained:

src/exo/routing/snapshot_receiver.py

@@ -0,0 +1,109 @@
+"""Reassembles a snapshot from a stream of `SnapshotChunk`s.
+
+A receiver belongs to one node; it ignores chunks addressed to other
+requesters and chunks from prior sessions. Once a transfer's chunks have
+all been collected and the SHA-256 checks out, the snapshot is decoded into
+a `State`. Concurrent transfers (for the same requester) are tolerated:
+each is keyed by `transfer_id`.
+"""
+
+from __future__ import annotations
+
+import hashlib
+from dataclasses import dataclass, field
+from typing import final
+
+import zstandard
+from loguru import logger
+
+from exo.shared.types.common import NodeId, SessionId
+from exo.shared.types.snapshots import SnapshotChunk, SnapshotTransferId
+from exo.shared.types.state import State
+
+
+@final
+@dataclass
+class _Assembly:
+    """Partial state for one in-flight snapshot transfer."""
+
+    total_chunks: int
+    sha256_hex: str
+    schema_version: int
+    last_event_applied_idx: int
+    chunks: dict[int, bytes] = field(default_factory=dict)
+
+    def is_complete(self) -> bool:
+        return len(self.chunks) == self.total_chunks
+
+    def assemble(self) -> bytes:
+        return b"".join(self.chunks[i] for i in range(self.total_chunks))
+
+
+@dataclass
+class ReceivedSnapshot:
+    last_event_applied_idx: int
+    state: State
+
+
+class SnapshotReceiver:
+    """Filters and reassembles inbound chunks into a `ReceivedSnapshot`.
+
+    Stateless w.r.t. delivery: callers feed `SnapshotChunk`s in via `ingest`
+    and check the return value for completion.
+    """
+
+    def __init__(self, my_node_id: NodeId, session_id: SessionId) -> None:
+        self._my_node_id = my_node_id
+        self._session_id = session_id
+        self._assemblies: dict[SnapshotTransferId, _Assembly] = {}
+
+    def ingest(self, chunk: SnapshotChunk) -> ReceivedSnapshot | None:
+        """Absorb a chunk; return the snapshot once a transfer completes.
+
+        Returns None for partial transfers, mismatched recipients, stale
+        sessions, version mismatches, or corrupt payloads.
+        """
+        if chunk.requester_node_id != self._my_node_id:
+            return None
+        if chunk.session_id != self._session_id:
+            return None
+
+        existing = self._assemblies.get(chunk.transfer_id)
+        if existing is None:
+            existing = _Assembly(
+                total_chunks=chunk.total_chunks,
+                sha256_hex=chunk.sha256_hex,
+                schema_version=chunk.schema_version,
+                last_event_applied_idx=chunk.last_event_applied_idx,
+            )
+            self._assemblies[chunk.transfer_id] = existing
+        existing.chunks[chunk.chunk_index] = chunk.data
+
+        if not existing.is_complete():
+            return None
+
+        # Transfer complete — finalise and remove from the in-flight map.
+        del self._assemblies[chunk.transfer_id]
+        body = existing.assemble()
+        if hashlib.sha256(body).hexdigest() != existing.sha256_hex:
+            logger.warning(f"Snapshot {chunk.transfer_id} failed checksum; discarding")
+            return None
+        try:
+            decompressed = zstandard.ZstdDecompressor().decompress(body)
+            state = State.model_validate_json(decompressed.decode("utf-8"))
+        except (zstandard.ZstdError, ValueError) as e:
+            logger.opt(exception=e).warning(
+                f"Snapshot {chunk.transfer_id} could not be decoded; discarding"
+            )
+            return None
+        if state.schema_version != existing.schema_version:
+            # Should not happen — the master writes schema_version into both
+            # the chunk meta and the State payload — but treat it as corrupt.
+            logger.warning(
+                f"Snapshot {chunk.transfer_id} schema version mismatch "
+                f"(chunk={existing.schema_version}, state={state.schema_version})"
+            )
+            return None
+        return ReceivedSnapshot(
+            last_event_applied_idx=existing.last_event_applied_idx, state=state
+        )

src/exo/routing/tests/test_event_buffer.py

@@ -141,3 +141,28 @@ async def test_drain_and_ingest_with_new_sequence(buffer: OrderedBuffer[Event]):
     assert [e[0] for e in drained] == [2]
     assert buffer.next_idx_to_release == 3
     assert 4 in buffer.store
+
+
+@pytest.mark.asyncio
+async def test_fast_forward_discards_buffered_stale_events(
+    buffer: OrderedBuffer[Event],
+):
+    buffer.ingest(*make_indexed_event(0))
+    buffer.ingest(*make_indexed_event(2))
+    buffer.ingest(*make_indexed_event(4))
+
+    buffer.fast_forward_to(3)
+
+    assert buffer.next_idx_to_release == 3
+    assert set(buffer.store) == {4}
+
+
+@pytest.mark.asyncio
+async def test_fast_forward_only_moves_forward(buffer: OrderedBuffer[Event]):
+    buffer.ingest(*make_indexed_event(0))
+    buffer.ingest(*make_indexed_event(1))
+    buffer.drain()
+
+    buffer.fast_forward_to(1)
+
+    assert buffer.next_idx_to_release == 2

src/exo/routing/tests/test_snapshot_receiver.py

@@ -0,0 +1,151 @@
+import hashlib
+
+import pytest
+import zstandard
+
+from exo.routing.snapshot_receiver import SnapshotReceiver
+from exo.shared.types.common import NodeId, SessionId
+from exo.shared.types.snapshots import SnapshotChunk, SnapshotTransferId
+from exo.shared.types.state import State
+
+
+@pytest.fixture
+def session_id() -> SessionId:
+    return SessionId(master_node_id=NodeId("master"), election_clock=0)
+
+
+@pytest.fixture
+def my_node() -> NodeId:
+    return NodeId("worker-1")
+
+
+def _encode(state: State) -> bytes:
+    return zstandard.ZstdCompressor().compress(state.model_dump_json().encode("utf-8"))
+
+
+def _make_chunks(
+    body: bytes,
+    *,
+    chunk_size: int,
+    requester_node_id: NodeId,
+    session_id: SessionId,
+    state: State,
+    transfer_id: SnapshotTransferId | None = None,
+) -> list[SnapshotChunk]:
+    sha256 = hashlib.sha256(body).hexdigest()
+    transfer_id = transfer_id or SnapshotTransferId()
+    pieces = [body[i : i + chunk_size] for i in range(0, len(body), chunk_size)] or [
+        b""
+    ]
+    return [
+        SnapshotChunk.from_data(
+            data=piece,
+            transfer_id=transfer_id,
+            requester_node_id=requester_node_id,
+            session_id=session_id,
+            schema_version=state.schema_version,
+            last_event_applied_idx=state.last_event_applied_idx,
+            chunk_index=i,
+            total_chunks=len(pieces),
+            sha256_hex=sha256,
+        )
+        for i, piece in enumerate(pieces)
+    ]
+
+
+def test_completes_on_full_transfer(my_node: NodeId, session_id: SessionId):
+    state = State(last_event_applied_idx=42)
+    chunks = _make_chunks(
+        _encode(state),
+        chunk_size=64,
+        requester_node_id=my_node,
+        session_id=session_id,
+        state=state,
+    )
+    receiver = SnapshotReceiver(my_node, session_id)
+
+    received = None
+    for chunk in chunks:
+        received = receiver.ingest(chunk)
+    assert received is not None
+    assert received.last_event_applied_idx == 42
+    assert received.state.last_event_applied_idx == 42
+
+
+def test_handles_out_of_order_chunks(my_node: NodeId, session_id: SessionId):
+    state = State(last_event_applied_idx=99)
+    chunks = _make_chunks(
+        _encode(state),
+        chunk_size=32,
+        requester_node_id=my_node,
+        session_id=session_id,
+        state=state,
+    )
+    receiver = SnapshotReceiver(my_node, session_id)
+
+    # Reverse them.
+    received = None
+    for chunk in reversed(chunks):
+        received = receiver.ingest(chunk)
+    assert received is not None
+    assert received.last_event_applied_idx == 99
+
+
+def test_ignores_chunks_for_other_recipients(my_node: NodeId, session_id: SessionId):
+    state = State(last_event_applied_idx=1)
+    other = NodeId("worker-2")
+    chunks = _make_chunks(
+        _encode(state),
+        chunk_size=64,
+        requester_node_id=other,
+        session_id=session_id,
+        state=state,
+    )
+    receiver = SnapshotReceiver(my_node, session_id)
+    for chunk in chunks:
+        assert receiver.ingest(chunk) is None
+
+
+def test_ignores_chunks_from_stale_session(my_node: NodeId, session_id: SessionId):
+    state = State(last_event_applied_idx=1)
+    other_session = SessionId(master_node_id=NodeId("other-master"), election_clock=99)
+    chunks = _make_chunks(
+        _encode(state),
+        chunk_size=64,
+        requester_node_id=my_node,
+        session_id=other_session,
+        state=state,
+    )
+    receiver = SnapshotReceiver(my_node, session_id)
+    for chunk in chunks:
+        assert receiver.ingest(chunk) is None
+
+
+def test_discards_on_checksum_mismatch(my_node: NodeId, session_id: SessionId):
+    state = State(last_event_applied_idx=1)
+    chunks = _make_chunks(
+        _encode(state),
+        chunk_size=64,
+        requester_node_id=my_node,
+        session_id=session_id,
+        state=state,
+    )
+    # Corrupt the last byte of the last chunk.
+    original = chunks[-1]
+    chunks[-1] = SnapshotChunk.from_data(
+        data=original.data + b"\x00garbage",
+        transfer_id=original.transfer_id,
+        requester_node_id=original.requester_node_id,
+        session_id=original.session_id,
+        schema_version=original.schema_version,
+        last_event_applied_idx=original.last_event_applied_idx,
+        chunk_index=original.chunk_index,
+        total_chunks=original.total_chunks,
+        sha256_hex=original.sha256_hex,
+    )
+
+    receiver = SnapshotReceiver(my_node, session_id)
+    received = None
+    for chunk in chunks:
+        received = receiver.ingest(chunk)
+    assert received is None

src/exo/routing/tests/test_snapshot_routing_types.py

@@ -0,0 +1,37 @@
+from exo.routing import topics
+from exo.shared.types.commands import ForwarderCommand, RequestSnapshot
+from exo.shared.types.common import NodeId, SessionId, SystemId
+from exo.shared.types.snapshots import SnapshotChunk, SnapshotTransferId
+
+
+def test_request_snapshot_round_trips_through_forwarder_command() -> None:
+    command = ForwarderCommand(
+        origin=SystemId("system-1"),
+        command=RequestSnapshot(requester_node_id=NodeId("worker-1")),
+    )
+
+    restored = ForwarderCommand.model_validate_json(command.model_dump_json())
+
+    assert isinstance(restored.command, RequestSnapshot)
+    assert restored.command.requester_node_id == NodeId("worker-1")
+
+
+def test_snapshot_response_topic_round_trips_chunk() -> None:
+    chunk = SnapshotChunk.from_data(
+        data=b"snapshot-bytes",
+        transfer_id=SnapshotTransferId("transfer-1"),
+        requester_node_id=NodeId("worker-1"),
+        session_id=SessionId(master_node_id=NodeId("master"), election_clock=1),
+        schema_version=1,
+        last_event_applied_idx=42,
+        chunk_index=0,
+        total_chunks=1,
+        sha256_hex="unused",
+    )
+
+    restored = topics.SNAPSHOT_RESPONSES.deserialize(
+        topics.SNAPSHOT_RESPONSES.serialize(chunk)
+    )
+
+    assert restored == chunk
+    assert restored.data == b"snapshot-bytes"

src/exo/routing/tests/test_transient_router.py

@@ -0,0 +1,69 @@
+import anyio
+import pytest
+
+from exo.routing import topics
+from exo.routing.transient_router import TransientRouter
+from exo.shared.types.common import NodeId, SessionId
+from exo.shared.types.events import (
+    GlobalForwarderTransientEvent,
+    TaskAcknowledged,
+)
+from exo.shared.types.tasks import TaskId
+from exo.utils.channels import channel
+
+
+def test_transient_topic_round_trips_wrapped_event() -> None:
+    wrapped = GlobalForwarderTransientEvent(
+        origin=NodeId("worker"),
+        session=SessionId(master_node_id=NodeId("master"), election_clock=1),
+        event=TaskAcknowledged(task_id=TaskId("task-1")),
+    )
+
+    restored = topics.TRANSIENT_EVENTS.deserialize(
+        topics.TRANSIENT_EVENTS.serialize(wrapped)
+    )
+
+    assert restored == wrapped
+
+
+@pytest.mark.asyncio
+async def test_transient_router_publishes_and_dispatches_session_events() -> None:
+    node_id = NodeId("worker")
+    session_id = SessionId(master_node_id=NodeId("master"), election_clock=1)
+    external_outbound_sender, external_outbound_receiver = channel[
+        GlobalForwarderTransientEvent
+    ]()
+    external_inbound_sender, external_inbound_receiver = channel[
+        GlobalForwarderTransientEvent
+    ]()
+    router = TransientRouter(
+        node_id=node_id,
+        session_id=session_id,
+        external_outbound=external_outbound_sender,
+        external_inbound=external_inbound_receiver,
+    )
+    local_sender = router.sender()
+    local_receiver = router.receiver()
+    event = TaskAcknowledged(task_id=TaskId("task-1"))
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(router.run)
+        await local_sender.send(event)
+
+        wrapped = await external_outbound_receiver.receive()
+        assert wrapped.origin == node_id
+        assert wrapped.session == session_id
+        assert wrapped.event == event
+
+        await external_inbound_sender.send(wrapped)
+        assert await local_receiver.receive() == event
+
+        stale_session = SessionId(master_node_id=NodeId("master"), election_clock=2)
+        await external_inbound_sender.send(
+            wrapped.model_copy(update={"session": stale_session})
+        )
+        await anyio.sleep(0)
+        assert local_receiver.collect() == []
+
+        router.shutdown()
+        tg.cancel_scope.cancel()

src/exo/routing/topics.py

@@ -6,8 +6,10 @@ from exo.shared.election import ElectionMessage
 from exo.shared.types.commands import ForwarderCommand, ForwarderDownloadCommand
 from exo.shared.types.events import (
     GlobalForwarderEvent,
+    GlobalForwarderTransientEvent,
     LocalForwarderEvent,
 )
+from exo.shared.types.snapshots import SnapshotChunk
 from exo.utils.pydantic_ext import FrozenModel
 
 
@@ -39,6 +41,9 @@ class TypedTopic[T: FrozenModel]:
 
 GLOBAL_EVENTS = TypedTopic("global_events", PublishPolicy.Always, GlobalForwarderEvent)
 LOCAL_EVENTS = TypedTopic("local_events", PublishPolicy.Always, LocalForwarderEvent)
+TRANSIENT_EVENTS = TypedTopic(
+    "transient_events", PublishPolicy.Always, GlobalForwarderTransientEvent
+)
 COMMANDS = TypedTopic("commands", PublishPolicy.Always, ForwarderCommand)
 ELECTION_MESSAGES = TypedTopic(
     "election_messages", PublishPolicy.Always, ElectionMessage
@@ -49,3 +54,6 @@ CONNECTION_MESSAGES = TypedTopic(
 DOWNLOAD_COMMANDS = TypedTopic(
     "download_commands", PublishPolicy.Always, ForwarderDownloadCommand
 )
+SNAPSHOT_RESPONSES = TypedTopic(
+    "snapshot_responses", PublishPolicy.Always, SnapshotChunk
+)

src/exo/routing/transient_router.py

@@ -0,0 +1,84 @@
+from dataclasses import dataclass, field
+
+from anyio import BrokenResourceError, ClosedResourceError
+from loguru import logger
+
+from exo.shared.types.common import NodeId, SessionId
+from exo.shared.types.events import (
+    GlobalForwarderTransientEvent,
+    TransientEvent,
+)
+from exo.utils.channels import Receiver, Sender, channel
+from exo.utils.task_group import TaskGroup
+
+
+@dataclass
+class TransientRouter:
+    """Routes unordered, non-durable events over the transient-events topic."""
+
+    node_id: NodeId
+    session_id: SessionId
+    external_outbound: Sender[GlobalForwarderTransientEvent]
+    external_inbound: Receiver[GlobalForwarderTransientEvent]
+    _outbound: list[Sender[TransientEvent]] = field(init=False, default_factory=list)
+    _inbound: list[Receiver[TransientEvent]] = field(init=False, default_factory=list)
+    _tg: TaskGroup = field(init=False, default_factory=TaskGroup)
+
+    def sender(self) -> Sender[TransientEvent]:
+        send, recv = channel[TransientEvent]()
+        if self._tg.is_running():
+            self._tg.start_soon(self._publish, recv)
+        else:
+            self._inbound.append(recv)
+        return send
+
+    def receiver(self) -> Receiver[TransientEvent]:
+        assert not self._tg.is_running(), (
+            "TransientRouter receivers must be registered before run()"
+        )
+        send, recv = channel[TransientEvent]()
+        self._outbound.append(send)
+        return recv
+
+    def shutdown(self) -> None:
+        self._tg.cancel_tasks()
+
+    async def run(self) -> None:
+        try:
+            async with self._tg as tg:
+                for recv in self._inbound:
+                    tg.start_soon(self._publish, recv)
+                tg.start_soon(self._dispatch_inbound)
+        finally:
+            self.external_outbound.close()
+            for send in self._outbound:
+                send.close()
+
+    async def _publish(self, recv: Receiver[TransientEvent]) -> None:
+        with recv as events:
+            async for event in events:
+                await self.external_outbound.send(
+                    GlobalForwarderTransientEvent(
+                        origin=self.node_id,
+                        session=self.session_id,
+                        event=event,
+                    )
+                )
+
+    async def _dispatch_inbound(self) -> None:
+        with self.external_inbound as wrapped_events:
+            async for wrapped in wrapped_events:
+                if wrapped.session != self.session_id:
+                    continue
+                stale: set[int] = set()
+                for index, send in enumerate(self._outbound):
+                    try:
+                        await send.send(wrapped.event)
+                    except (ClosedResourceError, BrokenResourceError):
+                        stale.add(index)
+                if stale:
+                    for index in sorted(stale, reverse=True):
+                        self._outbound.pop(index)
+                    logger.debug(
+                        f"TransientRouter dropped {len(stale)} closed receivers"
+                    )

src/exo/shared/apply.py

@@ -4,7 +4,8 @@ from datetime import datetime
 
 from loguru import logger
 
-from exo.shared.types.common import NodeId
+from exo.shared.models.model_cards import ModelCard
+from exo.shared.types.common import ModelId, NodeId
 from exo.shared.types.events import (
     ChunkGenerated,
     CustomModelCardAdded,
@@ -40,7 +41,14 @@ from exo.shared.types.profiling import (
     ThunderboltBridgeStatus,
 )
 from exo.shared.types.state import State
-from exo.shared.types.tasks import Task, TaskId, TaskStatus
+from exo.shared.types.tasks import (
+    ImageEdits,
+    ImageGeneration,
+    Task,
+    TaskId,
+    TaskStatus,
+    TextGeneration,
+)
 from exo.shared.types.topology import Connection, RDMAConnection
 from exo.shared.types.worker.downloads import DownloadProgress
 from exo.shared.types.worker.instances import Instance, InstanceId
@@ -72,13 +80,14 @@ def event_apply(event: Event, state: State) -> State:
             TestEvent()
             | ChunkGenerated()
             | TaskAcknowledged()
-            | InputChunkReceived()
             | TracesCollected()
             | TracesMerged()
-            | CustomModelCardAdded()
-            | CustomModelCardDeleted()
         ):  # Pass-through events that don't modify state
             return state
+        case CustomModelCardAdded():
+            return apply_custom_model_card_added(event, state)
+        case CustomModelCardDeleted():
+            return apply_custom_model_card_deleted(event, state)
         case InstanceCreated():
             return apply_instance_created(event, state)
         case InstanceDeleted():
@@ -93,6 +102,8 @@ def event_apply(event: Event, state: State) -> State:
             return apply_runner_status_updated(event, state)
         case TaskCreated():
             return apply_task_created(event, state)
+        case InputChunkReceived():
+            return apply_input_chunk_received(event, state)
         case TaskDeleted():
             return apply_task_deleted(event, state)
         case TaskFailed():
@@ -157,10 +168,32 @@ def apply_task_created(event: TaskCreated, state: State) -> State:
     return state.model_copy(update={"tasks": new_tasks})
 
 
+def apply_input_chunk_received(event: InputChunkReceived, state: State) -> State:
+    command_chunks = {
+        **state.input_chunks.get(event.command_id, {}),
+        event.chunk.chunk_index: event.chunk,
+    }
+    return state.model_copy(
+        update={
+            "input_chunks": {**state.input_chunks, event.command_id: command_chunks}
+        }
+    )
+
+
 def apply_task_deleted(event: TaskDeleted, state: State) -> State:
+    task = state.tasks.get(event.task_id)
     new_tasks: Mapping[TaskId, Task] = {
         tid: task for tid, task in state.tasks.items() if tid != event.task_id
     }
+    if isinstance(task, (TextGeneration, ImageGeneration, ImageEdits)):
+        new_input_chunks = {
+            command_id: chunks
+            for command_id, chunks in state.input_chunks.items()
+            if command_id != task.command_id
+        }
+        return state.model_copy(
+            update={"tasks": new_tasks, "input_chunks": new_input_chunks}
+        )
     return state.model_copy(update={"tasks": new_tasks})
 
 
@@ -447,3 +480,22 @@ def apply_topology_edge_deleted(event: TopologyEdgeDeleted, state: State) -> Sta
     topology.remove_connection(event.conn)
     # TODO: Clean up removing the reverse connection
     return state.model_copy(update={"topology": topology})
+
+
+def apply_custom_model_card_added(event: CustomModelCardAdded, state: State) -> State:
+    new_cards: Mapping[ModelId, ModelCard] = {
+        **state.custom_model_cards,
+        event.model_card.model_id: event.model_card,
+    }
+    return state.model_copy(update={"custom_model_cards": new_cards})
+
+
+def apply_custom_model_card_deleted(
+    event: CustomModelCardDeleted, state: State
+) -> State:
+    new_cards: Mapping[ModelId, ModelCard] = {
+        model_id: card
+        for model_id, card in state.custom_model_cards.items()
+        if model_id != event.model_id
+    }
+    return state.model_copy(update={"custom_model_cards": new_cards})

src/exo/shared/tests/test_apply/test_apply_custom_model_cards.py

@@ -0,0 +1,44 @@
+from exo.shared.apply import apply
+from exo.shared.models.model_cards import ModelCard, ModelTask
+from exo.shared.types.common import ModelId
+from exo.shared.types.events import (
+    CustomModelCardAdded,
+    CustomModelCardDeleted,
+    IndexedEvent,
+)
+from exo.shared.types.memory import Memory
+from exo.shared.types.state import State
+
+
+def _model_card(model_id: ModelId) -> ModelCard:
+    return ModelCard(
+        model_id=model_id,
+        n_layers=1,
+        storage_size=Memory.from_bytes(1),
+        hidden_size=1,
+        supports_tensor=True,
+        tasks=[ModelTask.TextGeneration],
+    )
+
+
+def test_custom_model_card_added_is_reduced_into_state() -> None:
+    card = _model_card(ModelId("custom/model"))
+
+    state = apply(
+        State(),
+        IndexedEvent(idx=0, event=CustomModelCardAdded(model_card=card)),
+    )
+
+    assert state.custom_model_cards == {card.model_id: card}
+
+
+def test_custom_model_card_deleted_removes_card_from_state() -> None:
+    card = _model_card(ModelId("custom/model"))
+    state = State(custom_model_cards={card.model_id: card}, last_event_applied_idx=0)
+
+    state = apply(
+        state,
+        IndexedEvent(idx=1, event=CustomModelCardDeleted(model_id=card.model_id)),
+    )
+
+    assert state.custom_model_cards == {}

src/exo/shared/tests/test_apply/test_apply_input_chunks.py

@@ -0,0 +1,85 @@
+from exo.shared.apply import apply
+from exo.shared.models.model_cards import ModelId
+from exo.shared.types.chunks import InputImageChunk
+from exo.shared.types.common import CommandId
+from exo.shared.types.events import (
+    IndexedEvent,
+    InputChunkReceived,
+    TaskCreated,
+    TaskDeleted,
+)
+from exo.shared.types.state import State
+from exo.shared.types.tasks import TaskId, TaskStatus, TextGeneration
+from exo.shared.types.text_generation import (
+    InputMessage,
+    InputMessageContent,
+    TextGenerationTaskParams,
+)
+from exo.shared.types.worker.instances import InstanceId
+
+
+def test_apply_input_chunk_received_stores_chunk_in_state() -> None:
+    command_id = CommandId("command-1")
+    chunk = InputImageChunk(
+        model=ModelId("mlx-community/test-model"),
+        command_id=command_id,
+        data="abc",
+        chunk_index=0,
+        total_chunks=1,
+        image_index=0,
+    )
+
+    state = apply(
+        State(),
+        IndexedEvent(
+            idx=0,
+            event=InputChunkReceived(command_id=command_id, chunk=chunk),
+        ),
+    )
+
+    assert state.input_chunks == {command_id: {0: chunk}}
+
+
+def test_apply_task_deleted_removes_chunks_for_generation_command() -> None:
+    command_id = CommandId("command-1")
+    task_id = TaskId("task-1")
+    chunk = InputImageChunk(
+        model=ModelId("mlx-community/test-model"),
+        command_id=command_id,
+        data="abc",
+        chunk_index=0,
+        total_chunks=1,
+        image_index=0,
+    )
+    task = TextGeneration(
+        task_id=task_id,
+        instance_id=InstanceId("instance-1"),
+        task_status=TaskStatus.Pending,
+        command_id=command_id,
+        task_params=TextGenerationTaskParams(
+            model=ModelId("mlx-community/test-model"),
+            input=[
+                InputMessage(role="user", content=InputMessageContent("hello")),
+            ],
+        ),
+    )
+
+    state = State()
+    state = apply(
+        state,
+        IndexedEvent(
+            idx=0,
+            event=InputChunkReceived(command_id=command_id, chunk=chunk),
+        ),
+    )
+    state = apply(
+        state,
+        IndexedEvent(idx=1, event=TaskCreated(task_id=task_id, task=task)),
+    )
+    state = apply(
+        state,
+        IndexedEvent(idx=2, event=TaskDeleted(task_id=task_id)),
+    )
+
+    assert state.tasks == {}
+    assert state.input_chunks == {}

src/exo/shared/tests/test_state_serialization.py

@@ -25,6 +25,7 @@ def test_state_serialization_roundtrip() -> None:
     json_repr = state.model_dump_json()
     restored_state = State.model_validate_json(json_repr)
 
+    assert restored_state.schema_version == state.schema_version
     assert (
         state.topology.to_snapshot().nodes
         == restored_state.topology.to_snapshot().nodes

src/exo/shared/types/commands.py

@@ -67,6 +67,12 @@ class RequestEventLog(BaseCommand):
     since_idx: int
 
 
+class RequestSnapshot(BaseCommand):
+    """Ask the current master to send a State snapshot to this node."""
+
+    requester_node_id: NodeId
+
+
 class StartDownload(BaseCommand):
     target_node_id: NodeId
     shard_metadata: ShardMetadata
@@ -106,6 +112,7 @@ DownloadCommand = StartDownload | DeleteDownload | CancelDownload
 Command = (
     TestCommand
     | RequestEventLog
+    | RequestSnapshot
     | TextGeneration
     | ImageGeneration
     | ImageEdits

src/exo/shared/types/events.py

@@ -172,6 +172,9 @@ Event = (
 )
 
 
+TransientEvent = TaskAcknowledged | ChunkGenerated | TracesCollected | TracesMerged
+
+
 class IndexedEvent(FrozenModel):
     """An event indexed by the master, with a globally unique index"""
 
@@ -195,3 +198,11 @@ class LocalForwarderEvent(FrozenModel):
     origin: SystemId
     session: SessionId
     event: Event
+
+
+class GlobalForwarderTransientEvent(FrozenModel):
+    """An unordered, non-durable event published to the cluster."""
+
+    origin: NodeId
+    session: SessionId
+    event: TransientEvent

src/exo/shared/types/snapshots.py

@@ -0,0 +1,54 @@
+"""Wire types for snapshot transfer between master and a joining node.
+
+Snapshots can be tens of MB; the gossipsub message ceiling is around 1 MB.
+We slice the compressed snapshot body into chunks and publish each chunk on
+the SNAPSHOT_RESPONSES topic. The receiver collects chunks for its own
+`requester_node_id`, validates the SHA-256 of the reassembled body, and
+materialises the State.
+"""
+
+import base64
+
+from exo.shared.types.common import Id, NodeId, SessionId
+from exo.utils.pydantic_ext import FrozenModel
+
+
+class SnapshotTransferId(Id):
+    """Identifies a single snapshot transfer (one master response to one
+    `RequestSnapshot`). Distinct transfers may interleave; the id lets
+    receivers keep them apart."""
+
+
+class SnapshotChunk(FrozenModel):
+    """One slice of a snapshot in flight.
+
+    `data_b64` carries a base64-encoded slice of the zstd-compressed JSON
+    dump of State. Concatenating the *decoded* bytes of all chunks for a
+    `transfer_id` in order of `chunk_index` yields the full compressed
+    body; `sha256_hex` is the SHA-256 of that decoded blob.
+
+    We use base64 explicitly because the topic layer JSON-encodes messages,
+    and JSON can't carry raw binary. Helpers `from_data` / `data` keep the
+    base64 detail at the boundaries.
+    """
+
+    transfer_id: SnapshotTransferId
+    requester_node_id: NodeId
+    session_id: SessionId
+    schema_version: int
+    last_event_applied_idx: int
+    chunk_index: int
+    total_chunks: int
+    sha256_hex: str
+    data_b64: str
+
+    @classmethod
+    def from_data(cls, *, data: bytes, **kwargs: object) -> "SnapshotChunk":
+        return cls(data_b64=base64.b64encode(data).decode("ascii"), **kwargs)  # pyright: ignore[reportArgumentType]
+
+    @property
+    def data(self) -> bytes:
+        return base64.b64decode(self.data_b64)
+
+
+__all__ = ["SnapshotChunk", "SnapshotTransferId"]

src/exo/shared/types/state.py

@@ -5,8 +5,10 @@ from typing import Any, cast
 from pydantic import ConfigDict, Field, field_serializer, field_validator
 from pydantic.alias_generators import to_camel
 
+from exo.shared.models.model_cards import ModelCard
 from exo.shared.topology import Topology, TopologySnapshot
-from exo.shared.types.common import NodeId
+from exo.shared.types.chunks import InputImageChunk
+from exo.shared.types.common import CommandId, ModelId, NodeId
 from exo.shared.types.instance_link import InstanceLink, InstanceLinkId
 from exo.shared.types.profiling import (
     DiskUsage,
@@ -41,10 +43,16 @@ class State(FrozenModel):
         strict=True,
         arbitrary_types_allowed=True,
     )
+    # Bump when a State change makes older snapshots unsafe to restore.
+    schema_version: int = Field(default=1, ge=1)
+
     instances: Mapping[InstanceId, Instance] = {}
     runners: Mapping[RunnerId, RunnerStatus] = {}
     downloads: Mapping[NodeId, Sequence[DownloadProgress]] = {}
     tasks: Mapping[TaskId, Task] = {}
+    # Durable request input chunks for active image requests. Workers rebuild
+    # local image caches from this state instead of reading events directly.
+    input_chunks: Mapping[CommandId, Mapping[int, InputImageChunk]] = {}
     last_seen: Mapping[NodeId, datetime] = {}
     topology: Topology = Field(default_factory=Topology)
     last_event_applied_idx: int = Field(default=-1, ge=-1)
@@ -65,6 +73,10 @@ class State(FrozenModel):
     instance_links: Mapping[InstanceLinkId, InstanceLink] = {}
     prefill_server_ports: Mapping[RunnerId, int] = {}
 
+    # User-added model cards. Workers can reconcile their on-disk custom card
+    # cache from this state after snapshot bootstrap.
+    custom_model_cards: Mapping[ModelId, ModelCard] = {}
+
     @field_serializer("topology", mode="plain")
     def _encode_topology(self, value: Topology) -> TopologySnapshot:
         return value.to_snapshot()

src/exo/utils/event_buffer.py

@@ -47,6 +47,18 @@ class OrderedBuffer[T]:
         logger.trace(f"Releasing event {ret}")
         return ret
 
+    def fast_forward_to(self, idx: int) -> None:
+        """Skip every event before idx.
+
+        Snapshot restore uses this after applying state that already includes
+        events before idx. Any buffered or future event below idx is stale.
+        """
+        if idx <= self.next_idx_to_release:
+            return
+        self.next_idx_to_release = idx
+        for stale_idx in [i for i in self.store if i < idx]:
+            del self.store[stale_idx]
+
 
 class MultiSourceBuffer[SourceId, T]:
     """

src/exo/utils/keyed_backoff.py

@@ -29,6 +29,10 @@ class KeyedBackoff[K]:
         """Return the number of recorded attempts for a key."""
         return self._attempts.get(key, 0)
 
+    def tracked_keys(self) -> set[K]:
+        """Return keys that currently have recorded backoff state."""
+        return set(self._attempts) | set(self._last_time)
+
     def reset(self, key: K) -> None:
         """Reset backoff state for a key (e.g., on success)."""
         self._attempts.pop(key, None)

src/exo/utils/tests/test_keyed_backoff.py

@@ -0,0 +1,13 @@
+from exo.utils.keyed_backoff import KeyedBackoff
+
+
+def test_tracked_keys_reports_and_resets_backoff_state() -> None:
+    backoff = KeyedBackoff[str]()
+
+    backoff.record_attempt("instance-a")
+
+    assert backoff.tracked_keys() == {"instance-a"}
+
+    backoff.reset("instance-a")
+
+    assert backoff.tracked_keys() == set()

src/exo/worker/main.py

@@ -8,24 +8,28 @@ from loguru import logger
 
 from exo.api.types import ImageEditsTaskParams
 from exo.download.download_utils import is_read_only_model_dir, resolve_existing_model
+from exo.routing.event_router import EventRouter
+from exo.routing.snapshot_receiver import SnapshotReceiver
 from exo.shared.apply import apply
 from exo.shared.constants import EXO_MAX_INSTANCE_RETRIES
-from exo.shared.models.model_cards import ModelId, add_to_card_cache, delete_custom_card
+from exo.shared.models.model_cards import (
+    ModelCard,
+    ModelId,
+    add_to_card_cache,
+    delete_custom_card,
+)
 from exo.shared.types.chunks import InputImageChunk
 from exo.shared.types.commands import (
     DeleteInstance,
     ForwarderCommand,
     ForwarderDownloadCommand,
+    RequestSnapshot,
     StartDownload,
 )
-from exo.shared.types.common import CommandId, NodeId, SystemId
+from exo.shared.types.common import CommandId, NodeId, SessionId, SystemId
 from exo.shared.types.events import (
-    CustomModelCardAdded,
-    CustomModelCardDeleted,
     Event,
     IndexedEvent,
-    InputChunkReceived,
-    InstanceDeleted,
     NodeDownloadProgress,
     NodeGatheredInfo,
     TaskCreated,
@@ -34,6 +38,7 @@ from exo.shared.types.events import (
     TopologyEdgeDeleted,
 )
 from exo.shared.types.multiaddr import Multiaddr
+from exo.shared.types.snapshots import SnapshotChunk
 from exo.shared.types.state import State
 from exo.shared.types.tasks import (
     CancelTask,
@@ -59,14 +64,19 @@ from exo.utils.task_group import TaskGroup
 from exo.worker.plan import plan
 from exo.worker.runner.supervisor import RunnerSupervisor
 
+_SNAPSHOT_FETCH_TIMEOUT_SECONDS = 30
+
 
 class Worker:
     def __init__(
         self,
         node_id: NodeId,
+        session_id: SessionId,
         *,
+        event_router: EventRouter,
         event_receiver: Receiver[IndexedEvent],
         event_sender: Sender[Event],
+        snapshot_chunk_receiver: Receiver[SnapshotChunk],
         # This is for requesting updates. It doesn't need to be a general command sender right now,
         # but I think it's the correct way to be thinking about commands
         command_sender: Sender[ForwarderCommand],
@@ -74,8 +84,11 @@ class Worker:
         api_port: int,
     ):
         self.node_id: NodeId = node_id
+        self.session_id: SessionId = session_id
+        self.event_router = event_router
         self.event_receiver = event_receiver
         self.event_sender = event_sender
+        self.snapshot_chunk_receiver = snapshot_chunk_receiver
         self.command_sender = command_sender
         self.download_command_sender = download_command_sender
         self.api_port = api_port
@@ -95,6 +108,7 @@ class Worker:
         self._instance_backoff: KeyedBackoff[InstanceId] = KeyedBackoff(
             base=0.5, cap=10.0
         )
+        self._synced_custom_cards: dict[ModelId, ModelCard] = {}
         self._stopped: anyio.Event = anyio.Event()
 
     async def run(self):
@@ -105,21 +119,59 @@ class Worker:
 
         try:
             async with self._tg as tg:
-                tg.start_soon(info_gatherer.run)
-                tg.start_soon(self._forward_info, info_recv)
-                tg.start_soon(self.plan_step)
-                tg.start_soon(self._event_applier)
-                tg.start_soon(self._poll_connection_updates)
+                tg.start_soon(self._bootstrap_then_run, info_gatherer, info_recv)
         finally:
             # Actual shutdown code - waits for all tasks to complete before executing.
             logger.info("Stopping Worker")
             self.event_sender.close()
+            self.snapshot_chunk_receiver.close()
             self.command_sender.close()
             self.download_command_sender.close()
             for runner in self.runners.values():
                 runner.shutdown()
             self._stopped.set()
 
+    async def _bootstrap_then_run(
+        self, info_gatherer: InfoGatherer, info_recv: Receiver[GatheredInfo]
+    ) -> None:
+        await self._fetch_snapshot()
+        self._sync_input_views_from_state()
+        self._tg.start_soon(info_gatherer.run)
+        self._tg.start_soon(self._forward_info, info_recv)
+        self._tg.start_soon(self.plan_step)
+        self._tg.start_soon(self._event_applier)
+        self._tg.start_soon(self._reconcile_instance_backoff)
+        self._tg.start_soon(self._reconcile_custom_cards)
+        self._tg.start_soon(self._poll_connection_updates)
+
+    async def _fetch_snapshot(self) -> None:
+        receiver = SnapshotReceiver(self.node_id, self.session_id)
+        await self.command_sender.send(
+            ForwarderCommand(
+                origin=self._system_id,
+                command=RequestSnapshot(requester_node_id=self.node_id),
+            )
+        )
+
+        with anyio.move_on_after(_SNAPSHOT_FETCH_TIMEOUT_SECONDS):
+            with self.snapshot_chunk_receiver as chunks:
+                async for chunk in chunks:
+                    received = receiver.ingest(chunk)
+                    if received is None:
+                        continue
+                    self.state = received.state
+                    self.event_router.set_buffer_start(
+                        received.last_event_applied_idx + 1
+                    )
+                    logger.info(
+                        f"Worker bootstrapped from snapshot at idx "
+                        f"{received.last_event_applied_idx}"
+                    )
+                    return
+        logger.info(
+            "No snapshot received before timeout; falling back to full event-log replay"
+        )
+
     async def _forward_info(self, recv: Receiver[GatheredInfo]):
         with recv as info_stream:
             async for info in info_stream:
@@ -134,50 +186,69 @@ class Worker:
     async def _event_applier(self):
         with self.event_receiver as events:
             async for event in events:
+                if event.idx <= self.state.last_event_applied_idx:
+                    continue
                 # 2. for each event, apply it to the state
                 self.state = apply(self.state, event=event)
-                event = event.event
+                self._sync_input_views_from_state()
 
-                if isinstance(event, InstanceDeleted):
-                    self._instance_backoff.reset(event.instance_id)
+    async def _reconcile_instance_backoff(self) -> None:
+        while True:
+            await anyio.sleep(1)
+            self._reconcile_instance_backoff_once()
 
-                # Buffer input image chunks for image editing
-                if isinstance(event, InputChunkReceived):
-                    cmd_id = event.command_id
-                    if cmd_id not in self.input_chunk_buffer:
-                        self.input_chunk_buffer[cmd_id] = {}
-                        self.input_chunk_counts[cmd_id] = event.chunk.total_chunks
+    def _reconcile_instance_backoff_once(self) -> None:
+        live_instances = set(self.state.instances)
+        for instance_id in self._instance_backoff.tracked_keys():
+            if instance_id not in live_instances:
+                self._instance_backoff.reset(instance_id)
 
-                    self.input_chunk_buffer[cmd_id][event.chunk.chunk_index] = (
-                        event.chunk
-                    )
+    async def _reconcile_custom_cards(self) -> None:
+        while True:
+            await anyio.sleep(1)
+            await self._sync_custom_cards_from_state()
 
-                    if (
-                        len(self.input_chunk_buffer[cmd_id])
-                        == self.input_chunk_counts[cmd_id]
-                    ):
-                        per_image: defaultdict[int, list[InputImageChunk]] = (
-                            defaultdict(list)
-                        )
-                        for chunk in self.input_chunk_buffer[cmd_id].values():
-                            per_image[chunk.image_index].append(chunk)
-                        for chunks_for_image in per_image.values():
-                            sorted_chunks = sorted(
-                                chunks_for_image, key=lambda c: c.chunk_index
-                            )
-                            img = Base64Image("".join(c.data for c in sorted_chunks))
-                            self.image_cache[
-                                Base64ImageHash(
-                                    hashlib.sha256(img.encode("ascii")).hexdigest()
-                                )
-                            ] = img
+    async def _sync_custom_cards_from_state(self) -> None:
+        target = dict(self.state.custom_model_cards)
+        for model_id, card in target.items():
+            if self._synced_custom_cards.get(model_id) == card:
+                continue
+            await card.save_to_custom_dir()
+            add_to_card_cache(card)
+            self._synced_custom_cards[model_id] = card
 
-                if isinstance(event, CustomModelCardAdded):
-                    await event.model_card.save_to_custom_dir()
-                    add_to_card_cache(event.model_card)
+        for model_id in list(self._synced_custom_cards):
+            if model_id in target:
+                continue
+            await delete_custom_card(model_id)
+            self._synced_custom_cards.pop(model_id, None)
 
-                if isinstance(event, CustomModelCardDeleted):
-                    await delete_custom_card(event.model_id)
+    def _sync_input_views_from_state(self) -> None:
+        self.input_chunk_buffer = {
+            command_id: dict(chunks)
+            for command_id, chunks in self.state.input_chunks.items()
+        }
+        self.input_chunk_counts = {
+            command_id: next(iter(chunks.values())).total_chunks
+            for command_id, chunks in self.input_chunk_buffer.items()
+            if chunks
+        }
+
+        self.image_cache = {}
+        for command_id, chunks in self.input_chunk_buffer.items():
+            expected_chunks = self.input_chunk_counts.get(command_id)
+            if expected_chunks is None or len(chunks) != expected_chunks:
+                continue
+
+            per_image: defaultdict[int, list[InputImageChunk]] = defaultdict(list)
+            for chunk in chunks.values():
+                per_image[chunk.image_index].append(chunk)
+            for chunks_for_image in per_image.values():
+                sorted_chunks = sorted(chunks_for_image, key=lambda c: c.chunk_index)
+                image = Base64Image("".join(chunk.data for chunk in sorted_chunks))
+                self.image_cache[
+                    Base64ImageHash(hashlib.sha256(image.encode("ascii")).hexdigest())
+                ] = image
 
     async def plan_step(self):
         while True:
@@ -189,7 +260,7 @@ class Worker:
                 self.state.instances,
                 self.state.runners,
                 self.state.tasks,
-                self.input_chunk_buffer,
+                self.state.input_chunks,
                 self.image_cache,
                 self._instance_backoff,
                 self._download_backoff,
@@ -321,15 +392,9 @@ class Worker:
                             advanced_params=task.task_params.advanced_params,
                         ),
                     )
-                    # Cleanup buffers
-                    if cmd_id in self.input_chunk_buffer:
-                        del self.input_chunk_buffer[cmd_id]
-                    if cmd_id in self.input_chunk_counts:
-                        del self.input_chunk_counts[cmd_id]
                     await self._start_runner_task(modified_task)
 
                 case TextGeneration() if task.task_params.image_hashes:
-                    cmd_id = task.command_id
                     resolved_images = [
                         self.image_cache[h]
                         for _, h in sorted(task.task_params.image_hashes.items())
@@ -341,10 +406,6 @@ class Worker:
                             )
                         }
                     )
-                    if cmd_id in self.input_chunk_buffer:
-                        del self.input_chunk_buffer[cmd_id]
-                    if cmd_id in self.input_chunk_counts:
-                        del self.input_chunk_counts[cmd_id]
                     await self._start_runner_task(modified_task)
                 case LoadModel(instance_id=instance_id):
                     if (instance := self.state.instances.get(instance_id)) is not None:

src/exo/worker/tests/unittests/test_worker_custom_cards.py

@@ -0,0 +1,78 @@
+# pyright: reportPrivateUsage=false
+
+import pytest
+
+import exo.worker.main as worker_main
+from exo.shared.models.model_cards import ModelCard, ModelTask
+from exo.shared.types.common import ModelId
+from exo.shared.types.memory import Memory
+from exo.shared.types.state import State
+from exo.worker.main import Worker
+
+
+def _model_card(model_id: ModelId) -> ModelCard:
+    return ModelCard(
+        model_id=model_id,
+        n_layers=1,
+        storage_size=Memory.from_bytes(1),
+        hidden_size=1,
+        supports_tensor=True,
+        tasks=[ModelTask.TextGeneration],
+    )
+
+
+def _worker() -> Worker:
+    worker = object.__new__(Worker)
+    worker.state = State()
+    worker._synced_custom_cards = {}
+    return worker
+
+
+@pytest.mark.asyncio
+async def test_worker_syncs_custom_cards_from_state(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    saved: list[ModelId] = []
+    cached: list[ModelId] = []
+
+    async def save_to_custom_dir(card: ModelCard) -> None:
+        saved.append(card.model_id)
+
+    def add_to_card_cache(card: ModelCard) -> None:
+        cached.append(card.model_id)
+
+    monkeypatch.setattr(ModelCard, "save_to_custom_dir", save_to_custom_dir)
+    monkeypatch.setattr(worker_main, "add_to_card_cache", add_to_card_cache)
+
+    card = _model_card(ModelId("custom/model"))
+    worker = _worker()
+    worker.state = State(custom_model_cards={card.model_id: card})
+
+    await worker._sync_custom_cards_from_state()
+    await worker._sync_custom_cards_from_state()
+
+    assert saved == [card.model_id]
+    assert cached == [card.model_id]
+    assert worker._synced_custom_cards == {card.model_id: card}
+
+
+@pytest.mark.asyncio
+async def test_worker_deletes_custom_cards_missing_from_state(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    deleted: list[ModelId] = []
+
+    async def delete_custom_card(model_id: ModelId) -> bool:
+        deleted.append(model_id)
+        return True
+
+    monkeypatch.setattr(worker_main, "delete_custom_card", delete_custom_card)
+
+    card = _model_card(ModelId("custom/model"))
+    worker = _worker()
+    worker._synced_custom_cards = {card.model_id: card}
+
+    await worker._sync_custom_cards_from_state()
+
+    assert deleted == [card.model_id]
+    assert worker._synced_custom_cards == {}

src/exo/worker/tests/unittests/test_worker_instance_backoff.py

@@ -0,0 +1,36 @@
+# pyright: reportPrivateUsage=false
+
+from exo.shared.types.common import ModelId, NodeId
+from exo.shared.types.state import State
+from exo.shared.types.worker.instances import InstanceId, MlxRingInstance
+from exo.shared.types.worker.runners import ShardAssignments
+from exo.utils.keyed_backoff import KeyedBackoff
+from exo.worker.main import Worker
+
+
+def _make_instance(instance_id: InstanceId) -> MlxRingInstance:
+    return MlxRingInstance(
+        instance_id=instance_id,
+        shard_assignments=ShardAssignments(
+            model_id=ModelId("test-model"),
+            node_to_runner={},
+            runner_to_shard={},
+        ),
+        hosts_by_node={NodeId("node-1"): []},
+        ephemeral_port=1,
+    )
+
+
+def test_worker_reconciles_instance_backoff_from_state() -> None:
+    live_instance_id = InstanceId("inst-live")
+    deleted_instance_id = InstanceId("inst-deleted")
+    worker = object.__new__(Worker)
+    worker.state = State(instances={live_instance_id: _make_instance(live_instance_id)})
+    worker._instance_backoff = KeyedBackoff[InstanceId]()
+    worker._instance_backoff.record_attempt(live_instance_id)
+    worker._instance_backoff.record_attempt(deleted_instance_id)
+
+    worker._reconcile_instance_backoff_once()
+
+    assert worker._instance_backoff.attempts(live_instance_id) == 1
+    assert worker._instance_backoff.attempts(deleted_instance_id) == 0

src/exo/worker/tests/unittests/test_worker_snapshot_bootstrap.py

@@ -0,0 +1,126 @@
+# pyright: reportPrivateUsage=false
+
+import hashlib
+
+import anyio
+import pytest
+import zstandard
+
+from exo.routing.event_router import EventRouter
+from exo.shared.types.commands import (
+    ForwarderCommand,
+    ForwarderDownloadCommand,
+    RequestSnapshot,
+)
+from exo.shared.types.common import NodeId, SessionId
+from exo.shared.types.events import (
+    Event,
+    GlobalForwarderEvent,
+    IndexedEvent,
+    LocalForwarderEvent,
+    TestEvent,
+)
+from exo.shared.types.snapshots import SnapshotChunk, SnapshotTransferId
+from exo.shared.types.state import State
+from exo.utils.channels import Receiver, Sender, channel
+from exo.worker.main import Worker
+
+
+def _snapshot_chunk(
+    state: State, *, requester_node_id: NodeId, session_id: SessionId
+) -> SnapshotChunk:
+    body = zstandard.ZstdCompressor().compress(state.model_dump_json().encode("utf-8"))
+    return SnapshotChunk.from_data(
+        data=body,
+        transfer_id=SnapshotTransferId("transfer-1"),
+        requester_node_id=requester_node_id,
+        session_id=session_id,
+        schema_version=state.schema_version,
+        last_event_applied_idx=state.last_event_applied_idx,
+        chunk_index=0,
+        total_chunks=1,
+        sha256_hex=hashlib.sha256(body).hexdigest(),
+    )
+
+
+def _worker(
+    node_id: NodeId, session_id: SessionId
+) -> tuple[
+    Worker,
+    EventRouter,
+    Receiver[ForwarderCommand],
+    Sender[SnapshotChunk],
+    Sender[IndexedEvent],
+]:
+    router_command_sender, _router_command_receiver = channel[ForwarderCommand]()
+    _global_event_sender, global_event_receiver = channel[GlobalForwarderEvent]()
+    local_event_sender, _local_event_receiver = channel[LocalForwarderEvent]()
+    event_router = EventRouter(
+        session_id=session_id,
+        command_sender=router_command_sender,
+        external_inbound=global_event_receiver,
+        external_outbound=local_event_sender,
+    )
+
+    event_sender, event_receiver = channel[IndexedEvent]()
+    local_event_output_sender, _local_event_output_receiver = channel[Event]()
+    command_sender, command_receiver = channel[ForwarderCommand]()
+    download_command_sender, _download_command_receiver = channel[
+        ForwarderDownloadCommand
+    ]()
+    snapshot_sender, snapshot_receiver = channel[SnapshotChunk]()
+
+    worker = Worker(
+        node_id,
+        session_id,
+        event_router=event_router,
+        event_receiver=event_receiver,
+        event_sender=local_event_output_sender,
+        snapshot_chunk_receiver=snapshot_receiver,
+        command_sender=command_sender,
+        download_command_sender=download_command_sender,
+        api_port=52415,
+    )
+    return worker, event_router, command_receiver, snapshot_sender, event_sender
+
+
+@pytest.mark.asyncio
+async def test_worker_fetch_snapshot_applies_state_and_fast_forwards_router() -> None:
+    node_id = NodeId("worker")
+    session_id = SessionId(master_node_id=NodeId("master"), election_clock=1)
+    worker, event_router, command_receiver, snapshot_sender, _event_sender = _worker(
+        node_id, session_id
+    )
+    state = State(last_event_applied_idx=7)
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(worker._fetch_snapshot)
+        command = await command_receiver.receive()
+        assert isinstance(command.command, RequestSnapshot)
+        assert command.command.requester_node_id == node_id
+
+        await snapshot_sender.send(
+            _snapshot_chunk(state, requester_node_id=node_id, session_id=session_id)
+        )
+
+    assert worker.state.last_event_applied_idx == 7
+    assert event_router.event_buffer.next_idx_to_release == 8
+
+
+@pytest.mark.asyncio
+async def test_worker_event_applier_ignores_events_covered_by_snapshot() -> None:
+    node_id = NodeId("worker")
+    session_id = SessionId(master_node_id=NodeId("master"), election_clock=1)
+    worker, _event_router, _command_receiver, _snapshot_sender, event_sender = _worker(
+        node_id, session_id
+    )
+    worker.state = State(last_event_applied_idx=7)
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(worker._event_applier)
+        await event_sender.send(IndexedEvent(idx=7, event=TestEvent()))
+        await event_sender.send(IndexedEvent(idx=8, event=TestEvent()))
+
+        while worker.state.last_event_applied_idx != 8:
+            await anyio.sleep(0.001)
+        tg.cancel_scope.cancel()