From a51ac2decb032587c9f8915902c7e49e4623f3e3 Mon Sep 17 00:00:00 2001 From: Agent Runtime Fixture Date: Sun, 27 Sep 2026 23:07:58 +0100 Subject: [PATCH 1/6] fix(web): close wedged websockets with an app.version heartbeat A half-open socket answers nothing and never fires close, so every recovery mechanism (reconnect, hydration, subscription replay) waited forever while the UI showed stale streaming state. A 30s interval now issues the existing app.version RPC with a 10s deadline; an unanswered probe closes the socket locally and the established onclose ladder owns recovery. agent.send gains a 20s timeout so a composer submit cannot park forever on a dead socket. The wedged-socket test now answers the liveness probe, since a socket that ignores it is exactly what the watchdog closes. --- apps/web/src/__tests__/connection.test.ts | 48 +++++++++++++++++++ .../__tests__/rpc-timeout-recovery.test.ts | 5 ++ apps/web/src/transport/ws-transport.ts | 43 ++++++++++++++++- 3 files changed, 95 insertions(+), 1 deletion(-) 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/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; From 64d4cdd8359b97623fcc3950b56dd04f9a381c30 Mon Sep 17 00:00:00 2001 From: Agent Runtime Fixture Date: Sun, 27 Sep 2026 23:08:29 +0100 Subject: [PATCH 2/6] feat(agent-model): stamp turn records with their execution identity The reducer now writes routing.executionId onto the stored turn so clients can correlate a canonical turn to the local runtime without threading the event envelope through state. A payload-supplied executionId that disagrees with routing is rejected as a routing conflict. --- packages/agent-model/src/records.ts | 4 ++++ packages/agent-model/src/reducer.ts | 8 +++++++- 2 files changed, 11 insertions(+), 1 deletion(-) 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", }; } From 5059a3fffaf2b4de723c24c9122fd54940beda97 Mon Sep 17 00:00:00 2001 From: Agent Runtime Fixture Date: Sun, 27 Sep 2026 23:08:29 +0100 Subject: [PATCH 3/6] fix(web): derive thread runtime phase from canonical turn state Composer, project tree, and running-thread indicators now follow the canonical replica once a turn correlates by execution identity, so a dropped legacy terminal event can no longer strand runningThreadIds. The optimistic window before turnStarted and the finalizing phase stay on the legacy path until a canonical turn proves correlation. The narrative projection keeps its existing child-only lifecycle gate. --- .../canonical-runtime-reconcile.test.ts | 147 ++++++++++++++++++ apps/web/src/stores/thread-lifecycle.ts | 54 +++++-- apps/web/src/stores/threadStore.ts | 40 ++++- 3 files changed, 228 insertions(+), 13 deletions(-) create mode 100644 apps/web/src/__tests__/canonical-runtime-reconcile.test.ts 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/stores/thread-lifecycle.ts b/apps/web/src/stores/thread-lifecycle.ts index cd30db2f3..6d7cc57cc 100644 --- a/apps/web/src/stores/thread-lifecycle.ts +++ b/apps/web/src/stores/thread-lifecycle.ts @@ -1,24 +1,47 @@ -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 { + if (record.runtimePhase === "running" || record.runtimePhase === "finalizing") return 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; 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; +} + +function phaseForTurnStatus(status: AgentTurnStatus): TurnRuntimePhase { + switch (status) { case "Pending": case "Running": return "running"; case "Completed": return "completed"; @@ -28,6 +51,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..19f5bf881 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 { getCanonicalRuntimePhase, isThreadRuntimeActive } from "./thread-lifecycle"; function deriveRunningThreadIds(records: Map): Set { return new Set( @@ -2715,6 +2715,31 @@ 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 phase = getCanonicalRuntimePhase(threadId, record); + if (phase === null) return { records, runningThreadIds }; + let nextRecords = records; + if (record.runtimePhase !== phase) { + nextRecords = patchThreadRecord(records, threadId, { runtimePhase: phase }); + } + 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 +2753,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 +2764,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 +2788,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 ( From 6e510d2775af10ff0d6e0f2da0be3103375ee2e0 Mon Sep 17 00:00:00 2001 From: Agent Runtime Fixture Date: Sun, 27 Sep 2026 23:21:31 +0100 Subject: [PATCH 4/6] fix(desktop): reconnect the IPC push relay with backoff The relay opened one socket and never recovered: a dropped connection silently degraded the renderer to WebSocket-only delivery for the rest of the session. Reconnect in the main process with bounded exponential backoff, reset on successful connect, and stop when the window dies or the relay is torn down. The renderer re-suppresses push channels as frames resume, so no reconnect handshake is needed. --- .../connection/__tests__/ipc-relay.test.ts | 93 +++++++++++++++ .../server-runtime/connection/ipc-relay.ts | 110 +++++++++++------- 2 files changed, 164 insertions(+), 39 deletions(-) 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(); + }; } From eeb139c237514e2b267b235359c904ee22a66ec5 Mon Sep 17 00:00:00 2001 From: Agent Runtime Fixture Date: Sun, 27 Sep 2026 23:21:36 +0100 Subject: [PATCH 5/6] refactor(web): drop the unused isThreadRunning selector No caller remains: consumers read runningThreadIds directly. --- apps/web/src/stores/threadStore.ts | 7 ------- 1 file changed, 7 deletions(-) diff --git a/apps/web/src/stores/threadStore.ts b/apps/web/src/stores/threadStore.ts index 19f5bf881..d10ef1e9e 100644 --- a/apps/web/src/stores/threadStore.ts +++ b/apps/web/src/stores/threadStore.ts @@ -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. */ @@ -3370,11 +3368,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); From fc9e113959118d0e40534c8923e5b7be572295db Mon Sep 17 00:00:00 2001 From: Agent Runtime Fixture Date: Tue, 29 Sep 2026 21:15:07 +0100 Subject: [PATCH 6/6] fix(web): keep the canonical lifecycle gate when canonical owns the phase The runtime reconcile stamps runtimePhase from a correlated canonical turn but never correlated the record's execution identity, so a canonically owned "running" phase made getCanonicalLifecycleTurn treat the provider-owned child turn as a local execution and suppress its projection. A later terminal canonical event could then no longer clear the stamped phase either. reconcileCanonicalRuntime now stamps turnExecutionId from the claiming turn's executionId on live claims, and the lifecycle gate only treats runtimePhase as locally owned when the canonical runtime turn is not the latest turn. --- .../__tests__/turn-lifecycle.test.tsx | 1 + apps/web/src/stores/thread-lifecycle.ts | 12 +++++++++--- apps/web/src/stores/threadStore.ts | 19 ++++++++++++++----- 3 files changed, 24 insertions(+), 8 deletions(-) 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 6d7cc57cc..491a7149e 100644 --- a/apps/web/src/stores/thread-lifecycle.ts +++ b/apps/web/src/stores/thread-lifecycle.ts @@ -11,10 +11,15 @@ function latestCanonicalTurn(threadId: string, record: ThreadRecord): AgentTurn /** 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 = 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; } @@ -40,7 +45,8 @@ export function getCanonicalRuntimeTurn(threadId: string, record: ThreadRecord): return latest; } -function phaseForTurnStatus(status: AgentTurnStatus): TurnRuntimePhase { +/** Runtime phase equivalent of one canonical turn status. */ +export function phaseForTurnStatus(status: AgentTurnStatus): TurnRuntimePhase { switch (status) { case "Pending": case "Running": return "running"; diff --git a/apps/web/src/stores/threadStore.ts b/apps/web/src/stores/threadStore.ts index d10ef1e9e..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 { getCanonicalRuntimePhase, isThreadRuntimeActive } from "./thread-lifecycle"; +import { getCanonicalRuntimeTurn, isThreadRuntimeActive, phaseForTurnStatus } from "./thread-lifecycle"; function deriveRunningThreadIds(records: Map): Set { return new Set( @@ -2722,11 +2722,20 @@ export const useThreadStore = create((zustandSet, get) => { threadId: string, ): { records: Map; runningThreadIds: Set } => { const record = getThreadRecord(records, threadId); - const phase = getCanonicalRuntimePhase(threadId, record); - if (phase === null) return { records, runningThreadIds }; + const turn = getCanonicalRuntimeTurn(threadId, record); + if (!turn) return { records, runningThreadIds }; + const phase = phaseForTurnStatus(turn.status); let nextRecords = records; - if (record.runtimePhase !== phase) { - nextRecords = patchThreadRecord(records, threadId, { runtimePhase: phase }); + 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) {