fix(desktop): reconnect the IPC push relay with backoff

apps/desktop/src/features/server-runtime/connection/__tests__/ipc-relay.test.ts

@@ -224,4 +224,97 @@ describe("startIpcRelay", () => {
       expect(win.webContents.send).toHaveBeenCalledWith("ipc-push-message", message);
     });
   });
+
+  describe("reconnect", () => {
+    interface TrackedSocket {
+      on: ReturnType<typeof vi.fn>;
+      destroy: ReturnType<typeof vi.fn>;
+      handlers: Record<string, (...args: unknown[]) => void>;
+    }
+
+    function trackedSockets(): TrackedSocket[] {
+      const sockets: TrackedSocket[] = [];
+      vi.mocked(NodeNet.connect).mockImplementation(() => {
+        const socket: TrackedSocket = {
+          on: vi.fn((event: string, handler: (...args: unknown[]) => void) => {
+            socket.handlers[event] = handler;
+            return socket;
+          }),
+          destroy: vi.fn(),
+          handlers: {},
+        };
+        sockets.push(socket);
+        return socket as unknown as ReturnType<typeof NodeNet.connect>;
+      });
+      return sockets;
+    }
+
+    it("reconnects with backoff after the socket closes", async () => {
+      vi.useFakeTimers();
+      try {
+        const sockets = trackedSockets();
+        startIpcRelay("/tmp/mcode.sock", makeWindow() as never);
+
+        sockets[0]!.handlers["close"]?.();
+        await vi.advanceTimersByTimeAsync(500);
+        expect(sockets).toHaveLength(2);
+
+        sockets[1]!.handlers["close"]?.();
+        await vi.advanceTimersByTimeAsync(999);
+        expect(sockets).toHaveLength(2);
+        await vi.advanceTimersByTimeAsync(1);
+        expect(sockets).toHaveLength(3);
+      } finally {
+        vi.useRealTimers();
+      }
+    });
+
+    it("resets the backoff after a successful connect", async () => {
+      vi.useFakeTimers();
+      try {
+        const sockets = trackedSockets();
+        startIpcRelay("/tmp/mcode.sock", makeWindow() as never);
+
+        sockets[0]!.handlers["close"]?.();
+        await vi.advanceTimersByTimeAsync(500);
+        sockets[1]!.handlers["connect"]?.();
+        sockets[1]!.handlers["close"]?.();
+        await vi.advanceTimersByTimeAsync(500);
+        expect(sockets).toHaveLength(3);
+      } finally {
+        vi.useRealTimers();
+      }
+    });
+
+    it("does not reconnect after cleanup", async () => {
+      vi.useFakeTimers();
+      try {
+        const sockets = trackedSockets();
+        const cleanup = startIpcRelay("/tmp/mcode.sock", makeWindow() as never);
+
+        cleanup();
+        sockets[0]!.handlers["close"]?.();
+        await vi.advanceTimersByTimeAsync(30_000);
+        expect(sockets).toHaveLength(1);
+      } finally {
+        vi.useRealTimers();
+      }
+    });
+
+    it("stops reconnecting when the window is destroyed", async () => {
+      vi.useFakeTimers();
+      try {
+        const sockets = trackedSockets();
+        const win = makeWindow(false);
+        startIpcRelay("/tmp/mcode.sock", win as never);
+        win.isDestroyed.mockReturnValue(true);
+
+        sockets[0]!.handlers["close"]?.();
+        await vi.advanceTimersByTimeAsync(30_000);
+        expect(sockets).toHaveLength(1);
+      } finally {
+        vi.useRealTimers();
+      }
+    });
+  });
 });

apps/desktop/src/features/server-runtime/connection/ipc-relay.ts

@@ -12,6 +12,11 @@ import * as NodeNet from "node:net";
  *  a corrupt or malicious length prefix; the socket is destroyed immediately. */
 const MAX_FRAME_SIZE = 8 * 1024 * 1024;
 
