fix(mount-io): bound token-harvester gcsfuse reads behind a MountIO guard

.env.example

@@ -15,3 +15,22 @@ MEDIA_ROOT=~/.dispatch/media
 # TLS (optional — both must be set to enable HTTPS)
 # TLS_CERT=~/.dispatch/tls/cert.pem
 # TLS_KEY=~/.dispatch/tls/key.pem
+
+# gcsfuse mount-io guard (token harvester). Defaults shown.
+# TIMEOUT_MS is an idle (no-progress) deadline for streaming reads and a
+# whole-op deadline for readdir. Keep it generous enough to exceed worst-case
+# large-file / cold-cache latency on your mount — a too-tight value makes a
+# healthy-but-slow mount trip the breaker and pause all harvesting for the
+# cooldown. Raise it on high-latency mounts.
+# DISPATCH_MOUNT_IO_TIMEOUT_MS=10000
+# DISPATCH_MOUNT_IO_CONCURRENCY=2
+# DISPATCH_MOUNT_IO_BREAKER_THRESHOLD=3
+# DISPATCH_MOUNT_IO_BREAKER_COOLDOWN_MS=60000
+
+# I/O thread-pool size. A stalled gcsfuse mount leaves uncancellable read/readdir
+# syscalls blocked on pool threads; enlarging the pool keeps a few wedged
+# harvester threads from starving other fs/dns/crypto work. Set before launch
+# (the launchd wrapper defaults this to 16). NOTE: the production Bun runtime
+# uses its own threadpool, not libuv, and may ignore this — it mainly hardens
+# Node dev/test; the mount-io breaker is the cross-runtime backstop.
+# UV_THREADPOOL_SIZE=16

apps/server/src/agents/token-harvester.ts

@@ -6,6 +6,7 @@ import { createInterface } from "node:readline";
 
 import type { Pool } from "pg";
 
