From af1ddda3186cfdf052f86acb51c0fe66f242f453 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Tue, 16 Jun 2026 22:20:49 -0700 Subject: [PATCH] FN-6511: bound planning and subtask generation timeouts Ensure stalled planning and subtask agent setup reaches terminal error states instead of hanging sessions.\n\n- Pass abort signals through GenerationGuard-managed operations.\n- Wrap planning and subtask agent construction in generation timeouts and dispose aborted agents.\n- Add regression coverage for stalled agent construction, stalled prompts, SSE error replay, and retry behavior.\n\nFiles changed:\n .../src/__tests__/session-error-recovery.test.ts | 83 ++++++++++\n .../src/__tests__/subtask-breakdown.test.ts | 171 ++++++++++++++++++++-\n packages/dashboard/src/ai-session-timeout.ts | 4 +-\n packages/dashboard/src/planning.ts | 56 +++++--\n packages/dashboard/src/subtask-breakdown.ts | 101 ++++++++----\n 5 files changed, 366 insertions(+), 49 deletions(-) Fusion-Task-Id: FN-6511 Fusion-Task-Lineage: a1727b80-23f7-4dfd-8ac1-cafd540a6abb --- .../__tests__/session-error-recovery.test.ts | 83 +++++++++ .../src/__tests__/subtask-breakdown.test.ts | 171 +++++++++++++++++- packages/dashboard/src/ai-session-timeout.ts | 4 +- packages/dashboard/src/planning.ts | 56 ++++-- packages/dashboard/src/subtask-breakdown.ts | 101 +++++++---- 5 files changed, 366 insertions(+), 49 deletions(-) diff --git a/packages/dashboard/src/__tests__/session-error-recovery.test.ts b/packages/dashboard/src/__tests__/session-error-recovery.test.ts index 78f6c4c28c..afa7eded87 100644 --- a/packages/dashboard/src/__tests__/session-error-recovery.test.ts +++ b/packages/dashboard/src/__tests__/session-error-recovery.test.ts @@ -14,8 +14,11 @@ import { Database, TaskStore } from "@fusion/core"; import { AiSessionStore } from "../ai-session-store.js"; import { __resetPlanningState, + __getActiveGenerationForTests, __setCreateFnAgent, createSession, + createSessionWithAgent, + GENERATION_TIMEOUT_MS as PLANNING_GENERATION_TIMEOUT_MS, getSession, planningStreamManager, retrySession, @@ -106,6 +109,7 @@ describe("session error recovery", () => { }); afterEach(async () => { + vi.useRealTimers(); __setCreateFnAgent(undefined as any); __resetPlanningState(); __resetSubtaskBreakdownState(); @@ -189,6 +193,85 @@ describe("session error recovery", () => { unsubscribeError(); }); + it("times out planning sessions when createFnAgent construction stalls", async () => { + vi.useFakeTimers(); + + __setCreateFnAgent(async () => { + await new Promise(() => undefined); + }); + + const sessionId = await createSessionWithAgent( + "127.0.0.150", + "Planning construction stall", + "/tmp/project", + taskStore, + ); + const errorEvents: string[] = []; + const unsubscribe = planningStreamManager.subscribe(sessionId, (event) => { + if (event.type === "error") { + errorEvents.push(String(event.data)); + } + }); + + planningStreamManager.consumeInitialTurn(sessionId)?.(); + await vi.advanceTimersByTimeAsync(0); + expect(aiSessionStore.get(sessionId)?.status).toBe("generating"); + expect(__getActiveGenerationForTests(sessionId)).toBeDefined(); + + await vi.advanceTimersByTimeAsync(PLANNING_GENERATION_TIMEOUT_MS); + await vi.advanceTimersByTimeAsync(0); + + expect(aiSessionStore.get(sessionId)?.status).toBe("error"); + expect(aiSessionStore.get(sessionId)?.error).toMatch(/timed out/i); + expect(errorEvents).toContainEqual(expect.stringMatching(/timed out/i)); + expect(__getActiveGenerationForTests(sessionId)).toBeUndefined(); + + unsubscribe(); + }); + + it("times out planning sessions when prompt stalls and disposes the agent", async () => { + vi.useFakeTimers(); + + const dispose = vi.fn(); + __setCreateFnAgent(async () => ({ + session: { + state: { messages: [] }, + prompt: vi.fn(async () => { + await new Promise(() => undefined); + }), + dispose, + }, + })); + + const sessionId = await createSessionWithAgent( + "127.0.0.151", + "Planning prompt stall", + "/tmp/project", + taskStore, + ); + const errorEvents: string[] = []; + const unsubscribe = planningStreamManager.subscribe(sessionId, (event) => { + if (event.type === "error") { + errorEvents.push(String(event.data)); + } + }); + + planningStreamManager.consumeInitialTurn(sessionId)?.(); + await vi.advanceTimersByTimeAsync(0); + expect(__getActiveGenerationForTests(sessionId)).toBeDefined(); + + await vi.advanceTimersByTimeAsync(PLANNING_GENERATION_TIMEOUT_MS); + await vi.advanceTimersByTimeAsync(0); + + expect(aiSessionStore.get(sessionId)?.status).toBe("error"); + expect(aiSessionStore.get(sessionId)?.error).toMatch(/timed out/i); + expect(errorEvents).toContainEqual(expect.stringMatching(/timed out/i)); + expect(__getActiveGenerationForTests(sessionId)).toBeUndefined(); + expect(dispose).toHaveBeenCalled(); + + unsubscribe(); + }); + it("captures subtask generation errors, broadcasts SSE error, and retries to completion", async () => { const subtaskErrors: string[] = []; diff --git a/packages/dashboard/src/__tests__/subtask-breakdown.test.ts b/packages/dashboard/src/__tests__/subtask-breakdown.test.ts index de9f04edf8..181e50d3ce 100644 --- a/packages/dashboard/src/__tests__/subtask-breakdown.test.ts +++ b/packages/dashboard/src/__tests__/subtask-breakdown.test.ts @@ -33,8 +33,10 @@ import { InvalidSessionStateError, setAiSessionStore, stopSubtaskGeneration, + subtaskStreamManager, SubtaskStreamManager, GENERATION_TIMEOUT_MS, + generationGuard, } from "../subtask-breakdown.js"; const UUID_REGEX = @@ -921,10 +923,43 @@ describe("SessionNotFoundError", () => { }); describe("subtask generation timeout / abort", () => { - it("marks the session as error and stops the prompt() promise when generation exceeds GENERATION_TIMEOUT_MS", async () => { + it("marks the session as error and broadcasts a terminal error when createFnAgent construction stalls", async () => { + vi.useFakeTimers(); + + mockCreateFnAgent.mockImplementation(async () => { + await new Promise(() => undefined); + }); + + const created = await createSubtaskSession( + "Hung subtask agent construction", + undefined, + "/tmp/project", + ); + const events: Array<{ type: string; data?: unknown }> = []; + const unsubscribe = subtaskStreamManager.subscribe(created.sessionId, (event) => { + events.push(event); + }); + + await vi.advanceTimersByTimeAsync(0); + expect(getSubtaskSession(created.sessionId)?.status).toBe("generating"); + + await vi.advanceTimersByTimeAsync(GENERATION_TIMEOUT_MS); + await vi.advanceTimersByTimeAsync(0); + + const after = getSubtaskSession(created.sessionId); + expect(after?.status).toBe("error"); + expect(after?.error).toMatch(/timed out/i); + expect(events).toContainEqual(expect.objectContaining({ type: "error", data: expect.stringMatching(/timed out/i) })); + + unsubscribe(); + vi.useRealTimers(); + }); + + it("marks the session as error and broadcasts a terminal error when prompt() stalls", async () => { vi.useFakeTimers(); let resolveHungPrompt: (() => void) | undefined; + const dispose = vi.fn(); const hungPromptCallable = vi.fn(async () => { // Simulate a stalled provider stream that never terminates on its own. await new Promise((resolve) => { resolveHungPrompt = resolve; }); @@ -935,7 +970,7 @@ describe("subtask generation timeout / abort", () => { session: { state: { messages: [] }, prompt: hungPromptCallable, - dispose: vi.fn(), + dispose, }, })); @@ -944,6 +979,10 @@ describe("subtask generation timeout / abort", () => { undefined, "/tmp/project", ); + const events: Array<{ type: string; data?: unknown }> = []; + const unsubscribe = subtaskStreamManager.subscribe(created.sessionId, (event) => { + events.push(event); + }); // Yield once so startSubtaskGeneration's microtasks run before we advance time. await Promise.resolve(); @@ -956,11 +995,139 @@ describe("subtask generation timeout / abort", () => { const after = getSubtaskSession(created.sessionId); expect(after?.status).toBe("error"); expect(after?.error).toMatch(/timed out/i); + expect(events).toContainEqual(expect.objectContaining({ type: "error", data: expect.stringMatching(/timed out/i) })); // The hung prompt is still pending; release it so its microtask completes. resolveHungPrompt?.(); await vi.advanceTimersByTimeAsync(0); + expect(dispose).toHaveBeenCalled(); + unsubscribe(); + vi.useRealTimers(); + }); + + it("keeps other subtask sessions responsive while one agent construction is stalled", async () => { + vi.useFakeTimers(); + + mockCreateFnAgent + .mockImplementationOnce(async () => { + await new Promise(() => undefined); + }) + .mockImplementationOnce(async () => createMockSubtaskAgent( + JSON.stringify({ + subtasks: [ + { + id: "subtask-responsive", + title: "Responsive session completed", + description: "The second session should not wait for the first hung construction.", + suggestedSize: "S", + dependsOn: [], + }, + ], + }), + )); + + const stalled = await createSubtaskSession( + "Hung construction should not monopolize generation", + undefined, + "/tmp/project", + ); + await vi.advanceTimersByTimeAsync(0); + expect(getSubtaskSession(stalled.sessionId)?.status).toBe("generating"); + expect(generationGuard.has(stalled.sessionId)).toBe(true); + + const responsive = await createSubtaskSession( + "Second subtask session should complete", + undefined, + "/tmp/project", + ); + await vi.advanceTimersByTimeAsync(0); + + const responsiveSession = getSubtaskSession(responsive.sessionId); + expect(responsiveSession?.status).toBe("complete"); + expect(responsiveSession?.subtasks).toEqual([ + expect.objectContaining({ id: "subtask-responsive" }), + ]); + expect(generationGuard.has(responsive.sessionId)).toBe(false); + + await vi.advanceTimersByTimeAsync(GENERATION_TIMEOUT_MS); + await vi.advanceTimersByTimeAsync(0); + + expect(getSubtaskSession(stalled.sessionId)?.status).toBe("error"); + expect(generationGuard.has(stalled.sessionId)).toBe(false); + + vi.useRealTimers(); + }); + + it("replays terminal timeout errors to late subscribers and still allows retry", async () => { + vi.useFakeTimers(); + + mockCreateFnAgent.mockImplementationOnce(async () => { + await new Promise(() => undefined); + }); + + const created = await createSubtaskSession( + "Hung construction with late subscriber", + undefined, + "/tmp/project", + ); + + await vi.advanceTimersByTimeAsync(0); + await vi.advanceTimersByTimeAsync(GENERATION_TIMEOUT_MS); + await vi.advanceTimersByTimeAsync(0); + + const errored = getSubtaskSession(created.sessionId); + expect(errored?.status).toBe("error"); + expect(errored?.error).toMatch(/timed out/i); + expect(generationGuard.has(created.sessionId)).toBe(false); + + const bufferedEvents = subtaskStreamManager.getBufferedEvents(created.sessionId, 0); + expect(bufferedEvents).toContainEqual( + expect.objectContaining({ + event: "error", + data: JSON.stringify(errored?.error), + }), + ); + + const errorEventId = bufferedEvents.find((event) => event.event === "error")?.id; + expect(errorEventId).toBeGreaterThan(0); + expect(subtaskStreamManager.getBufferedEvents(created.sessionId, (errorEventId ?? 0) - 1)).toContainEqual( + expect.objectContaining({ event: "error" }), + ); + expect(subtaskStreamManager.getBufferedEvents(created.sessionId, errorEventId ?? 0)).toEqual([]); + + const retryEvents: Array<{ type: string; data?: unknown }> = []; + const unsubscribe = subtaskStreamManager.subscribe(created.sessionId, (event) => { + retryEvents.push(event); + }); + + mockCreateFnAgent.mockImplementationOnce(async () => createMockSubtaskAgent( + JSON.stringify({ + subtasks: [ + { + id: "subtask-retry-success", + title: "Retry succeeds", + description: "Retry should use a fresh bounded generation after timeout.", + suggestedSize: "S", + dependsOn: [], + }, + ], + }), + )); + + await retrySubtaskSession(created.sessionId, "/tmp/project"); + await vi.advanceTimersByTimeAsync(0); + + const retried = getSubtaskSession(created.sessionId); + expect(retried?.status).toBe("complete"); + expect(retried?.subtasks).toEqual([ + expect.objectContaining({ id: "subtask-retry-success" }), + ]); + expect(retryEvents).toContainEqual(expect.objectContaining({ type: "subtasks" })); + expect(retryEvents).toContainEqual(expect.objectContaining({ type: "complete" })); + expect(generationGuard.has(created.sessionId)).toBe(false); + + unsubscribe(); vi.useRealTimers(); }); diff --git a/packages/dashboard/src/ai-session-timeout.ts b/packages/dashboard/src/ai-session-timeout.ts index e8c2eaf185..7e6c0ec3e6 100644 --- a/packages/dashboard/src/ai-session-timeout.ts +++ b/packages/dashboard/src/ai-session-timeout.ts @@ -42,7 +42,7 @@ export class GenerationGuard { sessionId: string, timeoutMs: number, handlers: TimeoutHandlers, - op: () => Promise, + op: (abortSignal: AbortSignal) => Promise, ): Promise { this.cancelInternal(sessionId, "displaced"); @@ -69,7 +69,7 @@ export class GenerationGuard { }); try { - return await Promise.race([op(), abortPromise]); + return await Promise.race([op(abort.signal), abortPromise]); } catch (err) { if (isAbortError(err)) { const cause = this.abortCause.get(abort) ?? "user-stop"; diff --git a/packages/dashboard/src/planning.ts b/packages/dashboard/src/planning.ts index 9ea875173e..5c7fa068b1 100644 --- a/packages/dashboard/src/planning.ts +++ b/packages/dashboard/src/planning.ts @@ -1341,21 +1341,54 @@ async function initializeAgent( customQuestionCount?: number, ): Promise { try { - session.agent = await createPlanningAgent( - session, - rootDir, - store, - modelProvider, - modelId, - promptOverrides, - planningDepth, - customQuestionCount, - ); - session.updatedAt = new Date(); + await runGenerationWithTimeout(session, async (abortSignal) => { + /* + FNXC:PlanningSession 2026-06-16-20:23: + FN-6511 requires planning agent construction to be bounded before the first prompt starts. Keep createFnAgent inside the active generation timeout so model-registry or extension-discovery stalls transition the SSE session to a terminal error instead of leaving it pinned in generating. + */ + const agentPromise = createPlanningAgent( + session, + rootDir, + store, + modelProvider, + modelId, + promptOverrides, + planningDepth, + customQuestionCount, + ); + + void agentPromise.then((lateAgent) => { + if (abortSignal.aborted) { + nonfatal( + () => lateAgent?.session?.dispose?.(), + diagnostics, + "Error disposing late-created planning agent", + { sessionId: session.id, operation: "dispose-late-agent" }, + ); + } + }, () => undefined); + + const agent = await agentPromise; + if (abortSignal.aborted) { + nonfatal( + () => agent?.session?.dispose?.(), + diagnostics, + "Error disposing aborted planning agent", + { sessionId: session.id, operation: "dispose-aborted-agent" }, + ); + throw createAbortError(); + } + session.agent = agent; + session.updatedAt = new Date(); + }); // Send initial message to get first question await continueAgentConversation(session, session.initialPlan); } catch (err) { + if (err instanceof Error && err.name === "AbortError") { + return; + } + const errorMessage = err instanceof Error ? err.message : "Failed to initialize AI agent"; diagnostics.errorFromException("Agent initialization error for session", err, { sessionId: session.id, operation: "initialize-agent" }); session.error = errorMessage; @@ -1561,6 +1594,7 @@ async function runGenerationWithTimeout(session: Session, operation: (abortSi const timer = setTimeout(() => { timeoutTriggered = true; setSessionError(session, "AI generation timed out. You can retry or start a new session."); + disposeSessionAgentForRetry(session); abortController.abort(); }, GENERATION_TIMEOUT_MS); const generationRecord = { abortController, timer }; diff --git a/packages/dashboard/src/subtask-breakdown.ts b/packages/dashboard/src/subtask-breakdown.ts index 3de9486e7a..23432711e7 100644 --- a/packages/dashboard/src/subtask-breakdown.ts +++ b/packages/dashboard/src/subtask-breakdown.ts @@ -9,7 +9,7 @@ import { createSessionDiagnostics, resetDiagnosticsSink, } from "./ai-session-diagnostics.js"; -import { GenerationGuard, isAbortError } from "./ai-session-timeout.js"; +import { GenerationGuard, createAbortError, isAbortError } from "./ai-session-timeout.js"; import { createFnAgent as engineCreateFnAgent } from "@fusion/engine"; @@ -88,13 +88,13 @@ const CLEANUP_INTERVAL_MS = 5 * 60 * 1000; */ export const GENERATION_TIMEOUT_MS = 90_000; -const generationGuard = new GenerationGuard(); +export const generationGuard = new GenerationGuard(); /** Minimal interface for the agent object created by createFnAgent */ interface SubtaskAgent { session: { dispose?: () => void; - prompt: (input: string) => Promise; + prompt: (input: string, options?: { signal?: AbortSignal }) => Promise; state: { messages: Array<{ role: string; content?: string | Array<{ type: string; text: string }> }> }; }; } @@ -452,42 +452,75 @@ async function generateSubtasks( const systemPrompt = resolvePrompt("subtask-breakdown-system", promptOverrides) || SUBTASK_BREAKDOWN_PROMPT; if (createFnAgent) { - const agent = await createFnAgent({ - cwd, - systemPrompt, - tools: "readonly", - onThinking: (delta: string) => { - const current = sessions.get(sessionId); - if (!current) return; - current.thinkingOutput += delta; - current.updatedAt = new Date(); - persistSubtaskThinking(sessionId, current.thinkingOutput); - subtaskStreamManager.broadcast(sessionId, { type: "thinking", data: delta }); - }, - onText: (delta: string) => { - const current = sessions.get(sessionId); - if (!current) return; - current.thinkingOutput += delta; - }, - }); - - session.agent = agent; - await generationGuard.run( sessionId, GENERATION_TIMEOUT_MS, { - onTimeout: () => setSubtaskError( - sessionId, - "AI generation timed out. You can retry or start a new session.", - ), - onUserStop: () => setSubtaskError( - sessionId, - "Generation stopped by user. You can retry or start a new session.", - ), + onTimeout: () => { + disposeSubtaskAgentForRetry(session); + setSubtaskError( + sessionId, + "AI generation timed out. You can retry or start a new session.", + ); + }, + onUserStop: () => { + disposeSubtaskAgentForRetry(session); + setSubtaskError( + sessionId, + "Generation stopped by user. You can retry or start a new session.", + ); + }, }, - async () => { - await agent.session.prompt(session.initialDescription); + async (abortSignal) => { + /* + FNXC:SubtaskBreakdown 2026-06-16-20:15: + FN-6511 requires the full subtask generation lifecycle to be timeout-bounded, including createFnAgent construction before prompt() starts. Keep construction and prompt inside one GenerationGuard entry so a model-registry or extension-discovery stall cannot pin the SSE session in generating forever. + */ + const agentPromise = createFnAgent({ + cwd, + systemPrompt, + tools: "readonly", + onThinking: (delta: string) => { + const current = sessions.get(sessionId); + if (!current) return; + current.thinkingOutput += delta; + current.updatedAt = new Date(); + persistSubtaskThinking(sessionId, current.thinkingOutput); + subtaskStreamManager.broadcast(sessionId, { type: "thinking", data: delta }); + }, + onText: (delta: string) => { + const current = sessions.get(sessionId); + if (!current) return; + current.thinkingOutput += delta; + }, + }) as Promise; + + void agentPromise.then((lateAgent) => { + if (abortSignal.aborted) { + try { + lateAgent?.session?.dispose?.(); + } catch { + // ignore late cleanup errors + } + } + }, () => undefined); + + const agent = await agentPromise; + if (abortSignal.aborted) { + try { + agent?.session?.dispose?.(); + } catch { + // ignore cleanup errors + } + throw createAbortError(); + } + session.agent = agent; + + await agent.session.prompt(session.initialDescription, { signal: abortSignal }); + + if (abortSignal.aborted) { + throw createAbortError(); + } const messages = agent.session.state.messages as Array<{ role: string; content?: string | Array<{ type: string; text: string }> }>; const lastAssistant = messages.filter((m) => m.role === "assistant").pop();