api cancellation

closing the http request to the api now
- sends a cancellation from the api
- writes that canellation in the master
- worker plans off the cancellation
- runner observes that cancellation after every generation step (+1
communication per token)
- cancellation happens synchronously to prevent gpu locks

.mlx_typings/mlx_lm/models/step3p5.pyi

@@ -1,151 +0,0 @@
-from dataclasses import dataclass
-from typing import Any, Dict, List, Optional
-
-import mlx.core as mx
-import mlx.nn as nn
-
-from .base import BaseModelArgs
-from .switch_layers import SwitchGLU
-
-@dataclass
-class ModelArgs(BaseModelArgs):
-    model_type: str
-    hidden_size: int
-    num_hidden_layers: int
-    vocab_size: int
-    num_attention_heads: int
-    num_attention_groups: int
-    head_dim: int
-    intermediate_size: int
-    rms_norm_eps: float
-    rope_theta: float
-    rope_scaling: Optional[Dict[str, Any]]
-    max_position_embeddings: int
-    sliding_window: int
-    layer_types: Optional[List[str]]
-    yarn_only_types: Optional[List[str]]
-    partial_rotary_factors: Optional[List[float]]
-    attention_other_setting: Optional[Dict[str, Any]]
-    use_head_wise_attn_gate: bool
-    moe_num_experts: int
-    moe_top_k: int
-    moe_intermediate_size: int
-    share_expert_dim: int
-    moe_layers_enum: Optional[str]
-    moe_router_scaling_factor: float
-    norm_expert_weight: bool
-    swiglu_limits: Optional[List[float]]
-    swiglu_limits_shared: Optional[List[float]]
-    tie_word_embeddings: bool
-
-class Step3p5MLP(nn.Module):
-    hidden_size: int
-    intermediate_size: int
-    gate_proj: nn.Linear
-    up_proj: nn.Linear
-    down_proj: nn.Linear
-    limit: Optional[float]
-
-    def __init__(
-        self, args: ModelArgs, intermediate_size: int, swiglu_limit: float = 0
-    ) -> None: ...
-    def __call__(self, x: mx.array) -> mx.array: ...
-
-class Step3p5MoEGate(nn.Module):
-    top_k: int
-    n_routed_experts: int
-    routed_scaling_factor: float
-    norm_topk_prob: bool
-    gate: nn.Linear
-    router_bias: mx.array
-
-    def __init__(self, args: ModelArgs) -> None: ...
-    def __call__(self, x: mx.array) -> tuple[mx.array, mx.array]: ...
-
-class Step3p5MoE(nn.Module):
-    gate: Step3p5MoEGate
-    switch_mlp: SwitchGLU
-    share_expert: Step3p5MLP
-    sharding_group: Optional[mx.distributed.Group]
-
-    def __init__(self, args: ModelArgs, layer_idx: int) -> None: ...
-    def __call__(self, x: mx.array) -> mx.array: ...
-
-class Step3p5Attention(nn.Module):
-    is_sliding: bool
-    num_heads: int
-    num_kv_heads: int
-    head_dim: int
-    scale: float
-    q_proj: nn.Linear
-    k_proj: nn.Linear
-    v_proj: nn.Linear
-    o_proj: nn.Linear
-    q_norm: nn.Module
-    k_norm: nn.Module
-    use_head_wise_attn_gate: bool
-    g_proj: nn.Linear
-    rope: nn.Module
-
-    def __init__(self, args: ModelArgs, layer_idx: int) -> None: ...
-    def __call__(
-        self,
-        x: mx.array,
-        mask: Optional[mx.array] = None,
-        cache: Optional[Any] = None,
-    ) -> mx.array: ...
-
-class Step3p5DecoderLayer(nn.Module):
-    self_attn: Step3p5Attention
-    is_sliding: bool
-    is_moe_layer: bool
-    mlp: Step3p5MLP | Step3p5MoE
-    input_layernorm: nn.Module
-    post_attention_layernorm: nn.Module
-
-    def __init__(self, args: ModelArgs, layer_idx: int) -> None: ...
-    def __call__(
-        self,
-        x: mx.array,
-        mask: Optional[mx.array] = None,
-        cache: Optional[Any] = None,
-    ) -> mx.array: ...
-
-class Step3p5Model(nn.Module):
-    args: ModelArgs
-    vocab_size: int
-    num_layers: int
-    embed_tokens: nn.Embedding
-    layers: list[Step3p5DecoderLayer]
-    norm: nn.Module
-    _swa_idx: Optional[int]
-    _full_idx: Optional[int]
-
-    def __init__(self, args: ModelArgs) -> None: ...
-    def __call__(
-        self,
-        x: mx.array,
-        cache: Optional[List[Any]] = None,
-    ) -> mx.array: ...
-
-class Model(nn.Module):
-    args: ModelArgs
-    model_type: str
-    model: Step3p5Model
-    lm_head: nn.Linear
-
-    def __init__(self, args: ModelArgs) -> None: ...
-    def __call__(
-        self,
-        inputs: mx.array,
-        cache: Optional[List[Any]] = None,
-    ) -> mx.array: ...
-    def sanitize(self, weights: dict[str, Any]) -> dict[str, Any]: ...
-    def shard(self, group: Optional[mx.distributed.Group] = None) -> None: ...
-    @property
-    def layers(self) -> list[Step3p5DecoderLayer]: ...
-    def make_cache(self) -> list[Any]: ...
-    @property
-    def cast_predicate(self) -> Any: ...
-    @property
-    def quant_predicate(self) -> Any: ...

MISSED_THINGS.md

@@ -5,21 +5,21 @@
 [X] Fetching download status of all models on start
 [X] Deduplication of tasks in plan_step.
 [X] resolve_allow_patterns should just be wildcard now.
-[] no mx_barrier in genreate.py mlx_generate at the end.
+[X] no mx_barrier in genreate.py mlx_generate at the end.
 [] cache assertion not needed in auto_parallel.py PipelineLastLayer.
-[] GPTOSS support dropped in auto_parallel.py.
-[] sharding changed "all-to-sharded" became _all_to_sharded in auto_parallel.py.
-[] same as above with "sharded-to-all" became _sharded_to_all in auto_parallel.py.
-[] Dropped support for Ministral3Model, DeepseekV32Model, Glm4MoeModel, Qwen3NextModel, GptOssMode in auto_parallel.py.
+[X] GPTOSS support dropped in auto_parallel.py.
+[X] sharding changed "all-to-sharded" became _all_to_sharded in auto_parallel.py.
+[X] same as above with "sharded-to-all" became _sharded_to_all in auto_parallel.py.
+[X] Dropped support for Ministral3Model, DeepseekV32Model, Glm4MoeModel, Qwen3NextModel, GptOssMode in auto_parallel.py.
 [] Dropped prefill/decode code in auto_parallel.py and utils_mlx.py.
 [X] KV_CACHE_BITS should be None to disable quantized KV cache.
-[] Dropped _set_nofile_limit in utils_mlx.py.
-[] We have group optional in load_mlx_items in utils_mlx.py.
-[] Dropped add_missing_chat_templates for GptOss in load_mlx_items in utils_mlx.py.
-[] Dropped model.make_cache in make_kv_cache in utils_mlx.py.
+[X] Dropped _set_nofile_limit in utils_mlx.py.
+[X] We have group optional in load_mlx_items in utils_mlx.py.
+[X] Dropped add_missing_chat_templates for GptOss in load_mlx_items in utils_mlx.py.
+[X] Dropped model.make_cache in make_kv_cache in utils_mlx.py.
 [X] We put cache limit back in utils_mlx.py.
-[] topology.py remove_node removes the connections after checking if node is is in self._node_id_to_rx_id_map. on beta_1 it checks after, so would remove stale connections I guess?
-[] Missing Glm 4.7 model cards (this isn't ready yet but should be picked up, probably create an issue... the blocker is transforemrs version doesn't support the tokenizer for Glm 4.7. rc-1 does but we can't upgrade as it breaks other things.)
+[X] topology.py remove_node removes the connections after checking if node is is in self._node_id_to_rx_id_map. on beta_1 it checks after, so would remove stale connections I guess?
+[X] Missing Glm 4.7 model cards (this isn't ready yet but should be picked up, probably create an issue... the blocker is transforemrs version doesn't support the tokenizer for Glm 4.7. rc-1 does but we can't upgrade as it breaks other things.)
 [] try-except in _command_processor only excepts ValueError. This was silently failing leading to un-debuggable errors (we had a KeyError that was happening ). Changed this to catch Exception instead of ValueError. See exo-v2 89ae38405e0052e3c22405daf094b065878aa873 and fb99fea69b5a39017efc90c5dad0072e677455f0.
 [X] In placement.py, place_instance no longer looks at model_meta.supports_tensor and check if this tensor parallel number of nodes is supported by the model's tensor dimensions.
 [X] In placement.py, place_instanec, we no longer have the special case to exclude DeepSeek v3.1 pipeline parallel (it doesn't work).

bench/exo_bench.py

@@ -19,11 +19,6 @@ from urllib.parse import urlencode
 from loguru import logger
 from transformers import AutoTokenizer
 
-# Backoff constants for cluster settling retry
-_SETTLE_INITIAL_BACKOFF_S = 1.0
-_SETTLE_MAX_BACKOFF_S = 60.0
-_SETTLE_BACKOFF_MULTIPLIER = 2.0
-
 # Monkey-patch for transformers 5.x compatibility
 # Kimi's tokenization_kimi.py imports bytes_to_unicode from the old location
 # which was moved in transformers 5.0.0rc2
@@ -393,66 +388,6 @@ class PromptSizer:
         return content, tok
 
 
-def fetch_and_filter_placements(
-    client: ExoClient, full_model_id: str, args: argparse.Namespace
-) -> list[dict[str, Any]]:
-    previews_resp = client.request_json(
-        "GET", "/instance/previews", params={"model_id": full_model_id}
-    )
-    previews = previews_resp.get("previews") or []
-
-    selected: list[dict[str, Any]] = []
-    for p in previews:
-        if p.get("error") is not None:
-            continue
-        if not placement_filter(str(p.get("instance_meta", "")), args.instance_meta):
-            continue
-        if not sharding_filter(str(p.get("sharding", "")), args.sharding):
-            continue
-
-        instance = p.get("instance")
-        if not isinstance(instance, dict):
-            continue
-
-        n = nodes_used_in_instance(instance)
-        # Skip tensor ring single node as it is pointless when pipeline ring
-        if n == 1 and (
-            (args.sharding == "both" and "tensor" in p.get("sharding", "").lower())
-            or (
-                args.instance_meta == "both"
-                and "jaccl" in p.get("instance_meta", "").lower()
-            )
-        ):
-            continue
-
-        if (
-            args.skip_pipeline_jaccl
-            and (
-                args.instance_meta == "both"
-                and "jaccl" in p.get("instance_meta", "").lower()
-            )
-            and (
-                args.sharding == "both" and "pipeline" in p.get("sharding", "").lower()
-            )
-        ):
-            continue
-
-        if (
-            args.skip_tensor_ring
-            and (
-                args.instance_meta == "both"
-                and "ring" in p.get("instance_meta", "").lower()
-            )
-            and (args.sharding == "both" and "tensor" in p.get("sharding", "").lower())
-        ):
-            continue
-
-        if args.min_nodes <= n <= args.max_nodes:
-            selected.append(p)
-
-    return selected
-
-
 def main() -> int:
     ap = argparse.ArgumentParser(
         prog="exo-bench",
@@ -529,12 +464,6 @@ def main() -> int:
         action="store_true",
         help="Force all pp×tg combinations (cartesian product) even when lists have equal length.",
     )
-    ap.add_argument(
-        "--settle-timeout",
-        type=float,
-        default=0,
-        help="Max seconds to wait for the cluster to produce valid placements (0 = try once).",
-    )
     args = ap.parse_args()
 
     pp_list = parse_int_list(args.pp)
@@ -558,6 +487,11 @@ def main() -> int:
     client = ExoClient(args.host, args.port, timeout_s=args.timeout)
     short_id, full_model_id = resolve_model_short_id(client, args.model)
 
+    previews_resp = client.request_json(
+        "GET", "/instance/previews", params={"model_id": full_model_id}
+    )
+    previews = previews_resp.get("previews") or []
+
     tokenizer = load_tokenizer_for_bench(full_model_id)
     if tokenizer is None:
         raise RuntimeError("[exo-bench] tokenizer load failed")
@@ -569,20 +503,54 @@ def main() -> int:
         logger.error("[exo-bench] tokenizer usable but prompt sizing failed")
         raise
 
-    selected = fetch_and_filter_placements(client, full_model_id, args)
+    selected: list[dict[str, Any]] = []
+    for p in previews:
+        if p.get("error") is not None:
+            continue
+        if not placement_filter(str(p.get("instance_meta", "")), args.instance_meta):
+            continue
+        if not sharding_filter(str(p.get("sharding", "")), args.sharding):
+            continue
 
-    if not selected and args.settle_timeout > 0:
-        backoff = _SETTLE_INITIAL_BACKOFF_S
-        deadline = time.monotonic() + args.settle_timeout
-        while not selected and time.monotonic() < deadline:
-            remaining = deadline - time.monotonic()
-            logger.warning(
-                f"No valid placements yet (cluster may still be settling). "
-                f"Retrying in {backoff:.1f}s ({remaining:.0f}s remaining)..."
+        instance = p.get("instance")
+        if not isinstance(instance, dict):
+            continue
+
+        n = nodes_used_in_instance(instance)
+        # Skip tensor ring single node as it is pointless when pipeline ring
+        if n == 1 and (
+            (args.sharding == "both" and "tensor" in p.get("sharding", "").lower())
+            or (
+                args.instance_meta == "both"
+                and "jaccl" in p.get("instance_meta", "").lower()
             )
-            time.sleep(min(backoff, remaining))
-            backoff = min(backoff * _SETTLE_BACKOFF_MULTIPLIER, _SETTLE_MAX_BACKOFF_S)
-            selected = fetch_and_filter_placements(client, full_model_id, args)
+        ):
+            continue
+
+        if (
+            args.skip_pipeline_jaccl
+            and (
+                args.instance_meta == "both"
+                and "jaccl" in p.get("instance_meta", "").lower()
+            )
+            and (
+                args.sharding == "both" and "pipeline" in p.get("sharding", "").lower()
+            )
+        ):
+            continue
+
+        if (
+            args.skip_tensor_ring
+            and (
+                args.instance_meta == "both"
+                and "ring" in p.get("instance_meta", "").lower()
+            )
+            and (args.sharding == "both" and "tensor" in p.get("sharding", "").lower())
+        ):
+            continue
+
+        if args.min_nodes <= n <= args.max_nodes:
+            selected.append(p)
 
     if not selected:
         logger.error("No valid placements matched your filters.")

dashboard/src/lib/components/ChatForm.svelte

@@ -12,8 +12,6 @@
     ttftMs,
     tps,
     totalTokens,
-    thinkingEnabled as thinkingEnabledStore,
-    setConversationThinking,
   } from "$lib/stores/app.svelte";
   import ChatAttachments from "./ChatAttachments.svelte";
   import ImageParamsPanel from "./ImageParamsPanel.svelte";
@@ -27,7 +25,6 @@
     autofocus?: boolean;
     showModelSelector?: boolean;
     modelTasks?: Record<string, string[]>;
-    modelCapabilities?: Record<string, string[]>;
   }
 
   let {
@@ -37,7 +34,6 @@
     autofocus = true,
     showModelSelector = false,
     modelTasks = {},
-    modelCapabilities = {},
   }: Props = $props();
 
   let message = $state("");
@@ -45,7 +41,6 @@
   let fileInputRef: HTMLInputElement | undefined = $state();
   let uploadedFiles = $state<ChatUploadedFile[]>([]);
   let isDragOver = $state(false);
-  const thinkingEnabled = $derived(thinkingEnabledStore());
   let loading = $derived(isLoading());
   const currentModel = $derived(selectedChatModel());
   const instanceData = $derived(instances());
@@ -100,12 +95,6 @@
     );
   });
 
-  const modelSupportsThinking = $derived(() => {
-    if (!currentModel) return false;
-    const caps = modelCapabilities[currentModel] || [];
-    return caps.includes("thinking") && caps.includes("text");
-  });
-
   const isEditOnlyWithoutImage = $derived(
     currentModel !== null &&
       modelSupportsOnlyImageEditing(currentModel) &&
@@ -293,11 +282,7 @@
       // Use image generation for text-to-image models
       generateImage(content);
     } else {
-      sendMessage(
-        content,
-        files,
-        modelSupportsThinking() ? thinkingEnabled : null,
-      );
+      sendMessage(content, files);
     }
 
     // Refocus the textarea after sending
@@ -535,35 +520,6 @@
             </div>
           {/if}
         </div>
-        <!-- Thinking toggle -->
-        {#if modelSupportsThinking()}
-          <button
-            type="button"
-            onclick={() => setConversationThinking(!thinkingEnabled)}
-            class="flex items-center gap-1.5 px-2 py-1 rounded text-xs font-mono tracking-wide transition-all duration-200 flex-shrink-0 cursor-pointer border {thinkingEnabled
-              ? 'bg-exo-yellow/15 border-exo-yellow/40 text-exo-yellow'
-              : 'bg-exo-medium-gray/30 border-exo-medium-gray/50 text-exo-light-gray/60 hover:text-exo-light-gray'}"
-            title={thinkingEnabled
-              ? "Thinking enabled — click to disable"
-              : "Thinking disabled — click to enable"}
-          >
-            <svg
-              class="w-3.5 h-3.5"
-              viewBox="0 0 24 24"
-              fill="none"
-              stroke="currentColor"
-              stroke-width="1.5"
-            >
-              <path
-                d="M12 2a7 7 0 0 0-7 7c0 2.38 1.19 4.47 3 5.74V17a1 1 0 0 0 1 1h6a1 1 0 0 0 1-1v-2.26c1.81-1.27 3-3.36 3-5.74a7 7 0 0 0-7-7zM9 20h6M10 22h4"
-                stroke-linecap="round"
-                stroke-linejoin="round"
-              />
-            </svg>
-            <span>{thinkingEnabled ? "THINK" : "NO THINK"}</span>
-          </button>
-        {/if}
-
         <!-- Performance stats -->
         {#if currentTtft !== null || currentTps !== null}
           <div class="flex items-center gap-4 text-xs font-mono flex-shrink-0">

dashboard/src/lib/components/ChatSidebar.svelte

@@ -185,7 +185,11 @@
 
     let instanceType: string | null = null;
     if (instanceTag === "MlxRingInstance") instanceType = "MLX Ring";
-    else if (instanceTag === "MlxJacclInstance") instanceType = "MLX RDMA";
+    else if (
+      instanceTag === "MlxIbvInstance" ||
+      instanceTag === "MlxJacclInstance"
+    )
+      instanceType = "MLX RDMA";
 
     let sharding: string | null = null;
     const inst = instance as {

dashboard/src/lib/components/ModelCard.svelte

@@ -21,7 +21,7 @@
     } | null;
     nodes?: Record<string, NodeInfo>;
     sharding?: "Pipeline" | "Tensor";
-    runtime?: "MlxRing" | "MlxJaccl";
+    runtime?: "MlxRing" | "MlxIbv" | "MlxJaccl";
     onLaunch?: () => void;
     tags?: string[];
     apiPreview?: PlacementPreview | null;
@@ -348,7 +348,7 @@
   // Debug mode state
   const isDebugMode = $derived(debugMode());
   const topology = $derived(topologyData());
-  const isRdma = $derived(runtime === "MlxJaccl");
+  const isRdma = $derived(runtime === "MlxIbv" || runtime === "MlxJaccl");
 
   // Get interface name for an IP from node data
   function getInterfaceForIp(nodeId: string, ip?: string): string | null {
@@ -575,7 +575,7 @@
       >
         {runtime === "MlxRing"
           ? "MLX Ring"
-          : runtime === "MlxJaccl"
+          : runtime === "MlxIbv" || runtime === "MlxJaccl"
             ? "MLX RDMA"
             : runtime}
       </span>

dashboard/src/lib/components/ModelPickerModal.svelte

@@ -806,7 +806,6 @@
                 isFavorite={favorites.has(group.id)}
                 {selectedModelId}
                 {canModelFit}
-                {getModelFitStatus}
                 onToggleExpand={() => toggleGroupExpanded(group.id)}
                 onSelectModel={handleSelect}
                 {onToggleFavorite}

dashboard/src/lib/components/TopologyGraph.svelte

@@ -7,7 +7,6 @@
     debugMode,
     nodeThunderboltBridge,
     nodeRdmaCtl,
-    nodeIdentities,
     type NodeInfo,
   } from "$lib/stores/app.svelte";
 
@@ -34,7 +33,6 @@
   const debugEnabled = $derived(debugMode());
   const tbBridgeData = $derived(nodeThunderboltBridge());
   const rdmaCtlData = $derived(nodeRdmaCtl());
-  const identitiesData = $derived(nodeIdentities());
 
   function getNodeLabel(nodeId: string): string {
     const node = data?.nodes?.[nodeId];
@@ -1179,22 +1177,6 @@
             .attr("font-size", debugFontSize)
             .attr("font-family", "SF Mono, Monaco, monospace")
             .text(rdmaText);
-          debugLabelY += debugLineHeight;
-        }
-
-        const identity = identitiesData[nodeInfo.id];
-        if (identity?.osVersion) {
-          nodeG
-            .append("text")
-            .attr("x", nodeInfo.x)
-            .attr("y", debugLabelY)
-            .attr("text-anchor", "middle")
-            .attr("fill", "rgba(179,179,179,0.7)")
-            .attr("font-size", debugFontSize)
-            .attr("font-family", "SF Mono, Monaco, monospace")
-            .text(
-              `macOS ${identity.osVersion}${identity.osBuildVersion ? ` (${identity.osBuildVersion})` : ""}`,
-            );
         }
       }
     });

dashboard/src/lib/stores/app.svelte.ts

