diff --git a/README.md b/README.md
index cdb7c5f063..2e84ac43a5 100644
--- a/README.md
+++ b/README.md
@@ -538,6 +538,13 @@ The `exo-bench` tool measures model prefill and token generation speed across di
 - Nodes should be running with `uv run exo` before benchmarking
 - The tool uses the `/bench/chat/completions` endpoint
 
+For Ring Attention on Metal, start every node with MLX's low-latency CPU/GPU
+synchronization enabled:
+
+```bash
+MLX_METAL_FAST_SYNCH=1 uv run exo
+```
+
 **Basic usage:**
 
 ```bash
@@ -554,7 +561,7 @@ uv run bench/exo_bench.py \
 - `--tg`: Generation lengths (comma-separated integers)
 - `--max-nodes`: Limit placements to N nodes (default: 4)
 - `--instance-meta`: Filter by `ring`, `jaccl`, or `both` (default: both)
-- `--sharding`: Filter by `pipeline`, `tensor`, or `both` (default: both)
+- `--sharding`: Filter by `pipeline`, `tensor`, `ring`, or `both` (default: both; `both` retains the pipeline/tensor comparison)
 - `--repeat`: Number of repetitions per configuration (default: 1)
 - `--warmup`: Warmup runs per placement (default: 0)
 - `--json-out`: Output file for results (default: bench/results.json)
@@ -572,6 +579,29 @@ uv run bench/exo_bench.py \
   --json-out my-results.json
 ```
 
+To compare Ring Attention against replicated pipeline prefill on the same
+Ring-compatible model and node count, run both placements with identical prompt
+lengths, generation lengths, repetitions, and warmups:
+
+```bash
+uv run bench/exo_bench.py \
+  --model Llama-3.2-1B-Instruct-4bit \
+  --pp 4096,16384,65536 --tg 128,128,128 \
+  --min-nodes 2 --max-nodes 2 --instance-meta ring \
+  --sharding ring --warmup 1 --repeat 3 \
+  --json-out bench/ring-attention.json
+
+uv run bench/exo_bench.py \
+  --model Llama-3.2-1B-Instruct-4bit \
+  --pp 4096,16384,65536 --tg 128,128,128 \
+  --min-nodes 2 --max-nodes 2 --instance-meta ring \
+  --sharding pipeline --warmup 1 --repeat 3 \
+  --json-out bench/replicated-pipeline.json
+```
+
+Use `prompt_tps` as the primary Ring prefill metric. Keep prefix caching disabled
+(the default), and record node hardware and network topology alongside results.
+
 The tool outputs performance metrics including prompt tokens per second (prompt_tps), generation tokens per second (generation_tps), and peak memory usage for each configuration.
 
 ---
diff --git a/dashboard/src/lib/components/ModelCard.svelte b/dashboard/src/lib/components/ModelCard.svelte
index 842871587b..d5b10a93c8 100644
--- a/dashboard/src/lib/components/ModelCard.svelte
+++ b/dashboard/src/lib/components/ModelCard.svelte
@@ -22,7 +22,7 @@
       }>;
     } | null;
     nodes?: Record<string, NodeInfo>;
-    sharding?: "Pipeline" | "Tensor";
+    sharding?: "Pipeline" | "Tensor" | "Ring";
     runtime?: "MlxRing" | "MlxJaccl";
     onLaunch?: () => void;
     tags?: string[];
diff --git a/dashboard/src/lib/stores/app.svelte.ts b/dashboard/src/lib/stores/app.svelte.ts
index a717c3afbe..cb45cd72a1 100644
--- a/dashboard/src/lib/stores/app.svelte.ts
+++ b/dashboard/src/lib/stores/app.svelte.ts
@@ -173,7 +173,7 @@ export interface ModelDownloadStatus {
 // Placement preview from the API
 export interface PlacementPreview {
   model_id: string;
-  sharding: "Pipeline" | "Tensor";
+  sharding: "Pipeline" | "Tensor" | "Ring";
   instance_meta: "MlxRing" | "MlxJaccl";
   instance: unknown | null;
   memory_delta_by_node: Record<string, number> | null;
diff --git a/dashboard/src/routes/+page.svelte b/dashboard/src/routes/+page.svelte
index 5db620795e..f5f577ae2f 100644
--- a/dashboard/src/routes/+page.svelte
+++ b/dashboard/src/routes/+page.svelte
@@ -783,6 +783,7 @@
       quantization?: string;
       base_model?: string;
       capabilities?: string[];
+      supports_ring?: boolean;
     }>
   >([]);
   type ModelMemoryFitStatus =
@@ -885,14 +886,14 @@
     sendMessage(content, files, thinkingEnabled());
   }
 
-  let selectedSharding = $state<"Pipeline" | "Tensor">("Pipeline");
+  let selectedSharding = $state<"Pipeline" | "Tensor" | "Ring">("Pipeline");
   type InstanceMeta = "MlxRing" | "MlxJaccl";
 
   // Launch defaults persistence
   const LAUNCH_DEFAULTS_KEY = "exo-launch-defaults-v2";
   interface LaunchDefaults {
     modelId: string | null;
-    sharding: "Pipeline" | "Tensor";
+    sharding: "Pipeline" | "Tensor" | "Ring";
     instanceType: InstanceMeta;
     minNodes: number;
   }
@@ -932,7 +933,9 @@
     // Apply sharding and instance type unconditionally
     selectedSharding = defaults.sharding;
     selectedInstanceType =
-      defaults.instanceType === "MlxRing" ? "MlxRing" : "MlxJaccl";
+      defaults.sharding === "Ring" || defaults.instanceType === "MlxRing"
+        ? "MlxRing"
+        : "MlxJaccl";
 
     // Apply minNodes if valid (between 1 and maxNodes)
     if (
@@ -954,6 +957,20 @@
   }
 
   let selectedInstanceType = $state<InstanceMeta>("MlxRing");
+
+  const selectedModelSupportsRing = $derived(
+    models.find(
+      (model) =>
+        model.id === selectedModelId ||
+        model.hugging_face_id === selectedModelId,
+    )?.supports_ring === true,
+  );
+
+  $effect(() => {
+    if (selectedSharding === "Ring" && !selectedModelSupportsRing) {
+      selectedSharding = "Pipeline";
+    }
+  });
   let selectedMinNodes = $state<number>(1);
   let minNodesInitialized = $state(false);
   let launchingModelId = $state<string | null>(null);
@@ -2043,7 +2060,7 @@
     return inst.shardAssignments?.modelId || "Unknown Model";
   }
 
-  // Get instance details: type (MLX Ring/IBV), sharding (Pipeline/Tensor), and node names
+  // Get instance details: type (MLX Ring/IBV), sharding strategy, and node names
   function getInstanceInfo(instanceWrapped: unknown): {
     instanceType: string;
     sharding: string;
@@ -2082,6 +2099,7 @@
       const [shardTag] = getTagged(firstShardWrapped);
       if (shardTag === "PipelineShardMetadata") sharding = "Pipeline";
       else if (shardTag === "TensorShardMetadata") sharding = "Tensor";
+      else if (shardTag === "RingShardMetadata") sharding = "Ring";
       else if (shardTag === "PrefillDecodeShardMetadata")
         sharding = "Prefill/Decode";
     }
@@ -2193,7 +2211,7 @@
 
   function getOrderedRunnerNodes(
     instance: Record<string, unknown>,
-    shardType: "Pipeline" | "Tensor",
+    shardType: "Pipeline" | "Tensor" | "Ring",
   ) {
     const runnerToShard =
       (
@@ -5772,6 +5790,37 @@
                         </span>
                         Tensor
                       </button>
+                      <button
+                        disabled={!selectedModelSupportsRing}
+                        title={selectedModelSupportsRing
+                          ? "Sequence-parallel Ring Attention"
+                          : "Ring Attention is not verified for this model"}
+                        onclick={() => {
+                          if (!selectedModelSupportsRing) return;
+                          selectedSharding = "Ring";
+                          selectedInstanceType = "MlxRing";
+                          saveLaunchDefaults();
+                        }}
+                        class="flex items-center gap-2 py-1.5 px-3 text-xs font-mono border rounded transition-all duration-200 cursor-pointer {selectedSharding ===
+                        'Ring'
+                          ? 'bg-transparent text-exo-yellow border-exo-yellow'
+                          : selectedModelSupportsRing
+                            ? 'bg-transparent text-white/70 border-exo-medium-gray/50 hover:border-exo-yellow/50'
+                            : 'bg-transparent text-white/30 border-exo-medium-gray/30 cursor-not-allowed'}"
+                      >
+                        <span
+                          class="w-3 h-3 rounded-full border-2 flex items-center justify-center {selectedSharding ===
+                          'Ring'
+                            ? 'border-exo-yellow'
+                            : 'border-exo-medium-gray'}"
+                        >
+                          {#if selectedSharding === "Ring"}
+                            <span class="w-1.5 h-1.5 rounded-full bg-exo-yellow"
+                            ></span>
+                          {/if}
+                        </span>
+                        Ring Attention
+                      </button>
                     </div>
                   </div>
 
@@ -5807,6 +5856,9 @@
                       <button
                         onclick={() => {
                           selectedInstanceType = "MlxJaccl";
+                          if (selectedSharding === "Ring") {
+                            selectedSharding = "Pipeline";
+                          }
                           saveLaunchDefaults();
                         }}
                         class="flex items-center gap-2 py-1.5 px-3 text-xs font-mono border rounded transition-all duration-200 cursor-pointer {selectedInstanceType ===
diff --git a/resources/inference_model_cards/mlx-community--Llama-3.2-1B-Instruct-4bit.toml b/resources/inference_model_cards/mlx-community--Llama-3.2-1B-Instruct-4bit.toml
index 4a70c63a35..c88b649198 100644
--- a/resources/inference_model_cards/mlx-community--Llama-3.2-1B-Instruct-4bit.toml
+++ b/resources/inference_model_cards/mlx-community--Llama-3.2-1B-Instruct-4bit.toml
@@ -3,6 +3,7 @@ n_layers = 16
 hidden_size = 2048
 num_key_value_heads = 8
 supports_tensor = true
+supports_ring = true
 tasks = ["TextGeneration"]
 family = "llama"
 quantization = "4bit"
diff --git a/resources/inference_model_cards/mlx-community--Llama-3.2-3B-Instruct-4bit.toml b/resources/inference_model_cards/mlx-community--Llama-3.2-3B-Instruct-4bit.toml
index 9e21b13126..f280c03e15 100644
--- a/resources/inference_model_cards/mlx-community--Llama-3.2-3B-Instruct-4bit.toml
+++ b/resources/inference_model_cards/mlx-community--Llama-3.2-3B-Instruct-4bit.toml
@@ -3,6 +3,7 @@ n_layers = 28
 hidden_size = 3072
 num_key_value_heads = 8
 supports_tensor = true
+supports_ring = true
 tasks = ["TextGeneration"]
 family = "llama"
 quantization = "4bit"
diff --git a/resources/inference_model_cards/mlx-community--Llama-3.2-3B-Instruct-8bit.toml b/resources/inference_model_cards/mlx-community--Llama-3.2-3B-Instruct-8bit.toml
index b3deee1bb3..80e7f0d927 100644
--- a/resources/inference_model_cards/mlx-community--Llama-3.2-3B-Instruct-8bit.toml
+++ b/resources/inference_model_cards/mlx-community--Llama-3.2-3B-Instruct-8bit.toml
@@ -3,6 +3,7 @@ n_layers = 28
 hidden_size = 3072
 num_key_value_heads = 8
 supports_tensor = true
+supports_ring = true
 tasks = ["TextGeneration"]
 family = "llama"
 quantization = "8bit"
diff --git a/resources/inference_model_cards/mlx-community--Llama-3.3-70B-Instruct-4bit.toml b/resources/inference_model_cards/mlx-community--Llama-3.3-70B-Instruct-4bit.toml
index cd2d960071..e3cfc5c095 100644
--- a/resources/inference_model_cards/mlx-community--Llama-3.3-70B-Instruct-4bit.toml
+++ b/resources/inference_model_cards/mlx-community--Llama-3.3-70B-Instruct-4bit.toml
@@ -3,6 +3,7 @@ n_layers = 80
 hidden_size = 8192
 num_key_value_heads = 8
 supports_tensor = true
+supports_ring = true
 tasks = ["TextGeneration"]
 family = "llama"
 quantization = "4bit"
diff --git a/resources/inference_model_cards/mlx-community--Llama-3.3-70B-Instruct-8bit.toml b/resources/inference_model_cards/mlx-community--Llama-3.3-70B-Instruct-8bit.toml
index 0ffae4af12..fb3015f4c1 100644
--- a/resources/inference_model_cards/mlx-community--Llama-3.3-70B-Instruct-8bit.toml
+++ b/resources/inference_model_cards/mlx-community--Llama-3.3-70B-Instruct-8bit.toml
@@ -3,6 +3,7 @@ n_layers = 80
 hidden_size = 8192
 num_key_value_heads = 8
 supports_tensor = true
+supports_ring = true
 tasks = ["TextGeneration"]
 family = "llama"
 quantization = "8bit"
diff --git a/resources/inference_model_cards/mlx-community--Meta-Llama-3.1-70B-Instruct-4bit.toml b/resources/inference_model_cards/mlx-community--Meta-Llama-3.1-70B-Instruct-4bit.toml
index 5a2d14db5e..5dc3d41ffc 100644
--- a/resources/inference_model_cards/mlx-community--Meta-Llama-3.1-70B-Instruct-4bit.toml
+++ b/resources/inference_model_cards/mlx-community--Meta-Llama-3.1-70B-Instruct-4bit.toml
@@ -3,6 +3,7 @@ n_layers = 80
 hidden_size = 8192
 num_key_value_heads = 8
 supports_tensor = true
+supports_ring = true
 tasks = ["TextGeneration"]
 family = "llama"
 quantization = "4bit"
diff --git a/resources/inference_model_cards/mlx-community--Meta-Llama-3.1-8B-Instruct-4bit.toml b/resources/inference_model_cards/mlx-community--Meta-Llama-3.1-8B-Instruct-4bit.toml
index cfa7e57b7a..a58a518bc3 100644
--- a/resources/inference_model_cards/mlx-community--Meta-Llama-3.1-8B-Instruct-4bit.toml
+++ b/resources/inference_model_cards/mlx-community--Meta-Llama-3.1-8B-Instruct-4bit.toml
@@ -3,6 +3,7 @@ n_layers = 32
 hidden_size = 4096
 num_key_value_heads = 8
 supports_tensor = true
+supports_ring = true
 tasks = ["TextGeneration"]
 family = "llama"
 quantization = "4bit"
diff --git a/resources/inference_model_cards/mlx-community--Meta-Llama-3.1-8B-Instruct-8bit.toml b/resources/inference_model_cards/mlx-community--Meta-Llama-3.1-8B-Instruct-8bit.toml
index eb2163a94f..d40a194a78 100644
--- a/resources/inference_model_cards/mlx-community--Meta-Llama-3.1-8B-Instruct-8bit.toml
+++ b/resources/inference_model_cards/mlx-community--Meta-Llama-3.1-8B-Instruct-8bit.toml
@@ -3,6 +3,7 @@ n_layers = 32
 hidden_size = 4096
 num_key_value_heads = 8
 supports_tensor = true
+supports_ring = true
 tasks = ["TextGeneration"]
 family = "llama"
 quantization = "8bit"
diff --git a/resources/inference_model_cards/mlx-community--Meta-Llama-3.1-8B-Instruct-bf16.toml b/resources/inference_model_cards/mlx-community--Meta-Llama-3.1-8B-Instruct-bf16.toml
index 4caefc1493..8b570758c4 100644
--- a/resources/inference_model_cards/mlx-community--Meta-Llama-3.1-8B-Instruct-bf16.toml
+++ b/resources/inference_model_cards/mlx-community--Meta-Llama-3.1-8B-Instruct-bf16.toml
@@ -3,6 +3,7 @@ n_layers = 32
 hidden_size = 4096
 num_key_value_heads = 8
 supports_tensor = true
+supports_ring = true
 tasks = ["TextGeneration"]
 family = "llama"
 quantization = "bf16"
diff --git a/resources/inference_model_cards/mlx-community--llama-3.3-70b-instruct-fp16.toml b/resources/inference_model_cards/mlx-community--llama-3.3-70b-instruct-fp16.toml
index f37400aa91..c301b5239e 100644
--- a/resources/inference_model_cards/mlx-community--llama-3.3-70b-instruct-fp16.toml
+++ b/resources/inference_model_cards/mlx-community--llama-3.3-70b-instruct-fp16.toml
@@ -3,6 +3,7 @@ n_layers = 80
 hidden_size = 8192
 num_key_value_heads = 8
 supports_tensor = true
+supports_ring = true
 tasks = ["TextGeneration"]
 family = "llama"
 quantization = "fp16"
diff --git a/rust/exo_rs/tests/test_python.py b/rust/exo_rs/tests/test_python.py
index fd6976c3db..d7800996be 100644
--- a/rust/exo_rs/tests/test_python.py
+++ b/rust/exo_rs/tests/test_python.py
@@ -13,7 +13,9 @@
 @pytest.mark.asyncio
 async def test_sleep_on_multiple_items() -> None:
     print("PYTHON: starting handle")
-    h = NetworkingHandle.new(os.urandom(16).hex().lstrip("0"), 52414, 52413)
+    h = NetworkingHandle.new(
+        os.urandom(16).hex().lstrip("0"), "exo-test", 52414, 52413
+    )
     print("PYTHON: handle started")
 
     rt = asyncio.create_task(_await_recv(h))
diff --git a/src/exo/api/main.py b/src/exo/api/main.py
index fcd54c9315..853d88dc62 100644
--- a/src/exo/api/main.py
+++ b/src/exo/api/main.py
@@ -535,6 +535,11 @@ async def get_placement_previews(
                         )
                     ]
                 )
