feat: add E2E chaos/networking tests

Add 17 E2E chaos tests across 6 test modules exercising the coordination
layer without Docker, networking, or GPU dependencies:

- Networking resilience: disconnect/reconnect, node timeout, concurrent writers
- Failure recovery: master crash/re-election, runner failure, rapid node joins
- Client disconnect: task cancellation, rapid cancel/no stuck tasks
- Node join/leave: dynamic registration, removal cleanup, join/leave churn
- Distributed model loading: multi-node sharding, single-node, 3-node sharding
- Concurrent requests: no corruption, multi-model routing, monotonic indexing

Uses MiniCluster harness wiring Master + Workers via in-process channels.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

.dockerignore

@@ -0,0 +1,15 @@
+.venv/
+.direnv/
+target/
+.git/
+.idea/
+.pytest_cache/
+.ruff_cache/
+dashboard/node_modules/
+dashboard/.svelte-kit/
+dashboard/build/
+dist/
+*.pdb
+**/__pycache__
+**/.DS_Store
+.mlx_typings/

.github/workflows/e2e.yml

@@ -0,0 +1,44 @@
+name: e2e-tests
+
+on:
+  push:
+    branches:
+      - e2e-tests
+  pull_request:
+    branches:
+      - staging
+      - main
+
+jobs:
+  e2e:
+    runs-on: ubuntu-latest
+    timeout-minutes: 45
+    steps:
+      - name: Free up disk space
+        run: |
+          sudo rm -rf /usr/share/dotnet /usr/local/lib/android /opt/ghc \
+            /opt/hostedtoolcache /usr/local/share/boost /usr/share/swift \
+            /opt/microsoft /opt/az
+          docker system prune -af
+          df -h /
+
+      - name: Checkout repository
+        uses: actions/checkout@v4
+        with:
+          lfs: false
+
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v3
+
+      - name: Build E2E image with cache
+        uses: docker/build-push-action@v6
+        with:
+          context: .
+          file: e2e/Dockerfile
+          tags: exo-e2e:latest
+          load: true
+          cache-from: type=gha
+          cache-to: type=gha,mode=max
+
+      - name: Run E2E tests
+        run: python3 e2e/run_all.py

conftest.py

@@ -0,0 +1 @@
+collect_ignore = ["tests/start_distributed_test.py"]

e2e/Dockerfile