@@ -168,7 +168,7 @@ export interface ModelDownloadStatus {
 export interface PlacementPreview {
   model_id: string;
   sharding: "Pipeline" | "Tensor";
-  instance_meta: "MlxRing" | "MlxJaccl";
+  instance_meta: "MlxRing" | "MlxIbv" | "MlxJaccl";
   instance: unknown | null;
   memory_delta_by_node: Record<string, number> | null;
   error: string | null;
@@ -219,6 +219,7 @@ interface RawStateResponse {
     string,
     {
       MlxRingInstance?: Instance;
+      MlxIbvInstance?: Instance;
       MlxJacclInstance?: Instance;
     }
   >;
@@ -249,20 +250,6 @@ interface RawStateResponse {
   >;
   // Thunderbolt bridge cycles (nodes with bridge enabled forming loops)
   thunderboltBridgeCycles?: string[][];
-  // MetaInstances (declarative instance constraints)
-  metaInstances?: Record<string, MetaInstanceData>;
-}
-
-export interface MetaInstanceData {
-  metaInstanceId: string;
-  modelId: string;
-  sharding: string;
-  instanceMeta: string;
-  minNodes: number;
-  nodeIds: string[] | null;
-  placementError: string | null;
-  consecutiveFailures: number;
-  lastFailureError: string | null;
 }
 
 export interface MessageAttachment {
@@ -309,7 +296,6 @@ export interface Conversation {
   modelId: string | null;
   sharding: string | null;
   instanceType: string | null;
-  enableThinking: boolean | null;
 }
 
 const STORAGE_KEY = "exo-conversations";
@@ -549,7 +535,6 @@ class AppStore {
   isLoadingPreviews = $state(false);
   previewNodeFilter = $state<Set<string>>(new Set());
   lastUpdate = $state<number | null>(null);
-  metaInstances = $state<Record<string, MetaInstanceData>>({});
   nodeIdentities = $state<Record<string, RawNodeIdentity>>({});
   thunderboltBridgeCycles = $state<string[][]>([]);
   nodeThunderbolt = $state<
@@ -620,7 +605,6 @@ class AppStore {
           modelId: conversation.modelId ?? null,
           sharding: conversation.sharding ?? null,
           instanceType: conversation.instanceType ?? null,
-          enableThinking: conversation.enableThinking ?? null,
         }));
       }
     } catch (error) {
@@ -810,7 +794,6 @@ class AppStore {
       modelId: derivedModelId,
       sharding: derivedSharding,
       instanceType: derivedInstanceType,
-      enableThinking: null,
     };
 
     this.conversations.unshift(conversation);
@@ -836,7 +819,6 @@ class AppStore {
     this.hasStartedChat = true;
     this.isTopologyMinimized = true;
     this.isSidebarOpen = true; // Auto-open sidebar when chatting
-    this.thinkingEnabled = conversation.enableThinking ?? true;
     this.refreshConversationModelFromInstances();
 
     return true;
@@ -909,7 +891,11 @@ class AppStore {
 
     let instanceType: string | null = null;
     if (instanceTag === "MlxRingInstance") instanceType = "MLX Ring";
-    else if (instanceTag === "MlxJacclInstance") instanceType = "MLX RDMA";
+    else if (
+      instanceTag === "MlxIbvInstance" ||
+      instanceTag === "MlxJacclInstance"
+    )
+      instanceType = "MLX RDMA";
 
     let sharding: string | null = null;
     const inst = instance as {
@@ -1274,8 +1260,6 @@ class AppStore {
       if (data.downloads) {
         this.downloads = data.downloads;
       }
-      // MetaInstances
-      this.metaInstances = data.metaInstances ?? {};
       if (data.nodeDisk) {
         this.nodeDisk = data.nodeDisk;
       }
@@ -1948,11 +1932,6 @@ class AppStore {
     }
   }
 
-  /**
-   * Whether thinking is enabled for the current conversation
-   */
-  thinkingEnabled = $state(true);
-
   /**
    * Selected model for chat (can be set by the UI)
    */
@@ -2131,7 +2110,6 @@ class AppStore {
       textContent?: string;
       preview?: string;
     }[],
-    enableThinking?: boolean | null,
   ): Promise<void> {
     if ((!content.trim() && (!files || files.length === 0)) || this.isLoading)
       return;
@@ -2279,9 +2257,6 @@ class AppStore {
           stream: true,
           logprobs: true,
           top_logprobs: 5,
-          ...(enableThinking != null && {
-            enable_thinking: enableThinking,
-          }),
         }),
       });
 
@@ -2940,18 +2915,6 @@ class AppStore {
     );
   }
 
-  /**
-   * Update the thinking preference for the active conversation
-   */
-  setConversationThinking(enabled: boolean) {
-    this.thinkingEnabled = enabled;
-    const conv = this.getActiveConversation();
-    if (conv) {
-      conv.enableThinking = enabled;
-      this.saveConversationsToStorage();
-    }
-  }
-
   /**
    * Start a download on a specific node
    */
@@ -3056,7 +3019,6 @@ export const tps = () => appStore.tps;
 export const totalTokens = () => appStore.totalTokens;
 export const topologyData = () => appStore.topologyData;
 export const instances = () => appStore.instances;
-export const metaInstances = () => appStore.metaInstances;
 export const runners = () => appStore.runners;
 export const downloads = () => appStore.downloads;
 export const nodeDisk = () => appStore.nodeDisk;
@@ -3066,7 +3028,6 @@ export const isLoadingPreviews = () => appStore.isLoadingPreviews;
 export const lastUpdate = () => appStore.lastUpdate;
 export const isTopologyMinimized = () => appStore.isTopologyMinimized;
 export const selectedChatModel = () => appStore.selectedChatModel;
-export const thinkingEnabled = () => appStore.thinkingEnabled;
 export const debugMode = () => appStore.getDebugMode();
 export const topologyOnlyMode = () => appStore.getTopologyOnlyMode();
 export const chatSidebarVisible = () => appStore.getChatSidebarVisible();
@@ -3082,8 +3043,7 @@ export const sendMessage = (
     textContent?: string;
     preview?: string;
   }[],
-  enableThinking?: boolean | null,
-) => appStore.sendMessage(content, files, enableThinking);
+) => appStore.sendMessage(content, files);
 export const generateImage = (prompt: string, modelId?: string) =>
   appStore.generateImage(prompt, modelId);
 export const editImage = (
@@ -3126,8 +3086,6 @@ export const deleteAllConversations = () => appStore.deleteAllConversations();
 export const renameConversation = (id: string, name: string) =>
   appStore.renameConversation(id, name);
 export const getActiveConversation = () => appStore.getActiveConversation();
-export const setConversationThinking = (enabled: boolean) =>
-  appStore.setConversationThinking(enabled);
 
 // Sidebar actions
 export const isSidebarOpen = () => appStore.isSidebarOpen;

resources/inference_model_cards/mlx-community--Step-3.5-Flash-4bit.toml

@@ -1,12 +0,0 @@
-model_id = "mlx-community/Step-3.5-Flash-4bit"
-n_layers = 45
-hidden_size = 4096
-supports_tensor = true
-tasks = ["TextGeneration"]
-family = "step"
-quantization = "4bit"
-base_model = "Step 3.5 Flash"
-capabilities = ["text", "thinking"]
-
-[storage_size]
-in_bytes = 114572190076

resources/inference_model_cards/mlx-community--Step-3.5-Flash-6bit.toml

@@ -1,12 +0,0 @@
-model_id = "mlx-community/Step-3.5-Flash-6bit"
-n_layers = 45
-hidden_size = 4096
-supports_tensor = true
-tasks = ["TextGeneration"]
-family = "step"
-quantization = "6bit"
-base_model = "Step 3.5 Flash"
-capabilities = ["text", "thinking"]
-
-[storage_size]
-in_bytes = 159039627774

resources/inference_model_cards/mlx-community--Step-3.5-Flash-8Bit.toml

@@ -1,12 +0,0 @@
-model_id = "mlx-community/Step-3.5-Flash-8Bit"
-n_layers = 45
-hidden_size = 4096
-supports_tensor = true
-tasks = ["TextGeneration"]
-family = "step"
-quantization = "8bit"
-base_model = "Step 3.5 Flash"
-capabilities = ["text", "thinking"]
-
-[storage_size]
-in_bytes = 209082699847

src/exo/master/adapters/chat_completions.py

@@ -79,7 +79,6 @@ def chat_request_to_text_generation(
         seed=request.seed,
         stream=request.stream,
         tools=request.tools,
-        enable_thinking=request.enable_thinking,
         chat_template_messages=chat_template_messages
         if chat_template_messages
         else None,
@@ -176,7 +175,7 @@ async def generate_chat_stream(
 async def collect_chat_response(
     command_id: CommandId,
     chunk_stream: AsyncGenerator[ErrorChunk | ToolCallChunk | TokenChunk, None],
-) -> ChatCompletionResponse:
+) -> AsyncGenerator[str]:
     """Collect all token chunks and return a single ChatCompletionResponse."""
     text_parts: list[str] = []
     tool_calls: list[ToolCall] = []
@@ -223,7 +222,7 @@ async def collect_chat_response(
     combined_text = "".join(text_parts)
     assert model is not None
 
-    return ChatCompletionResponse(
+    yield ChatCompletionResponse(
         id=command_id,
         created=int(time.time()),
         model=model,
@@ -241,4 +240,5 @@ async def collect_chat_response(
                 finish_reason=finish_reason,
             )
         ],
-    )
+    ).model_dump_json()
+    return

src/exo/master/api.py

@@ -71,11 +71,8 @@ from exo.shared.types.api import (
     ChatCompletionResponse,
     CreateInstanceParams,
     CreateInstanceResponse,
-    CreateMetaInstanceParams,
-    CreateMetaInstanceResponse,
     DeleteDownloadResponse,
     DeleteInstanceResponse,
-    DeleteMetaInstanceResponse,
     ErrorInfo,
     ErrorResponse,
     FinishReason,
@@ -118,10 +115,8 @@ from exo.shared.types.claude_api import (
 from exo.shared.types.commands import (
     Command,
     CreateInstance,
-    CreateMetaInstance,
     DeleteDownload,
     DeleteInstance,
-    DeleteMetaInstance,
     DownloadCommand,
     ForwarderCommand,
     ForwarderDownloadCommand,
@@ -130,10 +125,11 @@ from exo.shared.types.commands import (
     PlaceInstance,
     SendInputChunk,
     StartDownload,
+    TaskCancelled,
     TaskFinished,
     TextGeneration,
 )
-from exo.shared.types.common import CommandId, Id, MetaInstanceId, NodeId, SessionId
+from exo.shared.types.common import CommandId, Id, NodeId, SessionId
 from exo.shared.types.events import (
     ChunkGenerated,
     Event,
@@ -142,7 +138,6 @@ from exo.shared.types.events import (
     TracesMerged,
 )
 from exo.shared.types.memory import Memory
-from exo.shared.types.meta_instance import MetaInstance
 from exo.shared.types.openai_responses import (
     ResponsesRequest,
     ResponsesResponse,
@@ -281,8 +276,6 @@ class API:
         self.app.get("/instance/previews")(self.get_placement_previews)
         self.app.get("/instance/{instance_id}")(self.get_instance)
         self.app.delete("/instance/{instance_id}")(self.delete_instance)
-        self.app.post("/meta_instance")(self.create_meta_instance)
-        self.app.delete("/meta_instance/{meta_instance_id}")(self.delete_meta_instance)
         self.app.get("/models")(self.get_models)
         self.app.get("/v1/models")(self.get_models)
         self.app.post("/models/add")(self.add_custom_model)
@@ -529,46 +522,6 @@ class API:
             instance_id=instance_id,
         )
 
-    async def create_meta_instance(
-        self, payload: CreateMetaInstanceParams
-    ) -> CreateMetaInstanceResponse:
-        meta_instance = MetaInstance(
-            model_id=payload.model_id,
-            sharding=payload.sharding,
-            instance_meta=payload.instance_meta,
-            min_nodes=payload.min_nodes,
-            node_ids=payload.node_ids,
-        )
-        command = CreateMetaInstance(meta_instance=meta_instance)
-        await self._send(command)
-        return CreateMetaInstanceResponse(
-            message="Command received.",
-            command_id=command.command_id,
-            meta_instance_id=meta_instance.meta_instance_id,
-        )
-
-    async def delete_meta_instance(
-        self, meta_instance_id: MetaInstanceId
-    ) -> DeleteMetaInstanceResponse:
-        meta = self.state.meta_instances.get(meta_instance_id)
-        if not meta:
-            raise HTTPException(status_code=404, detail="MetaInstance not found")
-
-        # Delete MetaInstance first to prevent reconciler from re-placing
-        command = DeleteMetaInstance(meta_instance_id=meta_instance_id)
-        await self._send(command)
-
-        # Then cascade-delete any backing instances
-        for instance_id, instance in self.state.instances.items():
-            if instance.meta_instance_id == meta_instance_id:
-                await self._send(DeleteInstance(instance_id=instance_id))
-
-        return DeleteMetaInstanceResponse(
-            message="Command received.",
-            command_id=command.command_id,
-            meta_instance_id=meta_instance_id,
-        )
-
     async def _token_chunk_stream(
         self, command_id: CommandId
     ) -> AsyncGenerator[ErrorChunk | ToolCallChunk | TokenChunk, None]:
@@ -588,16 +541,14 @@ class API:
                         break
 
         except anyio.get_cancelled_exc_class():
-            # TODO: TaskCancelled
-            """
-            self.command_sender.send_nowait(
-                ForwarderCommand(origin=self.node_id, command=command)
-            )
-            """
+            command = TaskCancelled(cancelled_command_id=command_id)
+            with anyio.CancelScope(shield=True):
+                await self.command_sender.send(
+                    ForwarderCommand(origin=self.node_id, command=command)
+                )
             raise
         finally:
-            command = TaskFinished(finished_command_id=command_id)
-            await self._send(command)
+            await self._send(TaskFinished(finished_command_id=command_id))
             if command_id in self._text_generation_queues:
                 del self._text_generation_queues[command_id]
 
@@ -692,11 +643,14 @@ class API:
                     "X-Accel-Buffering": "no",
                 },
             )
-
-        return await collect_chat_response(
-            command.command_id,
-            self._token_chunk_stream(command.command_id),
-        )
+        else:
+            return StreamingResponse(
+                collect_chat_response(
+                    command.command_id,
+                    self._token_chunk_stream(command.command_id),
+                ),
+                media_type="application/json",
+            )
 
     async def bench_chat_completions(
         self, payload: BenchChatCompletionRequest
@@ -712,8 +666,7 @@ class API:
         command = TextGeneration(task_params=task_params)
         await self._send(command)
 
-        response = await self._collect_text_generation_with_stats(command.command_id)
-        return response
+        return await self._collect_text_generation_with_stats(command.command_id)
 
     async def _resolve_and_validate_text_model(self, model_id: ModelId) -> ModelId:
         """Validate a text model exists and return the resolved model ID.
@@ -931,6 +884,11 @@ class API:
                         del image_metadata[key]
 
         except anyio.get_cancelled_exc_class():
+            command = TaskCancelled(cancelled_command_id=command_id)
+            with anyio.CancelScope(shield=True):
+                await self.command_sender.send(
+                    ForwarderCommand(origin=self.node_id, command=command)
+                )
             raise
         finally:
             await self._send(TaskFinished(finished_command_id=command_id))
@@ -1012,6 +970,11 @@ class API:
 
             return (images, stats if capture_stats else None)
         except anyio.get_cancelled_exc_class():
+            command = TaskCancelled(cancelled_command_id=command_id)
+            with anyio.CancelScope(shield=True):
+                await self.command_sender.send(
+                    ForwarderCommand(origin=self.node_id, command=command)
+                )
             raise
         finally:
             await self._send(TaskFinished(finished_command_id=command_id))

src/exo/master/main.py

@@ -1,5 +1,4 @@
-from collections.abc import Sequence
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 
 import anyio
 from anyio.abc import TaskGroup
@@ -13,19 +12,11 @@ from exo.master.placement import (
     get_transition_events,
     place_instance,
 )
-from exo.master.process_managers import ProcessManager
-from exo.master.process_managers.instance_health import InstanceHealthReconciler
-from exo.master.process_managers.meta_instance import MetaInstanceReconciler
-from exo.master.process_managers.node_timeout import NodeTimeoutReconciler
-from exo.master.reconcile import try_place_for_meta_instance
 from exo.shared.apply import apply
 from exo.shared.constants import EXO_EVENT_LOG_DIR, EXO_TRACING_ENABLED
-from exo.shared.models.model_cards import ModelCard
 from exo.shared.types.commands import (
     CreateInstance,
-    CreateMetaInstance,
     DeleteInstance,
-    DeleteMetaInstance,
     ForwarderCommand,
     ForwarderDownloadCommand,
     ImageEdits,
@@ -33,6 +24,7 @@ from exo.shared.types.commands import (
     PlaceInstance,
     RequestEventLog,
     SendInputChunk,
+    TaskCancelled,
     TaskFinished,
     TestCommand,
     TextGeneration,
@@ -43,13 +35,12 @@ from exo.shared.types.events import (
     ForwarderEvent,
     IndexedEvent,
     InputChunkReceived,
-    JacclSideChannelData,
-    JacclSideChannelGathered,
-    MetaInstanceCreated,
-    MetaInstanceDeleted,
+    InstanceDeleted,
     NodeGatheredInfo,
+    NodeTimedOut,
     TaskCreated,
     TaskDeleted,
+    TaskStatusUpdated,
     TraceEventData,
     TracesCollected,
     TracesMerged,
@@ -69,8 +60,7 @@ from exo.shared.types.tasks import (
     TextGeneration as TextGenerationTask,
 )
 from exo.shared.types.worker.instances import InstanceId
-from exo.shared.types.worker.runners import RunnerId
-from exo.utils.channels import Receiver, Sender
+from exo.utils.channels import Receiver, Sender, channel
 from exo.utils.event_buffer import MultiSourceBuffer
 
 
@@ -94,16 +84,16 @@ class Master:
         self.local_event_receiver = local_event_receiver
         self.global_event_sender = global_event_sender
         self.download_command_sender = download_command_sender
+        send, recv = channel[Event]()
+        self.event_sender: Sender[Event] = send
+        self._loopback_event_receiver: Receiver[Event] = recv
+        self._loopback_event_sender: Sender[ForwarderEvent] = (
+            local_event_receiver.clone_sender()
+        )
         self._multi_buffer = MultiSourceBuffer[NodeId, Event]()
         self._event_log = DiskEventLog(EXO_EVENT_LOG_DIR / "master")
         self._pending_traces: dict[TaskId, dict[int, list[TraceEventData]]] = {}
         self._expected_ranks: dict[TaskId, set[int]] = {}
-        self._jaccl_pending: dict[InstanceId, dict[int, dict[RunnerId, bytes]]] = {}
-        self._process_managers: Sequence[ProcessManager] = [
-            InstanceHealthReconciler(),
-            NodeTimeoutReconciler(),
-            MetaInstanceReconciler(),
-        ]
 
     async def run(self):
         logger.info("Starting Master")
@@ -112,12 +102,15 @@ class Master:
             async with self._tg as tg:
                 tg.start_soon(self._event_processor)
                 tg.start_soon(self._command_processor)
-                tg.start_soon(self._reconcile)
+                tg.start_soon(self._loopback_processor)
+                tg.start_soon(self._plan)
         finally:
             self._event_log.close()
             self.global_event_sender.close()
             self.local_event_receiver.close()
             self.command_receiver.close()
+            self._loopback_event_sender.close()
+            self._loopback_event_receiver.close()
 
     async def shutdown(self):
         logger.info("Stopping Master")
@@ -288,7 +281,7 @@ class Master:
                         case DeleteInstance():
                             placement = delete_instance(command, self.state.instances)
                             transition_events = get_transition_events(
-                                self.state.instances, placement
+                                self.state.instances, placement, self.state.tasks
                             )
                             for cmd in cancel_unnecessary_downloads(
                                 placement, self.state.downloads
@@ -299,29 +292,6 @@ class Master:
                                     )
                                 )
                             generated_events.extend(transition_events)
-                        case CreateMetaInstance():
-                            generated_events.append(
-                                MetaInstanceCreated(meta_instance=command.meta_instance)
-                            )
-                            # Immediate placement attempt for responsiveness
-                            model_card = await ModelCard.load(
-                                command.meta_instance.model_id
-                            )
-                            result = try_place_for_meta_instance(
-                                command.meta_instance,
-                                model_card,
-                                self.state.topology,
-                                self.state.instances,
-                                self.state.node_memory,
-                                self.state.node_network,
-                            )
-                            generated_events.extend(result.events)
-                        case DeleteMetaInstance():
-                            generated_events.append(
-                                MetaInstanceDeleted(
-                                    meta_instance_id=command.meta_instance_id
-                                )
-                            )
                         case PlaceInstance():
                             placement = place_instance(
                                 command,
@@ -331,7 +301,7 @@ class Master:
                                 self.state.node_network,
                             )
                             transition_events = get_transition_events(
-                                self.state.instances, placement
+                                self.state.instances, placement, self.state.tasks
                             )
                             generated_events.extend(transition_events)
                         case CreateInstance():
@@ -341,7 +311,7 @@ class Master:
                                 self.state.instances,
                             )
                             transition_events = get_transition_events(
-                                self.state.instances, placement
+                                self.state.instances, placement, self.state.tasks
                             )
                             generated_events.extend(transition_events)
                         case SendInputChunk(chunk=chunk):
@@ -351,6 +321,18 @@ class Master:
                                     chunk=chunk,
                                 )
                             )
+                        case TaskCancelled():
+                            if (
+                                task_id := self.command_task_mapping.get(
+                                    command.cancelled_command_id
+                                )
+                            ) is not None:
+                                generated_events.append(
+                                    TaskStatusUpdated(
+                                        task_status=TaskStatus.Cancelled,
+                                        task_id=task_id,
+                                    )
+                                )
                         case TaskFinished():
                             generated_events.append(
                                 TaskDeleted(
@@ -359,10 +341,9 @@ class Master:
                                     ]
                                 )
                             )
-                            if command.finished_command_id in self.command_task_mapping:
-                                del self.command_task_mapping[
-                                    command.finished_command_id
-                                ]
+                            self.command_task_mapping.pop(
+                                command.finished_command_id, None
+                            )
                         case RequestEventLog():
                             # We should just be able to send everything, since other buffers will ignore old messages
                             # rate limit to 1000 at a time
@@ -373,32 +354,31 @@ class Master:
                             ):
                                 await self._send_event(IndexedEvent(idx=i, event=event))
                     for event in generated_events:
-                        await self._apply_and_broadcast(event)
+                        await self.event_sender.send(event)
                 except ValueError as e:
                     logger.opt(exception=e).warning("Error in command processor")
 
-    async def _apply_and_broadcast(self, event: Event) -> None:
-        """Apply event to state, persist to disk, and broadcast to workers.
-
-        State is updated synchronously (before any await), so callers can
-        rely on ``self.state`` reflecting this event immediately after the
-        call.  Python's cooperative scheduling guarantees no interleaving
-        between the state read and write.
-        """
-        logger.debug(f"Master indexing event: {str(event)[:100]}")
-        indexed = IndexedEvent(event=event, idx=len(self._event_log))
-        self.state = apply(self.state, indexed)
-        event._master_time_stamp = datetime.now(tz=timezone.utc)  # pyright: ignore[reportPrivateUsage]
-        self._event_log.append(event)
-        await self._send_event(indexed)
-
-    async def _reconcile(self) -> None:
+    # These plan loops are the cracks showing in our event sourcing architecture - more things could be commands
+    async def _plan(self) -> None:
         while True:
-            for pm in self._process_managers:
-                events = await pm.reconcile(self.state)
-                for event in events:
-                    await self._apply_and_broadcast(event)
-            await anyio.sleep(1)
+            # kill broken instances
+            connected_node_ids = set(self.state.topology.list_nodes())
+            for instance_id, instance in self.state.instances.items():
+                for node_id in instance.shard_assignments.node_to_runner:
+                    if node_id not in connected_node_ids:
+                        await self.event_sender.send(
+                            InstanceDeleted(instance_id=instance_id)
+                        )
+                        break
+
+            # time out dead nodes
+            for node_id, time in self.state.last_seen.items():
+                now = datetime.now(tz=timezone.utc)
+                if now - time > timedelta(seconds=30):
+                    logger.info(f"Manually removing node {node_id} due to inactivity")
+                    await self.event_sender.send(NodeTimedOut(node_id=node_id))
+
+            await anyio.sleep(10)
 
     async def _event_processor(self) -> None:
         with self.local_event_receiver as local_events:
@@ -416,15 +396,32 @@ class Master:
                         await self._handle_traces_collected(event)
                         continue
 
-                    if isinstance(event, JacclSideChannelData):
-                        await self._apply_and_broadcast(event)
-                        await self._handle_jaccl_side_channel(event)
-                        continue
+                    logger.debug(f"Master indexing event: {str(event)[:100]}")
+                    indexed = IndexedEvent(event=event, idx=len(self._event_log))
+                    self.state = apply(self.state, indexed)
 
+                    event._master_time_stamp = datetime.now(tz=timezone.utc)  # pyright: ignore[reportPrivateUsage]
                     if isinstance(event, NodeGatheredInfo):
                         event.when = str(datetime.now(tz=timezone.utc))
 
-                    await self._apply_and_broadcast(event)
+                    self._event_log.append(event)
+                    await self._send_event(indexed)
+
+    async def _loopback_processor(self) -> None:
+        # this would ideally not be necessary.
+        # this is WAY less hacky than how I was working around this before
+        local_index = 0
+        with self._loopback_event_receiver as events:
+            async for event in events:
+                await self._loopback_event_sender.send(
+                    ForwarderEvent(
+                        origin=NodeId(f"master_{self.node_id}"),
+                        origin_idx=local_index,
+                        session=self.session_id,
+                        event=event,
+                    )
+                )
+                local_index += 1
 
     # This function is re-entrant, take care!
     async def _send_event(self, event: IndexedEvent):
@@ -456,49 +453,10 @@ class Master:
         for trace_data in self._pending_traces[task_id].values():
             all_trace_data.extend(trace_data)
 
-        await self._apply_and_broadcast(
+        await self.event_sender.send(
             TracesMerged(task_id=task_id, traces=all_trace_data)
         )
 
         del self._pending_traces[task_id]
         if task_id in self._expected_ranks:
             del self._expected_ranks[task_id]
-
-    async def _handle_jaccl_side_channel(self, event: JacclSideChannelData) -> None:
-        """Accumulate SideChannel contributions; when all runners for an instance
-        have submitted for the same sequence, emit JacclSideChannelGathered."""
-        iid = event.instance_id
-        seq = event.sequence
-
-        if iid not in self._jaccl_pending:
-            self._jaccl_pending[iid] = {}
-        if seq not in self._jaccl_pending[iid]:
-            self._jaccl_pending[iid][seq] = {}
-        self._jaccl_pending[iid][seq][event.runner_id] = event.data
-
-        instance = self.state.instances.get(iid)
-        if instance is None:
-            logger.warning(f"JacclSideChannelData for unknown instance {iid}")
-            return
-
-        expected_runners = set(instance.shard_assignments.runner_to_shard.keys())
-        submitted = set(self._jaccl_pending[iid][seq].keys())
-
-        logger.info(
-            f"JACCL side channel: instance={iid} seq={seq} "
-            f"submitted={len(submitted)}/{len(expected_runners)}"
-        )
-
-        if submitted >= expected_runners:
-            gathered = dict(self._jaccl_pending[iid][seq])
-            del self._jaccl_pending[iid][seq]
-            if not self._jaccl_pending[iid]:
-                del self._jaccl_pending[iid]
-
-            await self._apply_and_broadcast(
-                JacclSideChannelGathered(
-                    instance_id=iid,
-                    sequence=seq,
-                    gathered_data=gathered,
-                )
-            )

src/exo/master/placement.py

@@ -22,9 +22,15 @@ from exo.shared.types.commands import (
     PlaceInstance,
 )
 from exo.shared.types.common import NodeId
-from exo.shared.types.events import Event, InstanceCreated, InstanceDeleted
+from exo.shared.types.events import (
+    Event,
+    InstanceCreated,
+    InstanceDeleted,
+    TaskStatusUpdated,
+)
 from exo.shared.types.memory import Memory
 from exo.shared.types.profiling import MemoryUsage, NodeNetworkInfo
+from exo.shared.types.tasks import Task, TaskId, TaskStatus
 from exo.shared.types.worker.downloads import (
     DownloadOngoing,
     DownloadProgress,
@@ -106,11 +112,7 @@ def place_instance(
         cycle for cycle in smallest_cycles if topology.is_rdma_cycle(cycle)
     ]
 
-    if command.instance_meta == InstanceMeta.MlxJaccl:
-        if not smallest_rdma_cycles:
-            raise ValueError(
-                "Requested RDMA (MlxJaccl) but no RDMA-connected cycles available"
-            )
+    if command.instance_meta == InstanceMeta.MlxJaccl and smallest_rdma_cycles != []:
         smallest_cycles = smallest_rdma_cycles
 
     cycles_with_leaf_nodes: list[Cycle] = [
@@ -190,6 +192,7 @@ def delete_instance(
 def get_transition_events(
     current_instances: Mapping[InstanceId, Instance],
     target_instances: Mapping[InstanceId, Instance],
+    tasks: Mapping[TaskId, Task],
 ) -> Sequence[Event]:
     events: list[Event] = []
 
@@ -205,6 +208,18 @@ def get_transition_events(
     # find instances to delete
     for instance_id in current_instances:
         if instance_id not in target_instances:
+            for task in tasks.values():
+                if task.instance_id == instance_id and task.task_status in [
+                    TaskStatus.Pending,
+                    TaskStatus.Running,
+                ]:
+                    events.append(
+                        TaskStatusUpdated(
+                            task_status=TaskStatus.Cancelled,
+                            task_id=task.task_id,
+                        )
+                    )
+
             events.append(
                 InstanceDeleted(
                     instance_id=instance_id,

src/exo/master/process_managers/__init__.py

@@ -1,12 +0,0 @@
-from collections.abc import Sequence
-from typing import Protocol, runtime_checkable
-
-from exo.shared.types.events import Event
-from exo.shared.types.state import State
-
-
-@runtime_checkable
-class ProcessManager(Protocol):
-    """A reconciliation step that examines state and returns corrective events."""
-
-    async def reconcile(self, state: State) -> Sequence[Event]: ...

src/exo/master/process_managers/instance_health.py

@@ -1,49 +0,0 @@
-from collections.abc import Sequence
-from typing import final
-
-from exo.master.reconcile import instance_connections_healthy, instance_runners_failed
-from exo.shared.types.events import Event, InstanceDeleted, InstanceRetrying
-from exo.shared.types.state import State
-
-MAX_INSTANCE_RETRIES = 3
-
-
-@final
-class InstanceHealthReconciler:
-    """Delete instances whose network connections are broken or whose runners have all failed."""
-
-    async def reconcile(self, state: State) -> Sequence[Event]:
-        events: list[Event] = []
-        for instance_id, instance in state.instances.items():
-            if not instance_connections_healthy(instance, state.topology):
-                events.append(
-                    InstanceDeleted(
-                        instance_id=instance_id,
-                        failure_error="Network connection lost",
-                    )
-                )
-                continue
-
-            is_failed, error_message = instance_runners_failed(
-                instance, state.runners, state.node_identities
-            )
-            if is_failed:
-                # Retry within the same instance if backed by a MetaInstance
-                mid = instance.meta_instance_id
-                mi = state.meta_instances.get(mid) if mid else None
-                if mid and mi and mi.consecutive_failures < MAX_INSTANCE_RETRIES:
-                    events.append(
-                        InstanceRetrying(
-                            instance_id=instance_id,
-                            meta_instance_id=mid,
-                            failure_error=error_message or "Runner failed",
-                        )
-                    )
-                else:
-                    events.append(
-                        InstanceDeleted(
-                            instance_id=instance_id,
-                            failure_error=error_message,
-                        )
-                    )
-        return events

src/exo/master/process_managers/meta_instance.py

@@ -1,56 +0,0 @@
-from collections.abc import Sequence
-from typing import final
-
-from exo.master.reconcile import (
-    find_unsatisfied_meta_instances,
-    try_place_for_meta_instance,
-)
-from exo.shared.models.model_cards import ModelCard
-from exo.shared.types.events import Event, InstanceCreated, MetaInstancePlacementFailed
-from exo.shared.types.state import State
-from exo.shared.types.worker.instances import Instance, InstanceId
-
-
-@final
-class MetaInstanceReconciler:
-    """Place instances for unsatisfied MetaInstances."""
-
-    async def reconcile(self, state: State) -> Sequence[Event]:
-        all_events: list[Event] = []
-        # Local copy for intermediate tracking — so placement of B
-        # sees A's instance and doesn't double-place on same resources.
-        current_instances: dict[InstanceId, Instance] = dict(state.instances)
-
-        unsatisfied = find_unsatisfied_meta_instances(
-            state.meta_instances,
-            current_instances,
-            state.topology,
-        )
-        for meta_instance in unsatisfied:
-            model_card = await ModelCard.load(meta_instance.model_id)
-            result = try_place_for_meta_instance(
-                meta_instance,
-                model_card,
-                state.topology,
-                current_instances,
-                state.node_memory,
-                state.node_network,
-            )
-            # Update local instance map so next placement sees this one
-            for event in result.events:
-                if isinstance(event, InstanceCreated):
-                    current_instances[event.instance.instance_id] = event.instance
-            all_events.extend(result.events)
-
-            # Emit placement failure if error differs from what's already in state
-            if (
-                result.error is not None
-                and meta_instance.placement_error != result.error
-            ):
-                all_events.append(
-                    MetaInstancePlacementFailed(
-                        meta_instance_id=meta_instance.meta_instance_id,
-                        reason=result.error,
-                    )
-                )
-        return all_events

src/exo/master/process_managers/node_timeout.py

@@ -1,27 +0,0 @@
-from collections.abc import Sequence
-from datetime import datetime, timedelta, timezone
-from typing import final
-
-from loguru import logger
-
-from exo.shared.types.events import Event, NodeTimedOut
-from exo.shared.types.state import State
-
-_DEFAULT_TIMEOUT = timedelta(seconds=30)
-
-
-@final
-class NodeTimeoutReconciler:
-    """Time out nodes that haven't been seen recently."""
-
-    def __init__(self, timeout: timedelta = _DEFAULT_TIMEOUT) -> None:
-        self.timeout = timeout
-
-    async def reconcile(self, state: State) -> Sequence[Event]:
-        now = datetime.now(tz=timezone.utc)
-        events: list[Event] = []
-        for node_id, last_seen in state.last_seen.items():
-            if now - last_seen > self.timeout:
-                logger.info(f"Removing node {node_id} due to inactivity")
-                events.append(NodeTimedOut(node_id=node_id))
-        return events

src/exo/master/reconcile.py

@@ -1,240 +0,0 @@
-from collections.abc import Mapping, Sequence
-from typing import NamedTuple
-
-from loguru import logger
-
-from exo.master.placement import get_transition_events, place_instance
-from exo.shared.models.model_cards import ModelCard
-from exo.shared.topology import Topology
-from exo.shared.types.commands import PlaceInstance
-from exo.shared.types.common import MetaInstanceId, NodeId
-from exo.shared.types.events import Event
-from exo.shared.types.meta_instance import MetaInstance
-from exo.shared.types.profiling import MemoryUsage, NodeIdentity, NodeNetworkInfo
-from exo.shared.types.topology import RDMAConnection, SocketConnection
-from exo.shared.types.worker.instances import (
-    BaseInstance,
-    Instance,
-    InstanceId,
-    MlxJacclInstance,
-    MlxRingInstance,
-)
-from exo.shared.types.worker.runners import (
-    RunnerFailed,
-    RunnerId,
-    RunnerShutdown,
-    RunnerStatus,
-)
-
-
-class PlacementResult(NamedTuple):
-    """Result of a placement attempt: events to apply and optional error reason."""
-
-    events: Sequence[Event]
-    error: str | None
-
-
-def _get_ring_order(instance: BaseInstance) -> list[NodeId]:
-    """Reconstruct ring order from shard device_rank."""
-    node_ranks: list[tuple[NodeId, int]] = []
-    for node_id, runner_id in instance.shard_assignments.node_to_runner.items():
-        shard = instance.shard_assignments.runner_to_shard[runner_id]
-        node_ranks.append((node_id, shard.device_rank))
-    node_ranks.sort(key=lambda x: x[1])
-    return [node_id for node_id, _ in node_ranks]
-
-
-def _ring_connections_healthy(instance: MlxRingInstance, topology: Topology) -> bool:
-    """Check that the specific IPs used by a ring instance still exist in the topology."""
-    ring = _get_ring_order(instance)
-    n = len(ring)
-    for node in ring:
-        hosts = instance.hosts_by_node[node]
-        for idx in range(n):
-            host = hosts[idx]
-            if host.ip in ("0.0.0.0", "198.51.100.1"):
-                continue  # self or placeholder
-            # Real connection: node → ring[idx]. Check specific IP.
-            connections = topology.get_all_connections_between(node, ring[idx])
-            if not any(
-                isinstance(c, SocketConnection)
-                and c.sink_multiaddr.ip_address == host.ip
-                for c in connections
-            ):
-                return False
-    return True
-
-
-def _jaccl_connections_healthy(instance: MlxJacclInstance, topology: Topology) -> bool:
-    """Check that the specific RDMA interfaces used by a JACCL instance still exist."""
-    ring = _get_ring_order(instance)
-    n = len(ring)
-    for i in range(n):
-        for j in range(n):
-            iface = instance.jaccl_devices[i][j]
-            if iface is None:
-                continue
-            connections = topology.get_all_connections_between(ring[i], ring[j])
-            if not any(
-                isinstance(c, RDMAConnection) and c.source_rdma_iface == iface
-                for c in connections
-            ):
-                return False
-    return True
-
-
-def instance_connections_healthy(instance: Instance, topology: Topology) -> bool:
-    """Check that an instance's nodes and specific connections are still in the topology."""
-    instance_nodes = set(instance.shard_assignments.node_to_runner.keys())
-    if not all(topology.contains_node(n) for n in instance_nodes):
-        return False
-    if len(instance_nodes) <= 1:
-        return True
-    match instance:
-        case MlxRingInstance():
-            return _ring_connections_healthy(instance, topology)
-        case MlxJacclInstance():
-            return _jaccl_connections_healthy(instance, topology)
-
-
-def instance_runners_failed(
-    instance: Instance,
-    runners: Mapping[RunnerId, RunnerStatus],
-    node_identities: Mapping[NodeId, NodeIdentity],
-) -> tuple[bool, str | None]:
-    """Check if an instance's runners have all reached terminal failure states.
-
-    Returns ``(True, error_message)`` when ALL runners are terminal
-    (``RunnerFailed`` or ``RunnerShutdown``) and at least one is ``RunnerFailed``.
-
-    Returns ``(False, None)`` when runners are still active, haven't reported
-    yet, or all gracefully shut down (no ``RunnerFailed``).
-    """
-    instance_runner_ids = set(instance.shard_assignments.node_to_runner.values())
-
-    if not instance_runner_ids:
-        return False, None
-
-    # Build reverse mapping: runner_id -> node_id
-    runner_to_node: dict[RunnerId, NodeId] = {
-        runner_id: node_id
-        for node_id, runner_id in instance.shard_assignments.node_to_runner.items()
-    }
-
-    has_any_failed = False
-    error_messages: list[str] = []
-
-    for runner_id in instance_runner_ids:
-        status = runners.get(runner_id)
-        if status is None:
-            # Runner hasn't reported yet — instance is still starting
-            return False, None
-        if isinstance(status, RunnerFailed):
-            has_any_failed = True
-            if status.error_message:
-                node_id = runner_to_node.get(runner_id)
-                name = (
-                    node_identities[node_id].friendly_name
-                    if node_id and node_id in node_identities
-                    else node_id or "unknown"
-                )
-                error_messages.append(f"{name}: {status.error_message}")
-        elif isinstance(status, RunnerShutdown):
-            pass  # Terminal but not a failure indicator on its own
-        else:
-            # Runner is still active (connecting, loading, running, etc.)
-            return False, None
-
-    if has_any_failed:
-        return True, "; ".join(error_messages) if error_messages else "Runner failed"
-
-    # All runners are Shutdown but none Failed — graceful shutdown, not a failure
-    return False, None
-
-
-def instance_satisfies_meta_instance(
-    meta_instance: MetaInstance,
-    instance: Instance,
-) -> bool:
-    """Check if a single instance satisfies a meta-instance's constraints.
-
-    This is a pure constraint check (model, min_nodes, node_ids).
-    Use ``instance_connections_healthy`` separately for topology health.
-    """
-    if instance.shard_assignments.model_id != meta_instance.model_id:
-        return False
-
-    instance_nodes = set(instance.shard_assignments.node_to_runner.keys())
-
-    if len(instance_nodes) < meta_instance.min_nodes:
-        return False
-
-    return meta_instance.node_ids is None or set(meta_instance.node_ids).issubset(
-        instance_nodes
-    )
-
-
-def find_unsatisfied_meta_instances(
-    meta_instances: Mapping[MetaInstanceId, MetaInstance],
-    instances: Mapping[InstanceId, Instance],
-    topology: Topology,
-) -> Sequence[MetaInstance]:
-    """Return meta-instances that have no healthy backing instance."""
-    unsatisfied: list[MetaInstance] = []
-    for meta_id, meta_instance in meta_instances.items():
-        has_healthy_backing = any(
-            instance.meta_instance_id == meta_id
-            and instance_connections_healthy(instance, topology)
-            for instance in instances.values()
-        )
-        if not has_healthy_backing:
-            unsatisfied.append(meta_instance)
-    return unsatisfied
-
-
-def try_place_for_meta_instance(
-    meta_instance: MetaInstance,
-    model_card: ModelCard,
-    topology: Topology,
-    current_instances: Mapping[InstanceId, Instance],
-    node_memory: Mapping[NodeId, MemoryUsage],
-    node_network: Mapping[NodeId, NodeNetworkInfo],
-) -> PlacementResult:
-    """Try to place an instance satisfying the meta-instance constraints.
-
-    Returns a :class:`PlacementResult` with events on success, or an error
-    reason on failure.
-    """
-    command = PlaceInstance(
-        model_card=model_card,
-        sharding=meta_instance.sharding,
-        instance_meta=meta_instance.instance_meta,
-        min_nodes=meta_instance.min_nodes,
-    )
-    try:
-        target_instances = place_instance(
-            command,
-            topology,
-            current_instances,
-            node_memory,
-            node_network,
-            required_nodes=(
-                set(meta_instance.node_ids) if meta_instance.node_ids else None
-            ),
-        )
-        # Tag the new instance with meta_instance_id
-        new_instance_ids = set(target_instances.keys()) - set(current_instances.keys())
-        if new_instance_ids:
-            new_id = next(iter(new_instance_ids))
-            target_instances[new_id] = target_instances[new_id].model_copy(
-                update={"meta_instance_id": meta_instance.meta_instance_id}
-            )
-        return PlacementResult(
-            events=list(get_transition_events(current_instances, target_instances)),
-            error=None,
-        )
-    except ValueError as e:
-        logger.debug(
-            f"MetaInstance placement not possible for {meta_instance.model_id}: {e}"
-        )
-        return PlacementResult(events=[], error=str(e))

src/exo/master/tests/test_meta_instance_edge_cases.py

@@ -1,573 +0,0 @@
-"""Edge-case and regression tests for MetaInstance lifecycle, concurrent operations, and error handling."""
-
-from exo.master.process_managers.instance_health import (
-    MAX_INSTANCE_RETRIES,
-    InstanceHealthReconciler,
-)
-from exo.master.process_managers.meta_instance import MetaInstanceReconciler
-from exo.master.reconcile import (
-    find_unsatisfied_meta_instances,
-    instance_connections_healthy,
-    instance_runners_failed,
-    instance_satisfies_meta_instance,
-)
-from exo.shared.apply import apply
-from exo.shared.models.model_cards import ModelCard, ModelId, ModelTask
-from exo.shared.topology import Topology
-from exo.shared.types.common import Host, MetaInstanceId, NodeId
-from exo.shared.types.events import (
-    IndexedEvent,
-    InstanceCreated,
-    InstanceDeleted,
-    InstanceRetrying,
-    MetaInstanceCreated,
-    MetaInstanceDeleted,
-    MetaInstancePlacementFailed,
-)
-from exo.shared.types.memory import Memory
-from exo.shared.types.meta_instance import MetaInstance
-from exo.shared.types.multiaddr import Multiaddr
-from exo.shared.types.profiling import NodeIdentity
-from exo.shared.types.state import State
-from exo.shared.types.topology import Connection, SocketConnection
-from exo.shared.types.worker.instances import (
-    InstanceId,
-    MlxRingInstance,
-)
-from exo.shared.types.worker.runners import (
-    RunnerFailed,
-    RunnerId,
-    RunnerReady,
-    ShardAssignments,
-)
-from exo.shared.types.worker.shards import PipelineShardMetadata
-
-# --- Helpers (copied from test_reconcile.py for independence) ---
-
-
-def _model_card(model_id: str = "test-org/test-model") -> ModelCard:
-    return ModelCard(
-        model_id=ModelId(model_id),
-        storage_size=Memory.from_kb(1000),
-        n_layers=10,
-        hidden_size=30,
-        supports_tensor=True,
-        tasks=[ModelTask.TextGeneration],
-    )
-
-
-def _topology(*node_ids: str, connect: bool = True) -> Topology:
-    t = Topology()
-    nodes = [NodeId(n) for n in node_ids]
-    for n in nodes:
-        t.add_node(n)
-    if connect and len(nodes) > 1:
-        for i in range(len(nodes)):
-            j = (i + 1) % len(nodes)
-            t.add_connection(
-                Connection(
-                    source=nodes[i],
-                    sink=nodes[j],
-                    edge=SocketConnection(
-                        sink_multiaddr=Multiaddr(
-                            address=f"/ip4/10.0.0.{j + 1}/tcp/50000"
-                        )
-                    ),
-                )
-            )
-            t.add_connection(
-                Connection(
-                    source=nodes[j],
-                    sink=nodes[i],
-                    edge=SocketConnection(
-                        sink_multiaddr=Multiaddr(
-                            address=f"/ip4/10.0.0.{i + 1}/tcp/50000"
-                        )
-                    ),
-                )
-            )
-    return t
-
-
-def _meta_instance(
-    model_id: str = "test-org/test-model",
-    *,
-    min_nodes: int = 1,
-    node_ids: list[NodeId] | None = None,
-    meta_instance_id: MetaInstanceId | None = None,
-    consecutive_failures: int = 0,
-    last_failure_error: str | None = None,
-    placement_error: str | None = None,
-) -> MetaInstance:
-    return MetaInstance(
-        meta_instance_id=meta_instance_id or MetaInstanceId(),
-        model_id=ModelId(model_id),
-        min_nodes=min_nodes,
-        node_ids=node_ids,
-        consecutive_failures=consecutive_failures,
-        last_failure_error=last_failure_error,
-        placement_error=placement_error,
-    )
-
-
-def _instance(
-    model_id: str = "test-org/test-model",
-    node_ids: list[str] | None = None,
-    instance_id: InstanceId | None = None,
-    meta_instance_id: MetaInstanceId | None = None,
-) -> tuple[InstanceId, MlxRingInstance]:
-    iid = instance_id or InstanceId()
-    nodes = node_ids or ["node-a"]
-    n = len(nodes)
-    mc = _model_card(model_id)
-    ephemeral_port = 50000
-    node_to_runner = {NodeId(nd): RunnerId() for nd in nodes}
-    runner_to_shard = {
-        runner_id: PipelineShardMetadata(
-            model_card=mc,
-            device_rank=i,
-            world_size=n,
-            start_layer=0,
-            end_layer=mc.n_layers,
-            n_layers=mc.n_layers,
-        )
-        for i, runner_id in enumerate(node_to_runner.values())
-    }
-    hosts_by_node: dict[NodeId, list[Host]] = {}
-    for r, node_str in enumerate(nodes):
-        hosts: list[Host] = []
-        for idx in range(n):
-            if idx == r:
-                hosts.append(Host(ip="0.0.0.0", port=ephemeral_port))
-            elif n > 1 and idx in ((r - 1) % n, (r + 1) % n):
-                hosts.append(Host(ip=f"10.0.0.{idx + 1}", port=ephemeral_port))
-            else:
-                hosts.append(Host(ip="198.51.100.1", port=0))
-        hosts_by_node[NodeId(node_str)] = hosts
-    return iid, MlxRingInstance(
-        instance_id=iid,
-        shard_assignments=ShardAssignments(
-            model_id=ModelId(model_id),
-            runner_to_shard=runner_to_shard,
-            node_to_runner=node_to_runner,
-        ),
-        hosts_by_node=hosts_by_node,
-        ephemeral_port=ephemeral_port,
-        meta_instance_id=meta_instance_id,
-    )
-
-
-# =============================================================================
-# 1. MetaInstance lifecycle edge cases
-# =============================================================================
-
-
-def test_meta_instance_model_is_frozen():
-    """MetaInstance should be immutable (frozen model)."""
-    meta = _meta_instance()
-    try:
-        meta.model_id = ModelId("something-else")
-        raise AssertionError("Should have raised")
-    except Exception:
-        pass  # Expected — frozen model
-
-
-def test_meta_instance_created_then_deleted_roundtrip():
-    """Create and delete a MetaInstance through apply — state should be clean."""
-    state = State()
-    meta = _meta_instance()
-    state = apply(state, IndexedEvent(idx=0, event=MetaInstanceCreated(meta_instance=meta)))
-    assert meta.meta_instance_id in state.meta_instances
-    state = apply(
-        state,
-        IndexedEvent(idx=1, event=MetaInstanceDeleted(meta_instance_id=meta.meta_instance_id)),
-    )
-    assert meta.meta_instance_id not in state.meta_instances
-    assert len(state.meta_instances) == 0
-
-
-def test_delete_nonexistent_meta_instance_is_safe():
-    """Deleting a MetaInstance that doesn't exist should not crash."""
-    state = State()
-    event = MetaInstanceDeleted(meta_instance_id=MetaInstanceId("nonexistent"))
-    new_state = apply(state, IndexedEvent(idx=0, event=event))
-    assert len(new_state.meta_instances) == 0
-
-
-def test_placement_failed_for_nonexistent_meta_instance_is_safe():
-    """MetaInstancePlacementFailed for unknown ID should not crash."""
-    state = State()
-    event = MetaInstancePlacementFailed(
-        meta_instance_id=MetaInstanceId("nonexistent"),
-        reason="test",
-    )
-    new_state = apply(state, IndexedEvent(idx=0, event=event))
-    assert len(new_state.meta_instances) == 0
-
-
-def test_multiple_meta_instances_for_same_model():
-    """Multiple MetaInstances for the same model are tracked independently."""
-    state = State()
-    meta_a = _meta_instance("test-org/model-x")
-    meta_b = _meta_instance("test-org/model-x")
-    state = apply(state, IndexedEvent(idx=0, event=MetaInstanceCreated(meta_instance=meta_a)))
-    state = apply(state, IndexedEvent(idx=1, event=MetaInstanceCreated(meta_instance=meta_b)))
-    assert len(state.meta_instances) == 2
-    assert meta_a.meta_instance_id in state.meta_instances
-    assert meta_b.meta_instance_id in state.meta_instances
-
-
-# =============================================================================
-# 2. Retry logic edge cases
-# =============================================================================
-
-
-def test_retry_counter_resets_on_successful_instance_creation():
-    """When a new instance is created for a meta-instance, failures should reset."""
-    meta = _meta_instance(consecutive_failures=2, last_failure_error="old")
-    _, inst = _instance(node_ids=["node-a"], meta_instance_id=meta.meta_instance_id)
-    state = State(meta_instances={meta.meta_instance_id: meta})
-    state = apply(state, IndexedEvent(idx=0, event=InstanceCreated(instance=inst)))
-    mi = state.meta_instances[meta.meta_instance_id]
-    assert mi.consecutive_failures == 0
-    # last_failure_error is preserved (for UI display)
-    assert mi.last_failure_error == "old"
-
-
-async def test_retry_count_increments_through_full_cycle():
-    """Walk through MAX_INSTANCE_RETRIES worth of retries, then verify delete."""
-    meta = _meta_instance()
-    iid, inst = _instance(node_ids=["node-a"], meta_instance_id=meta.meta_instance_id)
-    topology = _topology("node-a")
-    state = State(
-        meta_instances={meta.meta_instance_id: meta},
-        instances={iid: inst},
-        topology=topology,
-    )
-
-    runner_ids = list(inst.shard_assignments.node_to_runner.values())
-    for idx, i in enumerate(range(MAX_INSTANCE_RETRIES)):
-        # Simulate runners failing
-        state_with_runners = state.model_copy(
-            update={"runners": {runner_ids[0]: RunnerFailed(error_message=f"fail-{i}")}}
-        )
-        reconciler = InstanceHealthReconciler()
-        events = await reconciler.reconcile(state_with_runners)
-        assert len(events) == 1
-        assert isinstance(events[0], InstanceRetrying), f"iteration {i}"
-        state = apply(state, IndexedEvent(idx=idx, event=events[0]))
-
-    # After MAX_INSTANCE_RETRIES retries, failure counter should be at max
-    mi = state.meta_instances[meta.meta_instance_id]
-    assert mi.consecutive_failures == MAX_INSTANCE_RETRIES
-
-    # Next failure should result in deletion
-    state_with_runners = state.model_copy(
-        update={"runners": {runner_ids[0]: RunnerFailed(error_message="final")}}
-    )
-    reconciler = InstanceHealthReconciler()
-    events = await reconciler.reconcile(state_with_runners)
-    assert len(events) == 1
-    assert isinstance(events[0], InstanceDeleted)
-
-
-async def test_health_reconciler_respects_exact_limit():
-    """At exactly MAX_INSTANCE_RETRIES, reconciler should delete, not retry."""
-    meta = _meta_instance(consecutive_failures=MAX_INSTANCE_RETRIES)
-    iid, inst = _instance(node_ids=["node-a"], meta_instance_id=meta.meta_instance_id)
-    runner_ids = list(inst.shard_assignments.node_to_runner.values())
-    state = State(
-        meta_instances={meta.meta_instance_id: meta},
-        instances={iid: inst},
-        runners={runner_ids[0]: RunnerFailed(error_message="OOM")},
-        topology=_topology("node-a"),
-    )
-    reconciler = InstanceHealthReconciler()
-    events = await reconciler.reconcile(state)
-    assert len(events) == 1
-    assert isinstance(events[0], InstanceDeleted)
-
-
-async def test_health_reconciler_at_limit_minus_one_retries():
-    """At MAX_INSTANCE_RETRIES - 1, reconciler should still retry."""
-    meta = _meta_instance(consecutive_failures=MAX_INSTANCE_RETRIES - 1)
-    iid, inst = _instance(node_ids=["node-a"], meta_instance_id=meta.meta_instance_id)
-    runner_ids = list(inst.shard_assignments.node_to_runner.values())
-    state = State(
-        meta_instances={meta.meta_instance_id: meta},
-        instances={iid: inst},
-        runners={runner_ids[0]: RunnerFailed(error_message="OOM")},
-        topology=_topology("node-a"),
-    )
-    reconciler = InstanceHealthReconciler()
-    events = await reconciler.reconcile(state)
-    assert len(events) == 1
-    assert isinstance(events[0], InstanceRetrying)
-
-
-# =============================================================================
-# 3. Error handling edge cases
-# =============================================================================
-
-
-def test_runners_failed_with_empty_error_message():
-    """RunnerFailed with empty error_message should still report as failed."""
-    _, inst = _instance(node_ids=["node-a"])
-    runners = {
-        rid: RunnerFailed(error_message="")
-        for rid in inst.shard_assignments.node_to_runner.values()
-    }
-    is_failed, error = instance_runners_failed(inst, runners, {})
-    assert is_failed is True
-    # Empty error message means we get the fallback
-    assert error == "Runner failed"
-
-
-def test_runners_failed_with_none_error_message():
-    """RunnerFailed with None error_message should still report as failed."""
-    _, inst = _instance(node_ids=["node-a"])
-    runners = {
-        rid: RunnerFailed(error_message=None)
-        for rid in inst.shard_assignments.node_to_runner.values()
-    }
-    is_failed, error = instance_runners_failed(inst, runners, {})
-    assert is_failed is True
-    assert error == "Runner failed"
-
-
-def test_runners_failed_collects_all_error_messages():
-    """With multiple failed runners, all error messages should be collected."""
-    _, inst = _instance(node_ids=["node-a", "node-b", "node-c"])
-    runner_ids = list(inst.shard_assignments.node_to_runner.values())
-    runners = {
-        runner_ids[0]: RunnerFailed(error_message="OOM on GPU 0"),
-        runner_ids[1]: RunnerFailed(error_message="OOM on GPU 1"),
-        runner_ids[2]: RunnerFailed(error_message="OOM on GPU 2"),
-    }
-    is_failed, error = instance_runners_failed(inst, runners, {})
-    assert is_failed is True
-    assert error is not None
-    assert "OOM on GPU 0" in error
-    assert "OOM on GPU 1" in error
-    assert "OOM on GPU 2" in error
-
-
-def test_runners_failed_includes_friendly_name():
-    """Error messages should include node friendly names when available."""
-    _, inst = _instance(node_ids=["node-a"])
-    node_id = NodeId("node-a")
-    runner_ids = list(inst.shard_assignments.node_to_runner.values())
-    runners = {runner_ids[0]: RunnerFailed(error_message="OOM")}
-    identities = {node_id: NodeIdentity(friendly_name="My Mac Studio")}
-    is_failed, error = instance_runners_failed(inst, runners, identities)
-    assert is_failed is True
-    assert error is not None
-    assert "My Mac Studio" in error
-
-
-def test_instance_retrying_for_missing_instance_is_safe():
-    """InstanceRetrying for an instance not in state should not crash.
-
-    NOTE: When the instance is missing, the handler returns early WITHOUT
-    incrementing the MetaInstance failure counter. This means stale retry
-    events for already-deleted instances are silently dropped. This is
-    acceptable since the InstanceDeleted handler already increments failures.
-    """
-    meta = _meta_instance()
-    state = State(meta_instances={meta.meta_instance_id: meta})
-    event = InstanceRetrying(
-        instance_id=InstanceId("nonexistent"),
-        meta_instance_id=meta.meta_instance_id,
-        failure_error="crash",
-    )
-    new_state = apply(state, IndexedEvent(idx=0, event=event))
-    # Does not crash, but failure count is NOT incremented (early return)
-    mi = new_state.meta_instances[meta.meta_instance_id]
-    assert mi.consecutive_failures == 0
-
-
-# =============================================================================
-# 4. Backward compatibility
-# =============================================================================
-
-
-def test_instance_without_meta_instance_id_works():
-    """Instances created without meta_instance_id should still function normally."""
-    _, inst = _instance(node_ids=["node-a"])
-    assert inst.meta_instance_id is None
-    topology = _topology("node-a")
-    assert instance_connections_healthy(inst, topology) is True
-
-
-def test_instance_deleted_without_meta_does_not_affect_meta_instances():
-    """Deleting an instance without meta_instance_id should not affect meta_instances."""
-    meta = _meta_instance()
-    iid, inst = _instance(node_ids=["node-a"])  # no meta_instance_id
-    state = State(
-        meta_instances={meta.meta_instance_id: meta},
-        instances={iid: inst},
-    )
-    event = InstanceDeleted(instance_id=iid, failure_error="crash")
-    new_state = apply(state, IndexedEvent(idx=0, event=event))
-    mi = new_state.meta_instances[meta.meta_instance_id]
-    assert mi.consecutive_failures == 0  # unchanged
-
-
-def test_satisfies_ignores_meta_instance_id_binding():
-    """instance_satisfies_meta_instance checks constraints only, not binding."""
-    meta = _meta_instance()
-    _, inst = _instance(node_ids=["node-a"])  # no meta_instance_id set
-    # Should match on constraints (model, min_nodes) regardless of binding
-    assert instance_satisfies_meta_instance(meta, inst) is True
-
-
-def test_find_unsatisfied_uses_binding_not_constraints():
-    """find_unsatisfied checks meta_instance_id binding, not just constraint matching."""
-    meta = _meta_instance()
-    # Instance matches constraints but is NOT bound to this meta_instance
-    iid, inst = _instance(node_ids=["node-a"])
-    topology = _topology("node-a")
-    result = find_unsatisfied_meta_instances(
-        {meta.meta_instance_id: meta}, {iid: inst}, topology
-    )
-    # Should be unsatisfied because instance.meta_instance_id != meta.meta_instance_id
-    assert list(result) == [meta]
-
-
-# =============================================================================
-# 5. Concurrent / multi-instance scenarios
-# =============================================================================
-
-
-async def test_health_reconciler_handles_multiple_failing_instances():
-    """Multiple instances failing simultaneously should each get their own event."""
-    meta_a = _meta_instance()
-    meta_b = _meta_instance()
-    iid_a, inst_a = _instance(node_ids=["node-a"], meta_instance_id=meta_a.meta_instance_id)
-    iid_b, inst_b = _instance(node_ids=["node-b"], meta_instance_id=meta_b.meta_instance_id)
-    runner_ids_a = list(inst_a.shard_assignments.node_to_runner.values())
-    runner_ids_b = list(inst_b.shard_assignments.node_to_runner.values())
-    state = State(
-        meta_instances={
-            meta_a.meta_instance_id: meta_a,
-            meta_b.meta_instance_id: meta_b,
-        },
-        instances={iid_a: inst_a, iid_b: inst_b},
-        runners={
-            runner_ids_a[0]: RunnerFailed(error_message="OOM"),
-            runner_ids_b[0]: RunnerFailed(error_message="OOM"),
-        },
-        topology=_topology("node-a", "node-b"),
-    )
-    reconciler = InstanceHealthReconciler()
-    events = await reconciler.reconcile(state)
-    assert len(events) == 2
-    # Both should be InstanceRetrying since failures < MAX
-    assert all(isinstance(e, InstanceRetrying) for e in events)
-    instance_ids = {e.instance_id for e in events}  # type: ignore[union-attr]
-    assert instance_ids == {iid_a, iid_b}
-
-
-async def test_health_reconciler_mixed_healthy_and_failing():
-    """Only failing instances should produce events; healthy ones should not."""
-    meta_healthy = _meta_instance()
-    meta_failing = _meta_instance()
-    iid_h, inst_h = _instance(
-        node_ids=["node-a"], meta_instance_id=meta_healthy.meta_instance_id
-    )
-    iid_f, inst_f = _instance(
-        node_ids=["node-b"], meta_instance_id=meta_failing.meta_instance_id
-    )
-    runner_ids_h = list(inst_h.shard_assignments.node_to_runner.values())
-    runner_ids_f = list(inst_f.shard_assignments.node_to_runner.values())
-    state = State(
-        meta_instances={
-            meta_healthy.meta_instance_id: meta_healthy,
-            meta_failing.meta_instance_id: meta_failing,
-        },
-        instances={iid_h: inst_h, iid_f: inst_f},
-        runners={
-            runner_ids_h[0]: RunnerReady(),
-            runner_ids_f[0]: RunnerFailed(error_message="crash"),
-        },
-        topology=_topology("node-a", "node-b"),
-    )
-    reconciler = InstanceHealthReconciler()
-    events = await reconciler.reconcile(state)
-    assert len(events) == 1
-    assert isinstance(events[0], InstanceRetrying)
-    assert events[0].instance_id == iid_f
-
-
-async def test_meta_instance_reconciler_empty_state():
-    """MetaInstanceReconciler with no meta_instances should produce no events."""
-    state = State()
-    reconciler = MetaInstanceReconciler()
-    events = await reconciler.reconcile(state)
-    assert len(events) == 0
-
-
-# =============================================================================
-# 6. Placement error tracking
-# =============================================================================
-
-
-def test_placement_failed_sets_error():
-    """MetaInstancePlacementFailed should set placement_error on the MetaInstance."""
-    meta = _meta_instance()
-    state = State(meta_instances={meta.meta_instance_id: meta})
-    event = MetaInstancePlacementFailed(
-        meta_instance_id=meta.meta_instance_id,
-        reason="Not enough memory",
-    )
-    new_state = apply(state, IndexedEvent(idx=0, event=event))
-    mi = new_state.meta_instances[meta.meta_instance_id]
-    assert mi.placement_error == "Not enough memory"
-
-
-def test_instance_created_clears_placement_error():
-    """InstanceCreated should clear placement_error on the MetaInstance."""
-    meta = _meta_instance(placement_error="Not enough memory")
-    _, inst = _instance(node_ids=["node-a"], meta_instance_id=meta.meta_instance_id)
-    state = State(meta_instances={meta.meta_instance_id: meta})
-    state = apply(state, IndexedEvent(idx=0, event=InstanceCreated(instance=inst)))
-    mi = state.meta_instances[meta.meta_instance_id]
-    assert mi.placement_error is None
-
-
-def test_placement_error_does_not_increment_failures():
-    """Placement failures should only set placement_error, not increment consecutive_failures."""
-    meta = _meta_instance()
-    state = State(meta_instances={meta.meta_instance_id: meta})
-    event = MetaInstancePlacementFailed(
-        meta_instance_id=meta.meta_instance_id,
-        reason="No resources",
-    )
-    new_state = apply(state, IndexedEvent(idx=0, event=event))
-    mi = new_state.meta_instances[meta.meta_instance_id]
-    assert mi.consecutive_failures == 0
-    assert mi.placement_error == "No resources"
-
-
-# =============================================================================
-# 7. State serialization roundtrip
-# =============================================================================
-
-
-def test_state_with_meta_instances_serializes():
-    """State with meta_instances should serialize and deserialize correctly."""
-    meta = _meta_instance(consecutive_failures=2, last_failure_error="test")
-    iid, inst = _instance(node_ids=["node-a"], meta_instance_id=meta.meta_instance_id)
-    state = State(
-        meta_instances={meta.meta_instance_id: meta},
-        instances={iid: inst},
-    )
-    json_str = state.model_dump_json()
-    restored = State.model_validate_json(json_str)
-    assert meta.meta_instance_id in restored.meta_instances
-    mi = restored.meta_instances[meta.meta_instance_id]
-    assert mi.model_id == meta.model_id
-    assert mi.consecutive_failures == 2
-    assert mi.last_failure_error == "test"
-    assert iid in restored.instances
-    assert restored.instances[iid].meta_instance_id == meta.meta_instance_id

src/exo/master/tests/test_placement.py

@@ -239,7 +239,7 @@ def test_get_transition_events_no_change(instance: Instance):
     target_instances = {instance_id: instance}
 
     # act
-    events = get_transition_events(current_instances, target_instances)
+    events = get_transition_events(current_instances, target_instances, {})
 
     # assert
     assert len(events) == 0
@@ -252,7 +252,7 @@ def test_get_transition_events_create_instance(instance: Instance):
     target_instances: dict[InstanceId, Instance] = {instance_id: instance}
 
     # act
-    events = get_transition_events(current_instances, target_instances)
+    events = get_transition_events(current_instances, target_instances, {})
 
     # assert
     assert len(events) == 1
@@ -266,7 +266,7 @@ def test_get_transition_events_delete_instance(instance: Instance):
     target_instances: dict[InstanceId, Instance] = {}
 
     # act
-    events = get_transition_events(current_instances, target_instances)
+    events = get_transition_events(current_instances, target_instances, {})
 
     # assert
     assert len(events) == 1

src/exo/master/tests/test_reconcile.py

@@ -1,742 +0,0 @@
-from exo.master.process_managers.instance_health import InstanceHealthReconciler
-from exo.master.reconcile import (
-    find_unsatisfied_meta_instances,
-    instance_connections_healthy,
-    instance_runners_failed,
-    instance_satisfies_meta_instance,
-)
-from exo.shared.apply import apply
-from exo.shared.models.model_cards import ModelCard, ModelId, ModelTask
-from exo.shared.topology import Topology
-from exo.shared.types.common import Host, MetaInstanceId, NodeId
-from exo.shared.types.events import (
-    IndexedEvent,
-    InstanceCreated,
-    InstanceDeleted,
-    InstanceRetrying,
-    MetaInstanceCreated,
-    MetaInstanceDeleted,
-)
-from exo.shared.types.memory import Memory
-from exo.shared.types.meta_instance import MetaInstance
-from exo.shared.types.multiaddr import Multiaddr
-from exo.shared.types.state import State
-from exo.shared.types.topology import Connection, SocketConnection
-from exo.shared.types.worker.instances import (
-    InstanceId,
-    MlxRingInstance,
-)
-from exo.shared.types.worker.runners import (
-    RunnerFailed,
-    RunnerId,
-    RunnerLoading,
-    RunnerReady,
-    RunnerShutdown,
-    ShardAssignments,
-)
-from exo.shared.types.worker.shards import PipelineShardMetadata
-
-
-def _model_card(model_id: str = "test-org/test-model") -> ModelCard:
-    return ModelCard(
-        model_id=ModelId(model_id),
-        storage_size=Memory.from_kb(1000),
-        n_layers=10,
-        hidden_size=30,
-        supports_tensor=True,
-        tasks=[ModelTask.TextGeneration],
-    )
-
-
-def _topology(*node_ids: str, connect: bool = True) -> Topology:
-    """Build a topology with nodes connected in a bidirectional ring with unique IPs.
-
-    Node at index ``i`` gets IP ``10.0.0.{i+1}``. Edges go in both directions
-    between consecutive nodes (including wrap-around).
-    """
-    t = Topology()
-    nodes = [NodeId(n) for n in node_ids]
-    for n in nodes:
-        t.add_node(n)
-    if connect and len(nodes) > 1:
-        for i in range(len(nodes)):
-            j = (i + 1) % len(nodes)
-            t.add_connection(
-                Connection(
-                    source=nodes[i],
-                    sink=nodes[j],
-                    edge=SocketConnection(
-                        sink_multiaddr=Multiaddr(
-                            address=f"/ip4/10.0.0.{j + 1}/tcp/50000"
-                        )
-                    ),
-                )
-            )
-            t.add_connection(
-                Connection(
-                    source=nodes[j],
-                    sink=nodes[i],
-                    edge=SocketConnection(
-                        sink_multiaddr=Multiaddr(
-                            address=f"/ip4/10.0.0.{i + 1}/tcp/50000"
-                        )
-                    ),
-                )
-            )
-    return t
-
-
-def _meta_instance(
-    model_id: str = "test-org/test-model",
-    *,
-    min_nodes: int = 1,
-    node_ids: list[NodeId] | None = None,
-    meta_instance_id: MetaInstanceId | None = None,
-) -> MetaInstance:
-    return MetaInstance(
-        meta_instance_id=meta_instance_id or MetaInstanceId(),
-        model_id=ModelId(model_id),
-        min_nodes=min_nodes,
-        node_ids=node_ids,
-    )
-
-
-def _instance(
-    model_id: str = "test-org/test-model",
-    node_ids: list[str] | None = None,
-    instance_id: InstanceId | None = None,
-    meta_instance_id: MetaInstanceId | None = None,
-) -> tuple[InstanceId, MlxRingInstance]:
-    """Create a test instance with hosts_by_node matching ``_topology()`` IPs."""
-    iid = instance_id or InstanceId()
-    nodes = node_ids or ["node-a"]
-    n = len(nodes)
-    mc = _model_card(model_id)
-    ephemeral_port = 50000
-    node_to_runner = {NodeId(nd): RunnerId() for nd in nodes}
-    runner_to_shard = {
-        runner_id: PipelineShardMetadata(
-            model_card=mc,
-            device_rank=i,
-            world_size=n,
-            start_layer=0,
-            end_layer=mc.n_layers,
-            n_layers=mc.n_layers,
-        )
-        for i, runner_id in enumerate(node_to_runner.values())
-    }
-    # Build hosts_by_node with IPs matching _topology() convention:
-    # node at index idx has IP 10.0.0.{idx+1}
-    hosts_by_node: dict[NodeId, list[Host]] = {}
-    for r, node_str in enumerate(nodes):
-        hosts: list[Host] = []
-        for idx in range(n):
-            if idx == r:
-                hosts.append(Host(ip="0.0.0.0", port=ephemeral_port))
-            elif n > 1 and idx in ((r - 1) % n, (r + 1) % n):
-                hosts.append(Host(ip=f"10.0.0.{idx + 1}", port=ephemeral_port))
-            else:
-                hosts.append(Host(ip="198.51.100.1", port=0))
-        hosts_by_node[NodeId(node_str)] = hosts
-    return iid, MlxRingInstance(
-        instance_id=iid,
-        shard_assignments=ShardAssignments(
-            model_id=ModelId(model_id),
-            runner_to_shard=runner_to_shard,
-            node_to_runner=node_to_runner,
-        ),
-        hosts_by_node=hosts_by_node,
-        ephemeral_port=ephemeral_port,
-        meta_instance_id=meta_instance_id,
-    )
-
-
-# --- instance_satisfies_meta_instance (pure constraint matching) ---
-
-
-def test_satisfies_matching_model():
-    meta = _meta_instance()
-    _, inst = _instance(node_ids=["node-a"])
-    assert instance_satisfies_meta_instance(meta, inst) is True
-
-
-def test_not_satisfies_wrong_model():
-    meta = _meta_instance("test-org/model-a")
-    _, inst = _instance("test-org/model-b")
-    assert instance_satisfies_meta_instance(meta, inst) is False
-
-
-def test_not_satisfies_missing_required_node():
-    meta = _meta_instance(node_ids=[NodeId("node-c")])
-    _, inst = _instance(node_ids=["node-a", "node-b"])
-    assert instance_satisfies_meta_instance(meta, inst) is False
-
-
-def test_not_satisfies_fewer_than_min_nodes():
-    meta = _meta_instance(min_nodes=3)
-    _, inst = _instance(node_ids=["node-a", "node-b"])
-    assert instance_satisfies_meta_instance(meta, inst) is False
-
-
-def test_satisfies_with_node_ids_specified():
-    meta = _meta_instance(node_ids=[NodeId("node-a"), NodeId("node-b")], min_nodes=2)
-    _, inst = _instance(node_ids=["node-a", "node-b", "node-c"])
-    assert instance_satisfies_meta_instance(meta, inst) is True
-
-
-# --- instance_connections_healthy ---
-
-
-def test_healthy_single_node_present():
-    _, inst = _instance(node_ids=["node-a"])
-    topology = _topology("node-a")
-    assert instance_connections_healthy(inst, topology) is True
-
-
-def test_unhealthy_single_node_missing():
-    _, inst = _instance(node_ids=["node-a"])
-    topology = Topology()  # empty
-    assert instance_connections_healthy(inst, topology) is False
-
-
-def test_healthy_two_node_ring():
-    _, inst = _instance(node_ids=["node-a", "node-b"])
-    topology = _topology("node-a", "node-b")
-    assert instance_connections_healthy(inst, topology) is True
-
-
-def test_unhealthy_two_node_edge_removed():
-    """Nodes present but edge removed — ring broken."""
-    _, inst = _instance(node_ids=["node-a", "node-b"])
-    topology = _topology("node-a", "node-b", connect=False)
-    assert instance_connections_healthy(inst, topology) is False
-
-
-def test_unhealthy_two_node_ip_changed():
-    """Edge exists but with a different IP than instance was configured with."""
-    _, inst = _instance(node_ids=["node-a", "node-b"])
-    # Build topology with different IPs than _instance() expects
-    topology = Topology()
-    topology.add_node(NodeId("node-a"))
-    topology.add_node(NodeId("node-b"))
-    topology.add_connection(
-        Connection(
-            source=NodeId("node-a"),
-            sink=NodeId("node-b"),
-            edge=SocketConnection(
-                sink_multiaddr=Multiaddr(address="/ip4/192.168.99.99/tcp/50000")
-            ),
-        )
-    )
-    topology.add_connection(
-        Connection(
-            source=NodeId("node-b"),
-            sink=NodeId("node-a"),
-            edge=SocketConnection(
-                sink_multiaddr=Multiaddr(address="/ip4/192.168.99.98/tcp/50000")
-            ),
-        )
-    )
-    assert instance_connections_healthy(inst, topology) is False
-
-
-def test_healthy_three_node_ring():
-    _, inst = _instance(node_ids=["node-a", "node-b", "node-c"])
-    topology = _topology("node-a", "node-b", "node-c")
-    assert instance_connections_healthy(inst, topology) is True
-
-
-def test_unhealthy_three_node_one_edge_removed():
-    """Remove one edge from a three-node ring — instance unhealthy."""
-    _, inst = _instance(node_ids=["node-a", "node-b", "node-c"])
-    # Build topology with one direction of one edge missing
-    topology = Topology()
-    nodes = [NodeId("node-a"), NodeId("node-b"), NodeId("node-c")]
-    for n in nodes:
-        topology.add_node(n)
-    # Add all edges except node-a → node-b
-    topology.add_connection(
-        Connection(
-            source=nodes[1],
-            sink=nodes[0],
-            edge=SocketConnection(
-                sink_multiaddr=Multiaddr(address="/ip4/10.0.0.1/tcp/50000")
-            ),
-        )
-    )
-    topology.add_connection(
-        Connection(
-            source=nodes[1],
-            sink=nodes[2],
-            edge=SocketConnection(
-                sink_multiaddr=Multiaddr(address="/ip4/10.0.0.3/tcp/50000")
-            ),
-        )
-    )
-    topology.add_connection(
-        Connection(
-            source=nodes[2],
-            sink=nodes[1],
-            edge=SocketConnection(
-                sink_multiaddr=Multiaddr(address="/ip4/10.0.0.2/tcp/50000")
-            ),
-        )
-    )
-    topology.add_connection(
-        Connection(
-            source=nodes[2],
-            sink=nodes[0],
-            edge=SocketConnection(
-                sink_multiaddr=Multiaddr(address="/ip4/10.0.0.1/tcp/50000")
-            ),
-        )
-    )
-    topology.add_connection(
-        Connection(
-            source=nodes[0],
-            sink=nodes[2],
-            edge=SocketConnection(
-                sink_multiaddr=Multiaddr(address="/ip4/10.0.0.3/tcp/50000")
-            ),
-        )
-    )
-    # Missing: node-a → node-b (ip 10.0.0.2)
-    assert instance_connections_healthy(inst, topology) is False
-
-
-def test_unhealthy_node_missing_from_topology():
-    """Instance has a node that's not in the topology at all."""
-    _, inst = _instance(node_ids=["node-a", "node-b"])
-    topology = _topology("node-a")  # node-b not present
-    assert instance_connections_healthy(inst, topology) is False
-
-
-def test_healthy_extra_nodes_in_topology():
-    """Extra nodes in topology don't affect instance health."""
-    _, inst = _instance(node_ids=["node-a", "node-b"])
-    topology = _topology("node-a", "node-b", "node-c")
-    assert instance_connections_healthy(inst, topology) is True
-
-
-# --- find_unsatisfied_meta_instances ---
-
-
-def test_unsatisfied_no_meta_instances():
-    result = find_unsatisfied_meta_instances({}, {}, Topology())
-    assert list(result) == []
-
-
-def test_unsatisfied_one_satisfied():
-    meta = _meta_instance()
-    id_a, inst_a = _instance(meta_instance_id=meta.meta_instance_id)
-    topology = _topology("node-a")
-    result = find_unsatisfied_meta_instances(
-        {meta.meta_instance_id: meta},
-        {id_a: inst_a},
-        topology,
-    )
-    assert list(result) == []
-
-
-def test_unsatisfied_one_not_satisfied():
-    meta = _meta_instance("test-org/model-x")
-    id_a, inst_a = _instance("test-org/model-y")
-    topology = _topology("node-a")
-    result = find_unsatisfied_meta_instances(
-        {meta.meta_instance_id: meta}, {id_a: inst_a}, topology
-    )
-    assert list(result) == [meta]
-
-
-def test_unsatisfied_mix():
-    meta_satisfied = _meta_instance("test-org/model-a")
-    meta_unsatisfied = _meta_instance("test-org/model-b")
-    id_a, inst_a = _instance(
-        "test-org/model-a", meta_instance_id=meta_satisfied.meta_instance_id
-    )
-    topology = _topology("node-a")
-    result = find_unsatisfied_meta_instances(
-        {
-            meta_satisfied.meta_instance_id: meta_satisfied,
-            meta_unsatisfied.meta_instance_id: meta_unsatisfied,
-        },
-        {id_a: inst_a},
-        topology,
-    )
-    assert list(result) == [meta_unsatisfied]
-
-
-def test_unsatisfied_node_disconnect():
-    meta = _meta_instance()
-    id_a, inst_a = _instance(
-        node_ids=["node-a", "node-b"], meta_instance_id=meta.meta_instance_id
-    )
-    topology = _topology("node-a")  # node-b disconnected
-    result = find_unsatisfied_meta_instances(
-        {meta.meta_instance_id: meta},
-        {id_a: inst_a},
-        topology,
-    )
-    assert list(result) == [meta]
-
-
-def test_unsatisfied_edge_break():
-    """Instance exists but its connections broke — meta-instance becomes unsatisfied."""
-    meta = _meta_instance()
-    id_a, inst_a = _instance(
-        node_ids=["node-a", "node-b"], meta_instance_id=meta.meta_instance_id
-    )
-    topology = _topology("node-a", "node-b", connect=False)  # nodes present, no edges
-    result = find_unsatisfied_meta_instances(
-        {meta.meta_instance_id: meta},
-        {id_a: inst_a},
-        topology,
-    )
-    assert list(result) == [meta]
-
-
-def test_unsatisfied_idempotent():
-    meta = _meta_instance("test-org/model-x")
-    topology = _topology("node-a")
-    meta_instances = {meta.meta_instance_id: meta}
-    instances: dict[InstanceId, MlxRingInstance] = {}
-    result_1 = list(
-        find_unsatisfied_meta_instances(meta_instances, instances, topology)
-    )
-    result_2 = list(
-        find_unsatisfied_meta_instances(meta_instances, instances, topology)
-    )
-    assert result_1 == result_2
-
-
-def test_unsatisfied_exclusive_binding():
-    """Two MetaInstances for the same model: one is bound via meta_instance_id, the other is unsatisfied."""
-    meta_a = _meta_instance("test-org/model-x")
-    meta_b = _meta_instance("test-org/model-x")
-    id_inst, inst = _instance(
-        "test-org/model-x", meta_instance_id=meta_a.meta_instance_id
-    )
-    topology = _topology("node-a")
-    result = find_unsatisfied_meta_instances(
-        {
-            meta_a.meta_instance_id: meta_a,
-            meta_b.meta_instance_id: meta_b,
-        },
-        {id_inst: inst},
-        topology,
-    )
-    assert list(result) == [meta_b]
-
-
-# --- apply handlers ---
-
-
-def test_apply_meta_instance_created():
-    state = State()
-    meta = _meta_instance()
-    event = MetaInstanceCreated(meta_instance=meta)
-    new_state = apply(state, IndexedEvent(idx=0, event=event))
-    assert meta.meta_instance_id in new_state.meta_instances
-    assert new_state.meta_instances[meta.meta_instance_id] == meta
-
-
-def test_apply_meta_instance_deleted():
-    meta = _meta_instance()
-    state = State(meta_instances={meta.meta_instance_id: meta})
-    event = MetaInstanceDeleted(meta_instance_id=meta.meta_instance_id)
-    new_state = apply(state, IndexedEvent(idx=0, event=event))
-    assert meta.meta_instance_id not in new_state.meta_instances
-
-
-def test_apply_meta_instance_deleted_clears_failure_info():
-    meta = _meta_instance().model_copy(
-        update={"consecutive_failures": 2, "last_failure_error": "OOM"}
-    )
-    state = State(meta_instances={meta.meta_instance_id: meta})
-    event = MetaInstanceDeleted(meta_instance_id=meta.meta_instance_id)
-    new_state = apply(state, IndexedEvent(idx=0, event=event))
-    assert meta.meta_instance_id not in new_state.meta_instances
-
-
-# --- instance_runners_failed ---
-
-
-def test_runners_failed_all_failed():
-    """All runners in RunnerFailed -> instance is failed."""
-    _, inst = _instance(node_ids=["node-a", "node-b"])
-    runners = {
-        rid: RunnerFailed(error_message="OOM")
-        for rid in inst.shard_assignments.node_to_runner.values()
-    }
-    is_failed, error = instance_runners_failed(inst, runners, {})
-    assert is_failed is True
-    assert error is not None
-    assert "OOM" in error
-
-
-def test_runners_failed_mixed_failed_shutdown():
-    """One Failed + one Shutdown = failed."""
-    _, inst = _instance(node_ids=["node-a", "node-b"])
-    runner_ids = list(inst.shard_assignments.node_to_runner.values())
-    runners = {
-        runner_ids[0]: RunnerFailed(error_message="crash"),
-        runner_ids[1]: RunnerShutdown(),
-    }
-    is_failed, error = instance_runners_failed(inst, runners, {})
-    assert is_failed is True
-    assert error is not None
-    assert "crash" in error
-
-
-def test_runners_not_failed_all_shutdown():
-    """All Shutdown (graceful) = not a failure."""
-    _, inst = _instance(node_ids=["node-a"])
-    runners = {
-        rid: RunnerShutdown() for rid in inst.shard_assignments.node_to_runner.values()
-    }
-    is_failed, _ = instance_runners_failed(inst, runners, {})
-    assert is_failed is False
-
-
-def test_runners_not_failed_still_active():
-    """Some runners still active = not failed yet."""
-    _, inst = _instance(node_ids=["node-a", "node-b"])
-    runner_ids = list(inst.shard_assignments.node_to_runner.values())
-    runners = {
-        runner_ids[0]: RunnerFailed(error_message="OOM"),
-        runner_ids[1]: RunnerLoading(),
-    }
-    is_failed, _ = instance_runners_failed(inst, runners, {})
-    assert is_failed is False
-
-
-def test_runners_not_failed_no_status():
-    """Runner not yet reported = not failed."""
-    _, inst = _instance(node_ids=["node-a"])
-    is_failed, _ = instance_runners_failed(inst, {}, {})
-    assert is_failed is False
-
-
-def test_runners_not_failed_healthy():
-    """Runners in Ready state = not failed."""
-    _, inst = _instance(node_ids=["node-a"])
-    runners = {
-        rid: RunnerReady() for rid in inst.shard_assignments.node_to_runner.values()
-    }
-    is_failed, _ = instance_runners_failed(inst, runners, {})
-    assert is_failed is False
-
-
-# --- failure tracking in apply_instance_deleted ---
-
-
-def test_apply_instance_deleted_tracks_failure():
-    """InstanceDeleted with failure_error increments meta instance failure count."""
-    meta = _meta_instance()
-    iid, inst = _instance(node_ids=["node-a"], meta_instance_id=meta.meta_instance_id)
-    state = State(
-        meta_instances={meta.meta_instance_id: meta},
-        instances={iid: inst},
-    )
-    event = InstanceDeleted(instance_id=iid, failure_error="Runner OOM")
-    new_state = apply(state, IndexedEvent(idx=0, event=event))
-    mi = new_state.meta_instances[meta.meta_instance_id]
-    assert mi.consecutive_failures == 1
-    assert mi.last_failure_error == "Runner OOM"
-
-
-def test_apply_instance_deleted_increments_failure():
-    """Subsequent failures increment the counter."""
-    meta = _meta_instance().model_copy(
-        update={"consecutive_failures": 2, "last_failure_error": "previous error"}
-    )
-    iid, inst = _instance(node_ids=["node-a"], meta_instance_id=meta.meta_instance_id)
-    state = State(
-        meta_instances={meta.meta_instance_id: meta},
-        instances={iid: inst},
-    )
-    event = InstanceDeleted(instance_id=iid, failure_error="new error")
-    new_state = apply(state, IndexedEvent(idx=0, event=event))
-    mi = new_state.meta_instances[meta.meta_instance_id]
-    assert mi.consecutive_failures == 3
-    assert mi.last_failure_error == "new error"
-
-
-def test_apply_instance_deleted_no_failure_no_tracking():
-    """InstanceDeleted without failure_error does not track."""
-    meta = _meta_instance()
-    iid, inst = _instance(node_ids=["node-a"], meta_instance_id=meta.meta_instance_id)
-    state = State(
-        meta_instances={meta.meta_instance_id: meta},
-        instances={iid: inst},
-    )
-    event = InstanceDeleted(instance_id=iid)
-    new_state = apply(state, IndexedEvent(idx=0, event=event))
-    mi = new_state.meta_instances[meta.meta_instance_id]
-    assert mi.consecutive_failures == 0
-
-
-def test_apply_instance_deleted_orphan_no_tracking():
-    """InstanceDeleted for orphan instance (no meta_instance_id) does not track."""
-    iid, inst = _instance(node_ids=["node-a"])
-    state = State(instances={iid: inst})
-    event = InstanceDeleted(instance_id=iid, failure_error="crash")
-    new_state = apply(state, IndexedEvent(idx=0, event=event))
-    assert len(new_state.meta_instances) == 0
-
-
-# --- InstanceRetrying ---
-
-
-def test_apply_instance_retrying_removes_runners():
-    """InstanceRetrying removes the instance's runners from state but keeps the instance."""
-    meta = _meta_instance()
-    iid, inst = _instance(
-        node_ids=["node-a", "node-b"], meta_instance_id=meta.meta_instance_id
-    )
-    runner_ids = list(inst.shard_assignments.node_to_runner.values())
-    runners = {
-        runner_ids[0]: RunnerFailed(error_message="OOM"),
-        runner_ids[1]: RunnerShutdown(),
-    }
-    state = State(
-        meta_instances={meta.meta_instance_id: meta},
-        instances={iid: inst},
-        runners=runners,
-    )
-    event = InstanceRetrying(
-        instance_id=iid,
-        meta_instance_id=meta.meta_instance_id,
-        failure_error="OOM",
-    )
-    new_state = apply(state, IndexedEvent(idx=0, event=event))
-    # Instance still exists
-    assert iid in new_state.instances
-    # Runners removed
-    assert runner_ids[0] not in new_state.runners
-    assert runner_ids[1] not in new_state.runners
-
-
-def test_apply_instance_retrying_increments_failure():
-    """InstanceRetrying increments consecutive_failures on the MetaInstance."""
-    meta = _meta_instance()
-    iid, inst = _instance(node_ids=["node-a"], meta_instance_id=meta.meta_instance_id)
-    state = State(
-        meta_instances={meta.meta_instance_id: meta},
-        instances={iid: inst},
-    )
-    event = InstanceRetrying(
-        instance_id=iid,
-        meta_instance_id=meta.meta_instance_id,
-        failure_error="crash",
-    )
-    new_state = apply(state, IndexedEvent(idx=0, event=event))
-    mi = new_state.meta_instances[meta.meta_instance_id]
-    assert mi.consecutive_failures == 1
-    assert mi.last_failure_error == "crash"
-
-
-def test_apply_instance_retrying_skips_missing_runners():
-    """InstanceRetrying doesn't assert if runners haven't reported yet."""
-    meta = _meta_instance()
-    iid, inst = _instance(node_ids=["node-a"], meta_instance_id=meta.meta_instance_id)
-    # No runners in state at all
-    state = State(
-        meta_instances={meta.meta_instance_id: meta},
-        instances={iid: inst},
-    )
-    event = InstanceRetrying(
-        instance_id=iid,
-        meta_instance_id=meta.meta_instance_id,
-        failure_error="crash",
-    )
-    # Should not raise
-    new_state = apply(state, IndexedEvent(idx=0, event=event))
-    assert iid in new_state.instances
-
-
-def test_apply_instance_created_resets_failure_counter():
-    """InstanceCreated resets consecutive_failures but preserves last_failure_error."""
-    meta = _meta_instance().model_copy(
-        update={"consecutive_failures": 3, "last_failure_error": "old error"}
-    )
-    _, inst = _instance(node_ids=["node-a"], meta_instance_id=meta.meta_instance_id)
-    state = State(meta_instances={meta.meta_instance_id: meta})
-    event = InstanceCreated(instance=inst)
-    new_state = apply(state, IndexedEvent(idx=0, event=event))
-    mi = new_state.meta_instances[meta.meta_instance_id]
-    assert mi.consecutive_failures == 0
-    assert mi.last_failure_error == "old error"
-    assert mi.placement_error is None
-
-
-# --- InstanceHealthReconciler retry-vs-delete ---
-
-
-async def test_health_reconciler_retries_when_under_limit():
-    """InstanceHealthReconciler emits InstanceRetrying when consecutive_failures < 3."""
-    meta = _meta_instance()
-    iid, inst = _instance(node_ids=["node-a"], meta_instance_id=meta.meta_instance_id)
-    runner_ids = list(inst.shard_assignments.node_to_runner.values())
-    state = State(
-        meta_instances={meta.meta_instance_id: meta},
-        instances={iid: inst},
-        runners={runner_ids[0]: RunnerFailed(error_message="OOM")},
-        topology=_topology("node-a"),
-    )
-    reconciler = InstanceHealthReconciler()
-    events = await reconciler.reconcile(state)
-    assert len(events) == 1
-    assert isinstance(events[0], InstanceRetrying)
-    assert events[0].instance_id == iid
-    assert events[0].meta_instance_id == meta.meta_instance_id
-
-
-async def test_health_reconciler_deletes_when_limit_reached():
-    """InstanceHealthReconciler emits InstanceDeleted when consecutive_failures >= 3."""
-    meta = _meta_instance().model_copy(update={"consecutive_failures": 3})
-    iid, inst = _instance(node_ids=["node-a"], meta_instance_id=meta.meta_instance_id)
-    runner_ids = list(inst.shard_assignments.node_to_runner.values())
-    state = State(
-        meta_instances={meta.meta_instance_id: meta},
-        instances={iid: inst},
-        runners={runner_ids[0]: RunnerFailed(error_message="OOM")},
-        topology=_topology("node-a"),
-    )
-    reconciler = InstanceHealthReconciler()
-    events = await reconciler.reconcile(state)
-    assert len(events) == 1
-    assert isinstance(events[0], InstanceDeleted)
-
-
-async def test_health_reconciler_deletes_without_meta_instance():
-    """Instances without a MetaInstance are deleted immediately on runner failure."""
-    iid, inst = _instance(node_ids=["node-a"])
-    runner_ids = list(inst.shard_assignments.node_to_runner.values())
-    state = State(
-        instances={iid: inst},
-        runners={runner_ids[0]: RunnerFailed(error_message="crash")},
-        topology=_topology("node-a"),
-    )
-    reconciler = InstanceHealthReconciler()
-    events = await reconciler.reconcile(state)
-    assert len(events) == 1
-    assert isinstance(events[0], InstanceDeleted)
-
-
-async def test_health_reconciler_network_failure_always_deletes():
-    """Network failure always triggers InstanceDeleted regardless of retry count."""
-    meta = _meta_instance()
-    iid, inst = _instance(
-        node_ids=["node-a", "node-b"], meta_instance_id=meta.meta_instance_id
-    )
-    state = State(
-        meta_instances={meta.meta_instance_id: meta},
-        instances={iid: inst},
-        topology=_topology("node-a"),  # node-b missing
-    )
-    reconciler = InstanceHealthReconciler()
-    events = await reconciler.reconcile(state)
-    assert len(events) == 1
-    assert isinstance(events[0], InstanceDeleted)
-    assert events[0].failure_error == "Network connection lost"

src/exo/shared/apply.py

@@ -4,7 +4,7 @@ from datetime import datetime
 
 from loguru import logger
 
-from exo.shared.types.common import MetaInstanceId, NodeId
+from exo.shared.types.common import NodeId
 from exo.shared.types.events import (
     ChunkGenerated,
     Event,
@@ -12,12 +12,6 @@ from exo.shared.types.events import (
     InputChunkReceived,
     InstanceCreated,
     InstanceDeleted,
-    InstanceRetrying,
-    JacclSideChannelData,
-    JacclSideChannelGathered,
-    MetaInstanceCreated,
-    MetaInstanceDeleted,
-    MetaInstancePlacementFailed,
     NodeDownloadProgress,
     NodeGatheredInfo,
     NodeTimedOut,
@@ -34,7 +28,6 @@ from exo.shared.types.events import (
     TracesCollected,
     TracesMerged,
 )
-from exo.shared.types.meta_instance import MetaInstance
 from exo.shared.types.profiling import (
     NodeIdentity,
     NodeNetworkInfo,
@@ -73,22 +66,12 @@ def event_apply(event: Event, state: State) -> State:
             | InputChunkReceived()
             | TracesCollected()
             | TracesMerged()
-            | JacclSideChannelData()
-            | JacclSideChannelGathered()
         ):  # Pass-through events that don't modify state
             return state
         case InstanceCreated():
             return apply_instance_created(event, state)
         case InstanceDeleted():
             return apply_instance_deleted(event, state)
-        case InstanceRetrying():
-            return apply_instance_retrying(event, state)
-        case MetaInstanceCreated():
-            return apply_meta_instance_created(event, state)
-        case MetaInstanceDeleted():
-            return apply_meta_instance_deleted(event, state)
-        case MetaInstancePlacementFailed():
-            return apply_meta_instance_placement_failed(event, state)
         case NodeTimedOut():
             return apply_node_timed_out(event, state)
         case NodeDownloadProgress():
@@ -191,120 +174,20 @@ def apply_task_failed(event: TaskFailed, state: State) -> State:
     return state.model_copy(update={"tasks": new_tasks})
 
 
-def _update_meta_instance(
-    state: State, mid: MetaInstanceId, **fields: object
-) -> Mapping[MetaInstanceId, MetaInstance]:
-    mi = state.meta_instances[mid]
-    return {**state.meta_instances, mid: mi.model_copy(update=fields)}
-
-
 def apply_instance_created(event: InstanceCreated, state: State) -> State:
     instance = event.instance
     new_instances: Mapping[InstanceId, Instance] = {
         **state.instances,
         instance.instance_id: instance,
     }
-    update: dict[str, object] = {"instances": new_instances}
-    # Reset failure tracking when a new instance is created for a meta-instance
-    if instance.meta_instance_id and instance.meta_instance_id in state.meta_instances:
-        mi = state.meta_instances[instance.meta_instance_id]
-        if mi.placement_error is not None or mi.consecutive_failures > 0:
-            update["meta_instances"] = _update_meta_instance(
-                state,
-                instance.meta_instance_id,
-                placement_error=None,
-                consecutive_failures=0,
-            )
-    return state.model_copy(update=update)
+    return state.model_copy(update={"instances": new_instances})
 
 
 def apply_instance_deleted(event: InstanceDeleted, state: State) -> State:
-    deleted_instance = state.instances.get(event.instance_id)
     new_instances: Mapping[InstanceId, Instance] = {
         iid: inst for iid, inst in state.instances.items() if iid != event.instance_id
     }
-    update: dict[str, object] = {"instances": new_instances}
-
-    # Track failure on the MetaInstance itself
-    if (
-        event.failure_error
-        and deleted_instance
-        and deleted_instance.meta_instance_id
-        and deleted_instance.meta_instance_id in state.meta_instances
-    ):
-        mid = deleted_instance.meta_instance_id
-        mi = state.meta_instances[mid]
-        update["meta_instances"] = {
-            **state.meta_instances,
-            mid: mi.model_copy(
-                update={
-                    "consecutive_failures": mi.consecutive_failures + 1,
-                    "last_failure_error": event.failure_error,
-                }
-            ),
-        }
-
-    return state.model_copy(update=update)
-
-
-def apply_instance_retrying(event: InstanceRetrying, state: State) -> State:
-    """Runners failed but retry limit not reached — remove runners, keep instance."""
-    instance = state.instances.get(event.instance_id)
-    if instance is None:
-        return state
-
-    # Remove all runners belonging to this instance from state
-    runner_ids_to_remove = set(instance.shard_assignments.node_to_runner.values())
-    new_runners: Mapping[RunnerId, RunnerStatus] = {
-        rid: rs for rid, rs in state.runners.items() if rid not in runner_ids_to_remove
-    }
-
-    update: dict[str, object] = {"runners": new_runners}
-
-    # Increment failure count on the MetaInstance
-    if event.meta_instance_id in state.meta_instances:
-        update["meta_instances"] = _update_meta_instance(
-            state,
-            event.meta_instance_id,
-            consecutive_failures=state.meta_instances[
-                event.meta_instance_id
-            ].consecutive_failures
-            + 1,
-            last_failure_error=event.failure_error,
-        )
-
-    return state.model_copy(update=update)
-
-
-def apply_meta_instance_created(event: MetaInstanceCreated, state: State) -> State:
-    new_meta: Mapping[MetaInstanceId, MetaInstance] = {
-        **state.meta_instances,
-        event.meta_instance.meta_instance_id: event.meta_instance,
-    }
-    return state.model_copy(update={"meta_instances": new_meta})
-
-
-def apply_meta_instance_deleted(event: MetaInstanceDeleted, state: State) -> State:
-    new_meta: Mapping[MetaInstanceId, MetaInstance] = {
-        mid: mi
-        for mid, mi in state.meta_instances.items()
-        if mid != event.meta_instance_id
-    }
-    return state.model_copy(update={"meta_instances": new_meta})
-
-
-def apply_meta_instance_placement_failed(
-    event: MetaInstancePlacementFailed, state: State
-) -> State:
-    if event.meta_instance_id not in state.meta_instances:
-        return state
-    return state.model_copy(
-        update={
-            "meta_instances": _update_meta_instance(
-                state, event.meta_instance_id, placement_error=event.reason
-            )
-        }
-    )
+    return state.model_copy(update={"instances": new_instances})
 
 
 def apply_runner_status_updated(event: RunnerStatusUpdated, state: State) -> State:

src/exo/shared/models/model_cards.py

@@ -189,7 +189,6 @@ class ConfigData(BaseModel):
             ["MiniMaxM2ForCausalLM"],
             ["LlamaForCausalLM"],
             ["GptOssForCausalLM"],
-            ["Step3p5ForCausalLM"],
         ]
 
     @model_validator(mode="before")

src/exo/shared/types/api.py

@@ -3,10 +3,11 @@ from collections.abc import Generator
 from typing import Annotated, Any, Literal
 from uuid import uuid4
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, field_validator
+from pydantic_core import PydanticUseDefault
 
 from exo.shared.models.model_cards import ModelCard, ModelId
-from exo.shared.types.common import CommandId, MetaInstanceId, NodeId
+from exo.shared.types.common import CommandId, NodeId
 from exo.shared.types.memory import Memory
 from exo.shared.types.worker.instances import Instance, InstanceId, InstanceMeta
 from exo.shared.types.worker.shards import Sharding, ShardMetadata
@@ -198,7 +199,6 @@ class ChatCompletionRequest(BaseModel):
     top_p: float | None = None
     top_k: int | None = None
     tools: list[dict[str, Any]] | None = None
-    enable_thinking: bool | None = None
     tool_choice: str | dict[str, Any] | None = None
     parallel_tool_calls: bool | None = None
     user: str | None = None
@@ -227,6 +227,13 @@ class PlaceInstanceParams(BaseModel):
     instance_meta: InstanceMeta = InstanceMeta.MlxRing
     min_nodes: int = 1
 
+    @field_validator("sharding", "instance_meta", mode="plain")
+    @classmethod
+    def use_default(cls, v: object):
+        if not v or not isinstance(v, (Sharding, InstanceMeta)):
+            raise PydanticUseDefault()
+        return v
+
 
 class CreateInstanceParams(BaseModel):
     instance: Instance
@@ -262,26 +269,6 @@ class DeleteInstanceResponse(BaseModel):
     instance_id: InstanceId
 
 
-class CreateMetaInstanceParams(BaseModel):
-    model_id: ModelId
-    sharding: Sharding = Sharding.Pipeline
-    instance_meta: InstanceMeta = InstanceMeta.MlxRing
-    min_nodes: int = 1
-    node_ids: list[NodeId] | None = None
-
-
-class CreateMetaInstanceResponse(BaseModel):
-    message: str
-    command_id: CommandId
-    meta_instance_id: MetaInstanceId
-
-
-class DeleteMetaInstanceResponse(BaseModel):
-    message: str
-    command_id: CommandId
-    meta_instance_id: MetaInstanceId
-
-
 class AdvancedImageParams(BaseModel):
     seed: Annotated[int, Field(ge=0)] | None = None
     num_inference_steps: Annotated[int, Field(ge=1, le=100)] | None = None

src/exo/shared/types/commands.py

@@ -6,8 +6,7 @@ from exo.shared.types.api import (
     ImageGenerationTaskParams,
 )
 from exo.shared.types.chunks import InputImageChunk
-from exo.shared.types.common import CommandId, MetaInstanceId, NodeId
-from exo.shared.types.meta_instance import MetaInstance
+from exo.shared.types.common import CommandId, NodeId
 from exo.shared.types.text_generation import TextGenerationTaskParams
 from exo.shared.types.worker.instances import Instance, InstanceId, InstanceMeta
 from exo.shared.types.worker.shards import Sharding, ShardMetadata
@@ -49,12 +48,8 @@ class DeleteInstance(BaseCommand):
     instance_id: InstanceId
 
 
-class CreateMetaInstance(BaseCommand):
-    meta_instance: MetaInstance
-
-
-class DeleteMetaInstance(BaseCommand):
-    meta_instance_id: MetaInstanceId
+class TaskCancelled(BaseCommand):
+    cancelled_command_id: CommandId
 
 
 class TaskFinished(BaseCommand):
@@ -98,8 +93,7 @@ Command = (
     | PlaceInstance
     | CreateInstance
     | DeleteInstance
-    | CreateMetaInstance
-    | DeleteMetaInstance
+    | TaskCancelled
     | TaskFinished
     | SendInputChunk
 )

src/exo/shared/types/common.py

@@ -42,10 +42,6 @@ class CommandId(Id):
     pass
 
 
-class MetaInstanceId(Id):
-    """Identifier for a MetaInstance."""
-
-
 class Host(CamelCaseModel):
     ip: str
     port: int

src/exo/shared/types/events.py

@@ -1,14 +1,11 @@
-import base64
-from collections.abc import Mapping
 from datetime import datetime
-from typing import Annotated, final
+from typing import final
 
-from pydantic import BeforeValidator, Field, PlainSerializer
+from pydantic import Field
 
 from exo.shared.topology import Connection
 from exo.shared.types.chunks import GenerationChunk, InputImageChunk
-from exo.shared.types.common import CommandId, Id, MetaInstanceId, NodeId, SessionId
-from exo.shared.types.meta_instance import MetaInstance
+from exo.shared.types.common import CommandId, Id, NodeId, SessionId
 from exo.shared.types.tasks import Task, TaskId, TaskStatus
 from exo.shared.types.worker.downloads import DownloadProgress
 from exo.shared.types.worker.instances import Instance, InstanceId
@@ -17,28 +14,6 @@ from exo.utils.info_gatherer.info_gatherer import GatheredInfo
 from exo.utils.pydantic_ext import CamelCaseModel, FrozenModel, TaggedModel
 
 
-def _decode_base64_bytes(v: bytes | str) -> bytes:
-    if isinstance(v, bytes):
-        return v
-    return base64.b64decode(v)
-
-
-def _encode_base64_bytes(v: bytes) -> str:
-    return base64.b64encode(v).decode("ascii")
-
-
-Base64Bytes = Annotated[
-    bytes,
-    BeforeValidator(_decode_base64_bytes),
-    PlainSerializer(_encode_base64_bytes, return_type=str),
-]
-"""bytes that serialize to/from base64 strings in JSON.
-
-Needed because TaggedModel's wrap validator converts JSON→Python validation
-context, which breaks strict-mode bytes deserialization from JSON strings.
-"""
-
-
 class EventId(Id):
     """
     Newtype around `ID`
@@ -91,30 +66,6 @@ class InstanceCreated(BaseEvent):
 
 class InstanceDeleted(BaseEvent):
     instance_id: InstanceId
-    failure_error: str | None = None
-
-
-class MetaInstanceCreated(BaseEvent):
-    meta_instance: MetaInstance
-
-
-class MetaInstanceDeleted(BaseEvent):
-    meta_instance_id: MetaInstanceId
-
-
-@final
-class MetaInstancePlacementFailed(BaseEvent):
-    meta_instance_id: MetaInstanceId
-    reason: str
-
-
-@final
-class InstanceRetrying(BaseEvent):
-    """Runners failed but retry count is below the limit — restart runners, keep instance."""
-
-    instance_id: InstanceId
-    meta_instance_id: MetaInstanceId
-    failure_error: str
 
 
 class RunnerStatusUpdated(BaseEvent):
@@ -181,25 +132,6 @@ class TracesMerged(BaseEvent):
     traces: list[TraceEventData]
 
 
-@final
-class JacclSideChannelData(BaseEvent):
-    """A runner's local contribution to a JACCL SideChannel all_gather round."""
-
-    instance_id: InstanceId
-    runner_id: RunnerId
-    sequence: int
-    data: Base64Bytes
-
-
-@final
-class JacclSideChannelGathered(BaseEvent):
-    """Gathered result of a JACCL SideChannel all_gather round."""
-
-    instance_id: InstanceId
-    sequence: int
-    gathered_data: Mapping[RunnerId, Base64Bytes]
-
-
 Event = (
     TestEvent
     | TaskCreated
@@ -209,10 +141,6 @@ Event = (
     | TaskAcknowledged
     | InstanceCreated
     | InstanceDeleted
-    | InstanceRetrying
-    | MetaInstanceCreated
-    | MetaInstanceDeleted
-    | MetaInstancePlacementFailed
     | RunnerStatusUpdated
     | RunnerDeleted
     | NodeTimedOut
@@ -224,8 +152,6 @@ Event = (
     | TopologyEdgeDeleted
     | TracesCollected
     | TracesMerged
-    | JacclSideChannelData
-    | JacclSideChannelGathered
 )
 
 

src/exo/shared/types/meta_instance.py

@@ -1,25 +0,0 @@
-from typing import final
-
-from pydantic import Field
-
-from exo.shared.models.model_cards import ModelId
-from exo.shared.types.common import MetaInstanceId, NodeId
-from exo.shared.types.worker.instances import InstanceMeta
-from exo.shared.types.worker.shards import Sharding
-from exo.utils.pydantic_ext import FrozenModel
-
-
-@final
-class MetaInstance(FrozenModel):
-    """Declarative constraint: ensure an instance matching these parameters always exists."""
-
-    meta_instance_id: MetaInstanceId = Field(default_factory=MetaInstanceId)
-    model_id: ModelId
-    sharding: Sharding = Sharding.Pipeline
-    instance_meta: InstanceMeta = InstanceMeta.MlxRing
-    min_nodes: int = 1
-    node_ids: list[NodeId] | None = None
-    # Failure tracking
-    placement_error: str | None = None
-    consecutive_failures: int = 0
-    last_failure_error: str | None = None

src/exo/shared/types/state.py

@@ -6,8 +6,7 @@ from pydantic import ConfigDict, Field, field_serializer, field_validator
 from pydantic.alias_generators import to_camel
 
 from exo.shared.topology import Topology, TopologySnapshot
-from exo.shared.types.common import MetaInstanceId, NodeId
-from exo.shared.types.meta_instance import MetaInstance
+from exo.shared.types.common import NodeId
 from exo.shared.types.profiling import (
     DiskUsage,
     MemoryUsage,
@@ -42,7 +41,6 @@ class State(CamelCaseModel):
         arbitrary_types_allowed=True,
     )
     instances: Mapping[InstanceId, Instance] = {}
-    meta_instances: Mapping[MetaInstanceId, MetaInstance] = {}
     runners: Mapping[RunnerId, RunnerStatus] = {}
     downloads: Mapping[NodeId, Sequence[DownloadProgress]] = {}
     tasks: Mapping[TaskId, Task] = {}

src/exo/shared/types/tasks.py

@@ -24,6 +24,7 @@ class TaskStatus(str, Enum):
     Complete = "Complete"
     TimedOut = "TimedOut"
     Failed = "Failed"
+    Cancelled = "Cancelled"
 
 
 class BaseTask(TaggedModel):
@@ -60,6 +61,11 @@ class TextGeneration(BaseTask):  # emitted by Master
     error_message: str | None = Field(default=None)
 
 
+class CancelTask(BaseTask):
+    cancelled_task_id: TaskId
+    runner_id: RunnerId
+
+
 class ImageGeneration(BaseTask):  # emitted by Master
     command_id: CommandId
     task_params: ImageGenerationTaskParams
@@ -87,6 +93,7 @@ Task = (
     | LoadModel
     | StartWarmup
     | TextGeneration
+    | CancelTask
     | ImageGeneration
     | ImageEdits
     | Shutdown

src/exo/shared/types/text_generation.py

@@ -40,6 +40,5 @@ class TextGenerationTaskParams(BaseModel, frozen=True):
     stop: str | list[str] | None = None
     seed: int | None = None
     chat_template_messages: list[dict[str, Any]] | None = None
-    enable_thinking: bool | None = None
     logprobs: bool = False
     top_logprobs: int | None = None

src/exo/shared/types/worker/instances.py

@@ -2,7 +2,7 @@ from enum import Enum
 
 from pydantic import model_validator
 
-from exo.shared.types.common import Host, Id, MetaInstanceId, NodeId
+from exo.shared.types.common import Host, Id, NodeId
 from exo.shared.types.worker.runners import RunnerId, ShardAssignments, ShardMetadata
 from exo.utils.pydantic_ext import CamelCaseModel, TaggedModel
 
@@ -19,7 +19,6 @@ class InstanceMeta(str, Enum):
 class BaseInstance(TaggedModel):
     instance_id: InstanceId
     shard_assignments: ShardAssignments
-    meta_instance_id: MetaInstanceId | None = None
 
     def shard(self, runner_id: RunnerId) -> ShardMetadata | None:
         return self.shard_assignments.runner_to_shard.get(runner_id, None)

src/exo/utils/channels.py

@@ -125,7 +125,9 @@ class MpSender[T]:
             self._state.buffer.put(item, block=True)
 
     async def send_async(self, item: T) -> None:
-        await to_thread.run_sync(self.send, item, limiter=CapacityLimiter(1))
+        await to_thread.run_sync(
+            self.send, item, limiter=CapacityLimiter(1), abandon_on_cancel=True
+        )
 
     def close(self) -> None:
         if not self._state.closed.is_set():

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

@@ -35,9 +35,6 @@ from mlx_lm.models.qwen3_moe import Model as Qwen3MoeModel
 from mlx_lm.models.qwen3_moe import Qwen3MoeSparseMoeBlock
 from mlx_lm.models.qwen3_next import Model as Qwen3NextModel
 from mlx_lm.models.qwen3_next import Qwen3NextDecoderLayer, Qwen3NextSparseMoeBlock
-from mlx_lm.models.step3p5 import Model as Step35Model
-from mlx_lm.models.step3p5 import Step3p5MLP as Step35MLP
-from mlx_lm.models.step3p5 import Step3p5Model as Step35InnerModel
 from transformers.models.qwen3.modeling_qwen3 import Qwen3DecoderLayer
 
 from exo.shared.logging import logger
@@ -267,19 +264,6 @@ def pipeline_auto_parallel(
             )
         )
 
-    if isinstance(inner_model_instance, Step35InnerModel):
-        inner_model_instance.num_layers = len(layers)
-        sliding_layers = [
-            i for i, layer in enumerate(layers) if getattr(layer, "is_sliding", False)
-        ]
-        full_layers = [
-            i
-            for i, layer in enumerate(layers)
-            if not getattr(layer, "is_sliding", True)
-        ]
-        inner_model_instance._swa_idx = 0 if not sliding_layers else sliding_layers[0]
-        inner_model_instance._full_idx = 0 if not full_layers else full_layers[0]
-
     _set_layers(model, layers)
 
     assert isinstance(layers, list), (
@@ -443,14 +427,6 @@ def tensor_auto_parallel(
             all_to_sharded_linear_in_place,
             sharded_to_all_linear_in_place,
         )
-    elif isinstance(model, Step35Model):
-        tensor_parallel_sharding_strategy = Step35ShardingStrategy(
-            group,
-            all_to_sharded_linear,
-            sharded_to_all_linear,
-            all_to_sharded_linear_in_place,
-            sharded_to_all_linear_in_place,
-        )
     else:
         raise ValueError(f"Unsupported model type: {type(model)}")
 
@@ -1005,46 +981,3 @@ class GptOssShardingStrategy(TensorParallelShardingStrategy):
             layer.mlp.sharding_group = self.group  # pyright: ignore[reportAttributeAccessIssue]
             mx.eval(layer)
         return model
-
-
-class Step35ShardingStrategy(TensorParallelShardingStrategy):
-    def shard_model(
-        self,
-        model: nn.Module,
-        timeout_seconds: float,
-        on_timeout: TimeoutCallback | None,
-    ) -> nn.Module:
-        model = cast(Step35Model, model)
-
-        for layer in model.layers:
-            eval_with_timeout(
-                layer.parameters(), timeout_seconds / len(model.layers), on_timeout
-            )
-            layer.self_attn.q_proj = self.all_to_sharded_linear(layer.self_attn.q_proj)
-            layer.self_attn.k_proj = self.all_to_sharded_linear(layer.self_attn.k_proj)
-            layer.self_attn.v_proj = self.all_to_sharded_linear(layer.self_attn.v_proj)
-            layer.self_attn.o_proj = self.sharded_to_all_linear(layer.self_attn.o_proj)
-
-            layer.self_attn.num_heads //= self.N
-            layer.self_attn.num_kv_heads //= self.N
-
-            if getattr(layer.self_attn, "use_head_wise_attn_gate", False):
-                layer.self_attn.g_proj = self.all_to_sharded_linear(
-                    layer.self_attn.g_proj
-                )
-
-            if isinstance(layer.mlp, Step35MLP):
-                layer.mlp.gate_proj = self.all_to_sharded_linear(layer.mlp.gate_proj)
-                layer.mlp.up_proj = self.all_to_sharded_linear(layer.mlp.up_proj)
-                layer.mlp.down_proj = self.sharded_to_all_linear(layer.mlp.down_proj)
-            else:
-                layer.mlp.sharding_group = self.group
-                self.all_to_sharded_linear_in_place(layer.mlp.share_expert.gate_proj)
-                self.all_to_sharded_linear_in_place(layer.mlp.share_expert.up_proj)
-                self.sharded_to_all_linear_in_place(layer.mlp.share_expert.down_proj)
-                self.all_to_sharded_linear_in_place(layer.mlp.switch_mlp.gate_proj)
-                self.all_to_sharded_linear_in_place(layer.mlp.switch_mlp.up_proj)
-                self.sharded_to_all_linear_in_place(layer.mlp.switch_mlp.down_proj)
-
-            mx.eval(layer)
-        return model

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

@@ -64,8 +64,6 @@ from exo.worker.runner.bootstrap import logger
 Group = mx.distributed.Group
 
 
-# TODO: Test this
-#  ALSO https://github.com/exo-explore/exo/pull/233#discussion_r2549683673
 def get_weights_size(model_shard_meta: ShardMetadata) -> Memory:
     return Memory.from_float_kb(
         (model_shard_meta.end_layer - model_shard_meta.start_layer)
@@ -83,30 +81,6 @@ class ModelLoadingTimeoutError(Exception):
     pass
 
 
-def mx_barrier(group: Group | None = None):
-    mx.eval(
-        mx.distributed.all_sum(
-            mx.array(1.0),
-            stream=mx.default_stream(mx.Device(mx.cpu)),
-            group=group,
-        )
-    )
-
-
-def broadcast_from_zero(value: int, group: Group | None = None):
-    if group is None:
-        return value
-
-    if group.rank() == 0:
-        a = mx.array([value], dtype=mx.int32)
-    else:
-        a = mx.array([0], dtype=mx.int32)
-
-    m = mx.distributed.all_sum(a, stream=mx.Device(mx.DeviceType.cpu), group=group)
-    mx.eval(m)
-    return int(m.item())
-
-
 class HostList(RootModel[list[str]]):
     @classmethod
     def from_hosts(cls, hosts: list[Host]) -> "HostList":
@@ -462,19 +436,11 @@ def apply_chat_template(
         partial_assistant_content = cast(str, formatted_messages[-1].get("content", ""))
         formatted_messages = formatted_messages[:-1]
 
-    extra_kwargs: dict[str, Any] = {}
-    if task_params.enable_thinking is not None:
-        # Qwen3 and GLM use "enable_thinking"; DeepSeek uses "thinking".
-        # Jinja ignores unknown variables, so passing both is safe.
-        extra_kwargs["enable_thinking"] = task_params.enable_thinking
-        extra_kwargs["thinking"] = task_params.enable_thinking
-
     prompt: str = tokenizer.apply_chat_template(
         formatted_messages,
         tokenize=False,
         add_generation_prompt=True,
         tools=task_params.tools,
-        **extra_kwargs,
     )
 
     if partial_assistant_content:
@@ -591,3 +557,23 @@ def mlx_cleanup(
     import gc
 
     gc.collect()
+
+
+def mx_any(bool_: bool, group: Group | None) -> bool:
+    if group is None:
+        return bool_
+    num_true = mx.distributed.all_sum(
+        mx.array(bool_), group=group, stream=mx.default_stream(mx.Device(mx.cpu))
+    )
+    mx.eval(num_true)
+    return num_true.item() > 0
+
+
+def mx_barrier(group: Group | None):
+    if group is None:
+        return
+    mx.eval(
+        mx.distributed.all_sum(
+            mx.array(1.0), group=group, stream=mx.default_stream(mx.Device(mx.cpu))
+        )
+    )

src/exo/worker/main.py

@@ -24,7 +24,6 @@ from exo.shared.types.events import (
     ForwarderEvent,
     IndexedEvent,
     InputChunkReceived,
-    JacclSideChannelGathered,
     NodeGatheredInfo,
     TaskCreated,
     TaskStatusUpdated,
@@ -34,6 +33,7 @@ from exo.shared.types.events import (
 from exo.shared.types.multiaddr import Multiaddr
 from exo.shared.types.state import State
 from exo.shared.types.tasks import (
+    CancelTask,
     CreateRunner,
     DownloadModel,
     ImageEdits,
@@ -159,15 +159,6 @@ class Worker:
                 for idx, event in indexed_events:
                     self.state = apply(self.state, IndexedEvent(idx=idx, event=event))
 
-                    # Dispatch JACCL gathered events to the relevant RunnerSupervisor
-                    if isinstance(event, JacclSideChannelGathered):
-                        for runner in self.runners.values():
-                            if (
-                                runner.bound_instance.instance.instance_id
-                                == event.instance_id
-                            ):
-                                runner.notify_gathered(event)
-
                     # Buffer input image chunks for image editing
                     if isinstance(event, InputChunkReceived):
                         cmd_id = event.command_id
@@ -234,15 +225,22 @@ class Worker:
                         )
                     )
                 case Shutdown(runner_id=runner_id):
+                    runner = self.runners.pop(runner_id)
                     try:
                         with fail_after(3):
-                            await self.runners.pop(runner_id).start_task(task)
+                            await runner.start_task(task)
                     except TimeoutError:
                         await self.event_sender.send(
                             TaskStatusUpdated(
                                 task_id=task.task_id, task_status=TaskStatus.TimedOut
                             )
                         )
+                    finally:
+                        runner.shutdown()
+                case CancelTask(
+                    cancelled_task_id=cancelled_task_id, runner_id=runner_id
+                ):
+                    await self.runners[runner_id].cancel_task(cancelled_task_id)
                 case ImageEdits() if task.task_params.total_input_chunks > 0:
                     # Assemble image from chunks and inject into task
                     cmd_id = task.command_id
@@ -280,18 +278,18 @@ class Worker:
                         del self.input_chunk_buffer[cmd_id]
                     if cmd_id in self.input_chunk_counts:
                         del self.input_chunk_counts[cmd_id]
-                    await self.runners[self._task_to_runner_id(task)].start_task(
-                        modified_task
-                    )
+                    await self._start_runner_task(modified_task)
                 case task:
-                    await self.runners[self._task_to_runner_id(task)].start_task(task)
+                    await self._start_runner_task(task)
 
     def shutdown(self):
         self._tg.cancel_scope.cancel()
 
-    def _task_to_runner_id(self, task: Task):
-        instance = self.state.instances[task.instance_id]
-        return instance.shard_assignments.node_to_runner[self.node_id]
+    async def _start_runner_task(self, task: Task):
+        if (instance := self.state.instances.get(task.instance_id)) is not None:
+            await self.runners[
+                instance.shard_assignments.node_to_runner[self.node_id]
+            ].start_task(task)
 
     async def _nack_request(self, since_idx: int) -> None:
         # We request all events after (and including) the missing index.
@@ -330,8 +328,6 @@ class Worker:
             for event in self.out_for_delivery.copy().values():
                 await self.local_event_sender.send(event)
 
-    ## Op Executors
-
     def _create_supervisor(self, task: CreateRunner) -> RunnerSupervisor:
         """Creates and stores a new AssignedRunner with initial downloading status."""
         runner = RunnerSupervisor.create(

src/exo/worker/plan.py

@@ -4,6 +4,7 @@ from collections.abc import Mapping, Sequence
 
 from exo.shared.types.common import CommandId, NodeId
 from exo.shared.types.tasks import (
+    CancelTask,
     ConnectToGroup,
     CreateRunner,
     DownloadModel,
@@ -34,7 +35,6 @@ from exo.shared.types.worker.runners import (
     RunnerLoading,
     RunnerReady,
     RunnerRunning,
-    RunnerShutdown,
     RunnerStatus,
     RunnerWarmingUp,
 )
@@ -54,13 +54,14 @@ def plan(
 ) -> Task | None:
     # Python short circuiting OR logic should evaluate these sequentially.
     return (
-        _kill_runner(runners, all_runners, instances)
-        or _create_runner(node_id, runners, instances, all_runners)
+        _cancel_tasks(runners, tasks)
+        or _kill_runner(runners, all_runners, instances)
+        or _create_runner(node_id, runners, instances)
         or _model_needs_download(node_id, runners, global_download_status)
         or _init_distributed_backend(runners, all_runners)
         or _load_model(runners, all_runners, global_download_status)
         or _ready_to_warmup(runners, all_runners)
-        or _pending_tasks(runners, tasks, all_runners, input_chunk_buffer)
+        or _pending_tasks(runners, tasks, all_runners, input_chunk_buffer or {})
     )
 
 
@@ -74,12 +75,6 @@ def _kill_runner(
         if (instance_id := runner.bound_instance.instance.instance_id) not in instances:
             return Shutdown(instance_id=instance_id, runner_id=runner_id)
 
-        # Master removed our runner from state (retry signal) and process is dead
-        if runner_id not in all_runners and isinstance(
-            runner.status, (RunnerFailed, RunnerShutdown)
-        ):
-            return Shutdown(instance_id=instance_id, runner_id=runner_id)
-
         for (
             global_runner_id
         ) in runner.bound_instance.instance.shard_assignments.node_to_runner.values():
@@ -97,7 +92,6 @@ def _create_runner(
     node_id: NodeId,
     runners: Mapping[RunnerId, RunnerSupervisor],
     instances: Mapping[InstanceId, Instance],
-    all_runners: Mapping[RunnerId, RunnerStatus],
 ) -> CreateRunner | None:
     for instance in instances.values():
         runner_id = instance.shard_assignments.node_to_runner.get(node_id, None)
@@ -107,16 +101,6 @@ def _create_runner(
         if runner_id in runners:
             continue
 
-        # Don't create while any peer runner is in a terminal state — wait for
-        # the master to emit InstanceRetrying which removes them from state.
-        has_terminal_peer = any(
-            isinstance(all_runners.get(peer_rid), (RunnerFailed, RunnerShutdown))
-            for peer_rid in instance.shard_assignments.node_to_runner.values()
-            if peer_rid != runner_id
-        )
-        if has_terminal_peer:
-            continue
-
         shard = instance.shard(runner_id)
         assert shard is not None
 
@@ -288,7 +272,7 @@ def _pending_tasks(
     runners: Mapping[RunnerId, RunnerSupervisor],
     tasks: Mapping[TaskId, Task],
     all_runners: Mapping[RunnerId, RunnerStatus],
-    input_chunk_buffer: Mapping[CommandId, dict[int, str]] | None = None,
+    input_chunk_buffer: Mapping[CommandId, dict[int, str]],
 ) -> Task | None:
     for task in tasks.values():
         # for now, just forward chat completions
@@ -302,7 +286,7 @@ def _pending_tasks(
         if isinstance(task, ImageEdits) and task.task_params.total_input_chunks > 0:
             cmd_id = task.command_id
             expected = task.task_params.total_input_chunks
-            received = len((input_chunk_buffer or {}).get(cmd_id, {}))
+            received = len(input_chunk_buffer.get(cmd_id, {}))
             if received < expected:
                 continue  # Wait for all chunks to arrive
 
@@ -310,16 +294,33 @@ def _pending_tasks(
             if task.instance_id != runner.bound_instance.instance.instance_id:
                 continue
 
-            # I have a design point here; this is a state race in disguise as the task status doesn't get updated to completed fast enough
-            # however, realistically the task status should be set to completed by the LAST runner, so this is a true race
-            # the actual solution is somewhat deeper than this bypass - TODO!
+            # the task status _should_ be set to completed by the LAST runner
+            # it is currently set by the first
+            # this is definitely a hack
             if task.task_id in runner.completed:
                 continue
 
-            # TODO: Check ordering aligns with MLX distributeds expectations.
-
             if isinstance(runner.status, RunnerReady) and all(
                 isinstance(all_runners[global_runner_id], (RunnerReady, RunnerRunning))
                 for global_runner_id in runner.bound_instance.instance.shard_assignments.runner_to_shard
             ):
                 return task
+
+
+def _cancel_tasks(
+    runners: Mapping[RunnerId, RunnerSupervisor],
+    tasks: Mapping[TaskId, Task],
+) -> Task | None:
+    for task in tasks.values():
+        if task.task_status != TaskStatus.Cancelled:
+            continue
+        for runner_id, runner in runners.items():
+            if task.instance_id != runner.bound_instance.instance.instance_id:
+                continue
+            if task.task_id in runner.cancelled:
+                continue
+            return CancelTask(
+                instance_id=task.instance_id,
+                cancelled_task_id=task.task_id,
+                runner_id=runner_id,
+            )

src/exo/worker/runner/bootstrap.py

@@ -3,7 +3,7 @@ import os
 import loguru
 
 from exo.shared.types.events import Event, RunnerStatusUpdated
-from exo.shared.types.tasks import Task
+from exo.shared.types.tasks import Task, TaskId
 from exo.shared.types.worker.instances import BoundInstance, MlxJacclInstance
 from exo.shared.types.worker.runners import RunnerFailed
 from exo.utils.channels import ClosedResourceError, MpReceiver, MpSender
@@ -15,8 +15,8 @@ def entrypoint(
     bound_instance: BoundInstance,
     event_sender: MpSender[Event],
     task_receiver: MpReceiver[Task],
+    cancel_receiver: MpReceiver[TaskId],
     _logger: "loguru.Logger",
-    pipe_fifo_paths: tuple[str, str] | None = None,
 ) -> None:
     fast_synch_override = os.environ.get("EXO_FAST_SYNCH")
     if fast_synch_override == "on" or (
@@ -30,16 +30,6 @@ def entrypoint(
     else:
         os.environ["MLX_METAL_FAST_SYNCH"] = "0"
 
-    # Open JACCL FIFOs by path and set env vars for C++ SideChannel.
-    # Named pipes (FIFOs) work across multiprocessing spawn (macOS default).
-    if pipe_fifo_paths is not None:
-        fifo_c2p, fifo_p2c = pipe_fifo_paths
-        # C++ reads gathered data from p2c (PIPE_IN), writes local data to c2p (PIPE_OUT)
-        pipe_in_fd = os.open(fifo_p2c, os.O_RDONLY)
-        pipe_out_fd = os.open(fifo_c2p, os.O_WRONLY)
-        os.environ["MLX_JACCL_PIPE_IN"] = str(pipe_in_fd)
-        os.environ["MLX_JACCL_PIPE_OUT"] = str(pipe_out_fd)
-
     global logger
     logger = _logger
 
@@ -49,7 +39,7 @@ def entrypoint(
     try:
         from exo.worker.runner.runner import main
 
-        main(bound_instance, event_sender, task_receiver)
+        main(bound_instance, event_sender, task_receiver, cancel_receiver)
     except ClosedResourceError:
         logger.warning("Runner communication closed unexpectedly")
     except Exception as e:

src/exo/worker/runner/runner.py

@@ -1,5 +1,6 @@
 import base64
 import json
+import math
 import resource
 import time
 from collections.abc import Generator
@@ -88,6 +89,7 @@ from exo.worker.engines.mlx.utils_mlx import (
     initialize_mlx,
     load_mlx_items,
     mlx_force_oom,
+    mx_any,
 )
 from exo.worker.runner.bootstrap import logger
 
@@ -112,6 +114,7 @@ def main(
     bound_instance: BoundInstance,
     event_sender: MpSender[Event],
     task_receiver: MpReceiver[Task],
+    cancel_receiver: MpReceiver[TaskId],
 ):
     soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE)
     resource.setrlimit(resource.RLIMIT_NOFILE, (min(max(soft, 2048), hard), hard))
@@ -129,11 +132,15 @@ def main(
         time.sleep(timeout)
 
     setup_start_time = time.time()
+    cancelled_tasks = set[TaskId]()
 
-    model: Model | DistributedImageModel | None = None
+    # type checker was unhappy with me - splitting these fixed it
+    inference_model: Model | None = None
+    image_model: DistributedImageModel | None = None
     tokenizer = None
     group = None
     kv_prefix_cache: KVPrefixCache | None = None
+    check_for_cancel_every: int | None = None
 
     current_status: RunnerStatus = RunnerIdle()
     logger.info("runner created")
@@ -146,6 +153,7 @@ def main(
             if task.task_id in seen:
                 logger.warning("repeat task - potential error")
             seen.add(task.task_id)
+            cancelled_tasks.discard(TaskId("CANCEL_CURRENT_TASK"))
             event_sender.send(
                 TaskStatusUpdated(task_id=task.task_id, task_status=TaskStatus.Running)
             )
@@ -191,7 +199,7 @@ def main(
                         time.sleep(0.5)
 
                     if ModelTask.TextGeneration in shard_metadata.model_card.tasks:
-                        model, tokenizer = load_mlx_items(
+                        inference_model, tokenizer = load_mlx_items(
                             bound_instance, group, on_timeout=on_model_load_timeout
                         )
                         logger.info(
@@ -203,7 +211,7 @@ def main(
                         ModelTask.TextToImage in shard_metadata.model_card.tasks
                         or ModelTask.ImageToImage in shard_metadata.model_card.tasks
                     ):
-                        model = initialize_image_model(bound_instance)
+                        image_model = initialize_image_model(bound_instance)
                     else:
                         raise ValueError(
                             f"Unknown model task(s): {shard_metadata.model_card.tasks}"
@@ -211,8 +219,6 @@ def main(
                     current_status = RunnerLoaded()
                     logger.info("runner loaded")
                 case StartWarmup() if isinstance(current_status, RunnerLoaded):
-                    assert model
-
                     current_status = RunnerWarmingUp()
                     logger.info("runner warming up")
                     event_sender.send(
@@ -224,16 +230,31 @@ def main(
 
                     logger.info(f"warming up inference for instance: {instance}")
                     if ModelTask.TextGeneration in shard_metadata.model_card.tasks:
-                        assert not isinstance(model, DistributedImageModel)
+                        assert inference_model
                         assert tokenizer
 
+                        t = time.perf_counter()
                         toks = warmup_inference(
-                            model=model,
+                            model=inference_model,
                             tokenizer=tokenizer,
                             group=group,
-                            # kv_prefix_cache=kv_prefix_cache,  # supply for warmup-time prefix caching
                         )
                         logger.info(f"warmed up by generating {toks} tokens")
+                        check_for_cancel_every = min(
+                            math.ceil(toks / (time.perf_counter() - t)), 100
+                        )
+                        if group is not None:
+                            check_for_cancel_every = int(
+                                mx.max(
+                                    mx.distributed.all_gather(
+                                        mx.array([check_for_cancel_every]), group=group
+                                    )
+                                ).item()
+                            )
+
+                        logger.info(
+                            f"runner checking for cancellation every {check_for_cancel_every} tokens"
+                        )
                         logger.info(
                             f"runner initialized in {time.time() - setup_start_time} seconds"
                         )
@@ -241,8 +262,8 @@ def main(
                         ModelTask.TextToImage in shard_metadata.model_card.tasks
                         or ModelTask.ImageToImage in shard_metadata.model_card.tasks
                     ):
-                        assert isinstance(model, DistributedImageModel)
-                        image = warmup_image_generator(model=model)
+                        assert image_model
+                        image = warmup_image_generator(model=image_model)
                         if image is not None:
                             logger.info(f"warmed up by generating {image.size} image")
                         else:
@@ -262,9 +283,9 @@ def main(
                         )
                     )
                     event_sender.send(TaskAcknowledged(task_id=task.task_id))
-
-                    assert model and not isinstance(model, DistributedImageModel)
+                    assert inference_model
                     assert tokenizer
+                    assert check_for_cancel_every
 
                     try:
                         _check_for_debug_prompts(task_params)
@@ -274,7 +295,7 @@ def main(
 
                         # Generate responses using the actual MLX generation
                         mlx_generator = mlx_generate(
-                            model=model,
+                            model=inference_model,
                             tokenizer=tokenizer,
                             task=task_params,
                             prompt=prompt,
@@ -299,11 +320,11 @@ def main(
                             patch_glm_tokenizer(tokenizer)
 
                         # GPT-OSS specific parsing to match other model formats.
-                        elif isinstance(model, GptOssModel):
+                        elif isinstance(inference_model, GptOssModel):
                             mlx_generator = parse_gpt_oss(mlx_generator)
 
                         if tokenizer.has_tool_calling and not isinstance(
-                            model, GptOssModel
+                            inference_model, GptOssModel
                         ):
                             assert tokenizer.tool_call_start
                             assert tokenizer.tool_call_end
@@ -316,7 +337,18 @@ def main(
                             )
 
                         completion_tokens = 0
+                        tokens_since_last_cancel_check = 0
                         for response in mlx_generator:
+                            tokens_since_last_cancel_check += 1
+                            if tokens_since_last_cancel_check >= check_for_cancel_every:
+                                tokens_since_last_cancel_check = 0
+                                cancelled_tasks.update(cancel_receiver.collect())
+                                want_to_cancel = (task.task_id in cancelled_tasks) or (
+                                    TaskId("CANCEL_CURRENT_TASK") in cancelled_tasks
+                                )
+                                if mx_any(want_to_cancel, group):
+                                    break
+
                             match response:
                                 case GenerationResponse():
                                     completion_tokens += 1
@@ -388,7 +420,7 @@ def main(
                 case ImageGeneration(
                     task_params=task_params, command_id=command_id
                 ) if isinstance(current_status, RunnerReady):
-                    assert isinstance(model, DistributedImageModel)
+                    assert image_model
                     logger.info(f"received image generation request: {str(task)[:500]}")
                     current_status = RunnerRunning()
                     logger.info("runner running")
@@ -401,7 +433,9 @@ def main(
 
                     try:
                         image_index = 0
-                        for response in generate_image(model=model, task=task_params):
+                        for response in generate_image(
+                            model=image_model, task=task_params
+                        ):
                             is_primary_output = _is_primary_output_node(shard_metadata)
 
                             if is_primary_output:
@@ -451,7 +485,7 @@ def main(
                 case ImageEdits(task_params=task_params, command_id=command_id) if (
                     isinstance(current_status, RunnerReady)
                 ):
-                    assert isinstance(model, DistributedImageModel)
+                    assert image_model
                     logger.info(f"received image edits request: {str(task)[:500]}")
                     current_status = RunnerRunning()
                     logger.info("runner running")
@@ -464,7 +498,9 @@ def main(
 
                     try:
                         image_index = 0
-                        for response in generate_image(model=model, task=task_params):
+                        for response in generate_image(
+                            model=image_model, task=task_params
+                        ):
                             if _is_primary_output_node(shard_metadata):
                                 match response:
                                     case PartialImageResponse():
@@ -530,7 +566,7 @@ def main(
                 RunnerStatusUpdated(runner_id=runner_id, runner_status=current_status)
             )
             if isinstance(current_status, RunnerShutdown):
-                del model, tokenizer, group
+                del inference_model, image_model, tokenizer, group
                 mx.clear_cache()
                 import gc
 

src/exo/worker/runner/runner_supervisor.py

@@ -1,10 +1,6 @@
 import contextlib
-import os
 import signal
-import struct
-import tempfile
 from dataclasses import dataclass, field
-from functools import partial
 from multiprocessing import Process
 from typing import Self
 
@@ -18,14 +14,12 @@ from loguru import logger
 
 from exo.shared.types.events import (
     Event,
-    JacclSideChannelData,
-    JacclSideChannelGathered,
     RunnerStatusUpdated,
     TaskAcknowledged,
     TaskStatusUpdated,
 )
 from exo.shared.types.tasks import Task, TaskId, TaskStatus
-from exo.shared.types.worker.instances import BoundInstance, MlxJacclInstance
+from exo.shared.types.worker.instances import BoundInstance
 from exo.shared.types.worker.runners import (
     RunnerConnecting,
     RunnerFailed,
@@ -40,26 +34,6 @@ from exo.shared.types.worker.shards import ShardMetadata
 from exo.utils.channels import MpReceiver, MpSender, Sender, mp_channel
 from exo.worker.runner.bootstrap import entrypoint
 
-
-def _pipe_read_exact(fd: int, n: int) -> bytes | None:
-    """Read exactly n bytes from a file descriptor. Returns None on EOF."""
-    data = b""
-    while len(data) < n:
-        chunk = os.read(fd, n - len(data))
-        if not chunk:
-            return None
-        data += chunk
-    return data
-
-
-def _pipe_write_all(fd: int, data: bytes) -> None:
-    """Write all bytes to a file descriptor."""
-    view = memoryview(data)
-    while view:
-        written = os.write(fd, view)
-        view = view[written:]
-
-
 PREFILL_TIMEOUT_SECONDS = 60
 DECODE_TIMEOUT_SECONDS = 5
 
@@ -73,18 +47,11 @@ class RunnerSupervisor:
     _ev_recv: MpReceiver[Event]
     _task_sender: MpSender[Task]
     _event_sender: Sender[Event]
-    _pipe_read_fd: int | None = None  # Python reads runner's pipe output
-    _pipe_write_fd: int | None = None  # Python writes gathered data to runner
-    _child_pipe_fds: tuple[int, int] | None = None  # fds to close after fork
-    _fifo_dir: str | None = None  # Temp dir for FIFO files (for cleanup)
-    _fifo_c2p: str | None = None  # FIFO path: C++ writes → Python reads
-    _fifo_p2c: str | None = None  # FIFO path: Python writes → C++ reads
+    _cancel_sender: MpSender[TaskId]
     status: RunnerStatus = field(default_factory=RunnerIdle, init=False)
     pending: dict[TaskId, anyio.Event] = field(default_factory=dict, init=False)
     completed: set[TaskId] = field(default_factory=set, init=False)
-    _gathered_waiters: dict[
-        int, tuple[anyio.Event, JacclSideChannelGathered | None]
-    ] = field(default_factory=dict, init=False)
+    cancelled: set[TaskId] = field(default_factory=set, init=False)
 
     @classmethod
     def create(
@@ -95,25 +62,8 @@ class RunnerSupervisor:
         initialize_timeout: float = 400,
     ) -> Self:
         ev_send, ev_recv = mp_channel[Event]()
-        # A task is kind of a runner command
         task_sender, task_recv = mp_channel[Task]()
-
-        # For MlxJaccl instances, create named pipes (FIFOs) for SideChannel relay.
-        # Named pipes work across multiprocessing.Process spawn (macOS default).
-        # FIFO c2p: C++ writes local data → Python reads it
-        # FIFO p2c: Python writes gathered data → C++ reads it
-        fifo_dir: str | None = None
-        fifo_c2p: str | None = None
-        fifo_p2c: str | None = None
-        pipe_fifo_paths: tuple[str, str] | None = None
-
-        if isinstance(bound_instance.instance, MlxJacclInstance):
-            fifo_dir = tempfile.mkdtemp(prefix="exo_jaccl_")
-            fifo_c2p = os.path.join(fifo_dir, "c2p")  # C++ → Python
-            fifo_p2c = os.path.join(fifo_dir, "p2c")  # Python → C++
-            os.mkfifo(fifo_c2p)
-            os.mkfifo(fifo_p2c)
-            pipe_fifo_paths = (fifo_c2p, fifo_p2c)
+        cancel_sender, cancel_recv = mp_channel[TaskId]()
 
         runner_process = Process(
             target=entrypoint,
@@ -121,8 +71,8 @@ class RunnerSupervisor:
                 bound_instance,
                 ev_send,
                 task_recv,
+                cancel_recv,
                 logger,
-                pipe_fifo_paths,
             ),
             daemon=True,
         )
@@ -136,53 +86,23 @@ class RunnerSupervisor:
             initialize_timeout=initialize_timeout,
             _ev_recv=ev_recv,
             _task_sender=task_sender,
+            _cancel_sender=cancel_sender,
             _event_sender=event_sender,
-            _fifo_dir=fifo_dir,
-            _fifo_c2p=fifo_c2p,
-            _fifo_p2c=fifo_p2c,
         )
 
         return self
 
     async def run(self):
         self.runner_process.start()
-
-        if self._fifo_c2p is not None and self._fifo_p2c is not None:
-            # Open FIFOs from parent side. These block until child opens the other end,
-            # so we run them in threads concurrently to avoid deadlock.
-            fifo_c2p = self._fifo_c2p
-            fifo_p2c = self._fifo_p2c
-
-            async def open_read() -> None:
-                self._pipe_read_fd = await to_thread.run_sync(
-                    partial(os.open, fifo_c2p, os.O_RDONLY)
-                )
-
-            async def open_write() -> None:
-                self._pipe_write_fd = await to_thread.run_sync(
-                    partial(os.open, fifo_p2c, os.O_WRONLY)
-                )
-
-            async with anyio.create_task_group() as open_tg:
-                open_tg.start_soon(open_read)
-                open_tg.start_soon(open_write)
-
-            logger.info(
-                f"JACCL pipe relay: FIFOs opened (read_fd={self._pipe_read_fd}, write_fd={self._pipe_write_fd})"
-            )
-
-            async with anyio.create_task_group() as tg:
-                tg.start_soon(self._pipe_relay)
-                tg.start_soon(self._forward_events)
-        else:
-            await self._forward_events()
+        await self._forward_events()
 
     def shutdown(self):
         logger.info("Runner supervisor shutting down")
         self._ev_recv.close()
         self._task_sender.close()
         self._event_sender.close()
-        self._close_pipe_fds()
+        self._cancel_sender.send(TaskId("CANCEL_CURRENT_TASK"))
+        self._cancel_sender.close()
         self.runner_process.join(1)
         if not self.runner_process.is_alive():
             logger.info("Runner process succesfully terminated")
@@ -198,14 +118,6 @@ class RunnerSupervisor:
         logger.critical("Runner process didn't respond to SIGTERM, killing")
         self.runner_process.kill()
 
-        self.runner_process.join(1)
-        if not self.runner_process.is_alive():
-            return
-
-        logger.critical(
-            "Runner process didn't respond to SIGKILL. System resources may have leaked"
-        )
-
     async def start_task(self, task: Task):
         if task.task_id in self.pending:
             logger.warning(
@@ -227,6 +139,17 @@ class RunnerSupervisor:
             return
         await event.wait()
 
+    async def cancel_task(self, task_id: TaskId):
+        if task_id in self.completed:
+            logger.info(f"Unable to cancel {task_id} as it has been completed")
+            return
+        self.cancelled.add(task_id)
+        with anyio.move_on_after(0.5) as scope:
+            await self._cancel_sender.send_async(task_id)
+        if scope.cancel_called:
+            logger.error("RunnerSupervisor cancel pipe blocked")
+            await self._check_runner(TimeoutError("cancel pipe blocked"))
+
     async def _forward_events(self):
         with self._ev_recv as events:
             try:
@@ -258,110 +181,6 @@ class RunnerSupervisor:
                 for tid in self.pending:
                     self.pending[tid].set()
 
-    def _close_pipe_fds(self) -> None:
-        if self._pipe_read_fd is not None:
-            with contextlib.suppress(OSError):
-                os.close(self._pipe_read_fd)
-            self._pipe_read_fd = None
-        if self._pipe_write_fd is not None:
-            with contextlib.suppress(OSError):
-                os.close(self._pipe_write_fd)
-            self._pipe_write_fd = None
-        if self._child_pipe_fds is not None:
-            for fd in self._child_pipe_fds:
-                with contextlib.suppress(OSError):
-                    os.close(fd)
-            self._child_pipe_fds = None
-        # Clean up FIFO files
-        if self._fifo_c2p is not None:
-            with contextlib.suppress(OSError):
-                os.unlink(self._fifo_c2p)
-            self._fifo_c2p = None
-        if self._fifo_p2c is not None:
-            with contextlib.suppress(OSError):
-                os.unlink(self._fifo_p2c)
-            self._fifo_p2c = None
-        if self._fifo_dir is not None:
-            with contextlib.suppress(OSError):
-                os.rmdir(self._fifo_dir)
-            self._fifo_dir = None
-
-    async def _pipe_relay(self) -> None:
-        """Relay JACCL SideChannel all_gather rounds between runner pipes and exo events."""
-        assert self._pipe_read_fd is not None
-        assert self._pipe_write_fd is not None
-        read_fd = self._pipe_read_fd
-        write_fd = self._pipe_write_fd
-        sequence = 0
-
-        try:
-            while True:
-                # 1. Read local data from runner: [uint32 size][size bytes]
-                header = await to_thread.run_sync(partial(_pipe_read_exact, read_fd, 4))
-                if header is None:
-                    logger.info("JACCL pipe relay: runner closed pipe (EOF)")
-                    break
-                data_size: int = struct.unpack("<I", header)[0]  # pyright: ignore[reportAny]
-                local_data = await to_thread.run_sync(
-                    partial(_pipe_read_exact, read_fd, data_size)
-                )
-                if local_data is None:
-                    logger.warning("JACCL pipe relay: EOF reading data payload")
-                    break
-
-                logger.info(
-                    f"JACCL pipe relay: read {data_size} bytes from runner, seq={sequence}"
-                )
-
-                # 2. Emit JacclSideChannelData event
-                waiter = anyio.Event()
-                self._gathered_waiters[sequence] = (waiter, None)
-                await self._event_sender.send(
-                    JacclSideChannelData(
-                        instance_id=self.bound_instance.instance.instance_id,
-                        runner_id=self.bound_instance.bound_runner_id,
-                        sequence=sequence,
-                        data=local_data,
-                    )
-                )
-
-                # 3. Wait for gathered result
-                await waiter.wait()
-                _, gathered_event = self._gathered_waiters.pop(sequence)
-                assert gathered_event is not None
-
-                # 4. Order gathered data by runner rank and concatenate
-                instance = self.bound_instance.instance
-                assert isinstance(instance, MlxJacclInstance)
-                runner_order = list(instance.shard_assignments.runner_to_shard.keys())
-                ordered_data = b"".join(
-                    gathered_event.gathered_data[rid] for rid in runner_order
-                )
-
-                # 5. Write gathered data to runner: [uint32 total_size][total_size bytes]
-                total_size = len(ordered_data)
-                response = struct.pack("<I", total_size) + ordered_data
-                await to_thread.run_sync(partial(_pipe_write_all, write_fd, response))
-
-                logger.info(
-                    f"JACCL pipe relay: wrote {total_size} bytes to runner, seq={sequence}"
-                )
-                sequence += 1
-        except OSError as e:
-            logger.warning(f"JACCL pipe relay: OS error: {e}")
-        except Exception as e:
-            logger.opt(exception=e).error("JACCL pipe relay: unexpected error")
-
-    def notify_gathered(self, event: JacclSideChannelGathered) -> None:
-        """Called by the worker when a JacclSideChannelGathered event arrives."""
-        seq = event.sequence
-        if seq not in self._gathered_waiters:
-            logger.warning(f"JACCL: received gathered event for unknown sequence {seq}")
-            return
-        waiter, _ = self._gathered_waiters[seq]
-        self._gathered_waiters[seq] = (waiter, event)
-        waiter.set()
-
     def __del__(self) -> None:
         if self.runner_process.is_alive():
             logger.warning("RunnerSupervisor was not stopped cleanly.")

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

@@ -1,7 +1,9 @@
 # Check tasks are complete before runner is ever ready.
+import unittest.mock
 from collections.abc import Iterable
 from typing import Callable
 
+import mlx.core as mx
 import pytest
 
 import exo.worker.runner.runner as mlx_runner
@@ -19,6 +21,7 @@ from exo.shared.types.tasks import (
     Shutdown,
     StartWarmup,
     Task,
+    TaskId,
     TaskStatus,
     TextGeneration,
 )
@@ -113,6 +116,7 @@ def patch_out_mlx(monkeypatch: pytest.MonkeyPatch):
     monkeypatch.setattr(mlx_runner, "load_mlx_items", make_nothin((1, MockTokenizer)))
     monkeypatch.setattr(mlx_runner, "warmup_inference", make_nothin(1))
     monkeypatch.setattr(mlx_runner, "_check_for_debug_prompts", nothin)
+    monkeypatch.setattr(mlx_runner, "mx_any", make_nothin(False))
     # Mock apply_chat_template since we're using a fake tokenizer (integer 1).
     # Returns a prompt without thinking tag so detect_thinking_prompt_suffix returns None.
     monkeypatch.setattr(mlx_runner, "apply_chat_template", make_nothin("test prompt"))
@@ -163,6 +167,7 @@ def _run(tasks: Iterable[Task]):
     )
 
     task_sender, task_receiver = mp_channel[Task]()
+    _cancel_sender, cancel_receiver = mp_channel[TaskId]()
     event_sender = EventCollector()
 
     with task_sender:
@@ -173,8 +178,16 @@ def _run(tasks: Iterable[Task]):
         # this is some c++ nonsense
         task_receiver.close = nothin
         task_receiver.join = nothin
-
-        mlx_runner.main(bound_instance, event_sender, task_receiver)  # type: ignore[arg-type]
+        with unittest.mock.patch(
+            "exo.worker.runner.runner.mx.distributed.all_gather",
+            make_nothin(mx.array([1])),
+        ):
+            mlx_runner.main(
+                bound_instance,
+                event_sender,  # pyright: ignore[reportArgumentType]
+                task_receiver,
+                cancel_receiver,
+            )
 
         return event_sender.events