+import { mountIO } from "../shared/mount-io/index.js";
 import type { AgentRecord } from "./manager.js";
 
 type ModelTokenTotals = {
@@ -65,7 +66,10 @@ export function cwdToClaudeProjectDir(cwd: string): string {
 export async function discoverSessionFiles(dir: string): Promise<string[]> {
   let entries: string[];
   try {
-    entries = await readdir(dir);
+    // @types/node@24 readdir has no {signal} overload; bounded by MountIO timeout only
+    entries = await mountIO.run("readdir:claude-projects", (_signal) =>
+      readdir(dir)
+    );
   } catch {
     return [];
   }
@@ -78,57 +82,62 @@ async function parseClaudeSessionTokenUsage(
   filePath: string
 ): Promise<SessionTokenSummary> {
   const sessionId = path.basename(filePath, ".jsonl");
-  const totals = new Map<string, ModelTokenTotals>();
-  let sessionStart: string | null = null;
-  let sessionEnd: string | null = null;
+  return mountIO.run(`read:claude:${sessionId}`, async (signal, heartbeat) => {
+    const totals = new Map<string, ModelTokenTotals>();
+    let sessionStart: string | null = null;
+    let sessionEnd: string | null = null;
+
+    const rl = createInterface({
+      input: createReadStream(filePath, { encoding: "utf-8", signal }),
+      crlfDelay: Infinity,
+    });
+
+    for await (const line of rl) {
+      // Progress made — re-arm the idle deadline so a slow-but-streaming read
+      // is not mistaken for a stall.
+      heartbeat();
+      if (!line.trim()) continue;
+      let entry: Record<string, unknown>;
+      try {
+        entry = JSON.parse(line);
+      } catch {
+        continue;
+      }
 
-  const rl = createInterface({
-    input: createReadStream(filePath, { encoding: "utf-8" }),
-    crlfDelay: Infinity,
-  });
+      if (entry.type !== "assistant") continue;
+      const message = entry.message as Record<string, unknown> | undefined;
+      if (!message?.usage) continue;
 
-  for await (const line of rl) {
-    if (!line.trim()) continue;
-    let entry: Record<string, unknown>;
-    try {
-      entry = JSON.parse(line);
-    } catch {
-      continue;
-    }
-
-    if (entry.type !== "assistant") continue;
-    const message = entry.message as Record<string, unknown> | undefined;
-    if (!message?.usage) continue;
+      const usage = message.usage as Record<string, number>;
+      const model = (message.model as string) ?? "unknown";
+      const ts = entry.timestamp as string | undefined;
 
-    const usage = message.usage as Record<string, number>;
-    const model = (message.model as string) ?? "unknown";
-    const ts = entry.timestamp as string | undefined;
+      if (ts) {
+        if (!sessionStart) sessionStart = ts;
+        sessionEnd = ts;
+      }
 
-    if (ts) {
-      if (!sessionStart) sessionStart = ts;
-      sessionEnd = ts;
-    }
+      let bucket = totals.get(model);
+      if (!bucket) {
+        bucket = {
+          inputTokens: 0,
+          cacheCreationTokens: 0,
+          cacheReadTokens: 0,
+          outputTokens: 0,
+          messageCount: 0,
+        };
+        totals.set(model, bucket);
+      }
 
-    let bucket = totals.get(model);
-    if (!bucket) {
-      bucket = {
-        inputTokens: 0,
-        cacheCreationTokens: 0,
-        cacheReadTokens: 0,
-        outputTokens: 0,
-        messageCount: 0,
-      };
-      totals.set(model, bucket);
+      bucket.inputTokens += usage.input_tokens ?? 0;
+      bucket.cacheCreationTokens += usage.cache_creation_input_tokens ?? 0;
+      bucket.cacheReadTokens += usage.cache_read_input_tokens ?? 0;
+      bucket.outputTokens += usage.output_tokens ?? 0;
+      bucket.messageCount += 1;
     }
 
-    bucket.inputTokens += usage.input_tokens ?? 0;
-    bucket.cacheCreationTokens += usage.cache_creation_input_tokens ?? 0;
-    bucket.cacheReadTokens += usage.cache_read_input_tokens ?? 0;
-    bucket.outputTokens += usage.output_tokens ?? 0;
-    bucket.messageCount += 1;
-  }
-
-  return { sessionId, totals, sessionStart, sessionEnd };
+    return { sessionId, totals, sessionStart, sessionEnd };
+  });
 }
 
 async function harvestClaudeTokenUsage(
@@ -151,6 +160,13 @@ async function harvestClaudeTokenUsage(
   }
 
   for (const file of files) {
+    if (!mountIO.available()) {
+      logger?.warn(
+        { projectDir },
+        "mount unavailable, aborting Claude harvest"
+      );
+      break;
+    }
     try {
       const summary = await parseClaudeSessionTokenUsage(file);
 
@@ -195,15 +211,17 @@ type CodexTokenUsage = {
  * Recursively discover all rollout JSONL files under ~/.codex/sessions/.
  * Files are organized as YYYY/MM/DD/rollout-*.jsonl.
  */
-async function discoverCodexRolloutFiles(): Promise<string[]> {
+/** Exported for testing (stall behavior); no production caller besides harvest. */
+export async function discoverCodexRolloutFiles(): Promise<string[]> {
   const files: string[] = [];
 
   async function walk(dir: string): Promise<void> {
     let entries: import("node:fs").Dirent[];
     try {
-      entries = (await readdir(dir, {
-        withFileTypes: true,
-      })) as import("node:fs").Dirent[];
+      // @types/node@24 readdir has no {signal} overload; bounded by MountIO timeout only
+      entries = await mountIO.run("readdir:codex-sessions", (_signal) =>
+        readdir(dir, { withFileTypes: true })
+      );
     } catch {
       return;
     }
@@ -235,59 +253,60 @@ async function parseCodexRolloutForAgent(
   sessionStart: string | null;
   sessionEnd: string | null;
 } | null> {
-  let matched = false;
-  let lastUsage: CodexTokenUsage | null = null;
-  let model = "unknown";
-  let sessionStart: string | null = null;
-  let sessionEnd: string | null = null;
-  let linesRead = 0;
-
-  const rl = createInterface({
-    input: createReadStream(filePath, { encoding: "utf-8" }),
-    crlfDelay: Infinity,
-  });
-
-  for await (const line of rl) {
-    linesRead++;
-
-    // Check for agent tag in the first 20 lines
-    if (!matched) {
-      if (linesRead > 20) break;
-      const tagMatch = DISPATCH_TAG_RE.exec(line);
-      if (tagMatch && tagMatch[1] === agentId) matched = true;
-      continue;
-    }
-
-    if (!line.trim()) continue;
-    let entry: Record<string, unknown>;
-    try {
-      entry = JSON.parse(line);
-    } catch {
-      continue;
-    }
-
-    const ts = entry.timestamp as string | undefined;
-    if (ts && !sessionStart) sessionStart = ts;
-    if (ts) sessionEnd = ts;
-
-    if (entry.type === "turn_context") {
-      const payload = entry.payload as Record<string, unknown> | undefined;
-      if (payload?.model) model = payload.model as string;
-    }
-
-    if (entry.type === "event_msg") {
-      const payload = entry.payload as Record<string, unknown> | undefined;
-      if (payload?.type === "token_count") {
-        const info = payload.info as Record<string, unknown> | undefined;
-        if (info?.total_token_usage) {
-          lastUsage = info.total_token_usage as CodexTokenUsage;
+  return mountIO.run(
+    `read:codex:${path.basename(filePath)}`,
+    async (signal, heartbeat) => {
+      let matched = false;
+      let lastUsage: CodexTokenUsage | null = null;
+      let model = "unknown";
+      let sessionStart: string | null = null;
+      let sessionEnd: string | null = null;
+      let linesRead = 0;
+
+      const rl = createInterface({
+        input: createReadStream(filePath, { encoding: "utf-8", signal }),
+        crlfDelay: Infinity,
+      });
+
+      for await (const line of rl) {
+        // Progress made — re-arm the idle deadline (see Claude parser above).
+        heartbeat();
+        linesRead++;
+        if (!matched) {
+          if (linesRead > 20) break;
+          const tagMatch = DISPATCH_TAG_RE.exec(line);
+          if (tagMatch && tagMatch[1] === agentId) matched = true;
+          continue;
+        }
+        if (!line.trim()) continue;
+        let entry: Record<string, unknown>;
+        try {
+          entry = JSON.parse(line);
+        } catch {
+          continue;
+        }
+        const ts = entry.timestamp as string | undefined;
+        if (ts && !sessionStart) sessionStart = ts;
+        if (ts) sessionEnd = ts;
+        if (entry.type === "turn_context") {
+          const payload = entry.payload as Record<string, unknown> | undefined;
+          if (payload?.model) model = payload.model as string;
+        }
+        if (entry.type === "event_msg") {
+          const payload = entry.payload as Record<string, unknown> | undefined;
+          if (payload?.type === "token_count") {
+            const info = payload.info as Record<string, unknown> | undefined;
+            if (info?.total_token_usage) {
+              lastUsage = info.total_token_usage as CodexTokenUsage;
+            }
+          }
         }
       }
-    }
-  }
 
-  if (!matched || !lastUsage) return null;
-  return { usage: lastUsage, model, sessionStart, sessionEnd };
+      if (!matched || !lastUsage) return null;
+      return { usage: lastUsage, model, sessionStart, sessionEnd };
+    }
+  );
 }
 
 async function harvestCodexTokenUsage(
@@ -299,6 +318,10 @@ async function harvestCodexTokenUsage(
   if (rolloutFiles.length === 0) return;
 
   for (const file of rolloutFiles) {
+    if (!mountIO.available()) {
+      logger?.warn({}, "mount unavailable, aborting Codex harvest");
+      break;
+    }
     try {
       const result = await parseCodexRolloutForAgent(file, agent.id);
       if (!result) continue;

apps/server/src/server.ts

@@ -36,6 +36,7 @@ import { deleteSetting, getSetting, setSetting } from "./db/settings.js";
 import { runCommand } from "./shared/lib/run-command.js";
 import { shouldSkipAutomaticMacPathProbe } from "./shared/mac-path-privacy.js";
 import { mimeType, resolveMediaDir } from "./shared/media.js";
+import { mountIO } from "./shared/mount-io/index.js";
 import { handleMcpRequest } from "./shared/mcp/server.js";
 import { readReleaseStore, writeReleaseStore } from "./release-store.js";
 import {
@@ -151,6 +152,9 @@ const app = Fastify({
   ...(config.tls && { https: { cert: config.tls.cert, key: config.tls.key } }),
 });
 const pool = createPool(config);
+// Route mount-io breaker open/recover transitions to the app logger so
+// operators get clear timestamps for a stalled gcsfuse mount.
+mountIO.setLogger(app.log);
 const agentManager = new AgentManager(pool, app.log, config);
 const focusTracker = new FocusTracker();
 const slackNotifier = new SlackNotifier(pool, app.log);

apps/server/src/shared/mount-io/circuit-breaker.ts

@@ -0,0 +1,43 @@
+export type BreakerState = "closed" | "open" | "half-open";
+
+export class CircuitBreaker {
+  private failures = 0;
+  private openedAt: number | null = null;
+
+  constructor(
+    private readonly threshold: number,
+    private readonly cooldownMs: number,
+    private readonly now: () => number = Date.now
+  ) {}
+
+  state(): BreakerState {
+    if (this.openedAt === null) return "closed";
+    if (this.now() - this.openedAt >= this.cooldownMs) return "half-open";
+    return "open";
+  }
+
+  canProceed(): boolean {
+    return this.state() !== "open";
+  }
+
+  recordSuccess(): void {
+    this.failures = 0;
+    this.openedAt = null;
+  }
+
+  recordFailure(): void {
+    if (this.state() === "half-open") {
+      this.openedAt = this.now();
+      return;
+    }
+    this.failures++;
+    if (this.failures >= this.threshold) {
+      this.openedAt = this.now();
+    }
+  }
+
+  reset(): void {
+    this.failures = 0;
+    this.openedAt = null;
+  }
+}