+/** Reconnect backoff bounds. A dropped relay silently degraded the renderer to
+ *  the WebSocket path forever before this existed. */
+const MIN_RECONNECT_MS = 500;
+const MAX_RECONNECT_MS = 15_000;
+
 /** Minimal subset of BrowserWindow required by the relay. */
 interface RelayWindow {
   isDestroyed(): boolean;
@@ -28,57 +33,84 @@ interface RelayWindow {
  * Wire format: each frame is a 4-byte big-endian length prefix followed by
  * the UTF-8 encoded JSON body.
  *
+ * On socket close the renderer is told to fall back to WebSocket and the
+ * relay reconnects with exponential backoff. The renderer re-suppresses
+ * channels as frames resume, so no reconnect handshake is required.
+ *
  * @returns A cleanup function. Call it when the window closes to destroy
  *   the socket and prevent a named-pipe handle leak on Windows.
  */
 export function startIpcRelay(ipcPath: string, window: RelayWindow): () => void {
   if (!ipcPath) return () => { /* no-op: no socket was opened */ };
 
-  const socket = NodeNet.connect(ipcPath);
-  const chunks: Buffer[] = [];
-  let totalLen = 0;
+  let stopped = false;
+  let socket: NodeNet.Socket | null = null;
+  let reconnectTimer: NodeJS.Timeout | null = null;
+  let reconnectDelay = MIN_RECONNECT_MS;
 
-  socket.on("data", (chunk: Buffer) => {
-    chunks.push(chunk);
-    totalLen += chunk.length;
+  const windowAlive = () => !window.isDestroyed() && !window.webContents.isDestroyed();
 
-    // Avoid concat overhead when only one chunk is buffered.
-    let buffer = chunks.length === 1 ? chunks[0] : Buffer.concat(chunks, totalLen);
-    chunks.length = 0;
-    totalLen = 0;
+  const connect = (): void => {
+    if (stopped || !windowAlive()) return;
+    const current = NodeNet.connect(ipcPath);
+    socket = current;
+    const chunks: Buffer[] = [];
+    let totalLen = 0;
 
-    while (buffer.length >= 4) {
-      const frameLen = buffer.readUInt32BE(0);
-      if (frameLen > MAX_FRAME_SIZE) {
-        socket.destroy();
-        return;
-      }
-      if (buffer.length < 4 + frameLen) break;
+    current.on("connect", () => {
+      reconnectDelay = MIN_RECONNECT_MS;
+    });
+
+    current.on("data", (chunk: Buffer) => {
+      chunks.push(chunk);
+      totalLen += chunk.length;
 
-      const json = buffer.subarray(4, 4 + frameLen).toString("utf-8");
-      buffer = buffer.subarray(4 + frameLen);
+      // Avoid concat overhead when only one chunk is buffered.
+      let buffer = chunks.length === 1 ? chunks[0] : Buffer.concat(chunks, totalLen);
+      chunks.length = 0;
+      totalLen = 0;
 
-      try {
-        const data = JSON.parse(json) as unknown;
-        if (!window.isDestroyed() && !window.webContents.isDestroyed()) {
-          window.webContents.send("ipc-push-message", data);
+      while (buffer.length >= 4) {
+        const frameLen = buffer.readUInt32BE(0);
+        if (frameLen > MAX_FRAME_SIZE) {
+          current.destroy();
+          return;
         }
-      } catch { /* malformed frame - skip */ }
-    }
-
-    // Retain leftover bytes for the next data event.
-    if (buffer.length > 0) {
-      chunks.push(buffer);
-      totalLen = buffer.length;
-    }
-  });
-
-  socket.on("error", () => socket.destroy());
-  socket.on("close", () => {
-    if (!window.isDestroyed() && !window.webContents.isDestroyed()) {
+        if (buffer.length < 4 + frameLen) break;
+
+        const json = buffer.subarray(4, 4 + frameLen).toString("utf-8");
+        buffer = buffer.subarray(4 + frameLen);
+
+        try {
+          const data = JSON.parse(json) as unknown;
+          if (windowAlive()) {
+            window.webContents.send("ipc-push-message", data);
+          }
+        } catch { /* malformed frame - skip */ }
+      }
+
+      // Retain leftover bytes for the next data event.
+      if (buffer.length > 0) {
+        chunks.push(buffer);
+        totalLen = buffer.length;
+      }
+    });
+
+    current.on("error", () => current.destroy());
+    current.on("close", () => {
+      if (socket === current) socket = null;
+      if (stopped || !windowAlive()) return;
       window.webContents.send("ipc-push-disconnect");
-    }
-  });
+      reconnectTimer = setTimeout(connect, reconnectDelay);
+      reconnectDelay = Math.min(reconnectDelay * 2, MAX_RECONNECT_MS);
+    });
+  };
+
+  connect();
 
-  return () => socket.destroy();
+  return () => {
+    stopped = true;
+    if (reconnectTimer) clearTimeout(reconnectTimer);
+    socket?.destroy();
+  };
 }

apps/web/src/__tests__/canonical-runtime-reconcile.test.ts

@@ -0,0 +1,147 @@
+import { describe, expect, it, beforeEach, vi } from "vitest";
+import type { CanonicalAgentEventEnvelope } from "@mcode/contracts";
+import { useThreadStore } from "@/stores/threadStore";
+import {
+  resetThreadStoreForTests,
+  seedThreadRecord,
+  readThreadField,
+} from "@/stores/thread-store-test-utils";
+import { clearRecordCache } from "@/features/conversation/hydration/record-cache";
+import { mockTransport } from "./mocks/transport";
+
+vi.mock("@/transport", async () => ({
+  ...(await vi.importActual("@/transport")),
+  getTransport: () => mockTransport,
+}));
+
+const THREAD_ID = "thread-1";
+const TURN_ID = "turn-1";
+const EXECUTION_ID = "00000000-0000-4000-8000-000000000001";
+const OTHER_EXECUTION_ID = "00000000-0000-4000-8000-000000000099";
+const NOW = "2026-08-11T12:00:00.000Z";
+
+function envelope(
+  eventId: string,
+  acceptedSequence: number,
+  executionId: string,
+  payload: CanonicalAgentEventEnvelope["payload"],
+): CanonicalAgentEventEnvelope {
+  return {
+    eventId,
+    routing: {
+      threadId: THREAD_ID,
+      turnId: payload.type === "thread.recorded" ? undefined : TURN_ID,
+      executionId,
+    },
+    sourceProviderId: "codex",
+    sourceIdentities: [],
+    acceptedSequence,
+    durableRevision: acceptedSequence,
+    serverTimestamps: { acceptedAt: NOW, persistedAt: NOW },
+    payload,
+  };
+}
+
+function turnEvents(executionId: string, terminal?: "completed" | "interrupted"): CanonicalAgentEventEnvelope[] {
+  const events: CanonicalAgentEventEnvelope[] = [
+    envelope("e1", 1, executionId, {
+      type: "thread.recorded",
+      thread: {
+        id: THREAD_ID,
+        workspaceId: "workspace-1",
+        rootThreadId: THREAD_ID,
+        providerId: "codex",
+        providerIdentities: [],
+        activityState: "Active",
+        conversationRevision: 1,
+        rosterRevision: 0,
+        createdAt: NOW,
+        updatedAt: NOW,
+      },
+    }),
+    envelope("e2", 2, executionId, {
+      type: "turn.created",
+      turn: {
+        id: TURN_ID,
+        threadId: THREAD_ID,
+        status: "Pending",
+        trigger: { kind: "user" },
+        permissionMode: "full",
+        approvalReviewMode: "manual",
+        approvalReviewReason: "manual-requested",
+        providerIdentities: [],
+        startedAt: null,
+        endedAt: null,
+        createdAt: NOW,
+        updatedAt: NOW,
+      },
+    }),
+    envelope("e3", 3, executionId, { type: "turn.started", startedAt: NOW }),
+  ];
+  if (terminal === "completed") {
+    events.push(envelope("e4", 4, executionId, { type: "turn.completed", endedAt: NOW }));
+  } else if (terminal === "interrupted") {
+    events.push(envelope("e4", 4, executionId, { type: "turn.interrupted", endedAt: NOW, reason: "restart" }));
+  }
+  return events;
+}
+
+function seedRuntime(phase: "running" | "idle", turnExecutionId: string | null, running: boolean) {
+  useThreadStore.setState({
+    records: seedThreadRecord(THREAD_ID, { runtimePhase: phase, turnExecutionId }),
+    runningThreadIds: running ? new Set([THREAD_ID]) : new Set(),
+  });
+}
+
+describe("canonical runtime reconciliation", () => {
+  beforeEach(() => {
+    clearRecordCache();
+    resetThreadStoreForTests();
+    vi.clearAllMocks();
+  });
+
+  it("clears a stale running flag when the correlated canonical turn terminates", () => {
+    seedRuntime("running", EXECUTION_ID, true);
+
+    useThreadStore.getState().handleCanonicalAgentEvents(THREAD_ID, turnEvents(EXECUTION_ID, "completed"));
+
+    expect(readThreadField(THREAD_ID, (r) => r.runtimePhase)).toBe("completed");
+    expect(useThreadStore.getState().runningThreadIds.has(THREAD_ID)).toBe(false);
+  });
+
+  it("claims an idle record when canonical reports a running turn", () => {
+    seedRuntime("idle", null, false);
+
+    useThreadStore.getState().handleCanonicalAgentEvents(THREAD_ID, turnEvents(EXECUTION_ID));
+
+    expect(readThreadField(THREAD_ID, (r) => r.runtimePhase)).toBe("running");
+    expect(useThreadStore.getState().runningThreadIds.has(THREAD_ID)).toBe(true);
+  });
+
+  it("keeps the optimistic running window while no canonical turn correlates", () => {
+    seedRuntime("running", null, true);
+
+    useThreadStore.getState().handleCanonicalAgentEvents(THREAD_ID, turnEvents(EXECUTION_ID, "completed"));
+
+    expect(readThreadField(THREAD_ID, (r) => r.runtimePhase)).toBe("running");
+    expect(useThreadStore.getState().runningThreadIds.has(THREAD_ID)).toBe(true);
+  });
+
+  it("ignores a canonical turn carrying a different execution identity", () => {
+    seedRuntime("running", OTHER_EXECUTION_ID, true);
+
+    useThreadStore.getState().handleCanonicalAgentEvents(THREAD_ID, turnEvents(EXECUTION_ID, "completed"));
+
+    expect(readThreadField(THREAD_ID, (r) => r.runtimePhase)).toBe("running");
+    expect(useThreadStore.getState().runningThreadIds.has(THREAD_ID)).toBe(true);
+  });
+
+  it("maps an interrupted canonical turn to the interrupted phase", () => {
+    seedRuntime("running", EXECUTION_ID, true);
+
+    useThreadStore.getState().handleCanonicalAgentEvents(THREAD_ID, turnEvents(EXECUTION_ID, "interrupted"));
+
+    expect(readThreadField(THREAD_ID, (r) => r.runtimePhase)).toBe("interrupted");
+    expect(useThreadStore.getState().runningThreadIds.has(THREAD_ID)).toBe(false);
+  });
+});