+        if model_card.supports_ring:
+            instance_combinations.extend(
+                (Sharding.Ring, InstanceMeta.MlxRing, i)
+                for i in range(2, len(list(self.state.topology.list_nodes())) + 1)
+            )
         # TODO: PDD
         # instance_combinations.append((Sharding.PrefillDecodeDisaggregation, InstanceMeta.MlxRing, 1))
 
@@ -1805,6 +1810,7 @@ async def get_models(self, status: str | None = Query(default=None)) -> ModelLis
                     tags=[],
                     storage_size_megabytes=card.storage_size.in_mb,
                     supports_tensor=card.supports_tensor,
+                    supports_ring=card.supports_ring,
                     tasks=[task.value for task in card.tasks],
                     is_custom=card.is_custom,
                     family=card.family,
@@ -1846,6 +1852,7 @@ async def add_custom_model(self, payload: AddCustomModelParams) -> ModelListMode
             tags=[],
             storage_size_megabytes=int(card.storage_size.in_mb),
             supports_tensor=card.supports_tensor,
+            supports_ring=card.supports_ring,
             tasks=[task.value for task in card.tasks],
             is_custom=True,
         )
diff --git a/src/exo/api/types/api.py b/src/exo/api/types/api.py
index 02e397adaa..05298aeb81 100644
--- a/src/exo/api/types/api.py
+++ b/src/exo/api/types/api.py
@@ -42,6 +42,7 @@ class ModelListModel(BaseModel):
     tags: list[str] = Field(default=[])
     storage_size_megabytes: int = Field(default=0)
     supports_tensor: bool = Field(default=False)
+    supports_ring: bool = Field(default=False)
     tasks: list[str] = Field(default=[])
     is_custom: bool = Field(default=False)
     family: str = Field(default="")