@@ -0,0 +1,58 @@
+# Stage 1: Build the dashboard
+FROM node:22-slim AS dashboard
+WORKDIR /app/dashboard
+COPY dashboard/package.json dashboard/package-lock.json ./
+RUN npm ci
+COPY dashboard/ .
+RUN npm run build
+
+# Stage 2: Build and run exo
+FROM python:3.13-slim
+
+# Install system dependencies
+# libblas-dev/liblapack-dev/liblapacke-dev are required by MLX CPU backend on Linux
+RUN apt-get update && apt-get install -y \
+    build-essential \
+    pkg-config \
+    libssl-dev \
+    libblas-dev \
+    liblapack-dev \
+    liblapacke-dev \
+    curl \
+    protobuf-compiler \
+    iptables \
+    && rm -rf /var/lib/apt/lists/*
+
+# Install Rust nightly
+RUN curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y --default-toolchain nightly
+ENV PATH="/root/.cargo/bin:${PATH}"
+
+# Wrap g++ with -fpermissive to fix MLX CPU JIT compilation with GCC 14
+# (GCC 14 treats _Float128/_Float32/_Float64 as built-in types, conflicting with MLX-generated code)
+# Must be done BEFORE uv sync so any source builds also get the fix
+RUN mv /usr/bin/g++ /usr/bin/g++.real && \
+    printf '#!/bin/sh\nexec /usr/bin/g++.real -fpermissive "$@"\n' > /usr/bin/g++ && \
+    chmod +x /usr/bin/g++
+
+# Install uv
+COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv
+
+WORKDIR /app
+
+# Copy dependency files first for better layer caching
+COPY pyproject.toml Cargo.toml uv.lock README.md ./
+COPY rust/ ./rust/
+COPY bench/pyproject.toml ./bench/pyproject.toml
+
+# Copy source and resources
+COPY src/ ./src/
+COPY resources/ ./resources/
+
+# Copy built dashboard from stage 1
+COPY --from=dashboard /app/dashboard/build ./dashboard/build/
+
+# Install Python deps and build Rust bindings, then clean up build artifacts
+# to keep the layer small (Rust target/ and cargo registry can be 1-2 GB)
+RUN uv sync && rm -rf /app/rust/target /root/.cargo/registry /root/.cargo/git
+
+CMD [".venv/bin/exo", "-v"]

e2e/conftest.py

@@ -0,0 +1,195 @@
+"""Shared E2E test infrastructure for exo cluster tests."""
+
+import asyncio
+import json
+import os
+import sys
+from pathlib import Path
+from urllib.error import URLError
+from urllib.request import Request, urlopen
+
+E2E_DIR = Path(__file__).parent.resolve()
+TIMEOUT = int(os.environ.get("E2E_TIMEOUT", "120"))
+
+
+class Cluster:
+    """Async wrapper around a docker compose exo cluster."""
+
+    def __init__(self, name: str, overrides: list[str] | None = None):
+        self.name = name
+        self.project = f"e2e-{name}"
+        compose_files = [str(E2E_DIR / "docker-compose.yml")]
+        for path in overrides or []:
+            compose_files.append(str(E2E_DIR / path))
+        self._compose_base = [
+            "docker",
+            "compose",
+            "-p",
+            self.project,
+            *[arg for f in compose_files for arg in ("-f", f)],
+        ]
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *exc):
+        await self.stop()
+
+    async def _run(self, *args: str, check: bool = True) -> str:
+        proc = await asyncio.create_subprocess_exec(
+            *self._compose_base,
+            *args,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.STDOUT,
+        )
+        stdout, _ = await proc.communicate()
+        output = stdout.decode()
+        if check and proc.returncode != 0:
+            print(output, file=sys.stderr)
+            raise RuntimeError(
+                f"docker compose {' '.join(args)} failed (rc={proc.returncode})"
+            )
+        return output
+
+    async def build(self):
+        # Skip build if the image was pre-built (e.g. in CI with buildx cache)
+        proc = await asyncio.create_subprocess_exec(
+            "docker",
+            "image",
+            "inspect",
+            "exo-e2e:latest",
+            stdout=asyncio.subprocess.DEVNULL,
+            stderr=asyncio.subprocess.DEVNULL,
+        )
+        await proc.wait()
+        if proc.returncode == 0:
+            print("  Using pre-built image (exo-e2e:latest)")
+            return
+        print("  Building images...")
+        await self._run("build", "--quiet")
+
+    async def start(self):
+        print("  Starting cluster...")
+        await self._run("up", "-d")
+
+    async def stop(self):
+        print("  Cleaning up...")
+        await self._run("down", "--timeout", "5", check=False)
+
+    async def logs(self) -> str:
+        return await self._run("logs", check=False)
+
+    async def exec(
+        self, service: str, *cmd: str, check: bool = True
+    ) -> tuple[int, str]:
+        """Run a command inside a running container. Returns (returncode, output)."""
+        proc = await asyncio.create_subprocess_exec(
+            *self._compose_base,
+            "exec",
+            "-T",
+            service,
+            *cmd,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.STDOUT,
+        )
+        stdout, _ = await proc.communicate()
+        output = stdout.decode()
+        if check and proc.returncode != 0:
+            raise RuntimeError(
+                f"exec {' '.join(cmd)} in {service} failed (rc={proc.returncode})"
+            )
+        return proc.returncode, output
+
+    async def wait_for(self, description: str, check_fn, timeout: int = TIMEOUT):
+        """Poll check_fn every 2s until it returns True or timeout expires."""
+        print(f"  Waiting for {description}...")
+        deadline = asyncio.get_event_loop().time() + timeout
+        while asyncio.get_event_loop().time() < deadline:
+            if await check_fn():
+                print(f"  {description}")
+                return
+            await asyncio.sleep(2)
+        output = await self.logs()
+        print(f"--- cluster logs ---\n{output}\n---", file=sys.stderr)
+        raise TimeoutError(f"Timed out waiting for {description}")
+
+    async def assert_healthy(self):
+        """Verify the cluster formed correctly: nodes started, discovered each other, elected a master, API responds."""
+
+        async def both_nodes_started():
+            log = await self.logs()
+            return log.count("Starting node") >= 2
+
+        async def nodes_discovered():
+            log = await self.logs()
+            return log.count("ConnectionMessageType.Connected") >= 2
+
+        async def master_elected():
+            log = await self.logs()
+            return "demoting self" in log
+
+        async def api_responding():
+            try:
+                with urlopen("http://localhost:52415/v1/models", timeout=3) as resp:
+                    return resp.status == 200
+            except (URLError, OSError):
+                return False
+
+        await self.wait_for("Both nodes started", both_nodes_started)
+        await self.wait_for("Nodes discovered each other", nodes_discovered)
+        await self.wait_for("Master election resolved", master_elected)
+        await self.wait_for("API responding", api_responding)
+
+    async def _api(
+        self, method: str, path: str, body: dict | None = None, timeout: int = 30
+    ) -> dict:
+        """Make an API request to the cluster. Returns parsed JSON."""
+        url = f"http://localhost:52415{path}"
+        data = json.dumps(body).encode() if body else None
+        req = Request(
+            url, data=data, headers={"Content-Type": "application/json"}, method=method
+        )
+        loop = asyncio.get_event_loop()
+        resp_bytes = await loop.run_in_executor(
+            None, lambda: urlopen(req, timeout=timeout).read()
+        )
+        return json.loads(resp_bytes)
+
+    async def place_model(self, model: str, timeout: int = 600):
+        """Place a model instance on the cluster (triggers download) and wait until it's ready."""
+        await self._api("POST", "/place_instance", {"model_id": model})
+
+        async def model_ready():
+            try:
+                resp = await self._api("GET", "/v1/models")
+                return any(m.get("id") == model for m in resp.get("data", []))
+            except Exception:
+                return False
+
+        await self.wait_for(f"Model {model} ready", model_ready, timeout=timeout)
+
+    async def chat(
+        self, model: str, messages: list[dict], timeout: int = 600, **kwargs
+    ) -> dict:
+        """Send a chat completion request. Retries until model is downloaded and inference completes."""
+        body = json.dumps({"model": model, "messages": messages, **kwargs}).encode()
+        deadline = asyncio.get_event_loop().time() + timeout
+        last_error = None
+
+        while asyncio.get_event_loop().time() < deadline:
+            try:
+                req = Request(
+                    "http://localhost:52415/v1/chat/completions",
+                    data=body,
+                    headers={"Content-Type": "application/json"},
+                )
+                loop = asyncio.get_event_loop()
+                resp_bytes = await loop.run_in_executor(
+                    None, lambda r=req: urlopen(r, timeout=300).read()
+                )
+                return json.loads(resp_bytes)
+            except Exception as e:
+                last_error = e
+                await asyncio.sleep(5)
+
+        raise TimeoutError(f"Chat request failed after {timeout}s: {last_error}")

e2e/docker-compose.yml

@@ -0,0 +1,20 @@
+services:
+  exo-node-1:
+    image: exo-e2e:latest
+    build:
+      context: ..
+      dockerfile: e2e/Dockerfile
+    environment:
+      - EXO_LIBP2P_NAMESPACE=docker-e2e
+    command: [".venv/bin/exo", "-v"]
+    ports:
+      - "52415:52415"
+
+  exo-node-2:
+    image: exo-e2e:latest
+    build:
+      context: ..
+      dockerfile: e2e/Dockerfile
+    environment:
+      - EXO_LIBP2P_NAMESPACE=docker-e2e
+    command: [".venv/bin/exo", "-v"]

e2e/run_all.py

@@ -0,0 +1,83 @@
+#!/usr/bin/env python3
+"""Discovers and runs all E2E tests in e2e/test_*.py.
+
+Tests with '# slow' on the first line of their docstring are skipped
+unless --slow is passed or E2E_SLOW=1 is set.
+"""
+
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+E2E_DIR = Path(__file__).parent.resolve()
+
+
+def is_slow(test_file: Path) -> bool:
+    """Check if the test file is marked as slow (has '# slow' in first 3 lines)."""
+    with open(test_file) as f:
+        for line in f:
+            if line.strip().startswith("#"):
+                continue
+            if line.strip().startswith('"""') or line.strip().startswith("'''"):
+                # Read into the docstring
+                for doc_line in f:
+                    if "slow" in doc_line.lower() and doc_line.strip().startswith(
+                        "slow"
+                    ):
+                        return True
+                    if '"""' in doc_line or "'''" in doc_line:
+                        break
+            break
+    return False
+
+
+def main():
+    run_slow = "--slow" in sys.argv or os.environ.get("E2E_SLOW") == "1"
+    if "--update-snapshots" in sys.argv:
+        os.environ["UPDATE_SNAPSHOTS"] = "1"
+    test_files = sorted(E2E_DIR.glob("test_*.py"))
+    if not test_files:
+        print("No test files found")
+        sys.exit(1)
+
+    passed = 0
+    failed = 0
+    skipped = 0
+    failures = []
+
+    for test_file in test_files:
+        name = test_file.stem
+        if is_slow(test_file) and not run_slow:
+            print(f"=== {name} === SKIPPED (slow, use --slow to run)")
+            skipped += 1
+            continue
+
+        print(f"=== {name} ===")
+        result = subprocess.run([sys.executable, str(test_file)])
+        if result.returncode == 0:
+            passed += 1
+        else:
+            # Retry once — Docker networking (mDNS) can be slow on first boot
+            print(f"\n=== {name} === RETRYING (attempt 2/2)")
+            result = subprocess.run([sys.executable, str(test_file)])
+            if result.returncode == 0:
+                passed += 1
+            else:
+                failed += 1
+                failures.append(name)
+        print()
+
+    total = passed + failed + skipped
+    print("================================")
+    print(
+        f"{passed}/{total} tests passed" + (f", {skipped} skipped" if skipped else "")
+    )
+
+    if failed:
+        print(f"Failed: {' '.join(failures)}")
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()

e2e/snapshot.py

@@ -0,0 +1,78 @@
+"""Snapshot testing infrastructure for E2E tests.
+
+Provides deterministic regression testing by comparing inference output
+against committed baseline snapshots. Tests FAIL if no baseline exists —
+baselines must be explicitly generated and committed.
+
+Generate baselines: UPDATE_SNAPSHOTS=1 python3 e2e/run_all.py --slow
+Update after intentional changes: UPDATE_SNAPSHOTS=1 python3 e2e/run_all.py --slow
+
+Snapshots are stored per-architecture (e.g. snapshots/x86_64/, snapshots/arm64/)
+since floating-point results differ between CPU architectures.
+"""
+
+import difflib
+import json
+import os
+import platform
+from pathlib import Path
+
+ARCH = platform.machine()
+SNAPSHOTS_DIR = Path(__file__).parent / "snapshots" / ARCH
+
+
+def assert_snapshot(
+    name: str,
+    content: str,
+    metadata: dict,
+) -> None:
+    """Compare content against a saved snapshot, or create one if missing.
+
+    Args:
+        name: Snapshot identifier (used as filename: snapshots/{arch}/{name}.json).
+        content: The actual inference output to compare.
+        metadata: Additional context stored alongside content (model, seed, etc.).
+                  Not used for comparison -- purely documentary.
+
+    Raises:
+        AssertionError: If content doesn't match the saved snapshot.
+
+    Environment:
+        UPDATE_SNAPSHOTS=1: Overwrite existing snapshot with actual content.
+    """
+    snapshot_file = SNAPSHOTS_DIR / f"{name}.json"
+    update = os.environ.get("UPDATE_SNAPSHOTS") == "1"
+
+    if update:
+        # Explicitly regenerate snapshot
+        SNAPSHOTS_DIR.mkdir(parents=True, exist_ok=True)
+        snapshot_data = {**metadata, "arch": ARCH, "content": content}
+        snapshot_file.write_text(json.dumps(snapshot_data, indent=2) + "\n")
+        print(f"  Updated snapshot: {ARCH}/{snapshot_file.name}")
+    elif not snapshot_file.exists():
+        raise AssertionError(
+            f"No baseline snapshot for '{name}' on {ARCH}.\n"
+            f"Expected file: {snapshot_file}\n\n"
+            f"Generate baselines with: UPDATE_SNAPSHOTS=1 python3 e2e/run_all.py --slow"
+        )
+    else:
+        snapshot = json.loads(snapshot_file.read_text())
+        expected = snapshot["content"]
+        if content != expected:
+            diff = "\n".join(
+                difflib.unified_diff(
+                    expected.splitlines(),
+                    content.splitlines(),
+                    fromfile=f"expected ({snapshot_file.relative_to(SNAPSHOTS_DIR.parent.parent)})",
+                    tofile="actual",
+                    lineterm="",
+                )
+            )
+            raise AssertionError(
+                f"Snapshot mismatch for '{name}' on {ARCH}!\n\n"
+                f"{diff}\n\n"
+                f"Expected: {expected!r}\n"
+                f"Actual:   {content!r}\n\n"
+                f"To update: UPDATE_SNAPSHOTS=1 python3 e2e/run_all.py --slow"
+            )
+        print(f"  Output matches snapshot ({ARCH}/{snapshot_file.name})")

e2e/test_cluster_formation.py

@@ -0,0 +1,22 @@
+"""Test: Basic cluster formation.
+
+Verifies two nodes discover each other, elect a master, and the API responds.
+"""
+
+import asyncio
+import sys
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).parent))
+from conftest import Cluster
+
+
+async def main():
+    async with Cluster("cluster_formation") as cluster:
+        await cluster.build()
+        await cluster.start()
+        await cluster.assert_healthy()
+        print("PASSED: cluster_formation")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

e2e/test_inference_snapshot.py

@@ -0,0 +1,61 @@
+"""Test: Deterministic inference output (snapshot test).
+
+Sends a chat completion request with a fixed seed,
+then verifies the output matches a known-good snapshot. This ensures
+inference produces consistent results across runs.
+
+Uses MLX CPU backend in Docker on x86 Linux.
+"""
+
+import asyncio
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent))
+from snapshot import assert_snapshot
+
+from conftest import Cluster
+
+MODEL = "mlx-community/Qwen3-0.6B-4bit"
+SEED = 42
+PROMPT = "What is 2+2? Reply with just the number."
+MAX_TOKENS = 32
+
+
+async def main():
+    async with Cluster("inference_snapshot") as cluster:
+        await cluster.build()
+        await cluster.start()
+        await cluster.assert_healthy()
+
+        print(f"  Launching model {MODEL}...")
+        await cluster.place_model(MODEL)
+
+        print(f"  Sending chat completion (seed={SEED})...")
+        resp = await cluster.chat(
+            model=MODEL,
+            messages=[{"role": "user", "content": PROMPT}],
+            seed=SEED,
+            temperature=0,
+            max_tokens=MAX_TOKENS,
+        )
+
+        content = resp["choices"][0]["message"]["content"]
+        print(f"  Response: {content!r}")
+
+        assert_snapshot(
+            name="inference_snapshot",
+            content=content,
+            metadata={
+                "model": MODEL,
+                "seed": SEED,
+                "prompt": PROMPT,
+                "max_tokens": MAX_TOKENS,
+            },
+        )
+
+        print("PASSED: inference_snapshot")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

e2e/test_no_internet.py

@@ -0,0 +1,47 @@
+"""Test: Cluster works without internet access.
+
+Verifies exo functions correctly when containers can talk to each other
+but cannot reach the internet. Uses iptables to block all outbound traffic
+except private subnets and multicast (for mDNS discovery).
+"""
+
+import asyncio
+import sys
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).parent))
+from conftest import Cluster
+
+
+async def main():
+    async with Cluster(
+        "no_internet",
+        overrides=["tests/no_internet/docker-compose.override.yml"],
+    ) as cluster:
+        await cluster.build()
+        await cluster.start()
+        await cluster.assert_healthy()
+
+        # Verify internet is actually blocked from inside the containers
+        for node in ["exo-node-1", "exo-node-2"]:
+            rc, _ = await cluster.exec(
+                node,
+                "curl",
+                "-sf",
+                "--max-time",
+                "3",
+                "https://huggingface.co",
+                check=False,
+            )
+            assert rc != 0, f"{node} should not be able to reach the internet"
+            print(f"  {node}: internet correctly blocked")
+
+        # Verify exo detected no internet connectivity
+        log = await cluster.logs()
+        assert "Internet connectivity: False" in log, "exo should detect no internet"
+        print("  exo correctly detected no internet connectivity")
+
+        print("PASSED: no_internet")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

e2e/test_runner_chaos.py

@@ -0,0 +1,65 @@
+"""Test: Runner chaos — abrupt runner death detection.
+slow
+
+Sends a chat completion with the EXO_RUNNER_MUST_DIE trigger, which causes
+the runner process to call os._exit(1) (simulating an OOM kill). Verifies that
+the RunnerSupervisor health check detects the death and the system doesn't hang.
+
+Requires a machine that can run MLX inference at reasonable speed (Apple Silicon).
+Run with: python3 e2e/run_all.py --slow  or  E2E_SLOW=1 python3 e2e/run_all.py
+"""
+
+import asyncio
+import contextlib
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent))
+from conftest import Cluster
+
+MODEL = "mlx-community/Qwen3-0.6B-4bit"
+
+
+async def main():
+    async with Cluster("runner_chaos") as cluster:
+        await cluster.build()
+        await cluster.start()
+        await cluster.assert_healthy()
+
+        # Place the model so a runner is loaded and ready
+        print(f"  Launching model {MODEL}...")
+        await cluster.place_model(MODEL)
+
+        # Send a chat request with the die trigger.
+        # The runner will call os._exit(1) mid-inference, simulating OOM kill.
+        # The chat request itself will fail — that's expected.
+        print("  Sending EXO_RUNNER_MUST_DIE trigger...")
+        with contextlib.suppress(Exception):
+            await cluster.chat(
+                model=MODEL,
+                messages=[{"role": "user", "content": "EXO RUNNER MUST DIE"}],
+                timeout=60,
+            )
+
+        # Wait for the health check to detect the death and emit RunnerFailed
+        async def health_check_detected():
+            log = await cluster.logs()
+            return "runner process died unexpectedly" in log
+
+        await cluster.wait_for(
+            "Health check detected runner death",
+            health_check_detected,
+            timeout=30,
+        )
+
+        # Verify RunnerFailed was emitted (visible in logs)
+        log = await cluster.logs()
+        assert "runner process died unexpectedly" in log, (
+            f"Expected health check to detect runner death but it didn't.\nLogs:\n{log}"
+        )
+
+        print("PASSED: runner_chaos")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

e2e/test_snapshot_code_gen.py

@@ -0,0 +1,60 @@
+"""Test: Code generation snapshot.
+slow
+
+Verifies deterministic output for a code generation prompt.
+"""
+
+import asyncio
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent))
+from snapshot import assert_snapshot
+
+from conftest import Cluster
+
+MODEL = "mlx-community/Qwen3-0.6B-4bit"
+SEED = 42
+PROMPT = (
+    "Write a Python function to reverse a string. Only output the code, no explanation."
+)
+MAX_TOKENS = 64
+
+
+async def main():
+    async with Cluster("snapshot_code_gen") as cluster:
+        await cluster.build()
+        await cluster.start()
+        await cluster.assert_healthy()
+
+        print(f"  Launching model {MODEL}...")
+        await cluster.place_model(MODEL)
+
+        print(f"  Sending chat completion (seed={SEED})...")
+        resp = await cluster.chat(
+            model=MODEL,
+            messages=[{"role": "user", "content": PROMPT}],
+            seed=SEED,
+            temperature=0,
+            max_tokens=MAX_TOKENS,
+        )
+
+        content = resp["choices"][0]["message"]["content"]
+        print(f"  Response: {content!r}")
+
+        assert_snapshot(
+            name="snapshot_code_gen",
+            content=content,
+            metadata={
+                "model": MODEL,
+                "seed": SEED,
+                "prompt": PROMPT,
+                "max_tokens": MAX_TOKENS,
+            },
+        )
+
+        print("PASSED: snapshot_code_gen")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

e2e/test_snapshot_edge.py

@@ -0,0 +1,65 @@
+"""Test: Edge case snapshots.
+slow
+
+Verifies deterministic output for edge-case prompts: single word input,
+special characters, and unicode.
+"""
+
+import asyncio
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent))
+from snapshot import assert_snapshot
+
+from conftest import Cluster
+
+MODEL = "mlx-community/Qwen3-0.6B-4bit"
+SEED = 42
+MAX_TOKENS = 32
+
+CASES = [
+    ("edge_single_word", "Hi"),
+    ("edge_special_chars", "What does 2 * (3 + 4) / 7 - 1 equal? Use <math> tags."),
+    ("edge_unicode", "Translate 'hello' to Japanese, Chinese, and Korean."),
+]
+
+
+async def main():
+    async with Cluster("snapshot_edge") as cluster:
+        await cluster.build()
+        await cluster.start()
+        await cluster.assert_healthy()
+
+        print(f"  Launching model {MODEL}...")
+        await cluster.place_model(MODEL)
+
+        for snapshot_name, prompt in CASES:
+            print(f"  [{snapshot_name}] Sending: {prompt!r}")
+            resp = await cluster.chat(
+                model=MODEL,
+                messages=[{"role": "user", "content": prompt}],
+                seed=SEED,
+                temperature=0,
+                max_tokens=MAX_TOKENS,
+            )
+
+            content = resp["choices"][0]["message"]["content"]
+            print(f"  [{snapshot_name}] Response: {content!r}")
+
+            assert_snapshot(
+                name=snapshot_name,
+                content=content,
+                metadata={
+                    "model": MODEL,
+                    "seed": SEED,
+                    "prompt": prompt,
+                    "max_tokens": MAX_TOKENS,
+                },
+            )
+
+        print("PASSED: snapshot_edge")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

e2e/test_snapshot_long_output.py

@@ -0,0 +1,58 @@
+"""Test: Longer output snapshot.
+slow
+
+Verifies deterministic output with a higher max_tokens (128).
+"""
+
+import asyncio
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent))
+from snapshot import assert_snapshot
+
+from conftest import Cluster
+
+MODEL = "mlx-community/Qwen3-0.6B-4bit"
+SEED = 42
+PROMPT = "Explain how a binary search algorithm works."
+MAX_TOKENS = 128
+
+
+async def main():
+    async with Cluster("snapshot_long_output") as cluster:
+        await cluster.build()
+        await cluster.start()
+        await cluster.assert_healthy()
+
+        print(f"  Launching model {MODEL}...")
+        await cluster.place_model(MODEL)
+
+        print(f"  Sending chat completion (seed={SEED}, max_tokens={MAX_TOKENS})...")
+        resp = await cluster.chat(
+            model=MODEL,
+            messages=[{"role": "user", "content": PROMPT}],
+            seed=SEED,
+            temperature=0,
+            max_tokens=MAX_TOKENS,
+        )
+
+        content = resp["choices"][0]["message"]["content"]
+        print(f"  Response: {content!r}")
+
+        assert_snapshot(
+            name="snapshot_long_output",
+            content=content,
+            metadata={
+                "model": MODEL,
+                "seed": SEED,
+                "prompt": PROMPT,
+                "max_tokens": MAX_TOKENS,
+            },
+        )
+
+        print("PASSED: snapshot_long_output")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

e2e/test_snapshot_multi_model.py

@@ -0,0 +1,73 @@
+"""Test: Multi-model snapshot tests.
+slow
+
+Verifies deterministic output across different model architectures to catch
+model-specific regressions. Each model uses its own snapshot file.
+Run with: python3 e2e/run_all.py --slow  or  E2E_SLOW=1 python3 e2e/run_all.py
+"""
+
+import asyncio
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent))
+from snapshot import assert_snapshot
+
+from conftest import Cluster
+
+SEED = 42
+PROMPT = "What is the capital of France?"
+MAX_TOKENS = 32
+
+MODELS = [
+    "mlx-community/SmolLM2-135M-Instruct",
+    "mlx-community/Llama-3.2-1B-Instruct-4bit",
+    "mlx-community/gemma-2-2b-it-4bit",
+]
+
+
+async def main():
+    async with Cluster("snapshot_multi_model") as cluster:
+        await cluster.build()
+        await cluster.start()
+        await cluster.assert_healthy()
+
+        for model in MODELS:
+            short_name = (
+                model.split("/")[-1].lower().replace("-", "_").replace(".", "_")
+            )
+            snapshot_name = f"snapshot_multi_{short_name}"
+
+            print(f"  Launching model {model}...")
+            await cluster.place_model(model)
+
+            print(f"  Sending chat completion (seed={SEED})...")
+            resp = await cluster.chat(
+                model=model,
+                messages=[{"role": "user", "content": PROMPT}],
+                seed=SEED,
+                temperature=0,
+                max_tokens=MAX_TOKENS,
+            )
+
+            content = resp["choices"][0]["message"]["content"]
+            print(f"  [{short_name}] Response: {content!r}")
+
+            assert_snapshot(
+                name=snapshot_name,
+                content=content,
+                metadata={
+                    "model": model,
+                    "seed": SEED,
+                    "prompt": PROMPT,
+                    "max_tokens": MAX_TOKENS,
+                },
+            )
+
+            print(f"  [{short_name}] PASSED")
+
+        print("PASSED: snapshot_multi_model")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

e2e/test_snapshot_reasoning.py

@@ -0,0 +1,58 @@
+"""Test: Reasoning/math snapshot.
+slow
+
+Verifies deterministic output for a simple reasoning prompt.
+"""
+
+import asyncio
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent))
+from snapshot import assert_snapshot
+
+from conftest import Cluster
+
+MODEL = "mlx-community/Qwen3-0.6B-4bit"
+SEED = 42
+PROMPT = "If I have 3 apples and give away 1, how many do I have? Think step by step."
+MAX_TOKENS = 64
+
+
+async def main():
+    async with Cluster("snapshot_reasoning") as cluster:
+        await cluster.build()
+        await cluster.start()
+        await cluster.assert_healthy()
+
+        print(f"  Launching model {MODEL}...")
+        await cluster.place_model(MODEL)
+
+        print(f"  Sending chat completion (seed={SEED})...")
+        resp = await cluster.chat(
+            model=MODEL,
+            messages=[{"role": "user", "content": PROMPT}],
+            seed=SEED,
+            temperature=0,
+            max_tokens=MAX_TOKENS,
+        )
+
+        content = resp["choices"][0]["message"]["content"]
+        print(f"  Response: {content!r}")
+
+        assert_snapshot(
+            name="snapshot_reasoning",
+            content=content,
+            metadata={
+                "model": MODEL,
+                "seed": SEED,
+                "prompt": PROMPT,
+                "max_tokens": MAX_TOKENS,
+            },
+        )
+
+        print("PASSED: snapshot_reasoning")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

e2e/tests/no_internet/docker-compose.override.yml

@@ -0,0 +1,32 @@
+# Block all outbound internet traffic using iptables while preserving:
+#   - Multicast (224.0.0.0/4) for mDNS peer discovery
+#   - Private subnets (10/8, 172.16/12, 192.168/16) for inter-container communication
+#   - Loopback (127/8)
+# Requires NET_ADMIN capability for iptables.
+services:
+  exo-node-1:
+    cap_add:
+      - NET_ADMIN
+    entrypoint: ["/bin/sh", "-c"]
+    command:
+      - |
+        iptables -A OUTPUT -d 127.0.0.0/8 -j ACCEPT
+        iptables -A OUTPUT -d 10.0.0.0/8 -j ACCEPT
+        iptables -A OUTPUT -d 172.16.0.0/12 -j ACCEPT
+        iptables -A OUTPUT -d 192.168.0.0/16 -j ACCEPT
+        iptables -A OUTPUT -d 224.0.0.0/4 -j ACCEPT
+        iptables -A OUTPUT -j REJECT
+        exec .venv/bin/exo -v
+  exo-node-2:
+    cap_add:
+      - NET_ADMIN
+    entrypoint: ["/bin/sh", "-c"]
+    command:
+      - |
+        iptables -A OUTPUT -d 127.0.0.0/8 -j ACCEPT
+        iptables -A OUTPUT -d 10.0.0.0/8 -j ACCEPT
+        iptables -A OUTPUT -d 172.16.0.0/12 -j ACCEPT
+        iptables -A OUTPUT -d 192.168.0.0/16 -j ACCEPT
+        iptables -A OUTPUT -d 224.0.0.0/4 -j ACCEPT
+        iptables -A OUTPUT -j REJECT
+        exec .venv/bin/exo -v

resources/inference_model_cards/mlx-community--SmolLM2-135M-Instruct.toml

@@ -0,0 +1,12 @@
+model_id = "mlx-community/SmolLM2-135M-Instruct"
+n_layers = 30
+hidden_size = 576
+supports_tensor = true
+tasks = ["TextGeneration"]
+family = "llama"
+quantization = "bf16"
+base_model = "SmolLM2 135M"
+capabilities = ["text"]
+
+[storage_size]
+in_bytes = 269060381

resources/inference_model_cards/mlx-community--gemma-2-2b-it-4bit.toml

@@ -0,0 +1,12 @@
+model_id = "mlx-community/gemma-2-2b-it-4bit"
+n_layers = 26
+hidden_size = 2304
+supports_tensor = false
+tasks = ["TextGeneration"]
+family = "gemma2"
+quantization = "4bit"
+base_model = "Gemma 2 2B"
+capabilities = ["text"]
+
+[storage_size]
+in_bytes = 1492755242

src/exo/tests/e2e_chaos/conftest.py

@@ -0,0 +1,307 @@
+"""Shared fixtures and helpers for E2E chaos/networking tests.
+
+Provides a ``MiniCluster`` that wires Master + Worker(s) + Election together
+using in-process channels.  No Docker, no network, no GPU -- pure async
+integration testing of the coordination layer.
+"""
+
+from collections.abc import Iterator
+from datetime import datetime, timezone
+from typing import Final
+
+import pytest
+from _pytest.logging import LogCaptureFixture
+from loguru import logger
+
+from exo.master.main import Master
+from exo.shared.models.model_cards import ModelCard, ModelTask
+from exo.shared.types.commands import (
+    CommandId,
+    ForwarderCommand,
+    ForwarderDownloadCommand,
+    PlaceInstance,
+    TextGeneration,
+)
+from exo.shared.types.common import ModelId, NodeId, SessionId
+from exo.shared.types.events import (
+    ForwarderEvent,
+    IndexedEvent,
+    NodeGatheredInfo,
+    TopologyEdgeCreated,
+)
+from exo.shared.types.memory import Memory
+from exo.shared.types.multiaddr import Multiaddr
+from exo.shared.types.profiling import MemoryUsage
+from exo.shared.types.text_generation import InputMessage, TextGenerationTaskParams
+from exo.shared.types.topology import Connection, SocketConnection
+from exo.shared.types.worker.instances import InstanceMeta
+from exo.shared.types.worker.shards import Sharding
+from exo.utils.channels import Receiver, Sender, channel
+from exo.worker.main import Worker
+
+# ---------------------------------------------------------------------------
+# Constants
+# ---------------------------------------------------------------------------
+
+TEST_MODEL_ID: Final[ModelId] = ModelId("test-model/chaos-test-1b")
+TEST_MODEL_CARD: Final[ModelCard] = ModelCard(
+    model_id=TEST_MODEL_ID,
+    n_layers=16,
+    storage_size=Memory.from_bytes(678_948),
+    hidden_size=2048,
+    supports_tensor=True,
+    tasks=[ModelTask.TextGeneration],
+)
+
+FAST_ELECTION_TIMEOUT: Final[float] = 0.1
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def make_node_id(label: str) -> NodeId:
+    return NodeId(f"node-{label}")
+
+
+def make_session_id(master_node: NodeId) -> SessionId:
+    return SessionId(master_node_id=master_node, election_clock=0)
+
+
+def make_memory_info() -> MemoryUsage:
+    return MemoryUsage(
+        ram_total=Memory.from_bytes(16 * 1024 * 1024 * 1024),
+        ram_available=Memory.from_bytes(8 * 1024 * 1024 * 1024),
+        swap_total=Memory.from_bytes(0),
+        swap_available=Memory.from_bytes(0),
+    )
+
+
+def make_gathered_info_event(
+    node_id: NodeId, sender_id: NodeId, session_id: SessionId, origin_idx: int
+) -> ForwarderEvent:
+    return ForwarderEvent(
+        origin_idx=origin_idx,
+        origin=sender_id,
+        session=session_id,
+        event=NodeGatheredInfo(
+            when=str(datetime.now(tz=timezone.utc)),
+            node_id=node_id,
+            info=make_memory_info(),
+        ),
+    )
+
+
+def make_topology_edge_event(
+    source: NodeId,
+    sink: NodeId,
+    sender_id: NodeId,
+    session_id: SessionId,
+    origin_idx: int,
+    ip_suffix: int = 1,
+) -> ForwarderEvent:
+    """Create a ForwarderEvent wrapping a TopologyEdgeCreated event."""
+    return ForwarderEvent(
+        origin_idx=origin_idx,
+        origin=sender_id,
+        session=session_id,
+        event=TopologyEdgeCreated(
+            conn=Connection(
+                source=source,
+                sink=sink,
+                edge=SocketConnection(
+                    sink_multiaddr=Multiaddr(
+                        address=f"/ip4/10.0.0.{ip_suffix}/tcp/52415"
+                    )
+                ),
+            )
+        ),
+    )
+
+
+class EventCollector:
+    """Collects ForwarderEvents from a global event receiver."""
+
+    def __init__(self, receiver: Receiver[ForwarderEvent]) -> None:
+        self._receiver = receiver
+        self.indexed_events: list[IndexedEvent] = []
+
+    def collect(self) -> list[IndexedEvent]:
+        raw = self._receiver.collect()
+        for fe in raw:
+            self.indexed_events.append(
+                IndexedEvent(event=fe.event, idx=len(self.indexed_events))
+            )
+        return self.indexed_events
+
+    async def wait_for_event_count(
+        self, count: int, *, timeout: float = 5.0, poll_interval: float = 0.01
+    ) -> list[IndexedEvent]:
+        import anyio
+
+        with anyio.fail_after(timeout):
+            while len(self.collect()) < count:
+                await anyio.sleep(poll_interval)
+        return self.indexed_events
+
+
+class MiniCluster:
+    """An in-process cluster with one Master and N Workers wired via channels.
+
+    No networking, no real model loading -- exercises the coordination logic
+    (event sourcing, command routing, election) in a deterministic, fast
+    test harness.
+    """
+
+    def __init__(self, node_count: int = 2) -> None:
+        self.node_count = node_count
+        self.master_node_id = make_node_id("master")
+        self.session_id = make_session_id(self.master_node_id)
+
+        # -- shared bus channels --
+        self.global_event_sender: Sender[ForwarderEvent]
+        self.global_event_internal_receiver: Receiver[ForwarderEvent]
+        self.global_event_sender, self.global_event_internal_receiver = channel[
+            ForwarderEvent
+        ]()
+
+        self.command_sender: Sender[ForwarderCommand]
+        self.command_receiver: Receiver[ForwarderCommand]
+        self.command_sender, self.command_receiver = channel[ForwarderCommand]()
+
+        self.local_event_sender: Sender[ForwarderEvent]
+        self.local_event_receiver: Receiver[ForwarderEvent]
+        self.local_event_sender, self.local_event_receiver = channel[ForwarderEvent]()
+
+        self.download_cmd_sender: Sender[ForwarderDownloadCommand]
+        self._download_cmd_receiver: Receiver[ForwarderDownloadCommand]
+        self.download_cmd_sender, self._download_cmd_receiver = channel[
+            ForwarderDownloadCommand
+        ]()
+
+        # -- event collector (taps global events) --
+        self.event_collector = EventCollector(
+            self.global_event_internal_receiver.clone()
+        )
+
+        # -- master --
+        self.master = Master(
+            self.master_node_id,
+            self.session_id,
+            global_event_sender=self.global_event_sender.clone(),
+            local_event_receiver=self.local_event_receiver.clone(),
+            command_receiver=self.command_receiver.clone(),
+            download_command_sender=self.download_cmd_sender.clone(),
+        )
+
+        # -- workers --
+        self.worker_node_ids: list[NodeId] = []
+        self.workers: list[Worker] = []
+        for i in range(node_count):
+            wid = make_node_id(f"worker-{i}")
+            self.worker_node_ids.append(wid)
+
+            counter: Iterator[int] = iter(range(1_000_000))
+            worker = Worker(
+                wid,
+                self.session_id,
+                global_event_receiver=self.global_event_internal_receiver.clone(),
+                local_event_sender=self.local_event_sender.clone(),
+                command_sender=self.command_sender.clone(),
+                download_command_sender=self.download_cmd_sender.clone(),
+                event_index_counter=counter,
+            )
+            self.workers.append(worker)
+
+    async def inject_node_info(self, node_id: NodeId, sender_suffix: str = "") -> None:
+        """Inject a NodeGatheredInfo event for a node into the local event bus."""
+        sender_id = NodeId(f"{node_id}_sender{sender_suffix}")
+        await self.local_event_sender.send(
+            make_gathered_info_event(node_id, sender_id, self.session_id, 0)
+        )
+
+    async def wait_for_topology_nodes(
+        self, count: int, *, timeout: float = 5.0
+    ) -> None:
+        import anyio
+
+        with anyio.fail_after(timeout):
+            while len(list(self.master.state.topology.list_nodes())) < count:
+                await anyio.sleep(0.01)
+
+    async def place_model(
+        self,
+        model_card: ModelCard | None = None,
+        min_nodes: int = 1,
+    ) -> None:
+        card = model_card or TEST_MODEL_CARD
+        await self.command_sender.send(
+            ForwarderCommand(
+                origin=self.master_node_id,
+                command=PlaceInstance(
+                    command_id=CommandId(),
+                    model_card=card,
+                    sharding=Sharding.Pipeline,
+                    instance_meta=InstanceMeta.MlxRing,
+                    min_nodes=min_nodes,
+                ),
+            )
+        )
+
+    async def wait_for_instances(self, count: int, *, timeout: float = 5.0) -> None:
+        import anyio
+
+        with anyio.fail_after(timeout):
+            while len(self.master.state.instances) < count:
+                await anyio.sleep(0.01)
+
+    async def send_chat(
+        self,
+        message: str,
+        model: ModelId | None = None,
+    ) -> CommandId:
+        cmd_id = CommandId()
+        await self.command_sender.send(
+            ForwarderCommand(
+                origin=self.master_node_id,
+                command=TextGeneration(
+                    command_id=cmd_id,
+                    task_params=TextGenerationTaskParams(
+                        model=model or TEST_MODEL_ID,
+                        input=[InputMessage(role="user", content=message)],
+                    ),
+                ),
+            )
+        )
+        return cmd_id
+
+    async def shutdown_master(self) -> None:
+        await self.master.shutdown()
+
+    def shutdown_workers(self) -> None:
+        for w in self.workers:
+            w.shutdown()
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture(autouse=True)
+def fast_election_timeout(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr("exo.shared.election.DEFAULT_ELECTION_TIMEOUT", 0.1)
+
+
+@pytest.fixture
+def caplog(caplog: LogCaptureFixture) -> Iterator[LogCaptureFixture]:
+    handler_id = logger.add(
+        caplog.handler,
+        format="{message}",
+        level=0,
+        filter=lambda record: record["level"].no >= caplog.handler.level,
+        enqueue=True,
+    )
+    yield caplog
+    logger.remove(handler_id)

src/exo/tests/e2e_chaos/test_client_disconnect.py

@@ -0,0 +1,255 @@
+"""E2E Chaos Test: Client disconnect.
+
+Scenarios:
+1. Task cancellation after client disconnect -- a TextGeneration command is
+   sent, then immediately cancelled (simulating browser tab close).
+   Verify the master correctly transitions the task to Cancelled status.
+2. Multiple rapid cancellations -- several chat commands are sent and
+   cancelled in quick succession; no tasks should remain in a stuck state.
+"""
+
+import anyio
+import pytest
+
+from exo.master.main import Master
+from exo.shared.types.commands import (
+    CommandId,
+    ForwarderCommand,
+    ForwarderDownloadCommand,
+    PlaceInstance,
+    TaskCancelled,
+    TextGeneration,
+)
+from exo.shared.types.common import NodeId, SessionId
+from exo.shared.types.events import (
+    ForwarderEvent,
+)
+from exo.shared.types.tasks import TaskStatus
+from exo.shared.types.text_generation import InputMessage, TextGenerationTaskParams
+from exo.shared.types.worker.instances import InstanceMeta
+from exo.shared.types.worker.shards import Sharding
+from exo.utils.channels import channel
+
+from .conftest import (
+    TEST_MODEL_CARD,
+    TEST_MODEL_ID,
+    EventCollector,
+    make_gathered_info_event,
+    make_node_id,
+)
+
+
+@pytest.mark.slow
+@pytest.mark.asyncio
+async def test_task_cancelled_after_client_disconnect() -> None:
+    """Simulate a browser tab close by sending a TextGeneration command
+    followed immediately by a TaskCancelled command.  Verify the task
+    transitions to Cancelled status.
+    """
+    master_nid = make_node_id("master-cancel")
+    session_id = SessionId(master_node_id=master_nid, election_clock=0)
+
+    ge_sender, ge_receiver = channel[ForwarderEvent]()
+    cmd_sender, cmd_receiver = channel[ForwarderCommand]()
+    le_sender, le_receiver = channel[ForwarderEvent]()
+    dl_sender, _dl_receiver = channel[ForwarderDownloadCommand]()
+
+    master = Master(
+        master_nid,
+        session_id,
+        global_event_sender=ge_sender,
+        local_event_receiver=le_receiver,
+        command_receiver=cmd_receiver,
+        download_command_sender=dl_sender,
+    )
+
+    _collector = EventCollector(ge_receiver.clone())
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(master.run)
+
+        # Register node
+        sender_id = NodeId(f"{master_nid}_sender")
+        await le_sender.send(
+            make_gathered_info_event(master_nid, sender_id, session_id, 0)
+        )
+
+        with anyio.fail_after(3):
+            while len(list(master.state.topology.list_nodes())) == 0:
+                await anyio.sleep(0.01)
+
+        # Place instance
+        await cmd_sender.send(
+            ForwarderCommand(
+                origin=master_nid,
+                command=PlaceInstance(
+                    command_id=CommandId(),
+                    model_card=TEST_MODEL_CARD,
+                    sharding=Sharding.Pipeline,
+                    instance_meta=InstanceMeta.MlxRing,
+                    min_nodes=1,
+                ),
+            )
+        )
+
+        with anyio.fail_after(3):
+            while len(master.state.instances) == 0:
+                await anyio.sleep(0.01)
+
+        # Send a chat command
+        chat_cmd_id = CommandId()
+        await cmd_sender.send(
+            ForwarderCommand(
+                origin=master_nid,
+                command=TextGeneration(
+                    command_id=chat_cmd_id,
+                    task_params=TextGenerationTaskParams(
+                        model=TEST_MODEL_ID,
+                        input=[InputMessage(role="user", content="Hello world")],
+                    ),
+                ),
+            )
+        )
+
+        # Wait for the task to be created
+        with anyio.fail_after(3):
+            while len(master.state.tasks) == 0:
+                await anyio.sleep(0.01)
+
+        # Immediately cancel -- simulating browser tab close
+        await cmd_sender.send(
+            ForwarderCommand(
+                origin=master_nid,
+                command=TaskCancelled(
+                    command_id=CommandId(),
+                    cancelled_command_id=chat_cmd_id,
+                ),
+            )
+        )
+
+        # Wait for the task status to be updated to Cancelled
+        with anyio.fail_after(3):
+            while True:
+                tasks_cancelled = [
+                    t
+                    for t in master.state.tasks.values()
+                    if t.task_status == TaskStatus.Cancelled
+                ]
+                if tasks_cancelled:
+                    break
+                await anyio.sleep(0.01)
+
+        assert len(tasks_cancelled) == 1
+
+        await master.shutdown()
+
+
+@pytest.mark.slow
+@pytest.mark.asyncio
+async def test_rapid_cancel_does_not_leave_stuck_tasks() -> None:
+    """Send multiple chat commands and cancel them all rapidly.
+    Verify no tasks remain in Pending or Running state.
+    """
+    master_nid = make_node_id("master-rapid-cancel")
+    session_id = SessionId(master_node_id=master_nid, election_clock=0)
+
+    ge_sender, _ge_receiver = channel[ForwarderEvent]()
+    cmd_sender, cmd_receiver = channel[ForwarderCommand]()
+    le_sender, le_receiver = channel[ForwarderEvent]()
+    dl_sender, _dl_receiver = channel[ForwarderDownloadCommand]()
+
+    master = Master(
+        master_nid,
+        session_id,
+        global_event_sender=ge_sender,
+        local_event_receiver=le_receiver,
+        command_receiver=cmd_receiver,
+        download_command_sender=dl_sender,
+    )
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(master.run)
+
+        # Register node and place instance
+        sender_id = NodeId(f"{master_nid}_sender")
+        await le_sender.send(
+            make_gathered_info_event(master_nid, sender_id, session_id, 0)
+        )
+
+        with anyio.fail_after(3):
+            while len(list(master.state.topology.list_nodes())) == 0:
+                await anyio.sleep(0.01)
+
+        await cmd_sender.send(
+            ForwarderCommand(
+                origin=master_nid,
+                command=PlaceInstance(
+                    command_id=CommandId(),
+                    model_card=TEST_MODEL_CARD,
+                    sharding=Sharding.Pipeline,
+                    instance_meta=InstanceMeta.MlxRing,
+                    min_nodes=1,
+                ),
+            )
+        )
+
+        with anyio.fail_after(3):
+            while len(master.state.instances) == 0:
+                await anyio.sleep(0.01)
+
+        # Send 5 chat commands and immediately cancel each
+        chat_cmd_ids: list[CommandId] = []
+        for i in range(5):
+            cmd_id = CommandId()
+            chat_cmd_ids.append(cmd_id)
+            await cmd_sender.send(
+                ForwarderCommand(
+                    origin=master_nid,
+                    command=TextGeneration(
+                        command_id=cmd_id,
+                        task_params=TextGenerationTaskParams(
+                            model=TEST_MODEL_ID,
+                            input=[InputMessage(role="user", content=f"Message {i}")],
+                        ),
+                    ),
+                )
+            )
+
+        # Wait for all tasks to be created
+        with anyio.fail_after(3):
+            while len(master.state.tasks) < 5:
+                await anyio.sleep(0.01)
+
+        # Cancel all of them
+        for cmd_id in chat_cmd_ids:
+            await cmd_sender.send(
+                ForwarderCommand(
+                    origin=master_nid,
+                    command=TaskCancelled(
+                        command_id=CommandId(),
+                        cancelled_command_id=cmd_id,
+                    ),
+                )
+            )
+
+        # Wait for all cancellations to be processed
+        with anyio.fail_after(3):
+            while True:
+                cancelled_count = sum(
+                    1
+                    for t in master.state.tasks.values()
+                    if t.task_status == TaskStatus.Cancelled
+                )
+                if cancelled_count == 5:
+                    break
+                await anyio.sleep(0.01)
+
+        # No tasks should be Pending or Running
+        stuck = [
+            t
+            for t in master.state.tasks.values()
+            if t.task_status in (TaskStatus.Pending, TaskStatus.Running)
+        ]
+        assert len(stuck) == 0
+
+        await master.shutdown()

src/exo/tests/e2e_chaos/test_concurrent_requests.py

@@ -0,0 +1,395 @@
+"""E2E Chaos Test: Concurrent requests.
+
+Scenarios:
+1. Multiple simultaneous inference requests -- verify they are all created
+   as tasks with no data corruption (unique task IDs, correct model IDs).
+2. Concurrent requests across multiple model instances -- verify tasks are
+   routed to the correct instances.
+3. Concurrent requests with load balancing -- when multiple instances of
+   the same model exist, verify tasks are distributed.
+"""
+
+import anyio
+import pytest
+
+from exo.master.main import Master
+from exo.shared.models.model_cards import ModelCard, ModelTask
+from exo.shared.types.commands import (
+    CommandId,
+    ForwarderCommand,
+    ForwarderDownloadCommand,
+    PlaceInstance,
+    TextGeneration,
+)
+from exo.shared.types.common import ModelId, NodeId, SessionId
+from exo.shared.types.events import (
+    ForwarderEvent,
+)
+from exo.shared.types.memory import Memory
+from exo.shared.types.tasks import TaskStatus
+from exo.shared.types.tasks import TextGeneration as TextGenerationTask
+from exo.shared.types.text_generation import InputMessage, TextGenerationTaskParams
+from exo.shared.types.worker.instances import InstanceMeta
+from exo.shared.types.worker.shards import Sharding
+from exo.utils.channels import channel
+
+from .conftest import (
+    TEST_MODEL_CARD,
+    TEST_MODEL_ID,
+    EventCollector,
+    make_gathered_info_event,
+    make_node_id,
+)
+
+
+@pytest.mark.slow
+@pytest.mark.asyncio
+async def test_concurrent_chat_requests_no_corruption() -> None:
+    """Send multiple TextGeneration commands concurrently and verify each
+    results in a unique task with the correct model and content mapping.
+    """
+    master_nid = make_node_id("master-concurrent")
+    session_id = SessionId(master_node_id=master_nid, election_clock=0)
+
+    ge_sender, ge_receiver = channel[ForwarderEvent]()
+    cmd_sender, cmd_receiver = channel[ForwarderCommand]()
+    le_sender, le_receiver = channel[ForwarderEvent]()
+    dl_sender, _dl_receiver = channel[ForwarderDownloadCommand]()
+
+    master = Master(
+        master_nid,
+        session_id,
+        global_event_sender=ge_sender,
+        local_event_receiver=le_receiver,
+        command_receiver=cmd_receiver,
+        download_command_sender=dl_sender,
+    )
+
+    _collector = EventCollector(ge_receiver.clone())
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(master.run)
+
+        # Set up node and instance
+        sender_id = NodeId(f"{master_nid}_sender")
+        await le_sender.send(
+            make_gathered_info_event(master_nid, sender_id, session_id, 0)
+        )
+
+        with anyio.fail_after(3):
+            while len(list(master.state.topology.list_nodes())) == 0:
+                await anyio.sleep(0.01)
+
+        await cmd_sender.send(
+            ForwarderCommand(
+                origin=master_nid,
+                command=PlaceInstance(
+                    command_id=CommandId(),
+                    model_card=TEST_MODEL_CARD,
+                    sharding=Sharding.Pipeline,
+                    instance_meta=InstanceMeta.MlxRing,
+                    min_nodes=1,
+                ),
+            )
+        )
+
+        with anyio.fail_after(3):
+            while len(master.state.instances) == 0:
+                await anyio.sleep(0.01)
+
+        # Send 10 concurrent chat requests
+        num_requests = 10
+        cmd_ids: list[CommandId] = []
+
+        async def send_chat(index: int) -> None:
+            cmd_id = CommandId()
+            cmd_ids.append(cmd_id)
+            await cmd_sender.send(
+                ForwarderCommand(
+                    origin=master_nid,
+                    command=TextGeneration(
+                        command_id=cmd_id,
+                        task_params=TextGenerationTaskParams(
+                            model=TEST_MODEL_ID,
+                            input=[
+                                InputMessage(
+                                    role="user",
+                                    content=f"Concurrent request #{index}",
+                                )
+                            ],
+                        ),
+                    ),
+                )
+            )
+
+        async with anyio.create_task_group() as send_tg:
+            for i in range(num_requests):
+                send_tg.start_soon(send_chat, i)
+
+        # Wait for all tasks to be created
+        with anyio.fail_after(5):
+            while len(master.state.tasks) < num_requests:
+                await anyio.sleep(0.01)
+
+        # Verify no corruption
+        assert len(master.state.tasks) == num_requests
+
+        # All task IDs should be unique
+        task_ids = list(master.state.tasks.keys())
+        assert len(set(task_ids)) == num_requests
+
+        # All tasks should target the correct model
+        for task in master.state.tasks.values():
+            assert isinstance(task, TextGenerationTask)
+            assert task.task_params.model == TEST_MODEL_ID
+            assert task.task_status == TaskStatus.Pending
+
+        # All tasks should reference the same instance
+        instance_ids = {task.instance_id for task in master.state.tasks.values()}
+        assert len(instance_ids) == 1
+
+        await master.shutdown()
+
+
+@pytest.mark.slow
+@pytest.mark.asyncio
+async def test_concurrent_requests_across_multiple_models() -> None:
+    """Place two different models, then send concurrent requests for each.
+    Verify tasks are routed to the correct model instances.
+    """
+    master_nid = make_node_id("master-multi-model")
+    session_id = SessionId(master_node_id=master_nid, election_clock=0)
+
+    ge_sender, _ge_receiver = channel[ForwarderEvent]()
+    cmd_sender, cmd_receiver = channel[ForwarderCommand]()
+    le_sender, le_receiver = channel[ForwarderEvent]()
+    dl_sender, _dl_receiver = channel[ForwarderDownloadCommand]()
+
+    master = Master(
+        master_nid,
+        session_id,
+        global_event_sender=ge_sender,
+        local_event_receiver=le_receiver,
+        command_receiver=cmd_receiver,
+        download_command_sender=dl_sender,
+    )
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(master.run)
+
+        # Register node
+        sender_id = NodeId(f"{master_nid}_sender")
+        await le_sender.send(
+            make_gathered_info_event(master_nid, sender_id, session_id, 0)
+        )
+
+        with anyio.fail_after(3):
+            while len(list(master.state.topology.list_nodes())) == 0:
+                await anyio.sleep(0.01)
+
+        # Place two different models
+        model_a_id = ModelId("test-model/model-a")
+        model_a_card = ModelCard(
+            model_id=model_a_id,
+            n_layers=16,
+            storage_size=Memory.from_bytes(500_000),
+            hidden_size=2048,
+            supports_tensor=True,
+            tasks=[ModelTask.TextGeneration],
+        )
+
+        model_b_id = ModelId("test-model/model-b")
+        model_b_card = ModelCard(
+            model_id=model_b_id,
+            n_layers=32,
+            storage_size=Memory.from_bytes(500_000),
+            hidden_size=4096,
+            supports_tensor=True,
+            tasks=[ModelTask.TextGeneration],
+        )
+
+        for card in [model_a_card, model_b_card]:
+            await cmd_sender.send(
+                ForwarderCommand(
+                    origin=master_nid,
+                    command=PlaceInstance(
+                        command_id=CommandId(),
+                        model_card=card,
+                        sharding=Sharding.Pipeline,
+                        instance_meta=InstanceMeta.MlxRing,
+                        min_nodes=1,
+                    ),
+                )
+            )
+
+        with anyio.fail_after(5):
+            while len(master.state.instances) < 2:
+                await anyio.sleep(0.01)
+
+        # Map instance IDs to models
+        instance_to_model: dict[str, ModelId] = {}
+        for iid, inst in master.state.instances.items():
+            instance_to_model[iid] = inst.shard_assignments.model_id
+
+        # Send concurrent requests for both models
+        async def send_for_model(model_id: ModelId, count: int) -> None:
+            for i in range(count):
+                await cmd_sender.send(
+                    ForwarderCommand(
+                        origin=master_nid,
+                        command=TextGeneration(
+                            command_id=CommandId(),
+                            task_params=TextGenerationTaskParams(
+                                model=model_id,
+                                input=[
+                                    InputMessage(
+                                        role="user",
+                                        content=f"Request for {model_id} #{i}",
+                                    )
+                                ],
+                            ),
+                        ),
+                    )
+                )
+
+        async with anyio.create_task_group() as send_tg:
+            send_tg.start_soon(send_for_model, model_a_id, 3)
+            send_tg.start_soon(send_for_model, model_b_id, 3)
+
+        # Wait for all 6 tasks
+        with anyio.fail_after(5):
+            while len(master.state.tasks) < 6:
+                await anyio.sleep(0.01)
+
+        # Verify task routing
+        model_a_tasks = [
+            t
+            for t in master.state.tasks.values()
+            if isinstance(t, TextGenerationTask) and t.task_params.model == model_a_id
+        ]
+        model_b_tasks = [
+            t
+            for t in master.state.tasks.values()
+            if isinstance(t, TextGenerationTask) and t.task_params.model == model_b_id
+        ]
+
+        assert len(model_a_tasks) == 3
+        assert len(model_b_tasks) == 3
+
+        # All model_a tasks should reference the model_a instance
+        model_a_instance_ids = {
+            iid for iid, mid in instance_to_model.items() if mid == model_a_id
+        }
+        for task in model_a_tasks:
+            assert task.instance_id in model_a_instance_ids
+
+        # All model_b tasks should reference the model_b instance
+        model_b_instance_ids = {
+            iid for iid, mid in instance_to_model.items() if mid == model_b_id
+        }
+        for task in model_b_tasks:
+            assert task.instance_id in model_b_instance_ids
+
+        await master.shutdown()
+
+
+@pytest.mark.slow
+@pytest.mark.asyncio
+async def test_event_index_monotonically_increases_under_load() -> None:
+    """Under heavy concurrent command load, verify the master's event log
+    index increases monotonically with no gaps or duplicates.
+    """
+    master_nid = make_node_id("master-monotonic")
+    session_id = SessionId(master_node_id=master_nid, election_clock=0)
+
+    ge_sender, ge_receiver = channel[ForwarderEvent]()
+    cmd_sender, cmd_receiver = channel[ForwarderCommand]()
+    le_sender, le_receiver = channel[ForwarderEvent]()
+    dl_sender, _dl_receiver = channel[ForwarderDownloadCommand]()
+
+    master = Master(
+        master_nid,
+        session_id,
+        global_event_sender=ge_sender,
+        local_event_receiver=le_receiver,
+        command_receiver=cmd_receiver,
+        download_command_sender=dl_sender,
+    )
+
+    collector = EventCollector(ge_receiver.clone())
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(master.run)
+
+        # Register node and place instance
+        sender_id = NodeId(f"{master_nid}_sender")
+        await le_sender.send(
+            make_gathered_info_event(master_nid, sender_id, session_id, 0)
+        )
+
+        with anyio.fail_after(3):
+            while len(list(master.state.topology.list_nodes())) == 0:
+                await anyio.sleep(0.01)
+
+        await cmd_sender.send(
+            ForwarderCommand(
+                origin=master_nid,
+                command=PlaceInstance(
+                    command_id=CommandId(),
+                    model_card=TEST_MODEL_CARD,
+                    sharding=Sharding.Pipeline,
+                    instance_meta=InstanceMeta.MlxRing,
+                    min_nodes=1,
+                ),
+            )
+        )
+
+        with anyio.fail_after(3):
+            while len(master.state.instances) == 0:
+                await anyio.sleep(0.01)
+
+        # Blast 20 concurrent commands
+        async def blast_commands(start: int, count: int) -> None:
+            for i in range(count):
+                await cmd_sender.send(
+                    ForwarderCommand(
+                        origin=master_nid,
+                        command=TextGeneration(
+                            command_id=CommandId(),
+                            task_params=TextGenerationTaskParams(
+                                model=TEST_MODEL_ID,
+                                input=[
+                                    InputMessage(
+                                        role="user",
+                                        content=f"Blast {start + i}",
+                                    )
+                                ],
+                            ),
+                        ),
+                    )
+                )
+
+        async with anyio.create_task_group() as blast_tg:
+            blast_tg.start_soon(blast_commands, 0, 10)
+            blast_tg.start_soon(blast_commands, 10, 10)
+
+        # Wait for all tasks
+        with anyio.fail_after(5):
+            while len(master.state.tasks) < 20:
+                await anyio.sleep(0.01)
+
+        # Collect all events and verify monotonic indexing
+        # NodeGatheredInfo(0) + InstanceCreated(1) + 20 TaskCreated = 22 events
+        await collector.wait_for_event_count(22, timeout=5.0)
+
+        events = collector.indexed_events
+        indices = [e.idx for e in events]
+
+        # Should be 0, 1, 2, ..., N-1 with no gaps
+        expected = list(range(len(indices)))
+        assert indices == expected
+
+        # last_event_applied_idx should match
+        assert master.state.last_event_applied_idx == len(events) - 1
+
+        await master.shutdown()

src/exo/tests/e2e_chaos/test_distributed_model_loading.py

@@ -0,0 +1,356 @@
+"""E2E Chaos Test: Large model distributed loading.
+
+Scenarios:
+1. Multi-node sharding -- place a model with min_nodes > 1, verify sharding
+   is distributed across multiple nodes with correct shard assignments.
+2. Single-node gets all layers -- place on 1 node, verify full assignment.
+3. Three-node sharding -- verify 3-way distribution.
+"""
+
+import anyio
+import pytest
+
+from exo.master.main import Master
+from exo.shared.models.model_cards import ModelCard, ModelTask
+from exo.shared.types.commands import (
+    CommandId,
+    ForwarderCommand,
+    ForwarderDownloadCommand,
+    PlaceInstance,
+)
+from exo.shared.types.common import ModelId, NodeId, SessionId
+from exo.shared.types.events import (
+    ForwarderEvent,
+)
+from exo.shared.types.memory import Memory
+from exo.shared.types.worker.instances import InstanceMeta, MlxRingInstance
+from exo.shared.types.worker.shards import PipelineShardMetadata, Sharding
+from exo.utils.channels import Sender, channel
+
+from .conftest import (
+    TEST_MODEL_CARD,
+    make_gathered_info_event,
+    make_node_id,
+    make_topology_edge_event,
+)
+
+# A model large enough to need sharding but small enough to fit in test node memory
+# Each test node has 8GB available, so 2 nodes = 16GB, 3 nodes = 24GB.
+# storage_size < total cluster memory to pass the memory filter.
+LARGE_MODEL_CARD = ModelCard(
+    model_id=ModelId("test-model/large-70b-4bit"),
+    n_layers=80,
+    storage_size=Memory.from_bytes(4 * 1024 * 1024 * 1024),
+    hidden_size=8192,
+    supports_tensor=True,
+    tasks=[ModelTask.TextGeneration],
+)
+
+
+async def _register_node(
+    le_sender: Sender[ForwarderEvent],
+    node_id: NodeId,
+    session_id: SessionId,
+) -> None:
+    """Register a node by injecting NodeGatheredInfo."""
+    sender_id = NodeId(f"{node_id}_sender")
+    await le_sender.send(make_gathered_info_event(node_id, sender_id, session_id, 0))
+
+
+async def _add_bidirectional_edge(
+    le_sender: Sender[ForwarderEvent],
+    node_a: NodeId,
+    node_b: NodeId,
+    session_id: SessionId,
+    sender_id: NodeId,
+    origin_idx_start: int,
+    ip_a: int,
+    ip_b: int,
+) -> None:
+    """Add bidirectional topology edges between two nodes."""
+    await le_sender.send(
+        make_topology_edge_event(
+            node_a, node_b, sender_id, session_id, origin_idx_start, ip_suffix=ip_b
+        )
+    )
+    await le_sender.send(
+        make_topology_edge_event(
+            node_b, node_a, sender_id, session_id, origin_idx_start + 1, ip_suffix=ip_a
+        )
+    )
+
+
+@pytest.mark.slow
+@pytest.mark.asyncio
+async def test_multi_node_sharding_distributes_layers() -> None:
+    """Place a model with min_nodes=2 on a cluster with 2 connected nodes.
+    Verify the resulting instance has shard assignments spanning both nodes.
+    """
+    master_nid = make_node_id("master-shard")
+    session_id = SessionId(master_node_id=master_nid, election_clock=0)
+
+    ge_sender, _ge_receiver = channel[ForwarderEvent]()
+    cmd_sender, cmd_receiver = channel[ForwarderCommand]()
+    le_sender, le_receiver = channel[ForwarderEvent]()
+    dl_sender, _dl_receiver = channel[ForwarderDownloadCommand]()
+
+    master = Master(
+        master_nid,
+        session_id,
+        global_event_sender=ge_sender,
+        local_event_receiver=le_receiver,
+        command_receiver=cmd_receiver,
+        download_command_sender=dl_sender,
+    )
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(master.run)
+
+        worker_a = make_node_id("shard-worker-a")
+        worker_b = make_node_id("shard-worker-b")
+
+        # Register both worker nodes (each sender uses origin_idx=0)
+        for nid in [worker_a, worker_b]:
+            await _register_node(le_sender, nid, session_id)
+
+        with anyio.fail_after(3):
+            while len(list(master.state.topology.list_nodes())) < 2:
+                await anyio.sleep(0.01)
+
+        # Add bidirectional edges to form a 2-node cycle (A <-> B)
+        edge_sender = NodeId("edge_sender")
+        await _add_bidirectional_edge(
+            le_sender, worker_a, worker_b, session_id, edge_sender, 0, 1, 2
+        )
+
+        # Wait for edges to be processed
+        with anyio.fail_after(3):
+            while len(list(master.state.topology.list_connections())) < 2:
+                await anyio.sleep(0.01)
+
+        # Place a large model requiring 2 nodes
+        await cmd_sender.send(
+            ForwarderCommand(
+                origin=master_nid,
+                command=PlaceInstance(
+                    command_id=CommandId(),
+                    model_card=LARGE_MODEL_CARD,
+                    sharding=Sharding.Pipeline,
+                    instance_meta=InstanceMeta.MlxRing,
+                    min_nodes=2,
+                ),
+            )
+        )
+
+        with anyio.fail_after(5):
+            while len(master.state.instances) == 0:
+                await anyio.sleep(0.01)
+
+        instance_id = next(iter(master.state.instances))
+        instance = master.state.instances[instance_id]
+        assert isinstance(instance, MlxRingInstance)
+
+        shard_assignments = instance.shard_assignments
+        runner_shards = shard_assignments.runner_to_shard
+
+        assert len(runner_shards) == 2
+
+        assigned_nodes = set(shard_assignments.node_to_runner.keys())
+        assert worker_a in assigned_nodes
+        assert worker_b in assigned_nodes
+
+        shards = list(runner_shards.values())
+        assert all(isinstance(s, PipelineShardMetadata) for s in shards)
+        pipeline_shards = [s for s in shards if isinstance(s, PipelineShardMetadata)]
+
+        assert all(s.world_size == 2 for s in pipeline_shards)
+        ranks = {s.device_rank for s in pipeline_shards}
+        assert ranks == {0, 1}
+
+        sorted_shards = sorted(pipeline_shards, key=lambda s: s.device_rank)
+        assert sorted_shards[0].start_layer == 0
+        assert sorted_shards[-1].end_layer == LARGE_MODEL_CARD.n_layers
+
+        total_layers = sum(s.end_layer - s.start_layer for s in sorted_shards)
+        assert total_layers == LARGE_MODEL_CARD.n_layers
+
+        await master.shutdown()
+
+
+@pytest.mark.slow
+@pytest.mark.asyncio
+async def test_single_node_gets_all_layers() -> None:
+    """Place a model with min_nodes=1 on a single node.  Verify the
+    instance has one runner assigned all layers (world_size=1).
+    """
+    master_nid = make_node_id("master-single")
+    session_id = SessionId(master_node_id=master_nid, election_clock=0)
+
+    ge_sender, _ge_receiver = channel[ForwarderEvent]()
+    cmd_sender, cmd_receiver = channel[ForwarderCommand]()
+    le_sender, le_receiver = channel[ForwarderEvent]()
+    dl_sender, _dl_receiver = channel[ForwarderDownloadCommand]()
+
+    master = Master(
+        master_nid,
+        session_id,
+        global_event_sender=ge_sender,
+        local_event_receiver=le_receiver,
+        command_receiver=cmd_receiver,
+        download_command_sender=dl_sender,
+    )
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(master.run)
+
+        worker_nid = make_node_id("single-worker")
+        await _register_node(le_sender, worker_nid, session_id)
+
+        with anyio.fail_after(3):
+            while len(list(master.state.topology.list_nodes())) < 1:
+                await anyio.sleep(0.01)
+
+        await cmd_sender.send(
+            ForwarderCommand(
+                origin=master_nid,
+                command=PlaceInstance(
+                    command_id=CommandId(),
+                    model_card=TEST_MODEL_CARD,
+                    sharding=Sharding.Pipeline,
+                    instance_meta=InstanceMeta.MlxRing,
+                    min_nodes=1,
+                ),
+            )
+        )
+
+        with anyio.fail_after(3):
+            while len(master.state.instances) == 0:
+                await anyio.sleep(0.01)
+
+        instance_id = next(iter(master.state.instances))
+        instance = master.state.instances[instance_id]
+        assert isinstance(instance, MlxRingInstance)
+
+        shards = list(instance.shard_assignments.runner_to_shard.values())
+        assert len(shards) == 1
+
+        shard = shards[0]
+        assert isinstance(shard, PipelineShardMetadata)
+        assert shard.world_size == 1
+        assert shard.device_rank == 0
+        assert shard.start_layer == 0
+        assert shard.end_layer == TEST_MODEL_CARD.n_layers
+
+        await master.shutdown()
+
+
+@pytest.mark.slow
+@pytest.mark.asyncio
+async def test_three_node_sharding_distributes_evenly() -> None:
+    """Place a model across 3 connected nodes.  Verify all 3 get shard assignments."""
+    master_nid = make_node_id("master-3way")
+    session_id = SessionId(master_node_id=master_nid, election_clock=0)
+
+    ge_sender, _ge_receiver = channel[ForwarderEvent]()
+    cmd_sender, cmd_receiver = channel[ForwarderCommand]()
+    le_sender, le_receiver = channel[ForwarderEvent]()
+    dl_sender, _dl_receiver = channel[ForwarderDownloadCommand]()
+
+    master = Master(
+        master_nid,
+        session_id,
+        global_event_sender=ge_sender,
+        local_event_receiver=le_receiver,
+        command_receiver=cmd_receiver,
+        download_command_sender=dl_sender,
+    )
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(master.run)
+
+        workers: list[NodeId] = []
+        for i in range(3):
+            nid = make_node_id(f"three-worker-{i}")
+            workers.append(nid)
+            await _register_node(le_sender, nid, session_id)
+
+        with anyio.fail_after(3):
+            while len(list(master.state.topology.list_nodes())) < 3:
+                await anyio.sleep(0.01)
+
+        # Add bidirectional edges to form a fully connected 3-node cycle:
+        # A <-> B, B <-> C, C <-> A
+        edge_sender = NodeId("edge_sender_3way")
+        idx = 0
+        ip_counter = 10
+        for i in range(3):
+            source = workers[i]
+            sink = workers[(i + 1) % 3]
+            # Forward edge
+            await le_sender.send(
+                make_topology_edge_event(
+                    source,
+                    sink,
+                    edge_sender,
+                    session_id,
+                    idx,
+                    ip_suffix=ip_counter,
+                )
+            )
+            idx += 1
+            ip_counter += 1
+            # Reverse edge
+            await le_sender.send(
+                make_topology_edge_event(
+                    sink,
+                    source,
+                    edge_sender,
+                    session_id,
+                    idx,
+                    ip_suffix=ip_counter,
+                )
+            )
+            idx += 1
+            ip_counter += 1
+
+        # Wait for all 6 edges (3 pairs x 2 directions)
+        with anyio.fail_after(3):
+            while len(list(master.state.topology.list_connections())) < 6:
+                await anyio.sleep(0.01)
+
+        await cmd_sender.send(
+            ForwarderCommand(
+                origin=master_nid,
+                command=PlaceInstance(
+                    command_id=CommandId(),
+                    model_card=LARGE_MODEL_CARD,
+                    sharding=Sharding.Pipeline,
+                    instance_meta=InstanceMeta.MlxRing,
+                    min_nodes=3,
+                ),
+            )
+        )
+
+        with anyio.fail_after(5):
+            while len(master.state.instances) == 0:
+                await anyio.sleep(0.01)
+
+        instance = next(iter(master.state.instances.values()))
+        assert isinstance(instance, MlxRingInstance)
+
+        assignments = instance.shard_assignments
+        assert len(assignments.runner_to_shard) == 3
+        assert len(assignments.node_to_runner) == 3
+
+        for w in workers:
+            assert w in assignments.node_to_runner
+
+        shards = list(assignments.runner_to_shard.values())
+        ranks = {s.device_rank for s in shards if isinstance(s, PipelineShardMetadata)}
+        assert ranks == {0, 1, 2}
+
+        pipeline_shards = [s for s in shards if isinstance(s, PipelineShardMetadata)]
+        total_layers = sum(s.end_layer - s.start_layer for s in pipeline_shards)
+        assert total_layers == LARGE_MODEL_CARD.n_layers
+
+        await master.shutdown()

src/exo/tests/e2e_chaos/test_failure_recovery.py

@@ -0,0 +1,272 @@
+"""E2E Chaos Test: Failure recovery.
+
+Scenarios:
+1. Master crash and re-election -- master shuts down, a new election round
+   produces a new master, workers re-converge.
+2. Worker crash during task execution -- runner death is detected, instance
+   is cleaned up, and cluster recovers.
+"""
+
+import anyio
+import pytest
+
+from exo.master.main import Master
+from exo.shared.types.commands import (
+    CommandId,
+    ForwarderCommand,
+    ForwarderDownloadCommand,
+    PlaceInstance,
+)
+from exo.shared.types.common import NodeId, SessionId
+from exo.shared.types.events import (
+    ForwarderEvent,
+    RunnerStatusUpdated,
+)
+from exo.shared.types.worker.instances import InstanceMeta
+from exo.shared.types.worker.runners import RunnerFailed
+from exo.shared.types.worker.shards import Sharding
+from exo.utils.channels import channel
+
+from .conftest import (
+    TEST_MODEL_CARD,
+    EventCollector,
+    MiniCluster,
+    make_gathered_info_event,
+    make_node_id,
+)
+
+
+@pytest.mark.slow
+@pytest.mark.asyncio
+async def test_master_crash_and_reelection() -> None:
+    """Simulate master crash by shutting it down, then verify a new master
+    can be started with fresh state and begin accepting commands.
+
+    This tests the scenario where the elected master dies and a new election
+    must take place.  We simulate the election result directly (since
+    Election is tested separately) and verify the new master works.
+    """
+    cluster = MiniCluster(node_count=1)
+    old_instance_id: str = ""
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(cluster.master.run)
+
+        # Set up initial state
+        await cluster.inject_node_info(cluster.master_node_id)
+        await cluster.wait_for_topology_nodes(1)
+        await cluster.place_model()
+        await cluster.wait_for_instances(1)
+
+        # Verify initial state
+        assert len(cluster.master.state.instances) == 1
+        old_instance_id = next(iter(cluster.master.state.instances))
+
+        # --- Crash the master ---
+        await cluster.shutdown_master()
+
+    # --- Start a new master (simulating re-election) ---
+    new_master_nid = make_node_id("new-master")
+    new_session_id = SessionId(master_node_id=new_master_nid, election_clock=1)
+
+    ge_sender, ge_receiver = channel[ForwarderEvent]()
+    cmd_sender, cmd_receiver = channel[ForwarderCommand]()
+    le_sender, le_receiver = channel[ForwarderEvent]()
+    dl_sender, _dl_receiver = channel[ForwarderDownloadCommand]()
+
+    new_master = Master(
+        new_master_nid,
+        new_session_id,
+        global_event_sender=ge_sender,
+        local_event_receiver=le_receiver,
+        command_receiver=cmd_receiver,
+        download_command_sender=dl_sender,
+    )
+
+    _new_collector = EventCollector(ge_receiver.clone())
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(new_master.run)
+
+        # New master starts with clean state
+        assert len(new_master.state.instances) == 0
+        assert new_master.state.last_event_applied_idx == -1
+
+        # Re-register node with the new master
+        sender_id = NodeId(f"{new_master_nid}_sender_new")
+        await le_sender.send(
+            make_gathered_info_event(new_master_nid, sender_id, new_session_id, 0)
+        )
+
+        # Wait for topology to be rebuilt
+        with anyio.fail_after(3):
+            while len(list(new_master.state.topology.list_nodes())) == 0:
+                await anyio.sleep(0.01)
+
+        # Place a new model instance on the new master
+        await cmd_sender.send(
+            ForwarderCommand(
+                origin=new_master_nid,
+                command=PlaceInstance(
+                    command_id=CommandId(),
+                    model_card=TEST_MODEL_CARD,
+                    sharding=Sharding.Pipeline,
+                    instance_meta=InstanceMeta.MlxRing,
+                    min_nodes=1,
+                ),
+            )
+        )
+
+        with anyio.fail_after(3):
+            while len(new_master.state.instances) == 0:
+                await anyio.sleep(0.01)
+
+        # Verify new master is functional
+        assert len(new_master.state.instances) == 1
+        new_instance_id = next(iter(new_master.state.instances))
+        # New instance should be different from old one
+        assert new_instance_id != old_instance_id
+
+        await new_master.shutdown()
+
+
+@pytest.mark.slow
+@pytest.mark.asyncio
+async def test_runner_failure_triggers_instance_cleanup() -> None:
+    """Simulate a runner failure by injecting a RunnerStatusUpdated(RunnerFailed)
+    event.  Verify that the master's plan loop eventually detects the broken
+    instance (no connected node for the runner) and cleans it up.
+    """
+    master_nid = make_node_id("master-runner-fail")
+    session_id = SessionId(master_node_id=master_nid, election_clock=0)
+
+    ge_sender, ge_receiver = channel[ForwarderEvent]()
+    cmd_sender, cmd_receiver = channel[ForwarderCommand]()
+    le_sender, le_receiver = channel[ForwarderEvent]()
+    dl_sender, _dl_receiver = channel[ForwarderDownloadCommand]()
+
+    master = Master(
+        master_nid,
+        session_id,
+        global_event_sender=ge_sender,
+        local_event_receiver=le_receiver,
+        command_receiver=cmd_receiver,
+        download_command_sender=dl_sender,
+    )
+
+    _collector = EventCollector(ge_receiver.clone())
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(master.run)
+
+        # Register a worker node
+        worker_nid = make_node_id("worker-failing")
+        sender_id = NodeId(f"{worker_nid}_sender")
+        await le_sender.send(
+            make_gathered_info_event(worker_nid, sender_id, session_id, 0)
+        )
+
+        with anyio.fail_after(3):
+            while len(list(master.state.topology.list_nodes())) == 0:
+                await anyio.sleep(0.01)
+
+        # Place a model instance
+        await cmd_sender.send(
+            ForwarderCommand(
+                origin=master_nid,
+                command=PlaceInstance(
+                    command_id=CommandId(),
+                    model_card=TEST_MODEL_CARD,
+                    sharding=Sharding.Pipeline,
+                    instance_meta=InstanceMeta.MlxRing,
+                    min_nodes=1,
+                ),
+            )
+        )
+
+        with anyio.fail_after(3):
+            while len(master.state.instances) == 0:
+                await anyio.sleep(0.01)
+
+        instance_id = next(iter(master.state.instances))
+        instance = master.state.instances[instance_id]
+        runner_id = next(iter(instance.shard_assignments.runner_to_shard))
+
+        # Inject a RunnerFailed event from the worker
+        await le_sender.send(
+            ForwarderEvent(
+                origin_idx=1,
+                origin=sender_id,
+                session=session_id,
+                event=RunnerStatusUpdated(
+                    runner_id=runner_id,
+                    runner_status=RunnerFailed(
+                        error_message="Simulated OOM kill (exitcode=137)"
+                    ),
+                ),
+            )
+        )
+
+        # Wait for the runner failure to be processed
+        with anyio.fail_after(3):
+            while runner_id not in master.state.runners:
+                await anyio.sleep(0.01)
+
+        # The runner status should be RunnerFailed
+        assert isinstance(master.state.runners[runner_id], RunnerFailed)
+
+        await master.shutdown()
+
+
+@pytest.mark.slow
+@pytest.mark.asyncio
+async def test_election_recovers_after_multiple_node_joins() -> None:
+    """Verify that the election protocol correctly handles rapid node
+    join/leave events by running multiple election rounds.
+    """
+    from exo.routing.connection_message import ConnectionMessage, ConnectionMessageType
+    from exo.shared.election import Election, ElectionMessage, ElectionResult
+
+    em_out_tx, em_out_rx = channel[ElectionMessage]()
+    em_in_tx, em_in_rx = channel[ElectionMessage]()
+    er_tx, er_rx = channel[ElectionResult]()
+    cm_tx, cm_rx = channel[ConnectionMessage]()
+    co_tx, co_rx = channel[ForwarderCommand]()
+
+    election = Election(
+        node_id=NodeId("SURVIVOR"),
+        election_message_receiver=em_in_rx,
+        election_message_sender=em_out_tx,
+        election_result_sender=er_tx,
+        connection_message_receiver=cm_rx,
+        command_receiver=co_rx,
+        is_candidate=True,
+    )
+
+    async with anyio.create_task_group() as tg:
+        with anyio.fail_after(5):
+            tg.start_soon(election.run)
+
+            # Simulate rapid node joins via connection messages
+            for i in range(3):
+                await cm_tx.send(
+                    ConnectionMessage(
+                        node_id=NodeId(f"joiner-{i}"),
+                        connection_type=ConnectionMessageType.Connected,
+                        remote_ipv4=f"10.0.0.{i + 1}",
+                        remote_tcp_port=52415,
+                    )
+                )
+                # Each connection triggers a new election round
+                while True:
+                    got = await em_out_rx.receive()
+                    if got.proposed_session.master_node_id == NodeId("SURVIVOR"):
+                        break
+
+            # After all joins, an election result should eventually be produced
+            result = await er_rx.receive()
+            assert result.session_id.master_node_id == NodeId("SURVIVOR")
+
+            em_in_tx.close()
+            cm_tx.close()
+            co_tx.close()

src/exo/tests/e2e_chaos/test_networking_resilience.py

@@ -0,0 +1,227 @@
+"""E2E Chaos Test: Networking resilience.
+
+Scenarios:
+1. Node disconnect mid-inference -- a worker stops receiving global events, then
+   reconnects and catches up via the event buffer / nack mechanism.
+2. Master detects stale node and times it out, then the node re-announces.
+"""
+
+import anyio
+import pytest
+
+from exo.master.main import Master
+from exo.shared.types.commands import (
+    ForwarderCommand,
+    ForwarderDownloadCommand,
+)
+from exo.shared.types.common import NodeId, SessionId
+from exo.shared.types.events import (
+    ForwarderEvent,
+    InstanceCreated,
+    NodeGatheredInfo,
+    TaskCreated,
+)
+from exo.utils.channels import channel
+
+from .conftest import (
+    EventCollector,
+    MiniCluster,
+    make_gathered_info_event,
+    make_node_id,
+)
+
+
+@pytest.mark.slow
+@pytest.mark.asyncio
+async def test_node_disconnect_and_reconnect_event_replay() -> None:
+    """Simulate a node disconnecting by closing its global event receiver,
+    then reconnecting with a fresh receiver.
+
+    After reconnection, events that were broadcast while the node was
+    disconnected should be replayed to the new receiver via the shared
+    channel state.  The master's state should remain consistent.
+    """
+    cluster = MiniCluster(node_count=1)
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(cluster.master.run)
+
+        # Register the master node so topology is populated
+        await cluster.inject_node_info(cluster.master_node_id)
+        await cluster.wait_for_topology_nodes(1)
+
+        # Place a model instance
+        await cluster.place_model()
+        await cluster.wait_for_instances(1)
+
+        # Verify instance was created
+        assert len(cluster.master.state.instances) == 1
+
+        # --- Simulate disconnection ---
+        # The worker's global event receiver is independent; we just verify
+        # that the master continues to accept commands while a worker is gone.
+        _first_instance_id = next(iter(cluster.master.state.instances))
+
+        # Send a chat command while "disconnected" worker can't process
+        _cmd_id = await cluster.send_chat("Hello during disconnect")
+
+        # Give master time to process the command
+        await cluster.event_collector.wait_for_event_count(3, timeout=3.0)
+
+        events = cluster.event_collector.indexed_events
+        # Should have: NodeGatheredInfo, InstanceCreated, TaskCreated
+        assert any(isinstance(e.event, NodeGatheredInfo) for e in events)
+        assert any(isinstance(e.event, InstanceCreated) for e in events)
+        assert any(isinstance(e.event, TaskCreated) for e in events)
+
+        # --- Simulate reconnection ---
+        # A reconnecting node gets a fresh receiver clone and catches up
+        reconnect_receiver = cluster.global_event_internal_receiver.clone()
+        _reconnect_collector = EventCollector(reconnect_receiver)
+
+        # The new receiver should see future events; existing events are in
+        # the master's event log (which would be replayed via RequestEventLog
+        # in production). Here we verify the channel infrastructure works.
+        await cluster.send_chat("Hello after reconnect")
+        await anyio.sleep(0.1)
+
+        # Master state should now have 2 tasks
+        assert len(cluster.master.state.tasks) == 2
+
+        # The master's state is consistent throughout
+        assert len(cluster.master.state.instances) == 1
+        assert cluster.master.state.last_event_applied_idx >= 3
+
+        await cluster.shutdown_master()
+
+
+@pytest.mark.slow
+@pytest.mark.asyncio
+async def test_master_detects_timed_out_node_and_cleans_state() -> None:
+    """Verify that the master's plan loop detects a node that hasn't sent
+    a heartbeat (NodeGatheredInfo) recently and emits NodeTimedOut, cleaning
+    up topology and related state.
+    """
+    master_nid = make_node_id("master-timeout")
+    session_id = SessionId(master_node_id=master_nid, election_clock=0)
+
+    ge_sender, ge_receiver = channel[ForwarderEvent]()
+    _cmd_sender, cmd_receiver = channel[ForwarderCommand]()
+    le_sender, le_receiver = channel[ForwarderEvent]()
+    dl_sender, _dl_receiver = channel[ForwarderDownloadCommand]()
+
+    master = Master(
+        master_nid,
+        session_id,
+        global_event_sender=ge_sender,
+        local_event_receiver=le_receiver,
+        command_receiver=cmd_receiver,
+        download_command_sender=dl_sender,
+    )
+
+    _collector = EventCollector(ge_receiver.clone())
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(master.run)
+
+        # Register two nodes
+        stale_node = make_node_id("stale")
+        alive_node = make_node_id("alive")
+
+        for node_id, suffix in [(stale_node, "_s0"), (alive_node, "_a0")]:
+            sender_id = NodeId(f"{node_id}_sender{suffix}")
+            await le_sender.send(
+                make_gathered_info_event(node_id, sender_id, session_id, 0)
+            )
+
+        # Wait for both nodes in topology
+        with anyio.fail_after(3):
+            while len(list(master.state.topology.list_nodes())) < 2:
+                await anyio.sleep(0.01)
+
+        assert stale_node in master.state.last_seen
+        assert alive_node in master.state.last_seen
+
+        # Manually expire the stale node's last_seen time by patching the state
+        # (in production, the _plan loop checks every 10s with a 30s threshold)
+        from datetime import timedelta
+
+        old_time = master.state.last_seen[stale_node] - timedelta(seconds=60)
+        patched_last_seen = {**master.state.last_seen, stale_node: old_time}
+        master.state = master.state.model_copy(update={"last_seen": patched_last_seen})
+
+        # Trigger the plan loop manually to speed up the test
+        # The plan loop checks for stale nodes
+        # We wait for the NodeTimedOut event to be emitted
+        with anyio.fail_after(15):
+            while stale_node in master.state.last_seen:
+                await anyio.sleep(0.1)
+
+        # Stale node should be removed from topology
+        assert stale_node not in set(master.state.topology.list_nodes())
+
+        # Alive node should still be present
+        assert alive_node in set(master.state.topology.list_nodes())
+        assert alive_node in master.state.last_seen
+
+        await master.shutdown()
+
+
+@pytest.mark.slow
+@pytest.mark.asyncio
+async def test_event_ordering_preserved_under_concurrent_writers() -> None:
+    """Multiple sources writing local events concurrently.  Verify that the
+    master's MultiSourceBuffer correctly sequences events from each source
+    and the final state is consistent.
+    """
+    master_nid = make_node_id("master-ordering")
+    session_id = SessionId(master_node_id=master_nid, election_clock=0)
+
+    ge_sender, ge_receiver = channel[ForwarderEvent]()
+    _cmd_sender, cmd_receiver = channel[ForwarderCommand]()
+    le_sender, le_receiver = channel[ForwarderEvent]()
+    dl_sender, _dl_receiver = channel[ForwarderDownloadCommand]()
+
+    master = Master(
+        master_nid,
+        session_id,
+        global_event_sender=ge_sender,
+        local_event_receiver=le_receiver,
+        command_receiver=cmd_receiver,
+        download_command_sender=dl_sender,
+    )
+
+    _collector = EventCollector(ge_receiver.clone())
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(master.run)
+
+        # Inject events from 3 different "worker" sources concurrently
+        node_ids = [make_node_id(f"concurrent-{i}") for i in range(3)]
+
+        async def inject_events(node_id: NodeId, count: int) -> None:
+            for idx in range(count):
+                sender_id = NodeId(f"{node_id}_sender")
+                await le_sender.send(
+                    make_gathered_info_event(node_id, sender_id, session_id, idx)
+                )
+                await anyio.sleep(0.001)  # slight jitter
+
+        async with anyio.create_task_group() as inject_tg:
+            for nid in node_ids:
+                inject_tg.start_soon(inject_events, nid, 5)
+
+        # Wait for master to process all events (3 nodes * 5 events each = 15)
+        with anyio.fail_after(5):
+            while master.state.last_event_applied_idx < 14:
+                await anyio.sleep(0.01)
+
+        # All 3 nodes should be visible in topology
+        topo_nodes = set(master.state.topology.list_nodes())
+        for nid in node_ids:
+            assert nid in topo_nodes
+
+        # Event indices should be sequential with no gaps
+        assert master.state.last_event_applied_idx == 14
+
+        await master.shutdown()

src/exo/tests/e2e_chaos/test_node_join_leave.py

@@ -0,0 +1,267 @@
+"""E2E Chaos Test: Node join/leave during operation.
+
+Scenarios:
+1. Add nodes dynamically -- register new nodes with the master while
+   a model is already placed, verify topology grows.
+2. Remove nodes -- simulate node timeout, verify instances on that node
+   are cleaned up and remaining nodes are unaffected.
+3. Rapid join/leave churn -- nodes join and leave quickly, verify state
+   converges to a consistent snapshot.
+"""
+
+from datetime import timedelta
+
+import anyio
+import pytest
+
+from exo.master.main import Master
+from exo.shared.types.commands import (
+    CommandId,
+    ForwarderCommand,
+    ForwarderDownloadCommand,
+    PlaceInstance,
+)
+from exo.shared.types.common import NodeId, SessionId
+from exo.shared.types.events import (
+    ForwarderEvent,
+)
+from exo.shared.types.worker.instances import InstanceMeta
+from exo.shared.types.worker.shards import Sharding
+from exo.utils.channels import channel
+
+from .conftest import (
+    TEST_MODEL_CARD,
+    make_gathered_info_event,
+    make_node_id,
+)
+
+
+@pytest.mark.slow
+@pytest.mark.asyncio
+async def test_dynamic_node_registration_expands_topology() -> None:
+    """Start with one node, then add more dynamically.  Verify the topology
+    grows and all nodes are visible in state.
+    """
+    master_nid = make_node_id("master-join")
+    session_id = SessionId(master_node_id=master_nid, election_clock=0)
+
+    ge_sender, _ge_receiver = channel[ForwarderEvent]()
+    cmd_sender, cmd_receiver = channel[ForwarderCommand]()
+    le_sender, le_receiver = channel[ForwarderEvent]()
+    dl_sender, _dl_receiver = channel[ForwarderDownloadCommand]()
+
+    master = Master(
+        master_nid,
+        session_id,
+        global_event_sender=ge_sender,
+        local_event_receiver=le_receiver,
+        command_receiver=cmd_receiver,
+        download_command_sender=dl_sender,
+    )
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(master.run)
+
+        # Register initial node
+        initial_node = make_node_id("initial")
+        sender_id = NodeId(f"{initial_node}_sender")
+        await le_sender.send(
+            make_gathered_info_event(initial_node, sender_id, session_id, 0)
+        )
+
+        with anyio.fail_after(3):
+            while len(list(master.state.topology.list_nodes())) < 1:
+                await anyio.sleep(0.01)
+
+        # Place a model instance
+        await cmd_sender.send(
+            ForwarderCommand(
+                origin=master_nid,
+                command=PlaceInstance(
+                    command_id=CommandId(),
+                    model_card=TEST_MODEL_CARD,
+                    sharding=Sharding.Pipeline,
+                    instance_meta=InstanceMeta.MlxRing,
+                    min_nodes=1,
+                ),
+            )
+        )
+
+        with anyio.fail_after(3):
+            while len(master.state.instances) == 0:
+                await anyio.sleep(0.01)
+
+        # Dynamically add 3 more nodes
+        new_nodes: list[NodeId] = []
+        for i in range(3):
+            new_nid = make_node_id(f"dynamic-{i}")
+            new_nodes.append(new_nid)
+            new_sender = NodeId(f"{new_nid}_sender")
+            await le_sender.send(
+                make_gathered_info_event(new_nid, new_sender, session_id, 0)
+            )
+
+        with anyio.fail_after(3):
+            while len(list(master.state.topology.list_nodes())) < 4:
+                await anyio.sleep(0.01)
+
+        # All 4 nodes should be in topology
+        topo_nodes = set(master.state.topology.list_nodes())
+        assert initial_node in topo_nodes
+        for nid in new_nodes:
+            assert nid in topo_nodes
+
+        # Original instance should still exist
+        assert len(master.state.instances) >= 1
+
+        await master.shutdown()
+
+
+@pytest.mark.slow
+@pytest.mark.asyncio
+async def test_node_removal_cleans_up_instances() -> None:
+    """Place a model on a specific node, then time it out.  Verify the
+    instance assigned to that node is deleted by the master's plan loop.
+    """
+    master_nid = make_node_id("master-leave")
+    session_id = SessionId(master_node_id=master_nid, election_clock=0)
+
+    ge_sender, _ge_receiver = channel[ForwarderEvent]()
+    cmd_sender, cmd_receiver = channel[ForwarderCommand]()
+    le_sender, le_receiver = channel[ForwarderEvent]()
+    dl_sender, _dl_receiver = channel[ForwarderDownloadCommand]()
+
+    master = Master(
+        master_nid,
+        session_id,
+        global_event_sender=ge_sender,
+        local_event_receiver=le_receiver,
+        command_receiver=cmd_receiver,
+        download_command_sender=dl_sender,
+    )
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(master.run)
+
+        # Register a worker node
+        worker_nid = make_node_id("worker-leaving")
+        sender_id = NodeId(f"{worker_nid}_sender")
+        await le_sender.send(
+            make_gathered_info_event(worker_nid, sender_id, session_id, 0)
+        )
+
+        with anyio.fail_after(3):
+            while len(list(master.state.topology.list_nodes())) < 1:
+                await anyio.sleep(0.01)
+
+        # Place instance on the worker node
+        await cmd_sender.send(
+            ForwarderCommand(
+                origin=master_nid,
+                command=PlaceInstance(
+                    command_id=CommandId(),
+                    model_card=TEST_MODEL_CARD,
+                    sharding=Sharding.Pipeline,
+                    instance_meta=InstanceMeta.MlxRing,
+                    min_nodes=1,
+                ),
+            )
+        )
+
+        with anyio.fail_after(3):
+            while len(master.state.instances) == 0:
+                await anyio.sleep(0.01)
+
+        assert len(master.state.instances) == 1
+
+        # Simulate node leaving by expiring its last_seen
+        old_time = master.state.last_seen[worker_nid] - timedelta(seconds=60)
+        patched_last_seen = {**master.state.last_seen, worker_nid: old_time}
+        master.state = master.state.model_copy(update={"last_seen": patched_last_seen})
+
+        # The plan loop should detect the stale node and delete the instance
+        # because the node assigned to the instance is no longer in the topology
+        with anyio.fail_after(15):
+            while worker_nid in master.state.last_seen:
+                await anyio.sleep(0.1)
+
+        # After timeout, the node should be removed from topology
+        assert worker_nid not in set(master.state.topology.list_nodes())
+
+        # The instance should eventually be deleted since the assigned node
+        # is no longer connected (the _plan loop kills broken instances)
+        with anyio.fail_after(15):
+            while len(master.state.instances) > 0:
+                await anyio.sleep(0.1)
+
+        assert len(master.state.instances) == 0
+
+        await master.shutdown()
+
+
+@pytest.mark.slow
+@pytest.mark.asyncio
+async def test_rapid_join_leave_churn_converges() -> None:
+    """Rapidly join and leave nodes.  After the churn settles, verify the
+    master's state reflects only the surviving nodes.
+    """
+    master_nid = make_node_id("master-churn")
+    session_id = SessionId(master_node_id=master_nid, election_clock=0)
+
+    ge_sender, _ge_receiver = channel[ForwarderEvent]()
+    _cmd_sender, cmd_receiver = channel[ForwarderCommand]()
+    le_sender, le_receiver = channel[ForwarderEvent]()
+    dl_sender, _dl_receiver = channel[ForwarderDownloadCommand]()
+
+    master = Master(
+        master_nid,
+        session_id,
+        global_event_sender=ge_sender,
+        local_event_receiver=le_receiver,
+        command_receiver=cmd_receiver,
+        download_command_sender=dl_sender,
+    )
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(master.run)
+
+        # Register 5 nodes rapidly
+        all_nodes: list[NodeId] = []
+        for i in range(5):
+            nid = make_node_id(f"churn-{i}")
+            all_nodes.append(nid)
+            sender_id = NodeId(f"{nid}_sender")
+            await le_sender.send(
+                make_gathered_info_event(nid, sender_id, session_id, 0)
+            )
+
+        with anyio.fail_after(5):
+            while len(list(master.state.topology.list_nodes())) < 5:
+                await anyio.sleep(0.01)
+
+        assert len(list(master.state.topology.list_nodes())) == 5
+
+        # Expire the first 3 nodes (simulate leaving)
+        leaving_nodes = all_nodes[:3]
+        surviving_nodes = all_nodes[3:]
+
+        patched_last_seen = dict(master.state.last_seen)
+        for nid in leaving_nodes:
+            patched_last_seen[nid] = patched_last_seen[nid] - timedelta(seconds=60)
+        master.state = master.state.model_copy(update={"last_seen": patched_last_seen})
+
+        # Wait for master's plan loop to time out the expired nodes
+        with anyio.fail_after(15):
+            while any(nid in master.state.last_seen for nid in leaving_nodes):
+                await anyio.sleep(0.1)
+
+        # Verify only surviving nodes remain
+        topo_nodes = set(master.state.topology.list_nodes())
+        for nid in leaving_nodes:
+            assert nid not in topo_nodes
+        for nid in surviving_nodes:
+            assert nid in topo_nodes
+
+        assert len(list(master.state.topology.list_nodes())) == 2
+
+        await master.shutdown()

src/exo/worker/runner/bootstrap.py

@@ -1,4 +1,8 @@
+from __future__ import annotations
+
 import os
+import threading
+from multiprocessing.sharedctypes import Synchronized
 
 import loguru
 
@@ -10,6 +14,15 @@ from exo.utils.channels import ClosedResourceError, MpReceiver, MpSender
 
 logger: "loguru.Logger" = loguru.logger
 
+HEARTBEAT_INTERVAL_SECONDS = 0.5
+
+
+def _heartbeat_loop(heartbeat: Synchronized[int], stop: threading.Event) -> None:
+    """Daemon thread that periodically increments the heartbeat counter."""
+    while not stop.is_set():
+        heartbeat.value += 1
+        stop.wait(HEARTBEAT_INTERVAL_SECONDS)
+
 
 def entrypoint(
     bound_instance: BoundInstance,
@@ -17,6 +30,7 @@ def entrypoint(
     task_receiver: MpReceiver[Task],
     cancel_receiver: MpReceiver[TaskId],
     _logger: "loguru.Logger",
+    heartbeat: Synchronized[int] | None = None,
 ) -> None:
     fast_synch_override = os.environ.get("EXO_FAST_SYNCH")
     if fast_synch_override == "on" or (
@@ -35,6 +49,17 @@ def entrypoint(
 
     logger.info(f"Fast synch flag: {os.environ['MLX_METAL_FAST_SYNCH']}")
 
+    # Start heartbeat thread so the supervisor can detect if we freeze.
+    stop_heartbeat = threading.Event()
+    heartbeat_thread: threading.Thread | None = None
+    if heartbeat is not None:
+        heartbeat_thread = threading.Thread(
+            target=_heartbeat_loop,
+            args=(heartbeat, stop_heartbeat),
+            daemon=True,
+        )
+        heartbeat_thread.start()
+
     # Import main after setting global logger - this lets us just import logger from this module
     try:
         from exo.worker.runner.runner import main
@@ -53,6 +78,9 @@ def entrypoint(
             )
         )
     finally:
+        stop_heartbeat.set()
+        if heartbeat_thread is not None:
+            heartbeat_thread.join(timeout=1)
         try:
             event_sender.close()
             task_receiver.close()

src/exo/worker/runner/runner.py

@@ -1,6 +1,7 @@
 import base64
 import json
 import math
+import os
 import resource
 import time
 from collections.abc import Generator
@@ -999,6 +1000,7 @@ def _validate_single_tool(obj: dict[str, Any]) -> ToolCallItem:
 EXO_RUNNER_MUST_FAIL = "EXO RUNNER MUST FAIL"
 EXO_RUNNER_MUST_OOM = "EXO RUNNER MUST OOM"
 EXO_RUNNER_MUST_TIMEOUT = "EXO RUNNER MUST TIMEOUT"
+EXO_RUNNER_MUST_DIE = "EXO RUNNER MUST DIE"
 
 
 def _check_for_debug_prompts(task_params: TextGenerationTaskParams) -> None:
@@ -1014,6 +1016,9 @@ def _check_for_debug_prompts(task_params: TextGenerationTaskParams) -> None:
     if not prompt:
         return
 
+    if EXO_RUNNER_MUST_DIE in prompt:
+        logger.info("Abrupt process death triggered (simulates OOM kill)")
+        os._exit(1)
     if EXO_RUNNER_MUST_FAIL in prompt:
         logger.info("raising exception")
         raise Exception("Artificial runner exception - for testing purposes only.")

src/exo/worker/runner/runner_supervisor.py

@@ -1,12 +1,17 @@
+from __future__ import annotations
+
 import contextlib
+import multiprocessing
 import signal
 from dataclasses import dataclass, field
 from multiprocessing import Process
+from multiprocessing.sharedctypes import Synchronized
 from typing import Self
 
 import anyio
 from anyio import (
     BrokenResourceError,
+    CancelScope,
     ClosedResourceError,
     to_thread,
 )
@@ -26,6 +31,7 @@ from exo.shared.types.worker.runners import (
     RunnerIdle,
     RunnerLoading,
     RunnerRunning,
+    RunnerShutdown,
     RunnerShuttingDown,
     RunnerStatus,
     RunnerWarmingUp,
@@ -36,6 +42,8 @@ from exo.worker.runner.bootstrap import entrypoint
 
 PREFILL_TIMEOUT_SECONDS = 60
 DECODE_TIMEOUT_SECONDS = 5
+HEALTH_CHECK_INTERVAL_SECONDS = 1
+HEARTBEAT_STALE_CHECKS = 10
 
 
 @dataclass(eq=False)
@@ -48,10 +56,14 @@ class RunnerSupervisor:
     _task_sender: MpSender[Task]
     _event_sender: Sender[Event]
     _cancel_sender: MpSender[TaskId]
+    _heartbeat: Synchronized[int]
     status: RunnerStatus = field(default_factory=RunnerIdle, init=False)
     pending: dict[TaskId, anyio.Event] = field(default_factory=dict, init=False)
     completed: set[TaskId] = field(default_factory=set, init=False)
     cancelled: set[TaskId] = field(default_factory=set, init=False)
+    _death_handled: bool = field(default=False, init=False)
+    _last_heartbeat_value: int = field(default=0, init=False)
+    _heartbeat_stale_count: int = field(default=0, init=False)
 
     @classmethod
     def create(
@@ -65,6 +77,8 @@ class RunnerSupervisor:
         task_sender, task_recv = mp_channel[Task]()
         cancel_sender, cancel_recv = mp_channel[TaskId]()
 
+        heartbeat: Synchronized[int] = multiprocessing.Value("Q", 0)
+
         runner_process = Process(
             target=entrypoint,
             args=(
@@ -73,6 +87,7 @@ class RunnerSupervisor:
                 task_recv,
                 cancel_recv,
                 logger,
+                heartbeat,
             ),
             daemon=True,
         )
@@ -88,13 +103,16 @@ class RunnerSupervisor:
             _task_sender=task_sender,
             _cancel_sender=cancel_sender,
             _event_sender=event_sender,
+            _heartbeat=heartbeat,
         )
 
         return self
 
     async def run(self):
         self.runner_process.start()
-        await self._forward_events()
+        async with anyio.create_task_group() as tg:
+            tg.start_soon(self._forward_events)
+            tg.start_soon(self._health_check, tg.cancel_scope)
 
     def shutdown(self):
         logger.info("Runner supervisor shutting down")
@@ -177,9 +195,99 @@ class RunnerSupervisor:
                         self.completed.add(event.task_id)
                     await self._event_sender.send(event)
             except (ClosedResourceError, BrokenResourceError) as e:
-                await self._check_runner(e)
-                for tid in self.pending:
-                    self.pending[tid].set()
+                if not self._death_handled:
+                    self._death_handled = True
+                    await self._check_runner(e)
+                    for tid in self.pending:
+                        self.pending[tid].set()
+
+    async def _health_check(self, cancel_scope: CancelScope) -> None:
+        """Periodically check if the runner process is alive and responsive.
+
+        Detects two failure modes:
+        1. Process death (e.g. OOM kill) without cleanly closing the event
+           channel, which would leave _forward_events blocked on queue.get().
+        2. Unresponsive process (e.g. frozen by OS memory pressure, deadlock)
+           detected via a stale heartbeat counter.
+        """
+        while True:
+            await anyio.sleep(HEALTH_CHECK_INTERVAL_SECONDS)
+
+            if not self.runner_process.is_alive():
+                self._handle_process_exit(cancel_scope)
+                return
+
+            # Check heartbeat counter — if it hasn't changed between
+            # consecutive checks, the subprocess may be frozen.
+            current = self._heartbeat.value
+            if current > 0:
+                if current == self._last_heartbeat_value:
+                    self._heartbeat_stale_count += 1
+                    if self._heartbeat_stale_count >= HEARTBEAT_STALE_CHECKS:
+                        logger.error(
+                            f"Health check: runner process unresponsive "
+                            f"(heartbeat stale for {self._heartbeat_stale_count} checks), killing"
+                        )
+                        self._handle_unresponsive(cancel_scope)
+                        return
+                else:
+                    self._heartbeat_stale_count = 0
+                self._last_heartbeat_value = current
+
+    def _handle_process_exit(self, cancel_scope: CancelScope) -> None:
+        """Handle runner process that has exited."""
+        if not self._death_handled:
+            self._death_handled = True
+            if isinstance(
+                self.status, (RunnerShutdown, RunnerShuttingDown, RunnerFailed)
+            ):
+                logger.info("Health check: runner process exited (expected)")
+            else:
+                rc = self.runner_process.exitcode
+                if isinstance(rc, int) and rc < 0:
+                    sig = -rc
+                    try:
+                        cause = f"signal={sig} ({signal.strsignal(sig)})"
+                    except Exception:
+                        cause = f"signal={sig}"
+                else:
+                    cause = f"exitcode={rc}"
+
+                logger.error(
+                    f"Health check: runner process died unexpectedly ({cause})"
+                )
+                self._event_sender.send_nowait(
+                    RunnerStatusUpdated(
+                        runner_id=self.bound_instance.bound_runner_id,
+                        runner_status=RunnerFailed(
+                            error_message=f"Terminated ({cause})"
+                        ),
+                    )
+                )
+                self.shutdown()
+
+            for tid in self.pending:
+                self.pending[tid].set()
+
+        cancel_scope.cancel()
+
+    def _handle_unresponsive(self, cancel_scope: CancelScope) -> None:
+        """Handle runner process that is alive but unresponsive."""
+        if not self._death_handled:
+            self._death_handled = True
+            self._event_sender.send_nowait(
+                RunnerStatusUpdated(
+                    runner_id=self.bound_instance.bound_runner_id,
+                    runner_status=RunnerFailed(
+                        error_message="Runner process unresponsive (heartbeat timeout)"
+                    ),
+                )
+            )
+            for tid in self.pending:
+                self.pending[tid].set()
+            self.shutdown()
+
+        cancel_scope.cancel()
 
     def __del__(self) -> None:
         if self.runner_process.is_alive():

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

@@ -1 +1,204 @@
-# TODO:
+from __future__ import annotations
+
+import multiprocessing
+import os
+import signal as signal_module
+from collections.abc import Callable
+from multiprocessing.sharedctypes import Synchronized
+from typing import Any
+
+import anyio
+
+from exo.shared.types.events import Event, RunnerStatusUpdated
+from exo.shared.types.tasks import Task, TaskId
+from exo.shared.types.worker.runners import RunnerFailed, RunnerIdle, RunnerShutdown
+from exo.utils.channels import Receiver, Sender, channel, mp_channel
+from exo.worker.runner.runner_supervisor import (
+    HEALTH_CHECK_INTERVAL_SECONDS,
+    HEARTBEAT_STALE_CHECKS,
+    RunnerSupervisor,
+)
+
+from ...constants import (
+    INSTANCE_1_ID,
+    MODEL_A_ID,
+    NODE_A,
+    RUNNER_1_ID,
+)
+from ..conftest import get_bound_mlx_ring_instance
+
+
+def _die_immediately() -> None:
+    """Subprocess target that exits with a non-zero code."""
+    os._exit(1)
+
+
+def _die_with_signal() -> None:
+    """Subprocess target that kills itself with SIGKILL (simulates OOM)."""
+    os.kill(os.getpid(), signal_module.SIGKILL)
+
+
+def _exit_cleanly() -> None:
+    """Subprocess target that exits with code 0."""
+    os._exit(0)
+
+
+def _hang_forever() -> None:
+    """Subprocess target that hangs without updating heartbeat (simulates freeze)."""
+    import time
+
+    # Write one heartbeat so the supervisor starts tracking, then stop.
+    time.sleep(100000)
+
+
+def _build_supervisor(
+    event_sender: Sender[Event],
+    target: Callable[..., Any],
+) -> RunnerSupervisor:
+    """Build a RunnerSupervisor with a custom subprocess target.
+
+    Uses a clone of event_sender (matching real Worker behavior) so that
+    closing the supervisor's copy doesn't close the test's receiver.
+    """
+    bound_instance = get_bound_mlx_ring_instance(
+        instance_id=INSTANCE_1_ID,
+        model_id=MODEL_A_ID,
+        runner_id=RUNNER_1_ID,
+        node_id=NODE_A,
+    )
+
+    _ev_send, ev_recv = mp_channel[Event]()
+    task_sender, _task_recv = mp_channel[Task]()
+    cancel_sender, _cancel_recv = mp_channel[TaskId]()
+    runner_process = multiprocessing.Process(target=target, daemon=True)
+    heartbeat: Synchronized[int] = multiprocessing.Value("Q", 0)
+
+    return RunnerSupervisor(
+        bound_instance=bound_instance,
+        shard_metadata=bound_instance.bound_shard,
+        runner_process=runner_process,
+        initialize_timeout=10,
+        _ev_recv=ev_recv,
+        _task_sender=task_sender,
+        _cancel_sender=cancel_sender,
+        _event_sender=event_sender.clone(),
+        _heartbeat=heartbeat,
+    )
+
+
+def _collect_failed_events(
+    event_receiver: Receiver[Event],
+) -> list[RunnerFailed]:
+    """Drain the receiver and return all RunnerFailed statuses."""
+    out: list[RunnerFailed] = []
+    while True:
+        try:
+            event = event_receiver.receive_nowait()
+        except Exception:
+            break
+        if isinstance(event, RunnerStatusUpdated) and isinstance(
+            event.runner_status, RunnerFailed
+        ):
+            out.append(event.runner_status)
+    return out
+
+
+async def test_health_check_detects_dead_process():
+    """When the runner process dies with a non-zero exit code, the health check
+    should emit a RunnerFailed event and run() should return."""
+    event_sender, event_receiver = channel[Event]()
+    supervisor = _build_supervisor(event_sender, _die_immediately)
+
+    with anyio.fail_after(HEALTH_CHECK_INTERVAL_SECONDS + 5):
+        await supervisor.run()
+
+    failures = _collect_failed_events(event_receiver)
+    assert len(failures) == 1
+    assert failures[0].error_message is not None
+    assert "exitcode=1" in failures[0].error_message
+
+
+async def test_health_check_detects_signal_death():
+    """When the runner process is killed by a signal (e.g. OOM -> SIGKILL),
+    the health check should report the signal in the failure message."""
+    event_sender, event_receiver = channel[Event]()
+    supervisor = _build_supervisor(event_sender, _die_with_signal)
+
+    with anyio.fail_after(HEALTH_CHECK_INTERVAL_SECONDS + 5):
+        await supervisor.run()
+
+    failures = _collect_failed_events(event_receiver)
+    assert len(failures) == 1
+    assert failures[0].error_message is not None
+    assert "signal=9" in failures[0].error_message
+
+
+async def test_health_check_releases_pending_tasks():
+    """When the runner dies, any pending start_task() waiters should be unblocked."""
+    event_sender, _event_receiver = channel[Event]()
+    supervisor = _build_supervisor(event_sender, _die_immediately)
+
+    # Register a pending waiter as if start_task() was waiting for acknowledgement
+    task_event = anyio.Event()
+    tid = TaskId("pending-task")
+    supervisor.pending[tid] = task_event
+
+    with anyio.fail_after(HEALTH_CHECK_INTERVAL_SECONDS + 5):
+        await supervisor.run()
+
+    assert task_event.is_set()
+
+
+async def test_clean_exit_no_failure_when_shutdown_status():
+    """When the runner was in RunnerShutdown status and exits with code 0,
+    no RunnerFailed event should be emitted."""
+    event_sender, event_receiver = channel[Event]()
+    supervisor = _build_supervisor(event_sender, _exit_cleanly)
+
+    # Simulate that the runner had already reported shutdown via events
+    supervisor.status = RunnerShutdown()
+
+    with anyio.fail_after(HEALTH_CHECK_INTERVAL_SECONDS + 5):
+        await supervisor.run()
+
+    failures = _collect_failed_events(event_receiver)
+    assert len(failures) == 0
+
+
+async def test_unexpected_exit_code_zero_emits_failure():
+    """When the runner exits with code 0 but was NOT in a shutdown state,
+    this is unexpected and should still emit RunnerFailed."""
+    event_sender, event_receiver = channel[Event]()
+    supervisor = _build_supervisor(event_sender, _exit_cleanly)
+
+    assert isinstance(supervisor.status, RunnerIdle)
+
+    with anyio.fail_after(HEALTH_CHECK_INTERVAL_SECONDS + 5):
+        await supervisor.run()
+
+    failures = _collect_failed_events(event_receiver)
+    assert len(failures) == 1
+    assert failures[0].error_message is not None
+    assert "exitcode=0" in failures[0].error_message
+
+
+async def test_heartbeat_timeout_detects_unresponsive_process():
+    """When the runner process is alive but its heartbeat goes stale,
+    the health check should kill it and emit RunnerFailed."""
+    event_sender, event_receiver = channel[Event]()
+    supervisor = _build_supervisor(event_sender, _hang_forever)
+
+    # Pre-seed the heartbeat counter with a non-zero value and set the
+    # supervisor's last-seen value to match so it appears stale immediately.
+    # Set stale count to HEARTBEAT_STALE_CHECKS - 1 so a single check triggers.
+    supervisor._heartbeat.value = 42  # pyright: ignore[reportPrivateUsage]
+    supervisor._last_heartbeat_value = 42  # pyright: ignore[reportPrivateUsage]
+    supervisor._heartbeat_stale_count = HEARTBEAT_STALE_CHECKS - 1  # pyright: ignore[reportPrivateUsage]
+
+    with anyio.fail_after(HEALTH_CHECK_INTERVAL_SECONDS + 5):
+        await supervisor.run()
+
+    failures = _collect_failed_events(event_receiver)
+    assert len(failures) == 1
+    assert failures[0].error_message is not None
+    assert "unresponsive" in failures[0].error_message.lower()