Bump mlx

## Motivation

We want to use log mining tools like
[Drain3](https://github.com/logpai/Drain3) to get standardized error
formats, but for that we should record runner stdout/stderr in a massive
append-only log to gather training data for such tools. Also useful for
future opt-in telemetry.

## Changes

The stdout/stderr from runner now splits into 3 tasks: 
1) raw write to dedicated runner logs 
2) sanitized line-by-line logging with log-guru 
3) stub for further error-processing (i.e. turning lines into errors)

### Manual Testing
Works on 4x mac mini clusted connected as TB4 ring.

.gitignore

@@ -40,3 +40,4 @@ bench/**/*.json
 tmp/models
 /build/exo
 /.claude/skills
+/.claude

Cargo.lock

@@ -916,11 +916,13 @@ dependencies = [
  "libp2p",
  "log",
  "networking",
+ "pidfile-rs",
  "pin-project",
  "pyo3",
  "pyo3-async-runtimes",
  "pyo3-log",
  "pyo3-stub-gen",
+ "thiserror 2.0.17",
  "tokio",
  "util",
 ]
@@ -964,6 +966,16 @@ version = "0.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3a3076410a55c90011c298b04d0cfa770b00fa04e1e3c97d3f6c9de105a03844"
 
+[[package]]
+name = "flopen"
+version = "0.1.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fbfb8b5fbd1f27929f216650081a07b6ceb0741f0542c8c43ff7ef8e93a35a5d"
+dependencies = [
+ "libc",
+ "nix 0.31.2",
+]
+
 [[package]]
 name = "fnv"
 version = "1.0.7"
@@ -1789,9 +1801,9 @@ checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe"
 
 [[package]]
 name = "libc"
-version = "0.2.178"
+version = "0.2.186"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "37c93d8daa9d8a012fd8ab92f088405fb202ea0b6ab73ee2482ae66af4f42091"
+checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66"
 
 [[package]]
 name = "libp2p"
@@ -2807,6 +2819,18 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "nix"
+version = "0.31.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5d6d0705320c1e6ba1d912b5e37cf18071b6c2e9b7fa8215a1e8a7651966f5d3"
+dependencies = [
+ "bitflags 2.10.0",
+ "cfg-if",
+ "cfg_aliases",
+ "libc",
+]
+
 [[package]]
 name = "nohash-hasher"
 version = "0.2.0"
@@ -3060,6 +3084,18 @@ dependencies = [
  "siphasher",
 ]
 
+[[package]]
+name = "pidfile-rs"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d1a8aa9a30b1b65ef48b333931b80f2324a14e00208eb2b8f5788f1180791bcc"
+dependencies = [
+ "flopen",
+ "libc",
+ "log",
+ "thiserror 1.0.69",
+]
+
 [[package]]
 name = "pin-project"
 version = "1.1.10"
@@ -3668,7 +3704,7 @@ dependencies = [
  "netlink-packet-utils",
  "netlink-proto",
  "netlink-sys",
- "nix",
+ "nix 0.26.4",
  "thiserror 1.0.69",
  "tokio",
 ]

bench/eval_tool_calls.py

@@ -15,9 +15,8 @@ from pathlib import Path
 from typing import Any, Literal
 
 import httpx