diff --git a/src/exo/download/__init__.py b/src/exo/download/__init__.py
new file mode 100644
index 0000000000..bf77a4c7d3
--- /dev/null
+++ b/src/exo/download/__init__.py
@@ -0,0 +1 @@
+"""Model download orchestration and storage helpers."""
diff --git a/src/exo/master/placement.py b/src/exo/master/placement.py
index 0ed634e2ee..ad61e8a934 100644
--- a/src/exo/master/placement.py
+++ b/src/exo/master/placement.py
@@ -4,7 +4,9 @@
 
 from exo.master.placement_utils import (
     Cycle,
+    estimate_ring_node_memory,
     filter_cycles_by_memory,
+    filter_cycles_by_replicated_memory,
     get_mlx_jaccl_coordinators,
     get_mlx_jaccl_devices_matrix,
     get_mlx_ring_hosts_by_node,
@@ -114,6 +116,18 @@ def place_instance(
     download_status: Mapping[NodeId, Sequence[DownloadProgress]] | None = None,
     node_rdma_ctl: Mapping[NodeId, NodeRdmaCtlStatus] | None = None,
 ) -> dict[InstanceId, Instance]:
+    if (
+        command.sharding is Sharding.Ring
+        and command.instance_meta is not InstanceMeta.MlxRing
+    ):
+        raise ValueError("Ring attention requires the MlxRing transport")
+    if command.sharding is Sharding.Ring and command.min_nodes < 2:
+        raise ValueError("Ring attention requires at least two nodes")
+    if command.sharding is Sharding.Ring and not command.model_card.supports_ring:
+        raise ValueError(
+            f"Model does not declare Ring attention support: {command.model_card.model_id}"
+        )
+
     cycles = topology.get_cycles()
     candidate_cycles = list(filter(lambda it: len(it) >= command.min_nodes, cycles))
 
@@ -124,9 +138,18 @@ def place_instance(
             for cycle in candidate_cycles
             if required_nodes.issubset(cycle.node_ids)
         ]
-    cycles_with_sufficient_memory = filter_cycles_by_memory(
-        candidate_cycles, node_memory, command.model_card.storage_size
-    )
+    if command.sharding is Sharding.Ring:
+        # Every ring rank replicates the weights and must also hold the
+        # long-context prefill working set, not just the model file.
+        cycles_with_sufficient_memory = filter_cycles_by_replicated_memory(
+            candidate_cycles,
+            node_memory,
+            estimate_ring_node_memory(command.model_card),
+        )
+    else:
+        cycles_with_sufficient_memory = filter_cycles_by_memory(
+            candidate_cycles, node_memory, command.model_card.storage_size
+        )
     if len(cycles_with_sufficient_memory) == 0:
         raise ValueError("No cycles found with sufficient memory")
 
diff --git a/src/exo/master/placement_utils.py b/src/exo/master/placement_utils.py
index 0375e97e01..a3b7509849 100644
--- a/src/exo/master/placement_utils.py
+++ b/src/exo/master/placement_utils.py
@@ -12,6 +12,7 @@
 from exo.shared.types.worker.shards import (
     CfgShardMetadata,
     PipelineShardMetadata,
+    RingShardMetadata,
     Sharding,
     ShardMetadata,
     TensorShardMetadata,
@@ -29,7 +30,7 @@ def filter_cycles_by_memory(
             continue
 
         total_mem = sum(
-            (node_memory[node_id].ram_available for node_id in cycle.node_ids),
+            (node_memory[node_id].inference_available for node_id in cycle.node_ids),
             start=Memory(),
         )
         if total_mem >= required_memory:
@@ -37,6 +38,65 @@ def filter_cycles_by_memory(
     return filtered_cycles
 
 
+def filter_cycles_by_replicated_memory(
+    cycles: list[Cycle],
+    node_memory: Mapping[NodeId, MemoryUsage],
+    required_memory: Memory,
+) -> list[Cycle]:
+    """Keep cycles where every node can hold a fully replicated model."""
+    return [
+        cycle
+        for cycle in cycles
+        if all(
+            node_id in node_memory
+            and node_memory[node_id].inference_available >= required_memory
+            for node_id in cycle.node_ids
+        )
+    ]
+
+
+# K and V cache entries are stored unquantized at 2 bytes each during Ring
+# prefill; the working-set multiplier covers transient attention buffers and
+# allocator caching observed in practice (Metal peaked at ~3x the steady KV
+# size, CUDA at ~5x, for the same prefill).
+_KV_CACHE_BYTES_PER_ELEMENT = 2
+_RING_KV_WORKING_SET_MULTIPLIER = 4
+# Upper bound on head_dim across the verified Ring model families (Llama,
+# Qwen3), used when the card does not pin the exact geometry.
+_RING_HEAD_DIM_BOUND = 128
+# Ring exists for long-context prefill; admit against at least this context
+# even if requests may be shorter, and no more than this even for cards that
+# advertise 128K+ so admission stays achievable.
+_RING_ADMISSION_CONTEXT_TOKENS = 16384
+
+
+def estimate_ring_node_memory(model_card: ModelCard) -> Memory:
+    """Per-node memory a Ring rank needs: replicated weights plus the working
+    set of a long-context prefill (full KV cache and attention transients).
+
+    Weight-only admission demonstrably over-admits: a 4 GB-VRAM rank passed
+    the old check for a model whose 16K prefill peaks well past 4 GB.
+    """
+    admission_context = (
+        min(model_card.context_length, _RING_ADMISSION_CONTEXT_TOKENS)
+        if model_card.context_length > 0
+        else _RING_ADMISSION_CONTEXT_TOKENS
+    )
+    if model_card.num_key_value_heads is not None:
+        kv_width = model_card.num_key_value_heads * _RING_HEAD_DIM_BOUND
+    else:
+        kv_width = model_card.hidden_size
+    kv_cache_bytes = (
+        2  # keys and values
+        * _KV_CACHE_BYTES_PER_ELEMENT
+        * model_card.n_layers
+        * kv_width
+        * admission_context
+    )
+    working_set = Memory.from_bytes(kv_cache_bytes * _RING_KV_WORKING_SET_MULTIPLIER)
+    return model_card.storage_size + working_set
+
+
 def get_smallest_cycles(
     cycles: list[Cycle],
 ) -> list[Cycle]:
@@ -85,7 +145,7 @@ def _compute_total_memory(
     node_memory: Mapping[NodeId, MemoryUsage],
 ) -> Memory:
     total_memory = sum(
-        (node_memory[node_id].ram_available for node_id in node_ids),
+        (node_memory[node_id].inference_available for node_id in node_ids),
         start=Memory(),
     )
     if total_memory.in_bytes == 0:
@@ -102,7 +162,8 @@ def _allocate_and_validate_layers(
     layer_allocations = allocate_layers_proportionally(
         total_layers=model_card.n_layers,
         memory_fractions=[
-            node_memory[node_id].ram_available / total_memory for node_id in node_ids
+            node_memory[node_id].inference_available / total_memory
+            for node_id in node_ids
         ],
     )
 
@@ -111,7 +172,7 @@ def _allocate_and_validate_layers(
     for i, node_id in enumerate(node_ids):
         node_layers = layer_allocations[i]
         required_memory = (total_storage * node_layers) // total_layers
-        available_memory = node_memory[node_id].ram_available
+        available_memory = node_memory[node_id].inference_available
         if required_memory > available_memory:
             raise ValueError(
                 f"Node {i} ({node_id}) has insufficient memory: "
@@ -291,6 +352,11 @@ def get_shard_assignments(
                 model_card=model_card,
                 cycle=cycle,
             )
+        case Sharding.Ring:
+            return get_shard_assignments_for_ring_attention(
+                model_card=model_card,
+                cycle=cycle,
+            )
 
 
 def get_mlx_jaccl_devices_matrix(
@@ -459,3 +525,30 @@ def get_ip_for_node(n: NodeId) -> str:
         n: f"{get_ip_for_node(n)}:{coordinator_port}"
         for n in cycle_digraph.list_nodes()
     }
+
+
+def get_shard_assignments_for_ring_attention(
+    model_card: ModelCard,
+    cycle: Cycle,
+) -> ShardAssignments:
+    total_layers = model_card.n_layers
+    world_size = len(cycle)
+    runner_to_shard: dict[RunnerId, ShardMetadata] = {}
+    node_to_runner: dict[NodeId, RunnerId] = {}
+    for i, node_id in enumerate(cycle):
+        shard = RingShardMetadata(
+            model_card=model_card,
+            device_rank=i,
+            world_size=world_size,
+            start_layer=0,
+            end_layer=total_layers,
+            n_layers=total_layers,
+        )
+        runner_id = RunnerId()
+        runner_to_shard[runner_id] = shard
+        node_to_runner[node_id] = runner_id
+    return ShardAssignments(
+        model_id=model_card.model_id,
+        runner_to_shard=runner_to_shard,
+        node_to_runner=node_to_runner,
+    )
diff --git a/src/exo/master/tests/test_placement.py b/src/exo/master/tests/test_placement.py
index b891187266..9aec9993ef 100644
--- a/src/exo/master/tests/test_placement.py
+++ b/src/exo/master/tests/test_placement.py
@@ -73,6 +73,7 @@ def model_card() -> ModelCard:
         n_layers=10,
         hidden_size=30,
         supports_tensor=True,
+        supports_ring=True,
         tasks=[ModelTask.TextGeneration],
         backends=[Backend.MlxMetal],
     )
@@ -94,6 +95,45 @@ def place_instance_command(model_card: ModelCard) -> PlaceInstance:
     )
 
 
+def test_ring_attention_rejects_non_ring_transport(model_card: ModelCard) -> None:
+    command = PlaceInstance(
+        command_id=CommandId(),
+        model_card=model_card,
+        sharding=Sharding.Ring,
+        instance_meta=InstanceMeta.MlxJaccl,
+        min_nodes=2,
+    )
+
+    with pytest.raises(ValueError, match="requires the MlxRing transport"):
+        place_instance(command, Topology(), {}, {}, {}, {})
+
+
+def test_ring_attention_rejects_single_node(model_card: ModelCard) -> None:
+    command = PlaceInstance(
+        command_id=CommandId(),
+        model_card=model_card,
+        sharding=Sharding.Ring,
+        instance_meta=InstanceMeta.MlxRing,
+        min_nodes=1,
+    )
+
+    with pytest.raises(ValueError, match="requires at least two nodes"):
+        place_instance(command, Topology(), {}, {}, {}, {})
+
+
+def test_ring_attention_rejects_model_without_capability(model_card: ModelCard) -> None:
+    command = PlaceInstance(
+        command_id=CommandId(),
+        model_card=model_card.model_copy(update={"supports_ring": False}),
+        sharding=Sharding.Ring,
+        instance_meta=InstanceMeta.MlxRing,
+        min_nodes=2,
+    )
+
+    with pytest.raises(ValueError, match="does not declare Ring attention support"):
+        place_instance(command, Topology(), {}, {}, {}, {})
+
+
 @pytest.mark.parametrize(
     "available_memory,total_layers,expected_layers",
     [
diff --git a/src/exo/master/tests/test_placement_utils.py b/src/exo/master/tests/test_placement_utils.py
index b2f647294d..43914813ce 100644
--- a/src/exo/master/tests/test_placement_utils.py
+++ b/src/exo/master/tests/test_placement_utils.py
@@ -2,7 +2,9 @@
 
 from exo.master.placement_utils import (
     allocate_layers_proportionally,
+    estimate_ring_node_memory,
     filter_cycles_by_memory,
+    filter_cycles_by_replicated_memory,
     get_mlx_jaccl_coordinators,
     get_shard_assignments,
     get_shard_assignments_for_pipeline_parallel,
@@ -18,6 +20,7 @@
 from exo.shared.types.common import NodeId
 from exo.shared.types.memory import Memory
 from exo.shared.types.profiling import (
+    MemoryUsage,
     NetworkInterfaceInfo,
     NodeNetworkInfo,
 )
@@ -93,6 +96,29 @@ def test_filter_cycles_by_insufficient_memory():
     assert len(filtered_cycles) == 0
 
 
+def test_filter_cycles_by_replicated_memory_requires_each_node_to_fit():
+    node1_id = NodeId()
+    node2_id = NodeId()
+    topology = Topology()
+    topology.add_connection(
+        Connection(source=node1_id, sink=node2_id, edge=create_socket_connection(1))
+    )
+    topology.add_connection(
+        Connection(source=node2_id, sink=node1_id, edge=create_socket_connection(2))
+    )
+    cycles = [cycle for cycle in topology.get_cycles() if len(cycle) == 2]
+    node_memory = {
+        node1_id: create_node_memory(8 * 1024),
+        node2_id: create_node_memory(8 * 1024),
+    }
+
+    assert filter_cycles_by_memory(cycles, node_memory, Memory.from_kb(12)) == cycles
+    assert (
+        filter_cycles_by_replicated_memory(cycles, node_memory, Memory.from_kb(12))
+        == []
+    )
+
+
 def test_filter_multiple_cycles_by_memory():
     # arrange
     node_a_id = NodeId()
@@ -689,3 +715,107 @@ def test_two_nodes_non_cfg_model_uses_pipeline(self):
         # First shard starts at 0, last shard ends at 57
         assert layer_ranges[0][0] == 0
         assert layer_ranges[-1][1] == 57
+
+
+class TestRingMemoryAdmission:
+    @staticmethod
+    def _ring_card(
+        *,
+        context_length: int = 131072,
+        num_key_value_heads: int | None = 8,
+    ) -> ModelCard:
+        return ModelCard(
+            model_id=ModelId("ring-test"),
+            n_layers=16,
+            storage_size=Memory.from_mb(700),
+            hidden_size=2048,
+            supports_tensor=True,
+            supports_ring=True,
+            num_key_value_heads=num_key_value_heads,
+            context_length=context_length,
+            tasks=[ModelTask.TextGeneration],
+            backends=[Backend.MlxMetal, Backend.MlxCuda],
+        )
+
+    def test_estimate_exceeds_weights_alone(self) -> None:
+        card = self._ring_card()
+        estimate = estimate_ring_node_memory(card)
+        assert estimate > card.storage_size
+        # 16 layers x 8 kv heads x 128 head dim x 16384 tokens x 2 (K+V)
+        # x 2 bytes x 4 working-set multiplier = 4 GiB of working set.
+        assert estimate - card.storage_size == Memory.from_bytes(
+            2 * 2 * 16 * 8 * 128 * 16384 * 4
+        )
+
+    def test_admission_context_is_capped(self) -> None:
+        huge_context = self._ring_card(context_length=1_000_000)
+        capped = self._ring_card(context_length=16384)
+        assert estimate_ring_node_memory(huge_context) == estimate_ring_node_memory(
+            capped
+        )
+
+    def test_hidden_size_fallback_without_kv_heads(self) -> None:
+        card = self._ring_card(num_key_value_heads=None)
+        estimate = estimate_ring_node_memory(card)
+        assert estimate - card.storage_size == Memory.from_bytes(
+            2 * 2 * 16 * 2048 * 16384 * 4
+        )
+
+
+class TestAcceleratorMemoryAdmission:
+    def test_inference_available_bounded_by_accelerator(self) -> None:
+        usage = MemoryUsage(
+            ram_total=Memory.from_gb(8),
+            ram_available=Memory.from_gb(7),
+            swap_total=Memory(),
+            swap_available=Memory(),
+            accelerator_total=Memory.from_gb(4),
+            accelerator_available=Memory.from_gb(4),
+        )
+        assert usage.inference_available == Memory.from_gb(4)
+
+    def test_inference_available_defaults_to_ram(self) -> None:
+        usage = MemoryUsage(
+            ram_total=Memory.from_gb(8),
+            ram_available=Memory.from_gb(7),
+            swap_total=Memory(),
+            swap_available=Memory(),
+        )
+        assert usage.inference_available == Memory.from_gb(7)
+
+    def test_replicated_filter_rejects_small_vram_node(self) -> None:
+        node1_id = NodeId()
+        node2_id = NodeId()
+        connection = Connection(
+            source=node1_id, sink=node2_id, edge=create_socket_connection(1)
+        )
+        reverse = Connection(
+            source=node2_id, sink=node1_id, edge=create_socket_connection(2)
+        )
+        topology = Topology()
+        topology.add_connection(connection)
+        topology.add_connection(reverse)
+        cycles = topology.get_cycles()
+        big_ram_small_vram = MemoryUsage(
+            ram_total=Memory.from_gb(8),
+            ram_available=Memory.from_gb(7),
+            swap_total=Memory(),
+            swap_available=Memory(),
+            accelerator_total=Memory.from_gb(4),
+            accelerator_available=Memory.from_gb(4),
+        )
+        plenty = MemoryUsage(
+            ram_total=Memory.from_gb(24),
+            ram_available=Memory.from_gb(20),
+            swap_total=Memory(),
+            swap_available=Memory(),
+        )
+        node_memory = {node1_id: plenty, node2_id: big_ram_small_vram}
+
+        required = Memory.from_gb(5)
+        surviving = filter_cycles_by_replicated_memory(cycles, node_memory, required)
+        assert all(node2_id not in cycle.node_ids for cycle in surviving)
+        both_plenty = filter_cycles_by_replicated_memory(
+            cycles, {node1_id: plenty, node2_id: plenty}, required
+        )
+        assert any(node2_id in cycle.node_ids for cycle in both_plenty)
diff --git a/src/exo/shared/logging.py b/src/exo/shared/logging.py
index 80e4763545..e703420143 100644
--- a/src/exo/shared/logging.py
+++ b/src/exo/shared/logging.py
@@ -56,6 +56,10 @@ def logger_setup(log_file: Path | None, verbosity: int = 0):
     # replace all stdlib loggers with _InterceptHandlers that log to loguru
     logging.basicConfig(handlers=[_InterceptHandler()], level=0)
 
+    # diagnose=False everywhere: loguru's variable inspection calls repr() on
+    # every local in the traceback, and repr of MLX device objects can crash
+    # the process mid-log (observed as an opaque SIGSEGV replacing the actual
+    # runner exception on the CUDA backend).
     if verbosity == 0:
         logger.add(
             sys.__stderr__,  # type: ignore
@@ -63,6 +67,7 @@ def logger_setup(log_file: Path | None, verbosity: int = 0):
             level="INFO",
             colorize=True,
             enqueue=True,
+            diagnose=False,
         )
     else:
         logger.add(
@@ -71,6 +76,7 @@ def logger_setup(log_file: Path | None, verbosity: int = 0):
             level="DEBUG",
             colorize=True,
             enqueue=True,
+            diagnose=False,
         )
     if log_file:
         rotate_once = _once_then_never()
@@ -80,6 +86,7 @@ def logger_setup(log_file: Path | None, verbosity: int = 0):
             level="DEBUG" if verbosity > 0 else "INFO",
             colorize=False,
             enqueue=True,
+            diagnose=False,
             rotation=lambda _, __: next(rotate_once),
             retention=_MAX_LOG_ARCHIVES,
             compression=_zstd_compress,
diff --git a/src/exo/shared/models/model_cards.py b/src/exo/shared/models/model_cards.py
index 8911b9323c..dc2cd4085b 100644
--- a/src/exo/shared/models/model_cards.py
+++ b/src/exo/shared/models/model_cards.py
@@ -160,6 +160,7 @@ class ModelCard(FrozenModel):
     n_layers: PositiveInt
     hidden_size: PositiveInt
     supports_tensor: bool
+    supports_ring: bool = False
     num_key_value_heads: PositiveInt | None = None
     tasks: list[ModelTask]
     components: list[ComponentInfo] | None = None
@@ -250,6 +251,7 @@ async def fetch_from_hf(model_id: ModelId) -> "ModelCard":
             n_layers=num_layers,
             hidden_size=config_data.hidden_size or 0,
             supports_tensor=config_data.supports_tensor,
+            supports_ring=config_data.supports_ring,
             num_key_value_heads=config_data.num_key_value_heads,
             context_length=config_data.max_position_embeddings,
             tasks=[ModelTask.TextGeneration],
@@ -302,6 +304,14 @@ def supports_tensor(self) -> bool:
             ["Gemma4ForConditionalGeneration"],
         ]
 
+    @property
+    def supports_ring(self) -> bool:
+        """Whether the model uses an attention implementation verified for Ring."""
+        return self.architectures in [
+            ["LlamaForCausalLM"],
+            ["Qwen3ForCausalLM"],
+        ]
+
     @model_validator(mode="before")
     @classmethod
     def defer_to_text_config(cls, data: dict[str, Any], info: ValidationInfo):
diff --git a/src/exo/shared/tests/test_model_card_ring_support.py b/src/exo/shared/tests/test_model_card_ring_support.py
new file mode 100644
index 0000000000..6f55ad4d5b
--- /dev/null
+++ b/src/exo/shared/tests/test_model_card_ring_support.py
@@ -0,0 +1,34 @@
+from exo.shared.models.model_cards import ConfigData, ModelCard, ModelId, ModelTask
+from exo.shared.types.backends import Backend
+from exo.shared.types.memory import Memory
+
+
+def test_config_detects_verified_ring_architectures() -> None:
+    for architecture in ("LlamaForCausalLM", "Qwen3ForCausalLM"):
+        config = ConfigData.model_validate(
+            {"architectures": [architecture], "num_hidden_layers": 1}
+        )
+
+        assert config.supports_ring is True
+
+
+def test_config_rejects_unverified_ring_architecture() -> None:
+    config = ConfigData.model_validate(
+        {"architectures": ["Gemma2ForCausalLM"], "num_hidden_layers": 1}
+    )
+
+    assert config.supports_ring is False
+
+
+def test_model_card_defaults_ring_support_to_false() -> None:
+    card = ModelCard(
+        model_id=ModelId("test/model"),
+        storage_size=Memory.from_bytes(1),
+        n_layers=1,
+        hidden_size=1,
+        supports_tensor=False,
+        tasks=[ModelTask.TextGeneration],
+        backends=[Backend.MlxMetal],
+    )
+
+    assert card.supports_ring is False
diff --git a/src/exo/shared/types/profiling.py b/src/exo/shared/types/profiling.py
index a3548dc91a..ff695d9e7e 100644
--- a/src/exo/shared/types/profiling.py
+++ b/src/exo/shared/types/profiling.py
@@ -1,7 +1,7 @@
 import shutil
 from collections.abc import Sequence
 from pathlib import Path
-from typing import Literal, Self
+from typing import Literal, Self, cast
 
 import psutil
 
@@ -15,6 +15,21 @@ class MemoryUsage(FrozenModel):
     ram_available: Memory
     swap_total: Memory
     swap_available: Memory
+    # Dedicated accelerator memory (e.g. CUDA VRAM). None on unified-memory
+    # platforms, where system RAM is the accelerator memory.
+    accelerator_total: Memory | None = None
+    accelerator_available: Memory | None = None
+
+    @property
+    def inference_available(self) -> Memory:
+        """Memory actually available to hold model state on this node.
+
+        On discrete-GPU nodes the accelerator's free memory bounds what
+        inference can use, regardless of how much system RAM is free.
+        """
+        if self.accelerator_available is None:
+            return self.ram_available
+        return min(self.ram_available, self.accelerator_available)
 
     @classmethod
     def from_bytes(
@@ -31,15 +46,44 @@ def from_bytes(
     def from_psutil(cls, *, override_memory: int | None) -> Self:
         vm = psutil.virtual_memory()
         sm = psutil.swap_memory()
+        accelerator = _accelerator_memory()
 
-        return cls.from_bytes(
-            ram_total=vm.total,
-            ram_available=vm.available if override_memory is None else override_memory,
-            swap_total=sm.total,
-            swap_available=sm.free,
+        return cls(
+            ram_total=Memory.from_bytes(vm.total),
+            ram_available=Memory.from_bytes(
+                vm.available if override_memory is None else override_memory
+            ),
+            swap_total=Memory.from_bytes(sm.total),
+            swap_available=Memory.from_bytes(sm.free),
+            accelerator_total=accelerator[0],
+            accelerator_available=accelerator[1],
         )
 
 
+def _accelerator_memory() -> tuple[Memory | None, Memory | None]:
+    """Report dedicated accelerator memory where one exists (CUDA VRAM).
+
+    Returns (None, None) on unified-memory platforms or when no NVML-visible
+    device is present.
+    """
+    try:
+        import pynvml
+    except ImportError:
+        return None, None
+    try:
+        pynvml.nvmlInit()
+        try:
+            handle = pynvml.nvmlDeviceGetHandleByIndex(0)
+            info = pynvml.nvmlDeviceGetMemoryInfo(handle)  # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType]
+            total_bytes = cast(int, info.total)
+            free_bytes = cast(int, info.free)
+            return Memory.from_bytes(total_bytes), Memory.from_bytes(free_bytes)
+        finally:
+            pynvml.nvmlShutdown()
+    except Exception:
+        return None, None
+
+
 class DiskUsage(FrozenModel):
     """Disk space usage for the models directory."""
 
diff --git a/src/exo/shared/types/worker/shards.py b/src/exo/shared/types/worker/shards.py
index 59a6c54eb0..28bb6bacee 100644
--- a/src/exo/shared/types/worker/shards.py
+++ b/src/exo/shared/types/worker/shards.py
@@ -10,6 +10,7 @@
 class Sharding(str, Enum):
     Tensor = "Tensor"
     Pipeline = "Pipeline"
+    Ring = "Ring"
 
 
 class BaseShardMetadata(TaggedModel):
@@ -79,6 +80,19 @@ class TensorShardMetadata(BaseShardMetadata):
     pass
 
 
+@final
+class RingShardMetadata(BaseShardMetadata):
+    """Ring-attention shard metadata.
+
+    Unlike pipeline (layer-split) or tensor (weight-split), ring attention
+    replicates all layers on every device but partitions the **sequence**
+    dimension. Each device holds a contiguous block of the input sequence
+    and KV blocks rotate around the ring during attention computation.
+    """
+
+    sequence_block_size: int = Field(default=512, ge=1)
+
+
 ShardMetadata: TypeAlias = (
-    PipelineShardMetadata | CfgShardMetadata | TensorShardMetadata
+    PipelineShardMetadata | CfgShardMetadata | TensorShardMetadata | RingShardMetadata
 )
diff --git a/src/exo/utils/stall_watchdog.py b/src/exo/utils/stall_watchdog.py
new file mode 100644
index 0000000000..6b288c81f8
--- /dev/null
+++ b/src/exo/utils/stall_watchdog.py
@@ -0,0 +1,69 @@
+import faulthandler
+import os
+import sys
+import threading
+from types import TracebackType
+from typing import Self, final
+
+from loguru import logger
+
+STALL_EXIT_CODE = 70
+
+
+@final
+class StallWatchdog:
+    """Terminate the process when a monitored operation stops making progress.
+
+    A daemon thread trips after `timeout_seconds` elapse without a `kick()`.
+    Used around distributed prefill, where a cross-rank scheduling stall can
+    wedge every rank indefinitely: converting the hang into a runner death
+    lets the supervisor tear the instance down and surface an error instead
+    of hanging the request forever. All thread stacks are dumped first so the
+    stall site is preserved for diagnosis.
+
+    A `timeout_seconds` of 0 disables the watchdog entirely.
+    """
+
+    def __init__(self, timeout_seconds: float, description: str) -> None:
+        self._timeout_seconds = timeout_seconds
+        self._description = description
+        self._kicked = threading.Event()
+        self._closed = threading.Event()
+        self._thread: threading.Thread | None = None
+        if timeout_seconds > 0:
+            self._thread = threading.Thread(
+                target=self._watch, name="stall-watchdog", daemon=True
+            )
+            self._thread.start()
+
+    def kick(self) -> None:
+        """Record progress, granting the operation another full timeout window."""
+        self._kicked.set()
+
+    def close(self) -> None:
+        self._closed.set()
+
+    def __enter__(self) -> Self:
+        return self
+
+    def __exit__(
+        self,
+        exception_type: type[BaseException] | None,
+        exception: BaseException | None,
+        traceback: TracebackType | None,
+    ) -> None:
+        self.close()
+
+    def _watch(self) -> None:
+        while True:
+            self._kicked.clear()
+            if self._closed.wait(self._timeout_seconds):
+                return
+            if self._kicked.is_set():
+                continue
+            logger.critical(
+                f"{self._description} made no progress for "
+                f"{self._timeout_seconds:.0f}s - terminating runner"
+            )
+            faulthandler.dump_traceback(file=sys.stderr, all_threads=True)
+            os._exit(STALL_EXIT_CODE)
diff --git a/src/exo/utils/tests/test_stall_watchdog.py b/src/exo/utils/tests/test_stall_watchdog.py
new file mode 100644
index 0000000000..238a0ccbd9
--- /dev/null
+++ b/src/exo/utils/tests/test_stall_watchdog.py
@@ -0,0 +1,49 @@
+import threading
+import time
+
+import pytest
+
+import exo.utils.stall_watchdog as stall_watchdog_module
+from exo.utils.stall_watchdog import StallWatchdog
+
+
+@pytest.fixture
+def captured_exit(monkeypatch: pytest.MonkeyPatch) -> threading.Event:
+    tripped = threading.Event()
+
+    def fake_exit(code: int) -> None:
+        assert code == stall_watchdog_module.STALL_EXIT_CODE
+        tripped.set()
+
+    def fake_dump_traceback(**_kwargs: object) -> None:
+        pass
+
+    monkeypatch.setattr(stall_watchdog_module.os, "_exit", fake_exit)
+    monkeypatch.setattr(
+        stall_watchdog_module.faulthandler, "dump_traceback", fake_dump_traceback
+    )
+    return tripped
+
+
+def test_trips_when_no_progress(captured_exit: threading.Event) -> None:
+    with StallWatchdog(0.1, "test operation"):
+        assert captured_exit.wait(timeout=2.0)
+
+
+def test_kick_defers_the_trip(captured_exit: threading.Event) -> None:
+    with StallWatchdog(0.5, "test operation") as watchdog:
+        for _ in range(5):
+            time.sleep(0.1)
+            watchdog.kick()
+        assert not captured_exit.is_set()
+
+
+def test_close_stops_watching(captured_exit: threading.Event) -> None:
+    watchdog = StallWatchdog(0.1, "test operation")
+    watchdog.close()
+    assert not captured_exit.wait(timeout=0.5)
+
+
+def test_zero_timeout_disables(captured_exit: threading.Event) -> None:
+    with StallWatchdog(0, "test operation"):
+        assert not captured_exit.wait(timeout=0.3)
diff --git a/src/exo/worker/engines/mlx/generator/batch_generate.py b/src/exo/worker/engines/mlx/generator/batch_generate.py
index 5c7394ddec..78c04aa09a 100644
--- a/src/exo/worker/engines/mlx/generator/batch_generate.py
+++ b/src/exo/worker/engines/mlx/generator/batch_generate.py
@@ -45,6 +45,7 @@
     set_needs_topk,
     take_ready_topk,
 )
+from exo.worker.engines.mlx.ring_attention import uses_ring_sequence_parallel_prefill
 from exo.worker.engines.mlx.types import KVCacheType, Model
 from exo.worker.engines.mlx.utils_mlx import (
     fix_unmatched_think_end_tokens,
@@ -206,6 +207,9 @@ def submit(
         use_remote = (
             uncached_count > REMOTE_PREFILL_MIN_TOKENS
             and task_params.prefill_endpoint is not None
+            and not uses_ring_sequence_parallel_prefill(
+                self.model, len(prompt_tokens) - 1, self.group
+            )
         )
 
         _prefill_tps: float = 0.0
@@ -279,7 +283,13 @@ def submit(
                 prefill_tps=_prefill_tps,
             )
 
-        last_tokens = prompt_tokens[-2:]
+        last_tokens = (
+            prompt_tokens[-1:]
+            if uses_ring_sequence_parallel_prefill(
+                self.model, len(prompt_tokens) - 1, self.group
+            )
+            else prompt_tokens[-2:]
+        )
 
         logits_processors: list[Callable[[mx.array, mx.array], mx.array]] = (
             make_logits_processors(
diff --git a/src/exo/worker/engines/mlx/generator/generate.py b/src/exo/worker/engines/mlx/generator/generate.py
index 2e3d051251..3b2b6b6d76 100644
--- a/src/exo/worker/engines/mlx/generator/generate.py
+++ b/src/exo/worker/engines/mlx/generator/generate.py
@@ -1,6 +1,7 @@
 import contextlib
 import functools
 import math
+import os
 import time
 import uuid
 from typing import Callable, Generator, cast, get_args
@@ -31,6 +32,7 @@
 from exo.shared.types.worker.runner_response import (
     GenerationResponse,
 )
+from exo.utils.stall_watchdog import StallWatchdog
 from exo.worker.engines.mlx.auto_parallel import (
     PipelineFirstLayer,
     PipelineLastLayer,
@@ -56,6 +58,12 @@
     MAX_TOKENS,
 )
 from exo.worker.engines.mlx.generator.remote_prefill import remote_prefill
+from exo.worker.engines.mlx.ring_attention import (
+    ring_prefill_block_size,
+    set_ring_prefill,
+    uses_ring_sequence_parallel_prefill,
+    validate_ring_cache,
+)
 from exo.worker.engines.mlx.types import KVCacheType, Model
 from exo.worker.engines.mlx.utils_mlx import (
     apply_chat_template,
@@ -324,17 +332,49 @@ def combined_progress_callback(processed: int, total: int) -> None:
             distributed_prompt_progress_callback()
         progress_callback(processed, total)
 
+    is_ring_prefill = uses_ring_sequence_parallel_prefill(model, num_tokens, group)
+    if is_ring_prefill:
+        validate_ring_cache(cache)
     set_pipeline_prefill(model, is_prefill=True)
-
-    mx_barrier(group)
-    logger.info("Starting prefill")
+    set_ring_prefill(model, is_prefill=is_ring_prefill)
 
     is_pipeline = _has_pipeline_communication_layer(model)
 
     prefill_step_size = 4096
 
     try:
-        if is_pipeline and num_tokens >= prefill_step_size:
+        mx_barrier(group)
+        logger.info("Starting prefill")
+
+        if is_ring_prefill:
+            assert group is not None
+            rank = group.rank()
+            world_size = group.size()
+            max_chunk_size = world_size * ring_prefill_block_size(model)
+            chunk_ranges = list(range(0, num_tokens, max_chunk_size))
+            if len(chunk_ranges) > 1 and num_tokens - chunk_ranges[-1] < world_size:
+                chunk_ranges.pop()
+
+            stall_timeout_seconds = float(
+                os.environ.get("EXO_PREFILL_STALL_TIMEOUT", "300")
+            )
+            combined_progress_callback(0, num_tokens)
+            with StallWatchdog(
+                stall_timeout_seconds, f"Ring prefill ({num_tokens} tokens)"
+            ) as stall_watchdog:
+                for chunk_start in chunk_ranges:
+                    chunk_end = min(chunk_start + max_chunk_size, num_tokens)
+                    if chunk_start == chunk_ranges[-1]:
+                        chunk_end = num_tokens
+                    chunk_size = chunk_end - chunk_start
+                    start = chunk_start + (chunk_size * rank) // world_size
+                    end = chunk_start + (chunk_size * (rank + 1)) // world_size
+                    with mx.stream(generation_stream):
+                        model(prompt_tokens[start:end][None], cache=cache)
+                        mx.eval([c.state for c in cache])  # type: ignore
+                    stall_watchdog.kick()
+                    combined_progress_callback(chunk_end, num_tokens)
+        elif is_pipeline and num_tokens >= prefill_step_size:
             set_pipeline_queue_sends(model, queue_sends=True)
             assert group is not None, "Pipeline prefill requires a distributed group"
             pipeline_parallel_prefill(
@@ -364,27 +404,25 @@ def combined_progress_callback(processed: int, total: int) -> None:
                 prompt_progress_callback=combined_progress_callback,
             ):
                 break  # Stop after first iteration - cache is now filled
-    except PrefillCancelled:
+    finally:
         set_pipeline_queue_sends(model, queue_sends=False)
         set_pipeline_prefill(model, is_prefill=False)
-        raise
-
-    set_pipeline_queue_sends(model, queue_sends=False)
-    set_pipeline_prefill(model, is_prefill=False)
-
-    # stream_generate added 1 extra generated token to the cache, so we should trim it.
-    # Because of needing to roll back arrays cache, we will generate on 2 tokens so trim 1 more.
-    pre_gen = snapshots[-2] if has_ssm else None
-    for i, c in enumerate(cache):
-        non_trimmable = is_non_trimmable_cache_entry(c)
-        if has_ssm and non_trimmable:
-            assert pre_gen is not None
-            restored = copy_snapshot_entry(pre_gen.states[i])
-            if restored is not None:
-                cache[i] = restored  # type: ignore
-        else:
-            assert not non_trimmable
-            c.trim(2)
+        set_ring_prefill(model, is_prefill=False)
+
+    if not is_ring_prefill:
+        # stream_generate added 1 extra generated token to the cache, so we should trim it.
+        # Because of needing to roll back arrays cache, we will generate on 2 tokens so trim 1 more.
+        pre_gen = snapshots[-2] if has_ssm else None
+        for i, c in enumerate(cache):
+            non_trimmable = is_non_trimmable_cache_entry(c)
+            if has_ssm and non_trimmable:
+                assert pre_gen is not None
+                restored = copy_snapshot_entry(pre_gen.states[i])
+                if restored is not None:
+                    cache[i] = restored  # type: ignore
+            else:
+                assert not non_trimmable
+                c.trim(2)
 
     elapsed = time.perf_counter() - start_time
     tokens_per_sec = num_tokens / elapsed if elapsed > 0 else 0.0
@@ -640,6 +678,9 @@ def mlx_generate(
     use_remote = (
         len(prompt_tokens) > REMOTE_PREFILL_MIN_TOKENS
         and task.prefill_endpoint is not None
+        and not uses_ring_sequence_parallel_prefill(
+            model, len(prompt_tokens) - 1, group
+        )
     )
     remote_prefilled = False
     prefill_tps = 0.0
@@ -707,7 +748,11 @@ def mlx_generate(
             )
 
     # stream_generate starts from the last token
-    last_token = prompt_tokens[-2:]
+    last_token = (
+        prompt_tokens[-1:]
+        if uses_ring_sequence_parallel_prefill(model, len(prompt_tokens) - 1, group)
+        else prompt_tokens[-2:]
+    )
 
     max_tokens = task.max_output_tokens or MAX_TOKENS
     accumulated_text = ""
diff --git a/src/exo/worker/engines/mlx/ring_attention.py b/src/exo/worker/engines/mlx/ring_attention.py
new file mode 100644
index 0000000000..5a780a4861
--- /dev/null
+++ b/src/exo/worker/engines/mlx/ring_attention.py
@@ -0,0 +1,733 @@
+# pyright: reportAny=false, reportPrivateUsage=false
+# pyright: reportUnknownMemberType=false, reportUnknownArgumentType=false
+# pyright: reportArgumentType=false, reportCallIssue=false
+
+"""Ring Attention implementation for sequence-parallel distributed inference.
+
+Ring Attention splits the input sequence into blocks across devices in a ring
+topology. Each device computes partial attention against its local Q and a
+rotating KV block. Sends/receives of the next KV block overlap with attention
+computation on the current block, hiding network latency behind compute.
+
+During prefill, KV blocks rotate around the ring and every device accumulates
+the full KV cache from the received blocks — so decode has the complete cache
+available without any extra communication pass.
+
+During decode, every device has the complete cache populated during prefill.
+The normal synchronised generation path feeds the same next token to each
+replicated model, so each device can update its own full cache locally.
+
+References:
+    Liu et al., "Ring Attention with Blockwise Transformers for Near-Infinite Context" (2023)
+    https://arxiv.org/abs/2310.01889
+"""
+
+from __future__ import annotations
+
+from collections.abc import Generator, Sequence
+from typing import Protocol, cast
+
+import mlx.core as mx
+import mlx.nn as nn
+from mlx_lm.models.cache import KVCache
+
+from exo.shared.types.worker.runner_response import ModelLoadingResponse
+from exo.worker.engines.mlx.auto_parallel import (
+    CustomMlxLayer,
+    _LayerCallable,
+    get_inner_model,
+    get_layers,
+    patch_tensor_model,
+)
+
+_SUPPORTED_ATTENTION_TYPES = frozenset(
+    {
+        ("mlx_lm.models.llama", "Attention"),
+        ("mlx_lm.models.qwen3", "Attention"),
+    }
+)
+
+
+class _AttentionLayer(Protocol):
+    """Structural type for attention layers that expose q/k/v projections."""
+
+    def __call__(
+        self,
+        x: mx.array,
+        mask: mx.array | None = None,
+        cache: object | None = None,
+    ) -> mx.array: ...
+
+
+def _make_block_causal_mask(
+    local_seq_len: int,
+    kv_seq_len: int,
+    local_offset: int,
+    kv_offset: int,
+    *,
+    n_heads: int,
+) -> mx.array | None:
+    """Build a causal mask for a specific (query-block, kv-block) pair.
+
+    Returns ``None`` for full attention (no masking needed), or a boolean
+    mask of shape ``(1, local_seq_len, kv_seq_len)`` where ``True``
+    means "attend".
+    """
+    local_end = local_offset + local_seq_len
+    kv_end = kv_offset + kv_seq_len
+
+    if kv_end <= local_offset:
+        return None
+
+    if kv_offset >= local_end:
+        return mx.zeros((n_heads, local_seq_len, kv_seq_len), dtype=mx.bool_)
+
+    q_positions = mx.arange(local_offset, local_offset + local_seq_len)[:, None]
+    k_positions = mx.arange(kv_offset, kv_offset + kv_seq_len)[None, :]
+    causal = q_positions >= k_positions
+    return causal[None, :, :]
+
+
+class RingAttentionLayer(CustomMlxLayer):
+    """Wraps an attention layer with ring-attention distributed communication.
+
+    During prefill, the input sequence is split into blocks across devices.
+    Each device computes Q from its local block, then rotates KV blocks
+    around the ring. All received KV blocks are stored in the cache so
+    decode has the full KV available.
+
+    During decode, all ranks receive the same generated token and update their
+    replicated full cache locally.
+    """
+
+    def __init__(
+        self,
+        original_layer: _LayerCallable,
+        group: mx.distributed.Group,
+        sequence_block_size: int = 512,
+        send_stream: mx.Stream | None = None,
+        receive_stream: mx.Stream | None = None,
+    ):
+        super().__init__(original_layer)
+        self.group = group
+        self.rank: int = group.rank()
+        self.world_size: int = group.size()
+        self.sequence_block_size = sequence_block_size
+        # MLX distributed send/receive are CPU operations. Unified memory lets
+        # them consume Metal-produced KV arrays without an explicit copy.
+        self.send_stream = send_stream or mx.new_stream(mx.cpu)
+        self.receive_stream = receive_stream or mx.new_stream(mx.cpu)
+        self.is_prefill: bool = False
+
+    def __call__(
+        self,
+        x: mx.array,
+        mask: mx.array | None = None,
+        cache: object | None = None,
+    ) -> mx.array:
+        if self.world_size == 1:
+            return self.original_layer(x, mask=mask, cache=cache)
+
+        if not self.is_prefill:
+            return self._decode_step(x, mask=mask, cache=cache)
+
+        return self._prefill_step(x, mask=mask, cache=cache)
+
+    def _decode_step(
+        self,
+        x: mx.array,
+        mask: mx.array | None = None,
+        cache: object | None = None,
+    ) -> mx.array:
+        """Regular decode against the replicated full KV cache.
+
+        Ring parallelism is useful for prefill, where each rank owns a
+        different sequence block.  After prefill every rank has the complete
+        cache, and generation is synchronised by the normal runner protocol,
+        so every rank receives the *same* next token.  Gathering K/V here
+        would append that token once per rank and corrupt the cache.
+
+        Forwarding to the wrapped attention module keeps decode byte-for-byte
+        identical to the non-ring path, including support for every cache
+        type the model itself supports (BatchKVCache, quantized, rotating).
+        """
+        return self.original_layer(x, mask=mask, cache=cache)
+
+    def _prefill_step(
+        self,
+        x: mx.array,
+        mask: mx.array | None = None,
+        cache: object | None = None,
+    ) -> mx.array:
+        """Ring-attention prefill: rotate KV blocks around the ring.
+
+        Stores all received KV blocks in sequence order in the cache so
+        decode has the full KV available without extra communication.
+        """
+        # The verified model implementations create a standard causal mask from
+        # their local input. Ring prefill must rebuild that mask with global
+        # sequence offsets, so the local mask is intentionally not forwarded.
+        # `_validate_ring_attention_layer` rejects implementations whose mask
+        # semantics have not been verified equivalent.
+        del mask
+        original = cast(_AttentionLayer, self.original_layer)
+
+        _, local_seq_len, _ = x.shape
+        if local_seq_len == 0:
+            raise ValueError("Ring attention does not support empty sequence blocks")
+
+        attn = self._get_attn_module(original)
+        queries, keys, values, batch_dim, _, n_heads, n_kv_heads = self._project_qkv(
+            attn, x
+        )
+        head_dim = keys.shape[-1]
+
+        cache_obj = self._require_kv_cache(cache)
+        cache_offset = self._cache_offset(cache_obj)
+        block_lengths = self._block_lengths(local_seq_len)
+        local_offset = cache_offset + sum(block_lengths[: self.rank])
+
+        rope = getattr(attn, "rope", None)
+        if rope is not None:
+            queries = rope(queries, offset=local_offset)
+            keys = rope(keys, offset=local_offset)
+
+        # Communication ops must never depend on unfinished GPU work when they
+        # are posted: with MLX_METAL_FAST_SYNCH the CPU send would then block
+        # on a Metal shared event inside a bounded command-buffer queue, which
+        # can form a cross-rank circular wait (observed as an intermittent
+        # long-sequence prefill hang). Materialising the KV payload here keeps
+        # the send and receive streams pure-CPU so transport always drains.
+        mx.eval(keys, values)
+
+        all_keys: dict[int, mx.array] = {self.rank: keys}
+        all_values: dict[int, mx.array] = {self.rank: values}
+
+        current_keys = keys
+        current_values = values
+        current_rank = self.rank
+
+        running_max: mx.array | None = None
+        running_sum: mx.array | None = None
+        running_output: mx.array | None = None
+
+        cached_keys, cached_values = self._existing_cache(cache_obj)
+        if cached_keys is not None and cached_values is not None:
+            running_max, running_sum, running_output = self._accumulate_attention(
+                queries,
+                cached_keys,
+                cached_values,
+                scale=self._get_scale(attn),
+                mask=None,
+                running_max=running_max,
+                running_sum=running_sum,
+                running_output=running_output,
+            )
+
+        pending_sends: list[mx.array] = []
+        for step in range(self.world_size):
+            next_keys: mx.array | None = None
+            next_values: mx.array | None = None
+            next_rank: int | None = None
+            if step < self.world_size - 1:
+                destination_rank = (self.rank + 1) % self.world_size
+                source_peer_rank = (self.rank - 1) % self.world_size
+                next_rank = (self.rank - step - 1) % self.world_size
+
+                sent_keys = mx.distributed.send(
+                    current_keys,
+                    destination_rank,
+                    group=self.group,
+                    stream=self.send_stream,
+                )
+                sent_values = mx.distributed.send(
+                    current_values,
+                    destination_rank,
+                    group=self.group,
+                    stream=self.send_stream,
+                )
+                pending_sends.extend((sent_keys, sent_values))
+
+                # Templates are allocated on the CPU receive stream so the
+                # posted receive never waits on a GPU kernel (see the payload
+                # materialisation note above).
+                recv_keys = mx.distributed.recv_like(
+                    mx.zeros(
+                        (batch_dim, n_kv_heads, block_lengths[next_rank], head_dim),
+                        dtype=current_keys.dtype,
+                        stream=self.receive_stream,
+                    ),
+                    source_peer_rank,
+                    group=self.group,
+                    stream=self.receive_stream,
+                )
+                recv_values = mx.distributed.recv_like(
+                    mx.zeros(
+                        (batch_dim, n_kv_heads, block_lengths[next_rank], head_dim),
+                        dtype=current_values.dtype,
+                        stream=self.receive_stream,
+                    ),
+                    source_peer_rank,
+                    group=self.group,
+                    stream=self.receive_stream,
+                )
+
+                # Both directions are posted before attention is scheduled.
+                # Communication uses dedicated streams while the attention graph
+                # stays on the caller's compute stream.
+                mx.async_eval(sent_keys, sent_values)
+                mx.async_eval(recv_keys, recv_values)
+                next_keys = recv_keys
+                next_values = recv_values
+
+            block_mask = _make_block_causal_mask(
+                local_seq_len=local_seq_len,
+                kv_seq_len=current_keys.shape[2],
+                local_offset=local_offset,
+                kv_offset=cache_offset + sum(block_lengths[:current_rank]),
+                n_heads=n_heads,
+            )
+
+            running_max, running_sum, running_output = self._accumulate_attention(
+                queries,
+                current_keys,
+                current_values,
+                scale=self._get_scale(attn),
+                mask=block_mask,
+                running_max=running_max,
+                running_sum=running_sum,
+                running_output=running_output,
+            )
+
+            # MLX is lazy: explicitly scheduling the recurrence here is what
+            # creates a computation window in which the posted receive can run.
+            mx.async_eval(running_max, running_sum, running_output)
+
+            if (
+                next_keys is not None
+                and next_values is not None
+                and next_rank is not None
+            ):
+                mx.eval(next_keys, next_values)
+                current_keys = next_keys
+                current_values = next_values
+                current_rank = next_rank
+                all_keys[current_rank] = current_keys
+                all_values[current_rank] = current_values
+
+        # Ensure the communication streams no longer reference layer-local KV
+        # buffers before cache assembly and the next transformer layer begin.
+        if pending_sends:
+            mx.eval(pending_sends)
+
+        if cache_obj is not None:
+            full_keys = mx.concatenate(
+                [all_keys[i] for i in range(self.world_size)], axis=2
+            )
+            full_values = mx.concatenate(
+                [all_values[i] for i in range(self.world_size)], axis=2
+            )
+
+            cache_obj.update_and_fetch(full_keys, full_values)
+
+        assert running_output is not None and running_sum is not None
+        output = (
+            (running_output / running_sum)
+            .transpose(0, 2, 1, 3)
+            .reshape(batch_dim, local_seq_len, -1)
+        )
+
+        o_proj = getattr(attn, "o_proj", None)
+        if o_proj is not None:
+            output = o_proj(output)
+
+        return output
+
+    def _block_lengths(self, local_seq_len: int) -> list[int]:
+        """Return every rank's local sequence length for this prefill call."""
+        lengths = mx.distributed.all_gather(
+            mx.array([local_seq_len], dtype=mx.int32), group=self.group
+        )
+        mx.eval(lengths)
+        return cast(list[int], lengths.tolist())
+
+    def _cache_offset(self, cache: object | None) -> int:
+        offset = getattr(cache, "offset", 0)
+        if not isinstance(offset, int):
+            raise TypeError("Ring attention requires an integer cache offset")
+        return offset
+
+    def _require_kv_cache(self, cache: object | None) -> KVCache | None:
+        if cache is None:
+            return None
+        if not isinstance(cache, KVCache):
+            raise TypeError(
+                "Ring attention supports only the standard unquantized KVCache; "
+                f"received {type(cache).__name__}"
+            )
+        return cache
+
+    def _existing_cache(
+        self, cache: object | None
+    ) -> tuple[mx.array | None, mx.array | None]:
+        """Read the populated part of a standard, unquantized KV cache."""
+        if cache is None or self._cache_offset(cache) == 0:
+            return None, None
+
+        keys = getattr(cache, "keys", None)
+        values = getattr(cache, "values", None)
+        if not isinstance(keys, mx.array) or not isinstance(values, mx.array):
+            raise TypeError("Ring attention currently requires an unquantized KV cache")
+        offset = self._cache_offset(cache)
+        return keys[..., :offset, :], values[..., :offset, :]
+
+    def _accumulate_attention(
+        self,
+        queries: mx.array,
+        keys: mx.array,
+        values: mx.array,
+        *,
+        scale: float,
+        mask: mx.array | None,
+        running_max: mx.array | None,
+        running_sum: mx.array | None,
+        running_output: mx.array | None,
+    ) -> tuple[mx.array, mx.array, mx.array]:
+        """Merge one attention block using the numerically stable online softmax.
+
+        Individual block attention outputs cannot be added: each is normalized
+        by a different softmax denominator.  This keeps the running maximum,
+        denominator and weighted-value numerator needed to exactly reproduce
+        attention over the concatenation of all blocks.
+        """
+        if keys.shape[1] != queries.shape[1]:
+            if queries.shape[1] % keys.shape[1] != 0:
+                raise ValueError("Number of query heads must divide KV heads")
+            repeats = queries.shape[1] // keys.shape[1]
+            keys = mx.repeat(keys, repeats, axis=1)
+            values = mx.repeat(values, repeats, axis=1)
+
+        scores = (queries @ keys.transpose(0, 1, 3, 2)) * scale
+        if mask is None:
+            valid = mx.ones(scores.shape, dtype=mx.bool_)
+        else:
+            valid = mask[None, :, :, :]
+            scores = mx.where(valid, scores, -mx.inf)
+
+        block_has_values = mx.any(valid, axis=-1, keepdims=True)
+        block_max = mx.max(scores, axis=-1, keepdims=True)
+        safe_block_max = mx.where(block_has_values, block_max, 0.0)
+        weights = mx.where(valid, mx.exp(scores - safe_block_max), 0.0)
+        block_sum = mx.sum(weights, axis=-1, keepdims=True)
+        block_output = weights @ values
+
+        if running_max is None or running_sum is None or running_output is None:
+            return safe_block_max, block_sum, block_output
+
+        has_values = (running_sum + block_sum) > 0
+        new_max = mx.where(has_values, mx.maximum(running_max, safe_block_max), 0.0)
+        running_scale = mx.where(running_sum > 0, mx.exp(running_max - new_max), 0.0)
+        block_scale = mx.where(block_sum > 0, mx.exp(safe_block_max - new_max), 0.0)
+        return (
+            new_max,
+            running_sum * running_scale + block_sum * block_scale,
+            running_output * running_scale + block_output * block_scale,
+        )
+
+    @staticmethod
+    def _get_attn_module(original: _AttentionLayer) -> nn.Module:
+        if hasattr(original, "q_proj") or hasattr(original, "q_b_proj"):
+            return cast(nn.Module, original)
+        self_attn = getattr(original, "self_attn", None)
+        if self_attn is not None:
+            return cast(nn.Module, self_attn)
+        attn = getattr(original, "attn", None)
+        if attn is not None:
+            return cast(nn.Module, attn)
+        raise ValueError(
+            f"Cannot find attention projections on {type(original).__name__}"
+        )
+
+    def _project_qkv(
+        self, attn: nn.Module, x: mx.array
+    ) -> tuple[mx.array, mx.array, mx.array, int, int, int, int]:
+        """Project Q/K/V exactly as standard Llama and Qwen attention does."""
+        batch_dim, sequence_length, _ = x.shape
+        n_heads = self._get_n_heads(attn)
+        n_kv_heads = self._get_n_kv_heads(attn)
+        queries = self._project_q(attn, x)
+        keys = self._project_k(attn, x)
+        values = self._project_v(attn, x)
+        head_dim = queries.shape[-1] // n_heads
+
+        queries = queries.reshape(batch_dim, sequence_length, n_heads, head_dim)
+        keys = keys.reshape(batch_dim, sequence_length, n_kv_heads, head_dim)
+        values = values.reshape(batch_dim, sequence_length, n_kv_heads, head_dim)
+        if queries.shape[-1] != keys.shape[-1] or keys.shape[-1] != values.shape[-1]:
+            raise ValueError("Ring attention requires equal Q, K and V head dimensions")
+
+        q_norm = getattr(attn, "q_norm", None)
+        if q_norm is not None:
+            queries = q_norm(queries)
+        k_norm = getattr(attn, "k_norm", None)
+        if k_norm is not None:
+            keys = k_norm(keys)
+
+        return (
+            queries.transpose(0, 2, 1, 3),
+            keys.transpose(0, 2, 1, 3),
+            values.transpose(0, 2, 1, 3),
+            batch_dim,
+            sequence_length,
+            n_heads,
+            n_kv_heads,
+        )
+
+    def _project_q(self, attn: nn.Module, x: mx.array) -> mx.array:
+        q_proj = getattr(attn, "q_proj", None)
+        if q_proj is not None:
+            return q_proj(x)
+        raise ValueError(f"Cannot find Q projection on {type(attn).__name__}")
+
+    def _project_k(self, attn: nn.Module, x: mx.array) -> mx.array:
+        k_proj = getattr(attn, "k_proj", None)
+        if k_proj is not None:
+            return k_proj(x)
+        raise ValueError(f"Cannot find K projection on {type(attn).__name__}")
+
+    def _project_v(self, attn: nn.Module, x: mx.array) -> mx.array:
+        v_proj = getattr(attn, "v_proj", None)
+        if v_proj is not None:
+            return v_proj(x)
+        raise ValueError(f"Cannot find V projection on {type(attn).__name__}")
+
+    def _get_n_heads(self, attn: nn.Module) -> int:
+        for attr in ("n_heads", "num_heads", "num_attention_heads"):
+            val = getattr(attn, attr, None)
+            if isinstance(val, int):
+                return val
+        raise ValueError(f"Cannot find n_heads on {type(attn).__name__}")
+
+    def _get_n_kv_heads(self, attn: nn.Module) -> int:
+        for attr in ("n_kv_heads", "num_kv_heads", "num_key_value_heads"):
+            val = getattr(attn, attr, None)
+            if isinstance(val, int):
+                return val
+        return self._get_n_heads(attn)
+
+    def _get_scale(self, attn: nn.Module) -> float:
+        scale = getattr(attn, "scale", None)
+        if isinstance(scale, float):
+            return scale
+        n_heads = self._get_n_heads(attn)
+        head_dim = getattr(attn, "head_dim", None)
+        if isinstance(head_dim, int):
+            return head_dim**-0.5
+        return (1.0 / n_heads) ** 0.5
+
+
+def set_ring_prefill(model: nn.Module, is_prefill: bool) -> None:
+    """Toggle prefill mode on all RingAttentionLayer wrappers in the model."""
+    for layer in _ring_attention_layers(model):
+        layer.is_prefill = is_prefill
+
+
+def has_ring_attention_layer(model: nn.Module) -> bool:
+    """Check if the model contains any RingAttentionLayer wrappers."""
+    return any(_ring_attention_layers(model))
+
+
+def uses_ring_sequence_parallel_prefill(
+    model: nn.Module,
+    prompt_token_count: int,
+    group: mx.distributed.Group | None,
+) -> bool:
+    """Whether this prefill can be partitioned into non-empty rank blocks."""
+    return (
+        group is not None
+        and group.size() > 1
+        and prompt_token_count >= group.size()
+        and has_ring_attention_layer(model)
+    )
+
+
+def ring_prefill_block_size(model: nn.Module) -> int:
+    """Return the configured per-rank token block size for Ring prefill."""
+    block_sizes = {layer.sequence_block_size for layer in _ring_attention_layers(model)}
+    if len(block_sizes) != 1:
+        raise ValueError("Ring attention layers must share one sequence block size")
+    return block_sizes.pop()
+
+
+def _model_layers(model: nn.Module) -> list[_LayerCallable]:
+    """Find decoder layers on both flat and nested MLX model wrappers."""
+    try:
+        return get_layers(get_inner_model(model))
+    except ValueError:
+        layers = getattr(model, "layers", None)
+        if not isinstance(layers, list):
+            raise
+        return cast(list[_LayerCallable], layers)
+
+
+def _is_attention_layer(layer: object) -> bool:
+    """Heuristic to detect if a layer is an attention layer.
+
+    Checks for common attribute names used across MLX model implementations.
+    """
+    return any(hasattr(layer, attr) for attr in ("q_proj", "self_attn", "attn"))
+
+
+def _ring_attention_layers(model: nn.Module) -> list[RingAttentionLayer]:
+    """Find wrappers both as decoder layers and nested attention modules."""
+    wrappers: list[RingAttentionLayer] = []
+    for layer in _model_layers(model):
+        if isinstance(layer, RingAttentionLayer):
+            wrappers.append(layer)
+            continue
+        for attribute in ("self_attn", "attn"):
+            attention = getattr(layer, attribute, None)
+            if isinstance(attention, RingAttentionLayer):
+                wrappers.append(attention)
+    return wrappers
+
+
+def _attention_target(layer: _LayerCallable) -> tuple[nn.Module, str | None]:
+    """Return the attention module and its parent attribute, if nested."""
+    if getattr(layer, "q_proj", None) is not None:
+        return cast(nn.Module, layer), None
+    for attribute in ("self_attn", "attn"):
+        attention = getattr(layer, attribute, None)
+        if isinstance(attention, nn.Module):
+            return attention, attribute
+    raise ValueError(f"Cannot find an attention module on {type(layer).__name__}")
+
+
+def validate_ring_cache(cache_entries: Sequence[object]) -> None:
+    """Reject cache implementations whose semantics Ring prefill cannot preserve."""
+    unsupported_cache_types = {
+        type(entry).__name__
+        for entry in cache_entries
+        if not isinstance(entry, KVCache)
+    }
+    if unsupported_cache_types:
+        formatted_types = ", ".join(sorted(unsupported_cache_types))
+        raise TypeError(
+            "Ring attention supports only standard unquantized KVCache entries; "
+            f"received {formatted_types}"
+        )
+
+
+def _validate_ring_attention_layer(layer: _LayerCallable) -> None:
+    """Ensure the wrapper only replaces attention with known-equivalent semantics."""
+    attention, _ = _attention_target(layer)
+    missing_projections = [
+        attribute
+        for attribute in ("q_proj", "k_proj", "v_proj", "o_proj")
+        if getattr(attention, attribute, None) is None
+    ]
+    if missing_projections:
+        raise ValueError(
+            "Ring attention supports direct Q/K/V projection attention only; "
+            f"{type(attention).__name__} is missing {', '.join(missing_projections)}"
+        )
+
+    attention_type = (type(attention).__module__, type(attention).__name__)
+    if attention_type not in _SUPPORTED_ATTENTION_TYPES:
+        raise ValueError(
+            "Ring attention has only been verified for standard MLX Llama and "
+            f"Qwen3 attention; received {type(attention).__module__}."
+            f"{type(attention).__name__}"
+        )
+
+    unsupported_attributes = (
+        "use_sliding",
+        "use_sliding_window",
+        "is_sliding",
+        "is_sliding_window",
+        "sliding_window",
+        "window_size",
+        "attention_sinks",
+        "sinks",
+        "attn_logit_softcapping",
+    )
+    for owner in (layer, attention):
+        for attribute in unsupported_attributes:
+            value = getattr(owner, attribute, None)
+            if (
+                value is None
+                or value is False
+                or (isinstance(value, (int, float)) and value == 0)
+            ):
+                continue
+            raise ValueError(
+                f"Ring attention does not support {attribute} on {type(owner).__name__}"
+            )
+
+
+def _wrap_ring_attention_layer(
+    layer: _LayerCallable,
+    group: mx.distributed.Group,
+    sequence_block_size: int,
+    send_stream: mx.Stream,
+    receive_stream: mx.Stream,
+) -> _LayerCallable:
+    """Wrap only attention, never an entire decoder block with an MLP/residual."""
+    _validate_ring_attention_layer(layer)
+    attention, attribute = _attention_target(layer)
+    wrapped = RingAttentionLayer(
+        cast(_LayerCallable, attention),
+        group=group,
+        sequence_block_size=sequence_block_size,
+        send_stream=send_stream,
+        receive_stream=receive_stream,
+    )
+    if attribute is None:
+        return cast(_LayerCallable, cast(object, wrapped))
+    setattr(layer, attribute, wrapped)
+    return layer
+
+
+def ring_auto_parallel(
+    model: nn.Module,
+    group: mx.distributed.Group,
+    sequence_block_size: int = 512,
+) -> Generator[ModelLoadingResponse, None, nn.Module]:
+    """Set up ring-attention parallelism on a model.
+
+    Unlike pipeline (which splits layers) or tensor parallelism (which splits
+    weights), ring attention replicates all layers on all devices. Only the
+    attention layers are wrapped with :class:`RingAttentionLayer` to enable
+    sequence-parallel KV rotation.
+
+    MLP layers, normalisation, and embeddings run independently on each
+    device with no communication overhead — they process the local sequence
+    block without needing global context.
+    """
+    inner_model_instance = get_inner_model(model)
+    layers = get_layers(inner_model_instance)
+
+    total = len(layers)
+    wrapped_layers = 0
+    send_stream = mx.new_stream(mx.cpu)
+    receive_stream = mx.new_stream(mx.cpu)
+    for i, layer in enumerate(layers):
+        mx.eval(layer)
+        mx.clear_cache()
+
+        if _is_attention_layer(layer):
+            layers[i] = _wrap_ring_attention_layer(
+                layer,
+                group=group,
+                sequence_block_size=sequence_block_size,
+                send_stream=send_stream,
+                receive_stream=receive_stream,
+            )
+            wrapped_layers += 1
+
+        yield ModelLoadingResponse(layers_loaded=i, total=total)
+
+    if wrapped_layers == 0:
+        raise ValueError("Ring attention found no supported attention layers to wrap")
+
+    return patch_tensor_model(model)
diff --git a/src/exo/worker/engines/mlx/utils_mlx.py b/src/exo/worker/engines/mlx/utils_mlx.py
index 730abf64e3..5c6c14a89c 100644
--- a/src/exo/worker/engines/mlx/utils_mlx.py
+++ b/src/exo/worker/engines/mlx/utils_mlx.py
@@ -55,6 +55,7 @@
 from exo.shared.types.worker.shards import (
     CfgShardMetadata,
     PipelineShardMetadata,
+    RingShardMetadata,
     ShardMetadata,
     TensorShardMetadata,
 )
@@ -64,6 +65,7 @@
     pipeline_auto_parallel,
     tensor_auto_parallel,
 )
+from exo.worker.engines.mlx.ring_attention import ring_auto_parallel
 from exo.worker.engines.mlx.types import Model
 from exo.worker.runner.bootstrap import logger
 
@@ -268,6 +270,11 @@ def shard_and_load(
             logger.info(f"loading model from {model_path} with pipeline parallelism")
             model = yield from pipeline_auto_parallel(model, group, shard_metadata)
             mx.eval(model.parameters())
+        case RingShardMetadata():
+            logger.info(f"loading model from {model_path} with ring attention")
+            model = yield from ring_auto_parallel(
+                model, group, sequence_block_size=shard_metadata.sequence_block_size
+            )
         case CfgShardMetadata():
             raise ValueError(
                 "CfgShardMetadata is not supported for text model loading - "
diff --git a/src/exo/worker/runner/bootstrap.py b/src/exo/worker/runner/bootstrap.py
index f981667c0a..19d5b6cf9d 100644
--- a/src/exo/worker/runner/bootstrap.py
+++ b/src/exo/worker/runner/bootstrap.py
@@ -1,7 +1,10 @@
+import importlib.util
 import os
 import resource
+import sys
 import traceback
 from dataclasses import dataclass
+from pathlib import Path
 from typing import Self, cast
 
 import loguru
@@ -37,6 +40,30 @@ def __str__(self) -> str:
         return f"{self.exception_type}: {self.exception_message}\n{self.traceback}"
 
 
+def _ensure_cuda_home() -> None:
+    """Point MLX's CUDA backend at bundled CUDA headers when none are configured.
+
+    MLX on CUDA JIT-compiles kernels with NVRTC at runtime (the first
+    distributed send/recv triggers this) and fails with "Can not find
+    locations of CUDA headers" unless CUDA_HOME or CUDA_PATH is set. The pip
+    `nvidia-cuda-runtime` package ships the headers but nothing exports their
+    location, so resolve them from the `nvidia` namespace package.
+    """
+    if sys.platform != "linux":
+        return
+    if os.environ.get("CUDA_HOME") or os.environ.get("CUDA_PATH"):
+        return
+    specification = importlib.util.find_spec("nvidia")
+    if specification is None or not specification.submodule_search_locations:
+        return
+    for location in specification.submodule_search_locations:
+        candidate = Path(location) / "cuda_runtime"
+        if (candidate / "include").is_dir():
+            os.environ["CUDA_HOME"] = str(candidate)
+            logger.info(f"CUDA_HOME unset; using bundled CUDA headers at {candidate}")
+            return
+
+
 def entrypoint(
     bound_instance: BoundInstance,
     event_sender: MpSender[Event | RunnerTerminationError],
@@ -47,6 +74,8 @@ def entrypoint(
     global logger
     logger = _logger
 
+    _ensure_cuda_home()
+
     soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE)
     resource.setrlimit(resource.RLIMIT_NOFILE, (min(max(soft, 2048), hard), hard))
 
diff --git a/src/exo/worker/tests/unittests/test_mlx/test_has_ring_attention.py b/src/exo/worker/tests/unittests/test_mlx/test_has_ring_attention.py
new file mode 100644
index 0000000000..35cc561d10
--- /dev/null
+++ b/src/exo/worker/tests/unittests/test_mlx/test_has_ring_attention.py
@@ -0,0 +1,42 @@
+"""Tests for has_ring_attention_layer detection helper."""
+
+import mlx.core as mx
+import mlx.nn as nn
+
+from exo.worker.engines.mlx.ring_attention import (
+    RingAttentionLayer,
+    has_ring_attention_layer,
+)
+from exo.worker.tests.unittests.test_mlx.conftest import MockLayer
+
+
+class TestHasRingAttentionLayer:
+    def test_detects_ring_layer(self) -> None:
+        group = mx.distributed.init()
+        ring_layer = RingAttentionLayer(MockLayer(), group=group)
+
+        class FakeModel(nn.Module):
+            def __init__(self) -> None:
+                super().__init__()
+                self.layers = [ring_layer]
+
+        assert has_ring_attention_layer(FakeModel()) is True
+
+    def test_no_ring_layer_returns_false(self) -> None:
+        class FakeModel(nn.Module):
+            def __init__(self) -> None:
+                super().__init__()
+                self.layers = [MockLayer(), MockLayer()]
+
+        assert has_ring_attention_layer(FakeModel()) is False
+
+    def test_mixed_layers_detected(self) -> None:
+        group = mx.distributed.init()
+        ring_layer = RingAttentionLayer(MockLayer(), group=group)
+
+        class FakeModel(nn.Module):
+            def __init__(self) -> None:
+                super().__init__()
+                self.layers = [MockLayer(), ring_layer, MockLayer()]
+
+        assert has_ring_attention_layer(FakeModel()) is True
diff --git a/src/exo/worker/tests/unittests/test_mlx/test_ring_attention.py b/src/exo/worker/tests/unittests/test_mlx/test_ring_attention.py
new file mode 100644
index 0000000000..d85cf169fa
--- /dev/null
+++ b/src/exo/worker/tests/unittests/test_mlx/test_ring_attention.py
@@ -0,0 +1,969 @@
+# pyright: reportAny=false, reportPrivateUsage=false
+# pyright: reportUnknownVariableType=false, reportUnknownMemberType=false
+# pyright: reportUnknownArgumentType=false, reportArgumentType=false
+# pyright: reportAttributeAccessIssue=false, reportIndexIssue=false
+# pyright: reportCallIssue=false, reportOptionalSubscript=false
+
+"""Tests for RingAttentionLayer and ring-attention causal masking logic."""
+
+from __future__ import annotations
+
+import json
+import multiprocessing as mp
+import os
+import tempfile
+from copy import deepcopy
+from typing import Any
+
+import mlx.core as mx
+import mlx.nn as nn
+import pytest
+from mlx.utils import tree_map
+from mlx_lm.models import llama, qwen3
+from mlx_lm.models.cache import BatchKVCache, KVCache, QuantizedKVCache
+
+import exo.worker.engines.mlx.generator.generate as generate_module
+from exo.worker.engines.mlx.generator.generate import prefill
+from exo.worker.engines.mlx.ring_attention import (
+    RingAttentionLayer,
+    _is_attention_layer,
+    _make_block_causal_mask,
+    ring_auto_parallel,
+    set_ring_prefill,
+    validate_ring_cache,
+)
+from exo.worker.tests.unittests.test_mlx.conftest import MockLayer
+
+
+class TestBlockCausalMask:
+    def test_kv_entirely_before_returns_none(self) -> None:
+        result = _make_block_causal_mask(
+            local_seq_len=4, kv_seq_len=4, local_offset=4, kv_offset=0, n_heads=2
+        )
+        assert result is None
+
+    def test_kv_entirely_after_returns_all_false(self) -> None:
+        result = _make_block_causal_mask(
+            local_seq_len=4, kv_seq_len=4, local_offset=0, kv_offset=4, n_heads=2
+        )
+        assert result is not None
+        mx.eval(result)
+        assert result.shape == (2, 4, 4)
+        assert not bool(mx.any(result).item())
+
+    def test_same_block_causal_mask(self) -> None:
+        result = _make_block_causal_mask(
+            local_seq_len=4, kv_seq_len=4, local_offset=0, kv_offset=0, n_heads=2
+        )
+        assert result is not None
+        mx.eval(result)
+        mask = result.tolist()
+        for i in range(4):
+            for j in range(4):
+                assert mask[0][i][j] == (i >= j)
+
+    def test_overlapping_blocks(self) -> None:
+        result = _make_block_causal_mask(
+            local_seq_len=4, kv_seq_len=8, local_offset=4, kv_offset=0, n_heads=1
+        )
+        assert result is not None
+        mx.eval(result)
+        mask = result.tolist()
+        for i in range(4):
+            for j in range(8):
+                assert mask[0][i][j] == ((4 + i) >= j)
+
+    def test_mask_shape(self) -> None:
+        result = _make_block_causal_mask(
+            local_seq_len=8, kv_seq_len=6, local_offset=0, kv_offset=0, n_heads=4
+        )
+        assert result is not None
+        assert result.shape == (1, 8, 6)
+
+    def test_mask_dtype_is_bool(self) -> None:
+        result = _make_block_causal_mask(
+            local_seq_len=4, kv_seq_len=4, local_offset=0, kv_offset=0, n_heads=1
+        )
+        assert result is not None
+        assert result.dtype == mx.bool_
+
+
+class TestRingAttentionLayerAttributes:
+    def test_delegates_attributes(self) -> None:
+        mock = MockLayer()
+        group = mx.distributed.init()
+        wrapped = RingAttentionLayer(mock, group=group)
+        assert wrapped.custom_attr == "test_value"
+        assert wrapped.use_sliding is True
+
+    def test_rank_and_world_size(self) -> None:
+        mock = MockLayer()
+        group = mx.distributed.init()
+        wrapped = RingAttentionLayer(mock, group=group)
+        assert wrapped.rank == group.rank()
+        assert wrapped.world_size == group.size()
+
+    def test_is_prefill_default_false(self) -> None:
+        mock = MockLayer()
+        group = mx.distributed.init()
+        wrapped = RingAttentionLayer(mock, group=group)
+        assert wrapped.is_prefill is False
+
+    def test_missing_attribute_raises(self) -> None:
+        mock = MockLayer()
+        group = mx.distributed.init()
+        wrapped = RingAttentionLayer(mock, group=group)
+        with pytest.raises(AttributeError):
+            _ = wrapped.nonexistent_attr
+
+
+class TestRingAttentionLayerSingleDevice:
+    def test_single_device_delegates_to_original(self) -> None:
+        mock = MockLayer()
+        group = mx.distributed.init()
+        wrapped = RingAttentionLayer(mock, group=group)
+        x = mx.ones((1, 4))
+        result = wrapped(x)
+        mx.eval(result)
+        assert (result == 2.0).all()
+
+
+class TestOnlineAttentionMerge:
+    def test_block_merge_matches_full_causal_attention(self) -> None:
+        """Ring blocks must share one softmax denominator, not be summed."""
+        group = mx.distributed.init()
+        wrapped = RingAttentionLayer(MockLayer(), group=group)
+
+        queries = mx.array([[[[0.5, 1.0], [1.5, -0.5]]]])
+        first_keys = mx.array([[[[0.0, 1.0], [1.0, 0.0]]]])
+        first_values = mx.array([[[[2.0, 0.0], [0.0, 3.0]]]])
+        second_keys = mx.array([[[[1.0, 1.0], [-1.0, 0.5]]]])
+        second_values = mx.array([[[[1.0, 1.0], [4.0, -1.0]]]])
+
+        running_max, running_sum, running_output = wrapped._accumulate_attention(
+            queries,
+            first_keys,
+            first_values,
+            scale=0.5,
+            mask=None,
+            running_max=None,
+            running_sum=None,
+            running_output=None,
+        )
+        second_mask = mx.array([[[True, False], [True, True]]])
+        _, running_sum, running_output = wrapped._accumulate_attention(
+            queries,
+            second_keys,
+            second_values,
+            scale=0.5,
+            mask=second_mask,
+            running_max=running_max,
+            running_sum=running_sum,
+            running_output=running_output,
+        )
+
+        merged = running_output / running_sum
+        full_keys = mx.concatenate([first_keys, second_keys], axis=2)
+        full_values = mx.concatenate([first_values, second_values], axis=2)
+        full_mask = mx.array([[[True, True, True, False], [True, True, True, True]]])
+        expected = mx.fast.scaled_dot_product_attention(
+            queries, full_keys, full_values, scale=0.5, mask=full_mask
+        )
+
+        mx.eval(merged, expected)
+        assert mx.allclose(merged, expected, atol=1e-5).item()
+
+
+class TestRingPipelineScheduling:
+    def test_posts_communication_and_schedules_compute_before_waiting(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        class FakeGroup:
+            def rank(self) -> int:
+                return 0
+
+            def size(self) -> int:
+                return 2
+
+        class FakeAttention(nn.Module):
+            def __init__(self) -> None:
+                super().__init__()
+                self.q_proj = nn.Linear(8, 8)
+                self.k_proj = nn.Linear(8, 8)
+                self.v_proj = nn.Linear(8, 8)
+                self.o_proj = nn.Linear(8, 8)
+                self.n_heads = 2
+                self.n_kv_heads = 2
+                self.head_dim = 4
+                self.scale = 0.5
+
+        events: list[str] = []
+        array_kinds: dict[int, str] = {}
+
+        def fake_send(
+            value: mx.array,
+            destination: int,
+            *,
+            group: object,
+            stream: mx.Stream,
+        ) -> mx.array:
+            del destination, group, stream
+            result = value + 0
+            array_kinds[id(result)] = "send"
+            events.append("construct_send")
+            return result
+
+        def fake_receive(
+            template: mx.array,
+            source: int,
+            *,
+            group: object,
+            stream: mx.Stream,
+        ) -> mx.array:
+            del source, group, stream
+            result = template + 0
+            array_kinds[id(result)] = "receive"
+            events.append("construct_receive")
+            return result
+
+        def fake_async_eval(*arrays: object) -> None:
+            kinds = {
+                array_kinds.get(id(array))
+                for array in arrays
+                if isinstance(array, mx.array)
+            }
+            if kinds == {"send"}:
+                events.append("schedule_send")
+            elif kinds == {"receive"}:
+                events.append("schedule_receive")
+            else:
+                events.append("schedule_compute")
+
+        def fake_eval(*arrays: object) -> None:
+            kinds = {
+                array_kinds.get(id(array))
+                for array in arrays
+                if isinstance(array, mx.array)
+            }
+            events.append("wait_receive" if kinds == {"receive"} else "wait_send")
+
+        monkeypatch.setattr(mx.distributed, "send", fake_send)
+        monkeypatch.setattr(mx.distributed, "recv_like", fake_receive)
+        monkeypatch.setattr(mx, "async_eval", fake_async_eval)
+        monkeypatch.setattr(mx, "eval", fake_eval)
+
+        wrapped = RingAttentionLayer(FakeAttention(), FakeGroup())
+
+        def equal_block_lengths(local_sequence_length: int) -> list[int]:
+            del local_sequence_length
+            return [2, 2]
+
+        monkeypatch.setattr(wrapped, "_block_lengths", equal_block_lengths)
+        wrapped._prefill_step(mx.ones((1, 2, 8)), cache=KVCache())
+
+        first_compute = events.index("schedule_compute")
+        first_wait = events.index("wait_receive")
+        assert events.index("schedule_send") < first_compute
+        assert events.index("schedule_receive") < first_compute
+        assert first_compute < first_wait
+        assert events[-1] == "wait_send"
+
+
+class TestDecodePassthrough:
+    """Decode must forward to the wrapped attention unchanged, so every cache
+    type the model supports (including BatchKVCache) works during batched
+    generation on ring instances."""
+
+    class _TwoRankGroup:
+        def rank(self) -> int:
+            return 0
+
+        def size(self) -> int:
+            return 2
+
+    @staticmethod
+    def _tiny_llama_attention() -> nn.Module:
+        args = llama.ModelArgs(
+            model_type="llama",
+            hidden_size=16,
+            num_hidden_layers=1,
+            intermediate_size=32,
+            num_attention_heads=2,
+            num_key_value_heads=2,
+            rms_norm_eps=1e-5,
+            vocab_size=32,
+            max_position_embeddings=64,
+        )
+        return llama.Model(args).model.layers[0].self_attn
+
+    def test_decode_with_batch_kv_cache_matches_unwrapped(self) -> None:
+        attention = self._tiny_llama_attention()
+        wrapped = RingAttentionLayer(attention, self._TwoRankGroup())
+
+        wrapped_cache = BatchKVCache(left_padding=[1, 0])
+        unwrapped_cache = BatchKVCache(left_padding=[1, 0])
+
+        for step in range(3):
+            x = mx.arange(2 * 16, dtype=mx.float32).reshape(2, 1, 16) / (16 + step)
+            wrapped_out = wrapped(x, mask=None, cache=wrapped_cache)
+            unwrapped_out = attention(x, mask=None, cache=unwrapped_cache)
+            mx.eval(wrapped_out, unwrapped_out)
+            assert wrapped_out.shape == (2, 1, 16)
+            assert mx.array_equal(wrapped_out, unwrapped_out).item()
+
+    def test_decode_with_standard_kv_cache_matches_unwrapped(self) -> None:
+        attention = self._tiny_llama_attention()
+        wrapped = RingAttentionLayer(attention, self._TwoRankGroup())
+
+        wrapped_cache = KVCache()
+        unwrapped_cache = KVCache()
+
+        for step in range(3):
+            x = mx.arange(16, dtype=mx.float32).reshape(1, 1, 16) / (16 + step)
+            wrapped_out = wrapped(x, mask=None, cache=wrapped_cache)
+            unwrapped_out = attention(x, mask=None, cache=unwrapped_cache)
+            mx.eval(wrapped_out, unwrapped_out)
+            assert mx.array_equal(wrapped_out, unwrapped_out).item()
+
+
+class TestSetRingPrefill:
+    def test_set_ring_prefill_toggles(self) -> None:
+        mock = MockLayer()
+        group = mx.distributed.init()
+        wrapped = RingAttentionLayer(mock, group=group)
+
+        class FakeModel(nn.Module):
+            def __init__(self) -> None:
+                super().__init__()
+                self.layers = [wrapped]
+
+        model = FakeModel()
+        set_ring_prefill(model, is_prefill=True)
+        assert wrapped.is_prefill is True
+        set_ring_prefill(model, is_prefill=False)
+        assert wrapped.is_prefill is False
+
+    def test_set_ring_prefill_ignores_non_ring_layers(self) -> None:
+        mock = MockLayer()
+        group = mx.distributed.init()
+        ring_layer = RingAttentionLayer(mock, group=group)
+        plain_layer = MockLayer()
+
+        class FakeModel(nn.Module):
+            def __init__(self) -> None:
+                super().__init__()
+                self.layers = [ring_layer, plain_layer]
+
+        model = FakeModel()
+        set_ring_prefill(model, is_prefill=True)
+        assert ring_layer.is_prefill is True
+
+    def test_prefill_restores_ring_state_after_unexpected_error(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        wrapped = RingAttentionLayer(MockLayer(), group=mx.distributed.init())
+
+        class FakeModel(nn.Module):
+            def __init__(self) -> None:
+                super().__init__()
+                self.layers = [wrapped]
+
+        def fail_barrier(group: object) -> None:
+            del group
+            raise RuntimeError("simulated communication failure")
+
+        def force_ring_prefill(*args: object) -> bool:
+            del args
+            return True
+
+        def accept_cache(cache: object) -> None:
+            del cache
+
+        monkeypatch.setattr(
+            generate_module, "uses_ring_sequence_parallel_prefill", force_ring_prefill
+        )
+        monkeypatch.setattr(generate_module, "validate_ring_cache", accept_cache)
+        monkeypatch.setattr(generate_module, "mx_barrier", fail_barrier)
+
+        with pytest.raises(RuntimeError, match="communication failure"):
+            prefill(
+                FakeModel(),
+                None,
+                None,
+                mx.array([1, 2]),
+                [],
+                None,
+                None,
+                None,
+            )
+
+        assert wrapped.is_prefill is False
+
+
+class TestIsAttentionLayer:
+    def test_detects_q_proj(self) -> None:
+        class FakeAttn(nn.Module):
+            def __init__(self) -> None:
+                super().__init__()
+                self.q_proj = nn.Linear(8, 8)
+                self.k_proj = nn.Linear(8, 8)
+                self.v_proj = nn.Linear(8, 8)
+                self.o_proj = nn.Linear(8, 8)
+                self.n_heads = 2
+                self.n_kv_heads = 2
+                self.head_dim = 4
+
+        assert _is_attention_layer(FakeAttn()) is True
+
+    def test_detects_self_attn(self) -> None:
+        class FakeDecoder(nn.Module):
+            def __init__(self) -> None:
+                super().__init__()
+                self.self_attn = nn.Linear(8, 8)
+
+        assert _is_attention_layer(FakeDecoder()) is True
+
+    def test_detects_attn(self) -> None:
+        class FakeLayer(nn.Module):
+            def __init__(self) -> None:
+                super().__init__()
+                self.attn = nn.Linear(8, 8)
+
+        assert _is_attention_layer(FakeLayer()) is True
+
+    def test_rejects_plain_mlp(self) -> None:
+        class FakeMLP(nn.Module):
+            def __init__(self) -> None:
+                super().__init__()
+                self.gate_proj = nn.Linear(8, 8)
+                self.down_proj = nn.Linear(8, 8)
+
+        assert _is_attention_layer(FakeMLP()) is False
+
+    def test_rejects_empty_module(self) -> None:
+        class Empty(nn.Module):
+            def __init__(self) -> None:
+                super().__init__()
+
+        assert _is_attention_layer(Empty()) is False
+
+
+class TestRingAutoParallel:
+    def test_wraps_attention_layers(self) -> None:
+        args = llama.ModelArgs(
+            model_type="llama",
+            hidden_size=16,
+            num_hidden_layers=2,
+            intermediate_size=32,
+            num_attention_heads=2,
+            num_key_value_heads=2,
+            rms_norm_eps=1e-5,
+            vocab_size=32,
+            max_position_embeddings=64,
+        )
+        model = llama.Model(args)
+        group = mx.distributed.init()
+        responses = list(ring_auto_parallel(model, group, sequence_block_size=64))
+        assert len(responses) == 2
+        assert all(r.total == 2 for r in responses)
+        assert [r.layers_loaded for r in responses] == [0, 1]
+        inner = model.model
+        assert isinstance(inner.layers[0].self_attn, RingAttentionLayer)
+        assert inner.layers[0].self_attn.sequence_block_size == 64
+        assert isinstance(inner.layers[1].self_attn, RingAttentionLayer)
+
+    def test_preserves_non_attention_layers(self) -> None:
+        class FakeMLP(nn.Module):
+            def __init__(self) -> None:
+                super().__init__()
+                self.gate_proj = nn.Linear(8, 8)
+
+            def __call__(self, x: mx.array, *a: object, **k: object) -> mx.array:
+                return x
+
+        class FakeInner(nn.Module):
+            def __init__(self) -> None:
+                super().__init__()
+                self.layers = [FakeMLP(), FakeMLP()]
+
+        class FakeModel(nn.Module):
+            def __init__(self) -> None:
+                super().__init__()
+                self.model = FakeInner()
+
+            def __call__(self, x: mx.array, *a: object, **k: object) -> mx.array:
+                return x
+
+        model = FakeModel()
+        group = mx.distributed.init()
+        with pytest.raises(ValueError, match="no supported attention layers"):
+            list(ring_auto_parallel(model, group))
+
+
+class TestRingCompatibility:
+    def test_rejects_attention_without_direct_kv_projections(self) -> None:
+        class UnsupportedAttention(nn.Module):
+            def __init__(self) -> None:
+                super().__init__()
+                self.q_proj = nn.Linear(8, 8)
+
+        class FakeModel(nn.Module):
+            def __init__(self) -> None:
+                super().__init__()
+                self.model = nn.Module()
+                self.model.layers = [UnsupportedAttention()]
+
+        with pytest.raises(ValueError, match="direct Q/K/V projection"):
+            list(ring_auto_parallel(FakeModel(), mx.distributed.init()))
+
+    def test_rejects_unverified_direct_projection_attention(self) -> None:
+        class CustomAttention(nn.Module):
+            def __init__(self) -> None:
+                super().__init__()
+                self.q_proj = nn.Linear(8, 8)
+                self.k_proj = nn.Linear(8, 8)
+                self.v_proj = nn.Linear(8, 8)
+                self.o_proj = nn.Linear(8, 8)
+
+        class FakeModel(nn.Module):
+            def __init__(self) -> None:
+                super().__init__()
+                self.model = nn.Module()
+                self.model.layers = [CustomAttention()]
+
+        with pytest.raises(ValueError, match="only been verified"):
+            list(ring_auto_parallel(FakeModel(), mx.distributed.init()))
+
+    def test_rejects_parent_level_sliding_attention(self) -> None:
+        args = llama.ModelArgs(
+            model_type="llama",
+            hidden_size=16,
+            num_hidden_layers=1,
+            intermediate_size=32,
+            num_attention_heads=2,
+            num_key_value_heads=2,
+            rms_norm_eps=1e-5,
+            vocab_size=32,
+            max_position_embeddings=64,
+            sliding_window=16,
+        )
+        model = llama.Model(args)
+        model.model.layers[0].use_sliding = True
+
+        with pytest.raises(ValueError, match="use_sliding"):
+            list(ring_auto_parallel(model, mx.distributed.init()))
+
+    def test_rejects_quantized_kv_cache(self) -> None:
+        with pytest.raises(TypeError, match="unquantized KVCache"):
+            validate_ring_cache([QuantizedKVCache()])
+
+    def test_qwen_qk_norm_matches_native_attention_preparation(self) -> None:
+        args = qwen3.ModelArgs(
+            model_type="qwen3",
+            hidden_size=16,
+            num_hidden_layers=1,
+            intermediate_size=32,
+            num_attention_heads=2,
+            num_key_value_heads=2,
+            rms_norm_eps=1e-5,
+            vocab_size=32,
+            max_position_embeddings=64,
+            rope_theta=10000,
+            head_dim=8,
+            tie_word_embeddings=True,
+        )
+        attention = qwen3.Attention(args)
+        wrapped = RingAttentionLayer(attention, mx.distributed.init())
+        x = mx.arange(64, dtype=mx.float32).reshape(1, 4, 16) / 64
+
+        actual_queries, actual_keys, _, _, _, _, _ = wrapped._project_qkv(attention, x)
+        expected_queries = attention.q_norm(
+            attention.q_proj(x).reshape(1, 4, attention.n_heads, -1)
+        ).transpose(0, 2, 1, 3)
+        expected_keys = attention.k_norm(
+            attention.k_proj(x).reshape(1, 4, attention.n_kv_heads, -1)
+        ).transpose(0, 2, 1, 3)
+
+        mx.eval(actual_queries, actual_keys, expected_queries, expected_keys)
+        assert mx.allclose(actual_queries, expected_queries).item()
+        assert mx.allclose(actual_keys, expected_keys).item()
+
+
+def _run_ring_attention_device(
+    rank: int, world_size: int, hostfile_path: str, result_queue: Any
+) -> None:
+    os.environ["MLX_HOSTFILE"] = hostfile_path
+    os.environ["MLX_RANK"] = str(rank)
+    try:
+        group = mx.distributed.init(backend="ring", strict=True)
+
+        class MockAttn(nn.Module):
+            def __init__(self) -> None:
+                super().__init__()
+                self.q_proj = nn.Linear(8, 8)
+                self.k_proj = nn.Linear(8, 8)
+                self.v_proj = nn.Linear(8, 8)
+                self.o_proj = nn.Linear(8, 8)
+                self.n_heads = 2
+                self.head_dim = 4
+                self.scale = 0.5
+                for projection in (
+                    self.q_proj,
+                    self.k_proj,
+                    self.v_proj,
+                    self.o_proj,
+                ):
+                    projection.weight = mx.arange(64, dtype=mx.float32).reshape(8, 8)
+                    projection.bias = mx.zeros(8)
+
+            def __call__(
+                self, x: mx.array, *args: object, **kwargs: object
+            ) -> mx.array:
+                q = self.q_proj(x)
+                keys = self.k_proj(x)
+                v = self.v_proj(x)
+                batch, seq, dim = q.shape
+                q = q.reshape(batch, seq, self.n_heads, -1).transpose(0, 2, 1, 3)
+                keys = keys.reshape(batch, seq, self.n_heads, -1).transpose(0, 2, 1, 3)
+                v = v.reshape(batch, seq, self.n_heads, -1).transpose(0, 2, 1, 3)
+                scores = q @ keys.transpose(0, 1, 3, 2) * self.scale
+                attn = mx.softmax(scores, axis=-1)
+                out = attn @ v
+                out = out.transpose(0, 2, 1, 3).reshape(batch, seq, dim)
+                return self.o_proj(out)
+
+        mock_attn = MockAttn()
+        wrapped = RingAttentionLayer(mock_attn, group=group)
+        wrapped.is_prefill = True
+        x = (
+            mx.arange(rank * 32, (rank + 1) * 32, dtype=mx.float32).reshape(1, 4, 8)
+            / 64
+        )
+        result = wrapped(x)
+
+        total_sequence_length = world_size * 4
+        full_x = (
+            mx.arange(total_sequence_length * 8, dtype=mx.float32).reshape(
+                1, total_sequence_length, 8
+            )
+            / 64
+        )
+        queries = (
+            mock_attn.q_proj(full_x)
+            .reshape(1, total_sequence_length, 2, 4)
+            .transpose(0, 2, 1, 3)
+        )
+        keys = (
+            mock_attn.k_proj(full_x)
+            .reshape(1, total_sequence_length, 2, 4)
+            .transpose(0, 2, 1, 3)
+        )
+        values = (
+            mock_attn.v_proj(full_x)
+            .reshape(1, total_sequence_length, 2, 4)
+            .transpose(0, 2, 1, 3)
+        )
+        causal_mask = mx.tril(
+            mx.ones((total_sequence_length, total_sequence_length), dtype=mx.bool_)
+        )[None, :, :]
+        expected = mx.fast.scaled_dot_product_attention(
+            queries, keys, values, scale=mock_attn.scale, mask=causal_mask
+        )
+        expected = expected[:, :, rank * 4 : (rank + 1) * 4, :]
+        expected = expected.transpose(0, 2, 1, 3).reshape(1, 4, 8)
+        expected = mock_attn.o_proj(expected)
+        mx.eval(result)
+        mx.eval(expected)
+        result_queue.put(
+            (rank, True, (result.shape, mx.max(mx.abs(result - expected)).item()))
+        )
+    except Exception as e:
+        import traceback
+
+        result_queue.put((rank, False, f"{e}\n{traceback.format_exc()}"))
+
+
+def _run_ring_prefill_device(
+    rank: int, world_size: int, hostfile_path: str, result_queue: Any
+) -> None:
+    os.environ["MLX_HOSTFILE"] = hostfile_path
+    os.environ["MLX_RANK"] = str(rank)
+    try:
+        group = mx.distributed.init(backend="ring", strict=True)
+        mx.random.seed(42)
+
+        class MockAttn(nn.Module):
+            def __init__(self) -> None:
+                super().__init__()
+                self.q_proj = nn.Linear(8, 8)
+                self.k_proj = nn.Linear(8, 8)
+                self.v_proj = nn.Linear(8, 8)
+                self.o_proj = nn.Linear(8, 8)
+                self.n_heads = 2
+                self.head_dim = 4
+                self.scale = 0.5
+                for projection in (
+                    self.q_proj,
+                    self.k_proj,
+                    self.v_proj,
+                    self.o_proj,
+                ):
+                    projection.weight = mx.arange(64, dtype=mx.float32).reshape(8, 8)
+                    projection.bias = mx.zeros(8)
+
+        class FakeInner(nn.Module):
+            def __init__(self) -> None:
+                super().__init__()
+                self.layers = [
+                    RingAttentionLayer(MockAttn(), group, sequence_block_size=2)
+                ]
+
+        class FakeModel(nn.Module):
+            def __init__(self) -> None:
+                super().__init__()
+                self.model = FakeInner()
+                self.layers = self.model.layers
+
+            def __call__(self, tokens: mx.array, cache: object) -> mx.array:
+                embeddings = mx.eye(8)[tokens]
+                return self.model.layers[0](embeddings, cache=cache[0])
+
+        model = FakeModel()
+        cache = [KVCache()]
+        prompt = mx.array(list(range(8)))
+        prefill(
+            model,
+            None,
+            None,
+            prompt,
+            cache,
+            group,
+            None,
+            None,
+        )
+        full_embeddings = mx.eye(8)[prompt[None]]
+        original = model.layers[0].original_layer
+        expected_keys = original.k_proj(full_embeddings)
+        expected_keys = expected_keys.reshape(1, 8, 2, 4).transpose(0, 2, 1, 3)
+        actual_keys = cache[0].keys[..., : cache[0].offset, :]
+        mx.eval(expected_keys, actual_keys)
+        result_queue.put(
+            (
+                rank,
+                True,
+                (
+                    cache[0].offset,
+                    mx.allclose(actual_keys, expected_keys).item(),
+                    mx.max(mx.abs(actual_keys - expected_keys)).item(),
+                ),
+            )
+        )
+    except Exception as e:
+        import traceback
+
+        result_queue.put((rank, False, f"{e}\n{traceback.format_exc()}"))
+
+
+def _run_llama_ring_device(
+    rank: int, world_size: int, hostfile_path: str, result_queue: Any
+) -> None:
+    """Compare complete two-rank Ring inference against a real tiny Llama model."""
+    os.environ["MLX_HOSTFILE"] = hostfile_path
+    os.environ["MLX_RANK"] = str(rank)
+    try:
+        group = mx.distributed.init(backend="ring", strict=True)
+        mx.random.seed(42)
+        args = llama.ModelArgs(
+            model_type="llama",
+            hidden_size=16,
+            num_hidden_layers=1,
+            intermediate_size=32,
+            num_attention_heads=2,
+            num_key_value_heads=2,
+            rms_norm_eps=1e-5,
+            vocab_size=32,
+            max_position_embeddings=64,
+        )
+        baseline = llama.Model(args)
+
+        def constant_parameter(parameter: mx.array) -> mx.array:
+            return mx.full(parameter.shape, 0.01, dtype=parameter.dtype)
+
+        baseline.update(tree_map(constant_parameter, baseline.parameters()))
+        ring_model = deepcopy(baseline)
+        list(ring_auto_parallel(ring_model, group, sequence_block_size=2))
+
+        prompt = mx.array([1, 2, 3, 4, 5, 6, 7, 8])
+        baseline_cache = baseline.make_cache()
+        expected_prefill = baseline(prompt[None], cache=baseline_cache)
+
+        ring_cache = ring_model.make_cache()
+        set_ring_prefill(ring_model, is_prefill=True)
+        start = (len(prompt) * rank) // world_size
+        end = (len(prompt) * (rank + 1)) // world_size
+        actual_prefill = ring_model(prompt[start:end][None], cache=ring_cache)
+        set_ring_prefill(ring_model, is_prefill=False)
+
+        expected_prefill = expected_prefill[:, start:end, :]
+        next_token = mx.array([[9]])
+        expected_decode = baseline(next_token, cache=baseline_cache)
+        actual_decode = ring_model(next_token, cache=ring_cache)
+        mx.eval(actual_prefill, expected_prefill, actual_decode, expected_decode)
+
+        cache_offsets = [entry.offset for entry in ring_cache]
+        result_queue.put(
+            (
+                rank,
+                True,
+                (
+                    mx.max(mx.abs(actual_prefill - expected_prefill)).item(),
+                    mx.max(mx.abs(actual_decode - expected_decode)).item(),
+                    cache_offsets,
+                ),
+            )
+        )
+    except Exception as e:
+        import traceback
+
+        result_queue.put((rank, False, f"{e}\n{traceback.format_exc()}"))
+
+
+class TestRingAttentionDistributed:
+    @pytest.mark.parametrize("world_size", [2, 3])
+    def test_multi_device_ring_attention_completes(self, world_size: int) -> None:
+        ctx = mp.get_context("spawn")
+        starting_port = 29700 + world_size * 10
+        hosts = [f"127.0.0.1:{starting_port + i}" for i in range(world_size)]
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
+            json.dump(hosts, f)
+            hostfile_path = f.name
+        processes: list[Any] = []
+        try:
+            result_queue: Any = ctx.Queue()
+            for rank in range(world_size):
+                p = ctx.Process(
+                    target=_run_ring_attention_device,
+                    args=(rank, world_size, hostfile_path, result_queue),
+                )
+                p.start()
+                processes.append(p)
+            for p in processes:
+                p.join(timeout=30)
+            results: dict[int, Any] = {}
+            errors: dict[int, str] = {}
+            while not result_queue.empty():
+                rank, success, value = result_queue.get()
+                if success:
+                    results[rank] = value
+                else:
+                    errors[rank] = value
+            assert len(results) == world_size, f"Errors: {errors}"
+            for rank in range(world_size):
+                assert rank in results, f"Device {rank} failed: {errors.get(rank)}"
+                shape, maximum_error = results[rank]
+                assert shape == (1, 4, 8)
+                assert maximum_error < 1e-5
+        finally:
+            for process in processes:
+                if process.is_alive():
+                    process.terminate()
+                    process.join(timeout=5)
+            os.unlink(hostfile_path)
+
+    def test_single_device_matches_original(self) -> None:
+        group = mx.distributed.init()
+
+        class MockAttn(nn.Module):
+            def __init__(self) -> None:
+                super().__init__()
+                self.q_proj = nn.Linear(8, 8)
+                self.k_proj = nn.Linear(8, 8)
+                self.v_proj = nn.Linear(8, 8)
+                self.o_proj = nn.Linear(8, 8)
+                self.n_heads = 2
+                self.head_dim = 4
+                self.scale = 0.5
+
+            def __call__(self, x: mx.array, *a: object, **k: object) -> mx.array:
+                return self.o_proj(x * 3)
+
+        mock_attn = MockAttn()
+        wrapped = RingAttentionLayer(mock_attn, group=group)
+        x = mx.ones((1, 4, 8))
+        result = wrapped(x)
+        mx.eval(result)
+        expected = mock_attn(mx.ones((1, 4, 8)))
+        mx.eval(expected)
+        assert mx.allclose(result, expected).item()
+
+    def test_prefill_partitions_prompt_and_populates_full_cache(self) -> None:
+        ctx = mp.get_context("spawn")
+        world_size = 2
+        hosts = [f"127.0.0.1:{29800 + i}" for i in range(world_size)]
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
+            json.dump(hosts, f)
+            hostfile_path = f.name
+        try:
+            result_queue: Any = ctx.Queue()
+            processes: list[Any] = []
+            for rank in range(world_size):
+                process = ctx.Process(
+                    target=_run_ring_prefill_device,
+                    args=(rank, world_size, hostfile_path, result_queue),
+                )
+                process.start()
+                processes.append(process)
+            for process in processes:
+                process.join(timeout=30)
+
+            results: dict[int, Any] = {}
+            errors: dict[int, str] = {}
+            while not result_queue.empty():
+                rank, success, value = result_queue.get()
+                if success:
+                    results[rank] = value
+                else:
+                    errors[rank] = value
+
+            assert results == {0: (8, True, 0.0), 1: (8, True, 0.0)}, (
+                f"Errors: {errors}"
+            )
+        finally:
+            os.unlink(hostfile_path)
+
+    def test_tiny_llama_prefill_and_decode_match_single_device(self) -> None:
+        ctx = mp.get_context("spawn")
+        world_size = 2
+        hosts = [f"127.0.0.1:{29900 + i}" for i in range(world_size)]
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
+            json.dump(hosts, f)
+            hostfile_path = f.name
+        try:
+            result_queue: Any = ctx.Queue()
+            processes: list[Any] = []
+            for rank in range(world_size):
+                process = ctx.Process(
+                    target=_run_llama_ring_device,
+                    args=(rank, world_size, hostfile_path, result_queue),
+                )
+                process.start()
+                processes.append(process)
+            for process in processes:
+                process.join(timeout=30)
+
+            results: dict[int, Any] = {}
+            errors: dict[int, str] = {}
+            while not result_queue.empty():
+                rank, success, value = result_queue.get()
+                if success:
+                    results[rank] = value
+                else:
+                    errors[rank] = value
+
+            assert len(results) == world_size, f"Errors: {errors}"
+            for prefill_error, decode_error, cache_offsets in results.values():
+                assert prefill_error < 1e-5
+                assert decode_error < 1e-5
+                assert cache_offsets == [9]
+        finally:
+            os.unlink(hostfile_path)
diff --git a/src/exo/worker/tests/unittests/test_mlx/test_ring_shard_metadata.py b/src/exo/worker/tests/unittests/test_mlx/test_ring_shard_metadata.py
new file mode 100644
index 0000000000..efa2446809
--- /dev/null
+++ b/src/exo/worker/tests/unittests/test_mlx/test_ring_shard_metadata.py
@@ -0,0 +1,174 @@
+"""Tests for RingShardMetadata and the Ring sharding enum."""
+
+import pytest
+from pydantic import ValidationError
+
+from exo.shared.models.model_cards import ModelCard, ModelTask
+from exo.shared.types.backends import Backend
+from exo.shared.types.common import ModelId
+from exo.shared.types.memory import Memory
+from exo.shared.types.worker.shards import (
+    RingShardMetadata,
+    Sharding,
+)
+
+
+def _make_model_card() -> ModelCard:
+    return ModelCard(
+        model_id=ModelId("test/ring-model"),
+        storage_size=Memory.from_gb(1),
+        n_layers=4,
+        hidden_size=128,
+        supports_tensor=False,
+        tasks=[ModelTask.TextGeneration],
+        backends=[Backend.MlxMetal],
+    )
+
+
+class TestShardingEnum:
+    def test_ring_exists(self) -> None:
+        assert Sharding.Ring == "Ring"
+
+    def test_ring_is_distinct_from_others(self) -> None:
+        assert Sharding.Ring != Sharding.Tensor
+        assert Sharding.Ring != Sharding.Pipeline
+
+    def test_all_sharding_values(self) -> None:
+        values = {Sharding.Tensor, Sharding.Pipeline, Sharding.Ring}
+        assert len(values) == 3
+
+
+class TestRingShardMetadata:
+    def test_basic_construction(self) -> None:
+        meta = RingShardMetadata(
+            model_card=_make_model_card(),
+            device_rank=0,
+            world_size=2,
+            start_layer=0,
+            end_layer=4,
+            n_layers=4,
+        )
+        assert meta.device_rank == 0
+        assert meta.world_size == 2
+        assert meta.sequence_block_size == 512
+
+    def test_custom_block_size(self) -> None:
+        meta = RingShardMetadata(
+            model_card=_make_model_card(),
+            device_rank=1,
+            world_size=3,
+            start_layer=0,
+            end_layer=4,
+            n_layers=4,
+            sequence_block_size=1024,
+        )
+        assert meta.sequence_block_size == 1024
+
+    def test_block_size_must_be_positive(self) -> None:
+        with pytest.raises(ValidationError):
+            RingShardMetadata(
+                model_card=_make_model_card(),
+                device_rank=0,
+                world_size=2,
+                start_layer=0,
+                end_layer=4,
+                n_layers=4,
+                sequence_block_size=0,
+            )
+
+    def test_block_size_must_be_positive_int(self) -> None:
+        with pytest.raises(ValidationError):
+            RingShardMetadata(
+                model_card=_make_model_card(),
+                device_rank=0,
+                world_size=2,
+                start_layer=0,
+                end_layer=4,
+                n_layers=4,
+                sequence_block_size=-1,
+            )
+
+    def test_is_first_layer(self) -> None:
+        meta = RingShardMetadata(
+            model_card=_make_model_card(),
+            device_rank=0,
+            world_size=2,
+            start_layer=0,
+            end_layer=4,
+            n_layers=4,
+        )
+        assert meta.is_first_layer is True
+
+    def test_is_last_layer(self) -> None:
+        meta = RingShardMetadata(
+            model_card=_make_model_card(),
+            device_rank=0,
+            world_size=2,
+            start_layer=0,
+            end_layer=4,
+            n_layers=4,
+        )
+        assert meta.is_last_layer is True
+
+    def test_hash_is_deterministic(self) -> None:
+        meta1 = RingShardMetadata(
+            model_card=_make_model_card(),
+            device_rank=0,
+            world_size=2,
+            start_layer=0,
+            end_layer=4,
+            n_layers=4,
+        )
+        meta2 = RingShardMetadata(
+            model_card=_make_model_card(),
+            device_rank=0,
+            world_size=2,
+            start_layer=0,
+            end_layer=4,
+            n_layers=4,
+        )
+        assert hash(meta1) == hash(meta2)
+
+    def test_hash_differs_by_rank(self) -> None:
+        meta0 = RingShardMetadata(
+            model_card=_make_model_card(),
+            device_rank=0,
+            world_size=2,
+            start_layer=0,
+            end_layer=4,
+            n_layers=4,
+        )
+        meta1 = RingShardMetadata(
+            model_card=_make_model_card(),
+            device_rank=1,
+            world_size=2,
+            start_layer=0,
+            end_layer=4,
+            n_layers=4,
+        )
+        assert hash(meta0) != hash(meta1)
+
+    def test_frozen(self) -> None:
+        meta = RingShardMetadata(
+            model_card=_make_model_card(),
+            device_rank=0,
+            world_size=2,
+            start_layer=0,
+            end_layer=4,
+            n_layers=4,
+        )
+        with pytest.raises(ValidationError):
+            meta.sequence_block_size = 256
+
+    def test_isinstance_in_shard_metadata_union(self) -> None:
+        from exo.shared.types.worker.shards import ShardMetadata
+
+        meta = RingShardMetadata(
+            model_card=_make_model_card(),
+            device_rank=0,
+            world_size=2,
+            start_layer=0,
+            end_layer=4,
+            n_layers=4,
+        )
+        assert isinstance(meta, ShardMetadata)
diff --git a/src/exo/worker/tests/unittests/test_runner/test_cuda_home.py b/src/exo/worker/tests/unittests/test_runner/test_cuda_home.py
new file mode 100644
index 0000000000..2914c5746f
--- /dev/null
+++ b/src/exo/worker/tests/unittests/test_runner/test_cuda_home.py
@@ -0,0 +1,66 @@
+import importlib.machinery
+from pathlib import Path
+
+import pytest
+
+import exo.worker.runner.bootstrap as bootstrap
+from exo.worker.runner.bootstrap import (
+    _ensure_cuda_home,  # pyright: ignore[reportPrivateUsage]
+)
+
+
+@pytest.fixture
+def cuda_env(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv("CUDA_HOME", raising=False)
+    monkeypatch.delenv("CUDA_PATH", raising=False)
+
+
+def _fake_nvidia_spec(base: Path) -> importlib.machinery.ModuleSpec:
+    spec = importlib.machinery.ModuleSpec("nvidia", None, is_package=True)
+    spec.submodule_search_locations = [str(base)]
+    return spec
+
+
+def test_noop_off_linux(
+    cuda_env: None, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    monkeypatch.setattr(bootstrap.sys, "platform", "darwin")
+    _ensure_cuda_home()
+    assert "CUDA_HOME" not in bootstrap.os.environ
+
+
+def test_respects_existing_configuration(
+    cuda_env: None, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(bootstrap.sys, "platform", "linux")
+    monkeypatch.setenv("CUDA_PATH", "/opt/cuda")
+    _ensure_cuda_home()
+    assert "CUDA_HOME" not in bootstrap.os.environ
+
+
+def test_sets_cuda_home_from_nvidia_package(
+    cuda_env: None, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    include_dir = tmp_path / "cuda_runtime" / "include"
+    include_dir.mkdir(parents=True)
+
+    def fake_find_spec(name: str) -> importlib.machinery.ModuleSpec | None:
+        return _fake_nvidia_spec(tmp_path) if name == "nvidia" else None
+
+    monkeypatch.setattr(bootstrap.sys, "platform", "linux")
+    monkeypatch.setattr(bootstrap.importlib.util, "find_spec", fake_find_spec)
+    _ensure_cuda_home()
+    assert bootstrap.os.environ["CUDA_HOME"] == str(tmp_path / "cuda_runtime")
+    monkeypatch.delenv("CUDA_HOME")
+
+
+def test_noop_without_headers(
+    cuda_env: None, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    def fake_find_spec(name: str) -> importlib.machinery.ModuleSpec | None:
+        return _fake_nvidia_spec(tmp_path) if name == "nvidia" else None
+
+    monkeypatch.setattr(bootstrap.sys, "platform", "linux")
+    monkeypatch.setattr(bootstrap.importlib.util, "find_spec", fake_find_spec)
+    _ensure_cuda_home()
+    assert "CUDA_HOME" not in bootstrap.os.environ
diff --git a/tools/src/exo_tools/harness.py b/tools/src/exo_tools/harness.py
index a80da62545..91d0319ef5 100644
--- a/tools/src/exo_tools/harness.py
+++ b/tools/src/exo_tools/harness.py
@@ -501,7 +501,9 @@ def add_common_instance_args(ap: argparse.ArgumentParser) -> None:
         "--instance-meta", choices=["ring", "jaccl", "both"], default="both"
     )
     ap.add_argument(
-        "--sharding", choices=["pipeline", "tensor", "both"], default="both"
+        "--sharding",
+        choices=["pipeline", "tensor", "ring", "both"],
+        default="both",
     )
     ap.add_argument(
         "--skip-pipeline-jaccl",