From 263e31a25a7333827b8507a7ffab4ae6a9739f0e Mon Sep 17 00:00:00 2001 From: Serhii Vecherenko Date: Fri, 14 Aug 2026 20:56:20 -0700 Subject: [PATCH] fix(threads): resume prompts after missing sessions - Wait for in-flight launches before delivering input - Relaunch missing sessions from the latest thread state - Settle unknown interrupts and emit inactive state on close - Share missing-session detection with MCP tools - Add renderer resume coverage and update session tests --- src/main/app-controls/mcp/tools/threads.ts | 8 +- .../threadRuntimeActions.resume.test.ts | 210 ++++++++++++++++++ src/renderer/actions/threadRuntimeActions.ts | 68 +++++- src/shared/threadRelaunch.ts | 9 + .../threadSessionManager.startClose.test.ts | 112 +++++++++- .../runtime/threadSessionManager.ts | 36 ++- 6 files changed, 418 insertions(+), 25 deletions(-) create mode 100644 src/renderer/actions/threadRuntimeActions.resume.test.ts diff --git a/src/main/app-controls/mcp/tools/threads.ts b/src/main/app-controls/mcp/tools/threads.ts index b39e02aa3..887d35102 100644 --- a/src/main/app-controls/mcp/tools/threads.ts +++ b/src/main/app-controls/mcp/tools/threads.ts @@ -14,6 +14,7 @@ import { DEFAULT_TERMINAL_SIZE, resolveMcpLaunchSnapshot, } from "@/shared/contracts"; +import { isUnknownThreadSessionError } from "@/shared/threadRelaunch"; import { buildWorktreeLocation, normalizeWorktreePathForComparison } from "@/shared/worktree"; import { dbGetThreadRuntimeItemsPage } from "../../../db"; import { @@ -418,7 +419,7 @@ export const threadTools: ToolDomain = { await ctx.supervisor.sendThreadInput({ threadId, prompt: message, config: thread.config }); return { threadId, delivered: true, interruptedFirst: interruptFirst === true }; } catch (error) { - if (!isUnknownSessionError(error)) throw error; + if (!isUnknownThreadSessionError(error)) throw error; } // No live session — resume the thread the same way the app revives an // inactive thread (startThread with the persisted config + sessionRef), @@ -615,11 +616,6 @@ export const threadTools: ToolDomain = { }, }; -/** True when the supervisor rejected a call because the thread has no live session. */ -function isUnknownSessionError(error: unknown): boolean { - return error instanceof Error && /unknown thread session/i.test(error.message); -} - /** * Build the `startThread` payload that resumes an inactive thread, mirroring the * app's own resume path (`performInitialThreadLaunch` / `createAppThread`): the diff --git a/src/renderer/actions/threadRuntimeActions.resume.test.ts b/src/renderer/actions/threadRuntimeActions.resume.test.ts new file mode 100644 index 000000000..9a1f8e658 --- /dev/null +++ b/src/renderer/actions/threadRuntimeActions.resume.test.ts @@ -0,0 +1,210 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import type { Project, PromptSegment, SendThreadInputPayload, Thread } from "@/shared/contracts"; + +const mocks = vi.hoisted(() => ({ + appState: { + threads: [] as Thread[], + projects: [] as Project[], + applyRuntimeEvent: vi.fn<(threadId: string, event: unknown) => void>(), + updateThreadRuntime: vi.fn<(threadId: string, input: { status: string }) => void>(), + touchThread: vi.fn<(threadId: string) => void>(), + }, + bridge: { + sendThreadInput: vi.fn<(payload: SendThreadInputPayload) => Promise>(), + }, + performInitialThreadLaunch: vi.fn<(input: unknown) => Promise>(), +})); + +vi.mock("@/renderer/state/appStore", () => ({ + useAppStore: { getState: () => mocks.appState }, +})); +vi.mock("@/renderer/bridge", () => ({ + readBridge: () => mocks.bridge, +})); +vi.mock("@/renderer/state/remoteProjection", () => ({ + remoteOwner: () => undefined, +})); +vi.mock("@/renderer/state/fileCheckpointActions", () => ({ + captureFileCheckpoint: vi.fn<(input: unknown) => Promise>(), +})); +vi.mock("@/renderer/analytics/posthog", () => ({ + captureThreadPromptSubmitted: vi.fn<(...args: unknown[]) => void>(), + threadProductProperties: () => ({}), +})); +vi.mock("@/renderer/analytics/productAnalytics", () => ({ + captureProductEvent: vi.fn<(...args: unknown[]) => void>(), +})); +vi.mock("./threadLaunchActions", () => ({ + performInitialThreadLaunch: mocks.performInitialThreadLaunch, +})); + +import { performThreadInputSubmit, submitThreadInput } from "./threadRuntimeActions"; + +const project: Project = { + id: "project-1", + name: "Repo", + location: { kind: "posix", path: "/repo" }, + scripts: { actions: [] }, + createdAt: "2026-01-01T00:00:00.000Z", +}; + +function createThread(overrides: Partial = {}): Thread { + return { + id: "thread-1", + projectId: project.id, + title: "Thread", + agentKind: "codex", + config: { model: "codex/model" }, + status: "idle", + attention: "none", + canResumeWithConfig: true, + sessionRef: { providerSessionId: "ses_1", discoveredAt: "2026-01-01T00:00:00.000Z" }, + archived: false, + done: false, + starred: false, + presentationMode: "gui", + createdAt: "2026-01-01T00:00:00.000Z", + updatedAt: "2026-01-01T00:00:00.000Z", + ...overrides, + } as Thread; +} + +const segments: PromptSegment[] = [{ kind: "text", content: "hello" }]; + +function rejectingTransport(message: string) { + return { + sendThreadInput: vi.fn<() => Promise>(() => Promise.reject(new Error(message))), + }; +} + +/** The rollback write restores the pre-submit status; the optimistic one sets "working". */ +function rollbackCalls(): unknown[] { + return mocks.appState.updateThreadRuntime.mock.calls.filter( + ([, input]) => input.status !== "working", + ); +} + +describe("performThreadInputSubmit unknown-session resume", () => { + beforeEach(() => { + vi.clearAllMocks(); + mocks.bridge.sendThreadInput.mockResolvedValue(undefined); + mocks.performInitialThreadLaunch.mockResolvedValue(undefined); + mocks.appState.threads = []; + mocks.appState.projects = [project]; + }); + + it("resumes the thread instead of dropping the prompt when the session is gone", async () => { + const thread = createThread(); + const resumeLaunch = vi.fn<(args: unknown) => Promise>().mockResolvedValue(undefined); + + await expect( + performThreadInputSubmit({ + thread, + prompt: "hello", + segments, + transport: rejectingTransport("Unknown thread session: x"), + resumeLaunch, + }), + ).resolves.toBeUndefined(); + + expect(resumeLaunch).toHaveBeenCalledExactlyOnceWith({ + prompt: "hello", + segments, + userMessageItemId: expect.stringMatching(/^user-/), + }); + expect(rollbackCalls()).toEqual([]); + }); + + it("rolls back and rejects when the resume launch itself fails", async () => { + const thread = createThread(); + + await expect( + performThreadInputSubmit({ + thread, + prompt: "hello", + transport: rejectingTransport("Unknown thread session: x"), + resumeLaunch: () => Promise.reject(new Error("relaunch failed")), + }), + ).rejects.toThrow("relaunch failed"); + + expect(rollbackCalls()).toHaveLength(1); + }); + + it("rolls back and rejects for any other transport error", async () => { + const thread = createThread(); + const resumeLaunch = vi.fn<(args: unknown) => Promise>().mockResolvedValue(undefined); + + await expect( + performThreadInputSubmit({ + thread, + prompt: "hello", + transport: rejectingTransport("boom"), + resumeLaunch, + }), + ).rejects.toThrow("boom"); + + expect(resumeLaunch).not.toHaveBeenCalled(); + expect(rollbackCalls()).toHaveLength(1); + }); + + it("keeps the old failure behavior without a resume hook or a resumable thread", async () => { + const thread = createThread(); + await expect( + performThreadInputSubmit({ + thread, + prompt: "hello", + transport: rejectingTransport("Unknown thread session: x"), + }), + ).rejects.toThrow("Unknown thread session: x"); + expect(rollbackCalls()).toHaveLength(1); + + vi.clearAllMocks(); + const resumeLaunch = vi.fn<(args: unknown) => Promise>().mockResolvedValue(undefined); + await expect( + performThreadInputSubmit({ + thread: createThread({ canResumeWithConfig: false, sessionRef: undefined }), + prompt: "hello", + transport: rejectingTransport("Unknown thread session: x"), + resumeLaunch, + }), + ).rejects.toThrow("Unknown thread session: x"); + expect(resumeLaunch).not.toHaveBeenCalled(); + expect(rollbackCalls()).toHaveLength(1); + }); +}); + +describe("submitThreadInput resume wiring", () => { + beforeEach(() => { + vi.clearAllMocks(); + mocks.performInitialThreadLaunch.mockResolvedValue(undefined); + mocks.appState.projects = [project]; + mocks.appState.threads = [createThread()]; + }); + + it("relaunches with the freshest thread snapshot and the optimistic item id", async () => { + mocks.bridge.sendThreadInput.mockRejectedValueOnce( + new Error("Unknown thread session: thread-1"), + ); + // The store snapshot moved on since the submit started; the relaunch must + // carry the newly discovered session ref, not the stale one. + mocks.appState.threads = [ + createThread({ + sessionRef: { providerSessionId: "ses_2", discoveredAt: "2026-01-02T00:00:00.000Z" }, + }), + ]; + + await expect(submitThreadInput("thread-1", "hello", segments)).resolves.toBeUndefined(); + + expect(mocks.performInitialThreadLaunch).toHaveBeenCalledExactlyOnceWith({ + thread: expect.objectContaining({ + sessionRef: expect.objectContaining({ providerSessionId: "ses_2" }), + }), + projectLocation: { kind: "posix", path: "/repo" }, + prompt: "hello", + segments, + userMessageItemId: expect.stringMatching(/^user-/), + initialSize: { cols: 120, rows: 30 }, + }); + expect(rollbackCalls()).toEqual([]); + }); +}); diff --git a/src/renderer/actions/threadRuntimeActions.ts b/src/renderer/actions/threadRuntimeActions.ts index 69bd5bfef..66318fccf 100644 --- a/src/renderer/actions/threadRuntimeActions.ts +++ b/src/renderer/actions/threadRuntimeActions.ts @@ -8,8 +8,10 @@ import type { ThreadServerRequestId, } from "@/shared/contracts"; import { toast } from "@heroui/react"; +import { DEFAULT_TERMINAL_SIZE } from "@/shared/contracts"; import { isHomeProjectId } from "@/shared/homeScope"; import { friendlyError } from "@/shared/messages"; +import { isUnknownThreadSessionError } from "@/shared/threadRelaunch"; import { resolveProjectLocation } from "@/shared/worktree"; import { buildPromptContentBlocks } from "@/shared/promptContent"; import { readBridge } from "@/renderer/bridge"; @@ -21,6 +23,7 @@ import { captureProductEvent } from "@/renderer/analytics/productAnalytics"; import { useAppStore } from "@/renderer/state/appStore"; import { captureFileCheckpoint } from "@/renderer/state/fileCheckpointActions"; import { remoteOwner } from "@/renderer/state/remoteProjection"; +import { performInitialThreadLaunch } from "./threadLaunchActions"; /** Resolve a thread and its on-disk project location from the store. */ function resolveThreadProjectLocation( @@ -58,6 +61,16 @@ export async function performThreadInputSubmit(input: { transport: ThreadInputTransport; /** Desktop-only: capture a file checkpoint keyed to the optimistic user message. */ captureCheckpoint?: (checkpointItemId: string) => Promise; + /** + * Relaunch the thread and deliver this prompt as the resumed session's first + * input. Called only when the host has no session left for a thread that is + * still resumable, so the prompt is never dropped. + */ + resumeLaunch?: (args: { + prompt: string; + segments?: PromptSegment[]; + userMessageItemId?: string; + }) => Promise; }): Promise { const { thread, prompt, segments, transport } = input; @@ -95,6 +108,16 @@ export async function performThreadInputSubmit(input: { await input.captureCheckpoint(optimisticUserMessageItemId); } } + const rollbackOptimisticWorking = (): void => { + if (!markedWorking) return; + store.updateThreadRuntime(thread.id, { + status: thread.status, + attention: thread.attention, + canResumeWithConfig: thread.canResumeWithConfig, + forceCloseActiveTurn: true, + ...(thread.sessionRef ? { sessionRef: thread.sessionRef } : {}), + }); + }; try { await transport.sendThreadInput({ threadId: thread.id, @@ -104,15 +127,31 @@ export async function performThreadInputSubmit(input: { ...(optimisticUserMessageItemId ? { userMessageItemId: optimisticUserMessageItemId } : {}), }); } catch (error) { - if (markedWorking) { - store.updateThreadRuntime(thread.id, { - status: thread.status, - attention: thread.attention, - canResumeWithConfig: thread.canResumeWithConfig, - forceCloseActiveTurn: true, - ...(thread.sessionRef ? { sessionRef: thread.sessionRef } : {}), - }); + // The host session is gone (thread unloaded, supervisor restarted) but the + // thread can still be resumed: relaunch it with this prompt instead of + // dropping it. The optimistic paint stays — the relaunch reuses its item id. + if ( + input.resumeLaunch && + isUnknownThreadSessionError(error) && + (thread.sessionRef || thread.canResumeWithConfig) + ) { + try { + await input.resumeLaunch({ + prompt, + ...(segments ? { segments } : {}), + ...(optimisticUserMessageItemId + ? { userMessageItemId: optimisticUserMessageItemId } + : {}), + }); + } catch (resumeError) { + rollbackOptimisticWorking(); + throw resumeError; + } + // The relaunch captures its own prompt-submitted event. + store.touchThread(thread.id); + return; } + rollbackOptimisticWorking(); throw error; } captureThreadPromptSubmitted(thread, prompt, segments); @@ -140,6 +179,19 @@ export async function submitThreadInput( prompt, ...(segments ? { segments } : {}), transport: readBridge(), + resumeLaunch: async (resume) => { + // Re-resolve the thread: the pre-send snapshot can miss a sessionRef + // discovered since, and the resume payload must carry the latest one. + const latest = resolveThreadProjectLocation(threadId); + await performInitialThreadLaunch({ + thread: latest?.thread ?? thread, + projectLocation: latest?.projectLocation ?? projectLocation, + prompt: resume.prompt, + ...(resume.segments ? { segments: resume.segments } : {}), + ...(resume.userMessageItemId ? { userMessageItemId: resume.userMessageItemId } : {}), + initialSize: DEFAULT_TERMINAL_SIZE, + }); + }, ...(!owner ? { captureCheckpoint: async (checkpointItemId: string) => { diff --git a/src/shared/threadRelaunch.ts b/src/shared/threadRelaunch.ts index d50b3b5d5..0f691631f 100644 --- a/src/shared/threadRelaunch.ts +++ b/src/shared/threadRelaunch.ts @@ -1,6 +1,15 @@ import type { ProjectLocation, TerminalSize, Thread } from "./contracts"; import type { StartRemoteThreadInput } from "./remote/client"; +/** + * True when a supervisor call was rejected because the thread has no live + * session. Every caller that can revive a thread (the renderer composer, the + * `send_to_thread` MCP tool) branches on this to resume instead of failing. + */ +export function isUnknownThreadSessionError(error: unknown): boolean { + return error instanceof Error && /unknown thread session/i.test(error.message); +} + /** * Reopening a thread on its host relaunches it with an empty prompt. Only an * INACTIVE thread qualifies: every other status means the host session is diff --git a/src/supervisor/runtime/threadSessionManager.startClose.test.ts b/src/supervisor/runtime/threadSessionManager.startClose.test.ts index 840060896..5295000cd 100644 --- a/src/supervisor/runtime/threadSessionManager.startClose.test.ts +++ b/src/supervisor/runtime/threadSessionManager.startClose.test.ts @@ -3,6 +3,7 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import { afterEach, describe, expect, it, vi } from "vitest"; import type { AgentKind } from "@/shared/contracts"; +import type { SupervisorEvent } from "@/shared/ipc"; import type { AgentAdapter, StructuredSessionHandle } from "../agents/base"; import type { SessionRuntime } from "./sessionTypes"; @@ -69,11 +70,15 @@ function deferred(): { return { promise, resolve, reject }; } -function createManager(agentKind: AgentKind, adapter: AgentAdapter): ThreadSessionManager { +function createManager( + agentKind: AgentKind, + adapter: AgentAdapter, + emit: (event: SupervisorEvent) => void = vi.fn<(event: SupervisorEvent) => void>(), +): ThreadSessionManager { const tempDir = mkdtempSync(join(tmpdir(), "poracode-start-close-")); tempDirs.push(tempDir); const manager = new ThreadSessionManager({ - emit: vi.fn<() => void>(), + emit, isDev: false, logsDir: join(tempDir, "logs"), settingsPath: join(tempDir, "settings.json"), @@ -258,7 +263,31 @@ describe("ThreadSessionManager start guards", () => { ); }); - it("treats late input, write, and interrupt IPC after known removal as idempotent", async () => { + it("settles a closed working session so consumers never freeze at working", async () => { + const structuredSession = createStructuredSession(Promise.resolve()); + const adapter = createAdapter("codex", structuredSession); + const emit = vi.fn<(event: SupervisorEvent) => void>(); + const manager = createManager("codex", adapter, emit); + const runtime = createInactiveRuntime("codex", adapter, structuredSession); + runtime.threadId = "closed-thread"; + runtime.status = "working"; + runtime.attention = "working"; + manager.sessions.set(runtime.threadId, runtime); + + await manager.closeThread({ threadId: runtime.threadId }); + + expect(emit).toHaveBeenCalledWith( + expect.objectContaining({ + type: "thread-state", + threadId: "closed-thread", + status: "inactive", + attention: "none", + forceCloseActiveTurn: true, + }), + ); + }); + + it("rejects a prompt for a closed session instead of dropping it", async () => { const structuredSession = createStructuredSession(Promise.resolve()); const adapter = createAdapter("codex", structuredSession); const manager = createManager("codex", adapter); @@ -273,17 +302,84 @@ describe("ThreadSessionManager start guards", () => { prompt: "late", config: { model: "codex/model" }, }), - ).resolves.toBeUndefined(); + ).rejects.toThrow("Unknown thread session: closed-thread"); + // Raw keystrokes racing a close stay idempotent — only prompts must survive. await expect( manager.writeTerminal({ threadId: "closed-thread", data: "late" }), ).resolves.toBeUndefined(); + }); + + it.each(guardedStructuredProviders)( + "delivers a prompt sent while a %s start is still in flight", + async (agentKind) => { + const activation = deferred(); + const activationStarted = deferred(); + const structuredSession: StructuredSessionHandle = { + ...createStructuredSession(activation.promise, () => activationStarted.resolve()), + startTurn: vi.fn>( + async () => undefined, + ), + }; + const adapter = createAdapter(agentKind, structuredSession); + const manager = createManager(agentKind, adapter); + + const start = manager.startThread({ + threadId: `thread-${agentKind}`, + projectLocation: { kind: "windows", path: "C:\\repo" }, + agentKind, + config: { model: `${agentKind}/model` }, + prompt: "", + initialSize: { cols: 80, rows: 24 }, + presentationMode: "gui", + }); + await activationStarted.promise; + + // The session only lands in the map when the start settles; a prompt + // typed during the spawn must wait for it, not fail as unknown. + const send = manager.sendThreadInput({ + threadId: `thread-${agentKind}`, + prompt: "queued while starting", + config: { model: `${agentKind}/model` }, + }); + activation.resolve(); + await start; + await expect(send).resolves.toBeUndefined(); + expect(structuredSession.startTurn).toHaveBeenCalledWith( + "queued while starting", + expect.objectContaining({ model: `${agentKind}/model` }), + undefined, + expect.objectContaining({ userMessageItemId: expect.any(String) }), + ); + }, + ); + + it("recovers a closed thread's state on interrupt", async () => { + const structuredSession = createStructuredSession(Promise.resolve()); + const adapter = createAdapter("codex", structuredSession); + const emit = vi.fn<(event: SupervisorEvent) => void>(); + const manager = createManager("codex", adapter, emit); + const runtime = createInactiveRuntime("codex", adapter, structuredSession); + runtime.threadId = "closed-thread"; + manager.sessions.set(runtime.threadId, runtime); + await manager.closeThread({ threadId: runtime.threadId }); + emit.mockClear(); + await expect(manager.interruptThread({ threadId: "closed-thread" })).resolves.toBeUndefined(); + expect(emit).toHaveBeenCalledWith({ + type: "thread-state", + threadId: "closed-thread", + status: "inactive", + attention: "none", + canResumeWithConfig: false, + forceCloseActiveTurn: true, + }); }); it("preserves bookkeeping errors for never-known session ids", async () => { const structuredSession = createStructuredSession(Promise.resolve()); const adapter = createAdapter("codex", structuredSession); - const manager = createManager("codex", adapter); + const emit = vi.fn<(event: SupervisorEvent) => void>(); + const manager = createManager("codex", adapter, emit); await expect( manager.sendThreadInput({ @@ -295,8 +391,10 @@ describe("ThreadSessionManager start guards", () => { await expect(manager.writeTerminal({ threadId: "never-known", data: "late" })).rejects.toThrow( "Unknown thread session: never-known", ); - await expect(manager.interruptThread({ threadId: "never-known" })).rejects.toThrow( - "Unknown thread session: never-known", + // Interrupt is idempotent "ensure not running", so it settles rather than throws. + await expect(manager.interruptThread({ threadId: "never-known" })).resolves.toBeUndefined(); + expect(emit).toHaveBeenCalledWith( + expect.objectContaining({ threadId: "never-known", status: "inactive" }), ); }); diff --git a/src/supervisor/runtime/threadSessionManager.ts b/src/supervisor/runtime/threadSessionManager.ts index 465ba241a..58176734f 100644 --- a/src/supervisor/runtime/threadSessionManager.ts +++ b/src/supervisor/runtime/threadSessionManager.ts @@ -504,9 +504,20 @@ export class ThreadSessionManager { } async sendThreadInput(payload: SendThreadInputPayload): Promise { - const session = this.sessions.get(payload.threadId); + let session = this.sessions.get(payload.threadId); if (!session) { - if (this.recentlyRemovedThreadIds.has(payload.threadId)) return; + // A launch/resume can be mid-flight for seconds while the agent process + // spawns; the session only lands in the map at the end. Wait for it so a + // prompt typed during the spawn is delivered instead of failing. + const pendingStart = this.startLocks.get(payload.threadId); + if (pendingStart) { + await pendingStart; + session = this.sessions.get(payload.threadId); + } + } + if (!session) { + // Never swallow a full user prompt, even for a just-removed thread — + // callers (renderer composer, `send_to_thread`) resume on this error. throw new Error(`Unknown thread session: ${payload.threadId}`); } if (session.status === "inactive" && !session.sessionRef) { @@ -642,8 +653,19 @@ export class ThreadSessionManager { }); return; } - if (this.recentlyRemovedThreadIds.has(payload.threadId)) return; - throw new Error(`Unknown thread session: ${payload.threadId}`); + // Interrupt is idempotent "ensure this thread is not running". With no + // session there is nothing to stop, so emit the settled state instead of + // failing — this is the lever that unsticks a row whose session died + // while its persisted status still says `working`. + this.options.emit({ + type: "thread-state", + threadId: payload.threadId, + status: "inactive", + attention: "none", + canResumeWithConfig: false, + forceCloseActiveTurn: true, + }); + return; } this.options.crossagentMcp?.cancelForeground(payload.threadId); await this.structuredInterruptWatchdog.interruptStructuredTurn(session); @@ -906,6 +928,12 @@ export class ThreadSessionManager { this.outputPipeline.clearSessionTimers(existing); existing.stopSessionRefWatcher?.(); existing.stopSessionRefWatcher = undefined; + // Final state before the session disappears: without it a working thread + // freezes at `working` in the DB and in every renderer, with nothing left + // running to ever move it. + this.outputPipeline.updateState(existing, "inactive", "none", undefined, { + forceCloseActiveTurn: true, + }); this.sessions.delete(payload.threadId); if (existing.sessionRef?.providerSessionId) { this.sessionsBySessionId.delete(existing.sessionRef.providerSessionId);