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/4] 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/4] 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/4] 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 12ea4b4f77bcf12df779fade5da6ad8df4e24156 Mon Sep 17 00:00:00 2001 From: Agent Runtime Fixture Date: Tue, 29 Sep 2026 21:15:07 +0100 Subject: [PATCH 4/4] 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 19f5bf881..d8db3de92 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( @@ -2724,11 +2724,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) {