-from harness import (
-    ExoClient,
-    ExoHttpError,
+from exo_tools.client import ExoClient, ExoHttpError
+from exo_tools.harness import (
     add_common_instance_args,
     capture_cluster_snapshot,
     instance_id_from_instance,

bench/exo_bench.py

@@ -30,9 +30,8 @@ from pathlib import Path
 from statistics import mean
 from typing import Any
 
-from harness import (
-    ExoClient,
-    ExoHttpError,
+from exo_tools.client import ExoClient, ExoHttpError
+from exo_tools.harness import (
     add_common_instance_args,
     capture_cluster_snapshot,
     find_existing_instance,

bench/exo_eval.py

@@ -42,9 +42,8 @@ from pathlib import Path
 from typing import Any
 
 import httpx
-from harness import (
-    ExoClient,
-    ExoHttpError,
+from exo_tools.client import ExoClient, ExoHttpError
+from exo_tools.harness import (
     add_common_instance_args,
     capture_cluster_snapshot,
     find_existing_instance,

bench/prefill_decode_bench.py

@@ -35,9 +35,8 @@ from exo_bench import (
     load_tokenizer_for_bench,
     parse_int_list,
 )
-from harness import (
-    ExoClient,
-    ExoHttpError,
+from exo_tools.client import ExoClient, ExoHttpError
+from exo_tools.harness import (
     add_common_instance_args,
     instance_id_from_instance,
     node_ids_from_instance,

pyproject.toml

@@ -40,6 +40,7 @@ exo = "exo.main:main"
 dev = [
   "basedpyright>=1.29.0",
   "pyinstaller>=6.17.0",
+  "playwright>=1.52.0",
   "pytest>=8.4.0",
   "pytest-asyncio>=1.0.0",
   "pytest-env",
@@ -75,7 +76,7 @@ cuda13 = [
 ###
 
 [tool.uv.workspace]
-members = ["rust/exo_pyo3_bindings", "bench"]
+members = ["rust/exo_pyo3_bindings", "bench", "tools"]
 
 [tool.uv.sources]
 exo-pyo3-bindings = { workspace = true }
@@ -112,7 +113,7 @@ build-backend = "uv_build"
 ###
 
 [tool.basedpyright]
-include = ["src", "bench"]
+include = ["src", "bench", "tools"]
 typeCheckingMode = "strict"
 failOnWarnings = true
 
@@ -146,6 +147,13 @@ reportMissingModuleSource = false
 [[tool.basedpyright.executionEnvironments]]
 root = "src"
 
+[[tool.basedpyright.executionEnvironments]]
+root = "bench"
+extraPaths = ["tools/src"]
+
+[[tool.basedpyright.executionEnvironments]]
+root = "tools/src"
+
 
 ###
 # uv configuration
@@ -220,5 +228,5 @@ pythonpath = "."
 asyncio_mode = "auto"
 markers = ["slow: marks tests as slow (deselected by default)"]
 env = ["EXO_TESTS=1"]
-addopts = "-m 'not slow' --ignore=tests/start_distributed_test.py"
+addopts = "-m 'not slow' --ignore=tests"
 filterwarnings = ["ignore:builtin type Swig:DeprecationWarning"]

rust/exo_pyo3_bindings/Cargo.toml

@@ -46,9 +46,12 @@ pyo3-async-runtimes = { version = "0.27.0", features = [
 ] }
 pyo3-log = "0.13.2"
 
+pidfile-rs = "0.3"
+
 # macro dependencies
 extend = { workspace = true }
 delegate = { workspace = true }
+thiserror = "2.0"
 
 # async runtime
 tokio = { workspace = true, features = ["full", "tracing"] }

rust/exo_pyo3_bindings/exo_pyo3_bindings.pyi

@@ -2,6 +2,8 @@
 # ruff: noqa: E501, F401
 
 import builtins
+import os
+import pathlib
 import typing
 
 @typing.final
@@ -69,6 +71,48 @@ class NoPeersSubscribedToTopicError(builtins.Exception):
     def __repr__(self) -> builtins.str: ...
     def __str__(self) -> builtins.str: ...
 
+@typing.final
+class Pidfile:
+    r"""
+    A PID file protected with a lock.
+    
+    An instance of `Pidfile` can be used to manage a PID file: create it,
+    lock it, detect already running daemons. It is backed by [`pidfile`][]
+    functions of `libbsd`/`libutil` which use `flopen` to lock the PID
+    file.
+    
+    When a PID file is created, the process ID of the current process is
+    *not* written there, making it possible to lock the PID file before
+    forking and only write the ID of the forked process when it is ready.
+    
+    The PID file is deleted automatically when the `Pidfile` comes out of
+    the scope. To close the PID file without deleting it, for example, in
+    the parent process of a forked daemon, call `close()`.
+    
+    [`exit`]: https://doc.rust-lang.org/std/process/fn.exit.html
+    [`pidfile`]: https://linux.die.net/man/3/pidfile
+    [`daemon`(3)]: https://linux.die.net/man/3/daemon
+    """
+    def __new__(cls, path: builtins.str | os.PathLike | pathlib.Path, mode: builtins.int) -> Pidfile:
+        r"""
+        Creates a new PID file and locks it.
+        
+        If the PID file cannot be locked, returns `PidfileError::AlreadyRunning` with
+        a PID of the already running process, or `None` if no PID has been written to
+        the PID file yet.
+        """
+    def write(self) -> None:
+        r"""
+        Writes the current process ID to the PID file.
+        
+        The file is truncated before writing.
+        """
+
+@typing.final
+class PidfileError(builtins.Exception):
+    def __repr__(self) -> builtins.str: ...
+    def __str__(self) -> builtins.str: ...
+
 class PyFromSwarm:
     @typing.final
     class Connection(PyFromSwarm):

rust/exo_pyo3_bindings/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "maturin"
 
 [project]
 name = "exo_pyo3_bindings"
-version = "0.2.1"
+version = "0.2.2"
 description = "Add your description here"
 readme = "README.md"
 authors = [

rust/exo_pyo3_bindings/src/lib.rs

@@ -7,9 +7,11 @@
 mod allow_threading;
 mod ident;
 mod networking;
+mod pidfile;
 
 use crate::ident::PyKeypair;
 use crate::networking::networking_submodule;
+use crate::pidfile::pidfile_submodule;
 use pyo3::prelude::PyModule;
 use pyo3::types::PyModuleMethods;
 use pyo3::{Bound, PyResult, pyclass, pymodule};
@@ -164,6 +166,7 @@ fn main_module(m: &Bound<'_, PyModule>) -> PyResult<()> {
     //       too many importing issues...
     m.add_class::<PyKeypair>()?;
     networking_submodule(m)?;
+    pidfile_submodule(m)?;
 
     // top-level constructs
     // TODO: ...

rust/exo_pyo3_bindings/src/pidfile.rs

@@ -0,0 +1,87 @@
+use pidfile_rs::{Pidfile, PidfileError};
+use pyo3::exceptions::PyException;
+use pyo3::prelude::{PyModule, PyModuleMethods};
+use pyo3::{Bound, PyErr, PyResult, Python, pyclass, pymethods};
+use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pymethods};
+use std::fs::Permissions;
+use std::os::unix::prelude::PermissionsExt;
+use std::path::PathBuf;
+
+#[gen_stub_pyclass]
+#[pyclass(frozen, extends=PyException, name="PidfileError")]
+pub struct PyPidfileError(PidfileError);
+
+impl PyPidfileError {
+    // TODO: I actually like this pattern a LOT more but how to abstract??
+    fn into_pyerr(self, py: Python) -> PyErr {
+        match Bound::new(py, self) {
+            Ok(err) => PyErr::from_value(err.into_any()),
+            Err(err) => err,
+        }
+    }
+}
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl PyPidfileError {
+    fn __repr__(&self) -> String {
+        format!("PidfileError(\"{}\")", self.0)
+    }
+
+    fn __str__(&self) -> String {
+        self.0.to_string()
+    }
+}
+
+/// A PID file protected with a lock.
+///
+/// An instance of `Pidfile` can be used to manage a PID file: create it,
+/// lock it, detect already running daemons. It is backed by [`pidfile`][]
+/// functions of `libbsd`/`libutil` which use `flopen` to lock the PID
+/// file.
+///
+/// When a PID file is created, the process ID of the current process is
+/// *not* written there, making it possible to lock the PID file before
+/// forking and only write the ID of the forked process when it is ready.
+///
+/// The PID file is deleted automatically when the `Pidfile` comes out of
+/// the scope. To close the PID file without deleting it, for example, in
+/// the parent process of a forked daemon, call `close()`.
+///
+/// [`exit`]: https://doc.rust-lang.org/std/process/fn.exit.html
+/// [`pidfile`]: https://linux.die.net/man/3/pidfile
+/// [`daemon`(3)]: https://linux.die.net/man/3/daemon
+#[gen_stub_pyclass]
+#[pyclass(name = "Pidfile")]
+pub struct PyPidfile(Pidfile);
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl PyPidfile {
+    /// Creates a new PID file and locks it.
+    ///
+    /// If the PID file cannot be locked, returns `PidfileError::AlreadyRunning` with
+    /// a PID of the already running process, or `None` if no PID has been written to
+    /// the PID file yet.
+    #[new]
+    fn py_new(py: Python, path: PathBuf, mode: u32) -> PyResult<Self> {
+        Ok(Self(
+            Pidfile::new(&path, Permissions::from_mode(mode))
+                .map_err(|e| PyPidfileError(e).into_pyerr(py))?,
+        ))
+    }
+
+    /// Writes the current process ID to the PID file.
+    ///
+    /// The file is truncated before writing.
+    fn write<'py>(&mut self, py: Python<'py>) -> PyResult<()> {
+        self.0.write().map_err(|e| PyPidfileError(e).into_pyerr(py))
+    }
+}
+
+pub fn pidfile_submodule(m: &Bound<PyModule>) -> PyResult<()> {
+    m.add_class::<PyPidfileError>()?;
+    m.add_class::<PyPidfile>()?;
+
+    Ok(())
+}

rust/exo_pyo3_bindings/tests/test_python.py

@@ -1,10 +1,12 @@
 import asyncio
 
 import pytest
+from _pytest.capture import CaptureFixture
 from exo_pyo3_bindings import (
     Keypair,
     NetworkingHandle,
     NoPeersSubscribedToTopicError,
+    Pidfile,
     PyFromSwarm,
 )
 
@@ -26,6 +28,13 @@ async def test_sleep_on_multiple_items() -> None:
             print("caught it", e)
 
 
+def test_pidfile(capsys: CaptureFixture[str]):
+    with capsys.disabled():
+        print("\nbefore python")
+        scoped_lock_file()
+        print("after python")
+
+
 async def _await_recv(h: NetworkingHandle):
     while True:
         event = await h.recv()
@@ -34,3 +43,7 @@ async def _await_recv(h: NetworkingHandle):
                 print(f"PYTHON: connection update: {c}")
             case PyFromSwarm.Message() as m:
                 print(f"PYTHON: message: {m}")
+
+
+def scoped_lock_file():
+    a = Pidfile("/tmp/lock.pid", 0o0600)

src/exo/api/main.py

@@ -20,6 +20,7 @@ from fastapi.staticfiles import StaticFiles
 from hypercorn.asyncio import serve  # pyright: ignore[reportUnknownVariableType]
 from hypercorn.config import Config
 from hypercorn.typing import ASGIFramework
+from hypercorn.utils import LifespanTimeoutError
 from loguru import logger
 
 from exo.api.adapters.chat_completions import (
@@ -121,8 +122,6 @@ 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,
@@ -135,12 +134,10 @@ from exo.shared.constants import (
 )
 from exo.shared.election import ElectionMessage
 from exo.shared.logging import InterceptLogger
+from exo.shared.models import model_cards
 from exo.shared.models.model_cards import (
     ModelCard,
     ModelId,
-    add_to_card_cache,
-    get_card,
-    get_model_cards,
 )
 from exo.shared.tracing import TraceEvent, compute_stats, export_trace, load_trace_file
 from exo.shared.types.chunks import (
@@ -166,7 +163,6 @@ from exo.shared.types.commands import (
     ImageEdits,
     ImageGeneration,
     PlaceInstance,
-    RequestSnapshot,
     SendInputChunk,
     SetInstanceLink,
     StartDownload,
@@ -174,7 +170,7 @@ from exo.shared.types.commands import (
     TaskFinished,
     TextGeneration,
 )
-from exo.shared.types.common import CommandId, Id, NodeId, SessionId, SystemId
+from exo.shared.types.common import CommandId, Id, NodeId, SystemId
 from exo.shared.types.events import (
     ChunkGenerated,
     Event,
@@ -184,7 +180,6 @@ from exo.shared.types.events import (
 )
 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,
@@ -211,8 +206,6 @@ 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'}"
@@ -242,12 +235,9 @@ 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
@@ -256,16 +246,14 @@ 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()
@@ -303,29 +291,19 @@ class API:
         self._image_store = ImageStore(EXO_IMAGE_CACHE_DIR)
         self._tg: TaskGroup = TaskGroup()
 
-    def reset(
-        self,
-        result_clock: int,
-        session_id: SessionId,
-        event_router: EventRouter,
-        event_receiver: Receiver[IndexedEvent],
-        snapshot_chunk_receiver: Receiver[SnapshotChunk],
-    ):
+    def reset(self, result_clock: int, event_receiver: Receiver[IndexedEvent]):
         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.unpause(result_clock)
         self.event_receiver.close()
         self.event_receiver = event_receiver
-        self.snapshot_chunk_receiver.close()
-        self.snapshot_chunk_receiver = snapshot_chunk_receiver
-        self._tg.start_soon(self._bootstrap_then_apply_state)
+        self._tg.start_soon(self._apply_state)
+        self._sent_image_hashes = set()
 
     def unpause(self, result_clock: int):
         logger.info("Unpausing API")
@@ -502,6 +480,7 @@ class API:
                 topology=self.state.topology,
                 current_instances=self.state.instances,
                 download_status=self.state.downloads,
+                node_rdma_ctl=self.state.node_rdma_ctl,
             )
         except ValueError as exc:
             raise HTTPException(status_code=400, detail=str(exc)) from exc
@@ -565,6 +544,7 @@ class API:
                     current_instances=self.state.instances,
                     required_nodes=required_nodes,
                     download_status=self.state.downloads,
+                    node_rdma_ctl=self.state.node_rdma_ctl,
                 )
             except ValueError as exc:
                 if (model_card.model_id, sharding, instance_meta, 0) not in seen:
@@ -847,8 +827,18 @@ 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 enumerate(images):
+        for img_idx, img_data in new_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]))
 
@@ -1644,17 +1634,16 @@ class API:
     async def ollama_tags(self) -> OllamaTagsResponse:
         """Returns list of models in Ollama tags format. We return the downloaded ones only."""
 
-        def none_if_empty(value: str) -> str | None:
-            return value or None
-
-        downloaded_model_ids: set[str] = set()
+        downloaded_model_ids: set[ModelId] = set()
         for node_downloads in self.state.downloads.values():
             for dl in node_downloads:
                 if isinstance(dl, DownloadCompleted):
                     downloaded_model_ids.add(dl.shard_metadata.model_card.model_id)
 
         cards = [
-            c for c in await get_model_cards() if c.model_id in downloaded_model_ids
+            c
+            for c in await model_cards.card_cache.list_all()
+            if c.model_id in downloaded_model_ids
         ]
 
         now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
@@ -1667,8 +1656,8 @@ class API:
                     size=card.storage_size.in_bytes,
                     digest="sha256:000000000000",
                     details=OllamaModelDetails(
-                        family=none_if_empty(card.family),
-                        quantization_level=none_if_empty(card.quantization),
+                        family=card.family or None,
+                        quantization_level=card.quantization or None,
                     ),
                 )
                 for card in cards
@@ -1731,7 +1720,7 @@ class API:
 
     async def get_models(self, status: str | None = Query(default=None)) -> ModelList:
         """Returns list of available models, optionally filtered by being downloaded."""
-        cards = await get_model_cards()
+        cards = await model_cards.card_cache.list_all()
 
         if status == "downloaded":
             downloaded_model_ids: set[str] = set()
@@ -1782,7 +1771,7 @@ class API:
 
         # Immediately update the local cache so the subsequent GET /models
         # returns the new model without waiting for the event round-trip.
-        add_to_card_cache(card)
+        model_cards.card_cache.cc[card.model_id] = card
 
         return ModelListModel(
             id=card.model_id,
@@ -1798,7 +1787,7 @@ class API:
 
     async def delete_custom_model(self, model_id: ModelId) -> JSONResponse:
         """Delete a user-added custom model card and sync deletion across the cluster."""
-        card = get_card(model_id)
+        card = model_cards.card_cache.get(model_id)
         if card is None or not card.is_custom:
             raise HTTPException(status_code=404, detail="Custom model card not found")
 
@@ -1859,7 +1848,7 @@ class API:
         try:
             async with self._tg as tg:
                 logger.info("Starting API")
-                tg.start_soon(self._bootstrap_then_apply_state)
+                tg.start_soon(self._apply_state)
                 tg.start_soon(self._pause_on_new_election)
                 tg.start_soon(self._cleanup_expired_images)
                 print_startup_banner(self.port)
@@ -1868,12 +1857,20 @@ class API:
                     await anyio.sleep_forever()
                 finally:
                     with anyio.CancelScope(shield=True):
+                        # IMPORTANT: when new queues are added, update this (for proper shutdown semantics)
+                        self._shutdown_queues(self._text_generation_queues)
+                        self._shutdown_queues(self._image_generation_queues)
+
                         shutdown_ev.set()
         finally:
             self._event_log.close()
             self.command_sender.close()
             self.event_receiver.close()
-            self.snapshot_chunk_receiver.close()
+
+    @staticmethod
+    def _shutdown_queues[K, V](queues: dict[K, Sender[V]]):
+        for v in queues.values():
+            v.close()
 
     async def run_api(self, ev: anyio.Event):
         cfg = Config()
@@ -1882,50 +1879,27 @@ class API:
         cfg.accesslog = None
         cfg.errorlog = "-"
         cfg.logger_class = InterceptLogger
+
+        # prevents hangs when mid-request and connection refuses to close
+        cfg.graceful_timeout = 2  # seconds
+        cfg.shutdown_timeout = 3  # seconds
+
         with anyio.CancelScope(shield=True):
-            await serve(
-                cast(ASGIFramework, self.app),
-                cfg,
-                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"
-        )
+            try:
+                await serve(
+                    cast(ASGIFramework, self.app),
+                    cfg,
+                    shutdown_trigger=ev.wait,
+                )
+            except LifespanTimeoutError as e:
+                logger.warning(
+                    "Graceful server shutdown timed out, some connections forcebly closed"
+                )
+                logger.opt(exception=e).debug("")
 
     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

src/exo/api/tests/test_api_snapshot_bootstrap.py

@@ -1,136 +0,0 @@
-# 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/download/coordinator.py

@@ -16,7 +16,8 @@ from exo.download.download_utils import (
 )
 from exo.download.shard_downloader import ShardDownloader
 from exo.shared.constants import EXO_DEFAULT_MODELS_DIR, EXO_MODELS_READ_ONLY_DIRS
-from exo.shared.models.model_cards import ModelId, get_model_cards
+from exo.shared.models import model_cards
+from exo.shared.models.model_cards import ModelId
 from exo.shared.types.commands import (
     CancelDownload,
     DeleteDownload,
@@ -422,7 +423,7 @@ class DownloadCoordinator:
                     )
                 # Scan read-only directories for pre-downloaded models
                 if EXO_MODELS_READ_ONLY_DIRS:
-                    for card in await get_model_cards():
+                    for card in await model_cards.card_cache.list_all():
                         mid = card.model_id
                         if mid in self.active_downloads:
                             continue

src/exo/download/impl_shard_downloader.py

@@ -11,11 +11,11 @@ from exo.download.download_utils import (
     download_shard,
 )
 from exo.download.shard_downloader import ShardDownloader
+from exo.shared.models import model_cards
 from exo.shared.models.model_cards import (
     ModelCard,
     ModelId,
     ModelTask,
-    get_model_cards,
 )
 from exo.shared.types.memory import Memory
 from exo.shared.types.worker.shards import (
@@ -258,7 +258,7 @@ class ResumableShardDownloader(ShardDownloader):
 
         tasks = [
             create_task(download_with_semaphore(model_card))
-            for model_card in await get_model_cards()
+            for model_card in await model_cards.card_cache.list_all()
         ]
 
         for task in asyncio.as_completed(tasks):

src/exo/main.py

@@ -3,6 +3,7 @@ import multiprocessing as mp
 import os
 import resource
 import signal
+import sys
 from dataclasses import dataclass, field
 from typing import Self
 
@@ -22,6 +23,8 @@ from exo.shared.election import Election, ElectionResult
 from exo.shared.logging import logger_cleanup, logger_setup
 from exo.shared.types.common import NodeId, SessionId
 from exo.utils.channels import Receiver, channel
+from exo.utils.daemon import detach_stdio_to_devnull
+from exo.utils.pidfile import PidfileLockError, acquire_exo_pidfile
 from exo.utils.pydantic_ext import FrozenModel
 from exo.utils.task_group import TaskGroup
 from exo.worker.main import Worker
@@ -59,7 +62,6 @@ 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),
@@ -84,11 +86,8 @@ 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),
@@ -99,11 +98,8 @@ 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,
@@ -119,7 +115,6 @@ 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),
         )
 
@@ -218,9 +213,6 @@ 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
                         ),
@@ -257,13 +249,8 @@ 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
@@ -272,13 +259,7 @@ class Node:
                         )
                         self._tg.start_soon(self.worker.run)
                     if self.api:
-                        self.api.reset(
-                            result.won_clock,
-                            result.session_id,
-                            self.event_router,
-                            self.event_router.receiver(),
-                            self.router.receiver(topics.SNAPSHOT_RESPONSES),
-                        )
+                        self.api.reset(result.won_clock, self.event_router.receiver())
                     self._tg.start_soon(self.event_router.run)
                 else:
                     if self.api:
@@ -286,14 +267,26 @@ class Node:
 
 
 def main():
+    # Exit early if no PID file (not compatible with double-for daemonization yet)
+    try:
+        pidfile = acquire_exo_pidfile()
+    except PidfileLockError as exception:
+        print(exception, file=sys.stderr)
+        raise SystemExit(1) from exception
+
     args = Args.parse()
     soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE)
     target = min(max(soft, 65535), hard)
     resource.setrlimit(resource.RLIMIT_NOFILE, (target, hard))
 
     mp.set_start_method("spawn", force=True)
+
     # TODO: Refactor the current verbosity system
     logger_setup(EXO_LOG, args.verbosity)
+    if args.no_stdio:
+        detach_stdio_to_devnull()
+        logger.info("Detached stdio to /dev/null")
+
     logger.info(f"{'=' * 40}")
     logger.info(f"Starting EXO | pid={os.getpid()}")
     logger.info(f"{'=' * 40}")
@@ -328,6 +321,7 @@ def main():
     finally:
         logger.info("EXO Shutdown complete")
         logger_cleanup()
+        del pidfile
 
 
 class Args(FrozenModel):
@@ -341,6 +335,7 @@ class Args(FrozenModel):
     offline: bool = os.getenv("EXO_OFFLINE", "false").lower() == "true"
     no_batch: bool = False
     fast_synch: bool | None = None  # None = auto, True = force on, False = force off
+    no_stdio: bool = False
     bootstrap_peers: list[str] = []
     libp2p_port: int
 
@@ -400,6 +395,11 @@ class Args(FrozenModel):
             action="store_true",
             help="Disable continuous batching, use sequential generation",
         )
+        parser.add_argument(
+            "--no-stdio",
+            action="store_true",
+            help="Detach stdin/stdout/stderr to /dev/null after logging is configured",
+        )
         parser.add_argument(
             "--bootstrap-peers",
             type=lambda s: [p for p in s.split(",") if p],

src/exo/master/main.py

@@ -1,8 +1,6 @@
-import hashlib
 from datetime import datetime, timedelta, timezone
 
 import anyio
-from anyio import to_thread
 from loguru import logger
 
 from exo.master.placement import (
@@ -27,7 +25,6 @@ from exo.shared.types.commands import (
     ImageGeneration,
     PlaceInstance,
     RequestEventLog,
-    RequestSnapshot,
     SendInputChunk,
     SetInstanceLink,
     TaskCancelled,
@@ -57,7 +54,6 @@ 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,
@@ -78,15 +74,6 @@ 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)
@@ -138,7 +125,6 @@ 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
@@ -149,7 +135,6 @@ 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()
@@ -170,7 +155,6 @@ 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):
@@ -381,6 +365,7 @@ class Master:
                                 self.state.node_memory,
                                 self.state.node_network,
                                 download_status=self.state.downloads,
+                                node_rdma_ctl=self.state.node_rdma_ctl,
                             )
                             transition_events = get_transition_events(
                                 self.state.instances, placement, self.state.tasks
@@ -457,19 +442,12 @@ 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 + _MAX_EVENT_LOG_REPLAY_BATCH,
-                                len(self._event_log),
-                            )
+                            end = min(command.since_idx + 1000, 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:
@@ -529,42 +507,6 @@ 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/placement.py

@@ -28,7 +28,7 @@ from exo.shared.types.events import (
     TaskStatusUpdated,
 )
 from exo.shared.types.memory import Memory
-from exo.shared.types.profiling import MemoryUsage, NodeNetworkInfo
+from exo.shared.types.profiling import MemoryUsage, NodeNetworkInfo, NodeRdmaCtlStatus
 from exo.shared.types.tasks import Task, TaskId, TaskStatus
 from exo.shared.types.worker.downloads import (
     DownloadCompleted,
@@ -105,6 +105,7 @@ def place_instance(
     node_network: Mapping[NodeId, NodeNetworkInfo],
     required_nodes: set[NodeId] | None = None,
     download_status: Mapping[NodeId, Sequence[DownloadProgress]] | None = None,
+    node_rdma_ctl: Mapping[NodeId, NodeRdmaCtlStatus] | None = None,
 ) -> dict[InstanceId, Instance]:
     cycles = topology.get_cycles()
     candidate_cycles = list(filter(lambda it: len(it) >= command.min_nodes, cycles))
@@ -166,8 +167,18 @@ def place_instance(
 
     smallest_cycles = get_smallest_cycles(cycles_with_sufficient_memory)
 
+    rdma_ctl_status = node_rdma_ctl or {}
+
+    def _all_rdma_ctl_enabled(cycle: Cycle) -> bool:
+        return all(
+            ((status := rdma_ctl_status.get(node_id)) is not None and status.enabled)
+            for node_id in cycle
+        )
+
     smallest_rdma_cycles = [
-        cycle for cycle in smallest_cycles if topology.is_rdma_cycle(cycle)
+        cycle
+        for cycle in smallest_cycles
+        if topology.is_rdma_cycle(cycle) and _all_rdma_ctl_enabled(cycle)
     ]
 
     if command.instance_meta == InstanceMeta.MlxJaccl:

src/exo/master/tests/test_master.py

@@ -7,14 +7,12 @@ 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
@@ -31,7 +29,6 @@ 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 (
@@ -59,7 +56,6 @@ 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
@@ -96,7 +92,6 @@ 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")
@@ -234,52 +229,3 @@ 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/master/tests/test_placement.py

@@ -21,7 +21,11 @@ from exo.shared.types.events import (
 )
 from exo.shared.types.memory import Memory
 from exo.shared.types.multiaddr import Multiaddr
-from exo.shared.types.profiling import NetworkInterfaceInfo, NodeNetworkInfo
+from exo.shared.types.profiling import (
+    NetworkInterfaceInfo,
+    NodeNetworkInfo,
+    NodeRdmaCtlStatus,
+)
 from exo.shared.types.tasks import TaskId, TaskStatus, TextGeneration
 from exo.shared.types.text_generation import (
     InputMessage,
@@ -439,8 +443,21 @@ def test_tensor_rdma_backend_connectivity_matrix(
         min_nodes=1,
     )
 
+    node_rdma_ctl = {
+        node_a: NodeRdmaCtlStatus(enabled=True),
+        node_b: NodeRdmaCtlStatus(enabled=True),
+        node_c: NodeRdmaCtlStatus(enabled=True),
+    }
+
     # act
-    placements = place_instance(cic, topology, {}, node_memory, node_network)
+    placements = place_instance(
+        cic,
+        topology,
+        {},
+        node_memory,
+        node_network,
+        node_rdma_ctl=node_rdma_ctl,
+    )
 
     # assert
     assert len(placements) == 1
@@ -482,6 +499,131 @@ def test_tensor_rdma_backend_connectivity_matrix(
             assert len(ip_part.split(".")) == 4
 
 
+def _build_three_node_rdma_topology() -> tuple[
+    Topology, NodeId, NodeId, NodeId, dict[NodeId, NodeNetworkInfo]
+]:
+    topology = Topology()
+    node_a = NodeId()
+    node_b = NodeId()
+    node_c = NodeId()
+
+    ethernet_interface = NetworkInterfaceInfo(name="en0", ip_address="10.0.0.1")
+    ethernet_conn = SocketConnection(
+        sink_multiaddr=Multiaddr(address="/ip4/10.0.0.1/tcp/8000")
+    )
+    node_network = {
+        node_a: NodeNetworkInfo(interfaces=[ethernet_interface]),
+        node_b: NodeNetworkInfo(interfaces=[ethernet_interface]),
+        node_c: NodeNetworkInfo(interfaces=[ethernet_interface]),
+    }
+
+    for n in (node_a, node_b, node_c):
+        topology.add_node(n)
+
+    rdma_pairs = [
+        (node_a, node_b, 3),
+        (node_b, node_a, 3),
+        (node_b, node_c, 4),
+        (node_c, node_b, 4),
+        (node_a, node_c, 5),
+        (node_c, node_a, 5),
+    ]
+    for src, sink, iface in rdma_pairs:
+        topology.add_connection(
+            Connection(source=src, sink=sink, edge=create_rdma_connection(iface))
+        )
+
+    socket_pairs = [
+        (node_a, node_b),
+        (node_b, node_c),
+        (node_c, node_a),
+        (node_a, node_c),
+        (node_b, node_a),
+        (node_c, node_b),
+    ]
+    for src, sink in socket_pairs:
+        topology.add_connection(Connection(source=src, sink=sink, edge=ethernet_conn))
+
+    return topology, node_a, node_b, node_c, node_network
+
+
+def test_place_mlx_jaccl_rejects_when_a_node_has_rdma_ctl_disabled(
+    model_card: ModelCard,
+):
+    # arrange
+    model_card = model_card.model_copy(
+        update={"n_layers": 12, "storage_size": Memory.from_bytes(1500)}
+    )
+    topology, node_a, node_b, node_c, node_network = _build_three_node_rdma_topology()
+    node_memory = {
+        node_a: create_node_memory(500),
+        node_b: create_node_memory(500),
+        node_c: create_node_memory(500),
+    }
+    node_rdma_ctl = {
+        node_a: NodeRdmaCtlStatus(enabled=True),
+        node_b: NodeRdmaCtlStatus(enabled=True),
+        node_c: NodeRdmaCtlStatus(enabled=False),
+    }
+    cic = PlaceInstance(
+        sharding=Sharding.Tensor,
+        instance_meta=InstanceMeta.MlxJaccl,
+        command_id=CommandId(),
+        model_card=model_card,
+        min_nodes=3,
+    )
+
+    # act / assert
+    with pytest.raises(
+        ValueError, match="Requested RDMA \\(MlxJaccl\\) but no RDMA-connected cycles"
+    ):
+        place_instance(
+            cic,
+            topology,
+            {},
+            node_memory,
+            node_network,
+            node_rdma_ctl=node_rdma_ctl,
+        )
+
+
+def test_place_mlx_jaccl_rejects_when_node_rdma_ctl_missing(model_card: ModelCard):
+    """A node with no observed rdma_ctl status must not participate in RDMA placement."""
+    # arrange
+    model_card = model_card.model_copy(
+        update={"n_layers": 12, "storage_size": Memory.from_bytes(1500)}
+    )
+    topology, node_a, node_b, node_c, node_network = _build_three_node_rdma_topology()
+    node_memory = {
+        node_a: create_node_memory(500),
+        node_b: create_node_memory(500),
+        node_c: create_node_memory(500),
+    }
+    # node_c has no rdma_ctl entry at all
+    node_rdma_ctl = {
+        node_a: NodeRdmaCtlStatus(enabled=True),
+        node_b: NodeRdmaCtlStatus(enabled=True),
+    }
+    cic = PlaceInstance(
+        sharding=Sharding.Tensor,
+        instance_meta=InstanceMeta.MlxJaccl,
+        command_id=CommandId(),
+        model_card=model_card,
+        min_nodes=3,
+    )
+
+    # act / assert
+    with pytest.raises(ValueError):
+        place_instance(
+            cic,
+            topology,
+            {},
+            node_memory,
+            node_network,
+            node_rdma_ctl=node_rdma_ctl,
+        )
+
+
 def _make_task(
     instance_id: InstanceId,
     status: TaskStatus = TaskStatus.Running,

src/exo/routing/event_router.py

@@ -80,9 +80,6 @@ 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:
@@ -98,6 +95,7 @@ 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:
@@ -105,12 +103,12 @@ class EventRouter:
                 if event.origin != self.session_id.master_node_id:
                     continue
 
-                self.event_buffer.ingest(event.origin_idx, event.event)
+                buf.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 = self.event_buffer.drain_indexed()
+                drained = buf.drain_indexed()
                 if drained:
                     self._nack_attempts = 0
                     if self._nack_cancel_scope:
@@ -121,9 +119,7 @@ class EventRouter:
                     or self._nack_cancel_scope.cancel_called
                 ):
                     # Request the next index.
-                    self._tg.start_soon(
-                        self._nack_request, self.event_buffer.next_idx_to_release
-                    )
+                    self._tg.start_soon(self._nack_request, buf.next_idx_to_release)
                     continue
 
                 for idx, event in drained:

src/exo/routing/snapshot_receiver.py

@@ -1,109 +0,0 @@
-"""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,28 +141,3 @@ 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

@@ -1,151 +0,0 @@
-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

@@ -1,37 +0,0 @@
-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/topics.py

@@ -8,7 +8,6 @@ from exo.shared.types.events import (
     GlobalForwarderEvent,
     LocalForwarderEvent,
 )
-from exo.shared.types.snapshots import SnapshotChunk
 from exo.utils.pydantic_ext import FrozenModel
 
 
@@ -50,6 +49,3 @@ CONNECTION_MESSAGES = TypedTopic(
 DOWNLOAD_COMMANDS = TypedTopic(
     "download_commands", PublishPolicy.Always, ForwarderDownloadCommand
 )
-SNAPSHOT_RESPONSES = TypedTopic(
-    "snapshot_responses", PublishPolicy.Always, SnapshotChunk
-)

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,14 +41,7 @@ from exo.shared.types.profiling import (
     ThunderboltBridgeStatus,
 )
 from exo.shared.types.state import State
-from exo.shared.types.tasks import (
-    ImageEdits,
-    ImageGeneration,
-    Task,
-    TaskId,
-    TaskStatus,
-    TextGeneration,
-)
+from exo.shared.types.tasks import Task, TaskId, TaskStatus
 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,6 +66,18 @@ from exo.utils.info_gatherer.info_gatherer import (
 )
 
 
+def _is_rdma_ctl_enabled(
+    node_id: NodeId, node_rdma_ctl: Mapping[NodeId, NodeRdmaCtlStatus]
+) -> bool:
+    """A node is RDMA-capable only if rdma_ctl status has been observed as enabled.
+
+    Missing entries default to ``False`` — if we have not yet observed (or the node
+    cannot run) ``rdma_ctl``, it must not participate in an RDMA-backed instance.
+    """
+    status = node_rdma_ctl.get(node_id)
+    return status is not None and status.enabled
+
+
 def event_apply(event: Event, state: State) -> State:
     """Apply an event to state."""
     match event:
@@ -79,12 +85,15 @@ 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():
@@ -99,8 +108,6 @@ 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():
@@ -165,32 +172,10 @@ 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})
 
 
@@ -427,6 +412,9 @@ def apply_node_gathered_info(event: NodeGatheredInfo, state: State) -> State:
                 for nid in state.node_thunderbolt
                 for tb_ident in state.node_thunderbolt[nid].interfaces
             }
+            source_is_rdma_enabled = _is_rdma_ctl_enabled(
+                event.node_id, state.node_rdma_ctl
+            )
             as_rdma_conns = [
                 Connection(
                     source=event.node_id,
@@ -439,6 +427,10 @@ def apply_node_gathered_info(event: NodeGatheredInfo, state: State) -> State:
                 for tb_conn in info.conns
                 if tb_conn.source_uuid in conn_map
                 if tb_conn.sink_uuid in conn_map
+                if source_is_rdma_enabled
+                and _is_rdma_ctl_enabled(
+                    conn_map[tb_conn.sink_uuid][0], state.node_rdma_ctl
+                )
             ]
             topology.replace_all_out_rdma_connections(event.node_id, as_rdma_conns)
         case ThunderboltBridgeInfo():
@@ -462,6 +454,12 @@ def apply_node_gathered_info(event: NodeGatheredInfo, state: State) -> State:
                 **state.node_rdma_ctl,
                 event.node_id: NodeRdmaCtlStatus(enabled=info.enabled),
             }
+            # If RDMA just got disabled on this node, drop any RDMA edges touching it
+            # so placement / topology consumers cannot pick a disabled node for an
+            # RDMA-backed instance. (Edges will repopulate on the next
+            # MacThunderboltConnections poll once both endpoints are enabled again.)
+            if not info.enabled:
+                topology.remove_all_rdma_connections_touching(event.node_id)
 
     return state.model_copy(update=update)
 
@@ -477,3 +475,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/constants.py

@@ -68,7 +68,12 @@ DASHBOARD_DIR = (
 # Log files (data/logs or cache)
 EXO_LOG_DIR = EXO_CACHE_HOME / "exo_log"
 EXO_LOG = EXO_LOG_DIR / "exo.log"
+EXO_RUNNER_LOG_DIR = EXO_LOG_DIR / "runner_log"
+EXO_RUNNER_STDOUT_LOG = EXO_RUNNER_LOG_DIR / "stdout.log"
+EXO_RUNNER_STDERR_LOG = EXO_RUNNER_LOG_DIR / "stderr.log"
+
 EXO_TEST_LOG = EXO_CACHE_HOME / "exo_test.log"
+EXO_PID_FILE = EXO_CACHE_HOME / "exo.pid"
 
 # Identity (config)
 EXO_NODE_ID_KEYPAIR = EXO_CONFIG_HOME / "node_id.keypair"

src/exo/shared/models/model_cards.py

@@ -39,7 +39,57 @@ _BUILTIN_CARD_DIRS = [
     Path(RESOURCES_DIR) / "image_model_cards",
 ]
 
-_card_cache: dict[ModelId, "ModelCard"] = {}
+
+class _CardCache:
+    def __init__(self):
+        self.cc: dict[ModelId, "ModelCard"] = {}
+
+    def get(self, model_id: ModelId) -> "ModelCard | None":
+        return self.cc.get(model_id)
+
+    async def save(self, card: "ModelCard"):
+        self.cc[card.model_id] = card
+        try:
+            await card.save_to_custom_dir()
+        except OSError as e:
+            logger.warning(f"failed to save custom model card ({e.strerror})")
+
+    async def pop(self, model_id: ModelId) -> "ModelCard | None":
+        """Delete a user-added custom model card. Returns True if deleted."""
+        card_path = _custom_cards_dir / (ModelId(model_id).normalize() + ".toml")
+        try:
+            if await card_path.exists():
+                await card_path.unlink()
+                return self.cc.pop(model_id, None)
+        except OSError as e:
+            logger.warning(f"failed to delete custom model card ({e.strerror})")
+
+    async def list_all(self) -> list["ModelCard"]:
+        if len(self.cc) == 0:
+            await self.refresh()
+        if EXO_ENABLE_IMAGE_MODELS:
+            return list(self.cc.values())
+        return [c for c in self.cc.values() if not _is_image_card(c)]
+
+    async def _load_cards_from_dir(self, directory: Path, *, is_custom: bool) -> None:
+        """Load all TOML model cards from a directory into the cache."""
+        async for toml_file in directory.rglob("*.toml"):
+            try:
+                card = await ModelCard.load_from_path(toml_file)
+                if is_custom:
+                    card = card.model_copy(update={"is_custom": True})
+                if self.get(card.model_id) is None:
+                    self.cc[card.model_id] = card
+            except (ValidationError, TOMLKitError):
+                pass
+
+    async def refresh(self) -> None:
+        for path in _BUILTIN_CARD_DIRS:
+            await self._load_cards_from_dir(path, is_custom=False)
+        await self._load_cards_from_dir(_custom_cards_dir, is_custom=True)
+
+
+card_cache = _CardCache()
 
 
 def detect_vision_from_config(model_id: ModelId) -> "VisionCardConfig | None":
@@ -59,42 +109,10 @@ def detect_vision_from_config(model_id: ModelId) -> "VisionCardConfig | None":
     return None
 
 
-async def _load_cards_from_dir(directory: Path, *, is_custom: bool) -> None:
-    """Load all TOML model cards from a directory into the cache."""
-    async for toml_file in directory.rglob("*.toml"):
-        try:
-            card = await ModelCard.load_from_path(toml_file)
-            if is_custom:
-                card = card.model_copy(update={"is_custom": True})
-            if card.model_id not in _card_cache:
-                _card_cache[card.model_id] = card
-        except (ValidationError, TOMLKitError):
-            pass
-
-
-async def _refresh_card_cache() -> None:
-    for path in _BUILTIN_CARD_DIRS:
-        await _load_cards_from_dir(path, is_custom=False)
-    await _load_cards_from_dir(_custom_cards_dir, is_custom=True)
-
-
 def _is_image_card(card: "ModelCard") -> bool:
     return any(t in (ModelTask.TextToImage, ModelTask.ImageToImage) for t in card.tasks)
 
 
-def get_card(model_id: ModelId) -> "ModelCard | None":
-    """Look up a single model card from the cache by ID."""
-    return _card_cache.get(model_id)
-
-
-async def get_model_cards() -> list["ModelCard"]:
-    if len(_card_cache) == 0:
-        await _refresh_card_cache()
-    if EXO_ENABLE_IMAGE_MODELS:
-        return list(_card_cache.values())
-    return [c for c in _card_cache.values() if not _is_image_card(c)]
-
-
 class ModelTask(str, Enum):
     TextGeneration = "TextGeneration"
     TextToImage = "TextToImage"
@@ -196,14 +214,13 @@ class ModelCard(FrozenModel):
     # Is it okay that model card.load defaults to network access if the card doesn't exist? do we want to be more explicit here?
     @staticmethod
     async def load(model_id: ModelId) -> "ModelCard":
-        if model_id not in _card_cache:
-            await _refresh_card_cache()
-        if (mc := _card_cache.get(model_id)) is not None:
+        if card_cache.get(model_id) is None:
+            await card_cache.refresh()
+        if (mc := card_cache.get(model_id)) is not None:
             return mc
 
         mc = await ModelCard.fetch_from_hf(model_id)
         await mc.save_to_custom_dir()
-        _card_cache[model_id] = mc
         return mc
 
     @staticmethod
@@ -233,21 +250,6 @@ class ModelCard(FrozenModel):
         )
 
 
-def add_to_card_cache(card: "ModelCard") -> None:
-    """Add or update a model card in the in-memory cache."""
-    _card_cache[card.model_id] = card
-
-
-async def delete_custom_card(model_id: ModelId) -> bool:
-    """Delete a user-added custom model card. Returns True if deleted."""
-    card_path = _custom_cards_dir / (ModelId(model_id).normalize() + ".toml")
-    if await card_path.exists():
-        await card_path.unlink()
-        _card_cache.pop(model_id, None)
-        return True
-    return False
-
-
 class ConfigData(BaseModel):
     model_config = {"extra": "ignore"}  # Allow unknown fields
 

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

@@ -1,85 +0,0 @@
-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_apply/test_apply_rdma_gating.py

@@ -0,0 +1,231 @@
+from datetime import datetime, timezone
+
+from exo.shared.apply import apply_node_gathered_info
+from exo.shared.topology import Topology
+from exo.shared.types.common import NodeId
+from exo.shared.types.events import NodeGatheredInfo
+from exo.shared.types.profiling import (
+    NodeRdmaCtlStatus,
+    NodeThunderboltInfo,
+)
+from exo.shared.types.state import State
+from exo.shared.types.thunderbolt import ThunderboltConnection, ThunderboltIdentifier
+from exo.shared.types.topology import RDMAConnection
+from exo.utils.info_gatherer.info_gatherer import (
+    MacThunderboltConnections,
+    RdmaCtlStatus,
+)
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _make_state_with_thunderbolt_idents(
+    *node_ids_and_uuids: tuple[NodeId, str, str],
+    rdma_ctl: dict[NodeId, NodeRdmaCtlStatus] | None = None,
+) -> State:
+    """Build a State with Thunderbolt identifiers per node so the apply MacThunderboltConnections
+    case can resolve uuid -> (node, iface)."""
+    node_thunderbolt = {
+        nid: NodeThunderboltInfo(
+            interfaces=[ThunderboltIdentifier(rdma_interface=iface, domain_uuid=uuid)]
+        )
+        for nid, uuid, iface in node_ids_and_uuids
+    }
+    return State(
+        node_thunderbolt=node_thunderbolt,
+        node_rdma_ctl=rdma_ctl or {},
+    )
+
+
+def _has_rdma_edge(topology: Topology, source: NodeId, sink: NodeId) -> bool:
+    return any(
+        isinstance(edge, RDMAConnection)
+        for edge in topology.get_all_connections_between(source, sink)
+    )
+
+
+def test_mac_thunderbolt_connections_emits_rdma_when_both_endpoints_enabled():
+    node_a = NodeId()
+    node_b = NodeId()
+    state = _make_state_with_thunderbolt_idents(
+        (node_a, "uuid-a", "rdma_en1"),
+        (node_b, "uuid-b", "rdma_en1"),
+        rdma_ctl={
+            node_a: NodeRdmaCtlStatus(enabled=True),
+            node_b: NodeRdmaCtlStatus(enabled=True),
+        },
+    )
+
+    event = NodeGatheredInfo(
+        node_id=node_a,
+        when=_now(),
+        info=MacThunderboltConnections(
+            conns=[ThunderboltConnection(source_uuid="uuid-a", sink_uuid="uuid-b")]
+        ),
+    )
+
+    new_state = apply_node_gathered_info(event, state)
+
+    assert _has_rdma_edge(new_state.topology, node_a, node_b)
+
+
+def test_mac_thunderbolt_connections_skips_rdma_when_source_rdma_ctl_disabled():
+    node_a = NodeId()
+    node_b = NodeId()
+    state = _make_state_with_thunderbolt_idents(
+        (node_a, "uuid-a", "rdma_en1"),
+        (node_b, "uuid-b", "rdma_en1"),
+        rdma_ctl={
+            node_a: NodeRdmaCtlStatus(enabled=False),
+            node_b: NodeRdmaCtlStatus(enabled=True),
+        },
+    )
+
+    event = NodeGatheredInfo(
+        node_id=node_a,
+        when=_now(),
+        info=MacThunderboltConnections(
+            conns=[ThunderboltConnection(source_uuid="uuid-a", sink_uuid="uuid-b")]
+        ),
+    )
+
+    new_state = apply_node_gathered_info(event, state)
+
+    assert not _has_rdma_edge(new_state.topology, node_a, node_b)
+
+
+def test_mac_thunderbolt_connections_skips_rdma_when_sink_rdma_ctl_disabled():
+    node_a = NodeId()
+    node_b = NodeId()
+    state = _make_state_with_thunderbolt_idents(
+        (node_a, "uuid-a", "rdma_en1"),
+        (node_b, "uuid-b", "rdma_en1"),
+        rdma_ctl={
+            node_a: NodeRdmaCtlStatus(enabled=True),
+            node_b: NodeRdmaCtlStatus(enabled=False),
+        },
+    )
+
+    event = NodeGatheredInfo(
+        node_id=node_a,
+        when=_now(),
+        info=MacThunderboltConnections(
+            conns=[ThunderboltConnection(source_uuid="uuid-a", sink_uuid="uuid-b")]
+        ),
+    )
+
+    new_state = apply_node_gathered_info(event, state)
+
+    assert not _has_rdma_edge(new_state.topology, node_a, node_b)
+
+
+def test_mac_thunderbolt_connections_skips_rdma_when_rdma_ctl_status_missing():
+    """Missing rdma_ctl status defaults to not-enabled — node is RDMA-incapable."""
+    node_a = NodeId()
+    node_b = NodeId()
+    state = _make_state_with_thunderbolt_idents(
+        (node_a, "uuid-a", "rdma_en1"),
+        (node_b, "uuid-b", "rdma_en1"),
+        rdma_ctl={
+            node_a: NodeRdmaCtlStatus(enabled=True),
+            # node_b intentionally absent
+        },
+    )
+
+    event = NodeGatheredInfo(
+        node_id=node_a,
+        when=_now(),
+        info=MacThunderboltConnections(
+            conns=[ThunderboltConnection(source_uuid="uuid-a", sink_uuid="uuid-b")]
+        ),
+    )
+
+    new_state = apply_node_gathered_info(event, state)
+
+    assert not _has_rdma_edge(new_state.topology, node_a, node_b)
+
+
+def test_rdma_ctl_status_disabled_purges_existing_rdma_edges():
+    """When a node reports rdma_ctl disabled, all RDMA edges touching it must be removed."""
+    node_a = NodeId()
+    node_b = NodeId()
+
+    # Start with both nodes RDMA-enabled and existing RDMA edges in the topology.
+    state = _make_state_with_thunderbolt_idents(
+        (node_a, "uuid-a", "rdma_en1"),
+        (node_b, "uuid-b", "rdma_en1"),
+        rdma_ctl={
+            node_a: NodeRdmaCtlStatus(enabled=True),
+            node_b: NodeRdmaCtlStatus(enabled=True),
+        },
+    )
+    state = apply_node_gathered_info(
+        NodeGatheredInfo(
+            node_id=node_a,
+            when=_now(),
+            info=MacThunderboltConnections(
+                conns=[ThunderboltConnection(source_uuid="uuid-a", sink_uuid="uuid-b")]
+            ),
+        ),
+        state,
+    )
+    state = apply_node_gathered_info(
+        NodeGatheredInfo(
+            node_id=node_b,
+            when=_now(),
+            info=MacThunderboltConnections(
+                conns=[ThunderboltConnection(source_uuid="uuid-b", sink_uuid="uuid-a")]
+            ),
+        ),
+        state,
+    )
+    assert _has_rdma_edge(state.topology, node_a, node_b)
+    assert _has_rdma_edge(state.topology, node_b, node_a)
+
+    # Now node_a flips to rdma_ctl disabled — both directions of RDMA edge must drop.
+    state = apply_node_gathered_info(
+        NodeGatheredInfo(
+            node_id=node_a, when=_now(), info=RdmaCtlStatus(enabled=False)
+        ),
+        state,
+    )
+
+    assert not _has_rdma_edge(state.topology, node_a, node_b)
+    assert not _has_rdma_edge(state.topology, node_b, node_a)
+    assert state.node_rdma_ctl[node_a].enabled is False
+
+
+def test_topology_remove_all_rdma_connections_touching_keeps_socket_edges():
+    """Purging RDMA edges for a disabled node must not affect non-RDMA edges."""
+    from exo.shared.types.multiaddr import Multiaddr
+    from exo.shared.types.topology import Connection, SocketConnection
+
+    topology = Topology()
+    node_a = NodeId()
+    node_b = NodeId()
+    topology.add_node(node_a)
+    topology.add_node(node_b)
+    topology.add_connection(
+        Connection(
+            source=node_a,
+            sink=node_b,
+            edge=RDMAConnection(
+                source_rdma_iface="rdma_en1", sink_rdma_iface="rdma_en1"
+            ),
+        )
+    )
+    socket_edge = SocketConnection(
+        sink_multiaddr=Multiaddr(address="/ip4/10.0.0.1/tcp/8000")
+    )
+    topology.add_connection(Connection(source=node_a, sink=node_b, edge=socket_edge))
+
+    topology.remove_all_rdma_connections_touching(node_a)
+
+    assert not _has_rdma_edge(topology, node_a, node_b)
+    # Socket edge survives.
+    assert any(
+        isinstance(edge, SocketConnection)
+        for edge in topology.get_all_connections_between(node_a, node_b)
+    )

src/exo/shared/tests/test_state_serialization.py

@@ -25,7 +25,6 @@ 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/topology.py

@@ -169,6 +169,22 @@ class Topology:
         for conn in new_connections:
             self.add_connection(conn)
 
+    def remove_all_rdma_connections_touching(self, node_id: NodeId) -> None:
+        """Remove every RDMA edge incident to ``node_id`` (incoming or outgoing)."""
+        if node_id not in self._vertex_indices:
+            return
+        rx_idx = self._vertex_indices[node_id]
+        rdma_edge_idxs = [
+            edge_idx
+            for edge_idx in (
+                *self._graph.out_edge_indices(rx_idx),
+                *self._graph.in_edge_indices(rx_idx),
+            )
+            if isinstance(self._graph.get_edge_data_by_index(edge_idx), RDMAConnection)
+        ]
+        for edge_idx in rdma_edge_idxs:
+            self._graph.remove_edge_from_index(edge_idx)
+
     def remove_connection(self, conn: Connection) -> None:
         if (
             conn.source not in self._vertex_indices

src/exo/shared/types/commands.py

@@ -67,12 +67,6 @@ 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
@@ -112,7 +106,6 @@ DownloadCommand = StartDownload | DeleteDownload | CancelDownload
 Command = (
     TestCommand
     | RequestEventLog
-    | RequestSnapshot
     | TextGeneration
     | ImageGeneration
     | ImageEdits

src/exo/shared/types/snapshots.py

@@ -1,54 +0,0 @@
-"""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,9 +5,9 @@ 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.chunks import InputImageChunk
-from exo.shared.types.common import CommandId, NodeId
+from exo.shared.types.common import ModelId, NodeId
 from exo.shared.types.instance_link import InstanceLink, InstanceLinkId
 from exo.shared.types.profiling import (
     DiskUsage,
@@ -42,16 +42,10 @@ 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)
@@ -72,6 +66,9 @@ 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
+    custom_model_cards: Mapping[ModelId, ModelCard] = {}
+
     @field_serializer("topology", mode="plain")
     def _encode_topology(self, value: Topology) -> TopologySnapshot:
         return value.to_snapshot()

src/exo/shared/types/text_generation.py

@@ -135,9 +135,9 @@ class TextGenerationTaskParams(BaseModel, frozen=True):
     prefill_endpoint: str | None = None
 
     def with_card_sampling_defaults(self) -> "TextGenerationTaskParams":
-        from exo.shared.models.model_cards import get_card
+        from exo.shared.models import model_cards
 
-        card = get_card(self.model)
+        card = model_cards.card_cache.get(self.model)
         if card is None:
             return self
 

src/exo/utils/async_process.py

@@ -0,0 +1,290 @@
+from __future__ import annotations
+
+import contextlib
+import faulthandler
+import multiprocessing as mp
+import os
+import sys
+from collections.abc import Callable, Iterable, Mapping
+from multiprocessing.process import BaseProcess
+from multiprocessing.resource_sharer import DupFd
+from typing import final
+
+from anyio import (
+    TASK_STATUS_IGNORED,
+    BrokenResourceError,
+    CancelScope,
+    ClosedResourceError,
+    Event,
+    create_task_group,
+    move_on_after,
+    sleep,
+    wait_readable,
+)
+from anyio.abc import TaskStatus
+from loguru import logger
+
+from exo.utils.channels import Receiver, Sender, channel
+
+_STDOUT_FD = 1
+_STDERR_FD = 2
+_READ_CHUNK_SIZE = 64 * 1024
+_TERMINATE_GRACE_SECONDS = 10.0
+_TERMINATE_RETRY_GRACE_SECONDS = 2.0
+_TERMINATE_ATTEMPTS = 10
+_KILL_GRACE_SECONDS = 5.0
+
+
+@final
+class AsyncProcess:
+    def __init__(
+        self,
+        target: Callable[..., object] | None = None,
+        name: str | None = None,
+        args: Iterable[object] = (),
+        kwargs: Mapping[str, object] | None = None,
+        *,
+        daemon: bool | None = None,
+    ) -> None:
+        # setup state
+        self._target = target
+        self._name = name
+        self._args = args
+        self._kwargs = kwargs
+        self._daemon = daemon
+
+        # lifecycle state
+        self._process: BaseProcess | None = None
+        self._pid: int | None = None
+        self._stdout_tx, self._stdout_rx = channel[bytes]()
+        self._stderr_tx, self._stderr_rx = channel[bytes]()
+        self._started = Event()
+        self._done = Event()
+        self._run_cancel_scope: CancelScope | None = None
+        self._start_error: BaseException | None = None
+        self._exitcode: int | None = None
+
+    async def run(self, *, task_status: TaskStatus[None] = TASK_STATUS_IGNORED) -> None:
+        if self._run_cancel_scope is not None or self._done.is_set():
+            raise RuntimeError("process has already been started")
+
+        stdout_read_fd: int | None = None
+        stdout_write_fd: int | None = None
+        stderr_read_fd: int | None = None
+        stderr_write_fd: int | None = None
+
+        def cleanup_stdio_fd() -> None:
+            nonlocal stdout_read_fd, stdout_write_fd, stderr_read_fd, stderr_write_fd
+            stdout_read_fd = _close_fd(stdout_read_fd)
+            stdout_write_fd = _close_fd(stdout_write_fd)
+            stderr_read_fd = _close_fd(stderr_read_fd)
+            stderr_write_fd = _close_fd(stderr_write_fd)
+
+        try:
+            with CancelScope() as run_cancel_scope:
+                self._run_cancel_scope = run_cancel_scope
+                stdout_read_fd, stdout_write_fd = os.pipe()
+                stderr_read_fd, stderr_write_fd = os.pipe()
+
+                process = mp.Process(
+                    target=_run_with_captured_stdio,
+                    name=self._name,
+                    args=(
+                        DupFd(stdout_write_fd),
+                        DupFd(stderr_write_fd),
+                        self._target,
+                        *self._args,
+                    ),
+                    kwargs={} if self._kwargs is None else self._kwargs,
+                    daemon=self._daemon,
+                )
+                process.start()
+                pid = process.pid
+                if pid is None:
+                    raise RuntimeError("started process has no pid")
+
+                # important to close parent write-side FD to prevent hangs
+                stdout_write_fd = _close_fd(stdout_write_fd)
+                stderr_write_fd = _close_fd(stderr_write_fd)
+
+                self._process = process
+                self._pid = pid
+                self._started.set()
+
+                async with create_task_group() as tg:
+                    tg.start_soon(_drain_fd, stdout_read_fd, self._stdout_tx)
+                    stdout_read_fd = None
+                    tg.start_soon(_drain_fd, stderr_read_fd, self._stderr_tx)
+                    stderr_read_fd = None
+                    task_status.started()
+                    await self.wait()
+        except BaseException as exc:
+            if not self._started.is_set():
+                self._start_error = exc
+                self._started.set()
+            raise
+        finally:
+            try:
+                with CancelScope(shield=True):
+                    await self._terminate_if_still_alive()
+            finally:
+                cleanup_stdio_fd()
+                for tx in (self._stdout_tx, self._stderr_tx):
+                    with contextlib.suppress(Exception):
+                        await tx.aclose()
+                if self._process is not None:
+                    with contextlib.suppress(ValueError):
+                        self._process.close()
+                self._run_cancel_scope = None
+                self._done.set()
+
+    async def stop(self) -> None:
+        if self._run_cancel_scope is None and not self._done.is_set():
+            raise RuntimeError("process has not been started")
+        if self._run_cancel_scope is not None:
+            self._run_cancel_scope.cancel()
+        await self._done.wait()
+
+    async def aclose(self) -> None:
+        await self.stop()
+
+    async def wait(self) -> int:
+        if self._exitcode is not None:
+            return self._exitcode
+
+        await self._started.wait()
+        if self._start_error is not None:
+            raise self._start_error
+        assert self._process is not None
+
+        while True:
+            exitcode = self.exitcode
+            if exitcode is not None:
+                return exitcode
+            await sleep(0.01)
+
+    @property
+    def pid(self) -> int:
+        if self._pid is None:
+            raise RuntimeError("process has not been started")
+        return self._pid
+
+    @property
+    def exitcode(self) -> int | None:
+        if self._exitcode is not None:
+            return self._exitcode
+        if self._process is None:
+            return None
+
+        with contextlib.suppress(ValueError):
+            exitcode = self._process.exitcode
+            if exitcode is not None:
+                self._exitcode = exitcode
+            return exitcode
+        return None
+
+    def is_alive(self) -> bool:
+        if self._process is None:
+            return False
+
+        with contextlib.suppress(ValueError):
+            return self._process.is_alive()
+        return False
+
+    # TODO: maybe in the future if needed, create stdin that is also installed,
+    #       and a ByteSendStream handle is provided for it :)
+
+    @property
+    def stdout(self) -> Receiver[bytes]:
+        return self._stdout_rx
+
+    @property
+    def stderr(self) -> Receiver[bytes]:
+        return self._stderr_rx
+
+    async def _terminate_if_still_alive(self) -> None:
+        process = self._process
+        if process is None:
+            return
+
+        if self.exitcode is not None:
+            return
+
+        with contextlib.suppress(ValueError):
+            if not process.is_alive():
+                return
+
+            logger.warning("Child process didn't shut down successfully, terminating")
+            process.terminate()
+            with move_on_after(_TERMINATE_GRACE_SECONDS):
+                await self.wait()
+
+            if self.exitcode is not None or not process.is_alive():
+                logger.warning("Terminated nicely in the first attempt!")
+                return
+
+            for attempt in range(2, _TERMINATE_ATTEMPTS + 1):
+                process.terminate()
+                with move_on_after(_TERMINATE_RETRY_GRACE_SECONDS):
+                    await self.wait()
+
+                if self.exitcode is not None or not process.is_alive():
+                    logger.warning(f"That took {attempt} attempts :)")
+                    return
+
+            logger.critical("Child process didn't respond to SIGTERM, killing")
+            j = 0
+            while True:
+                process.kill()
+                with move_on_after(_KILL_GRACE_SECONDS):
+                    await self.wait()
+                j += 1
+                if self.exitcode is not None or not process.is_alive():
+                    break
+            logger.warning(f"That took {j} attempts :(")
+
+
+# Spawn-mode multiprocessing requires a module-level target that can be pickled.
+def _run_with_captured_stdio(
+    stdout: DupFd,
+    stderr: DupFd,
+    target: Callable[..., object] | None,
+    *target_args: object,
+    **target_kwargs: object,
+) -> None:
+    stdout_fd = stdout.detach()
+    stderr_fd = stderr.detach()
+
+    try:
+        os.dup2(stdout_fd, _STDOUT_FD)
+        os.dup2(stderr_fd, _STDERR_FD)
+    finally:
+        for fd in (stdout_fd, stderr_fd):
+            if fd not in (_STDOUT_FD, _STDERR_FD):
+                _close_fd(fd)
+
+    faulthandler.enable(file=sys.stderr, all_threads=True)
+    if target is not None:
+        target(*target_args, **target_kwargs)
+
+
+async def _drain_fd(fd: int, tx: Sender[bytes]) -> None:
+    try:
+        while True:
+            await wait_readable(fd)
+            chunk = os.read(fd, _READ_CHUNK_SIZE)
+            if not chunk:
+                return
+            await tx.send(chunk)
+    except (BrokenPipeError, BrokenResourceError, ClosedResourceError):
+        pass
+    finally:
+        _close_fd(fd)
+        await tx.aclose()
+
+
+def _close_fd(fd: int | None) -> None:
+    if fd is None:
+        return
+    with contextlib.suppress(OSError):
+        os.close(fd)

src/exo/utils/daemon.py

@@ -0,0 +1,28 @@
+import os
+import sys
+
+_STDIN_FD = 0
+_STDOUT_FD = 1
+_STDERR_FD = 2
+
+
+def detach_stdio_to_devnull() -> None:
+    """Redirect process stdio file descriptors to /dev/null."""
+
+    for stream in (sys.stdout, sys.stderr, sys.__stdout__, sys.__stderr__):
+        if stream is not None:
+            stream.flush()
+
+    stdin_fd = os.open(os.devnull, os.O_RDONLY)
+    stdout_fd = os.open(os.devnull, os.O_WRONLY)
+    stderr_fd = os.open(os.devnull, os.O_WRONLY)
+
+    try:
+        # dup2 closes the target fd first, but leaves the source fd open.
+        os.dup2(stdin_fd, _STDIN_FD)
+        os.dup2(stdout_fd, _STDOUT_FD)
+        os.dup2(stderr_fd, _STDERR_FD)
+    finally:
+        for fd in (stdin_fd, stdout_fd, stderr_fd):
+            if fd not in (_STDIN_FD, _STDOUT_FD, _STDERR_FD):
+                os.close(fd)

src/exo/utils/event_buffer.py

@@ -47,18 +47,6 @@ 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/pidfile.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+import os
+from typing import Final
+
+from exo_pyo3_bindings import Pidfile, PidfileError
+
+from exo.shared.constants import EXO_PID_FILE
+
+_PIDFILE_MODE: Final = 0o600
+
+
+class PidfileLockError(RuntimeError):
+    pass
+
+
+def acquire_exo_pidfile() -> Pidfile:
+    path = EXO_PID_FILE
+    os.makedirs(os.path.dirname(path), exist_ok=True)
+    try:
+        pidfile = Pidfile(path, _PIDFILE_MODE)
+        pidfile.write()
+    except (OSError, PidfileError) as exception:
+        raise PidfileLockError(
+            f"Failed to acquire EXO pidfile at {path}: {exception}"
+        ) from exception
+
+    return pidfile

src/exo/utils/power_sampler.py

@@ -19,19 +19,21 @@ class PowerSampler:
     ):
         self._get_node_system = get_node_system
         self._interval = interval
-        self._samples: defaultdict[NodeId, list[SystemPerformanceProfile]] = (
-            defaultdict(list)
-        )
+        self._samples: defaultdict[
+            NodeId, list[tuple[float, SystemPerformanceProfile]]
+        ] = defaultdict(list)
         self._start_time: float | None = None
         self._stopped = False
 
-    def _take_sample(self) -> None:
+    def _take_sample(self, t_rel: float | None = None) -> None:
+        assert self._start_time is not None
+        ts = t_rel if t_rel is not None else time.perf_counter() - self._start_time
         for node_id, profile in self._get_node_system().items():
-            self._samples[node_id].append(profile)
+            self._samples[node_id].append((ts, profile))
 
     async def run(self) -> None:
         self._start_time = time.perf_counter()
-        self._take_sample()
+        self._take_sample(t_rel=0.0)
         while not self._stopped:
             await anyio.sleep(self._interval)
             self._take_sample()
@@ -39,26 +41,51 @@ class PowerSampler:
     def result(self) -> PowerUsage:
         self._stopped = True
         assert self._start_time is not None, "result() called before run()"
-        self._take_sample()
         elapsed = time.perf_counter() - self._start_time
+        self._take_sample(t_rel=elapsed)
 
         node_stats: list[NodePowerStats] = []
-        for node_id, profiles in self._samples.items():
-            n = len(profiles)
+        total_energy_j = 0.0
+        for node_id, ts_profiles in self._samples.items():
+            n = len(ts_profiles)
             if n == 0:
                 continue
+            node_energy_j = trapezoidal_energy(ts_profiles, elapsed)
+            avg_power_w = node_energy_j / elapsed if elapsed > 0 else 0.0
+            total_energy_j += node_energy_j
             node_stats.append(
                 NodePowerStats(
                     node_id=node_id,
                     samples=n,
-                    avg_sys_power=sum(p.sys_power for p in profiles) / n,
+                    avg_sys_power=avg_power_w,
                 )
             )
 
-        total_avg_sys = sum(ns.avg_sys_power for ns in node_stats)
+        total_avg_sys_w = total_energy_j / elapsed if elapsed > 0 else 0.0
         return PowerUsage(
             elapsed_seconds=elapsed,
             nodes=node_stats,
-            total_avg_sys_power_watts=total_avg_sys,
-            total_energy_joules=total_avg_sys * elapsed,
+            total_avg_sys_power_watts=total_avg_sys_w,
+            total_energy_joules=total_energy_j,
         )
+
+
+def trapezoidal_energy(
+    ts_profiles: list[tuple[float, SystemPerformanceProfile]],
+    elapsed: float,
+) -> float:
+    """Integrate sys_power(t) over the sample window using the trapezoidal rule.
+    First sample is anchored at t=0 and last at t=elapsed (set by `run` /
+    `result`), so the integral spans the full request interval. Falls back to
+    power * elapsed when only one sample exists (constant-power assumption)."""
+    if len(ts_profiles) == 1:
+        return ts_profiles[0][1].sys_power * elapsed
+    energy_j = 0.0
+    for i in range(1, len(ts_profiles)):
+        t_prev, p_prev = ts_profiles[i - 1]
+        t_cur, p_cur = ts_profiles[i]
+        dt = t_cur - t_prev
+        if dt <= 0:
+            continue
+        energy_j += (p_prev.sys_power + p_cur.sys_power) / 2.0 * dt
+    return energy_j

src/exo/utils/tests/conftest.py

@@ -0,0 +1,8 @@
+import multiprocessing as mp
+
+import pytest
+
+
+@pytest.fixture(scope="session", autouse=True)
+def mp_force_spawn():
+    mp.set_start_method("spawn", force=True)

src/exo/utils/tests/test_async_process.py

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

src/exo/utils/tests/test_daemon.py

@@ -0,0 +1,168 @@
+import contextlib
+import os
+from collections.abc import AsyncIterator
+
+import anyio
+import pytest
+from anyio import EndOfStream, create_task_group, fail_after
+
+from exo.utils.async_process import AsyncProcess
+from exo.utils.channels import MpReceiver, MpSender, Receiver, mp_channel
+from exo.utils.daemon import detach_stdio_to_devnull
+
+
+def _write_before_and_after_detach() -> None:
+    os.write(1, b"before stdout\n")
+    os.write(2, b"before stderr\n")
+    detach_stdio_to_devnull()
+    os.write(1, b"after stdout\n")
+    os.write(2, b"after stderr\n")
+
+
+def _write_grandchild_stdio(label: str) -> None:
+    os.write(1, f"{label} stdout\n".encode())
+    os.write(2, f"{label} stderr\n".encode())
+
+
+async def _spawn_grandchild_and_report(
+    result_sender: MpSender[tuple[int, bytes, bytes]],
+    label: str,
+) -> None:
+    result_sender.send(await _collect_spawned_child(label))
+    result_sender.close()
+
+
+async def _collect_spawned_child(label: str) -> tuple[int, bytes, bytes]:
+    process = AsyncProcess(_write_grandchild_stdio, args=(label,))
+    async with _started_process(process):
+        return await _collect_process_output(process)
+
+
+def _detach_stdio_then_spawn_captured_child(
+    result_sender: MpSender[tuple[int, bytes, bytes]],
+) -> None:
+    detach_stdio_to_devnull()
+    anyio.run(_spawn_grandchild_and_report, result_sender, "grandchild")
+
+
+def _detach_stdio_then_spawn_captured_children_sequentially(
+    result_sender: MpSender[list[tuple[int, bytes, bytes]]],
+) -> None:
+    async def run_children() -> list[tuple[int, bytes, bytes]]:
+        results: list[tuple[int, bytes, bytes]] = []
+        for index in range(5):
+            results.append(await _collect_spawned_child(f"grandchild-{index}"))
+        return results
+
+    detach_stdio_to_devnull()
+    result_sender.send(anyio.run(run_children))
+    result_sender.close()
+
+
+async def _collect_stream(stream: Receiver[bytes], output: bytearray) -> None:
+    while True:
+        try:
+            output.extend(await stream.receive())
+        except EndOfStream:
+            return
+
+
+async def _collect_process_output(
+    process: AsyncProcess,
+) -> tuple[int, bytes, bytes]:
+    stdout = bytearray()
+    stderr = bytearray()
+    exitcodes: list[int] = []
+
+    async with create_task_group() as collect_group:
+        collect_group.start_soon(_collect_stream, process.stdout, stdout)
+        collect_group.start_soon(_collect_stream, process.stderr, stderr)
+        exitcodes.append(await process.wait())
+
+    if not exitcodes:
+        raise RuntimeError("process exited without a return code")
+    return exitcodes[0], bytes(stdout), bytes(stderr)
+
+
+@contextlib.asynccontextmanager
+async def _started_process(process: AsyncProcess) -> AsyncIterator[None]:
+    async with create_task_group() as task_group:
+        await task_group.start(process.run)
+        try:
+            yield
+        finally:
+            await process.stop()
+
+
+async def _run_process_and_receive[T](
+    process: AsyncProcess,
+    recv: MpReceiver[T],
+    *,
+    timeout: float,
+) -> tuple[int, T]:
+    async with _started_process(process):
+        with fail_after(timeout):
+            result = await recv.receive_async()
+            exitcode = await process.wait()
+
+    return exitcode, result
+
+
+@pytest.mark.anyio
+async def test_detach_stdio_to_devnull_redirects_stdio_away_from_capture() -> None:
+    process = AsyncProcess(_write_before_and_after_detach)
+
+    async with _started_process(process):
+        exitcode, stdout, stderr = await _collect_process_output(process)
+
+    assert exitcode == 0
+    assert stdout == b"before stdout\n"
+    assert stderr == b"before stderr\n"
+
+
+@pytest.mark.anyio
+async def test_detached_stdio_process_can_spawn_and_capture_child_stdio() -> None:
+    send, recv = mp_channel[tuple[int, bytes, bytes]]()
+    process = AsyncProcess(_detach_stdio_then_spawn_captured_child, args=(send,))
+
+    try:
+        daemonized_parent_exitcode, result = await _run_process_and_receive(
+            process, recv, timeout=5
+        )
+    finally:
+        recv.close()
+
+    child_exitcode, child_stdout, child_stderr = result
+
+    assert daemonized_parent_exitcode == 0
+    assert child_exitcode == 0
+    assert child_stdout == b"grandchild stdout\n"
+    assert child_stderr == b"grandchild stderr\n"
+
+
+@pytest.mark.anyio
+async def test_detached_stdio_process_can_spawn_captured_children_sequentially() -> (
+    None
+):
+    send, recv = mp_channel[list[tuple[int, bytes, bytes]]]()
+    process = AsyncProcess(
+        _detach_stdio_then_spawn_captured_children_sequentially,
+        args=(send,),
+    )
+
+    try:
+        daemonized_parent_exitcode, results = await _run_process_and_receive(
+            process, recv, timeout=10
+        )
+    finally:
+        recv.close()
+
+    assert daemonized_parent_exitcode == 0
+    assert results == [
+        (
+            0,
+            f"grandchild-{index} stdout\n".encode(),
+            f"grandchild-{index} stderr\n".encode(),
+        )
+        for index in range(5)
+    ]

src/exo/utils/tests/test_pidfile.py

@@ -0,0 +1,84 @@
+from __future__ import annotations
+
+import gc
+import os
+import subprocess
+import sys
+import textwrap
+from pathlib import Path
+from typing import Final
+
+import pytest
+
+import exo.utils.pidfile as pidfile
+from exo.utils.pidfile import acquire_exo_pidfile
+
+_CHILD_ACQUIRE_PIDFILE_SCRIPT: Final = textwrap.dedent(
+    """
+    import sys
+    from pathlib import Path
+    from unittest.mock import patch
+
+    import exo.utils.pidfile as pidfile
+    from exo.utils.pidfile import PidfileLockError, acquire_exo_pidfile
+
+    with patch.object(pidfile, "EXO_PID_FILE", Path(sys.argv[1])):
+        try:
+            handle = acquire_exo_pidfile()
+        except PidfileLockError as exception:
+            print(str(exception))
+            raise SystemExit(73) from exception
+
+        del handle
+    """
+)
+
+
+def _use_pidfile_path(monkeypatch: pytest.MonkeyPatch, path: Path) -> None:
+    monkeypatch.setattr(pidfile, "EXO_PID_FILE", path)
+
+
+def _run_child_acquire_pidfile(path: Path) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(
+        [sys.executable, "-c", _CHILD_ACQUIRE_PIDFILE_SCRIPT, str(path)],
+        check=False,
+        capture_output=True,
+        text=True,
+    )
+
+
+def test_acquire_exo_pidfile_writes_current_pid_and_removes_on_drop(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    path = tmp_path / "exo.pid"
+    _use_pidfile_path(monkeypatch, path)
+
+    handle = acquire_exo_pidfile()
+    assert path.read_text() == str(os.getpid())
+
+    del handle
+    gc.collect()
+
+    assert not path.exists()
+
+
+def test_acquire_exo_pidfile_rejects_second_process(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    path = tmp_path / "exo.pid"
+    _use_pidfile_path(monkeypatch, path)
+
+    handle = acquire_exo_pidfile()
+    try:
+        blocked_child = _run_child_acquire_pidfile(path)
+        assert blocked_child.returncode == 73
+        assert "Failed to acquire EXO pidfile" in blocked_child.stdout
+    finally:
+        del handle
+        gc.collect()
+
+    unblocked_child = _run_child_acquire_pidfile(path)
+    assert unblocked_child.returncode == 0
+    assert unblocked_child.stdout == ""

src/exo/utils/tests/test_power_sampler.py

@@ -111,6 +111,36 @@ async def test_empty_state() -> None:
     assert result.total_energy_joules == 0.0
 
 
+def test_trapezoidal_unit_dt_weighting() -> None:
+    """Pure unit test on the integration helper. Crafted samples where the
+    arithmetic mean is wildly wrong vs the time-weighted result."""
+    from exo.utils.power_sampler import trapezoidal_energy
+
+    # 5 s window. Power = 10 W for the first 4.9 s, then 100 W for the last 0.1 s.
+    # Three samples: t=0 W=10, t=4.9 W=10, t=5.0 W=100.
+    samples = [
+        (0.0, _make_profile(10.0)),
+        (4.9, _make_profile(10.0)),
+        (5.0, _make_profile(100.0)),
+    ]
+    energy = trapezoidal_energy(samples, elapsed=5.0)
+    # (10+10)/2 * 4.9 + (10+100)/2 * 0.1 = 49 + 5.5 = 54.5 J
+    assert abs(energy - 54.5) < 1e-9
+    avg = energy / 5.0  # 10.9 W
+    # Arithmetic mean of the three samples would be (10+10+100)/3 ≈ 40 W.
+    # Trapezoidal correctly weights each segment by its dt.
+    assert abs(avg - 10.9) < 1e-9
+
+
+def test_trapezoidal_unit_single_sample() -> None:
+    """One sample: no window to integrate over, so fall back to constant power
+    over the elapsed duration."""
+    from exo.utils.power_sampler import trapezoidal_energy
+
+    samples = [(0.0, _make_profile(42.0))]
+    assert trapezoidal_energy(samples, elapsed=3.0) == 42.0 * 3.0
+
+
 async def test_result_stops_sampling() -> None:
     """Calling result() should stop the sampler's run loop."""
     state: dict[NodeId, SystemPerformanceProfile] = {

src/exo/worker/engines/image/builder.py

@@ -143,6 +143,7 @@ class ImageEngine(Engine):
         Generator[tuple[TaskId, Chunk | FinishedResponse | CancelledResponse]] | None
     ) = field(init=False, default=None)
     queue: deque[ImageTask] = field(init=False, default_factory=deque)
+    _cancelled_tasks: set[TaskId] = field(init=False, default_factory=set)
 
     def warmup(self) -> None:
         image = warmup_image_generator(model=self.image_model)
@@ -168,7 +169,11 @@ class ImageEngine(Engine):
             task = self.queue.popleft()
             self.current_gen = self._run_image_task(task.task_id, task.task_params)
             resp = next(self.current_gen, None)
-        return (resp,) if resp is not None else ()
+        return (
+            (resp,)
+            if resp is not None and _is_primary_output_node(self.shard_metadata)
+            else ()
+        )
 
     def close(self) -> None:
         with contextlib.suppress(NameError, AttributeError):

src/exo/worker/engines/mlx/auto_parallel.py

@@ -71,14 +71,48 @@ if TYPE_CHECKING:
 
 _pending_prefill_sends: list[tuple[mx.array, int, mx.distributed.Group]] = []
 
+_last_dist_op: mx.array | None = None
+
+
+def _link(out: mx.array) -> mx.array:
+    global _last_dist_op
+    if _last_dist_op is not None:
+        out = mx.depends(out, _last_dist_op)
+    mx.async_eval(out)
+    _last_dist_op = out
+    return out
+
+
+def send(x: mx.array, dst: int, group: mx.distributed.Group) -> mx.array:
+    return _link(mx.distributed.send(x, dst, group=group, stream=mx.Device(mx.cpu)))
+
+
+def recv_like(x: mx.array, src: int, group: mx.distributed.Group) -> mx.array:
+    return _link(
+        mx.distributed.recv_like(x, src, group=group, stream=mx.Device(mx.cpu))
+    )
+
+
+def all_gather(x: mx.array, group: mx.distributed.Group) -> mx.array:
+    return _link(mx.distributed.all_gather(x, group=group, stream=mx.Device(mx.cpu)))
+
 
 def flush_prefill_sends() -> None:
     for output, dst, group in _pending_prefill_sends:
-        sent = mx.distributed.send(output, dst, group=group)
-        mx.async_eval(sent)
+        send(output, dst, group)
     _pending_prefill_sends.clear()
 
 
+def reset_chain_head() -> None:
+    global _last_dist_op
+    _last_dist_op = None
+
+
+def dist_chain_head() -> mx.array:
+    assert _last_dist_op is not None
+    return _last_dist_op
+
+
 def clear_prefill_sends() -> None:
     # Discard pending sends (e.g. on cancellation).
     _pending_prefill_sends.clear()
@@ -131,11 +165,7 @@ class PipelineFirstLayer(CustomMlxLayer):
 
     def __call__(self, x: mx.array, *args: object, **kwargs: object) -> mx.array:
         if self.r != 0:
-            # We want to avoid GPU timeout errors by evalling the distributed operation
-            # so that it stays on CPU, which does not have a timeout.
-            mx.eval(x)
-            x = mx.distributed.recv_like(x, (self.r - 1), group=self.group)
-            mx.eval(x)
+            x = recv_like(x, (self.r - 1), group=self.group)
         return self.original_layer(x, *args, **kwargs)
 
 
@@ -162,36 +192,31 @@ class PipelineLastLayer(CustomMlxLayer):
 
         output: mx.array = self.original_layer(x, *args, **kwargs)
 
-        # Eval layer output to materialize it before send — this splits the graph
-        # so the send is isolated and the receiving rank's recv can complete.
-        mx.eval(output)
-
         if self.r != self.s - 1:
             if self.queue_sends:
                 _pending_prefill_sends.append(
                     (output, (self.r + 1) % self.s, self.group)
                 )
+                sent_output = output
             else:
-                output = mx.distributed.send(
-                    output, (self.r + 1) % self.s, group=self.group
-                )
-            if cache is not None:
-                # CacheList (used by MLA models like DeepSeekV32, GLM MoE DSA)
-                # doesn't have .keys directly; access via first sub-cache.
-                _cache = cache[0] if hasattr(cache, "caches") else cache  # type: ignore
-                if hasattr(_cache, "keys"):  # pyright: ignore[reportAny]
-                    _cache.keys = mx.depends(_cache.keys, output)  # type: ignore
-            mx.eval(output)
-            if cache is not None and hasattr(_cache, "keys"):  # type: ignore
-                mx.eval(_cache.keys)  # type: ignore
+                sent_output = send(output, (self.r + 1) % self.s, self.group)
+                if cache is not None:
+                    # CacheList (used by MLA models like DeepSeekV32, GLM MoE DSA)
+                    # doesn't have .keys directly; access via first sub-cache.
+                    _cache = cache[0] if hasattr(cache, "caches") else cache  # type: ignore
+                    if hasattr(_cache, "keys"):  # pyright: ignore[reportAny]
+                        _cache.keys = mx.depends(_cache.keys, sent_output)  # type: ignore
+        else:
+            sent_output = output
 
         if not self.is_prefill:
-            output = mx.distributed.all_gather(output, group=self.group)[
+            all_gathered_output = all_gather(sent_output, group=self.group)[
                 -output.shape[0] :
             ]
-            mx.eval(output)
+        else:
+            all_gathered_output = sent_output
 
-        return output
+        return all_gathered_output
 
 
 def set_pipeline_prefill(model: nn.Module, is_prefill: bool) -> None:
@@ -392,64 +417,6 @@ def pipeline_auto_parallel(
         "Expected a list of layers after auto-parallel initialisation"
     )
 
-    return patch_pipeline_model(model, group)
-
-
-def patch_pipeline_model[T](model: T, group: mx.distributed.Group) -> T:
-    # Patch __call__ on the model's class
-    cls = model.__class__
-    original_call = cls.__call__  # type :ignore
-    call_signature = signature(original_call)  # type :ignore
-
-    def patched_call(
-        self: T,
-        *args: object,
-        **kwargs: object,
-    ) -> mx.array:
-        logits: mx.array = original_call(self, *args, **kwargs)  # type: ignore
-        cache = call_signature.bind_partial(self, *args, **kwargs).arguments.get(
-            "cache", None
-        )
-
-        # Add dependency to last cache entry to ensure distributed ops are evaluated
-        if cache is not None and len(cache) > 0:  # type: ignore
-            last = cache[-1]  # type: ignore
-            dep_cache = last[0] if hasattr(last, "caches") else last  # type: ignore
-            if hasattr(dep_cache, "keys") and dep_cache.keys is not None:  # type: ignore
-                dep_cache.keys = mx.depends(dep_cache.keys, logits)  # type: ignore
-
-        return logits
-
-    cls.__call__ = patched_call
-    return model
-
-
-def patch_tensor_model[T](model: T) -> T:
-    """Patch model's __call__ to ensure distributed ops sync during inference."""
-    cls = model.__class__
-    original_call = cls.__call__
-    call_signature = signature(original_call)
-
-    def patched_call(
-        self: T,
-        *args: object,
-        **kwargs: object,
-    ) -> mx.array:
-        logits: mx.array = original_call(self, *args, **kwargs)  # pyright: ignore[reportAny]
-        cache = call_signature.bind_partial(self, *args, **kwargs).arguments.get(
-            "cache", None
-        )
-
-        # Add dependency to last cache entry to ensure distributed ops are evaluated
-        if cache is not None and len(cache) > 0:  # pyright: ignore[reportAny]
-            last = cache[-1]  # pyright: ignore[reportAny]
-            dep_cache = last[0] if hasattr(last, "caches") else last  # pyright: ignore[reportAny]
-            if hasattr(dep_cache, "keys"):  # type: ignore
-                dep_cache.keys = mx.depends(dep_cache.keys, logits)  # pyright: ignore[reportAny]
-
-        return logits
-
-    cls.__call__ = patched_call
     return model
 
 
@@ -599,7 +566,7 @@ def tensor_auto_parallel(
         raise ValueError(f"Unsupported model type: {type(model)}")
 
     model = yield from tensor_parallel_sharding_strategy.shard_model(model)
-    return patch_tensor_model(model)
+    return model
 
 
 class TensorParallelShardingStrategy(ABC):
@@ -763,8 +730,10 @@ class ShardedMoE(CustomMlxLayer):
             x = sum_gradients(self.sharding_group)(x)
         y = self.original_layer.__call__(x)
         if self.sharding_group is not None:
-            y = mx.distributed.all_sum(y, group=self.sharding_group)
-        return y
+            z = mx.distributed.all_sum(y, group=self.sharding_group)
+        else:
+            z = y
+        return z
 
 
 class ShardedMoEV4(CustomMlxLayer):
@@ -780,8 +749,10 @@ class ShardedMoEV4(CustomMlxLayer):
             x = sum_gradients(self.sharding_group)(x)
         y = self._v4_inner(x, input_ids)
         if self.sharding_group is not None:
-            y = mx.distributed.all_sum(y, group=self.sharding_group)
-        return y
+            z = mx.distributed.all_sum(y, group=self.sharding_group)
+        else:
+            z = y
+        return z
 
 
 def _shard_quantized_rows(

src/exo/worker/engines/mlx/cache.py

@@ -1,5 +1,6 @@
 import gc
 import os
+import time
 from copy import deepcopy
 from typing import TYPE_CHECKING
 
@@ -93,6 +94,17 @@ def copy_rotating_kv_cache(cache: RotatingKVCache) -> RotatingKVCache | None:
     return snap
 
 
+def copy_kv_cache(cache: KVCache) -> KVCache:
+    snap = KVCache()
+    snap.offset = cache.offset
+    if cache.keys is not None:
+        assert cache.values is not None
+        snap.keys = _detached_copy(cache.keys)
+        snap.values = _detached_copy(cache.values)
+        mx.eval(snap.keys, snap.values)
+    return snap
+
+
 def _copy_arrays_cache(ac: ArraysCache) -> ArraysCache:
     entries: list[mx.array | None] = []
     for entry in ac.cache:  # type: ignore[reportUnknownMemberType]
@@ -181,6 +193,41 @@ def copy_snapshot_entry(
             return _copy_v4_cache(entry)
 
 
+def copy_cache_entry(
+    entry: KVCache
+    | RotatingKVCache
+    | QuantizedKVCache
+    | ArraysCache
+    | CacheList
+    | DeepseekV4Cache,
+) -> (
+    KVCache
+    | RotatingKVCache
+    | QuantizedKVCache
+    | ArraysCache
+    | CacheList
+    | DeepseekV4Cache
+):
+    match entry:
+        case KVCache():
+            return copy_kv_cache(entry)
+        case RotatingKVCache():
+            snap = copy_rotating_kv_cache(entry)
+            return snap if snap is not None else deepcopy(entry)
+        case QuantizedKVCache():
+            return deepcopy(entry)
+        case ArraysCache():
+            return _copy_arrays_cache(entry)
+        case CacheList():
+            return _copy_cache_list(entry)
+        case DeepseekV4Cache():
+            return _copy_v4_cache(entry)
+
+
+def copy_kv_cache_state(cache: KVCacheType) -> KVCacheType:
+    return [copy_cache_entry(entry) for entry in cache]
+
+
 def snapshot_ssm_states(cache: KVCacheType) -> CacheSnapshot:
     states: list[
         RotatingKVCache | ArraysCache | CacheList | DeepseekV4Cache | None
@@ -260,7 +307,7 @@ class KVPrefixCache:
         """Add a new cache entry. Evicts LRU entries if memory is high."""
         self._evict_if_needed()
         self.prompts.append(prompt_tokens)
-        self.caches.append(deepcopy(cache))
+        self.caches.append(copy_kv_cache_state(cache))
         self._snapshots.append(ssm_snapshots)
         self._media_regions.append(media_regions or [])
         self.prefill_tps.append(prefill_tps)
@@ -287,7 +334,7 @@ class KVPrefixCache:
             merged.extend(snapshots)
 
         self.prompts[index] = prompt_tokens
-        self.caches[index] = deepcopy(cache)
+        self.caches[index] = copy_kv_cache_state(cache)
         self._snapshots[index] = merged or None
         self._media_regions[index] = media_regions or []
         self.prefill_tps[index] = prefill_tps
@@ -376,7 +423,7 @@ class KVPrefixCache:
         if restore_snap is None and has_ssm:
             return make_kv_cache(model), prompt_tokens, None, False
 
-        prompt_cache = deepcopy(self.caches[best_index])
+        prompt_cache = copy_kv_cache_state(self.caches[best_index])
         tokens_to_trim = cached_length - restore_pos
         if tokens_to_trim > 0:
             trim_cache(prompt_cache, tokens_to_trim, restore_snap)
@@ -456,6 +503,7 @@ class KVPrefixCache:
             mx.clear_cache()
 
     def get_memory_used_percentage(self) -> float:
+        t0 = time.perf_counter()
         local_pressure: float = get_memory_used_percentage()
 
         if self._group is None:
@@ -464,9 +512,14 @@ class KVPrefixCache:
         all_pressure = mx.distributed.all_gather(
             mx.array([local_pressure], dtype=mx.float32),
             group=self._group,
+            stream=mx.Device(mx.cpu),
         )
         # .item() evals.
         max_pressure = float(mx.max(all_pressure).item())
+        elapsed_ms = (time.perf_counter() - t0) * 1000
+        logger.info(
+            f"get_memory_used_percentage took {elapsed_ms:.2f}ms (group size {self._group.size()})"
+        )
         return max_pressure
 
 

src/exo/worker/engines/mlx/generator/batch_generate.py

@@ -49,6 +49,7 @@ from exo.worker.engines.mlx.types import KVCacheType, Model
 from exo.worker.engines.mlx.utils_mlx import (
     fix_unmatched_think_end_tokens,
     system_prompt_token_count,
+    mx_barrier
 )
 from exo.worker.engines.mlx.vision import (
     MediaRegion,
@@ -105,7 +106,7 @@ class ExoBatchGenerator:
         self._mlx_gen = MlxBatchGenerator(
             model=self.model,
             stop_tokens=[[t] for t in eos_ids_from_tokenizer(self.tokenizer)],
-            prefill_step_size=4096,
+            prefill_step_size=1,
         )
         self._step_count = 0
 
@@ -231,6 +232,7 @@ class ExoBatchGenerator:
                     )
 
             if not remote_prefilled:
+                mx_barrier(self.group)
                 _prefill_tps, _prefill_tokens, cache_snapshots = prefill(
                     self.model,
                     self.tokenizer,
@@ -324,6 +326,8 @@ class ExoBatchGenerator:
             media_regions=media_regions,
         )
 
+        mx_barrier(self.group)
+
         return uid
 
     def step(self) -> list[tuple[int, GenerationResponse]]:

src/exo/worker/engines/mlx/generator/generate.py

@@ -35,7 +35,9 @@ from exo.worker.engines.mlx.auto_parallel import (
     PipelineFirstLayer,
     PipelineLastLayer,
     clear_prefill_sends,
+    dist_chain_head,
     flush_prefill_sends,
+    reset_chain_head,
     set_pipeline_prefill,
     set_pipeline_queue_sends,
 )
@@ -192,7 +194,8 @@ def pipeline_parallel_prefill(
     This function is designed to match mlx_lm's stream_generate exactly in terms of
     side effects (given the same prefill step size)
     """
-    prefill_step_size = prefill_step_size // min(4, group.size())
+    world_size = group.size()
+    prefill_step_size = prefill_step_size // min(4, world_size)
 
     quantize_cache_fn: Callable[..., None] = functools.partial(
         maybe_quantize_kv_cache,
@@ -203,14 +206,18 @@ def pipeline_parallel_prefill(
 
     _prompt_cache: KVCacheType = prompt_cache
     rank = group.rank()
-    world_size = group.size()
 
-    # Build list of real prompt chunk sizes
+    # Build list of real prompt chunk sizes.
+    # For pipeline parallel to overlap stages, we need at least world_size real chunks;
+    # otherwise a single-chunk prompt serializes across ranks (each rank's recv blocks).
     total = len(prompt)
-    real_chunk_sizes: list[int] = []
     remaining = total - 1
+    chunk_size = min(
+        prefill_step_size, max(1, (remaining + world_size - 1) // world_size)
+    )
+    real_chunk_sizes: list[int] = []
     while remaining:
-        n = min(prefill_step_size, remaining)
+        n = min(chunk_size, remaining)
         real_chunk_sizes.append(n)
         remaining -= n
     n_real = len(real_chunk_sizes)
@@ -232,43 +239,77 @@ def pipeline_parallel_prefill(
 
     try:
         with mx.stream(generation_stream):
+            t_leading = time.perf_counter()
             for _ in range(n_leading):
                 if distributed_prompt_progress_callback is not None:
                     distributed_prompt_progress_callback()
+            logger.info(
+                f"[R{rank}] leading dummies ({n_leading}) took {(time.perf_counter() - t_leading) * 1000:.1f}ms"
+            )
 
             for i in range(n_real):
+                t_iter = time.perf_counter()
                 chunk_size = real_chunk_sizes[i]
+                t_model = time.perf_counter()
                 model(
                     prompt[processed : processed + chunk_size][None],
                     cache=_prompt_cache,
                 )
                 quantize_cache_fn(_prompt_cache)
+                t_after_model = time.perf_counter()
                 processed += chunk_size
 
                 if distributed_prompt_progress_callback is not None:
                     distributed_prompt_progress_callback()
+                t_after_cb = time.perf_counter()
 
                 flush_prefill_sends()
+                t_after_flush = time.perf_counter()
+                mx.eval([c.state for c in _prompt_cache])  # type: ignore
+                t_after_eval = time.perf_counter()
 
                 prompt_progress_callback(processed, total)
+                logger.info(
+                    f"[R{rank}] iter {i}/{n_real} ({chunk_size} tok): "
+                    f"model+quant {(t_after_model - t_model) * 1000:.1f}ms, "
+                    f"cb {(t_after_cb - t_after_model) * 1000:.1f}ms, "
+                    f"flush {(t_after_flush - t_after_cb) * 1000:.1f}ms, "
+                    f"eval {(t_after_eval - t_after_flush) * 1000:.1f}ms, "
+                    f"total {(time.perf_counter() - t_iter) * 1000:.1f}ms"
+                )
 
+            t_trailing = time.perf_counter()
             for _ in range(n_trailing):
                 if distributed_prompt_progress_callback is not None:
                     distributed_prompt_progress_callback()
+            logger.info(
+                f"[R{rank}] trailing dummies ({n_trailing}) took {(time.perf_counter() - t_trailing) * 1000:.1f}ms"
+            )
 
     finally:
         clear_prefill_sends()
 
     # Post-loop: process remaining 1 token + add +1 entry to match stream_generate.
-    for _ in range(2):
+    for j in range(2):
+        t_post = time.perf_counter()
         with mx.stream(generation_stream):
             model(prompt[-1:][None], cache=_prompt_cache)
             quantize_cache_fn(_prompt_cache)
         flush_prefill_sends()
+        logger.info(
+            f"[R{rank}] post-loop iter {j} took {(time.perf_counter() - t_post) * 1000:.1f}ms"
+        )
 
     assert _prompt_cache is not None
+    t_final = time.perf_counter()
     with mx.stream(generation_stream):
-        mx.eval([c.state for c in _prompt_cache])  # type: ignore
+        mx.eval([c.state for c in _prompt_cache], dist_chain_head())  # type: ignore
+    mx.synchronize(mx.default_stream(mx.Device(mx.cpu)))
+    mx.synchronize(generation_stream)
+    reset_chain_head()
+    logger.info(
+        f"[R{rank}] final eval+chain drain took {(time.perf_counter() - t_final) * 1000:.1f}ms"
+    )
 
     # Final callback matching generate_step
     prompt_progress_callback(total, total)
@@ -302,6 +343,8 @@ def prefill(
         return 0.0, 0, []
 
     logger.debug(f"Prefilling {num_tokens} tokens...")
+
+    reset_chain_head()
     start_time = time.perf_counter()
     has_ssm = has_non_kv_caches(cache)
     snapshots: list[CacheSnapshot] = []
@@ -334,7 +377,10 @@ def prefill(
     prefill_step_size = 4096
 
     try:
-        if is_pipeline and num_tokens >= prefill_step_size:
+        if is_pipeline:
+            logger.info(
+                f"prefill path: pipeline_parallel_prefill ({num_tokens} tokens)"
+            )
             set_pipeline_queue_sends(model, queue_sends=True)
             assert group is not None, "Pipeline prefill requires a distributed group"
             pipeline_parallel_prefill(
@@ -349,6 +395,10 @@ def prefill(
                 group=group,
             )
         else:
+            logger.info(
+                f"prefill path: stream_generate ({num_tokens} tokens, is_pipeline={is_pipeline})"
+            )
+            t_stream = time.perf_counter()
             # Use max_tokens=1 because max_tokens=0 does not work.
             # We just throw away the generated token - we only care about filling the cache
             for _ in stream_generate(
@@ -364,6 +414,9 @@ def prefill(
                 prompt_progress_callback=combined_progress_callback,
             ):
                 break  # Stop after first iteration - cache is now filled
+            logger.info(
+                f"stream_generate prefill took {(time.perf_counter() - t_stream) * 1000:.1f}ms"
+            )
     except PrefillCancelled:
         set_pipeline_queue_sends(model, queue_sends=False)
         set_pipeline_prefill(model, is_prefill=False)
@@ -543,6 +596,7 @@ def mlx_generate(
 ) -> Generator[GenerationResponse]:
     # Ensure that generation stats only contains peak memory for this generation
     mx.reset_peak_memory()
+    reset_chain_head()
     # TODO: Randomise task seed and set in taskparams, instead of hard coding as 42.
     seed = task.seed or 42
     mx.random.seed(seed)
@@ -716,6 +770,7 @@ def mlx_generate(
     usage: Usage | None = None
     logger.info("Starting decode")
     mx_barrier(group)
+    reset_chain_head()
 
     for completion_tokens, out in enumerate(
         stream_generate(

src/exo/worker/engines/mlx/utils_mlx.py

@@ -115,7 +115,8 @@ def mlx_distributed_init(
 
                 os.environ["MLX_HOSTFILE"] = coordination_file
                 os.environ["MLX_RANK"] = str(rank)
-                os.environ["MLX_RING_VERBOSE"] = "1"
+                # os.environ["MLX_RING_VERBOSE"] = "1"  # NOTE: we don't use it enough to care (turn on again if need to)
+
                 group = mx.distributed.init(backend="ring", strict=True)
 
             case MlxJacclInstance(
@@ -310,12 +311,11 @@ def get_eos_token_ids_for_model(model_id: ModelId) -> list[int] | None:
     model_id_lower = model_id.lower()
     if "kimi-k2" in model_id_lower:
         return [163586]
-    elif "glm-5" in model_id_lower or "glm-4.7" in model_id_lower:
-        # For GLM-5 and GLM-4.7
+    elif "glm-5" in model_id_lower:
         # 154820: <|endoftext|>, 154827: <|user|>, 154829: <|observation|>
         return [154820, 154827, 154829]
     elif "glm" in model_id_lower:
-        # For GLM-4.5 and older
+        # For GLM-4.7 and older
         return [151336, 151329, 151338]
     elif "gpt-oss" in model_id_lower:
         return [200002, 200012]

src/exo/worker/main.py

@@ -8,25 +8,21 @@ 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 ModelId, card_cache
 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, SessionId, SystemId
+from exo.shared.types.common import CommandId, NodeId, SystemId
 from exo.shared.types.events import (
-    CustomModelCardAdded,
-    CustomModelCardDeleted,
     Event,
     IndexedEvent,
+    InputChunkReceived,
     InstanceDeleted,
     NodeDownloadProgress,
     NodeGatheredInfo,
@@ -36,7 +32,6 @@ 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,
@@ -62,19 +57,14 @@ 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],
@@ -82,11 +72,8 @@ 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
@@ -116,57 +103,23 @@ class Worker:
 
         try:
             async with self._tg as tg:
-                tg.start_soon(self._bootstrap_then_run, info_gatherer, info_recv)
+                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._reconcile_custom_cards)
+
         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._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:
@@ -181,8 +134,6 @@ 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
@@ -190,41 +141,48 @@ class Worker:
                 if isinstance(event, InstanceDeleted):
                     self._instance_backoff.reset(event.instance_id)
 
-                if isinstance(event, CustomModelCardAdded):
-                    await event.model_card.save_to_custom_dir()
-                    add_to_card_cache(event.model_card)
+                # 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
 
-                if isinstance(event, CustomModelCardDeleted):
-                    await delete_custom_card(event.model_id)
+                    self.input_chunk_buffer[cmd_id][event.chunk.chunk_index] = (
+                        event.chunk
+                    )
+                    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
 
-                self._sync_input_views_from_state()
+    async def _reconcile_custom_cards(self) -> None:
+        while True:
+            await anyio.sleep(1)
+            target = dict(self.state.custom_model_cards)
+            for model_id, card in target.items():
+                if card_cache.get(model_id) == card:
+                    continue
+                await card_cache.save(card)
 
-    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
+            for card in await card_cache.list_all():
+                if card.model_id not in target:
+                    await card_cache.pop(card.model_id)
 
     async def plan_step(self):
         while True:
@@ -236,7 +194,7 @@ class Worker:
                 self.state.instances,
                 self.state.runners,
                 self.state.tasks,
-                self.state.input_chunks,
+                self.input_chunk_buffer,
                 self.image_cache,
                 self._instance_backoff,
                 self._download_backoff,
@@ -265,7 +223,7 @@ class Worker:
             # lets not kill the worker if a runner is unresponsive
             match task:
                 case CreateRunner():
-                    self._create_supervisor(task)
+                    await self._create_supervisor(task)
                     self._instance_backoff.record_attempt(task.instance_id)
                     await self.event_sender.send(
                         TaskStatusUpdated(
@@ -368,9 +326,15 @@ 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())
@@ -382,6 +346,10 @@ 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:
@@ -402,9 +370,9 @@ class Worker:
                 instance.shard_assignments.node_to_runner[self.node_id]
             ].start_task(task)
 
-    def _create_supervisor(self, task: CreateRunner) -> RunnerSupervisor:
+    async def _create_supervisor(self, task: CreateRunner) -> RunnerSupervisor:
         """Creates and stores a new AssignedRunner with initial downloading status."""
-        runner = RunnerSupervisor.create(
+        runner = await RunnerSupervisor.create(
             bound_instance=task.bound_instance,
             event_sender=self.event_sender.clone(),
         )

src/exo/worker/runner/llm_inference/batch_generator.py

@@ -138,8 +138,10 @@ class SequentialGenerator(Engine):
     def agree_on_tasks(self) -> None:
         """Agree between all ranks about the task ordering (some may have received in different order or not at all)."""
         agreed, different = mx_all_gather_tasks(self._maybe_queue, self.group)
-        self._queue.extend(task for task in self._maybe_queue if task in agreed)
-        self._maybe_queue = [task for task in self._maybe_queue if task in different]
+        # Extend from `agreed` (sorted by task_id on all ranks) to guarantee every
+        # rank enqueues tasks in the same order, preventing TP collective deadlocks.
+        self._queue.extend(agreed)
+        self._maybe_queue = list(different)
 
     def agree_on_cancellations(self) -> None:
         """Agree between all ranks about which tasks to cancel."""
@@ -197,9 +199,14 @@ class SequentialGenerator(Engine):
             self._active = None
             raise
 
-        return itertools.chain(
-            output,
-            map(lambda task: (task, CancelledResponse()), self._cancelled_tasks),
+        return filter(
+            lambda chunk: (
+                not isinstance(chunk[1], GenerationChunk) or self.device_rank == 0
+            ),
+            itertools.chain(
+                output,
+                map(lambda task: (task, CancelledResponse()), self._cancelled_tasks),
+            ),
         )
 
     def _start_next(self) -> None:
@@ -368,8 +375,10 @@ class BatchGenerator(Engine):
     def agree_on_tasks(self) -> None:
         """Agree between all ranks about the task ordering (some may have received in different order or not at all)."""
         agreed, different = mx_all_gather_tasks(self._maybe_queue, self.group)
-        self._queue.extend(task for task in self._maybe_queue if task in agreed)
-        self._maybe_queue = [task for task in self._maybe_queue if task in different]
+        # Extend from `agreed` (sorted by task_id on all ranks) to guarantee every
+        # rank enqueues tasks in the same order, preventing TP collective deadlocks.
+        self._queue.extend(agreed)
+        self._maybe_queue = list(different)
 
     def agree_on_cancellations(self) -> None:
         """Agree between all ranks about which tasks to cancel."""
@@ -449,7 +458,12 @@ class BatchGenerator(Engine):
                 output.append((task.task_id, FinishedResponse()))
                 del self._active_tasks[uid]
 
-        return itertools.chain(output, self._apply_cancellations())
+        return filter(
+            lambda chunk: (
+                not isinstance(chunk[1], GenerationChunk) or self.device_rank == 0
+            ),
+            itertools.chain(output, self._apply_cancellations()),
+        )
 
     def _apply_cancellations(
         self,

src/exo/worker/runner/runner.py

@@ -390,5 +390,5 @@ class Runner:
         chunk: Chunk,
         command_id: CommandId,
     ):
-        if self.device_rank == 0:
-            self.event_sender.send(ChunkGenerated(command_id=command_id, chunk=chunk))
+        assert isinstance(self.generator, Engine)
+        self.event_sender.send(ChunkGenerated(command_id=command_id, chunk=chunk))

src/exo/worker/runner/supervisor.py

@@ -1,17 +1,20 @@
+import codecs
 import contextlib
-import multiprocessing as mp
 import signal
 from dataclasses import dataclass, field
-from typing import Self
+from os import PathLike
+from typing import Callable, Self
 
 import anyio
 from anyio import (
+    AsyncFile,
     BrokenResourceError,
+    CancelScope,
     ClosedResourceError,
-    to_thread,
 )
 from loguru import logger
 
+from exo.shared.constants import EXO_RUNNER_STDERR_LOG, EXO_RUNNER_STDOUT_LOG
 from exo.shared.types.chunks import ErrorChunk
 from exo.shared.types.events import (
     ChunkGenerated,
@@ -41,7 +44,9 @@ from exo.shared.types.worker.runners import (
     RunnerWarmingUp,
 )
 from exo.shared.types.worker.shards import ShardMetadata
-from exo.utils.channels import MpReceiver, MpSender, Sender, mp_channel
+from exo.utils.async_process import AsyncProcess
+from exo.utils.channels import MpReceiver, MpSender, Receiver, Sender, mp_channel
+from exo.utils.fs import ensure_parent_directory_exists
 from exo.utils.task_group import TaskGroup
 from exo.worker.runner.bootstrap import entrypoint
 
@@ -49,11 +54,127 @@ PREFILL_TIMEOUT_SECONDS = 60
 DECODE_TIMEOUT_SECONDS = 5
 
 
+@dataclass(eq=False)
+class RunnerStdioHandler:
+    _stdout_rx: Receiver[bytes]
+    _stderr_rx: Receiver[bytes]
+    _stdout_log: AsyncFile[str]
+    _stderr_log: AsyncFile[str]
+
+    _tg: TaskGroup = field(default_factory=TaskGroup, init=False)
+
+    @classmethod
+    async def create(
+        cls,
+        *,
+        stdout_rx: Receiver[bytes],
+        stderr_rx: Receiver[bytes],
+        stdout_log_path: PathLike[str] = EXO_RUNNER_STDOUT_LOG,
+        stderr_log_path: PathLike[str] = EXO_RUNNER_STDERR_LOG,
+    ) -> Self:
+        # these are append only logs used to gather data for log template mining
+        #
+        # TODO: in the future use [Drain3](https://github.com/logpai/Drain3)
+        #       to mine these logs
+        ensure_parent_directory_exists(stdout_log_path)
+        ensure_parent_directory_exists(stderr_log_path)
+        stdout_log = await anyio.open_file(stdout_log_path, "a")
+        stderr_log = await anyio.open_file(stderr_log_path, "a")
+
+        # instantiate and return
+        self = cls(
+            _stdout_rx=stdout_rx,
+            _stderr_rx=stderr_rx,
+            _stdout_log=stdout_log,
+            _stderr_log=stderr_log,
+        )
+        return self
+
+    async def run(self):
+        try:
+            async with self._tg as tg:
+                tg.start_soon(  # pyright: ignore[reportUnknownArgumentType]
+                    self._handle_runner_output,
+                    self._stdout_rx,
+                    self._stdout_log,
+                    lambda line: logger.info(f"Runner stdout: {line}"),  # pyright: ignore[reportUnknownLambdaType]
+                )
+                tg.start_soon(  # pyright: ignore[reportUnknownArgumentType]
+                    self._handle_runner_output,
+                    self._stderr_rx,
+                    self._stderr_log,
+                    lambda line: logger.warning(f"Runner stderr: {line}"),  # pyright: ignore[reportUnknownLambdaType]
+                )
+        finally:
+            with CancelScope(shield=True):
+                await self._stdout_log.aclose()
+                await self._stderr_log.aclose()
+
+    async def _handle_runner_output(
+        self,
+        rx: Receiver[bytes],
+        logfile: AsyncFile[str],
+        log_line: Callable[[str], None],
+    ):
+        # TODO: right now it logs them as warnings, but in the future they should be split
+        #       into being logged AND a seperate task which tries to best-effort figure out cause
+        #       of error and package into error enum, which then is used by rest of app to act on it;
+        #       inferring what the error is would be done by pattern-matching in the text for things
+        #       e.g. certain VLLM error codes and so on
+
+        # not using TextReceiveStream because it doesn't do final=True handling on errors
+        decoder = codecs.getincrementaldecoder("utf-8")(errors="replace")
+        pending_line = ""
+
+        async def handle_line(line: str):
+            # preserve whitespace for later log-mining
+            line = line.removesuffix("\r")
+            if not line:
+                return
+
+            # Send to logger & error recovery task
+            log_line(line)
+            # TODO: error recovery task
+
+        async def handle_text(text: str):
+            nonlocal pending_line
+
+            if not text:
+                return
+
+            await logfile.write(text)
+            await logfile.flush()
+
+            # newline buffering
+            pending_line += text
+            lines = pending_line.split("\n")
+            pending_line = lines.pop()
+
+            for line in lines:
+                await handle_line(line)
+
+        try:
+            with rx:
+                async for chunk in rx:
+                    await handle_text(decoder.decode(chunk, final=False))
+        except (ClosedResourceError, BrokenResourceError):
+            logger.warning("Runner stdio stream closed before clean EOF")
+        finally:
+            with CancelScope(shield=True):
+                await handle_text(decoder.decode(b"", final=True))
+                await logfile.flush()
+
+                if pending_line:
+                    await handle_line(pending_line)
+                    pending_line = ""
+
+
 @dataclass(eq=False)
 class RunnerSupervisor:
     shard_metadata: ShardMetadata
     bound_instance: BoundInstance
-    runner_process: mp.Process
+    runner_process: AsyncProcess
+    _runner_stdio_handler: RunnerStdioHandler
     initialize_timeout: float
     _ev_recv: MpReceiver[Event]
     _task_sender: MpSender[Task]
@@ -70,7 +191,7 @@ class RunnerSupervisor:
     )
 
     @classmethod
-    def create(
+    async def create(
         cls,
         *,
         bound_instance: BoundInstance,
@@ -81,7 +202,7 @@ class RunnerSupervisor:
         task_sender, task_recv = mp_channel[Task]()
         cancel_sender, cancel_recv = mp_channel[TaskId]()
 
-        runner_process = mp.Process(
+        runner_process = AsyncProcess(
             target=entrypoint,
             args=(
                 bound_instance,
@@ -92,6 +213,9 @@ class RunnerSupervisor:
             ),
             daemon=True,
         )
+        runner_stdio_handler = await RunnerStdioHandler.create(
+            stdout_rx=runner_process.stdout, stderr_rx=runner_process.stderr
+        )
 
         shard_metadata = bound_instance.bound_shard
 
@@ -99,6 +223,7 @@ class RunnerSupervisor:
             bound_instance=bound_instance,
             shard_metadata=shard_metadata,
             runner_process=runner_process,
+            _runner_stdio_handler=runner_stdio_handler,
             initialize_timeout=initialize_timeout,
             _ev_recv=ev_recv,
             _task_sender=task_sender,
@@ -109,9 +234,12 @@ class RunnerSupervisor:
         return self
 
     async def run(self):
-        self.runner_process.start()
         try:
             async with self._tg as tg:
+                # start the process itself & handle its stdout/stderr
+                await tg.start(self.runner_process.run)
+                tg.start_soon(self._runner_stdio_handler.run)
+
                 tg.start_soon(self._watch_runner)
                 tg.start_soon(self._forward_events)
         finally:
@@ -129,41 +257,11 @@ class RunnerSupervisor:
             with contextlib.suppress(ClosedResourceError):
                 self._cancel_sender.close()
 
-            await to_thread.run_sync(self.runner_process.join, 5)
-
-            if self.runner_process.is_alive():
-                logger.warning(
-                    "Runner process didn't shutdown succesfully, terminating"
+            with anyio.CancelScope(shield=True):
+                await self.runner_process.stop()
+                logger.info(
+                    f"Runner process successfully terminated: {self.runner_process.exitcode}"
                 )
-                self.runner_process.terminate()
-                self.runner_process.join(timeout=10)
-
-                if not self.runner_process.is_alive():
-                    logger.warning("Terminated nicely in the first attempt!")
-
-                else:
-                    # Try really hard to terminate
-                    for i in range(2, 11):
-                        self.runner_process.terminate()
-                        self.runner_process.join(timeout=2)
-                        if not self.runner_process.is_alive():
-                            logger.warning(f"That took {i} attempts :)")
-                            break
-                    # Try even harder to kill
-                    else:
-                        logger.critical(
-                            "Runner process didn't respond to SIGTERM, killing"
-                        )
-                        j = 0
-                        while self.runner_process.is_alive():
-                            j += 1
-                            self.runner_process.kill()
-                            self.runner_process.join(timeout=5)
-                            logger.warning(f"That took {j} attempts :(")
-            else:
-                logger.info("Runner process succesfully terminated")
-
-            self.runner_process.close()
 
     def shutdown(self):
         self._tg.cancel_tasks()
@@ -254,8 +352,9 @@ class RunnerSupervisor:
             self._cancel_watch_runner.cancel()
         logger.info("Checking runner's status")
         if self.runner_process.is_alive():
-            logger.info("Runner was found to be alive, attempting to join process")
-            await to_thread.run_sync(self.runner_process.join, 5)
+            logger.info("Runner was found to be alive, stopping process")
+            with anyio.CancelScope(shield=True):
+                await self.runner_process.stop()
         rc = self.runner_process.exitcode
         logger.info(f"Runner exited with exit code {rc}")
         if rc == 0:

src/exo/worker/tests/unittests/test_mlx/test_auto_parallel.py

@@ -12,7 +12,6 @@ from exo.worker.engines.mlx.auto_parallel import (
     CustomMlxLayer,
     PipelineFirstLayer,
     PipelineLastLayer,
-    patch_pipeline_model,
 )
 from exo.worker.tests.unittests.test_mlx.conftest import MockLayer
 
@@ -55,7 +54,7 @@ def run_pipeline_device(
 
         # Wrap in a mock model, then wrap in PipelineParallelModel for all_gather
         inner_model = MockModel([composed])
-        model = patch_pipeline_model(inner_model, group)
+        model = inner_model
 
         x = mx.ones((1, 4))
         result = model(x)

src/exo/worker/tests/unittests/test_mlx/test_tokenizers.py

@@ -16,7 +16,7 @@ from exo.download.download_utils import (
     fetch_file_list_with_cache,
     resolve_model_dir,
 )
-from exo.shared.models.model_cards import ModelCard, ModelId, get_model_cards
+from exo.shared.models.model_cards import ModelCard, ModelId, card_cache
 from exo.worker.engines.mlx.utils_mlx import (
     get_eos_token_ids_for_model,
     load_tokenizer_for_model_id,
@@ -76,7 +76,7 @@ def get_test_models() -> list[ModelCard]:
     """Get a representative sample of models to test."""
     # Pick one model from each family to test
     families: dict[str, ModelCard] = {}
-    for card in asyncio.run(get_model_cards()):
+    for card in asyncio.run(card_cache.list_all()):
         # Extract family name (e.g., "llama-3.1" from "llama-3.1-8b")
         parts = card.model_id.short().split("-")
         family = "-".join(parts[:2]) if len(parts) >= 2 else parts[0]
@@ -298,7 +298,7 @@ async def test_tokenizer_special_tokens(model_card: ModelCard) -> None:
 async def test_kimi_tokenizer_specifically():
     """Test Kimi tokenizer with its specific patches and quirks."""
     kimi_models = [
-        card for card in await get_model_cards() if "kimi" in card.model_id.lower()
+        card for card in await card_cache.list_all() if "kimi" in card.model_id.lower()
     ]
 
     if not kimi_models:
@@ -350,7 +350,7 @@ async def test_glm_tokenizer_specifically():
 
     glm_model_cards = [
         card
-        for card in await get_model_cards()
+        for card in await card_cache.list_all()
         if contains(card, "glm")
         and not contains(card, "-5")
         and not contains(card, "4.7")

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

@@ -1,4 +1,3 @@
-import multiprocessing as mp
 from typing import cast
 
 import anyio
@@ -16,31 +15,26 @@ from exo.shared.types.text_generation import (
 )
 from exo.shared.types.worker.instances import BoundInstance, InstanceId
 from exo.shared.types.worker.runners import RunnerFailed, RunnerId
+from exo.utils.async_process import AsyncProcess
 from exo.utils.channels import channel, mp_channel
-from exo.worker.runner.supervisor import RunnerSupervisor
+from exo.worker.runner.supervisor import RunnerStdioHandler, RunnerSupervisor
 from exo.worker.tests.unittests.conftest import get_bound_mlx_ring_instance
 
 
 class _DeadProcess:
-    exitcode = -6
+    def __init__(self):
+        rx1, _ = channel[bytes]()
+        rx2, _ = channel[bytes]()
+        self.stdout = rx1
+        self.stderr = rx2
 
-    def start(self) -> None:
-        return None
+    exitcode = -6
 
     def is_alive(self) -> bool:
         return False
 
-    def join(self, _timeout: float | None = None) -> None:
-        return None
 
-    def terminate(self) -> None:
-        return None
-
-    def kill(self) -> None:
-        return None
-
-
-@pytest.mark.asyncio
+@pytest.mark.anyio
 async def test_check_runner_emits_error_chunk_for_inflight_text_generation() -> None:
     event_sender, event_receiver = channel[Event]()
     task_sender, _ = mp_channel[Task]()
@@ -54,10 +48,15 @@ async def test_check_runner_emits_error_chunk_for_inflight_text_generation() ->
         node_id=NodeId("node-a"),
     )
 
+    proc = cast(AsyncProcess, cast(object, _DeadProcess()))
+    handler = await RunnerStdioHandler.create(
+        stdout_rx=proc.stdout, stderr_rx=proc.stderr
+    )
     supervisor = RunnerSupervisor(
         shard_metadata=bound_instance.bound_shard,
         bound_instance=bound_instance,
-        runner_process=cast("mp.Process", cast(object, _DeadProcess())),
+        runner_process=proc,
+        _runner_stdio_handler=handler,
         initialize_timeout=400,
         _ev_recv=ev_recv,
         _task_sender=task_sender,

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

@@ -1,126 +0,0 @@
-# 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()

tests/conftest.py

@@ -0,0 +1,181 @@
+# type: ignore
+"""Pytest configuration for marker-driven exo integration tests.
+
+Test authors declare requirements via markers:
+
+    @pytest.mark.cluster(count=2, thunderbolt='a2a')
+    @pytest.mark.instance('mlx-community/Llama-3.2-1B-Instruct-4bit',
+                          sharding='tensor', comm='jaccl')
+    def test_jaccl_inference(session):
+        resp = session.chat('What is 2+2?')
+        assert '4' in resp
+
+Clusters are cached by `ClusterSpec`; tests with the same cluster_spec
+share a deployment. Each test places its own instance (matching its
+`@pytest.mark.instance`), and instances are cleaned up after the test.
+
+Run with:
+    uv run pytest tests/ -v
+    uv run pytest tests/ -v --hosts s2,s4,s9,s10
+"""
+
+from __future__ import annotations
+
+import contextlib
+import json
+
+import pytest
+from exo_tools.cluster import ClusterInfo, EcoSession
+from exo_tools.harness import cleanup_all_instances, place_instance
+
+from .framework import (
+    ClusterSpec,
+    Session,
+    parse_cluster_marker,
+    parse_instance_marker,
+)
+
+# Single eco session for the entire test process.
+eco = EcoSession(user_prefix="test")
+
+# Cluster cache keyed by ClusterSpec — tests with the same spec share a deployment.
+# Cleared at session teardown.
+_cluster_cache: dict[ClusterSpec, ClusterInfo] = {}
+
+
+def pytest_addoption(parser):
+    parser.addoption(
+        "--hosts",
+        default=None,
+        help="Comma-separated list of hosts (e.g. s2,s4,s9,s10). "
+        "Overrides constraint-based reservation.",
+    )
+
+
+def pytest_configure(config):
+    """Register custom markers."""
+    config.addinivalue_line(
+        "markers",
+        "cluster(count=N, thunderbolt=Thunderbolt|None, min_memory=GB, chip=PATTERN): "
+        "declare cluster requirements for a test",
+    )
+    config.addinivalue_line(
+        "markers",
+        "instance(model_id, sharding=Sharding, comm=Comm, min_nodes=N): "
+        "declare instance placement for a test",
+    )
+
+
+def pytest_report_header(config):
+    """Show the eco user and hosts for this test session."""
+    hosts = config.getoption("--hosts")
+    lines = [f"eco user: {eco.user}"]
+    if hosts:
+        lines.append(f"hosts override: {hosts}")
+    return lines
+
+
+@pytest.fixture(scope="session")
+def _host_pool(request) -> list[str] | None:
+    raw = request.config.getoption("--hosts")
+    if raw:
+        return [h.strip() for h in raw.split(",") if h.strip()]
+    return None
+
+
+@pytest.fixture
+def session(request, _host_pool) -> Session:
+    """Per-test fixture providing a Session matching the test's markers.
+
+    Reads @pytest.mark.cluster and @pytest.mark.instance from the test, deploys
+    a matching cluster (cached across tests with the same spec), places the
+    model, and yields a Session for the test to interact with. Cleans up the
+    instance after the test, and invalidates the cluster cache if the test
+    left nodes disconnected.
+    """
+    cluster_marker = request.node.get_closest_marker("cluster")
+    instance_marker = request.node.get_closest_marker("instance")
+
+    cluster_spec = parse_cluster_marker(cluster_marker)
+    instance_spec = parse_instance_marker(instance_marker)
+
+    # Deploy or reuse a cluster matching the spec
+    cluster = _cluster_cache.get(cluster_spec)
+    if cluster is None:
+        if _host_pool:
+            cluster = eco.start_deploy(
+                hosts=_host_pool[: cluster_spec.count], wait=True
+            )
+        else:
+            cluster = eco.start_deploy(
+                count=cluster_spec.count,
+                thunderbolt=cluster_spec.thunderbolt,
+                chip=cluster_spec.chip,
+                min_memory_gb=cluster_spec.min_memory_gb,
+                wait=True,
+            )
+        _cluster_cache[cluster_spec] = cluster
+
+    # Place an instance for this test if the test specified one
+    instance_id = None
+    if instance_spec is not None:
+        client = cluster.make_client()
+        instance_id = place_instance(
+            client,
+            instance_spec.model_id,
+            sharding=instance_spec.sharding,
+            comm=instance_spec.comm,
+            min_nodes=instance_spec.min_nodes,
+        )
+
+    sess = Session(
+        cluster=cluster,
+        eco=eco,
+        instance_spec=instance_spec,
+        instance_id=instance_id,
+    )
+
+    yield sess
+
+    # ---- Teardown ----
+
+    # If the test left nodes disconnected, invalidate the cluster cache and
+    # stop the cluster so the next test deploys fresh.
+    if sess._stopped_hosts:
+        _cluster_cache.pop(cluster_spec, None)
+        with contextlib.suppress(Exception):
+            eco.stop(sess.cluster.hosts)
+        return
+
+    # Otherwise, clean up any instances created during the test
+    with contextlib.suppress(Exception):
+        cleanup_all_instances(sess.client)
+
+
+# ---------------------------------------------------------------------------
+# Session-level teardown — stop all cached clusters
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture(scope="session", autouse=True)
+def _teardown_clusters():
+    yield
+    for cluster in _cluster_cache.values():
+        with contextlib.suppress(Exception):
+            eco.stop(cluster.hosts)
+    _cluster_cache.clear()
+
+
+def pytest_runtest_makereport(item, call):
+    """Attach cluster logs to the test report when a test fails."""
+    if call.when != "call" or call.excinfo is None:
+        return
+
+    sess = item.funcargs.get("session")
+    if sess is None:
+        return
+    try:
+        logs = eco.logs(sess.cluster.hosts, lines=200)
+        item.add_report_section("call", "Cluster Logs", json.dumps(logs, indent=2))
+    except Exception:
+        pass

tests/framework.py

@@ -0,0 +1,199 @@
+"""Marker-driven test framework for exo integration tests.
+
+Test authors declare requirements via markers:
+
+    @pytest.mark.cluster(count=2, thunderbolt='a2a')
+    @pytest.mark.instance('mlx-community/Llama-3.2-1B-Instruct-4bit',
+                          sharding='tensor', comm='jaccl')
+    def test_jaccl_inference(session):
+        resp = session.chat('What is 2+2?')
+        assert '4' in resp
+
+The `session` fixture reads the markers, deploys the cluster, places the
+instance, and provides a `Session` object. All cluster/instance orchestration
+lives in `exo_tools.harness`; this module is purely the pytest-facing layer.
+"""
+
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass, field
+from typing import Any
+
+from exo_tools.client import ExoClient
+from exo_tools.cluster import (
+    Chip,
+    ClusterInfo,
+    EcoSession,
+    Thunderbolt,
+    make_client_from_url,
+)
+from exo_tools.harness import Comm, Sharding
+
+from exo.api.types.api import (
+    ChatCompletionChoice,
+    ChatCompletionRequest,
+    ChatCompletionResponse,
+)
+
+DEFAULT_MODEL = "mlx-community/Llama-3.2-1B-Instruct-4bit"
+
+
+def _extract_content(resp: ChatCompletionResponse) -> str:
+    """Extract plain-text content from a non-streaming chat completion."""
+    choice = resp.choices[0]
+    if not isinstance(choice, ChatCompletionChoice):
+        raise RuntimeError(
+            f"Expected non-streaming choice, got {type(choice).__name__}"
+        )
+    content = choice.message.content
+    if not isinstance(content, str):
+        raise RuntimeError(f"Expected string content, got {type(content).__name__}")
+    return content
+
+
+@dataclass(frozen=True)
+class ClusterSpec:
+    count: int = 1
+    thunderbolt: Thunderbolt | None = None
+    min_memory_gb: float | None = None
+    chip: Chip | None = None
+
+
+@dataclass(frozen=True)
+class InstanceSpec:
+    model_id: str
+    sharding: Sharding = Sharding.PIPELINE
+    comm: Comm = Comm.RING
+    min_nodes: int = 1
+
+
+def parse_cluster_marker(marker) -> ClusterSpec:
+    if marker is None:
+        return ClusterSpec()
+    return ClusterSpec(
+        count=marker.kwargs.get("count", 1),
+        thunderbolt=marker.kwargs.get("thunderbolt"),
+        min_memory_gb=marker.kwargs.get("min_memory"),
+        chip=marker.kwargs.get("chip"),
+    )
+
+
+def parse_instance_marker(marker) -> InstanceSpec | None:
+    if marker is None:
+        return None
+    if not marker.args:
+        raise ValueError(
+            "@pytest.mark.instance requires a positional model_id argument"
+        )
+    return InstanceSpec(
+        model_id=marker.args[0],
+        sharding=marker.kwargs.get("sharding", Sharding.PIPELINE),
+        comm=marker.kwargs.get("comm", Comm.RING),
+        min_nodes=marker.kwargs.get("min_nodes", 1),
+    )
+
+
+@dataclass
+class Session:
+    cluster: ClusterInfo
+    eco: EcoSession
+    instance_spec: InstanceSpec | None = None
+    instance_id: str | None = None
+    _stopped_hosts: set[str] = field(default_factory=set)
+
+    @property
+    def client(self) -> ExoClient:
+        for host in self.cluster.hosts:
+            if host not in self._stopped_hosts:
+                return make_client_from_url(self.cluster.api_endpoints[host])
+        return self.cluster.make_client()
+
+    @property
+    def state(self) -> dict[str, Any]:
+        return self.client.request_json("GET", "/state") or {}
+
+    @property
+    def instances(self) -> dict[str, Any]:
+        return self.state.get("instances", {})
+
+    # ---- Inference ----
+
+    def chat(self, prompt: str, max_tokens: int = 100) -> str:
+        resp = self.chat_raw(prompt, max_tokens=max_tokens)
+        return _extract_content(resp)
+
+    def chat_raw(self, prompt: str, **kwargs: Any) -> ChatCompletionResponse:
+        if not self.instance_spec:
+            raise RuntimeError(
+                "No instance placed; add @pytest.mark.instance to the test"
+            )
+        max_tokens = kwargs.pop("max_tokens", 100)
+        request = ChatCompletionRequest.model_validate(
+            {
+                "model": self.instance_spec.model_id,
+                "messages": [{"role": "user", "content": prompt}],
+                "max_tokens": max_tokens,
+                **kwargs,
+            }
+        )
+        return self._post_chat(request)
+
+    def multi_turn(self, messages: list[dict[str, str]], max_tokens: int = 100) -> str:
+        if not self.instance_spec:
+            raise RuntimeError(
+                "No instance placed; add @pytest.mark.instance to the test"
+            )
+        request = ChatCompletionRequest.model_validate(
+            {
+                "model": self.instance_spec.model_id,
+                "messages": messages,
+                "max_tokens": max_tokens,
+            }
+        )
+        return _extract_content(self._post_chat(request))
+
+    def _post_chat(self, request: ChatCompletionRequest) -> ChatCompletionResponse:
+        raw = self.client.request_json(
+            "POST",
+            "/v1/chat/completions",
+            body=request.model_dump(exclude_none=True),
+        )
+        return ChatCompletionResponse.model_validate(raw)
+
+    def disconnect_node(self, index: int) -> None:
+        """Stop exo on a node and wait for the cluster to observe the disconnect."""
+        host = self.cluster.hosts[index]
+        self.eco.stop([host], keep=True)
+        self._stopped_hosts.add(host)
+
+    def reconnect_node(self, index: int) -> None:
+        """Restart a previously disconnected node into the existing namespace."""
+        host = self.cluster.hosts[index]
+        self.eco.start_hosts([host], namespace=self.cluster.namespace)
+        self._stopped_hosts.discard(host)
+
+    def wait_ready(
+        self, expected_nodes: int | None = None, timeout: float = 60
+    ) -> None:
+        """Wait until the cluster has exactly `expected_nodes` visible and reporting memory.
+
+        Defaults to the count of non-stopped hosts. Use this after
+        `disconnect_node` / `reconnect_node` to wait for the cluster to settle.
+        """
+        if expected_nodes is None:
+            expected_nodes = len(self.cluster.hosts) - len(self._stopped_hosts)
+        start = time.time()
+        while time.time() - start < timeout:
+            try:
+                state = self.state
+                identities = len(state.get("nodeIdentities", {}))
+                memory = len(state.get("nodeMemory", {}))
+                if identities == expected_nodes and memory == expected_nodes:
+                    return
+            except Exception:
+                pass
+            time.sleep(2.0)
+        raise TimeoutError(
+            f"Cluster did not reach exactly {expected_nodes} ready nodes within {timeout}s"
+        )

tests/test_1node.py

@@ -0,0 +1,75 @@
+# type: ignore
+"""Single-node integration tests.
+
+Run with:
+    uv run pytest tests/test_1node.py -v
+"""
+
+from __future__ import annotations
+
+import time
+
+import pytest
+from exo_tools.harness import is_model_downloaded, place_instance
+
+from .framework import DEFAULT_MODEL, InstanceSpec
+
+
+@pytest.mark.cluster(count=1)
+@pytest.mark.instance(DEFAULT_MODEL)
+def test_place_instance_and_chat(session):
+    resp = session.chat("Say hello in one sentence.")
+    assert len(resp) > 0
+
+
+@pytest.mark.cluster(count=1)
+@pytest.mark.instance(DEFAULT_MODEL)
+def test_chat_multiple_turns(session):
+    first_reply = session.chat("What is 2 + 2?")
+    assert len(first_reply) > 0
+
+    second_reply = session.multi_turn(
+        [
+            {"role": "user", "content": "What is 2 + 2?"},
+            {"role": "assistant", "content": first_reply},
+            {"role": "user", "content": "Now multiply that by 3."},
+        ]
+    )
+    assert len(second_reply) > 0
+
+
+@pytest.mark.cluster(count=1)
+@pytest.mark.instance(DEFAULT_MODEL)
+def test_delete_instance(session):
+    from exo_tools.harness import wait_for_instance_gone
+
+    session.client.request_json("DELETE", f"/instance/{session.instance_id}")
+    wait_for_instance_gone(session.client, session.instance_id, timeout=30.0)
+    assert len(session.instances) == 0, (
+        f"Expected no instances, found {len(session.instances)}"
+    )
+
+
+@pytest.mark.cluster(count=1)
+def test_download_from_scratch(session):
+    """Ensure the model is not on the cluster, then place an instance to
+    trigger a fresh download and verify inference.
+    """
+    node_id = next(iter(session.state.get("nodeIdentities", {})))
+
+    # Delete any existing download — the API call is idempotent
+    session.client.request_json("DELETE", f"/download/{node_id}/{DEFAULT_MODEL}")
+
+    # Poll until the model is gone (it may already be gone)
+    deadline = time.time() + 60.0
+    while time.time() < deadline:
+        if not is_model_downloaded(session.client, DEFAULT_MODEL):
+            break
+        time.sleep(2.0)
+    else:
+        raise AssertionError(f"Expected {DEFAULT_MODEL} to be deleted from cluster")
+
+    place_instance(session.client, DEFAULT_MODEL, timeout=900.0)
+    session.instance_spec = InstanceSpec(model_id=DEFAULT_MODEL)
+    resp = session.chat("Say hello in one sentence.")
+    assert len(resp) > 0

tests/test_2node.py

@@ -0,0 +1,49 @@
+# type: ignore
+"""Two-node integration tests (ring + jaccl parallelism).
+
+Run with:
+    uv run pytest tests/test_2node.py -v
+"""
+
+from __future__ import annotations
+
+import pytest
+from exo_tools.cluster import Thunderbolt
+from exo_tools.harness import Comm, Sharding
+
+from .framework import DEFAULT_MODEL
+
+
+@pytest.mark.cluster(count=2, thunderbolt=Thunderbolt.A2A)
+@pytest.mark.instance(
+    DEFAULT_MODEL, sharding=Sharding.TENSOR, comm=Comm.JACCL, min_nodes=2
+)
+def test_2node_jaccl(session):
+    resp = session.chat("Say hello in one sentence.")
+    assert len(resp) > 0
+
+
+@pytest.mark.cluster(count=2, thunderbolt=Thunderbolt.A2A)
+@pytest.mark.instance(
+    DEFAULT_MODEL, sharding=Sharding.PIPELINE, comm=Comm.RING, min_nodes=2
+)
+def test_2node_ring(session):
+    resp = session.chat("Say hello in one sentence.")
+    assert len(resp) > 0
+
+
+@pytest.mark.cluster(count=2, thunderbolt=Thunderbolt.A2A)
+@pytest.mark.instance(
+    DEFAULT_MODEL, sharding=Sharding.TENSOR, comm=Comm.JACCL, min_nodes=2
+)
+def test_2node_jaccl_multi_turn(session):
+    first = session.chat("What is the capital of France?")
+    assert len(first) > 0
+    second = session.multi_turn(
+        [
+            {"role": "user", "content": "What is the capital of France?"},
+            {"role": "assistant", "content": first},
+            {"role": "user", "content": "What country is it in?"},
+        ]
+    )
+    assert len(second) > 0

tests/test_4node.py

@@ -0,0 +1,32 @@
+# type: ignore
+"""Four-node integration tests.
+
+Run with:
+    uv run pytest tests/test_4node.py -v
+"""
+
+from __future__ import annotations
+
+import pytest
+from exo_tools.cluster import Thunderbolt
+from exo_tools.harness import Comm, Sharding
+
+from .framework import DEFAULT_MODEL
+
+
+@pytest.mark.cluster(count=4, thunderbolt=Thunderbolt.A2A)
+@pytest.mark.instance(
+    DEFAULT_MODEL, sharding=Sharding.PIPELINE, comm=Comm.RING, min_nodes=4
+)
+def test_4node_pipeline_ring(session):
+    resp = session.chat("Say hello in one sentence.")
+    assert len(resp) > 0
+
+
+@pytest.mark.cluster(count=4, thunderbolt=Thunderbolt.A2A)
+@pytest.mark.instance(
+    DEFAULT_MODEL, sharding=Sharding.TENSOR, comm=Comm.JACCL, min_nodes=4
+)
+def test_4node_tensor_jaccl(session):
+    resp = session.chat("Say hello in one sentence.")
+    assert len(resp) > 0

tests/test_dashboard.py

@@ -0,0 +1,102 @@
+# type: ignore
+"""Dashboard end-to-end tests using Playwright (headless Chromium).
+
+Prerequisites:
+    uv run playwright install chromium
+
+Run with:
+    uv run pytest tests/test_dashboard.py -v
+"""
+
+from __future__ import annotations
+
+import contextlib
+
+import pytest
+
+try:
+    from playwright.sync_api import sync_playwright
+
+    _HAS_PLAYWRIGHT = True
+except ImportError:
+    _HAS_PLAYWRIGHT = False
+
+# Check if Chromium is installed by attempting a quick launch
+_HAS_CHROMIUM = False
+if _HAS_PLAYWRIGHT:
+    try:
+        with sync_playwright() as p:
+            browser = p.chromium.launch(headless=True)
+            browser.close()
+        _HAS_CHROMIUM = True
+    except Exception:
+        pass
+
+pytestmark = pytest.mark.skipif(
+    not _HAS_PLAYWRIGHT or not _HAS_CHROMIUM,
+    reason="playwright or chromium not installed (run: uv run playwright install chromium)",
+)
+
+
+def _mark_onboarding_complete(session) -> None:
+    """Mark onboarding complete on the server so the wizard doesn't auto-launch a model."""
+    with contextlib.suppress(Exception):
+        session.client.request_json("POST", "/onboarding")
+
+
+@pytest.mark.cluster(count=1)
+def test_dashboard_chat_inference(session):
+    """Full UI flow: open dashboard, pick a model, send a chat, verify response.
+
+    The instance is created via the dashboard UI (model picker → chat send
+    triggers the dashboard's auto-launch flow), not via @pytest.mark.instance.
+    """
+    _mark_onboarding_complete(session)
+
+    with sync_playwright() as p:
+        browser = p.chromium.launch(headless=True)
+        page = browser.new_page(viewport={"width": 1280, "height": 800})
+        page.goto(session.cluster.api_url, wait_until="networkidle")
+        page.wait_for_timeout(3000)
+        page.screenshot(path="/tmp/dashboard_initial.png")
+
+        # Open the model picker by clicking the "SELECT MODEL" button
+        page.get_by_text("SELECT MODEL", exact=False).first.click()
+        page.wait_for_timeout(1000)
+        page.screenshot(path="/tmp/dashboard_picker_open.png")
+
+        # Search for the model — uses the model id substring; the picker
+        # matches against name/id so "Llama-3.2-1B" filters to the small Llama.
+        search_input = page.locator('input[placeholder*="Search models"]').first
+        search_input.fill("Llama-3.2-1B")
+        page.wait_for_timeout(1500)
+        page.screenshot(path="/tmp/dashboard_picker_search.png")
+
+        # Click the only matching result. The picker shows the model's
+        # display name (e.g. "Llama 3.2 1B") which differs from the model_id.
+        # We click the first visible button-like row in the result list.
+        page.get_by_text("Llama 3.2 1B", exact=False).first.click()
+        page.wait_for_timeout(1500)
+        page.screenshot(path="/tmp/dashboard_model_selected.png")
+
+        # Type a chat message — sending triggers the dashboard's auto-launch
+        # flow: it picks an optimal placement for the selected model and POSTs
+        # to /instance, then sends the chat once the runner is ready.
+        chat_input = page.locator("textarea").first
+        chat_input.fill("Say hello")
+        chat_input.press("Enter")
+        page.screenshot(path="/tmp/dashboard_chat_sent.png")
+
+        # Wait for the instance to launch and respond. Generous timeout
+        # because this includes model placement + load + generation.
+        page.wait_for_timeout(60000)
+        page.screenshot(path="/tmp/dashboard_after_chat.png")
+
+        # Verify an instance was created and the chat got a response
+        instances = session.client.request_json("GET", "/state").get("instances", {})
+        assert len(instances) > 0, "Expected the dashboard to have created an instance"
+
+        body_text = page.text_content("body") or ""
+        assert len(body_text) > 0
+
+        browser.close()

tests/test_resilience.py

@@ -0,0 +1,56 @@
+# type: ignore
+"""Resilience tests: disconnect/reconnect nodes and verify cluster recovery.
+
+Run with:
+    uv run pytest tests/test_resilience.py -v
+"""
+
+from __future__ import annotations
+
+import pytest
+from exo_tools.cluster import Thunderbolt
+from exo_tools.harness import Comm, Sharding, cleanup_all_instances, place_instance
+
+from .framework import DEFAULT_MODEL, InstanceSpec
+
+
+@pytest.mark.cluster(count=2, thunderbolt=Thunderbolt.A2A)
+@pytest.mark.instance(
+    DEFAULT_MODEL, sharding=Sharding.PIPELINE, comm=Comm.RING, min_nodes=2
+)
+def test_node_recovery(session):
+    """Full disconnect/reconnect cycle.
+
+    1. Place a 2-node instance, verify inference
+    2. Disconnect one node
+    3. Place a 1-node instance on remaining node, verify inference
+    4. Reconnect the stopped node, wait for the cluster to reform
+    5. Place a 2-node instance again, verify inference
+    """
+    # --- Phase 1: 2-node inference ---
+    resp = session.chat("Hello")
+    assert len(resp) > 0
+
+    # --- Phase 2: disconnect one node ---
+    session.disconnect_node(1)
+    session.wait_ready(60)
+
+    # Clean up the now-broken 2-node instance
+    cleanup_all_instances(session.client)
+
+    # --- Phase 3: 1-node inference on the remaining node ---
+    place_instance(session.client, DEFAULT_MODEL, min_nodes=1)
+    session.instance_spec = InstanceSpec(model_id=DEFAULT_MODEL, min_nodes=1)
+    resp = session.chat("Hello")
+    assert len(resp) > 0
+
+    # --- Phase 4: reconnect and restore 2-node cluster ---
+    cleanup_all_instances(session.client)
+    session.reconnect_node(1)
+    session.wait_ready(60)
+
+    # --- Phase 5: 2-node inference again ---
+    place_instance(session.client, DEFAULT_MODEL, min_nodes=2)
+    session.instance_spec = InstanceSpec(model_id=DEFAULT_MODEL, min_nodes=2)
+    resp = session.chat("Hello again")
+    assert len(resp) > 0

tools/pyproject.toml

@@ -0,0 +1,10 @@
+[project]
+name = "exo-tools"
+version = "0.1.0"
+description = "Shared tooling for interacting with exo clusters"
+requires-python = ">=3.13"
+dependencies = ["loguru>=0.7.3"]
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"

tools/src/exo_tools/client.py

@@ -0,0 +1,117 @@
+# type: ignore
+"""HTTP client for the exo API."""
+
+from __future__ import annotations
+
+import http.client
+import json
+from collections.abc import Iterator
+from typing import Any
+from urllib.parse import urlencode
+
+
+class ExoHttpError(RuntimeError):
+    def __init__(self, status: int, reason: str, body_preview: str):
+        super().__init__(f"HTTP {status} {reason}: {body_preview}")
+        self.status = status
+
+
+class ExoClient:
+    def __init__(self, host: str, port: int, timeout_s: float = 7200.0):
+        self.host = host
+        self.port = port
+        self.timeout_s = timeout_s
+
+    def request_json(
+        self,
+        method: str,
+        path: str,
+        params: dict[str, Any] | None = None,
+        body: dict[str, Any] | None = None,
+        headers: dict[str, str] | None = None,
+    ) -> Any:
+        if not path.startswith("/"):
+            path = "/" + path
+        if params:
+            path = path + "?" + urlencode(params)
+
+        conn = http.client.HTTPConnection(self.host, self.port, timeout=self.timeout_s)
+        try:
+            payload: bytes | None = None
+            hdrs: dict[str, str] = {"Accept": "application/json"}
+
+            if body is not None:
+                payload = json.dumps(body).encode("utf-8")
+                hdrs["Content-Type"] = "application/json"
+            if headers:
+                hdrs.update(headers)
+
+            conn.request(method.upper(), path, body=payload, headers=hdrs)
+            resp = conn.getresponse()
+            raw = resp.read()
+            text = raw.decode("utf-8", errors="replace") if raw else ""
+
+            if resp.status >= 400:
+                raise ExoHttpError(resp.status, resp.reason, text[:300])
+
+            if not text:
+                return None
+            return json.loads(text)
+        finally:
+            conn.close()
+
+    def post_bench_chat_completions(self, payload: dict[str, Any]) -> dict[str, Any]:
+        return self.request_json("POST", "/bench/chat/completions", body=payload)
+
+    def stream_bench_chat_completions(self, payload: dict[str, Any]) -> Iterator[str]:
+        """POST /bench/chat/completions with stream=True, yielding raw SSE lines."""
+        payload = {**payload, "stream": True}
+        data = json.dumps(payload).encode("utf-8")
+        conn = http.client.HTTPConnection(self.host, self.port, timeout=self.timeout_s)
+        try:
+            conn.request(
+                "POST",
+                "/bench/chat/completions",
+                body=data,
+                headers={
+                    "Content-Type": "application/json",
+                    "Accept": "text/event-stream",
+                },
+            )
+            resp = conn.getresponse()
+            if resp.status >= 400:
+                raw = resp.read().decode("utf-8", errors="replace")
+                raise ExoHttpError(resp.status, resp.reason, raw[:300])
+            for line in resp:
+                yield line.decode("utf-8", errors="replace")
+        finally:
+            conn.close()
+
+    def get_state_path(self, path: str) -> Any:
+        try:
+            return self.request_json("GET", f"/state/{path}")
+        except ExoHttpError as e:
+            if e.status == 404:
+                return None
+            raise
+
+    def get_instance(self, instance_id: str) -> dict[str, Any] | None:
+        return self.get_state_path(f"instances/{instance_id}")
+
+    def get_runner(self, runner_id: str) -> dict[str, Any] | None:
+        return self.get_state_path(f"runners/{runner_id}")
+
+    def get_node_downloads(self, node_id: str) -> list[dict[str, Any]] | None:
+        return self.get_state_path(f"downloads/{node_id}")
+
+    def get_node_disk(self, node_id: str) -> dict[str, Any] | None:
+        return self.get_state_path(f"nodeDisk/{node_id}")
+
+    def get_node_system(self, node_id: str) -> dict[str, Any] | None:
+        return self.get_state_path(f"nodeSystem/{node_id}")
+
+    def get_node_identities(self) -> dict[str, Any] | None:
+        return self.get_state_path("nodeIdentities")
+
+    def get_topology(self) -> dict[str, Any] | None:
+        return self.get_state_path("topology")

tools/src/exo_tools/cluster.py

@@ -0,0 +1,243 @@
+# type: ignore
+"""Cluster lifecycle management via eco.
+
+Provides subprocess wrappers for eco commands (deploy, stop, start, release,
+logs, exec) and a ClusterInfo dataclass. Reusable by integration tests,
+bench, eval, and CI workflows.
+"""
+
+from __future__ import annotations
+
+import atexit
+import contextlib
+import json
+import logging
+import os
+import signal
+import subprocess
+import uuid
+from dataclasses import dataclass, field
+from enum import Enum
+
+from .client import ExoClient
+
+
+class Thunderbolt(str, Enum):
+    A2A = "a2a"  # all-to-all (eco --tb-a2a)
+    RING = "ring"  # ring topology (eco --tb-ring)
+
+
+class Chip(str, Enum):
+    M1 = "M1"
+    M1_PRO = "M1 Pro"
+    M1_MAX = "M1 Max"
+    M1_ULTRA = "M1 Ultra"
+    M2 = "M2"
+    M2_PRO = "M2 Pro"
+    M2_MAX = "M2 Max"
+    M2_ULTRA = "M2 Ultra"
+    M3 = "M3"
+    M3_PRO = "M3 Pro"
+    M3_MAX = "M3 Max"
+    M3_ULTRA = "M3 Ultra"
+    M4 = "M4"
+    M4_PRO = "M4 Pro"
+    M4_MAX = "M4 Max"
+    M4_ULTRA = "M4 Ultra"
+
+
+logger = logging.getLogger("exo_tools.cluster")
+
+# When set, deploy from a GitHub branch/tag instead of local source (rsync).
+_EXO_REF = os.environ.get("EXO_REF")
+
+
+@dataclass
+class ClusterInfo:
+    """Holds the result of an `eco start --deploy` invocation."""
+
+    hosts: list[str]
+    namespace: str
+    api_endpoints: dict[str, str]  # host -> url
+    api_url: str  # primary endpoint for ExoClient
+
+    primary_host: str = ""
+    _host: str = field(init=False, repr=False, default="")
+    _port: int = field(init=False, repr=False, default=52415)
+
+    def __post_init__(self) -> None:
+        if not self.primary_host:
+            self.primary_host = self.hosts[0]
+        url = self.api_url.replace("http://", "").replace("https://", "")
+        parts = url.split(":")
+        self._host = parts[0]
+        self._port = int(parts[1]) if len(parts) > 1 else 52415
+
+    def make_client(self, timeout_s: float = 7200.0) -> ExoClient:
+        return ExoClient(self._host, self._port, timeout_s=timeout_s)
+
+
+class EcoSession:
+    """Manages an eco session with a unique user and automatic cleanup.
+
+    Usage:
+        session = EcoSession(user_prefix="test")
+        cluster = session.start_deploy(count=2, thunderbolt=True)
+        ...
+        session.stop_all()  # or let atexit handle it
+
+    The session registers atexit and signal handlers to ensure cleanup
+    on normal exit, uncaught exceptions, SIGTERM, and SIGHUP. SIGINT
+    is left unhandled so KeyboardInterrupt propagates normally.
+    """
+
+    def __init__(self, user_prefix: str = "test") -> None:
+        self._session_id = uuid.uuid4().hex[:8]
+        self.user = f"{user_prefix}-{self._session_id}"
+        self._env = {**os.environ, "USER": self.user}
+
+        # Register cleanup handlers
+        atexit.register(self.stop_all)
+        for sig in (signal.SIGTERM, signal.SIGHUP):
+            signal.signal(sig, self._signal_handler)
+
+    def _signal_handler(self, signum: int, _frame: object) -> None:
+        self.stop_all()
+        raise SystemExit(128 + signum)
+
+    def stop_all(self) -> None:
+        """Stop all clusters and release all reservations for this session."""
+        with contextlib.suppress(Exception):
+            subprocess.run(
+                ["eco", "stop"],
+                capture_output=True,
+                text=True,
+                timeout=30,
+                env=self._env,
+            )
+
+    def _run(
+        self, args: list[str], *, check: bool = True, timeout: int = 120
+    ) -> subprocess.CompletedProcess[str]:
+        """Run an eco command as this session's user.
+
+        stdout is captured (JSON output), stderr is passed through to the
+        console so eco's progress messages are visible.
+        """
+        logger.info(f"eco: {' '.join(args)}")
+        return subprocess.run(
+            args,
+            stdout=subprocess.PIPE,
+            stderr=None,
+            text=True,
+            check=check,
+            timeout=timeout,
+            env=self._env,
+        )
+
+    def start_deploy(
+        self,
+        hosts: list[str] | None = None,
+        *,
+        count: int | None = None,
+        thunderbolt: Thunderbolt | None = None,
+        chip: Chip | None = None,
+        min_memory_gb: float | None = None,
+        wait: bool = True,
+        ref: str | None = _EXO_REF,
+        timeout: int = 600,
+    ) -> ClusterInfo:
+        """Start and deploy exo on a set of hosts via eco.
+
+        By default, deploys from local source via rsync. Set EXO_REF
+        or pass ref= to deploy from a GitHub branch/tag instead (for CI).
+        """
+        cmd: list[str] = ["eco", "--json", "start", "--deploy"]
+        if hosts:
+            cmd.extend(hosts)
+        if count is not None:
+            cmd.extend(["--count", str(count)])
+        if thunderbolt is not None:
+            cmd.append(f"--tb-{thunderbolt.value}")
+        if chip is not None:
+            cmd.extend(["--chip", chip.value])
+        if min_memory_gb is not None:
+            cmd.extend(["--min-memory", str(min_memory_gb)])
+        if wait:
+            cmd.append("--wait")
+        if ref:
+            cmd.extend(["--ref", ref])
+
+        result = self._run(cmd, timeout=timeout)
+        data = json.loads(result.stdout)["data"]
+        endpoints: dict[str, str] = data["api_endpoints"]
+        primary_host = data["hosts"][0]
+
+        return ClusterInfo(
+            hosts=data["hosts"],
+            namespace=data["namespace"],
+            api_endpoints=endpoints,
+            api_url=endpoints[primary_host],
+            primary_host=primary_host,
+        )
+
+    def stop(self, hosts: list[str], *, keep: bool = False, timeout: int = 120) -> None:
+        """Stop exo on the given hosts. If keep=True, keep the reservation."""
+        cmd: list[str] = ["eco", "stop"]
+        cmd.extend(hosts)
+        if keep:
+            cmd.append("--keep")
+        self._run(cmd, timeout=timeout)
+
+    def start_hosts(
+        self, hosts: list[str], *, namespace: str, timeout: int = 300
+    ) -> None:
+        """Start (previously stopped) hosts back into an existing namespace."""
+        cmd: list[str] = ["eco", "--json", "start"]
+        cmd.extend(hosts)
+        cmd.extend(["--namespace", namespace])
+        self._run(cmd, timeout=timeout)
+
+    def release(self, hosts: list[str], timeout: int = 120) -> None:
+        """Release hosts from the reservation."""
+        cmd: list[str] = ["eco", "release"]
+        cmd.extend(hosts)
+        self._run(cmd, timeout=timeout)
+
+    def logs(
+        self, hosts: list[str], lines: int = 500, timeout: int = 60
+    ) -> dict[str, list[str]]:
+        """Fetch recent logs from cluster hosts."""
+        cmd: list[str] = ["eco", "--json", "logs"]
+        cmd.extend(hosts)
+        cmd.extend(["-n", str(lines), "--raw"])
+        result = self._run(cmd, check=False, timeout=timeout)
+        if result.returncode != 0:
+            return {"_error": [result.stderr]}
+        try:
+            return json.loads(result.stdout)
+        except json.JSONDecodeError:
+            return {"_raw": result.stdout.splitlines()}
+
+    def exec(self, hosts: list[str], command: str, timeout: int = 120) -> str:
+        """Run an arbitrary command on the given hosts via eco."""
+        cmd: list[str] = ["eco", "exec"]
+        cmd.extend(hosts)
+        cmd.append("--")
+        cmd.extend(command.split())
+        result = self._run(cmd, check=False, timeout=timeout)
+        return result.stdout
+
+
+def make_client(cluster: ClusterInfo, timeout_s: float = 7200.0) -> ExoClient:
+    """Create an ExoClient from a ClusterInfo."""
+    return cluster.make_client(timeout_s=timeout_s)
+
+
+def make_client_from_url(url: str, timeout_s: float = 7200.0) -> ExoClient:
+    """Create an ExoClient from a URL string like 'http://host:port'."""
+    url_clean = url.replace("http://", "").replace("https://", "")
+    parts = url_clean.split(":")
+    host = parts[0]
+    port = int(parts[1]) if len(parts) > 1 else 52415
+    return ExoClient(host, port, timeout_s=timeout_s)

tools/src/exo_tools/harness.py

@@ -1,129 +1,39 @@
 # type: ignore
+"""Instance lifecycle helpers for exo clusters.
+
+Provides utilities for placing instances, waiting for readiness,
+managing downloads, filtering placements, and common CLI arguments.
+"""
+
 from __future__ import annotations
 
 import argparse
-import http.client
-import json
+import contextlib
 import os
 import time
-from collections.abc import Iterator
+from enum import Enum
 from typing import Any
-from urllib.parse import urlencode
 
 from loguru import logger
 
+from .client import ExoClient, ExoHttpError
+
+
+class Sharding(str, Enum):
+    PIPELINE = "Pipeline"  # layers split across nodes
+    TENSOR = "Tensor"  # layers split within (across nodes)
+
+
+class Comm(str, Enum):
+    RING = "MlxRing"  # ring all-reduce over network
+    JACCL = "MlxJaccl"  # RDMA over Thunderbolt
+
+
 _SETTLE_INITIAL_BACKOFF_S = 1.0
 _SETTLE_MAX_BACKOFF_S = 60.0
 _SETTLE_BACKOFF_MULTIPLIER = 2.0
 
 
-class ExoHttpError(RuntimeError):
-    def __init__(self, status: int, reason: str, body_preview: str):
-        super().__init__(f"HTTP {status} {reason}: {body_preview}")
-        self.status = status
-
-
-class ExoClient:
-    def __init__(self, host: str, port: int, timeout_s: float = 7200.0):
-        self.host = host
-        self.port = port
-        self.timeout_s = timeout_s
-
-    def request_json(
-        self,
-        method: str,
-        path: str,
-        params: dict[str, Any] | None = None,
-        body: dict[str, Any] | None = None,
-        headers: dict[str, str] | None = None,
-    ) -> Any:
-        if not path.startswith("/"):
-            path = "/" + path
-        if params:
-            path = path + "?" + urlencode(params)
-
-        conn = http.client.HTTPConnection(self.host, self.port, timeout=self.timeout_s)
-        try:
-            payload: bytes | None = None
-            hdrs: dict[str, str] = {"Accept": "application/json"}
-
-            if body is not None:
-                payload = json.dumps(body).encode("utf-8")
-                hdrs["Content-Type"] = "application/json"
-            if headers:
-                hdrs.update(headers)
-
-            conn.request(method.upper(), path, body=payload, headers=hdrs)
-            resp = conn.getresponse()
-            raw = resp.read()
-            text = raw.decode("utf-8", errors="replace") if raw else ""
-
-            if resp.status >= 400:
-                raise ExoHttpError(resp.status, resp.reason, text[:300])
-
-            if not text:
-                return None
-            return json.loads(text)
-        finally:
-            conn.close()
-
-    def post_bench_chat_completions(self, payload: dict[str, Any]) -> dict[str, Any]:
-        return self.request_json("POST", "/bench/chat/completions", body=payload)
-
-    def stream_bench_chat_completions(self, payload: dict[str, Any]) -> Iterator[str]:
-        """POST /bench/chat/completions with stream=True, yielding raw SSE lines."""
-        payload = {**payload, "stream": True}
-        data = json.dumps(payload).encode("utf-8")
-        conn = http.client.HTTPConnection(self.host, self.port, timeout=self.timeout_s)
-        try:
-            conn.request(
-                "POST",
-                "/bench/chat/completions",
-                body=data,
-                headers={
-                    "Content-Type": "application/json",
-                    "Accept": "text/event-stream",
-                },
-            )
-            resp = conn.getresponse()
-            if resp.status >= 400:
-                raw = resp.read().decode("utf-8", errors="replace")
-                raise ExoHttpError(resp.status, resp.reason, raw[:300])
-            for line in resp:
-                yield line.decode("utf-8", errors="replace")
-        finally:
-            conn.close()
-
-    def get_state_path(self, path: str) -> Any:
-        try:
-            return self.request_json("GET", f"/state/{path}")
-        except ExoHttpError as e:
-            if e.status == 404:
-                return None
-            raise
-
-    def get_instance(self, instance_id: str) -> dict[str, Any] | None:
-        return self.get_state_path(f"instances/{instance_id}")
-
-    def get_runner(self, runner_id: str) -> dict[str, Any] | None:
-        return self.get_state_path(f"runners/{runner_id}")
-
-    def get_node_downloads(self, node_id: str) -> list[dict[str, Any]] | None:
-        return self.get_state_path(f"downloads/{node_id}")
-
-    def get_node_disk(self, node_id: str) -> dict[str, Any] | None:
-        return self.get_state_path(f"nodeDisk/{node_id}")
-
-    def get_node_system(self, node_id: str) -> dict[str, Any] | None:
-        return self.get_state_path(f"nodeSystem/{node_id}")
-
-    def get_node_identities(self) -> dict[str, Any] | None:
-        return self.get_state_path("nodeIdentities")
-
-    def get_topology(self) -> dict[str, Any] | None:
-        return self.get_state_path("topology")
-
-
 def unwrap_instance(instance: dict[str, Any]) -> dict[str, Any]:
     if len(instance) != 1:
         raise KeyError(f"Expected 1 key, got keys={list(instance.keys())}")
@@ -555,7 +465,6 @@ def find_existing_instance(client: ExoClient, model_id: str) -> str | None:
     except Exception:
         return None
     for inst_id, inst in state.get("instances", {}).items():
-        # Instance structure is nested: {"MlxJacclInstance": {"shardAssignments": {"modelId": ...}}}
         for _inst_type, inner in inst.items():
             if not isinstance(inner, dict):
                 continue
@@ -623,3 +532,112 @@ def add_common_instance_args(ap: argparse.ArgumentParser) -> None:
         action="store_true",
         help="Reuse an existing running instance for this model instead of creating a new one.",
     )
+
+
+# ---------------------------------------------------------------------------
+# Cluster/instance orchestration helpers (used by tests, bench, eval)
+# ---------------------------------------------------------------------------
+
+
+def get_instance_ids(client: ExoClient) -> set[str]:
+    """Return the set of current instance IDs from cluster state."""
+    state = client.request_json("GET", "/state") or {}
+    result: set[str] = set()
+    for instance in state.get("instances", {}).values():
+        with contextlib.suppress(Exception):
+            result.add(instance_id_from_instance(instance))
+    return result
+
+
+def wait_for_cluster_ready(
+    client: ExoClient, expected_nodes: int = 1, timeout: float = 120.0
+) -> None:
+    """Wait until the cluster has all expected nodes visible and reporting memory.
+
+    Placement requires nodeMemory for all nodes in a cycle. This polls until
+    both nodeIdentities and nodeMemory have at least `expected_nodes` entries.
+    """
+    start = time.time()
+    while time.time() - start < timeout:
+        try:
+            state = client.request_json("GET", "/state") or {}
+            if (
+                len(state.get("nodeIdentities", {})) >= expected_nodes
+                and len(state.get("nodeMemory", {})) >= expected_nodes
+            ):
+                return
+        except Exception:
+            pass
+        time.sleep(1.0)
+    raise TimeoutError(f"Cluster not ready: expected {expected_nodes} nodes")
+
+
+def place_instance(
+    client: ExoClient,
+    model_id: str,
+    *,
+    sharding: Sharding = Sharding.PIPELINE,
+    comm: Comm = Comm.RING,
+    min_nodes: int = 1,
+    timeout: float = 600.0,
+    placement_retries: int = 10,
+    placement_retry_delay: float = 10.0,
+) -> str:
+    """Place an instance and wait for it to be ready. Returns the instance_id.
+
+    The /place_instance API returns a command_id, but instances are stored
+    under a separately-generated instance_id. This polls cluster state for the
+    new instance, retrying placement if the cluster is still settling.
+    """
+    wait_for_cluster_ready(client, expected_nodes=min_nodes)
+
+    body = {
+        "model_id": model_id,
+        "sharding": sharding.value,
+        "instance_meta": comm.value,
+        "min_nodes": min_nodes,
+    }
+
+    instance_id: str | None = None
+    for attempt in range(placement_retries):
+        before_ids = get_instance_ids(client)
+        client.request_json("POST", "/place_instance", body=body)
+
+        poll_deadline = time.time() + 30.0
+        while time.time() < poll_deadline:
+            new_ids = get_instance_ids(client) - before_ids
+            if new_ids:
+                instance_id = next(iter(new_ids))
+                break
+            time.sleep(1.0)
+
+        if instance_id is not None:
+            break
+
+        if attempt < placement_retries - 1:
+            time.sleep(placement_retry_delay)
+
+    if instance_id is None:
+        raise TimeoutError(
+            f"Placement failed after {placement_retries} attempts "
+            f"({sharding.value}/{comm.value} for {model_id})"
+        )
+
+    wait_for_instance_ready(client, instance_id, timeout=timeout)
+    return instance_id
+
+
+def cleanup_all_instances(client: ExoClient) -> None:
+    """Remove all running instances from the cluster."""
+    state = client.request_json("GET", "/state") or {}
+    for instance in state.get("instances", {}).values():
+        with contextlib.suppress(Exception):
+            iid = instance_id_from_instance(instance)
+            client.request_json("DELETE", f"/instance/{iid}")
+            wait_for_instance_gone(client, iid, timeout=30.0)
+
+
+def is_model_downloaded(client: ExoClient, model_id: str) -> bool:
+    response = client.request_json("GET", "/models", params={"status": "downloaded"})
+    data = (response or {}).get("data", [])
+    return all(model.get("id") == model_id for model in data)

uv.lock

@@ -23,6 +23,7 @@ members = [
     "exo",
     "exo-bench",
     "exo-pyo3-bindings",
+    "exo-tools",
 ]
 constraints = [{ name = "transformers", specifier = ">=5.6.2" }]
 overrides = [
@@ -394,7 +395,7 @@ dependencies = [
     { name = "loguru", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
     { name = "mflux", marker = "sys_platform == 'darwin' or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
     { name = "mlx", version = "0.31.1", source = { registry = "https://pypi.org/simple" }, marker = "sys_platform == 'linux' or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
-    { name = "mlx", version = "0.32.0.dev20260427+cc3f3e60", source = { git = "https://github.com/rltakashige/mlx-jaccl-fix-small-recv.git?branch=address-rdma-gpu-locks#cc3f3e60be1289506125f2fa19b73b05aa770df8" }, marker = "sys_platform == 'darwin' or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
+    { name = "mlx", version = "0.32.0.dev20260511+e9835615", source = { git = "https://github.com/rltakashige/mlx-jaccl-fix-small-recv.git?branch=address-rdma-gpu-locks#e9835615badaf4ad7c95301f2a9bd58d782b38db" }, marker = "sys_platform == 'darwin' or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
     { name = "mlx-lm", marker = "sys_platform == 'darwin' or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
     { name = "mlx-vlm", marker = "sys_platform == 'darwin' or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
     { name = "msgspec", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
@@ -416,21 +417,21 @@ build = [
 ]
 cpu = [
     { name = "mlx", version = "0.31.1", source = { registry = "https://pypi.org/simple" }, marker = "(sys_platform == 'linux' and extra == 'extra-3-exo-cpu') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
-    { name = "mlx", version = "0.32.0.dev20260427+cc3f3e60", source = { git = "https://github.com/rltakashige/mlx-jaccl-fix-small-recv.git?branch=address-rdma-gpu-locks#cc3f3e60be1289506125f2fa19b73b05aa770df8" }, marker = "(sys_platform == 'darwin' and extra == 'extra-3-exo-cpu') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
+    { name = "mlx", version = "0.32.0.dev20260511+e9835615", source = { git = "https://github.com/rltakashige/mlx-jaccl-fix-small-recv.git?branch=address-rdma-gpu-locks#e9835615badaf4ad7c95301f2a9bd58d782b38db" }, marker = "(sys_platform == 'darwin' and extra == 'extra-3-exo-cpu') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
     { name = "mlx-cpu", marker = "sys_platform == 'linux'" },
     { name = "mlx-lm", marker = "sys_platform == 'linux'" },
     { name = "mlx-vlm", marker = "sys_platform == 'linux'" },
 ]
 cuda12 = [
     { name = "mlx", version = "0.31.1", source = { registry = "https://pypi.org/simple" }, marker = "(sys_platform == 'linux' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13')" },
-    { name = "mlx", version = "0.32.0.dev20260427+cc3f3e60", source = { git = "https://github.com/rltakashige/mlx-jaccl-fix-small-recv.git?branch=address-rdma-gpu-locks#cc3f3e60be1289506125f2fa19b73b05aa770df8" }, marker = "(sys_platform == 'darwin' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13')" },
+    { name = "mlx", version = "0.32.0.dev20260511+e9835615", source = { git = "https://github.com/rltakashige/mlx-jaccl-fix-small-recv.git?branch=address-rdma-gpu-locks#e9835615badaf4ad7c95301f2a9bd58d782b38db" }, marker = "(sys_platform == 'darwin' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13')" },
     { name = "mlx-cuda-12", marker = "sys_platform == 'linux'" },
     { name = "mlx-lm", marker = "sys_platform == 'linux'" },
     { name = "mlx-vlm", marker = "sys_platform == 'linux'" },
 ]
 cuda13 = [
     { name = "mlx", version = "0.31.1", source = { registry = "https://pypi.org/simple" }, marker = "(sys_platform == 'linux' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13')" },
-    { name = "mlx", version = "0.32.0.dev20260427+cc3f3e60", source = { git = "https://github.com/rltakashige/mlx-jaccl-fix-small-recv.git?branch=address-rdma-gpu-locks#cc3f3e60be1289506125f2fa19b73b05aa770df8" }, marker = "(sys_platform == 'darwin' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13')" },
+    { name = "mlx", version = "0.32.0.dev20260511+e9835615", source = { git = "https://github.com/rltakashige/mlx-jaccl-fix-small-recv.git?branch=address-rdma-gpu-locks#e9835615badaf4ad7c95301f2a9bd58d782b38db" }, marker = "(sys_platform == 'darwin' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13')" },
     { name = "mlx-cuda-13", marker = "sys_platform == 'linux'" },
     { name = "mlx-lm", marker = "sys_platform == 'linux'" },
     { name = "mlx-vlm", marker = "sys_platform == 'linux'" },
@@ -439,6 +440,7 @@ cuda13 = [
 [package.dev-dependencies]
 dev = [
     { name = "basedpyright", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
+    { name = "playwright", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
     { name = "pyinstaller", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
     { name = "pytest", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
     { name = "pytest-asyncio", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
@@ -498,6 +500,7 @@ provides-extras = ["build", "cpu", "cuda12", "cuda13"]
 [package.metadata.requires-dev]
 dev = [
     { name = "basedpyright", specifier = ">=1.29.0" },
+    { name = "playwright", specifier = ">=1.52.0" },
     { name = "pyinstaller", specifier = ">=6.17.0" },
     { name = "pytest", specifier = ">=8.4.0" },
     { name = "pytest-asyncio", specifier = ">=1.0.0" },
@@ -542,7 +545,7 @@ requires-dist = [
 
 [[package]]
 name = "exo-pyo3-bindings"
-version = "0.2.1"
+version = "0.2.2"
 source = { editable = "rust/exo_pyo3_bindings" }
 
 [package.dev-dependencies]
@@ -561,6 +564,17 @@ dev = [
     { name = "pytest-asyncio", specifier = ">=1.0.0" },
 ]
 
+[[package]]
+name = "exo-tools"
+version = "0.1.0"
+source = { editable = "tools" }
+dependencies = [
+    { name = "loguru", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
+]
+
+[package.metadata]
+requires-dist = [{ name = "loguru", specifier = ">=0.7.3" }]
+
 [[package]]
 name = "fastapi"
 version = "0.128.0"
@@ -669,6 +683,24 @@ http = [
     { name = "aiohttp", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
 ]
 
+[[package]]
+name = "greenlet"
+version = "3.5.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/3c/3f/dbf99fb14bfeb88c28f16729215478c0e265cacd6dc22270c8f31bb6892f/greenlet-3.5.0.tar.gz", hash = "sha256:d419647372241bc68e957bf38d5c1f98852155e4146bd1e4121adea81f4f01e4", size = 196995, upload-time = "2026-04-27T13:37:15.544Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/0c/58/fc576f99037ce19c5aa16628e4c3226b6d1419f72a62c79f5f40576e6eb3/greenlet-3.5.0-cp313-cp313-macosx_11_0_universal2.whl", hash = "sha256:5a5ed18de6a0f6cc7087f1563f6bd93fc7df1c19165ca01e9bde5a5dc281d106", size = 285066, upload-time = "2026-04-27T12:23:05.033Z" },
+    { url = "https://files.pythonhosted.org/packages/4a/ba/b28ddbe6bfad6a8ac196ef0e8cff37bc65b79735995b9e410923fffeeb70/greenlet-3.5.0-cp313-cp313-manylinux_2_24_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:3a717fbc46d8a354fa675f7c1e813485b6ba3885f9bef0cd56e5ba27d758ff5b", size = 604414, upload-time = "2026-04-27T12:52:42.358Z" },
+    { url = "https://files.pythonhosted.org/packages/09/06/4b69f8f0b67603a8be2790e55107a190b376f2627fe0eaf5695d85ffb3cd/greenlet-3.5.0-cp313-cp313-manylinux_2_24_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:ddc090c5c1792b10246a78e8c2163ebbe04cf877f9d785c230a7b27b39ad038e", size = 617349, upload-time = "2026-04-27T12:59:43.32Z" },
+    { url = "https://files.pythonhosted.org/packages/6a/15/a643b4ecd09969e30b8a150d5919960caae0abe4f5af75ab040b1ab85e78/greenlet-3.5.0-cp313-cp313-manylinux_2_24_s390x.manylinux_2_28_s390x.whl", hash = "sha256:4964101b8585c144cbda5532b1aa644255126c08a265dae90c16e7a0e63aaa9d", size = 623234, upload-time = "2026-04-27T13:02:40.611Z" },
+    { url = "https://files.pythonhosted.org/packages/8a/17/a3918541fd0ddefe024a69de6d16aa7b46d36ac19562adaa63c7fa180eff/greenlet-3.5.0-cp313-cp313-manylinux_2_24_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:2094acd54b272cb6eae8c03dd87b3fa1820a4cef18d6889c378d503500a1dc13", size = 613927, upload-time = "2026-04-27T12:25:30.28Z" },
+    { url = "https://files.pythonhosted.org/packages/77/18/3b13d5ef1275b0ffaf933b05efa21408ac4ca95823c7411d79682e4fdcff/greenlet-3.5.0-cp313-cp313-manylinux_2_39_riscv64.whl", hash = "sha256:7022615368890680e67b9965d33f5773aade330d5343bbe25560135aaa849eae", size = 425243, upload-time = "2026-04-27T13:05:15.689Z" },
+    { url = "https://files.pythonhosted.org/packages/ee/e1/bd0af6213c7dd33175d8a462d4c1fe1175124ebed4855bc1475a5b5242c2/greenlet-3.5.0-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:5e05ba267789ea87b5a155cf0e810b1ab88bf18e9e8740813945ceb8ee4350ba", size = 1570893, upload-time = "2026-04-27T12:53:29.483Z" },
+    { url = "https://files.pythonhosted.org/packages/9b/2a/0789702f864f5382cb476b93d7a9c823c10472658102ccd65f415747d2e2/greenlet-3.5.0-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:0ecec963079cd58cbd14723582384f11f166fd58883c15dcbfb342e0bc9b5846", size = 1636060, upload-time = "2026-04-27T12:25:28.845Z" },
+    { url = "https://files.pythonhosted.org/packages/b2/8f/22bf9df92bbff0eb07842b60f7e63bf7675a9742df628437a9f02d09137f/greenlet-3.5.0-cp313-cp313-win_amd64.whl", hash = "sha256:728d9667d8f2f586644b748dbd9bb67e50d6a9381767d1357714ea6825bb3bf5", size = 238740, upload-time = "2026-04-27T12:24:01.341Z" },
+    { url = "https://files.pythonhosted.org/packages/b6/b7/9c5c3d653bd4ff614277c049ac676422e2c557db47b4fe43e6313fc005dc/greenlet-3.5.0-cp313-cp313-win_arm64.whl", hash = "sha256:47422135b1d308c14b2c6e758beedb1acd33bb91679f5670edf77bf46244722b", size = 235525, upload-time = "2026-04-27T12:23:12.308Z" },
+]
+
 [[package]]
 name = "h11"
 version = "0.16.0"
@@ -1213,7 +1245,7 @@ dependencies = [
     { name = "hf-transfer", marker = "sys_platform == 'darwin' or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
     { name = "huggingface-hub", marker = "sys_platform == 'darwin' or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
     { name = "matplotlib", marker = "sys_platform == 'darwin' or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
-    { name = "mlx", version = "0.32.0.dev20260427+cc3f3e60", source = { git = "https://github.com/rltakashige/mlx-jaccl-fix-small-recv.git?branch=address-rdma-gpu-locks#cc3f3e60be1289506125f2fa19b73b05aa770df8" }, marker = "sys_platform == 'darwin' or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
+    { name = "mlx", version = "0.32.0.dev20260511+e9835615", source = { git = "https://github.com/rltakashige/mlx-jaccl-fix-small-recv.git?branch=address-rdma-gpu-locks#e9835615badaf4ad7c95301f2a9bd58d782b38db" }, marker = "sys_platform == 'darwin' or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
     { name = "numpy", marker = "sys_platform == 'darwin' or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
     { name = "opencv-python", marker = "sys_platform == 'darwin' or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
     { name = "piexif", marker = "sys_platform == 'darwin' or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
@@ -1263,8 +1295,8 @@ wheels = [
 
 [[package]]
 name = "mlx"
-version = "0.32.0.dev20260427+cc3f3e60"
-source = { git = "https://github.com/rltakashige/mlx-jaccl-fix-small-recv.git?branch=address-rdma-gpu-locks#cc3f3e60be1289506125f2fa19b73b05aa770df8" }
+version = "0.32.0.dev20260511+e9835615"
+source = { git = "https://github.com/rltakashige/mlx-jaccl-fix-small-recv.git?branch=address-rdma-gpu-locks#e9835615badaf4ad7c95301f2a9bd58d782b38db" }
 resolution-markers = [
     "sys_platform == 'darwin'",
 ]
@@ -1315,7 +1347,7 @@ source = { git = "https://github.com/rltakashige/mlx-lm?branch=leo%2Fdeepseek-v4
 dependencies = [
     { name = "jinja2", marker = "sys_platform == 'darwin' or (sys_platform == 'linux' and extra == 'extra-3-exo-cpu') or (sys_platform == 'linux' and extra == 'extra-3-exo-cuda12') or (sys_platform == 'linux' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
     { name = "mlx", version = "0.31.1", source = { registry = "https://pypi.org/simple" }, marker = "(sys_platform == 'linux' and extra == 'extra-3-exo-cpu') or (sys_platform == 'linux' and extra == 'extra-3-exo-cuda12') or (sys_platform == 'linux' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
-    { name = "mlx", version = "0.32.0.dev20260427+cc3f3e60", source = { git = "https://github.com/rltakashige/mlx-jaccl-fix-small-recv.git?branch=address-rdma-gpu-locks#cc3f3e60be1289506125f2fa19b73b05aa770df8" }, marker = "sys_platform == 'darwin' or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
+    { name = "mlx", version = "0.32.0.dev20260511+e9835615", source = { git = "https://github.com/rltakashige/mlx-jaccl-fix-small-recv.git?branch=address-rdma-gpu-locks#e9835615badaf4ad7c95301f2a9bd58d782b38db" }, marker = "sys_platform == 'darwin' or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
     { name = "numpy", marker = "sys_platform == 'darwin' or (sys_platform == 'linux' and extra == 'extra-3-exo-cpu') or (sys_platform == 'linux' and extra == 'extra-3-exo-cuda12') or (sys_platform == 'linux' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
     { name = "protobuf", marker = "sys_platform == 'darwin' or (sys_platform == 'linux' and extra == 'extra-3-exo-cpu') or (sys_platform == 'linux' and extra == 'extra-3-exo-cuda12') or (sys_platform == 'linux' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
     { name = "pyyaml", marker = "sys_platform == 'darwin' or (sys_platform == 'linux' and extra == 'extra-3-exo-cpu') or (sys_platform == 'linux' and extra == 'extra-3-exo-cuda12') or (sys_platform == 'linux' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
@@ -1332,7 +1364,7 @@ dependencies = [
     { name = "fastapi", marker = "sys_platform == 'darwin' or (sys_platform == 'linux' and extra == 'extra-3-exo-cpu') or (sys_platform == 'linux' and extra == 'extra-3-exo-cuda12') or (sys_platform == 'linux' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
     { name = "miniaudio", marker = "sys_platform == 'darwin' or (sys_platform == 'linux' and extra == 'extra-3-exo-cpu') or (sys_platform == 'linux' and extra == 'extra-3-exo-cuda12') or (sys_platform == 'linux' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
     { name = "mlx", version = "0.31.1", source = { registry = "https://pypi.org/simple" }, marker = "(sys_platform == 'linux' and extra == 'extra-3-exo-cpu') or (sys_platform == 'linux' and extra == 'extra-3-exo-cuda12') or (sys_platform == 'linux' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
-    { name = "mlx", version = "0.32.0.dev20260427+cc3f3e60", source = { git = "https://github.com/rltakashige/mlx-jaccl-fix-small-recv.git?branch=address-rdma-gpu-locks#cc3f3e60be1289506125f2fa19b73b05aa770df8" }, marker = "sys_platform == 'darwin' or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
+    { name = "mlx", version = "0.32.0.dev20260511+e9835615", source = { git = "https://github.com/rltakashige/mlx-jaccl-fix-small-recv.git?branch=address-rdma-gpu-locks#e9835615badaf4ad7c95301f2a9bd58d782b38db" }, marker = "sys_platform == 'darwin' or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
     { name = "mlx-lm", marker = "sys_platform == 'darwin' or (sys_platform == 'linux' and extra == 'extra-3-exo-cpu') or (sys_platform == 'linux' and extra == 'extra-3-exo-cuda12') or (sys_platform == 'linux' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
     { name = "numpy", marker = "sys_platform == 'darwin' or (sys_platform == 'linux' and extra == 'extra-3-exo-cpu') or (sys_platform == 'linux' and extra == 'extra-3-exo-cuda12') or (sys_platform == 'linux' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
     { name = "opencv-python", marker = "sys_platform == 'darwin' or (sys_platform == 'linux' and extra == 'extra-3-exo-cpu') or (sys_platform == 'linux' and extra == 'extra-3-exo-cuda12') or (sys_platform == 'linux' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
@@ -1768,6 +1800,25 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/cb/28/3bfe2fa5a7b9c46fe7e13c97bda14c895fb10fa2ebf1d0abb90e0cea7ee1/platformdirs-4.5.1-py3-none-any.whl", hash = "sha256:d03afa3963c806a9bed9d5125c8f4cb2fdaf74a55ab60e5d59b3fde758104d31", size = 18731, upload-time = "2025-12-05T13:52:56.823Z" },
 ]
 
+[[package]]
+name = "playwright"
+version = "1.58.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "greenlet", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
+    { name = "pyee", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
+]
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/f8/c9/9c6061d5703267f1baae6a4647bfd1862e386fbfdb97d889f6f6ae9e3f64/playwright-1.58.0-py3-none-macosx_10_13_x86_64.whl", hash = "sha256:96e3204aac292ee639edbfdef6298b4be2ea0a55a16b7068df91adac077cc606", size = 42251098, upload-time = "2026-01-30T15:09:24.028Z" },
+    { url = "https://files.pythonhosted.org/packages/e0/40/59d34a756e02f8c670f0fee987d46f7ee53d05447d43cd114ca015cb168c/playwright-1.58.0-py3-none-macosx_11_0_arm64.whl", hash = "sha256:70c763694739d28df71ed578b9c8202bb83e8fe8fb9268c04dd13afe36301f71", size = 41039625, upload-time = "2026-01-30T15:09:27.558Z" },
+    { url = "https://files.pythonhosted.org/packages/e1/ee/3ce6209c9c74a650aac9028c621f357a34ea5cd4d950700f8e2c4b7fe2c4/playwright-1.58.0-py3-none-macosx_11_0_universal2.whl", hash = "sha256:185e0132578733d02802dfddfbbc35f42be23a45ff49ccae5081f25952238117", size = 42251098, upload-time = "2026-01-30T15:09:30.461Z" },
+    { url = "https://files.pythonhosted.org/packages/f1/af/009958cbf23fac551a940d34e3206e6c7eed2b8c940d0c3afd1feb0b0589/playwright-1.58.0-py3-none-manylinux1_x86_64.whl", hash = "sha256:c95568ba1eda83812598c1dc9be60b4406dffd60b149bc1536180ad108723d6b", size = 46235268, upload-time = "2026-01-30T15:09:33.787Z" },
+    { url = "https://files.pythonhosted.org/packages/d9/a6/0e66ad04b6d3440dae73efb39540c5685c5fc95b17c8b29340b62abbd952/playwright-1.58.0-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:8f9999948f1ab541d98812de25e3a8c410776aa516d948807140aff797b4bffa", size = 45964214, upload-time = "2026-01-30T15:09:36.751Z" },
+    { url = "https://files.pythonhosted.org/packages/0e/4b/236e60ab9f6d62ed0fd32150d61f1f494cefbf02304c0061e78ed80c1c32/playwright-1.58.0-py3-none-win32.whl", hash = "sha256:1e03be090e75a0fabbdaeab65ce17c308c425d879fa48bb1d7986f96bfad0b99", size = 36815998, upload-time = "2026-01-30T15:09:39.627Z" },
+    { url = "https://files.pythonhosted.org/packages/41/f8/5ec599c5e59d2f2f336a05b4f318e733077cd5044f24adb6f86900c3e6a7/playwright-1.58.0-py3-none-win_amd64.whl", hash = "sha256:a2bf639d0ce33b3ba38de777e08697b0d8f3dc07ab6802e4ac53fb65e3907af8", size = 36816005, upload-time = "2026-01-30T15:09:42.449Z" },
+    { url = "https://files.pythonhosted.org/packages/c8/c4/cc0229fea55c87d6c9c67fe44a21e2cd28d1d558a5478ed4d617e9fb0c93/playwright-1.58.0-py3-none-win_arm64.whl", hash = "sha256:32ffe5c303901a13a0ecab91d1c3f74baf73b84f4bedbb6b935f5bc11cc98e1b", size = 33085919, upload-time = "2026-01-30T15:09:45.71Z" },
+]
+
 [[package]]
 name = "pluggy"
 version = "1.6.0"
@@ -1941,6 +1992,18 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/73/7d/f2f9db34af103bea3e09735bb40b021788a5e834c81eedb541991badf8f5/pydantic_core-2.41.5-cp313-cp313-win_arm64.whl", hash = "sha256:3f84d5c1b4ab906093bdc1ff10484838aca54ef08de4afa9de0f5f14d69639cd", size = 1981005, upload-time = "2025-11-04T13:40:54.734Z" },
 ]
 
+[[package]]
+name = "pyee"
+version = "13.0.1"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "typing-extensions", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda12') or (extra == 'extra-3-exo-cpu' and extra == 'extra-3-exo-cuda13') or (extra == 'extra-3-exo-cuda12' and extra == 'extra-3-exo-cuda13')" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/8b/04/e7c1fe4dc78a6fdbfd6c337b1c3732ff543b8a397683ab38378447baa331/pyee-13.0.1.tar.gz", hash = "sha256:0b931f7c14535667ed4c7e0d531716368715e860b988770fc7eb8578d1f67fc8", size = 31655, upload-time = "2026-02-14T21:12:28.044Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/a0/c4/b4d4827c93ef43c01f599ef31453ccc1c132b353284fc6c87d535c233129/pyee-13.0.1-py3-none-any.whl", hash = "sha256:af2f8fede4171ef667dfded53f96e2ed0d6e6bd7ee3bb46437f77e3b57689228", size = 15659, upload-time = "2026-02-14T21:12:26.263Z" },
+]
+
 [[package]]
 name = "pygments"
 version = "2.19.2"