Fix rebase

app/EXO/EXO/Models/ClusterState.swift

@@ -264,7 +264,7 @@ struct NodeDownloadStatus {
     init?(statusKey: String, payload: NodeDownloadPayload) {
         guard let nodeId = payload.nodeId else { return nil }
         self.nodeId = nodeId
-        self.progress = statusKey == "DownloadOngoing" ? payload.downloadProgress : nil
+        self.progress = statusKey == "ModelDownloading" ? payload.downloadProgress : nil
     }
 }
 

bench/harness.py

@@ -389,8 +389,8 @@ def run_planning_phase(
         node_downloads = client.get_node_downloads(node_id) or []
 
         already_downloaded = any(
-            "DownloadCompleted" in p
-            and unwrap_instance(p["DownloadCompleted"]["shardMetadata"])["modelCard"][
+            "ModelReady" in p
+            and unwrap_instance(p["ModelReady"]["shardMetadata"])["modelCard"][
                 "modelId"
             ]
             == full_model_id
@@ -428,14 +428,13 @@ def run_planning_phase(
 
         completed = [
             (
-                unwrap_instance(p["DownloadCompleted"]["shardMetadata"])["modelCard"][
+                unwrap_instance(p["ModelReady"]["shardMetadata"])["modelCard"][
                     "modelId"
                 ],
-                p["DownloadCompleted"]["total"]["inBytes"],
+                p["ModelReady"]["total"]["inBytes"],
             )
             for p in node_downloads
-            if "DownloadCompleted" in p
-            and not p["DownloadCompleted"].get("readOnly", False)
+            if "ModelReady" in p and not p["ModelReady"].get("readOnly", False)
         ]
         for del_model, size in sorted(completed, key=lambda x: x[1]):
             logger.info(f"Deleting {del_model} from {node_id} ({size // (1024**2)}MB)")
@@ -469,20 +468,20 @@ def run_planning_phase(
         for node_id in node_ids:
             node_downloads = client.get_node_downloads(node_id) or []
             done = any(
-                "DownloadCompleted" in p
-                and unwrap_instance(p["DownloadCompleted"]["shardMetadata"])[
-                    "modelCard"
-                ]["modelId"]
+                "ModelReady" in p
+                and unwrap_instance(p["ModelReady"]["shardMetadata"])["modelCard"][
+                    "modelId"
+                ]
                 == full_model_id
                 for p in node_downloads
             )
             failed = [
-                p["DownloadFailed"]["errorMessage"]
+                p["ModelDownloadFailed"]["errorMessage"]
                 for p in node_downloads
-                if "DownloadFailed" in p
-                and unwrap_instance(p["DownloadFailed"]["shardMetadata"])["modelCard"][
-                    "modelId"
-                ]
+                if "ModelDownloadFailed" in p
+                and unwrap_instance(p["ModelDownloadFailed"]["shardMetadata"])[
+                    "modelCard"
+                ]["modelId"]
                 == full_model_id
             ]
             if failed:

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

@@ -254,6 +254,13 @@ interface RawStateResponse {
     string,
     { total: { inBytes: number }; available: { inBytes: number } }
   >;
+  nodeStorageConfig?: Record<
+    string,
+    {
+      maxStorage: { inBytes: number } | null;
+      storagePolicy: "manual" | "auto-evict";
+    }
+  >;
 }
 
 export interface MessageAttachment {
@@ -568,6 +575,15 @@ class AppStore {
     >
   >({});
   nodeRdmaCtl = $state<Record<string, { enabled: boolean }>>({});
+  nodeStorageConfig = $state<
+    Record<
+      string,
+      {
+        maxStorage: { inBytes: number } | null;
+        storagePolicy: "manual" | "auto-evict";
+      }
+    >
+  >({});
   nodeThunderboltBridge = $state<
     Record<
       string,
@@ -1326,6 +1342,7 @@ class AppStore {
       this.thunderboltBridgeCycles = data.thunderboltBridgeCycles ?? [];
       // Thunderbolt bridge status per node
       this.nodeThunderboltBridge = data.nodeThunderboltBridge ?? {};
+      this.nodeStorageConfig = data.nodeStorageConfig ?? {};
       this.lastUpdate = Date.now();
       // Connection recovered
       if (!this.isConnected) {
@@ -3279,6 +3296,29 @@ class AppStore {
     }
   }
 
+  async setStorageConfig(
+    nodeIds: string[] | null,
+    maxStorageGb: number | null,
+    storagePolicy: "manual" | "auto-evict",
+  ): Promise<void> {
+    try {
+      const response = await fetch("/storage", {
+        method: "PUT",
+        headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({ nodeIds, maxStorageGb, storagePolicy }),
+      });
+      if (!response.ok) {
+        const errorText = await response.text();
+        throw new Error(
+          `Failed to set storage config: ${response.status} - ${errorText}`,
+        );
+      }
+    } catch (error) {
+      console.error("Error setting storage config:", error);
+      throw error;
+    }
+  }
+
   /**
    * List all available traces
    */
@@ -3356,6 +3396,7 @@ export const instances = () => appStore.instances;
 export const runners = () => appStore.runners;
 export const downloads = () => appStore.downloads;
 export const nodeDisk = () => appStore.nodeDisk;
+export const nodeStorageConfig = () => appStore.nodeStorageConfig;
 export const placementPreviews = () => appStore.placementPreviews;
 export const selectedPreviewModelId = () => appStore.selectedPreviewModelId;
 export const isLoadingPreviews = () => appStore.isLoadingPreviews;
@@ -3479,6 +3520,11 @@ export const startDownload = (nodeId: string, shardMetadata: object) =>
   appStore.startDownload(nodeId, shardMetadata);
 export const deleteDownload = (nodeId: string, modelId: string) =>
   appStore.deleteDownload(nodeId, modelId);
+export const setStorageConfig = (
+  nodeIds: string[] | null,
+  maxStorageGb: number | null,
+  storagePolicy: "manual" | "auto-evict",
+) => appStore.setStorageConfig(nodeIds, maxStorageGb, storagePolicy);
 
 // Trace actions
 export const listTraces = () => appStore.listTraces();

dashboard/src/lib/utils/downloads.ts

@@ -5,7 +5,7 @@
  *   Record<NodeId, Array<TaggedDownloadEntry>>
  *
  * Each entry is a tagged union object like:
- *   { "DownloadCompleted": { shard_metadata: { "PipelineShardMetadata": { model_card: { model_id: "..." }, ... } }, ... } }
+ *   { "ModelReady": { shard_metadata: { "PipelineShardMetadata": { model_card: { model_id: "..." }, ... } }, ... } }
  */
 
 /** Unwrap one level of tagged-union envelope, returning [tag, payload]. */
@@ -49,7 +49,7 @@ export function extractShardMetadata(
   return shardMetadata as Record<string, unknown>;
 }
 
-/** Get the download tag (DownloadCompleted, DownloadOngoing, etc.) from a wrapped entry. */
+/** Get the download tag (ModelReady, ModelDownloading, etc.) from a wrapped entry. */
 export function getDownloadTag(
   entry: unknown,
 ): [string, Record<string, unknown>] | null {
@@ -73,7 +73,7 @@ function* iterNodeDownloads(
   }
 }
 
-/** Check if a specific model is fully downloaded (DownloadCompleted) on a specific node. */
+/** Check if a specific model is fully downloaded (ModelReady) on a specific node. */
 export function isModelDownloadedOnNode(
   downloadsData: Record<string, unknown[]>,
   nodeId: string,
@@ -83,12 +83,12 @@ export function isModelDownloadedOnNode(
   if (!Array.isArray(nodeDownloads)) return false;
 
   for (const [tag, , entryModelId] of iterNodeDownloads(nodeDownloads)) {
-    if (tag === "DownloadCompleted" && entryModelId === modelId) return true;
+    if (tag === "ModelReady" && entryModelId === modelId) return true;
   }
   return false;
 }
 
-/** Get all node IDs where a model is fully downloaded (DownloadCompleted). */
+/** Get all node IDs where a model is fully downloaded (ModelReady). */
 export function getNodesWithModelDownloaded(
   downloadsData: Record<string, unknown[]>,
   modelId: string,
@@ -122,7 +122,7 @@ export function getShardMetadataForModel(
       const shard = extractShardMetadata(payload);
       if (!shard) continue;
 
-      if (tag === "DownloadCompleted") return shard;
+      if (tag === "ModelReady") return shard;
       if (!fallback) fallback = shard;
     }
   }
@@ -131,7 +131,7 @@ export function getShardMetadataForModel(
 
 /**
  * Get the download status tag for a specific model on a specific node.
- * Returns the "best" status: DownloadCompleted > DownloadOngoing > others.
+ * Returns the "best" status: ModelReady > ModelDownloading > others.
  */
 export function getModelDownloadStatus(
   downloadsData: Record<string, unknown[]>,
@@ -144,8 +144,8 @@ export function getModelDownloadStatus(
   let best: string | null = null;
   for (const [tag, , entryModelId] of iterNodeDownloads(nodeDownloads)) {
     if (entryModelId !== modelId) continue;
-    if (tag === "DownloadCompleted") return tag;
-    if (tag === "DownloadOngoing") best = tag;
+    if (tag === "ModelReady") return tag;
+    if (tag === "ModelDownloading") best = tag;
     else if (!best) best = tag;
   }
   return best;

dashboard/src/routes/+page.svelte

@@ -1581,12 +1581,14 @@
     progress: DownloadProgress | null;
     perNode: NodeDownloadStatus[];
     failedError: string | null;
+    rejectedError: string | null;
   } {
     const empty = {
       isDownloading: false,
       progress: null,
       perNode: [] as NodeDownloadStatus[],
       failedError: null,
+      rejectedError: null,
     };
 
     if (!downloadsData || Object.keys(downloadsData).length === 0) {
@@ -1618,8 +1620,8 @@
         const downloadModelId = extractModelIdFromDownload(downloadPayload);
         if (!downloadModelId || downloadModelId !== modelId) continue;
 
-        // DownloadFailed — return with any data collected so far
-        if (downloadKind === "DownloadFailed") {
+        // ModelDownloadFailed — return with any data collected so far
+        if (downloadKind === "ModelDownloadFailed") {
           return {
             isDownloading: false,
             progress: null,
@@ -1628,12 +1630,25 @@
               (downloadPayload.errorMessage as string) ||
               (downloadPayload.error_message as string) ||
               "Download failed",
+            rejectedError: null,
+          };
+        }
+
+        // ModelRejected — storage limit exceeded
+        if (downloadKind === "ModelRejected") {
+          return {
+            isDownloading: false,
+            progress: null,
+            perNode: Array.from(perNodeMap.values()),
+            failedError: null,
+            rejectedError:
+              (downloadPayload.reason as string) || "Storage limit exceeded",
           };
         }
 
         if (
-          downloadKind !== "DownloadOngoing" &&
-          downloadKind !== "DownloadPending" &&
+          downloadKind !== "ModelDownloading" &&
+          downloadKind !== "ModelNotDownloading" &&
           downloadKind !== "DownloadCompleted"
         )
           continue;
@@ -1652,7 +1667,7 @@
           continue;
         }
 
-        if (downloadKind === "DownloadPending") {
+        if (downloadKind === "ModelNotDownloading") {
           const pendingDownloaded = getBytes(
             downloadPayload.downloaded ??
               downloadPayload.downloaded_bytes ??
@@ -1676,7 +1691,7 @@
           continue;
         }
 
-        // DownloadOngoing
+        // ModelDownloading
         const progress = parseDownloadProgress(downloadPayload);
         if (
           !progress ||
@@ -1722,6 +1737,7 @@
         progress: null,
         perNode,
         failedError: null,
+        rejectedError: null,
       };
     }
 
@@ -1742,6 +1758,7 @@
       },
       perNode,
       failedError: null,
+      rejectedError: null,
     };
   }
 
@@ -1826,6 +1843,17 @@
       };
     }
 
+    if (result.rejectedError) {
+      return {
+        isDownloading: false,
+        isFailed: true,
+        errorMessage: result.rejectedError,
+        progress: null,
+        statusText: "REJECTED",
+        perNode: [],
+      };
+    }
+
     if (!result.isDownloading) {
       const statusInfo = deriveInstanceStatus(instanceWrapped);
       return {
@@ -2475,7 +2503,13 @@
     if (Object.keys(prev).length > 0) {
       for (const [id, currentStatus] of Object.entries(currentStatuses)) {
         const prevStatus = prev[id];
-        if (!prevStatus || prevStatus === currentStatus) continue;
+        if (prevStatus === currentStatus) continue;
+        if (
+          !prevStatus &&
+          currentStatus !== "REJECTED" &&
+          currentStatus !== "FAILED"
+        )
+          continue;
 
         const modelId = getInstanceModelId(instanceData[id]);
         const shortName = modelId
@@ -2509,6 +2543,14 @@
           addToast({ type: "error", message: `Model failed: ${shortName}` });
         }
 
+        if (prevStatus !== "REJECTED" && currentStatus === "REJECTED") {
+          addToast({
+            type: "warning",
+            message: `Storage limit exceeded: ${shortName}`,
+            duration: 8000,
+          });
+        }
+
         // Any -> Shutdown
         if (prevStatus !== "SHUTDOWN" && currentStatus === "SHUTDOWN") {
           addToast({ type: "info", message: `Model shut down: ${shortName}` });

dashboard/src/routes/downloads/+page.svelte

@@ -6,16 +6,19 @@
     topologyData,
     downloads,
     nodeDisk,
+    nodeStorageConfig,
     refreshState,
     lastUpdate as lastUpdateStore,
     startDownload,
     deleteDownload,
+    setStorageConfig,
   } from "$lib/stores/app.svelte";
   import {
     getDownloadTag,
     extractModelIdFromDownload,
     extractShardMetadata,
   } from "$lib/utils/downloads";
+  import { addToast } from "$lib/stores/toast.svelte";
   import HeaderNav from "$lib/components/HeaderNav.svelte";
 
   type CellStatus =
@@ -36,6 +39,14 @@
         modelDirectory?: string;
       }
     | { kind: "failed"; modelDirectory?: string }
+    | {
+        kind: "rejected";
+        reason: string;
+        requiredBytes: number;
+        availableBytes: number;
+        limitBytes: number;
+        modelDirectory?: string;
+      }
     | { kind: "not_present" };
 
   type ModelCardInfo = {
@@ -61,11 +72,14 @@
     label: string;
     diskAvailable?: number;
     diskTotal?: number;
+    storageLimit?: number;
+    storagePolicy?: "manual" | "auto-evict";
   };
 
   const data = $derived(topologyData());
   const downloadsData = $derived(downloads());
   const nodeDiskData = $derived(nodeDisk());
+  const storageConfigData = $derived(nodeStorageConfig());
 
   function getNodeLabel(nodeId: string): string {
     const node = data?.nodes?.[nodeId];
@@ -122,10 +136,37 @@
     return Math.min(100, Math.max(0, value as number));
   }
 
+  function getNodeUsedStorage(nodeId: string): number {
+    const nodeDownloads = downloadsData?.[nodeId];
+    if (!nodeDownloads || !Array.isArray(nodeDownloads)) return 0;
+    let total = 0;
+    for (const entry of nodeDownloads) {
+      const tagged = getDownloadTag(entry);
+      if (!tagged) continue;
+      const [tag, payload] = tagged;
+      if (tag === "ModelReady") {
+        total += getBytes(payload.total);
+      } else if (tag === "ModelDownloading") {
+        const prog = (payload.download_progress ?? payload.downloadProgress) as
+          | Record<string, unknown>
+          | undefined;
+        if (prog) total += getBytes(prog.downloaded);
+      }
+    }
+    return total;
+  }
+
+  function storageBarColor(percent: number): string {
+    if (percent >= 90) return "bg-red-500";
+    if (percent >= 70) return "bg-yellow-500";
+    return "bg-green-500";
+  }
+
   const CELL_PRIORITY: Record<CellStatus["kind"], number> = {
-    completed: 4,
-    downloading: 3,
-    pending: 2,
+    completed: 5,
+    downloading: 4,
+    pending: 3,
+    rejected: 2,
     failed: 1,
     not_present: 0,
   };
@@ -178,6 +219,50 @@
   let nodeColumns = $state<NodeColumn[]>([]);
   let infoRow = $state<ModelRow | null>(null);
 
+  let storageConfigNode = $state<NodeColumn | null>(null);
+  let configMaxGb = $state<number | null>(null);
+  let configNoLimit = $state(true);
+  let configPolicy = $state<"manual" | "auto-evict">("manual");
+  let configSaving = $state(false);
+  let configApplyAll = $state(false);
+  let configDiskTotalGb = $derived(
+    storageConfigNode
+      ? Math.round((storageConfigNode.diskTotal ?? 0) / 1024 ** 3)
+      : 0,
+  );
+
+  function openStorageConfig(col: NodeColumn) {
+    storageConfigNode = col;
+    if (col.storageLimit != null) {
+      configNoLimit = false;
+      configMaxGb = Math.round(col.storageLimit / 1024 ** 3);
+    } else {
+      configNoLimit = true;
+      configMaxGb = null;
+    }
+    configPolicy = col.storagePolicy ?? "manual";
+    configApplyAll = false;
+  }
+
+  async function saveStorageConfig() {
+    if (!storageConfigNode) return;
+    configSaving = true;
+    try {
+      const maxGb = configNoLimit ? null : configMaxGb;
+      const nodeIds = configApplyAll ? null : [storageConfigNode.nodeId];
+      await setStorageConfig(nodeIds, maxGb, configPolicy);
+      storageConfigNode = null;
+      refreshState();
+    } catch (error) {
+      addToast({
+        type: "error",
+        message: `Failed to save storage config: ${error instanceof Error ? error.message : String(error)}`,
+      });
+    } finally {
+      configSaving = false;
+    }
+  }
+
   $effect(() => {
     try {
       if (!downloadsData || Object.keys(downloadsData).length === 0) {
@@ -189,11 +274,14 @@
       const allNodeIds = Object.keys(downloadsData);
       const columns: NodeColumn[] = allNodeIds.map((nodeId) => {
         const diskInfo = nodeDiskData?.[nodeId];
+        const storageConfig = storageConfigData?.[nodeId];
         return {
           nodeId,
           label: getNodeLabel(nodeId),
           diskAvailable: diskInfo?.available?.inBytes,
           diskTotal: diskInfo?.total?.inBytes,
+          storageLimit: storageConfig?.maxStorage?.inBytes ?? undefined,
+          storagePolicy: storageConfig?.storagePolicy,
         };
       });
 
@@ -234,10 +322,14 @@
             ((payload.model_directory ?? payload.modelDirectory) as string) ||
             undefined;
           let cell: CellStatus;
-          if (tag === "DownloadCompleted") {
+          if (tag === "ModelReady") {
             const totalBytes = getBytes(payload.total);
-            cell = { kind: "completed", totalBytes, modelDirectory };
-          } else if (tag === "DownloadOngoing") {
+            cell = {
+              kind: "completed",
+              totalBytes,
+              modelDirectory,
+            };
+          } else if (tag === "ModelDownloading") {
             const rawProgress =
               payload.download_progress ?? payload.downloadProgress ?? {};
             const prog = rawProgress as Record<string, unknown>;
@@ -257,7 +349,16 @@
               etaMs,
               modelDirectory,
             };
-          } else if (tag === "DownloadFailed") {
+          } else if (tag === "ModelRejected") {
+            cell = {
+              kind: "rejected",
+              reason: (payload.reason as string) ?? "Storage limit exceeded",
+              requiredBytes: getBytes(payload.required),
+              availableBytes: getBytes(payload.available),
+              limitBytes: getBytes(payload.limit),
+              modelDirectory,
+            };
+          } else if (tag === "ModelDownloadFailed") {
             cell = { kind: "failed", modelDirectory };
           } else {
             const downloaded = getBytes(
@@ -406,15 +507,52 @@
                 Model
               </th>
               {#each nodeColumns as col}
+                {@const usedStorage = getNodeUsedStorage(col.nodeId)}
+                {@const storageMax = col.storageLimit ?? col.diskTotal ?? 0}
+                {@const storagePercent =
+                  storageMax > 0
+                    ? Math.min(100, (usedStorage / storageMax) * 100)
+                    : 0}
                 <th
                   class="px-4 py-3 text-[11px] uppercase tracking-wider text-exo-light-gray font-medium text-center whitespace-nowrap min-w-[120px]"
                 >
-                  <div>{col.label}</div>
-                  {#if col.diskAvailable != null}
-                    <div
-                      class="text-[9px] text-white/70 normal-case tracking-normal mt-0.5"
+                  <div class="flex items-center justify-center gap-1">
+                    <span>{col.label}</span>
+                    <button
+                      type="button"
+                      class="p-0.5 rounded hover:bg-white/10 transition-colors"
+                      onclick={() => openStorageConfig(col)}
+                      title="Storage settings"
+                      aria-label="Storage settings for {col.label}"
                     >
-                      {formatBytes(col.diskAvailable)} free
+                      <svg
+                        class="w-3.5 h-3.5 text-white/40 hover:text-exo-yellow transition-colors"
+                        viewBox="0 0 20 20"
+                        fill="currentColor"
+                      >
+                        <path
+                          fill-rule="evenodd"
+                          d="M11.49 3.17c-.38-1.56-2.6-1.56-2.98 0a1.532 1.532 0 01-2.286.948c-1.372-.836-2.942.734-2.106 2.106.54.886.061 2.042-.947 2.287-1.561.379-1.561 2.6 0 2.978a1.532 1.532 0 01.947 2.287c-.836 1.372.734 2.942 2.106 2.106a1.532 1.532 0 012.287.947c.379 1.561 2.6 1.561 2.978 0a1.533 1.533 0 012.287-.947c1.372.836 2.942-.734 2.106-2.106a1.533 1.533 0 01.947-2.287c1.561-.379 1.561-2.6 0-2.978a1.532 1.532 0 01-.947-2.287c.836-1.372-.734-2.942-2.106-2.106a1.532 1.532 0 01-2.287-.947zM10 13a3 3 0 100-6 3 3 0 000 6z"
+                          clip-rule="evenodd"
+                        />
+                      </svg>
+                    </button>
+                  </div>
+                  {#if storageMax > 0}
+                    <div class="text-[9px] normal-case tracking-normal mt-1">
+                      <div
+                        class="w-full h-1.5 bg-white/10 rounded-full overflow-hidden"
+                      >
+                        <div
+                          class="h-full rounded-full transition-all duration-300 {storageBarColor(
+                            storagePercent,
+                          )}"
+                          style="width: {storagePercent.toFixed(1)}%"
+                        ></div>
+                      </div>
+                      <div class="text-white/60 mt-0.5">
+                        {formatBytes(usedStorage)} / {formatBytes(storageMax)}
+                      </div>
                     </div>
                   {/if}
                 </th>
@@ -610,6 +748,52 @@
                           <span class="text-white/40 text-sm">...</span>
                         {/if}
                       </div>
+                    {:else if cell.kind === "rejected"}
+                      <div
+                        class="flex flex-col items-center gap-1"
+                        title={cell.reason}
+                      >
+                        <svg
+                          class="w-7 h-7 text-orange-400"
+                          viewBox="0 0 20 20"
+                          fill="currentColor"
+                        >
+                          <path
+                            fill-rule="evenodd"
+                            d="M8.257 3.099c.765-1.36 2.722-1.36 3.486 0l5.58 9.92c.75 1.334-.213 2.98-1.742 2.98H4.42c-1.53 0-2.493-1.646-1.743-2.98l5.58-9.92zM11 13a1 1 0 11-2 0 1 1 0 012 0zm-1-8a1 1 0 00-1 1v3a1 1 0 002 0V6a1 1 0 00-1-1z"
+                            clip-rule="evenodd"
+                          ></path>
+                        </svg>
+                        <span class="text-[10px] text-orange-400/80"
+                          >Need {formatBytes(cell.requiredBytes)}</span
+                        >
+                        <span class="text-[10px] text-white/50"
+                          >{formatBytes(cell.availableBytes)} avail</span
+                        >
+                        {#if row.shardMetadata}
+                          <button
+                            type="button"
+                            class="text-white/50 hover:text-exo-yellow transition-colors cursor-pointer"
+                            onclick={() =>
+                              startDownload(col.nodeId, row.shardMetadata!)}
+                            title="Retry download on this node"
+                          >
+                            <svg
+                              class="w-5 h-5"
+                              viewBox="0 0 20 20"
+                              fill="none"
+                              stroke="currentColor"
+                              stroke-width="2"
+                            >
+                              <path
+                                d="M10 3v10m0 0l-3-3m3 3l3-3M3 17h14"
+                                stroke-linecap="round"
+                                stroke-linejoin="round"
+                              ></path>
+                            </svg>
+                          </button>
+                        {/if}
+                      </div>
                     {:else if cell.kind === "failed"}
                       <div
                         class="flex flex-col items-center gap-1"
@@ -812,9 +996,11 @@
                       ? 'bg-green-500/10 text-green-400/80 border border-green-500/20'
                       : cellStatus.kind === 'downloading'
                         ? 'bg-exo-yellow/10 text-exo-yellow/80 border border-exo-yellow/20'
-                        : cellStatus.kind === 'failed'
-                          ? 'bg-red-500/10 text-red-400/80 border border-red-500/20'
-                          : 'bg-white/5 text-white/50 border border-white/10'}"
+                        : cellStatus.kind === 'rejected'
+                          ? 'bg-orange-500/10 text-orange-400/80 border border-orange-500/20'
+                          : cellStatus.kind === 'failed'
+                            ? 'bg-red-500/10 text-red-400/80 border border-red-500/20'
+                            : 'bg-white/5 text-white/50 border border-white/10'}"
                   >
                     {col.label}
                     {#if cellStatus.kind === "downloading" && "percentage" in cellStatus}
@@ -839,8 +1025,198 @@
   </div>
 {/if}
 
+<!-- Storage config modal -->
+{#if storageConfigNode}
+  <div
+    class="fixed inset-0 z-[60] bg-black/60"
+    transition:fade={{ duration: 150 }}
+    onclick={() => (storageConfigNode = null)}
+    role="presentation"
+  ></div>
+  <div
+    class="fixed z-[60] top-1/2 left-1/2 -translate-x-1/2 -translate-y-1/2 w-[min(80vw,360px)] bg-exo-dark-gray border border-exo-yellow/10 rounded-lg shadow-2xl p-4"
+    transition:fly={{ y: 10, duration: 200, easing: cubicOut }}
+    role="dialog"
+    aria-modal="true"
+    onkeydown={(e) => {
+      if (e.key === "Escape") storageConfigNode = null;
+    }}
+  >
+    <div class="flex items-start justify-between mb-4">
+      <h3 class="font-mono text-sm text-white">
+        Storage — {configApplyAll ? "All nodes" : storageConfigNode.label}
+      </h3>
+      <button
+        type="button"
+        class="p-1 rounded hover:bg-white/10 transition-colors text-white/50"
+        onclick={() => (storageConfigNode = null)}
+        aria-label="Close storage settings"
+      >
+        <svg class="w-4 h-4" viewBox="0 0 24 24" fill="currentColor">
+          <path
+            d="M19 6.41L17.59 5 12 10.59 6.41 5 5 6.41 10.59 12 5 17.59 6.41 19 12 13.41 17.59 19 19 17.59 13.41 12 19 6.41z"
+          />
+        </svg>
+      </button>
+    </div>
+
+    <div class="space-y-4">
+      <!-- Apply to all nodes -->
+      {#if nodeColumns.length > 1}
+        <label class="flex items-center gap-2 cursor-pointer">
+          <input
+            type="checkbox"
+            bind:checked={configApplyAll}
+            class="accent-exo-yellow w-4 h-4"
+          />
+          <span class="text-xs font-mono text-white/80">Apply to all nodes</span
+          >
+        </label>
+      {/if}
+
+      <!-- No limit checkbox -->
+      <label class="flex items-center gap-2 cursor-pointer">
+        <input
+          type="checkbox"
+          bind:checked={configNoLimit}
+          onchange={() => {
+            if (!configNoLimit && configMaxGb == null) {
+              configMaxGb = configDiskTotalGb || 50;
+            }
+          }}
+          class="accent-exo-yellow w-4 h-4"
+        />
+        <span class="text-xs font-mono text-white/80">Unlimited storage</span>
+      </label>
+
+      <!-- Max storage slider -->
+      <div class="space-y-1.5">
+        <div class="flex items-baseline justify-between">
+          <label
+            class="text-[11px] font-mono text-white/50 uppercase tracking-wider"
+            for="storage-max-gb"
+          >
+            Max storage
+          </label>
+          <span
+            class="text-xs font-mono tabular-nums transition-opacity {configNoLimit
+              ? 'opacity-30'
+              : 'text-white'}"
+          >
+            {configMaxGb ?? 0} GB
+          </span>
+        </div>
+        <input
+          id="storage-max-gb"
+          type="range"
+          min="1"
+          max={Math.max(configDiskTotalGb, configMaxGb ?? 1)}
+          step="1"
+          bind:value={configMaxGb}
+          disabled={configNoLimit}
+          class="slider w-full h-1.5 rounded-full appearance-none cursor-pointer
+            disabled:opacity-30 disabled:cursor-not-allowed"
+        />
+        <div
+          class="flex justify-between text-[10px] font-mono text-white/30 transition-opacity {configNoLimit
+            ? 'opacity-30'
+            : ''}"
+        >
+          <span>1 GB</span>
+          <span>{Math.max(configDiskTotalGb, configMaxGb ?? 1)} GB</span>
+        </div>
+      </div>
+
+      <!-- Policy selector -->
+      <div class="space-y-1.5">
+        <div
+          class="text-[11px] font-mono text-white/50 uppercase tracking-wider"
+        >
+          Eviction policy
+        </div>
+        <div class="flex gap-1">
+          <button
+            type="button"
+            class="flex-1 px-3 py-1.5 rounded text-xs font-mono transition-colors
+              {configPolicy === 'manual'
+              ? 'bg-exo-yellow/20 text-exo-yellow border border-exo-yellow/40'
+              : 'bg-exo-black/40 text-white/50 border border-exo-medium-gray/30 hover:text-white/70'}"
+            onclick={() => (configPolicy = "manual")}
+          >
+            Manual
+          </button>
+          <button
+            type="button"
+            class="flex-1 px-3 py-1.5 rounded text-xs font-mono transition-colors
+              {configPolicy === 'auto-evict'
+              ? 'bg-exo-yellow/20 text-exo-yellow border border-exo-yellow/40'
+              : 'bg-exo-black/40 text-white/50 border border-exo-medium-gray/30 hover:text-white/70'}"
+            onclick={() => (configPolicy = "auto-evict")}
+          >
+            Auto-evict
+          </button>
+        </div>
+        <p class="text-[10px] text-white/40 font-mono">
+          {#if configPolicy === "manual"}
+            Downloads that exceed the limit are rejected. Delete models
+            manually.
+          {:else}
+            Oldest unused models are automatically removed to make room.
+          {/if}
+        </p>
+      </div>
+    </div>
+
+    <!-- Actions -->
+    <div class="flex justify-end gap-2 mt-5">
+      <button
+        type="button"
+        class="px-3 py-1.5 rounded text-xs font-mono text-white/50 hover:text-white/70 transition-colors"
+        onclick={() => (storageConfigNode = null)}
+      >
+        Cancel
+      </button>
+      <button
+        type="button"
+        class="px-3 py-1.5 rounded text-xs font-mono bg-exo-yellow/20 text-exo-yellow border border-exo-yellow/40 hover:bg-exo-yellow/30 transition-colors disabled:opacity-50"
+        onclick={saveStorageConfig}
+        disabled={configSaving ||
+          (!configNoLimit && (configMaxGb == null || configMaxGb <= 0))}
+      >
+        {configSaving ? "Saving..." : "Save"}
+      </button>
+    </div>
+  </div>
+{/if}
+
 <style>
   table {
     min-width: max-content;
   }
+
+  .slider {
+    background: rgba(255, 255, 255, 0.1);
+  }
+  .slider::-webkit-slider-thumb {
+    -webkit-appearance: none;
+    width: 14px;
+    height: 14px;
+    border-radius: 50%;
+    background: #f5c518;
+    cursor: pointer;
+  }
+  .slider::-moz-range-thumb {
+    width: 14px;
+    height: 14px;
+    border-radius: 50%;
+    border: none;
+    background: #f5c518;
+    cursor: pointer;
+  }
+  .slider:disabled::-webkit-slider-thumb {
+    cursor: not-allowed;
+  }
+  .slider:disabled::-moz-range-thumb {
+    cursor: not-allowed;
+  }
 </style>

src/exo/api/main.py

@@ -21,6 +21,7 @@ from hypercorn.asyncio import serve  # pyright: ignore[reportUnknownVariableType
 from hypercorn.config import Config
 from hypercorn.typing import ASGIFramework
 from loguru import logger
+from pydantic import Field
 
 from exo.api.adapters.chat_completions import (
     chat_request_to_text_generation,
@@ -136,6 +137,7 @@ from exo.shared.models.model_cards import (
     get_card,
     get_model_cards,
 )
+from exo.shared.storage import calculate_used_storage
 from exo.shared.tracing import TraceEvent, compute_stats, export_trace, load_trace_file
 from exo.shared.types.chunks import (
     ErrorChunk,
@@ -159,6 +161,7 @@ from exo.shared.types.commands import (
     ImageGeneration,
     PlaceInstance,
     SendInputChunk,
+    SetStorageConfig,
     StartDownload,
     TaskCancelled,
     TaskFinished,
@@ -173,16 +176,30 @@ from exo.shared.types.events import (
 )
 from exo.shared.types.memory import Memory
 from exo.shared.types.state import State
+from exo.shared.types.storage import StorageConfig, StoragePolicy
 from exo.shared.types.text_generation import TextGenerationTaskParams
-from exo.shared.types.worker.downloads import DownloadCompleted
+from exo.shared.types.worker.downloads import ModelReady
 from exo.shared.types.worker.instances import Instance, InstanceId, InstanceMeta
 from exo.shared.types.worker.shards import Sharding
 from exo.utils.banner import print_startup_banner
 from exo.utils.channels import Receiver, Sender, channel
 from exo.utils.disk_event_log import DiskEventLog
 from exo.utils.power_sampler import PowerSampler
+from exo.utils.pydantic_ext import CamelCaseModel
 from exo.utils.task_group import TaskGroup
 
+
+class SetStorageConfigRequest(CamelCaseModel):
+    node_ids: list[NodeId] | None = None
+    max_storage_gb: Annotated[float, Field(ge=0)] | None = None
+    storage_policy: StoragePolicy = "manual"
+
+
+class NodeStorageInfo(CamelCaseModel):
+    config: StorageConfig
+    used: Memory
+
+
 _API_EVENT_LOG_DIR = EXO_EVENT_LOG_DIR / "api"
 ONBOARDING_COMPLETE_FILE = EXO_CACHE_HOME / "onboarding_complete"
 
@@ -351,6 +368,9 @@ class API:
         self.app.get("/events")(self.stream_events)
         self.app.post("/download/start")(self.start_download)
         self.app.delete("/download/{node_id}/{model_id:path}")(self.delete_download)
+        self.app.get("/storage")(self.get_storage)
+        self.app.get("/storage/{node_id}")(self.get_storage_node)
+        self.app.put("/storage")(self.set_storage_config)
         self.app.get("/v1/traces")(self.list_traces)
         self.app.post("/v1/traces/delete")(self.delete_traces)
         self.app.get("/v1/traces/{task_id}")(self.get_trace)
@@ -1553,7 +1573,7 @@ class API:
         downloaded_model_ids: set[str] = set()
         for node_downloads in self.state.downloads.values():
             for dl in node_downloads:
-                if isinstance(dl, DownloadCompleted):
+                if isinstance(dl, ModelReady):
                     downloaded_model_ids.add(dl.shard_metadata.model_card.model_id)
 
         cards = [
@@ -1640,7 +1660,7 @@ class API:
             downloaded_model_ids: set[str] = set()
             for node_downloads in self.state.downloads.values():
                 for dl in node_downloads:
-                    if isinstance(dl, DownloadCompleted):
+                    if isinstance(dl, ModelReady):
                         downloaded_model_ids.add(dl.shard_metadata.model_card.model_id)
             cards = [c for c in cards if c.model_id in downloaded_model_ids]
 
@@ -1879,6 +1899,48 @@ class API:
         await self._send_download(command)
         return DeleteDownloadResponse(command_id=command.command_id)
 
+    async def get_storage(self) -> dict[str, NodeStorageInfo]:
+        result: dict[str, NodeStorageInfo] = {}
+        for node_id, config in self.state.node_storage_config.items():
+            downloads = list(self.state.downloads.get(node_id, ()))
+            used = calculate_used_storage(downloads)
+            result[node_id] = NodeStorageInfo(config=config, used=used)
+        return result
+
+    async def get_storage_node(self, node_id: NodeId) -> NodeStorageInfo:
+        config = self.state.node_storage_config.get(node_id, StorageConfig())
+        downloads = list(self.state.downloads.get(node_id, ()))
+        used = calculate_used_storage(downloads)
+        return NodeStorageInfo(config=config, used=used)
+
+    async def set_storage_config(
+        self, request: SetStorageConfigRequest
+    ) -> dict[str, str | list[str]]:
+        max_storage = (
+            Memory.from_gb(request.max_storage_gb)
+            if request.max_storage_gb is not None
+            else None
+        )
+
+        target_node_ids = (
+            request.node_ids
+            if request.node_ids is not None
+            else list(self.state.node_storage_config.keys())
+        )
+
+        command_ids: list[str] = []
+        for node_id in target_node_ids:
+            command = SetStorageConfig(
+                target_node_id=node_id,
+                max_storage=max_storage,
+                storage_policy=request.storage_policy,
+            )
+            await self.command_sender.send(
+                ForwarderCommand(origin=self._system_id, command=command)
+            )
+            command_ids.append(str(command.command_id))
+        return {"status": "ok", "commandIds": command_ids}
+
     @staticmethod
     def _get_trace_path(task_id: str) -> Path:
         trace_path = EXO_TRACING_CACHE_DIR / f"trace_{task_id}.json"

src/exo/download/coordinator.py

@@ -1,8 +1,12 @@
 from __future__ import annotations
 
+import json
 from dataclasses import dataclass, field
+from datetime import UTC, datetime
 from pathlib import Path
 
+import aiofiles
+import aiofiles.os as aios
 import anyio
 from anyio import current_time, to_thread
 from loguru import logger
@@ -15,8 +19,17 @@ from exo.download.download_utils import (
     resolve_existing_model,
 )
 from exo.download.shard_downloader import ShardDownloader
-from exo.shared.constants import EXO_DEFAULT_MODELS_DIR, EXO_MODELS_READ_ONLY_DIRS
+from exo.shared.constants import (
+    EXO_DEFAULT_MODELS_DIR,
+    EXO_MODEL_USAGE_FILE,
+    EXO_MODELS_READ_ONLY_DIRS,
+)
 from exo.shared.models.model_cards import ModelId, get_model_cards
+from exo.shared.storage import (
+    calculate_used_storage,
+    decide_storage_action,
+    persist_storage_config,
+)
 from exo.shared.types.commands import (
     CancelDownload,
     DeleteDownload,
@@ -26,15 +39,26 @@ from exo.shared.types.commands import (
 from exo.shared.types.common import NodeId
 from exo.shared.types.events import (
     Event,
+    IndexedEvent,
+    InstanceCreated,
+    InstanceDeleted,
     NodeDownloadProgress,
+    StorageConfigUpdated,
 )
 from exo.shared.types.memory import Memory
+from exo.shared.types.storage import (
+    StorageAllow,
+    StorageConfig,
+    StorageEvict,
+    StorageReject,
+)
 from exo.shared.types.worker.downloads import (
-    DownloadCompleted,
-    DownloadFailed,
-    DownloadOngoing,
-    DownloadPending,
-    DownloadProgress,
+    ModelDownloadFailed,
+    ModelDownloading,
+    ModelNotDownloading,
+    ModelReady,
+    ModelRejected,
+    ModelStatus,
 )
 from exo.shared.types.worker.shards import PipelineShardMetadata, ShardMetadata
 from exo.utils.channels import Receiver, Sender
@@ -46,12 +70,18 @@ class DownloadCoordinator:
     node_id: NodeId
     shard_downloader: ShardDownloader
     download_command_receiver: Receiver[ForwarderDownloadCommand]
+    event_receiver: Receiver[IndexedEvent]
     event_sender: Sender[Event]
     offline: bool = False
+    storage_config: StorageConfig = field(default_factory=StorageConfig)
 
     # Local state
-    download_status: dict[ModelId, DownloadProgress] = field(default_factory=dict)
+    download_status: dict[ModelId, ModelStatus] = field(default_factory=dict)
     active_downloads: dict[ModelId, anyio.CancelScope] = field(default_factory=dict)
+    _deleting: set[ModelId] = field(default_factory=set)
+
+    _model_last_used: dict[ModelId, datetime] = field(default_factory=dict)
+    _active_model_ids: set[ModelId] = field(default_factory=set)
 
     _tg: TaskGroup = field(init=False, default_factory=TaskGroup)
     _stopped: anyio.Event = field(init=False, default_factory=anyio.Event)
@@ -71,8 +101,8 @@ class DownloadCoordinator:
         shard: ShardMetadata,
         found: Path,
         total: Memory,
-    ) -> DownloadCompleted:
-        return DownloadCompleted(
+    ) -> ModelReady:
+        return ModelReady(
             shard_metadata=shard,
             node_id=self.node_id,
             total=total,
@@ -93,7 +123,7 @@ class DownloadCoordinator:
                     callback_shard, found, progress.total
                 )
             else:
-                completed = DownloadCompleted(
+                completed = ModelReady(
                     shard_metadata=callback_shard,
                     node_id=self.node_id,
                     total=progress.total,
@@ -109,7 +139,7 @@ class DownloadCoordinator:
             and current_time() - self._last_progress_time.get(model_id, 0.0)
             > throttle_interval_secs
         ):
-            ongoing = DownloadOngoing(
+            ongoing = ModelDownloading(
                 node_id=self.node_id,
                 shard_metadata=callback_shard,
                 download_progress=map_repo_download_progress_to_download_progress_data(
@@ -127,13 +157,40 @@ class DownloadCoordinator:
         logger.info(
             f"Starting DownloadCoordinator{' (offline mode)' if self.offline else ''}"
         )
+        await self._load_model_usage()
         try:
             async with self._tg as tg:
                 tg.start_soon(self._command_processor)
                 tg.start_soon(self._emit_existing_download_progress)
+                tg.start_soon(self._event_watcher)
         finally:
             self._stopped.set()
 
+    async def _event_watcher(self) -> None:
+        active_instances: dict[str, ModelId] = {}
+        with self.event_receiver as events:
+            async for indexed_event in events:
+                match indexed_event.event:
+                    case StorageConfigUpdated(node_id=node_id) if (
+                        node_id == self.node_id
+                    ):
+                        self.storage_config = indexed_event.event.storage_config
+                        await self.clear_rejections()
+                        await persist_storage_config(indexed_event.event.storage_config)
+                    case InstanceCreated(instance=instance):
+                        active_instances[instance.instance_id] = (
+                            instance.shard_assignments.model_id
+                        )
+                        await self.update_active_models(set(active_instances.values()))
+                    case InstanceDeleted(instance_id=instance_id):
+                        if instance_id in active_instances:
+                            del active_instances[instance_id]
+                            await self.update_active_models(
+                                set(active_instances.values())
+                            )
+                    case _:
+                        pass
+
     async def shutdown(self) -> None:
         self._tg.cancel_tasks()
         await self._stopped.wait()
@@ -158,7 +215,7 @@ class DownloadCoordinator:
             logger.info(f"Cancelling download for {model_id}")
             self.active_downloads[model_id].cancel()
             current_status = self.download_status[model_id]
-            pending = DownloadPending(
+            pending = ModelNotDownloading(
                 shard_metadata=current_status.shard_metadata,
                 node_id=self.node_id,
                 model_directory=self._default_model_dir(model_id),
@@ -174,11 +231,13 @@ class DownloadCoordinator:
         # Check if already downloading, complete, or recently failed
         if model_id in self.download_status:
             status = self.download_status[model_id]
-            if isinstance(status, (DownloadOngoing, DownloadCompleted, DownloadFailed)):
+            if isinstance(status, (ModelDownloading, ModelReady, ModelDownloadFailed)):
                 logger.debug(
                     f"Download for {model_id} already in progress, complete, or failed, skipping"
                 )
                 return
+            if isinstance(status, ModelRejected):
+                del self.download_status[model_id]
 
         # Check all model directories for pre-existing complete models
         found_path = await to_thread.run_sync(resolve_existing_model, model_id)
@@ -193,8 +252,25 @@ class DownloadCoordinator:
             )
             return
 
+        action = decide_storage_action(
+            shard.model_card.storage_size,
+            self.storage_config,
+            list(self.download_status.values()),
+            self._model_last_used,
+            frozenset(self._active_model_ids),
+        )
+        match action:
+            case StorageReject(reason=reason, available=available):
+                await self._reject_download(shard, reason, available)
+                return
+            case StorageEvict(model_ids=model_ids):
+                if not await self._execute_evictions(model_ids, shard):
+                    return
+            case StorageAllow():
+                pass
+
         # Emit pending status
-        progress = DownloadPending(
+        progress = ModelNotDownloading(
             shard_metadata=shard,
             node_id=self.node_id,
             model_directory=self._default_model_dir(model_id),
@@ -214,7 +290,7 @@ class DownloadCoordinator:
                     shard, found, initial_progress.total
                 )
             else:
-                completed = DownloadCompleted(
+                completed = ModelReady(
                     shard_metadata=shard,
                     node_id=self.node_id,
                     total=initial_progress.total,
@@ -230,7 +306,7 @@ class DownloadCoordinator:
             logger.warning(
                 f"Offline mode: model {model_id} is not fully available locally, cannot download"
             )
-            failed = DownloadFailed(
+            failed = ModelDownloadFailed(
                 shard_metadata=shard,
                 node_id=self.node_id,
                 error_message=f"Model files not found locally in offline mode: {model_id}",
@@ -249,7 +325,7 @@ class DownloadCoordinator:
         model_id = shard.model_card.model_id
 
         # Emit ongoing status
-        status = DownloadOngoing(
+        status = ModelDownloading(
             node_id=self.node_id,
             shard_metadata=shard,
             download_progress=map_repo_download_progress_to_download_progress_data(
@@ -266,7 +342,7 @@ class DownloadCoordinator:
                     await self.shard_downloader.ensure_shard(shard)
             except Exception as e:
                 logger.error(f"Download failed for {model_id}: {e}")
-                failed = DownloadFailed(
+                failed = ModelDownloadFailed(
                     shard_metadata=shard,
                     node_id=self.node_id,
                     error_message=str(e),
@@ -276,9 +352,6 @@ class DownloadCoordinator:
                 await self.event_sender.send(
                     NodeDownloadProgress(download_progress=failed)
                 )
-            except anyio.get_cancelled_exc_class():
-                # ignore cancellation - let cleanup do its thing
-                pass
             finally:
                 self.active_downloads.pop(model_id, None)
 
@@ -286,13 +359,15 @@ class DownloadCoordinator:
         self._tg.start_soon(download_wrapper, scope)
         self.active_downloads[model_id] = scope
 
-    async def _delete_download(self, model_id: ModelId) -> None:
+    async def _remove_model_from_disk(self, model_id: ModelId) -> bool:
         # Protect read-only models from deletion
         if model_id in self.download_status:
             current = self.download_status[model_id]
-            if isinstance(current, DownloadCompleted) and current.read_only:
-                logger.warning(f"Refusing to delete read-only model {model_id}")
-                return
+            if isinstance(current, ModelReady) and current.read_only:
+                logger.warning(
+                    f"Refusing to delete read-only model {model_id} (from EXO_MODELS_READ_ONLY_DIRS)"
+                )
+                return False
 
         # Cancel if active
         if model_id in self.active_downloads:
@@ -303,15 +378,22 @@ class DownloadCoordinator:
         logger.info(f"Deleting model files for {model_id}")
         deleted = await delete_model(model_id)
 
-        if deleted:
-            logger.info(f"Successfully deleted model {model_id}")
-        else:
-            logger.warning(f"Model {model_id} was not found on disk")
+        if not deleted:
+            logger.warning(f"Failed to delete model {model_id} from disk")
+            return False
+
+        logger.info(f"Successfully deleted model {model_id}")
+        return True
+
+    async def _delete_download(self, model_id: ModelId) -> bool:
+        success = await self._remove_model_from_disk(model_id)
+        if not success:
+            return False
 
         # Emit pending status to reset UI state, then remove from local tracking
         if model_id in self.download_status:
             current_status = self.download_status[model_id]
-            pending = DownloadPending(
+            pending = ModelNotDownloading(
                 shard_metadata=current_status.shard_metadata,
                 node_id=self.node_id,
                 model_directory=self._default_model_dir(model_id),
@@ -321,32 +403,42 @@ class DownloadCoordinator:
             )
             del self.download_status[model_id]
 
+        return True
+
     async def _emit_existing_download_progress(self) -> None:
         while True:
             try:
                 logger.debug(
                     "DownloadCoordinator: Fetching and emitting existing download progress..."
                 )
+
                 async for (
                     _,
                     progress,
                 ) in self.shard_downloader.get_shard_download_status():
                     model_id = progress.shard.model_card.model_id
 
+                    # Don't overwrite status while deletion is in progress
+                    if model_id in self._deleting:
+                        continue
+
                     # Active downloads emit progress via the callback — don't overwrite
                     if model_id in self.active_downloads:
                         continue
 
+                    if isinstance(self.download_status.get(model_id), ModelRejected):
+                        continue
+
                     if progress.status == "complete":
                         found = await to_thread.run_sync(
                             resolve_existing_model, model_id
                         )
                         if found is not None:
-                            status: DownloadProgress = self._completed_from_path(
+                            status: ModelStatus = self._completed_from_path(
                                 progress.shard, found, progress.total
                             )
                         else:
-                            status = DownloadCompleted(
+                            status = ModelReady(
                                 node_id=self.node_id,
                                 shard_metadata=progress.shard,
                                 total=progress.total,
@@ -354,7 +446,7 @@ class DownloadCoordinator:
                             )
                     elif progress.status in ["in_progress", "not_started"]:
                         if progress.downloaded_this_session.in_bytes == 0:
-                            status = DownloadPending(
+                            status = ModelNotDownloading(
                                 node_id=self.node_id,
                                 shard_metadata=progress.shard,
                                 model_directory=self._default_model_dir(model_id),
@@ -362,7 +454,7 @@ class DownloadCoordinator:
                                 total=progress.total,
                             )
                         else:
-                            status = DownloadOngoing(
+                            status = ModelDownloading(
                                 node_id=self.node_id,
                                 shard_metadata=progress.shard,
                                 download_progress=map_repo_download_progress_to_download_progress_data(
@@ -385,7 +477,7 @@ class DownloadCoordinator:
                             continue
                         if isinstance(
                             self.download_status.get(mid),
-                            (DownloadCompleted, DownloadOngoing, DownloadFailed),
+                            (ModelReady, ModelDownloading, ModelDownloadFailed),
                         ):
                             continue
                         found = await to_thread.run_sync(resolve_existing_model, mid)
@@ -398,7 +490,7 @@ class DownloadCoordinator:
                                 end_layer=card.n_layers,
                                 n_layers=card.n_layers,
                             )
-                            path_completed: DownloadProgress = (
+                            path_completed: ModelStatus = (
                                 self._completed_from_path(
                                     path_shard, found, card.storage_size
                                 )
@@ -416,3 +508,114 @@ class DownloadCoordinator:
                     f"DownloadCoordinator: Error emitting existing download progress: {e}"
                 )
             await anyio.sleep(60)
+
+    async def _reject_download(
+        self, shard: ShardMetadata, reason: str, available: Memory
+    ) -> None:
+        assert self.storage_config.max_storage is not None
+        model_id = shard.model_card.model_id
+        rejected = ModelRejected(
+            shard_metadata=shard,
+            node_id=self.node_id,
+            model_directory=self._default_model_dir(model_id),
+            reason=reason,
+            required=shard.model_card.storage_size,
+            available=available if available.in_bytes > 0 else Memory(),
+            limit=self.storage_config.max_storage,
+        )
+        self.download_status[model_id] = rejected
+        await self.event_sender.send(NodeDownloadProgress(download_progress=rejected))
+
+    async def _execute_evictions(
+        self, model_ids: list[ModelId], shard: ShardMetadata
+    ) -> bool:
+        """Execute disk deletions for the given model IDs. Returns False on failure."""
+        target_model_id = shard.model_card.model_id
+
+        for evict_model_id in model_ids:
+            logger.info(
+                f"Auto-evicting model {evict_model_id} to free space for {target_model_id}"
+            )
+            evicted_status = self.download_status.get(evict_model_id)
+            self._deleting.add(evict_model_id)
+            try:
+                success = await self._remove_model_from_disk(evict_model_id)
+            finally:
+                self._deleting.discard(evict_model_id)
+
+            if not success:
+                current_used = calculate_used_storage(
+                    list(self.download_status.values())
+                )
+                assert self.storage_config.max_storage is not None
+                current_available = self.storage_config.max_storage - current_used
+                await self._reject_download(
+                    shard,
+                    f"Failed to delete model {evict_model_id} from disk",
+                    current_available,
+                )
+                return False
+
+            if evicted_status is not None:
+                not_downloading = ModelNotDownloading(
+                    shard_metadata=evicted_status.shard_metadata,
+                    node_id=self.node_id,
+                    model_directory=self._default_model_dir(evict_model_id),
+                )
+                await self.event_sender.send(
+                    NodeDownloadProgress(download_progress=not_downloading)
+                )
+                del self.download_status[evict_model_id]
+
+        return True
+
+    async def clear_rejections(self) -> None:
+        rejected = [
+            (model_id, status)
+            for model_id, status in self.download_status.items()
+            if isinstance(status, ModelRejected)
+        ]
+        for model_id, status in rejected:
+            logger.info(
+                f"Clearing ModelRejected for {model_id} after storage config change"
+            )
+            pending = ModelNotDownloading(
+                shard_metadata=status.shard_metadata,
+                node_id=self.node_id,
+                model_directory=self._default_model_dir(model_id),
+            )
+            self.download_status[model_id] = pending
+            await self.event_sender.send(
+                NodeDownloadProgress(download_progress=pending)
+            )
+
+    async def update_active_models(self, active_model_ids: set[ModelId]) -> None:
+        new_models = active_model_ids - self._active_model_ids
+        for mid in new_models:
+            self._model_last_used[mid] = datetime.now(UTC)
+        self._active_model_ids = active_model_ids.copy()
+        if new_models:
+            await self._persist_model_usage()
+
+    async def _persist_model_usage(self) -> None:
+        try:
+            await aios.makedirs(EXO_MODEL_USAGE_FILE.parent, exist_ok=True)
+            data = {mid: ts.isoformat() for mid, ts in self._model_last_used.items()}
+            async with aiofiles.open(EXO_MODEL_USAGE_FILE, "w") as f:
+                await f.write(json.dumps(data))
+        except Exception as e:
+            logger.warning(f"Failed to persist model usage: {e}")
+
+    async def _load_model_usage(self) -> None:
+        try:
+            if await aios.path.exists(EXO_MODEL_USAGE_FILE):
+                async with aiofiles.open(EXO_MODEL_USAGE_FILE, "r") as f:
+                    raw: dict[str, str] = json.loads(await f.read())  # pyright: ignore[reportAny]
+                self._model_last_used = {
+                    ModelId(k): datetime.fromisoformat(v) for k, v in raw.items()
+                }
+                logger.debug(
+                    f"Loaded model usage for {len(self._model_last_used)} models"
+                )
+        except Exception as e:
+            logger.warning(f"Failed to load model usage: {e}")

src/exo/download/shard_downloader.py

@@ -1,6 +1,5 @@
 from abc import ABC, abstractmethod
 from collections.abc import Awaitable
-from copy import copy
 from datetime import timedelta
 from pathlib import Path
 from typing import AsyncIterator, Callable
@@ -77,9 +76,7 @@ class NoopShardDownloader(ShardDownloader):
     async def get_shard_download_status_for_shard(
         self, shard: ShardMetadata
     ) -> RepoDownloadProgress:
-        dp = copy(NOOP_DOWNLOAD_PROGRESS)
-        dp.shard = shard
-        return dp
+        return NOOP_DOWNLOAD_PROGRESS.model_copy(update={"shard": shard})
 
 
 NOOP_DOWNLOAD_PROGRESS = RepoDownloadProgress(

src/exo/download/tests/test_auto_eviction.py

@@ -0,0 +1,546 @@
+"""Tests for auto-eviction in the DownloadCoordinator.
+
+Tests exercise _start_download (the production entry point) to verify that
+storage quota checks and LRU eviction work end-to-end through the coordinator.
+"""
+
+from datetime import UTC, datetime
+from pathlib import Path
+from unittest.mock import AsyncMock, patch
+
+from anyio.streams.memory import MemoryObjectStreamState
+
+from exo.download.coordinator import DownloadCoordinator
+from exo.download.shard_downloader import NoopShardDownloader
+from exo.shared.models.model_cards import ModelCard, ModelId, ModelTask
+from exo.shared.types.commands import ForwarderDownloadCommand
+from exo.shared.types.common import NodeId
+from exo.shared.types.events import Event, IndexedEvent, NodeDownloadProgress
+from exo.shared.types.memory import Memory
+from exo.shared.types.storage import StorageConfig
+from exo.shared.types.worker.downloads import (
+    ModelNotDownloading,
+    ModelReady,
+    ModelRejected,
+)
+from exo.shared.types.worker.shards import PipelineShardMetadata, ShardMetadata
+from exo.utils.channels import Receiver, Sender
+
+MODEL_A = ModelId("org/model-a")
+MODEL_B = ModelId("org/model-b")
+MODEL_C = ModelId("org/model-c")
+MODEL_NEW = ModelId("org/model-new")
+NODE_ID = NodeId("test-node")
+
+
+def _shard(model_id: ModelId, size_gb: float) -> ShardMetadata:
+    return PipelineShardMetadata(
+        model_card=ModelCard(
+            model_id=model_id,
+            storage_size=Memory.from_gb(size_gb),
+            n_layers=32,
+            hidden_size=1000,
+            supports_tensor=True,
+            tasks=[ModelTask.TextGeneration],
+        ),
+        device_rank=0,
+        world_size=1,
+        start_layer=0,
+        end_layer=32,
+        n_layers=32,
+    )
+
+
+def _completed(model_id: ModelId, size_gb: float) -> ModelReady:
+    return ModelReady(
+        node_id=NODE_ID,
+        shard_metadata=_shard(model_id, size_gb),
+        total=Memory.from_gb(size_gb),
+    )
+
+
+def _make_coordinator(
+    storage_config: StorageConfig,
+    download_status: dict[ModelId, ModelReady | ModelRejected],
+    model_last_used: dict[ModelId, datetime] | None = None,
+) -> tuple[DownloadCoordinator, Receiver[Event]]:
+    state = MemoryObjectStreamState[Event](max_buffer_size=100)
+    event_sender = Sender[Event](_state=state)
+    event_receiver = Receiver[Event](_state=state)
+
+    cmd_state: MemoryObjectStreamState[ForwarderDownloadCommand] = (
+        MemoryObjectStreamState(max_buffer_size=100)
+    )
+    cmd_receiver: Receiver[ForwarderDownloadCommand] = Receiver(_state=cmd_state)
+
+    idx_state: MemoryObjectStreamState[IndexedEvent] = MemoryObjectStreamState(
+        max_buffer_size=100
+    )
+    idx_receiver: Receiver[IndexedEvent] = Receiver(_state=idx_state)
+
+    coordinator = DownloadCoordinator(
+        node_id=NODE_ID,
+        shard_downloader=NoopShardDownloader(),
+        download_command_receiver=cmd_receiver,
+        event_receiver=idx_receiver,
+        event_sender=event_sender,
+        storage_config=storage_config,
+    )
+    coordinator.download_status = dict(download_status)
+    if model_last_used is not None:
+        coordinator._model_last_used = model_last_used  # pyright: ignore[reportPrivateUsage]
+
+    return coordinator, event_receiver
+
+
+async def _start_download(
+    coordinator: DownloadCoordinator, shard: ShardMetadata
+) -> None:
+    await coordinator._start_download(shard)  # pyright: ignore[reportPrivateUsage]
+
+
+class TestStartDownloadAutoEviction:
+    """Tests that go through _start_download — the production entry point."""
+
+    @patch(
+        "exo.download.coordinator.delete_model",
+        new_callable=AsyncMock,
+        return_value=True,
+    )
+    @patch("exo.download.coordinator.resolve_existing_model", return_value=None)
+    async def test_evicts_oldest_model_to_fit_new_download(
+        self, _mock_resolve: AsyncMock, mock_delete: AsyncMock
+    ) -> None:
+        """_start_download should trigger auto-eviction of the oldest model."""
+        config = StorageConfig(
+            max_storage=Memory.from_gb(10), storage_policy="auto-evict"
+        )
+        coordinator, _ = _make_coordinator(
+            config,
+            {MODEL_A: _completed(MODEL_A, 4), MODEL_B: _completed(MODEL_B, 4)},
+            model_last_used={
+                MODEL_A: datetime(2024, 1, 1, tzinfo=UTC),
+                MODEL_B: datetime(2024, 6, 1, tzinfo=UTC),
+            },
+        )
+
+        await _start_download(coordinator, _shard(MODEL_NEW, 5))
+
+        # MODEL_A (oldest) should have been evicted
+        mock_delete.assert_called_once_with(MODEL_A)
+        assert MODEL_A not in coordinator.download_status
+
+    @patch(
+        "exo.download.coordinator.delete_model",
+        new_callable=AsyncMock,
+        return_value=True,
+    )
+    @patch("exo.download.coordinator.resolve_existing_model", return_value=None)
+    async def test_evicts_multiple_in_lru_order(
+        self, _mock_resolve: AsyncMock, mock_delete: AsyncMock
+    ) -> None:
+        """_start_download evicts multiple models oldest-first until space is freed."""
+        config = StorageConfig(
+            max_storage=Memory.from_gb(10), storage_policy="auto-evict"
+        )
+        coordinator, _ = _make_coordinator(
+            config,
+            {
+                MODEL_A: _completed(MODEL_A, 3),
+                MODEL_B: _completed(MODEL_B, 3),
+                MODEL_C: _completed(MODEL_C, 3),
+            },
+            model_last_used={
+                MODEL_A: datetime(2024, 1, 1, tzinfo=UTC),
+                MODEL_B: datetime(2024, 6, 1, tzinfo=UTC),
+                MODEL_C: datetime(2024, 12, 1, tzinfo=UTC),
+            },
+        )
+
+        # Need 8 GiB, have 1 GiB free — need to free 7 GiB
+        await _start_download(coordinator, _shard(MODEL_NEW, 8))
+
+        evicted = [call.args[0] for call in mock_delete.call_args_list]
+        assert evicted == [MODEL_A, MODEL_B, MODEL_C]
+
+    @patch(
+        "exo.download.coordinator.delete_model",
+        new_callable=AsyncMock,
+        return_value=True,
+    )
+    @patch("exo.download.coordinator.resolve_existing_model", return_value=None)
+    async def test_rejects_when_cannot_free_enough_space(
+        self, _mock_resolve: AsyncMock, mock_delete: AsyncMock
+    ) -> None:
+        """_start_download emits DownloadRejected when eviction can't free enough."""
+        config = StorageConfig(
+            max_storage=Memory.from_gb(10), storage_policy="auto-evict"
+        )
+        coordinator, _ = _make_coordinator(
+            config,
+            {MODEL_A: _completed(MODEL_A, 2)},
+        )
+
+        await _start_download(coordinator, _shard(MODEL_NEW, 20))
+
+        mock_delete.assert_not_called()
+        assert isinstance(coordinator.download_status[MODEL_NEW], ModelRejected)
+
+    @patch(
+        "exo.download.coordinator.delete_model",
+        new_callable=AsyncMock,
+        return_value=True,
+    )
+    @patch("exo.download.coordinator.resolve_existing_model", return_value=None)
+    async def test_no_eviction_when_space_available(
+        self, _mock_resolve: AsyncMock, mock_delete: AsyncMock
+    ) -> None:
+        """_start_download proceeds without evicting when enough space exists."""
+        config = StorageConfig(
+            max_storage=Memory.from_gb(10), storage_policy="auto-evict"
+        )
+        coordinator, _ = _make_coordinator(
+            config,
+            {MODEL_A: _completed(MODEL_A, 2)},
+        )
+
+        await _start_download(coordinator, _shard(MODEL_NEW, 5))
+
+        mock_delete.assert_not_called()
+        # Download should have started (not rejected)
+        assert MODEL_NEW in coordinator.download_status
+        assert not isinstance(coordinator.download_status[MODEL_NEW], ModelRejected)
+
+    @patch(
+        "exo.download.coordinator.delete_model",
+        new_callable=AsyncMock,
+        return_value=True,
+    )
+    @patch("exo.download.coordinator.resolve_existing_model", return_value=None)
+    async def test_manual_policy_rejects_instead_of_evicting(
+        self, _mock_resolve: AsyncMock, mock_delete: AsyncMock
+    ) -> None:
+        """With manual policy, _start_download rejects instead of auto-evicting."""
+        config = StorageConfig(max_storage=Memory.from_gb(10), storage_policy="manual")
+        coordinator, _ = _make_coordinator(
+            config,
+            {MODEL_A: _completed(MODEL_A, 4), MODEL_B: _completed(MODEL_B, 4)},
+        )
+
+        await _start_download(coordinator, _shard(MODEL_NEW, 5))
+
+        mock_delete.assert_not_called()
+        assert isinstance(coordinator.download_status[MODEL_NEW], ModelRejected)
+
+    @patch(
+        "exo.download.coordinator.delete_model",
+        new_callable=AsyncMock,
+        return_value=True,
+    )
+    @patch("exo.download.coordinator.resolve_existing_model", return_value=None)
+    async def test_eviction_emits_not_downloading_event_for_evicted_model(
+        self, _mock_resolve: AsyncMock, mock_delete: AsyncMock
+    ) -> None:
+        """Evicted models emit ModelNotDownloading events and are removed from status."""
+        config = StorageConfig(
+            max_storage=Memory.from_gb(10), storage_policy="auto-evict"
+        )
+        coordinator, event_receiver = _make_coordinator(
+            config,
+            {MODEL_A: _completed(MODEL_A, 4), MODEL_B: _completed(MODEL_B, 4)},
+            model_last_used={
+                MODEL_A: datetime(2024, 1, 1, tzinfo=UTC),
+                MODEL_B: datetime(2024, 6, 1, tzinfo=UTC),
+            },
+        )
+
+        await _start_download(coordinator, _shard(MODEL_NEW, 5))
+
+        events = event_receiver.collect()
+        eviction_events = [
+            e
+            for e in events
+            if isinstance(e, NodeDownloadProgress)
+            and isinstance(e.download_progress, ModelNotDownloading)
+            and e.download_progress.shard_metadata.model_card.model_id == MODEL_A
+        ]
+        assert len(eviction_events) == 1
+        assert MODEL_A not in coordinator.download_status
+
+
+class TestActiveModelProtection:
+    """Tests that update_active_models protects models from eviction.
+
+    These tests use the public API (update_active_models) to mark models as
+    active, then trigger eviction through _start_download.
+    """
+
+    @patch(
+        "exo.download.coordinator.delete_model",
+        new_callable=AsyncMock,
+        return_value=True,
+    )
+    @patch("exo.download.coordinator.resolve_existing_model", return_value=None)
+    async def test_active_model_not_evicted(
+        self, _mock_resolve: AsyncMock, mock_delete: AsyncMock
+    ) -> None:
+        """A model marked active via update_active_models must not be evicted."""
+        config = StorageConfig(
+            max_storage=Memory.from_gb(10), storage_policy="auto-evict"
+        )
+        coordinator, _ = _make_coordinator(
+            config,
+            {MODEL_A: _completed(MODEL_A, 4), MODEL_B: _completed(MODEL_B, 4)},
+            model_last_used={
+                MODEL_A: datetime(2024, 1, 1, tzinfo=UTC),  # oldest
+                MODEL_B: datetime(2024, 6, 1, tzinfo=UTC),
+            },
+        )
+
+        # Mark MODEL_A as active through the public API
+        await coordinator.update_active_models({MODEL_A})
+
+        await _start_download(coordinator, _shard(MODEL_NEW, 5))
+
+        # MODEL_A is active — MODEL_B should be evicted instead
+        mock_delete.assert_called_once_with(MODEL_B)
+        assert MODEL_A in coordinator.download_status
+
+    @patch(
+        "exo.download.coordinator.delete_model",
+        new_callable=AsyncMock,
+        return_value=True,
+    )
+    @patch("exo.download.coordinator.resolve_existing_model", return_value=None)
+    async def test_all_active_models_rejected(
+        self, _mock_resolve: AsyncMock, mock_delete: AsyncMock
+    ) -> None:
+        """When all models are active, eviction is impossible — download is rejected."""
+        config = StorageConfig(
+            max_storage=Memory.from_gb(10), storage_policy="auto-evict"
+        )
+        coordinator, _ = _make_coordinator(
+            config,
+            {MODEL_A: _completed(MODEL_A, 4), MODEL_B: _completed(MODEL_B, 4)},
+        )
+
+        await coordinator.update_active_models({MODEL_A, MODEL_B})
+
+        await _start_download(coordinator, _shard(MODEL_NEW, 5))
+
+        mock_delete.assert_not_called()
+        assert isinstance(coordinator.download_status[MODEL_NEW], ModelRejected)
+
+
+class TestDiskDeleteFailure:
+    """Tests that eviction fails properly when disk delete fails."""
+
+    @patch(
+        "exo.download.coordinator.delete_model",
+        new_callable=AsyncMock,
+        return_value=False,
+    )
+    @patch("exo.download.coordinator.resolve_existing_model", return_value=None)
+    async def test_eviction_rejected_on_disk_delete_failure(
+        self, _mock_resolve: AsyncMock, mock_delete: AsyncMock
+    ) -> None:
+        """When disk delete fails, auto-eviction emits DownloadRejected."""
+        config = StorageConfig(
+            max_storage=Memory.from_gb(10), storage_policy="auto-evict"
+        )
+        coordinator, _ = _make_coordinator(
+            config,
+            {MODEL_A: _completed(MODEL_A, 4), MODEL_B: _completed(MODEL_B, 4)},
+            model_last_used={
+                MODEL_A: datetime(2024, 1, 1, tzinfo=UTC),
+                MODEL_B: datetime(2024, 6, 1, tzinfo=UTC),
+            },
+        )
+
+        await _start_download(coordinator, _shard(MODEL_NEW, 5))
+
+        # Should have tried to delete oldest model and failed
+        mock_delete.assert_called_once_with(MODEL_A)
+        # New model should be rejected
+        assert isinstance(coordinator.download_status[MODEL_NEW], ModelRejected)
+        # Eviction target should still be in download_status (not removed)
+        assert MODEL_A in coordinator.download_status
+
+
+class TestLruPersistence:
+    """Tests for _persist_model_usage and _load_model_usage round-trip."""
+
+    async def test_persist_then_load_round_trip(self, tmp_path: Path) -> None:
+        """Persisting then loading recovers the same data."""
+        usage_file = tmp_path / "model_usage.json"
+        coordinator, _ = _make_coordinator(StorageConfig(), {})
+        coordinator._model_last_used = {  # pyright: ignore[reportPrivateUsage]
+            MODEL_A: datetime(2024, 1, 15, 12, 30, 0, tzinfo=UTC),
+            MODEL_B: datetime(2024, 6, 1, 0, 0, 0, tzinfo=UTC),
+        }
+
+        with patch("exo.download.coordinator.EXO_MODEL_USAGE_FILE", usage_file):
+            await coordinator._persist_model_usage()  # pyright: ignore[reportPrivateUsage]
+
+            # Create a fresh coordinator and load
+            coordinator2, _ = _make_coordinator(StorageConfig(), {})
+            await coordinator2._load_model_usage()  # pyright: ignore[reportPrivateUsage]
+
+        assert coordinator2._model_last_used == coordinator._model_last_used  # pyright: ignore[reportPrivateUsage]
+
+    async def test_load_missing_file_returns_empty(self, tmp_path: Path) -> None:
+        """Loading when file doesn't exist returns empty dict."""
+        usage_file = tmp_path / "nonexistent" / "model_usage.json"
+        coordinator, _ = _make_coordinator(StorageConfig(), {})
+
+        with patch("exo.download.coordinator.EXO_MODEL_USAGE_FILE", usage_file):
+            await coordinator._load_model_usage()  # pyright: ignore[reportPrivateUsage]
+
+        assert coordinator._model_last_used == {}  # pyright: ignore[reportPrivateUsage]
+
+    async def test_load_corrupt_json_returns_empty(self, tmp_path: Path) -> None:
+        """Loading corrupt JSON logs warning and returns empty dict."""
+        usage_file = tmp_path / "model_usage.json"
+        usage_file.write_text("not valid json {{{")
+        coordinator, _ = _make_coordinator(StorageConfig(), {})
+
+        with patch("exo.download.coordinator.EXO_MODEL_USAGE_FILE", usage_file):
+            await coordinator._load_model_usage()  # pyright: ignore[reportPrivateUsage]
+
+        assert coordinator._model_last_used == {}  # pyright: ignore[reportPrivateUsage]
+
+
+class TestEvictionEvents:
+    """Tests that eviction emits the correct sequence of events."""
+
+    @patch(
+        "exo.download.coordinator.delete_model",
+        new_callable=AsyncMock,
+        return_value=True,
+    )
+    @patch("exo.download.coordinator.resolve_existing_model", return_value=None)
+    async def test_eviction_emits_not_downloading_event(
+        self, _mock_resolve: AsyncMock, mock_delete: AsyncMock
+    ) -> None:
+        """Eviction emits a ModelNotDownloading event and removes from status."""
+        config = StorageConfig(
+            max_storage=Memory.from_gb(10), storage_policy="auto-evict"
+        )
+        coordinator, event_receiver = _make_coordinator(
+            config,
+            {MODEL_A: _completed(MODEL_A, 4), MODEL_B: _completed(MODEL_B, 4)},
+            model_last_used={
+                MODEL_A: datetime(2024, 1, 1, tzinfo=UTC),
+                MODEL_B: datetime(2024, 6, 1, tzinfo=UTC),
+            },
+        )
+
+        await _start_download(coordinator, _shard(MODEL_NEW, 5))
+
+        events = event_receiver.collect()
+        eviction_events = [
+            e
+            for e in events
+            if isinstance(e, NodeDownloadProgress)
+            and isinstance(e.download_progress, ModelNotDownloading)
+            and e.download_progress.shard_metadata.model_card.model_id == MODEL_A
+        ]
+        assert len(eviction_events) == 1
+        assert MODEL_A not in coordinator.download_status
+
+    @patch(
+        "exo.download.coordinator.delete_model",
+        new_callable=AsyncMock,
+        return_value=True,
+    )
+    @patch("exo.download.coordinator.resolve_existing_model", return_value=None)
+    async def test_multi_eviction_emits_event_per_model(
+        self, _mock_resolve: AsyncMock, _mock_delete: AsyncMock
+    ) -> None:
+        """Each evicted model gets its own ModelNotDownloading event."""
+        config = StorageConfig(
+            max_storage=Memory.from_gb(10), storage_policy="auto-evict"
+        )
+        coordinator, event_receiver = _make_coordinator(
+            config,
+            {
+                MODEL_A: _completed(MODEL_A, 3),
+                MODEL_B: _completed(MODEL_B, 3),
+                MODEL_C: _completed(MODEL_C, 3),
+            },
+            model_last_used={
+                MODEL_A: datetime(2024, 1, 1, tzinfo=UTC),
+                MODEL_B: datetime(2024, 6, 1, tzinfo=UTC),
+                MODEL_C: datetime(2024, 12, 1, tzinfo=UTC),
+            },
+        )
+
+        await _start_download(coordinator, _shard(MODEL_NEW, 8))
+
+        events = event_receiver.collect()
+        eviction_events = [
+            e
+            for e in events
+            if isinstance(e, NodeDownloadProgress)
+            and isinstance(e.download_progress, ModelNotDownloading)
+            and e.download_progress.shard_metadata.model_card.model_id != MODEL_NEW
+        ]
+        evicted_model_ids = [
+            e.download_progress.shard_metadata.model_card.model_id
+            for e in eviction_events
+        ]
+        assert evicted_model_ids == [MODEL_A, MODEL_B, MODEL_C]
+        for mid in [MODEL_A, MODEL_B, MODEL_C]:
+            assert mid not in coordinator.download_status
+
+
+class TestClearRejections:
+    """Tests for clear_rejections behavior."""
+
+    async def test_clear_rejections_resets_rejected(self) -> None:
+        """clear_rejections resets ModelRejected to ModelNotDownloading."""
+        config = StorageConfig(
+            max_storage=Memory.from_gb(10), storage_policy="auto-evict"
+        )
+        rejected = ModelRejected(
+            node_id=NODE_ID,
+            shard_metadata=_shard(MODEL_B, 4),
+            reason="Not enough space",
+            required=Memory.from_gb(4),
+            available=Memory.from_gb(1),
+            limit=Memory.from_gb(10),
+        )
+        coordinator, _ = _make_coordinator(
+            config,
+            {MODEL_A: _completed(MODEL_A, 4), MODEL_B: rejected},
+        )
+
+        await coordinator.clear_rejections()
+
+        # Completed should remain unchanged
+        assert isinstance(coordinator.download_status[MODEL_A], ModelReady)
+        # Rejected should be cleared
+        assert isinstance(coordinator.download_status[MODEL_B], ModelNotDownloading)
+
+    async def test_clear_rejections_on_policy_only_change(self) -> None:
+        """clear_rejections fires even when only the policy changes (no limit change)."""
+        config = StorageConfig(max_storage=Memory.from_gb(10), storage_policy="manual")
+        rejected = ModelRejected(
+            node_id=NODE_ID,
+            shard_metadata=_shard(MODEL_A, 4),
+            reason="Manual policy",
+            required=Memory.from_gb(4),
+            available=Memory.from_gb(1),
+            limit=Memory.from_gb(10),
+        )
+        coordinator, _ = _make_coordinator(
+            config,
+            {MODEL_A: rejected, MODEL_B: _completed(MODEL_B, 3)},
+        )
+
+        await coordinator.clear_rejections()
+
+        # Rejected should be cleared to Pending
+        assert isinstance(coordinator.download_status[MODEL_A], ModelNotDownloading)
+        # Completed should be unchanged
+        assert isinstance(coordinator.download_status[MODEL_B], ModelReady)

src/exo/download/tests/test_re_download.py

@@ -19,9 +19,9 @@ from exo.shared.types.commands import (
     StartDownload,
 )
 from exo.shared.types.common import NodeId, SystemId
-from exo.shared.types.events import Event, NodeDownloadProgress
+from exo.shared.types.events import Event, IndexedEvent, NodeDownloadProgress
 from exo.shared.types.memory import Memory
-from exo.shared.types.worker.downloads import DownloadCompleted
+from exo.shared.types.worker.downloads import ModelReady
 from exo.shared.types.worker.shards import PipelineShardMetadata, ShardMetadata
 from exo.utils.channels import Receiver, Sender, channel
 
@@ -132,6 +132,7 @@ async def test_re_download_after_delete_completes() -> None:
     cmd_send: Sender[ForwarderDownloadCommand]
     cmd_send, cmd_recv = channel[ForwarderDownloadCommand]()
     event_send, event_recv = channel[Event]()
+    _idx_send, idx_recv = channel[IndexedEvent]()
 
     fake_downloader = FakeShardDownloader()
     wrapped_downloader = SingletonShardDownloader(fake_downloader)
@@ -139,6 +140,7 @@ async def test_re_download_after_delete_completes() -> None:
         node_id=NODE_ID,
         shard_downloader=wrapped_downloader,
         download_command_receiver=cmd_recv,
+        event_receiver=idx_recv,
         event_sender=event_send,
     )
 
@@ -194,7 +196,7 @@ async def test_re_download_after_delete_completes() -> None:
 
 async def _wait_for_download_completed(
     event_recv: Receiver[Event], model_id: ModelId, timeout: float = 2.0
-) -> DownloadCompleted | None:
+) -> ModelReady | None:
     """Drain events until we see a DownloadCompleted for the given model, or timeout."""
     try:
         async with asyncio.timeout(timeout):
@@ -202,7 +204,7 @@ async def _wait_for_download_completed(
                 event = await event_recv.receive()
                 if (
                     isinstance(event, NodeDownloadProgress)
-                    and isinstance(event.download_progress, DownloadCompleted)
+                    and isinstance(event.download_progress, ModelReady)
                     and event.download_progress.shard_metadata.model_card.model_id
                     == model_id
                 ):

src/exo/main.py

@@ -20,7 +20,9 @@ from exo.routing.router import Router, get_node_id_keypair
 from exo.shared.constants import EXO_LOG
 from exo.shared.election import Election, ElectionResult
 from exo.shared.logging import logger_cleanup, logger_setup
+from exo.shared.storage import load_storage_config
 from exo.shared.types.common import NodeId, SessionId
+from exo.shared.types.storage import StoragePolicy
 from exo.utils.channels import Receiver, channel
 from exo.utils.pydantic_ext import CamelCaseModel
 from exo.utils.task_group import TaskGroup
@@ -67,14 +69,21 @@ class Node:
 
         logger.info(f"Starting node {node_id}")
 
+        storage_config = await load_storage_config(
+            max_storage_gb=args.max_storage_gb,
+            storage_policy=args.storage_policy,
+        )
+
         # Create DownloadCoordinator (unless --no-downloads)
         if not args.no_downloads:
             download_coordinator = DownloadCoordinator(
                 node_id,
                 exo_shard_downloader(offline=args.offline),
-                event_sender=event_router.sender(),
                 download_command_receiver=router.receiver(topics.DOWNLOAD_COMMANDS),
+                event_receiver=event_router.receiver(),
+                event_sender=event_router.sender(),
                 offline=args.offline,
+                storage_config=storage_config,
             )
         else:
             download_coordinator = None
@@ -229,15 +238,20 @@ class Node:
                 if result.is_new_master:
                     if self.download_coordinator:
                         await self.download_coordinator.shutdown()
+                        storage_config = self.download_coordinator.storage_config
+                        active_model_ids = self.download_coordinator._active_model_ids  # pyright: ignore[reportPrivateUsage]
                         self.download_coordinator = DownloadCoordinator(
                             self.node_id,
                             exo_shard_downloader(offline=self.offline),
-                            event_sender=self.event_router.sender(),
                             download_command_receiver=self.router.receiver(
                                 topics.DOWNLOAD_COMMANDS
                             ),
+                            event_receiver=self.event_router.receiver(),
+                            event_sender=self.event_router.sender(),
                             offline=self.offline,
+                            storage_config=storage_config,
                         )
+                        self.download_coordinator._active_model_ids = active_model_ids  # pyright: ignore[reportPrivateUsage]
                         self._tg.start_soon(self.download_coordinator.run)
                     if self.worker:
                         await self.worker.shutdown()
@@ -317,6 +331,8 @@ class Args(CamelCaseModel):
     fast_synch: bool | None = None  # None = auto, True = force on, False = force off
     bootstrap_peers: list[str] = []
     libp2p_port: int
+    max_storage_gb: float | None = None
+    storage_policy: StoragePolicy | None = None
 
     @classmethod
     def parse(cls) -> Self:
@@ -404,6 +420,20 @@ class Args(CamelCaseModel):
             dest="fast_synch",
             help="Force MLX FAST_SYNCH off",
         )
+        parser.add_argument(
+            "--max-storage-gb",
+            type=float,
+            dest="max_storage_gb",
+            default=None,
+            help="Maximum storage for downloaded models in GB (default: unlimited)",
+        )
+        parser.add_argument(
+            "--storage-policy",
+            choices=["manual", "auto-evict"],
+            dest="storage_policy",
+            default=None,
+            help="Storage policy: 'manual' rejects on exceed, 'auto-evict' removes LRU models (default: manual)",
+        )
 
         args = parser.parse_args()
         return cls(**vars(args))  # pyright: ignore[reportAny] - We are intentionally validating here, we can't do it statically

src/exo/master/main.py

@@ -12,6 +12,7 @@ from exo.master.placement import (
 )
 from exo.shared.apply import apply
 from exo.shared.constants import EXO_EVENT_LOG_DIR, EXO_TRACING_ENABLED
+from exo.shared.storage import get_download_rejected_events
 from exo.shared.types.commands import (
     AddCustomModelCard,
     CreateInstance,
@@ -24,6 +25,7 @@ from exo.shared.types.commands import (
     PlaceInstance,
     RequestEventLog,
     SendInputChunk,
+    SetStorageConfig,
     TaskCancelled,
     TaskFinished,
     TestCommand,
@@ -39,8 +41,10 @@ from exo.shared.types.events import (
     InputChunkReceived,
     InstanceDeleted,
     LocalForwarderEvent,
+    NodeDownloadProgress,
     NodeGatheredInfo,
     NodeTimedOut,
+    StorageConfigUpdated,
     TaskCreated,
     TaskDeleted,
     TaskStatusUpdated,
@@ -49,6 +53,7 @@ from exo.shared.types.events import (
     TracesMerged,
 )
 from exo.shared.types.state import State
+from exo.shared.types.storage import StorageConfig
 from exo.shared.types.tasks import (
     ImageEdits as ImageEditsTask,
 )
@@ -62,6 +67,7 @@ from exo.shared.types.tasks import (
 from exo.shared.types.tasks import (
     TextGeneration as TextGenerationTask,
 )
+from exo.shared.types.worker.downloads import ModelRejected
 from exo.shared.types.worker.instances import InstanceId
 from exo.utils.channels import Receiver, Sender
 from exo.utils.disk_event_log import DiskEventLog
@@ -357,6 +363,16 @@ class Master:
                             generated_events.append(
                                 CustomModelCardDeleted(model_id=command.model_id)
                             )
+                        case SetStorageConfig():
+                            generated_events.append(
+                                StorageConfigUpdated(
+                                    node_id=command.target_node_id,
+                                    storage_config=StorageConfig(
+                                        max_storage=command.max_storage,
+                                        storage_policy=command.storage_policy,
+                                    ),
+                                )
+                            )
                         case RequestEventLog():
                             # We should just be able to send everything, since other buffers will ignore old messages
                             # rate limit to 1000 at a time
@@ -413,6 +429,20 @@ class Master:
                     indexed = IndexedEvent(event=event, idx=len(self._event_log))
                     self.state = apply(self.state, indexed)
 
+                    if isinstance(event, NodeDownloadProgress) and isinstance(
+                        event.download_progress, ModelRejected
+                    ):
+                        dp = event.download_progress
+                        rejection_events = get_download_rejected_events(
+                            dp.shard_metadata.model_card.model_id,
+                            dp.node_id,
+                            self.state.instances,
+                            self.state.tasks,
+                        )
+                        for rejection_event in rejection_events:
+                            logger.info(f"Download rejected cleanup: {rejection_event}")
+                            await self.event_sender.send(rejection_event)
+
                     event._master_time_stamp = datetime.now(tz=timezone.utc)  # pyright: ignore[reportPrivateUsage]
                     if isinstance(event, NodeGatheredInfo):
                         event.when = str(datetime.now(tz=timezone.utc))

src/exo/master/placement.py

@@ -32,11 +32,12 @@ 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 (
-    DownloadCompleted,
-    DownloadFailed,
-    DownloadOngoing,
-    DownloadPending,
-    DownloadProgress,
+    ModelDownloadFailed,
+    ModelDownloading,
+    ModelNotDownloading,
+    ModelReady,
+    ModelRejected,
+    ModelStatus,
 )
 from exo.shared.types.worker.instances import (
     Instance,
@@ -66,26 +67,28 @@ def add_instance_to_placements(
 def _get_node_download_fraction(
     node_id: NodeId,
     model_id: ModelId,
-    download_status: Mapping[NodeId, Sequence[DownloadProgress]],
+    download_status: Mapping[NodeId, Sequence[ModelStatus]],
 ) -> float:
     """Return the download fraction (0.0–1.0) for a model on a given node."""
     for progress in download_status.get(node_id, []):
         if progress.shard_metadata.model_card.model_id != model_id:
             continue
         match progress:
-            case DownloadCompleted():
+            case ModelReady():
                 return 1.0
-            case DownloadOngoing():
+            case ModelDownloading():
                 total = progress.download_progress.total.in_bytes
                 return (
                     progress.download_progress.downloaded.in_bytes / total
                     if total > 0
                     else 0.0
                 )
-            case DownloadPending():
+            case ModelNotDownloading():
                 total = progress.total.in_bytes
                 return progress.downloaded.in_bytes / total if total > 0 else 0.0
-            case DownloadFailed():
+            case ModelDownloadFailed():
+                return 0.0
+            case ModelRejected():
                 return 0.0
     return 0.0
 
@@ -93,7 +96,7 @@ def _get_node_download_fraction(
 def _cycle_download_score(
     cycle: Cycle,
     model_id: ModelId,
-    download_status: Mapping[NodeId, Sequence[DownloadProgress]],
+    download_status: Mapping[NodeId, Sequence[ModelStatus]],
 ) -> float:
     """Sum of download fractions across all nodes in a cycle."""
     return sum(
@@ -109,7 +112,7 @@ def place_instance(
     node_memory: Mapping[NodeId, MemoryUsage],
     node_network: Mapping[NodeId, NodeNetworkInfo],
     required_nodes: set[NodeId] | None = None,
-    download_status: Mapping[NodeId, Sequence[DownloadProgress]] | None = None,
+    download_status: Mapping[NodeId, Sequence[ModelStatus]] | None = None,
 ) -> dict[InstanceId, Instance]:
     cycles = topology.get_cycles()
     candidate_cycles = list(filter(lambda it: len(it) >= command.min_nodes, cycles))
@@ -309,14 +312,14 @@ def get_transition_events(
 
 def cancel_unnecessary_downloads(
     instances: Mapping[InstanceId, Instance],
-    download_status: Mapping[NodeId, Sequence[DownloadProgress]],
+    download_status: Mapping[NodeId, Sequence[ModelStatus]],
 ) -> Sequence[DownloadCommand]:
     commands: list[DownloadCommand] = []
     currently_downloading = [
         (k, v.shard_metadata.model_card.model_id)
         for k, vs in download_status.items()
         for v in vs
-        if isinstance(v, (DownloadOngoing))
+        if isinstance(v, (ModelDownloading))
     ]
     active_models = set(
         (

src/exo/master/tests/test_placement.py

@@ -26,10 +26,10 @@ from exo.shared.types.tasks import TaskId, TaskStatus, TextGeneration
 from exo.shared.types.text_generation import InputMessage, TextGenerationTaskParams
 from exo.shared.types.topology import Connection, SocketConnection
 from exo.shared.types.worker.downloads import (
-    DownloadCompleted,
-    DownloadFailed,
-    DownloadOngoing,
     DownloadProgressData,
+    ModelDownloadFailed,
+    ModelDownloading,
+    ModelReady,
 )
 from exo.shared.types.worker.instances import (
     Instance,
@@ -624,7 +624,7 @@ def test_placement_prefers_cycle_with_downloaded_model(
     # node_b has the model fully downloaded, node_a does not
     download_status = {
         node_b: [
-            DownloadCompleted(
+            ModelReady(
                 node_id=node_b,
                 shard_metadata=shard_meta,
                 total=model_card.storage_size,
@@ -671,7 +671,7 @@ def test_placement_prefers_cycle_with_higher_download_progress(
     # node_a: 30% downloaded, node_b: 80% downloaded
     download_status = {
         node_a: [
-            DownloadOngoing(
+            ModelDownloading(
                 node_id=node_a,
                 shard_metadata=shard_meta,
                 download_progress=DownloadProgressData(
@@ -687,7 +687,7 @@ def test_placement_prefers_cycle_with_higher_download_progress(
             ),
         ],
         node_b: [
-            DownloadOngoing(
+            ModelDownloading(
                 node_id=node_b,
                 shard_metadata=shard_meta,
                 download_progress=DownloadProgressData(
@@ -744,7 +744,7 @@ def test_placement_does_not_prefer_cycle_with_failed_download(
     # node_b has a failed download — should not be preferred
     download_status = {
         node_b: [
-            DownloadFailed(
+            ModelDownloadFailed(
                 node_id=node_b,
                 shard_metadata=shard_meta,
                 error_message="connection reset",

src/exo/shared/apply.py

@@ -18,6 +18,7 @@ from exo.shared.types.events import (
     NodeGatheredInfo,
     NodeTimedOut,
     RunnerStatusUpdated,
+    StorageConfigUpdated,
     TaskAcknowledged,
     TaskCreated,
     TaskDeleted,
@@ -39,7 +40,7 @@ from exo.shared.types.profiling import (
 from exo.shared.types.state import State
 from exo.shared.types.tasks import Task, TaskId, TaskStatus
 from exo.shared.types.topology import Connection, RDMAConnection
-from exo.shared.types.worker.downloads import DownloadProgress
+from exo.shared.types.worker.downloads import ModelStatus
 from exo.shared.types.worker.instances import Instance, InstanceId
 from exo.shared.types.worker.runners import RunnerId, RunnerShutdown, RunnerStatus
 from exo.utils.info_gatherer.info_gatherer import (
@@ -95,6 +96,8 @@ def event_apply(event: Event, state: State) -> State:
             return apply_topology_edge_created(event, state)
         case TopologyEdgeDeleted():
             return apply_topology_edge_deleted(event, state)
+        case StorageConfigUpdated():
+            return apply_storage_config_updated(event, state)
 
 
 def apply(state: State, event: IndexedEvent) -> State:
@@ -114,18 +117,13 @@ def apply_node_download_progress(event: NodeDownloadProgress, state: State) -> S
     """
     dp = event.download_progress
     node_id = dp.node_id
+    model_id = dp.shard_metadata.model_card.model_id
 
     current = list(state.downloads.get(node_id, ()))
 
     replaced = False
     for i, existing_dp in enumerate(current):
-        # TODO(ciaran): deduplicate by model_id for now. Will need to use
-        # shard_metadata again when pipeline and tensor downloads differ.
-        # For now this is fine
-        if (
-            existing_dp.shard_metadata.model_card.model_id
-            == dp.shard_metadata.model_card.model_id
-        ):
+        if existing_dp.shard_metadata.model_card.model_id == model_id:
             current[i] = dp
             replaced = True
             break
@@ -133,7 +131,7 @@ def apply_node_download_progress(event: NodeDownloadProgress, state: State) -> S
     if not replaced:
         current.append(dp)
 
-    new_downloads: Mapping[NodeId, Sequence[DownloadProgress]] = {
+    new_downloads: Mapping[NodeId, Sequence[ModelStatus]] = {
         **state.downloads,
         node_id: current,
     }
@@ -245,6 +243,11 @@ def apply_node_timed_out(event: NodeTimedOut, state: State) -> State:
     node_rdma_ctl = {
         key: value for key, value in state.node_rdma_ctl.items() if key != event.node_id
     }
+    node_storage_config = {
+        key: value
+        for key, value in state.node_storage_config.items()
+        if key != event.node_id
+    }
     # Only recompute cycles if the leaving node had TB bridge enabled
     leaving_node_status = state.node_thunderbolt_bridge.get(event.node_id)
     leaving_node_had_tb_enabled = (
@@ -267,6 +270,7 @@ def apply_node_timed_out(event: NodeTimedOut, state: State) -> State:
             "node_thunderbolt": node_thunderbolt,
             "node_thunderbolt_bridge": node_thunderbolt_bridge,
             "node_rdma_ctl": node_rdma_ctl,
+            "node_storage_config": node_storage_config,
             "thunderbolt_bridge_cycles": thunderbolt_bridge_cycles,
         }
     )
@@ -298,7 +302,10 @@ def apply_node_gathered_info(event: NodeGatheredInfo, state: State) -> State:
         case NodeDiskUsage():
             update["node_disk"] = {**state.node_disk, event.node_id: info.disk_usage}
         case NodeConfig():
-            pass
+            update["node_storage_config"] = {
+                **state.node_storage_config,
+                event.node_id: info.storage_config,
+            }
         case MiscData():
             current_identity = state.node_identities.get(event.node_id, NodeIdentity())
             new_identity = current_identity.model_copy(
@@ -377,6 +384,14 @@ def apply_node_gathered_info(event: NodeGatheredInfo, state: State) -> State:
     return state.model_copy(update=update)
 
 
+def apply_storage_config_updated(event: StorageConfigUpdated, state: State) -> State:
+    new_node_storage_config = {
+        **state.node_storage_config,
+        event.node_id: event.storage_config,
+    }
+    return state.model_copy(update={"node_storage_config": new_node_storage_config})
+
+
 def apply_topology_edge_created(event: TopologyEdgeCreated, state: State) -> State:
     topology = copy.deepcopy(state.topology)
     topology.add_connection(event.conn)

src/exo/shared/constants.py

@@ -94,6 +94,8 @@ EXO_ENABLE_IMAGE_MODELS = (
 
 EXO_OFFLINE = os.getenv("EXO_OFFLINE", "false").lower() == "true"
 
+EXO_MODEL_USAGE_FILE = EXO_DATA_HOME / "model_usage.json"
+
 EXO_TRACING_ENABLED = os.getenv("EXO_TRACING_ENABLED", "false").lower() == "true"
 
 EXO_MAX_CONCURRENT_REQUESTS = int(os.getenv("EXO_MAX_CONCURRENT_REQUESTS", "8"))

src/exo/shared/storage.py

@@ -0,0 +1,221 @@
+import tomllib
+from collections.abc import Mapping, Sequence
+from datetime import UTC, datetime
+
+import anyio
+import tomlkit
+from loguru import logger
+from tomlkit.exceptions import TOMLKitError
+
+from exo.shared.constants import EXO_CONFIG_FILE
+from exo.shared.models.model_cards import ModelId
+from exo.shared.types.common import NodeId
+from exo.shared.types.events import InstanceDeleted, TaskStatusUpdated
+from exo.shared.types.memory import Memory
+from exo.shared.types.storage import (
+    StorageAllow,
+    StorageConfig,
+    StorageDecision,
+    StorageEvict,
+    StoragePolicy,
+    StorageReject,
+)
+from exo.shared.types.tasks import Task, TaskId, TaskStatus
+from exo.shared.types.worker.downloads import (
+    ModelDownloading,
+    ModelReady,
+    ModelStatus,
+)
+from exo.shared.types.worker.instances import Instance, InstanceId
+
+
+def calculate_used_storage(downloads: Sequence[ModelStatus]) -> Memory:
+    total = Memory()
+    for dp in downloads:
+        if isinstance(dp, ModelReady):
+            total = total + dp.total
+        elif isinstance(dp, ModelDownloading):
+            total = total + dp.download_progress.total
+    return total
+
+
+def check_storage_quota(
+    model_size: Memory,
+    config: StorageConfig,
+    downloads: Sequence[ModelStatus],
+) -> tuple[bool, str]:
+    if config.max_storage is None:
+        return True, ""
+
+    used = calculate_used_storage(downloads)
+    available = config.max_storage - used
+
+    if model_size <= available:
+        return True, ""
+
+    return (
+        False,
+        f"Need {model_size.in_gb:.1f} GiB, only {max(0, available.in_gb):.1f} GiB available within {config.max_storage.in_gb:.1f} GiB limit",
+    )
+
+
+def get_lru_eviction_candidates(
+    downloads: Sequence[ModelStatus],
+    model_last_used: Mapping[ModelId, datetime],
+    active_model_ids: frozenset[ModelId],
+) -> list[tuple[ModelId, ModelReady]]:
+    candidates: list[tuple[ModelId, ModelReady]] = []
+    for dp in downloads:
+        if not isinstance(dp, ModelReady):
+            continue
+        if dp.read_only:
+            continue
+        model_id = dp.shard_metadata.model_card.model_id
+        if model_id in active_model_ids:
+            continue
+        candidates.append((model_id, dp))
+
+    candidates.sort(
+        key=lambda item: model_last_used.get(item[0], datetime.min.replace(tzinfo=UTC))
+    )
+    return candidates
+
+
+def compute_evictions_needed(
+    model_size: Memory,
+    available: Memory,
+    candidates: list[tuple[ModelId, ModelReady]],
+) -> list[ModelId] | None:
+    if model_size <= available:
+        return []
+
+    space_needed = model_size - available
+    freed = Memory()
+    to_evict: list[ModelId] = []
+
+    for model_id, completed in candidates:
+        to_evict.append(model_id)
+        freed = freed + completed.total
+        if freed >= space_needed:
+            return to_evict
+
+    return None
+
+
+def decide_storage_action(
+    model_size: Memory,
+    config: StorageConfig,
+    downloads: Sequence[ModelStatus],
+    model_last_used: Mapping[ModelId, datetime],
+    active_model_ids: frozenset[ModelId],
+) -> StorageDecision:
+    """Pure decision function: given storage state, decide whether to allow, evict, or reject."""
+    if config.max_storage is None:
+        return StorageAllow()
+
+    allowed, reason = check_storage_quota(model_size, config, downloads)
+    if allowed:
+        return StorageAllow()
+
+    used = calculate_used_storage(downloads)
+    raw_available = config.max_storage - used
+    available = raw_available if raw_available.in_bytes >= 0 else Memory()
+
+    if config.storage_policy == "auto-evict":
+        candidates = get_lru_eviction_candidates(
+            downloads, model_last_used, active_model_ids
+        )
+        to_evict = compute_evictions_needed(model_size, available, candidates)
+
+        if to_evict is not None:
+            return StorageEvict(model_ids=to_evict)
+
+        return StorageReject(
+            reason="Cannot free enough space even after evicting all eligible models",
+            available=available,
+        )
+
+    return StorageReject(
+        reason=reason,
+        available=available,
+    )
+
+
+def get_download_rejected_events(
+    rejected_model_id: ModelId,
+    rejected_node_id: NodeId,
+    instances: Mapping[InstanceId, Instance],
+    tasks: Mapping[TaskId, Task],
+) -> list[TaskStatusUpdated | InstanceDeleted]:
+    """Pure function: compute events needed to clean up after a download rejection."""
+    events: list[TaskStatusUpdated | InstanceDeleted] = []
+    for instance_id, instance in instances.items():
+        if (
+            instance.shard_assignments.model_id == rejected_model_id
+            and rejected_node_id in instance.shard_assignments.node_to_runner
+        ):
+            for task in tasks.values():
+                if task.instance_id == instance_id and task.task_status in (
+                    TaskStatus.Pending,
+                    TaskStatus.Running,
+                ):
+                    events.append(
+                        TaskStatusUpdated(
+                            task_id=task.task_id,
+                            task_status=TaskStatus.Failed,
+                        )
+                    )
+            events.append(InstanceDeleted(instance_id=instance_id))
+    return events
+
+
+async def load_storage_config(
+    *,
+    max_storage_gb: float | None = None,
+    storage_policy: StoragePolicy | None = None,
+) -> StorageConfig:
+    """Load StorageConfig from config.toml, overlaying any CLI arg overrides."""
+    base = StorageConfig()
+    cfg_file = anyio.Path(EXO_CONFIG_FILE)
+    try:
+        await cfg_file.parent.mkdir(parents=True, exist_ok=True)
+        await cfg_file.touch(exist_ok=True)
+        raw = (await cfg_file.read_bytes()).decode("utf-8")
+        if raw.strip():
+            data = tomllib.loads(raw)
+            base = StorageConfig.from_disk(data)
+    except (OSError, tomllib.TOMLDecodeError, ValueError, KeyError):
+        logger.warning("Failed to read storage config from config file, using defaults")
+
+    resolved_max_storage = (
+        Memory.from_gb(max_storage_gb)
+        if max_storage_gb is not None
+        else base.max_storage
+    )
+    resolved_policy = (
+        storage_policy if storage_policy is not None else base.storage_policy
+    )
+    return StorageConfig(
+        max_storage=resolved_max_storage, storage_policy=resolved_policy
+    )
+
+
+async def persist_storage_config(config: StorageConfig) -> None:
+    """Persist StorageConfig to config.toml, preserving other config keys."""
+    cfg_path = anyio.Path(EXO_CONFIG_FILE)
+    await cfg_path.parent.mkdir(parents=True, exist_ok=True)
+
+    doc = tomlkit.document()
+    try:
+        raw = (await cfg_path.read_bytes()).decode("utf-8")
+        if raw.strip():
+            doc = tomlkit.parse(raw)
+    except (FileNotFoundError, TOMLKitError, UnicodeDecodeError):
+        pass
+
+    # Clear max_storage_gb so it doesn't linger when max_storage is None
+    doc.pop("max_storage_gb", None)  # pyright: ignore[reportUnknownMemberType]
+    doc.update(config.to_disk())  # pyright: ignore[reportUnknownMemberType]
+
+    await cfg_path.write_text(tomlkit.dumps(doc))  # pyright: ignore[reportUnknownMemberType]
+    logger.debug(f"Persisted storage config to {cfg_path}")

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

@@ -4,14 +4,14 @@ from exo.shared.types.common import NodeId
 from exo.shared.types.events import NodeDownloadProgress
 from exo.shared.types.memory import Memory
 from exo.shared.types.state import State
-from exo.shared.types.worker.downloads import DownloadCompleted
+from exo.shared.types.worker.downloads import ModelReady
 from exo.worker.tests.constants import MODEL_A_ID, MODEL_B_ID
 
 
 def test_apply_node_download_progress():
     state = State()
     shard1 = get_pipeline_shard_metadata(MODEL_A_ID, device_rank=0, world_size=2)
-    event = DownloadCompleted(
+    event = ModelReady(
         node_id=NodeId("node-1"),
         shard_metadata=shard1,
         total=Memory(),
@@ -27,12 +27,12 @@ def test_apply_node_download_progress():
 def test_apply_two_node_download_progress():
     shard1 = get_pipeline_shard_metadata(MODEL_A_ID, device_rank=0, world_size=2)
     shard2 = get_pipeline_shard_metadata(MODEL_B_ID, device_rank=0, world_size=2)
-    event1 = DownloadCompleted(
+    event1 = ModelReady(
         node_id=NodeId("node-1"),
         shard_metadata=shard1,
         total=Memory(),
     )
-    event2 = DownloadCompleted(
+    event2 = ModelReady(
         node_id=NodeId("node-1"),
         shard_metadata=shard2,
         total=Memory(),

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

@@ -0,0 +1,45 @@
+from exo.shared.apply import apply_node_timed_out, apply_storage_config_updated
+from exo.shared.types.common import NodeId
+from exo.shared.types.events import NodeTimedOut, StorageConfigUpdated
+from exo.shared.types.memory import Memory
+from exo.shared.types.state import State
+from exo.shared.types.storage import StorageConfig
+
+NODE_A = NodeId("node-a")
+NODE_B = NodeId("node-b")
+
+
+def test_storage_config_updated_adds_config() -> None:
+    state = State()
+    config = StorageConfig(max_storage=Memory.from_gb(10), storage_policy="manual")
+    event = StorageConfigUpdated(node_id=NODE_A, storage_config=config)
+
+    new_state = apply_storage_config_updated(event, state)
+
+    assert NODE_A in new_state.node_storage_config
+    assert new_state.node_storage_config[NODE_A].max_storage == Memory.from_gb(10)
+    assert new_state.node_storage_config[NODE_A].storage_policy == "manual"
+
+
+def test_storage_config_updated_overwrites_existing() -> None:
+    config1 = StorageConfig(max_storage=Memory.from_gb(10), storage_policy="manual")
+    state = State(node_storage_config={NODE_A: config1})
+
+    config2 = StorageConfig(max_storage=Memory.from_gb(20), storage_policy="auto-evict")
+    event = StorageConfigUpdated(node_id=NODE_A, storage_config=config2)
+
+    new_state = apply_storage_config_updated(event, state)
+
+    assert new_state.node_storage_config[NODE_A].max_storage == Memory.from_gb(20)
+    assert new_state.node_storage_config[NODE_A].storage_policy == "auto-evict"
+
+
+def test_node_timed_out_cleans_up_storage_config() -> None:
+    config = StorageConfig(max_storage=Memory.from_gb(10))
+    state = State(node_storage_config={NODE_A: config, NODE_B: config})
+
+    event = NodeTimedOut(node_id=NODE_A)
+    new_state = apply_node_timed_out(event, state)
+
+    assert NODE_A not in new_state.node_storage_config
+    assert NODE_B in new_state.node_storage_config

src/exo/shared/tests/test_storage.py

@@ -0,0 +1,634 @@
+from datetime import UTC, datetime
+from pathlib import Path
+from unittest.mock import patch
+
+from exo.shared.models.model_cards import ModelId
+from exo.shared.storage import (
+    calculate_used_storage,
+    check_storage_quota,
+    compute_evictions_needed,
+    decide_storage_action,
+    get_download_rejected_events,
+    get_lru_eviction_candidates,
+    load_storage_config,
+    persist_storage_config,
+)
+from exo.shared.tests.conftest import get_pipeline_shard_metadata
+from exo.shared.types.common import NodeId
+from exo.shared.types.events import InstanceDeleted, TaskStatusUpdated
+from exo.shared.types.memory import Memory
+from exo.shared.types.storage import (
+    StorageAllow,
+    StorageConfig,
+    StorageEvict,
+    StorageReject,
+)
+from exo.shared.types.tasks import LoadModel, TaskId, TaskStatus
+from exo.shared.types.worker.downloads import (
+    DownloadProgressData,
+    ModelDownloading,
+    ModelNotDownloading,
+    ModelReady,
+)
+from exo.shared.types.worker.instances import InstanceId, MlxRingInstance
+from exo.shared.types.worker.runners import RunnerId, ShardAssignments
+
+MODEL_A = ModelId("org/model-a")
+MODEL_B = ModelId("org/model-b")
+MODEL_C = ModelId("org/model-c")
+MODEL_D = ModelId("org/model-d")
+NODE_ID = "node-1"
+
+
+def _completed(
+    model_id: ModelId, size_gb: float, read_only: bool = False
+) -> ModelReady:
+    shard = get_pipeline_shard_metadata(model_id, device_rank=0)
+    return ModelReady(
+        node_id=NODE_ID,  # type: ignore[arg-type]
+        shard_metadata=shard,
+        total=Memory.from_gb(size_gb),
+        read_only=read_only,
+    )
+
+
+class TestCheckStorageQuota:
+    def test_unlimited_allows(self) -> None:
+        config = StorageConfig(max_storage=None)
+        allowed, _reason = check_storage_quota(Memory.from_gb(10), config, [])
+        assert allowed is True
+        assert _reason == ""
+
+    def test_under_limit_allows(self) -> None:
+        config = StorageConfig(max_storage=Memory.from_gb(20))
+        downloads = [_completed(MODEL_A, 5)]
+        allowed, _reason = check_storage_quota(Memory.from_gb(10), config, downloads)
+        assert allowed is True
+
+    def test_over_limit_rejects(self) -> None:
+        config = StorageConfig(max_storage=Memory.from_gb(10))
+        downloads = [_completed(MODEL_A, 5)]
+        allowed, reason = check_storage_quota(Memory.from_gb(8), config, downloads)
+        assert allowed is False
+        assert "Need" in reason
+        assert "available" in reason
+
+    def test_exact_fit_allows(self) -> None:
+        config = StorageConfig(max_storage=Memory.from_gb(10))
+        downloads = [_completed(MODEL_A, 5)]
+        allowed, _ = check_storage_quota(Memory.from_gb(5), config, downloads)
+        assert allowed is True
+
+
+class TestGetLruEvictionCandidates:
+    def test_excludes_active_models(self) -> None:
+        downloads = [_completed(MODEL_A, 5), _completed(MODEL_B, 3)]
+        last_used = {
+            MODEL_A: datetime(2024, 1, 1, tzinfo=UTC),
+            MODEL_B: datetime(2024, 1, 2, tzinfo=UTC),
+        }
+        candidates = get_lru_eviction_candidates(
+            downloads, last_used, frozenset({MODEL_A})
+        )
+        assert len(candidates) == 1
+        assert candidates[0][0] == MODEL_B
+
+    def test_excludes_read_only(self) -> None:
+        downloads = [_completed(MODEL_A, 5, read_only=True), _completed(MODEL_B, 3)]
+        candidates = get_lru_eviction_candidates(downloads, {}, frozenset())
+        assert len(candidates) == 1
+        assert candidates[0][0] == MODEL_B
+
+    def test_sorts_oldest_first(self) -> None:
+        downloads = [_completed(MODEL_A, 5), _completed(MODEL_B, 3)]
+        last_used = {
+            MODEL_A: datetime(2024, 6, 1, tzinfo=UTC),
+            MODEL_B: datetime(2024, 1, 1, tzinfo=UTC),
+        }
+        candidates = get_lru_eviction_candidates(downloads, last_used, frozenset())
+        assert candidates[0][0] == MODEL_B
+        assert candidates[1][0] == MODEL_A
+
+    def test_models_without_usage_get_min(self) -> None:
+        downloads = [_completed(MODEL_A, 5), _completed(MODEL_B, 3)]
+        last_used = {MODEL_A: datetime(2024, 6, 1, tzinfo=UTC)}
+        candidates = get_lru_eviction_candidates(downloads, last_used, frozenset())
+        assert candidates[0][0] == MODEL_B  # no usage -> datetime.min
+
+
+class TestComputeEvictionsNeeded:
+    def test_sufficient_candidates(self) -> None:
+        candidates = [
+            (MODEL_A, _completed(MODEL_A, 5)),
+            (MODEL_B, _completed(MODEL_B, 3)),
+        ]
+        result = compute_evictions_needed(
+            Memory.from_gb(6), Memory.from_gb(2), candidates
+        )
+        assert result is not None
+        assert MODEL_A in result
+
+    def test_insufficient_candidates(self) -> None:
+        candidates = [
+            (MODEL_A, _completed(MODEL_A, 2)),
+        ]
+        result = compute_evictions_needed(
+            Memory.from_gb(10), Memory.from_gb(2), candidates
+        )
+        assert result is None
+
+    def test_no_eviction_needed(self) -> None:
+        candidates = [(MODEL_A, _completed(MODEL_A, 5))]
+        result = compute_evictions_needed(
+            Memory.from_gb(3), Memory.from_gb(5), candidates
+        )
+        assert result == []
+
+    def test_evicts_in_lru_order(self) -> None:
+        """Eviction picks candidates in the order given (oldest first from LRU sort)."""
+        candidates = [
+            (MODEL_A, _completed(MODEL_A, 2)),  # oldest
+            (MODEL_B, _completed(MODEL_B, 2)),  # newer
+            (MODEL_C, _completed(MODEL_C, 2)),  # newest
+        ]
+        result = compute_evictions_needed(
+            Memory.from_gb(3), Memory.from_gb(1), candidates
+        )
+        assert result == [MODEL_A]
+
+    def test_evicts_multiple_until_enough_space(self) -> None:
+        """When one model isn't enough, evicts multiple in LRU order."""
+        candidates = [
+            (MODEL_A, _completed(MODEL_A, 1)),
+            (MODEL_B, _completed(MODEL_B, 1)),
+            (MODEL_C, _completed(MODEL_C, 1)),
+        ]
+        # Need 4 GiB, have 1 GiB available — need 3 GiB freed
+        result = compute_evictions_needed(
+            Memory.from_gb(4), Memory.from_gb(1), candidates
+        )
+        assert result == [MODEL_A, MODEL_B, MODEL_C]
+
+    def test_evicts_minimum_needed(self) -> None:
+        """Stops evicting as soon as enough space is freed."""
+        candidates = [
+            (MODEL_A, _completed(MODEL_A, 3)),
+            (MODEL_B, _completed(MODEL_B, 3)),
+        ]
+        # Need 5 GiB, have 2 GiB — need 3 GiB freed. Model A alone suffices.
+        result = compute_evictions_needed(
+            Memory.from_gb(5), Memory.from_gb(2), candidates
+        )
+        assert result == [MODEL_A]
+
+
+class TestCalculateUsedStorage:
+    def test_only_counts_completed_and_ongoing(self) -> None:
+        """Pending and rejected downloads should not count toward used storage."""
+        shard_a = get_pipeline_shard_metadata(MODEL_A, device_rank=0)
+        shard_b = get_pipeline_shard_metadata(MODEL_B, device_rank=0)
+        downloads = [
+            _completed(MODEL_A, 5),
+            ModelNotDownloading(
+                node_id=NODE_ID,  # type: ignore[arg-type]
+                shard_metadata=shard_a,
+            ),
+            ModelDownloading(
+                node_id=NODE_ID,  # type: ignore[arg-type]
+                shard_metadata=shard_b,
+                download_progress=DownloadProgressData(
+                    total=Memory.from_gb(10),
+                    downloaded=Memory.from_gb(3),
+                    downloaded_this_session=Memory.from_gb(3),
+                    completed_files=1,
+                    total_files=5,
+                    speed=0,
+                    eta_ms=0,
+                    files={},
+                ),
+            ),
+        ]
+        used = calculate_used_storage(downloads)
+        # 5 GiB completed + 10 GiB ongoing total = 15 GiB
+        assert abs(used.in_gb - 15.0) < 0.01
+
+    def test_empty_downloads(self) -> None:
+        assert calculate_used_storage([]).in_bytes == 0
+
+
+class TestGetLruEvictionCandidatesExtended:
+    def test_excludes_non_completed_downloads(self) -> None:
+        """Only DownloadCompleted entries are eviction candidates."""
+        shard_a = get_pipeline_shard_metadata(MODEL_A, device_rank=0)
+        downloads = [
+            _completed(MODEL_B, 3),
+            ModelNotDownloading(
+                node_id=NODE_ID,  # type: ignore[arg-type]
+                shard_metadata=shard_a,
+            ),
+        ]
+        candidates = get_lru_eviction_candidates(downloads, {}, frozenset())
+        assert len(candidates) == 1
+        assert candidates[0][0] == MODEL_B
+
+    def test_all_active_returns_empty(self) -> None:
+        """When all completed models are active, no candidates available."""
+        downloads = [_completed(MODEL_A, 5), _completed(MODEL_B, 3)]
+        candidates = get_lru_eviction_candidates(
+            downloads, {}, frozenset({MODEL_A, MODEL_B})
+        )
+        assert candidates == []
+
+    def test_three_models_lru_order(self) -> None:
+        """Three models sorted correctly: oldest used first."""
+        downloads = [
+            _completed(MODEL_A, 2),
+            _completed(MODEL_B, 3),
+            _completed(MODEL_C, 1),
+        ]
+        last_used = {
+            MODEL_A: datetime(2024, 3, 1, tzinfo=UTC),
+            MODEL_B: datetime(2024, 1, 1, tzinfo=UTC),
+            MODEL_C: datetime(2024, 6, 1, tzinfo=UTC),
+        }
+        candidates = get_lru_eviction_candidates(downloads, last_used, frozenset())
+        assert [c[0] for c in candidates] == [MODEL_B, MODEL_A, MODEL_C]
+
+    def test_mixed_active_readonly_and_regular(self) -> None:
+        """Only non-active, non-read-only completed models are candidates."""
+        downloads = [
+            _completed(MODEL_A, 5, read_only=True),  # excluded: read-only
+            _completed(MODEL_B, 3),  # excluded: active
+            _completed(MODEL_C, 2),  # candidate
+            _completed(MODEL_D, 1),  # candidate
+        ]
+        last_used = {
+            MODEL_C: datetime(2024, 6, 1, tzinfo=UTC),
+            MODEL_D: datetime(2024, 1, 1, tzinfo=UTC),
+        }
+        candidates = get_lru_eviction_candidates(
+            downloads, last_used, frozenset({MODEL_B})
+        )
+        assert [c[0] for c in candidates] == [MODEL_D, MODEL_C]
+
+
+class TestEndToEndEvictionScenario:
+    """Tests that combine LRU candidate selection with eviction computation."""
+
+    def test_evicts_oldest_model_to_fit_new_one(self) -> None:
+        """10 GiB limit, 3 completed models totaling 9 GiB,
+        need 3 GiB for new model — should evict the oldest."""
+        downloads = [
+            _completed(MODEL_A, 3),  # oldest used
+            _completed(MODEL_B, 3),
+            _completed(MODEL_C, 3),  # newest used
+        ]
+        last_used = {
+            MODEL_A: datetime(2024, 1, 1, tzinfo=UTC),
+            MODEL_B: datetime(2024, 6, 1, tzinfo=UTC),
+            MODEL_C: datetime(2024, 12, 1, tzinfo=UTC),
+        }
+        config = StorageConfig(max_storage=Memory.from_gb(10))
+        new_model_size = Memory.from_gb(3)
+
+        # Step 1: quota check fails
+        allowed, _ = check_storage_quota(new_model_size, config, downloads)
+        assert not allowed
+
+        # Step 2: get candidates in LRU order
+        candidates = get_lru_eviction_candidates(downloads, last_used, frozenset())
+        assert candidates[0][0] == MODEL_A  # oldest
+
+        # Step 3: compute what to evict
+        used = calculate_used_storage(downloads)
+        assert config.max_storage is not None
+        available = config.max_storage - used
+        to_evict = compute_evictions_needed(new_model_size, available, candidates)
+        assert to_evict == [MODEL_A]
+
+    def test_protects_currently_active_model(self) -> None:
+        """Active model should not be evicted even if it's the oldest."""
+        downloads = [
+            _completed(MODEL_A, 4),  # oldest but active
+            _completed(MODEL_B, 4),  # next oldest, evictable
+        ]
+        last_used = {
+            MODEL_A: datetime(2024, 1, 1, tzinfo=UTC),
+            MODEL_B: datetime(2024, 6, 1, tzinfo=UTC),
+        }
+        config = StorageConfig(max_storage=Memory.from_gb(10))
+        new_model_size = Memory.from_gb(5)
+
+        allowed, _ = check_storage_quota(new_model_size, config, downloads)
+        assert not allowed
+
+        # MODEL_A is active — should be excluded
+        candidates = get_lru_eviction_candidates(
+            downloads, last_used, frozenset({MODEL_A})
+        )
+        assert len(candidates) == 1
+        assert candidates[0][0] == MODEL_B
+
+        used = calculate_used_storage(downloads)
+        assert config.max_storage is not None
+        available = config.max_storage - used
+        to_evict = compute_evictions_needed(new_model_size, available, candidates)
+        assert to_evict == [MODEL_B]
+
+    def test_cannot_evict_enough_returns_none(self) -> None:
+        """When all evictable space isn't enough, returns None."""
+        downloads = [
+            _completed(MODEL_A, 4, read_only=True),  # can't evict
+            _completed(MODEL_B, 2),  # can evict but only 2 GiB
+        ]
+        config = StorageConfig(max_storage=Memory.from_gb(10))
+        new_model_size = Memory.from_gb(8)
+
+        candidates = get_lru_eviction_candidates(downloads, {}, frozenset())
+        used = calculate_used_storage(downloads)
+        assert config.max_storage is not None
+        available = config.max_storage - used
+        to_evict = compute_evictions_needed(new_model_size, available, candidates)
+        assert to_evict is None
+
+
+class TestDecideStorageAction:
+    """Tests for the decide_storage_action pure function."""
+
+    def test_unlimited_allows(self) -> None:
+        config = StorageConfig(max_storage=None)
+        action = decide_storage_action(Memory.from_gb(10), config, [], {}, frozenset())
+        assert isinstance(action, StorageAllow)
+
+    def test_under_limit_allows(self) -> None:
+        config = StorageConfig(max_storage=Memory.from_gb(20))
+        downloads = [_completed(MODEL_A, 5)]
+        action = decide_storage_action(
+            Memory.from_gb(10), config, downloads, {}, frozenset()
+        )
+        assert isinstance(action, StorageAllow)
+
+    def test_manual_policy_rejects(self) -> None:
+        config = StorageConfig(max_storage=Memory.from_gb(10), storage_policy="manual")
+        downloads = [_completed(MODEL_A, 5)]
+        action = decide_storage_action(
+            Memory.from_gb(8), config, downloads, {}, frozenset()
+        )
+        assert isinstance(action, StorageReject)
+        assert "Need" in action.reason
+
+    def test_auto_evict_returns_evict(self) -> None:
+        config = StorageConfig(
+            max_storage=Memory.from_gb(10), storage_policy="auto-evict"
+        )
+        downloads = [_completed(MODEL_A, 4), _completed(MODEL_B, 4)]
+        last_used = {
+            MODEL_A: datetime(2024, 1, 1, tzinfo=UTC),
+            MODEL_B: datetime(2024, 6, 1, tzinfo=UTC),
+        }
+        action = decide_storage_action(
+            Memory.from_gb(5), config, downloads, last_used, frozenset()
+        )
+        assert isinstance(action, StorageEvict)
+        assert MODEL_A in action.model_ids
+
+    def test_auto_evict_rejects_when_impossible(self) -> None:
+        config = StorageConfig(
+            max_storage=Memory.from_gb(10), storage_policy="auto-evict"
+        )
+        downloads = [_completed(MODEL_A, 2)]
+        action = decide_storage_action(
+            Memory.from_gb(20), config, downloads, {}, frozenset()
+        )
+        assert isinstance(action, StorageReject)
+        assert "Cannot free enough" in action.reason
+
+    def test_auto_evict_protects_active_models(self) -> None:
+        config = StorageConfig(
+            max_storage=Memory.from_gb(10), storage_policy="auto-evict"
+        )
+        downloads = [_completed(MODEL_A, 4), _completed(MODEL_B, 4)]
+        last_used = {
+            MODEL_A: datetime(2024, 1, 1, tzinfo=UTC),
+            MODEL_B: datetime(2024, 6, 1, tzinfo=UTC),
+        }
+        # MODEL_A is oldest but active — should evict MODEL_B instead
+        action = decide_storage_action(
+            Memory.from_gb(5), config, downloads, last_used, frozenset({MODEL_A})
+        )
+        assert isinstance(action, StorageEvict)
+        assert action.model_ids == [MODEL_B]
+
+    def test_auto_evict_all_active_rejects(self) -> None:
+        config = StorageConfig(
+            max_storage=Memory.from_gb(10), storage_policy="auto-evict"
+        )
+        downloads = [_completed(MODEL_A, 4), _completed(MODEL_B, 4)]
+        action = decide_storage_action(
+            Memory.from_gb(5),
+            config,
+            downloads,
+            {},
+            frozenset({MODEL_A, MODEL_B}),
+        )
+        assert isinstance(action, StorageReject)
+
+
+def _make_instance(
+    instance_id: InstanceId,
+    model_id: ModelId,
+    node_id: NodeId,
+) -> MlxRingInstance:
+    shard = get_pipeline_shard_metadata(model_id, device_rank=0)
+    runner_id = RunnerId()
+    return MlxRingInstance(
+        instance_id=instance_id,
+        shard_assignments=ShardAssignments(
+            model_id=model_id,
+            runner_to_shard={runner_id: shard},
+            node_to_runner={node_id: runner_id},
+        ),
+        hosts_by_node={node_id: []},
+        ephemeral_port=0,
+    )
+
+
+class TestGetDownloadRejectedEvents:
+    """Tests for the get_download_rejected_events pure function."""
+
+    def test_deletes_instance_for_rejected_model(self) -> None:
+        node_id = NodeId("node-1")
+        instance_id = InstanceId()
+        instance = _make_instance(instance_id, MODEL_A, node_id)
+        events = get_download_rejected_events(
+            MODEL_A, node_id, {instance_id: instance}, {}
+        )
+        assert len(events) == 1
+        assert isinstance(events[0], InstanceDeleted)
+        assert events[0].instance_id == instance_id
+
+    def test_fails_pending_tasks_before_deleting_instance(self) -> None:
+        node_id = NodeId("node-1")
+        instance_id = InstanceId()
+        instance = _make_instance(instance_id, MODEL_A, node_id)
+        task_id = TaskId()
+        task = LoadModel(
+            task_id=task_id,
+            instance_id=instance_id,
+            task_status=TaskStatus.Pending,
+        )
+        events = get_download_rejected_events(
+            MODEL_A, node_id, {instance_id: instance}, {task_id: task}
+        )
+        assert len(events) == 2
+        assert isinstance(events[0], TaskStatusUpdated)
+        assert events[0].task_status == TaskStatus.Failed
+        assert isinstance(events[1], InstanceDeleted)
+
+    def test_ignores_different_model(self) -> None:
+        node_id = NodeId("node-1")
+        instance_id = InstanceId()
+        instance = _make_instance(instance_id, MODEL_B, node_id)
+        events = get_download_rejected_events(
+            MODEL_A, node_id, {instance_id: instance}, {}
+        )
+        assert events == []
+
+    def test_ignores_different_node(self) -> None:
+        node_id = NodeId("node-1")
+        other_node = NodeId("node-2")
+        instance_id = InstanceId()
+        instance = _make_instance(instance_id, MODEL_A, other_node)
+        events = get_download_rejected_events(
+            MODEL_A, node_id, {instance_id: instance}, {}
+        )
+        assert events == []
+
+    def test_skips_completed_tasks(self) -> None:
+        node_id = NodeId("node-1")
+        instance_id = InstanceId()
+        instance = _make_instance(instance_id, MODEL_A, node_id)
+        task_id = TaskId()
+        task = LoadModel(
+            task_id=task_id,
+            instance_id=instance_id,
+            task_status=TaskStatus.Complete,
+        )
+        events = get_download_rejected_events(
+            MODEL_A, node_id, {instance_id: instance}, {task_id: task}
+        )
+        # Only InstanceDeleted, no TaskStatusUpdated for completed task
+        assert len(events) == 1
+        assert isinstance(events[0], InstanceDeleted)
+
+
+class TestPersistStorageConfig:
+    """Tests for persist_storage_config I/O."""
+
+    async def test_round_trip(self, tmp_path: Path) -> None:
+        cfg_file = tmp_path / "config.toml"
+        config = StorageConfig(
+            max_storage=Memory.from_gb(50), storage_policy="auto-evict"
+        )
+        with patch("exo.shared.storage.EXO_CONFIG_FILE", cfg_file):
+            await persist_storage_config(config)
+            loaded = await load_storage_config()
+
+        assert loaded.storage_policy == "auto-evict"
+        assert loaded.max_storage is not None
+        assert abs(loaded.max_storage.in_gb - 50.0) < 0.1
+
+    async def test_preserves_other_keys(self, tmp_path: Path) -> None:
+        cfg_file = tmp_path / "config.toml"
+        cfg_file.write_text('some_other_key = "hello"\n')
+        config = StorageConfig(max_storage=Memory.from_gb(10))
+        with patch("exo.shared.storage.EXO_CONFIG_FILE", cfg_file):
+            await persist_storage_config(config)
+
+        contents = cfg_file.read_text()
+        assert "some_other_key" in contents
+        assert "hello" in contents
+        assert "max_storage_gb" in contents
+
+    async def test_clears_max_storage_gb_when_unlimited(self, tmp_path: Path) -> None:
+        cfg_file = tmp_path / "config.toml"
+        cfg_file.write_text('max_storage_gb = 50\nstorage_policy = "auto-evict"\n')
+        config = StorageConfig(max_storage=None, storage_policy="manual")
+        with patch("exo.shared.storage.EXO_CONFIG_FILE", cfg_file):
+            await persist_storage_config(config)
+
+        contents = cfg_file.read_text()
+        assert "max_storage_gb" not in contents
+        assert "manual" in contents
+
+    async def test_creates_file_if_missing(self, tmp_path: Path) -> None:
+        cfg_file = tmp_path / "subdir" / "config.toml"
+        config = StorageConfig(max_storage=Memory.from_gb(25))
+        with patch("exo.shared.storage.EXO_CONFIG_FILE", cfg_file):
+            await persist_storage_config(config)
+
+        assert cfg_file.exists()
+        assert "max_storage_gb" in cfg_file.read_text()
+
+
+class TestLoadStorageConfig:
+    """Tests for load_storage_config I/O."""
+
+    async def test_defaults_when_empty_file(self, tmp_path: Path) -> None:
+        cfg_file = tmp_path / "config.toml"
+        cfg_file.write_text("")
+        with patch("exo.shared.storage.EXO_CONFIG_FILE", cfg_file):
+            config = await load_storage_config()
+
+        assert config.max_storage is None
+        assert config.storage_policy == "manual"
+
+    async def test_reads_from_file(self, tmp_path: Path) -> None:
+        cfg_file = tmp_path / "config.toml"
+        cfg_file.write_text('max_storage_gb = 30.0\nstorage_policy = "auto-evict"\n')
+        with patch("exo.shared.storage.EXO_CONFIG_FILE", cfg_file):
+            config = await load_storage_config()
+
+        assert config.storage_policy == "auto-evict"
+        assert config.max_storage is not None
+        assert abs(config.max_storage.in_gb - 30.0) < 0.1
+
+    async def test_cli_overrides_file(self, tmp_path: Path) -> None:
+        cfg_file = tmp_path / "config.toml"
+        cfg_file.write_text('max_storage_gb = 30.0\nstorage_policy = "manual"\n')
+        with patch("exo.shared.storage.EXO_CONFIG_FILE", cfg_file):
+            config = await load_storage_config(
+                max_storage_gb=100.0, storage_policy="auto-evict"
+            )
+
+        assert config.storage_policy == "auto-evict"
+        assert config.max_storage is not None
+        assert abs(config.max_storage.in_gb - 100.0) < 0.1
+
+    async def test_partial_cli_override(self, tmp_path: Path) -> None:
+        """CLI overrides only the fields provided, file values used for the rest."""
+        cfg_file = tmp_path / "config.toml"
+        cfg_file.write_text('max_storage_gb = 30.0\nstorage_policy = "auto-evict"\n')
+        with patch("exo.shared.storage.EXO_CONFIG_FILE", cfg_file):
+            config = await load_storage_config(max_storage_gb=50.0)
+
+        assert config.storage_policy == "auto-evict"  # from file
+        assert config.max_storage is not None
+        assert abs(config.max_storage.in_gb - 50.0) < 0.1  # from CLI
+
+    async def test_defaults_on_corrupt_file(self, tmp_path: Path) -> None:
+        cfg_file = tmp_path / "config.toml"
+        cfg_file.write_text("not valid toml {{{")
+        with patch("exo.shared.storage.EXO_CONFIG_FILE", cfg_file):
+            config = await load_storage_config()
+
+        assert config.max_storage is None
+        assert config.storage_policy == "manual"
+
+    async def test_creates_file_if_missing(self, tmp_path: Path) -> None:
+        cfg_file = tmp_path / "subdir" / "config.toml"
+        with patch("exo.shared.storage.EXO_CONFIG_FILE", cfg_file):
+            config = await load_storage_config()
+
+        assert config.max_storage is None
+        assert cfg_file.exists()

src/exo/shared/types/commands.py

@@ -7,6 +7,8 @@ from exo.api.types import (
 from exo.shared.models.model_cards import ModelCard, ModelId
 from exo.shared.types.chunks import InputImageChunk
 from exo.shared.types.common import CommandId, NodeId, SystemId
+from exo.shared.types.memory import Memory
+from exo.shared.types.storage import StoragePolicy
 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
@@ -92,6 +94,12 @@ class DeleteCustomModelCard(BaseCommand):
 DownloadCommand = StartDownload | DeleteDownload | CancelDownload
 
 
+class SetStorageConfig(BaseCommand):
+    target_node_id: NodeId
+    max_storage: Memory | None
+    storage_policy: StoragePolicy
+
+
 Command = (
     TestCommand
     | RequestEventLog
@@ -106,6 +114,7 @@ Command = (
     | SendInputChunk
     | AddCustomModelCard
     | DeleteCustomModelCard
+    | SetStorageConfig
 )
 
 

src/exo/shared/types/events.py

@@ -7,8 +7,9 @@ from exo.shared.models.model_cards import ModelCard
 from exo.shared.topology import Connection
 from exo.shared.types.chunks import GenerationChunk, InputImageChunk
 from exo.shared.types.common import CommandId, Id, ModelId, NodeId, SessionId, SystemId
+from exo.shared.types.storage import StorageConfig
 from exo.shared.types.tasks import Task, TaskId, TaskStatus
-from exo.shared.types.worker.downloads import DownloadProgress
+from exo.shared.types.worker.downloads import ModelStatus
 from exo.shared.types.worker.instances import Instance, InstanceId
 from exo.shared.types.worker.runners import RunnerId, RunnerStatus
 from exo.utils.info_gatherer.info_gatherer import GatheredInfo
@@ -86,7 +87,7 @@ class NodeGatheredInfo(BaseEvent):
 
 
 class NodeDownloadProgress(BaseEvent):
-    download_progress: DownloadProgress
+    download_progress: ModelStatus
 
 
 class ChunkGenerated(BaseEvent):
@@ -137,6 +138,12 @@ class TracesMerged(BaseEvent):
     traces: list[TraceEventData]
 
 
+@final
+class StorageConfigUpdated(BaseEvent):
+    node_id: NodeId
+    storage_config: StorageConfig
+
+
 Event = (
     TestEvent
     | TaskCreated
@@ -158,6 +165,7 @@ Event = (
     | TracesMerged
     | CustomModelCardAdded
     | CustomModelCardDeleted
+    | StorageConfigUpdated
 )
 
 

src/exo/shared/types/state.py

@@ -17,8 +17,9 @@ from exo.shared.types.profiling import (
     SystemPerformanceProfile,
     ThunderboltBridgeStatus,
 )
+from exo.shared.types.storage import StorageConfig
 from exo.shared.types.tasks import Task, TaskId
-from exo.shared.types.worker.downloads import DownloadProgress
+from exo.shared.types.worker.downloads import ModelStatus
 from exo.shared.types.worker.instances import Instance, InstanceId
 from exo.shared.types.worker.runners import RunnerId, RunnerStatus
 from exo.utils.pydantic_ext import CamelCaseModel
@@ -42,7 +43,7 @@ class State(CamelCaseModel):
     )
     instances: Mapping[InstanceId, Instance] = {}
     runners: Mapping[RunnerId, RunnerStatus] = {}
-    downloads: Mapping[NodeId, Sequence[DownloadProgress]] = {}
+    downloads: Mapping[NodeId, Sequence[ModelStatus]] = {}
     tasks: Mapping[TaskId, Task] = {}
     last_seen: Mapping[NodeId, datetime] = {}
     topology: Topology = Field(default_factory=Topology)
@@ -57,6 +58,7 @@ class State(CamelCaseModel):
     node_thunderbolt: Mapping[NodeId, NodeThunderboltInfo] = {}
     node_thunderbolt_bridge: Mapping[NodeId, ThunderboltBridgeStatus] = {}
     node_rdma_ctl: Mapping[NodeId, NodeRdmaCtlStatus] = {}
+    node_storage_config: Mapping[NodeId, StorageConfig] = {}
 
     # Detected cycles where all nodes have Thunderbolt bridge enabled (>2 nodes)
     thunderbolt_bridge_cycles: Sequence[Sequence[NodeId]] = []

src/exo/shared/types/storage.py

@@ -0,0 +1,52 @@
+from typing import Any, Literal, Self, final
+
+from exo.shared.models.model_cards import ModelId
+from exo.shared.types.memory import Memory
+from exo.utils.pydantic_ext import FrozenModel
+
+StoragePolicy = Literal["manual", "auto-evict"]
+
+
+@final
+class StorageConfig(FrozenModel):
+    max_storage: Memory | None = None
+    storage_policy: StoragePolicy = "manual"
+
+    @classmethod
+    def from_disk(cls, data: dict[str, Any]) -> Self:
+        """Parse from a TOML config dict (e.g. from tomllib)."""
+        max_storage: Memory | None = None
+        if "max_storage_gb" in data:
+            gb = float(data["max_storage_gb"])  # pyright: ignore[reportAny]
+            if gb < 0:
+                raise ValueError(f"max_storage_gb must be non-negative, got {gb}")
+            max_storage = Memory.from_gb(gb)
+        policy: StoragePolicy = data.get("storage_policy", "manual")  # pyright: ignore[reportAny]
+        return cls(max_storage=max_storage, storage_policy=policy)
+
+    def to_disk(self) -> dict[str, Any]:
+        """Serialize to a dict suitable for writing to TOML."""
+        result: dict[str, Any] = {}
+        if self.max_storage is not None:
+            result["max_storage_gb"] = round(self.max_storage.in_gb, 2)
+        result["storage_policy"] = self.storage_policy
+        return result
+
+
+@final
+class StorageAllow(FrozenModel):
+    pass
+
+
+@final
+class StorageEvict(FrozenModel):
+    model_ids: list[ModelId]
+
+
+@final
+class StorageReject(FrozenModel):
+    reason: str
+    available: Memory
+
+
+StorageDecision = StorageAllow | StorageEvict | StorageReject

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

@@ -23,32 +23,43 @@ class DownloadProgressData(CamelCaseModel):
     files: dict[str, "DownloadProgressData"]
 
 
-class BaseDownloadProgress(TaggedModel):
+class BaseModelStatus(TaggedModel):
     node_id: NodeId
     shard_metadata: ShardMetadata
     model_directory: str = ""
 
 
-class DownloadPending(BaseDownloadProgress):
+class ModelNotDownloading(BaseModelStatus):
     downloaded: Memory = Memory()
     total: Memory = Memory()
 
 
-class DownloadCompleted(BaseDownloadProgress):
+class ModelReady(BaseModelStatus):
     total: Memory
     read_only: bool = False
 
 
-class DownloadFailed(BaseDownloadProgress):
+class ModelDownloadFailed(BaseModelStatus):
     error_message: str
 
 
-class DownloadOngoing(BaseDownloadProgress):
+class ModelDownloading(BaseModelStatus):
     download_progress: DownloadProgressData
 
 
-DownloadProgress = (
-    DownloadPending | DownloadCompleted | DownloadFailed | DownloadOngoing
+class ModelRejected(BaseModelStatus):
+    reason: str
+    required: Memory
+    available: Memory
+    limit: Memory
+
+
+ModelStatus = (
+    ModelNotDownloading
+    | ModelReady
+    | ModelDownloadFailed
+    | ModelDownloading
+    | ModelRejected
 )
 
 

src/exo/utils/info_gatherer/info_gatherer.py

@@ -21,6 +21,7 @@ from exo.shared.types.profiling import (
     NetworkInterfaceInfo,
     ThunderboltBridgeStatus,
 )
+from exo.shared.types.storage import StorageConfig
 from exo.shared.types.thunderbolt import (
     ThunderboltConnection,
     ThunderboltConnectivity,
@@ -294,6 +295,8 @@ class ThunderboltBridgeInfo(TaggedModel):
 class NodeConfig(TaggedModel):
     """Node configuration from EXO_CONFIG_FILE, reloaded from the file only at startup. Other changes should come in through the API and propagate from there"""
 
+    storage_config: StorageConfig = StorageConfig()
+
     @classmethod
     async def gather(cls) -> Self | None:
         cfg_file = anyio.Path(EXO_CONFIG_FILE)
@@ -303,7 +306,7 @@ class NodeConfig(TaggedModel):
             try:
                 contents = (await f.read()).decode("utf-8")
                 data = tomllib.loads(contents)
-                return cls.model_validate(data)
+                return cls(storage_config=StorageConfig.from_disk(data))
             except (tomllib.TOMLDecodeError, UnicodeDecodeError, ValidationError):
                 logger.warning("Invalid config file, skipping...")
                 return None

src/exo/worker/main.py

@@ -43,7 +43,7 @@ from exo.shared.types.tasks import (
     TextGeneration,
 )
 from exo.shared.types.topology import Connection, SocketConnection
-from exo.shared.types.worker.downloads import DownloadCompleted
+from exo.shared.types.worker.downloads import ModelReady
 from exo.shared.types.worker.runners import RunnerId
 from exo.utils.channels import Receiver, Sender, channel
 from exo.utils.info_gatherer.info_gatherer import GatheredInfo, InfoGatherer
@@ -191,7 +191,7 @@ class Worker:
                         logger.info(f"Model {model_id} found at {found_path}")
                         await self.event_sender.send(
                             NodeDownloadProgress(
-                                download_progress=DownloadCompleted(
+                                download_progress=ModelReady(
                                     node_id=self.node_id,
                                     shard_metadata=shard,
                                     model_directory=str(found_path),

src/exo/worker/plan.py

@@ -20,10 +20,10 @@ from exo.shared.types.tasks import (
     TextGeneration,
 )
 from exo.shared.types.worker.downloads import (
-    DownloadCompleted,
-    DownloadFailed,
-    DownloadOngoing,
-    DownloadProgress,
+    ModelDownloadFailed,
+    ModelDownloading,
+    ModelReady,
+    ModelStatus,
 )
 from exo.shared.types.worker.instances import BoundInstance, Instance, InstanceId
 from exo.shared.types.worker.runners import (
@@ -46,7 +46,7 @@ def plan(
     node_id: NodeId,
     # Runners is expected to be FRESH and so should not come from state
     runners: Mapping[RunnerId, RunnerSupervisor],
-    global_download_status: Mapping[NodeId, Sequence[DownloadProgress]],
+    global_download_status: Mapping[NodeId, Sequence[ModelStatus]],
     instances: Mapping[InstanceId, Instance],
     all_runners: Mapping[RunnerId, RunnerStatus],  # all global
     tasks: Mapping[TaskId, Task],
@@ -115,7 +115,7 @@ def _create_runner(
 def _model_needs_download(
     node_id: NodeId,
     runners: Mapping[RunnerId, RunnerSupervisor],
-    global_download_status: Mapping[NodeId, Sequence[DownloadProgress]],
+    global_download_status: Mapping[NodeId, Sequence[ModelStatus]],
 ) -> DownloadModel | None:
     local_downloads = global_download_status.get(node_id, [])
     download_status = {
@@ -128,7 +128,7 @@ def _model_needs_download(
             model_id not in download_status
             or not isinstance(
                 download_status[model_id],
-                (DownloadOngoing, DownloadCompleted, DownloadFailed),
+                (ModelDownloading, ModelReady, ModelDownloadFailed),
             )
         ):
             # We don't invalidate download_status randomly in case a file gets deleted on disk
@@ -191,7 +191,7 @@ def _init_distributed_backend(
 def _load_model(
     runners: Mapping[RunnerId, RunnerSupervisor],
     all_runners: Mapping[RunnerId, RunnerStatus],
-    global_download_status: Mapping[NodeId, Sequence[DownloadProgress]],
+    global_download_status: Mapping[NodeId, Sequence[ModelStatus]],
 ) -> LoadModel | None:
     for runner in runners.values():
         instance = runner.bound_instance.instance
@@ -200,7 +200,7 @@ def _load_model(
         all_local_downloads_complete = all(
             nid in global_download_status
             and any(
-                isinstance(dp, DownloadCompleted)
+                isinstance(dp, ModelReady)
                 and dp.shard_metadata.model_card.model_id == shard_assignments.model_id
                 for dp in global_download_status[nid]
             )

src/exo/worker/tests/unittests/test_plan/test_download_and_loading.py

@@ -2,7 +2,7 @@ import exo.worker.plan as plan_mod
 from exo.shared.types.common import NodeId
 from exo.shared.types.memory import Memory
 from exo.shared.types.tasks import LoadModel
-from exo.shared.types.worker.downloads import DownloadCompleted, DownloadProgress
+from exo.shared.types.worker.downloads import ModelReady, ModelStatus
 from exo.shared.types.worker.instances import BoundInstance
 from exo.shared.types.worker.runners import (
     RunnerConnected,
@@ -89,12 +89,8 @@ def test_plan_loads_model_when_all_shards_downloaded_and_waiting():
     }
 
     global_download_status = {
-        NODE_A: [
-            DownloadCompleted(shard_metadata=shard1, node_id=NODE_A, total=Memory())
-        ],
-        NODE_B: [
-            DownloadCompleted(shard_metadata=shard2, node_id=NODE_B, total=Memory())
-        ],
+        NODE_A: [ModelReady(shard_metadata=shard1, node_id=NODE_A, total=Memory())],
+        NODE_B: [ModelReady(shard_metadata=shard2, node_id=NODE_B, total=Memory())],
     }
 
     result = plan_mod.plan(
@@ -132,10 +128,8 @@ def test_plan_does_not_request_download_when_shard_already_downloaded():
     all_runners = {RUNNER_1_ID: RunnerIdle()}
 
     # Global state shows shard is downloaded for NODE_A
-    global_download_status: dict[NodeId, list[DownloadProgress]] = {
-        NODE_A: [
-            DownloadCompleted(shard_metadata=shard, node_id=NODE_A, total=Memory())
-        ],
+    global_download_status: dict[NodeId, list[ModelStatus]] = {
+        NODE_A: [ModelReady(shard_metadata=shard, node_id=NODE_A, total=Memory())],
         NODE_B: [],
     }
 
@@ -180,9 +174,7 @@ def test_plan_does_not_load_model_until_all_shards_downloaded_globally():
     }
 
     global_download_status = {
-        NODE_A: [
-            DownloadCompleted(shard_metadata=shard1, node_id=NODE_A, total=Memory())
-        ],
+        NODE_A: [ModelReady(shard_metadata=shard1, node_id=NODE_A, total=Memory())],
         NODE_B: [],  # NODE_B has no downloads completed yet
     }
 
@@ -198,11 +190,9 @@ def test_plan_does_not_load_model_until_all_shards_downloaded_globally():
     assert result is None
 
     global_download_status = {
-        NODE_A: [
-            DownloadCompleted(shard_metadata=shard1, node_id=NODE_A, total=Memory())
-        ],
+        NODE_A: [ModelReady(shard_metadata=shard1, node_id=NODE_A, total=Memory())],
         NODE_B: [
-            DownloadCompleted(shard_metadata=shard2, node_id=NODE_B, total=Memory())
+            ModelReady(shard_metadata=shard2, node_id=NODE_B, total=Memory())
         ],  # NODE_B has no downloads completed yet
     }
 

tests/get_all_models_on_cluster.py

@@ -19,7 +19,7 @@ with urlopen(f"http://{ip}:52415/state", timeout=5) as r:
 
 def mid(x: dict[str, Any]) -> str | None:
     for k in (
-        "DownloadCompleted",
+        "ModelReady",
         "shardMetadata",
         "PipelineShardMetadata",
         "modelCard",