diff --git a/apps/desktop/src/features/server-runtime/connection/__tests__/ipc-relay.test.ts b/apps/desktop/src/features/server-runtime/connection/__tests__/ipc-relay.test.ts index 26997292b..559ba06e6 100644 --- a/apps/desktop/src/features/server-runtime/connection/__tests__/ipc-relay.test.ts +++ b/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; + destroy: ReturnType; + handlers: Record 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; + }); + 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(); + } + }); + }); }); diff --git a/apps/desktop/src/features/server-runtime/connection/ipc-relay.ts b/apps/desktop/src/features/server-runtime/connection/ipc-relay.ts index e3511bee9..c58df402f 100644 --- a/apps/desktop/src/features/server-runtime/connection/ipc-relay.ts +++ b/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(); + }; } diff --git a/apps/web/src/__tests__/canonical-runtime-reconcile.test.ts b/apps/web/src/__tests__/canonical-runtime-reconcile.test.ts new file mode 100644 index 000000000..c84d53093 --- /dev/null +++ b/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); + }); +}); diff --git a/apps/web/src/__tests__/connection.test.ts b/apps/web/src/__tests__/connection.test.ts index 5f8a69d02..2f7a063d8 100644 --- a/apps/web/src/__tests__/connection.test.ts +++ b/apps/web/src/__tests__/connection.test.ts @@ -234,6 +234,54 @@ describe("4001 auth failure handling", () => { }); }); +describe("liveness heartbeat", () => { + it("closes a silent socket when the heartbeat goes unanswered", async () => { + vi.useFakeTimers(); + const statusSpy = vi.fn(); + const transport = createWsTransport("ws://localhost:1234", { + onStatusChange: statusSpy, + }); + mockWsInstance.simulateOpen(); + + // 30s interval fires the probe; the mock never responds; the 10s + // heartbeat timeout closes the socket and drives the reconnect path. + await vi.advanceTimersByTimeAsync(30_000); + statusSpy.mockClear(); + await vi.advanceTimersByTimeAsync(10_000); + + expect(statusSpy).toHaveBeenCalledWith("reconnecting"); + transport.close(); + vi.useRealTimers(); + }); + + it("keeps the socket open when the heartbeat is answered", async () => { + vi.useFakeTimers(); + const statusSpy = vi.fn(); + const transport = createWsTransport("ws://localhost:1234", { + onStatusChange: statusSpy, + }); + mockWsInstance.simulateOpen(); + const sendSpy = vi.spyOn(mockWsInstance, "send"); + + await vi.advanceTimersByTimeAsync(30_000); + const heartbeat = sendSpy.mock.calls + .map(([raw]) => JSON.parse(raw as string) as { id: string; method?: string }) + .find((message) => message.method === "app.version"); + expect(heartbeat).toBeDefined(); + mockWsInstance.onmessage?.({ + data: JSON.stringify({ id: heartbeat!.id, result: "0.0.1" }), + }); + + await vi.advanceTimersByTimeAsync(10_000); + expect(mockWsInstance.readyState).toBe(1); + statusSpy.mockClear(); + await vi.advanceTimersByTimeAsync(10_000); + expect(statusSpy).not.toHaveBeenCalledWith("reconnecting"); + transport.close(); + vi.useRealTimers(); + }); +}); + /** * Browser-faithful socket: send() throws while CONNECTING and silently * discards while CLOSING/CLOSED, matching the WebSocket spec. The silent diff --git a/apps/web/src/features/conversation/messages/__tests__/turn-lifecycle.test.tsx b/apps/web/src/features/conversation/messages/__tests__/turn-lifecycle.test.tsx index 320346d86..c65a87e26 100644 --- a/apps/web/src/features/conversation/messages/__tests__/turn-lifecycle.test.tsx +++ b/apps/web/src/features/conversation/messages/__tests__/turn-lifecycle.test.tsx @@ -25,6 +25,7 @@ function canonicalTurn(status: AgentTurn["status"], trigger: AgentTurn["trigger" return { id: "canonical-turn", threadId: THREAD, status, trigger, permissionMode: "full", approvalReviewMode: "manual", approvalReviewReason: "manual-requested", + executionId: "00000000-0000-4000-8000-000000000042", providerIdentities: [], startedAt: NOW, endedAt: null, createdAt: NOW, updatedAt: NOW, }; } diff --git a/apps/web/src/stores/thread-lifecycle.ts b/apps/web/src/stores/thread-lifecycle.ts index cd30db2f3..491a7149e 100644 --- a/apps/web/src/stores/thread-lifecycle.ts +++ b/apps/web/src/stores/thread-lifecycle.ts @@ -1,24 +1,53 @@ -import type { AgentTurn, TurnRuntimePhase } from "@mcode/contracts"; +import type { AgentTurn, AgentTurnStatus, TurnRuntimePhase } from "@mcode/contracts"; import type { ThreadRecord } from "./thread-record"; -/** Selects the provider-owned child lifecycle when no local execution is active. */ -export function getCanonicalLifecycleTurn(threadId: string, record: ThreadRecord): AgentTurn | undefined { - if (record.runtimePhase === "running" || record.runtimePhase === "finalizing") return undefined; - const latest = Object.values(record.canonicalAgent.state.turns) +function latestCanonicalTurn(threadId: string, record: ThreadRecord): AgentTurn | undefined { + return Object.values(record.canonicalAgent.state.turns) .filter((turn) => turn.threadId === threadId) .sort((left, right) => Date.parse(left.startedAt ?? left.createdAt) - Date.parse(right.startedAt ?? right.createdAt) || left.id.localeCompare(right.id)) .at(-1); +} + +/** Selects the provider-owned child lifecycle when no local execution is active. */ +export function getCanonicalLifecycleTurn(threadId: string, record: ThreadRecord): AgentTurn | undefined { + const latest = latestCanonicalTurn(threadId, record); if (latest?.trigger.kind !== "child") return undefined; - if (record.runtimePhase !== "idle" && (latest.status === "Pending" || latest.status === "Running")) return undefined; + // A runtime phase stamped by this same canonical turn must not gate it out; + // only a locally-owned phase suppresses the child lifecycle. + const locallyOwnedPhase = getCanonicalRuntimeTurn(threadId, record)?.id === latest.id + ? "idle" + : record.runtimePhase; + if (locallyOwnedPhase === "running" || locallyOwnedPhase === "finalizing") return undefined; + if (locallyOwnedPhase !== "idle" && (latest.status === "Pending" || latest.status === "Running")) return undefined; return latest; } -/** Resolves the lifecycle shared by transcript, Composer, and running-thread indicators. */ -export function getThreadRuntimePhase(threadId: string, record: ThreadRecord): TurnRuntimePhase { - const turn = getCanonicalLifecycleTurn(threadId, record); - if (!turn) return record.runtimePhase; - switch (turn.status) { +/** + * Canonical turn that owns this record's runtime phase, once correlation is + * proven. A tracked local execution only follows the canonical turn carrying + * the same execution identity, so an older persisted turn cannot clear or + * resurrect a live run while a matching terminal can clear a stale one. + */ +export function getCanonicalRuntimeTurn(threadId: string, record: ThreadRecord): AgentTurn | undefined { + if (record.runtimePhase === "finalizing") return undefined; + const latest = latestCanonicalTurn(threadId, record); + if (!latest) return undefined; + if (record.turnExecutionId !== null) { + return latest.executionId === record.turnExecutionId ? latest : undefined; + } + // Optimistic sends run without an identity until turnStarted lands; a prior + // terminal turn must not cancel that window. + if (record.runtimePhase === "running") return undefined; + // Without an identity, canonical Pending or Running claims only idle + // records; terminal truth may claim any non-busy record. + if ((latest.status === "Pending" || latest.status === "Running") && record.runtimePhase !== "idle") return undefined; + return latest; +} + +/** Runtime phase equivalent of one canonical turn status. */ +export function phaseForTurnStatus(status: AgentTurnStatus): TurnRuntimePhase { + switch (status) { case "Pending": case "Running": return "running"; case "Completed": return "completed"; @@ -28,6 +57,17 @@ export function getThreadRuntimePhase(threadId: string, record: ThreadRecord): T } } +/** Canonical-owned runtime phase for this record, or null when legacy owns it. */ +export function getCanonicalRuntimePhase(threadId: string, record: ThreadRecord): TurnRuntimePhase | null { + const turn = getCanonicalRuntimeTurn(threadId, record); + return turn ? phaseForTurnStatus(turn.status) : null; +} + +/** Resolves the lifecycle shared by transcript, Composer, and running-thread indicators. */ +export function getThreadRuntimePhase(threadId: string, record: ThreadRecord): TurnRuntimePhase { + return getCanonicalRuntimePhase(threadId, record) ?? record.runtimePhase; +} + /** Whether the current lifecycle still owns execution, including final persistence. */ export function isThreadRuntimeActive(threadId: string, record: ThreadRecord): boolean { const phase = getThreadRuntimePhase(threadId, record); diff --git a/apps/web/src/stores/threadStore.ts b/apps/web/src/stores/threadStore.ts index e4616371e..61421fa3b 100644 --- a/apps/web/src/stores/threadStore.ts +++ b/apps/web/src/stores/threadStore.ts @@ -105,7 +105,7 @@ import { } from "./thread-store/usage"; export { mergeProviderUsageSnapshot } from "./thread-store/usage"; -import { isThreadRuntimeActive } from "./thread-lifecycle"; +import { getCanonicalRuntimeTurn, isThreadRuntimeActive, phaseForTurnStatus } from "./thread-lifecycle"; function deriveRunningThreadIds(records: Map): Set { return new Set( @@ -186,8 +186,6 @@ interface ThreadState { clearMessages: () => void; /** Deactivate the selected conversation and invalidate any active hydration commit. */ deactivateConversation: () => void; - /** Returns true if an agent is actively executing on the given thread. */ - isThreadRunning: (threadId: string) => boolean; /** Set questions received from the model and show the wizard. */ setPlanQuestions: (threadId: string, questions: PlanQuestion[]) => void; /** Record the user's answer for one question. */ @@ -2715,6 +2713,40 @@ export const useThreadStore = create((zustandSet, get) => { return false; }; + // Canonical lifecycle owns runtime state once its turn correlates to the + // record, so a dropped legacy terminal event cannot strand the running set + // or the composer gate. + const reconcileCanonicalRuntime = ( + records: Map, + runningThreadIds: Set, + threadId: string, + ): { records: Map; runningThreadIds: Set } => { + const record = getThreadRecord(records, threadId); + const turn = getCanonicalRuntimeTurn(threadId, record); + if (!turn) return { records, runningThreadIds }; + const phase = phaseForTurnStatus(turn.status); + let nextRecords = records; + const patch: Partial = {}; + if (record.runtimePhase !== phase) patch.runtimePhase = phase; + // Stamp the execution identity on live claims so later reconciles and the + // child lifecycle gate can keep correlating this turn to the record. + if (record.turnExecutionId === null && turn.executionId + && (turn.status === "Pending" || turn.status === "Running")) { + patch.turnExecutionId = turn.executionId; + } + if (Object.keys(patch).length > 0) { + nextRecords = patchThreadRecord(records, threadId, patch); + } + const running = phase === "running"; + if (runningThreadIds.has(threadId) === running) { + return { records: nextRecords, runningThreadIds }; + } + const nextRunning = new Set(runningThreadIds); + if (running) nextRunning.add(threadId); + else nextRunning.delete(threadId); + return { records: nextRecords, runningThreadIds: nextRunning }; + }; + return { records: new Map(), currentThreadId: null, @@ -2728,6 +2760,7 @@ export const useThreadStore = create((zustandSet, get) => { flushPendingTextDeltas(); set((state) => { let records = state.records; + let runningThreadIds = state.runningThreadIds; let changed = false; for (const recovery of recoveries) { const current = getThreadRecord(records, recovery.threadId); @@ -2738,9 +2771,14 @@ export const useThreadStore = create((zustandSet, get) => { canonicalAgent: update.replica, ...(update.installedSnapshot ? recoverParentNarrative(recovery.threadId, update.replica.state) : {}), }); + const reconciled = reconcileCanonicalRuntime(records, runningThreadIds, recovery.threadId); + records = reconciled.records; + runningThreadIds = reconciled.runningThreadIds; changed = true; } - return changed ? { records } : {}; + return changed + ? { records, ...(runningThreadIds === state.runningThreadIds ? {} : { runningThreadIds }) } + : {}; }); }, @@ -2757,8 +2795,11 @@ export const useThreadStore = create((zustandSet, get) => { const update = applyCanonicalPushEvents(current.canonicalAgent, threadId, events); if (update.replica === current.canonicalAgent) return {}; accepted = true; + const records = patchThreadRecord(state.records, threadId, { canonicalAgent: update.replica }); + const reconciled = reconcileCanonicalRuntime(records, state.runningThreadIds, threadId); return { - records: patchThreadRecord(state.records, threadId, { canonicalAgent: update.replica }), + records: reconciled.records, + ...(reconciled.runningThreadIds === state.runningThreadIds ? {} : { runningThreadIds: reconciled.runningThreadIds }), }; }); if ( @@ -3336,11 +3377,6 @@ export const useThreadStore = create((zustandSet, get) => { threadHydrator.deactivate(); }, - /** Check whether an agent is currently executing on the given thread. */ - isThreadRunning: (threadId) => { - return get().runningThreadIds.has(threadId); - }, - /** Return per-thread settings, preferring in-memory overrides then DB-persisted values then defaults. */ getThreadSettings: (threadId) => { const stored = get().records.get(threadId); diff --git a/apps/web/src/transport/__tests__/rpc-timeout-recovery.test.ts b/apps/web/src/transport/__tests__/rpc-timeout-recovery.test.ts index af0d5cc3c..28ffc90f0 100644 --- a/apps/web/src/transport/__tests__/rpc-timeout-recovery.test.ts +++ b/apps/web/src/transport/__tests__/rpc-timeout-recovery.test.ts @@ -49,6 +49,11 @@ class TimeoutSocket { const parsed: unknown = JSON.parse(raw); if (!isRpcRequest(parsed)) throw new Error("Expected an RPC request"); this.requests.push(parsed); + // A socket that ignores the liveness probe is closed by the watchdog; + // these tests exercise RPC timeouts on a live connection instead. + if (parsed.method === "app.version") { + queueMicrotask(() => this.respond(parsed, "0.0.1")); + } } open(): void { diff --git a/apps/web/src/transport/ws-transport.ts b/apps/web/src/transport/ws-transport.ts index 5329652c7..925d3caf4 100644 --- a/apps/web/src/transport/ws-transport.ts +++ b/apps/web/src/transport/ws-transport.ts @@ -107,6 +107,13 @@ const TERMINAL_LATE_CREATE_CLEANUP_TIMEOUT_MS = 10_000; /** Maximum Terminal creates whose late responses still need exact cleanup. */ const MAX_PENDING_TERMINAL_CREATE_CLEANUPS = 8; +/** Interval between liveness probes on an open socket. */ +const HEARTBEAT_INTERVAL_MS = 30_000; +/** How long a heartbeat may go unanswered before the socket is treated as dead. */ +const HEARTBEAT_TIMEOUT_MS = 10_000; +/** Deadline for `agent.send` so a composer submit cannot park forever on a dead socket. */ +const SEND_MESSAGE_TIMEOUT_MS = 20_000; + /** Last thread-list refresh timestamp per workspace, triggered on WS reconnect. */ const lastLoadThreadsAtByWorkspace = new Map(); /** Minimum interval between reconnect-triggered thread-list fetches to avoid rapid-reconnect storms. */ @@ -386,6 +393,8 @@ export function createWsTransport( let closed = false; let reconnectDelay = MIN_RECONNECT_MS; let reconnectTimer: ReturnType | null = null; + let heartbeatTimer: ReturnType | null = null; + let heartbeatInFlight = false; let terminalSelectionPromise: Promise | null = null; // Track consecutive auth failures so we apply backoff after 3 immediate // retries, preventing a tight loop when the token is persistently wrong. @@ -554,6 +563,7 @@ export function createWsTransport( setAttachmentTransportWsUrl(url); resolveReady(); options?.onStatusChange?.("connected"); + startHeartbeat(); invalidateLiveTurnDiff(); void selectTerminalClientWithRecovery(); @@ -614,6 +624,7 @@ export function createWsTransport( ws.onclose = (event: CloseEvent) => { freshTurnDiffThreads.clear(); + stopHeartbeat(); rejectPending("WebSocket disconnected"); lateResponseHandlers = new Map(); pendingTerminalCreateCleanups.clear(); @@ -640,6 +651,35 @@ export function createWsTransport( pending = new Map(); } + // A half-open socket answers nothing and never fires `close`; without a + // probe, reconnect and runtime resync wait forever while the UI shows stale + // streaming state. The probe is a cheap existing RPC bounded by timeoutMs; + // on timeout we close locally so onclose drives the recovery ladder. + function heartbeatTick(): void { + if (heartbeatInFlight || closed || ws.readyState !== WebSocket.OPEN) return; + heartbeatInFlight = true; + rpc("app.version", {}, { timeoutMs: HEARTBEAT_TIMEOUT_MS }) + .catch(() => { + if (ws.readyState === WebSocket.OPEN) ws.close(); + }) + .finally(() => { + heartbeatInFlight = false; + }); + } + + function startHeartbeat(): void { + stopHeartbeat(); + heartbeatTimer = setInterval(heartbeatTick, HEARTBEAT_INTERVAL_MS); + } + + function stopHeartbeat(): void { + if (heartbeatTimer) { + clearInterval(heartbeatTimer); + heartbeatTimer = null; + } + heartbeatInFlight = false; + } + function invalidateLiveTurnDiff(): void { void Promise.all([import("@/features/projects/state/workspaceStore"), import("@/stores/diffStore")]).then(([workspace, diff]) => { const threadId = workspace.useWorkspaceStore.getState().activeThreadId; @@ -1153,7 +1193,7 @@ export function createWsTransport( ...(replyToMessageId && { replyToMessageId }), ...(quotedText && { quotedText }), ...guardrails, - }); + }, { timeoutMs: SEND_MESSAGE_TIMEOUT_MS }); }, getRecoveryIncident: () => rpc("agent.recoveryIncident", {}), @@ -1543,6 +1583,7 @@ export function createWsTransport( // Lifecycle close: () => { closed = true; + stopHeartbeat(); if (reconnectTimer) { clearTimeout(reconnectTimer); reconnectTimer = null; diff --git a/packages/agent-model/src/records.ts b/packages/agent-model/src/records.ts index 16f103648..c5a2d244b 100644 --- a/packages/agent-model/src/records.ts +++ b/packages/agent-model/src/records.ts @@ -2,6 +2,7 @@ import { z } from "zod"; import { AgentItemIdSchema, AgentThreadIdSchema, + AgentTurnExecutionIdSchema, AgentTurnIdSchema, CanonicalTimestampSchema, CollaborationActionIdSchema, @@ -86,6 +87,9 @@ export const AgentTurnSchema = z threadId: AgentThreadIdSchema, status: AgentTurnStatusSchema, trigger: AgentTurnTriggerSchema, + // Stamped by the reducer from routing so clients can correlate the turn to + // the local runtime without threading the event envelope through state. + executionId: AgentTurnExecutionIdSchema.optional(), permissionMode: z.enum(["supervised", "full"]), approvalReviewMode: z.enum(["manual", "automatic"]), approvalReviewReason: z.string().min(1).max(128), diff --git a/packages/agent-model/src/reducer.ts b/packages/agent-model/src/reducer.ts index 83ab5ff85..29880fa9e 100644 --- a/packages/agent-model/src/reducer.ts +++ b/packages/agent-model/src/reducer.ts @@ -243,12 +243,18 @@ function reduceTurnCreated( if (event.routing.threadId !== turn.threadId || event.routing.turnId !== turn.id) { return { state, outcome: "routing-conflict" }; } + if (turn.executionId !== undefined && turn.executionId !== event.routing.executionId) { + return { state, outcome: "routing-conflict" }; + } const currentTurn = state.turns[turn.id]; if (currentTurn && TERMINAL_TURN_STATUSES.has(currentTurn.status)) { return { state: { ...state, ...acceptedInputState }, outcome: "terminal-outcome-confirmed" }; } + const stored = turn.executionId === event.routing.executionId + ? turn + : { ...turn, executionId: event.routing.executionId }; return { - state: { ...state, turns: { ...state.turns, [turn.id]: turn }, ...acceptedInputState }, + state: { ...state, turns: { ...state.turns, [turn.id]: stored }, ...acceptedInputState }, outcome: "applied", }; }