From f1b3bd8854e6fa814aa3eb7f7bacc4631de9de35 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Thu, 25 Jun 2026 00:38:41 -0700 Subject: [PATCH] FN-6978: fix Planning Mode liveness watchdog Replace the fixed Planning Mode generation wall-clock cap with progress-aware liveness handling.\n\n- Refresh the generation watchdog when meaningful thinking or text output advances.\n- Abort repeated identical output as a deterministic loop while preserving user-stop errors.\n- Cover inactivity, progress refresh, loop detection, displaced generations, and manual stop behavior.\n- Add a patch changeset for the Planning Mode liveness fix.\n\nFiles changed:\n .changeset/fn-6978-planning-liveness.md | 7 +\n .../src/__tests__/session-error-recovery.test.ts | 193 ++++++++++++++++++++-\n packages/dashboard/src/planning.ts | 99 +++++++++--\n 3 files changed, 281 insertions(+), 18 deletions(-) Fusion-Task-Id: FN-6978 Fusion-Task-Lineage: cf23956e-db79-47ac-a10b-657538aa5440 --- .changeset/fn-6978-planning-liveness.md | 7 + .../__tests__/session-error-recovery.test.ts | 193 +++++++++++++++++- packages/dashboard/src/planning.ts | 99 +++++++-- 3 files changed, 281 insertions(+), 18 deletions(-) create mode 100644 .changeset/fn-6978-planning-liveness.md diff --git a/.changeset/fn-6978-planning-liveness.md b/.changeset/fn-6978-planning-liveness.md new file mode 100644 index 0000000000..5f332534b3 --- /dev/null +++ b/.changeset/fn-6978-planning-liveness.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Allow Planning Mode generations to continue while meaningful AI output is progressing. +category: fix +dev: Replaces the fixed Planning Mode generation cap with inactivity and repeated-output detection. diff --git a/packages/dashboard/src/__tests__/session-error-recovery.test.ts b/packages/dashboard/src/__tests__/session-error-recovery.test.ts index a1e5495ad5..6251191cf3 100644 --- a/packages/dashboard/src/__tests__/session-error-recovery.test.ts +++ b/packages/dashboard/src/__tests__/session-error-recovery.test.ts @@ -18,12 +18,14 @@ import { __setCreateFnAgent, createSession, createSessionWithAgent, + GENERATION_LOOP_REPEAT_LIMIT, GENERATION_TIMEOUT_MS as PLANNING_GENERATION_TIMEOUT_MS, getSession, parseAgentResponse, planningStreamManager, retrySession, setAiSessionStore as setPlanningAiSessionStore, + stopGeneration, submitResponse, } from "../planning.js"; import { @@ -68,6 +70,16 @@ function makeTmpDir(): string { return mkdtempSync(join(tmpdir(), "kb-session-error-recovery-")); } +function createDeferred() { + let resolve!: (value: T | PromiseLike) => void; + let reject!: (reason?: unknown) => void; + const promise = new Promise((innerResolve, innerReject) => { + resolve = innerResolve; + reject = innerReject; + }); + return { promise, resolve, reject }; +} + function createMockAgent(responses: string[]) { const queue = [...responses]; const messages: Array<{ role: string; content: string }> = []; @@ -346,7 +358,74 @@ The actual planning response is: unsubscribeError(); }); - it("times out planning sessions when createFnAgent construction stalls", async () => { + it("allows meaningful planning progress beyond the old fixed generation deadline", async () => { + vi.useFakeTimers(); + + const promptDeferred = createDeferred(); + const messages: Array<{ role: string; content: string }> = []; + let streamOptions: { onThinking?: (delta: string) => void; onText?: (delta: string) => void } | undefined; + + __setCreateFnAgent(async (options: { onThinking?: (delta: string) => void; onText?: (delta: string) => void }) => { + streamOptions = options; + return { + session: { + state: { messages }, + prompt: vi.fn(async () => { + await promptDeferred.promise; + messages.push({ + role: "assistant", + content: JSON.stringify({ + type: "question", + data: { id: "q-long-progress", type: "text", question: "What should we build next?" }, + }), + }); + }), + dispose: vi.fn(), + }, + }; + }); + + const sessionId = await createSessionWithAgent( + "127.0.0.149", + "Planning long reasoning", + "/tmp/project", + taskStore, + ); + const errorEvents: string[] = []; + const questionEvents: string[] = []; + const unsubscribe = planningStreamManager.subscribe(sessionId, (event) => { + if (event.type === "error") errorEvents.push(String(event.data)); + if (event.type === "question") questionEvents.push(String((event.data as { id?: string }).id)); + }); + + planningStreamManager.consumeInitialTurn(sessionId)?.(); + await vi.advanceTimersByTimeAsync(0); + expect(__getActiveGenerationForTests(sessionId)).toBeDefined(); + + streamOptions?.onThinking?.("Considering the project context"); + await vi.advanceTimersByTimeAsync(PLANNING_GENERATION_TIMEOUT_MS / 2); + streamOptions?.onText?.("Drafting a focused planning question"); + await vi.advanceTimersByTimeAsync(PLANNING_GENERATION_TIMEOUT_MS / 2 + 10_000); + + expect(aiSessionStore.get(sessionId)?.status).toBe("generating"); + expect(aiSessionStore.get(sessionId)?.error).toBeNull(); + expect(errorEvents).toEqual([]); + expect(__getActiveGenerationForTests(sessionId)).toBeDefined(); + + promptDeferred.resolve(); + await vi.advanceTimersByTimeAsync(0); + + expect(aiSessionStore.get(sessionId)?.status).toBe("awaiting_input"); + expect(aiSessionStore.get(sessionId)?.error).toBeNull(); + expect(getSession(sessionId)?.currentQuestion?.id).toBe("q-long-progress"); + expect(questionEvents).toContain("q-long-progress"); + expect(errorEvents).toEqual([]); + expect(__getActiveGenerationForTests(sessionId)).toBeUndefined(); + + unsubscribe(); + }); + + it("marks planning sessions as stuck when createFnAgent construction stalls", async () => { vi.useFakeTimers(); __setCreateFnAgent(async () => { @@ -375,14 +454,14 @@ The actual planning response is: 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(aiSessionStore.get(sessionId)?.error).toMatch(/stuck with no new output/i); + expect(errorEvents).toContainEqual(expect.stringMatching(/stuck with no new output/i)); expect(__getActiveGenerationForTests(sessionId)).toBeUndefined(); unsubscribe(); }); - it("times out planning sessions when prompt stalls and disposes the agent", async () => { + it("marks planning sessions as stuck when prompt stalls and disposes the agent", async () => { vi.useFakeTimers(); const dispose = vi.fn(); @@ -417,8 +496,110 @@ The actual planning response is: 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(aiSessionStore.get(sessionId)?.error).toMatch(/stuck with no new output/i); + expect(errorEvents).toContainEqual(expect.stringMatching(/stuck with no new output/i)); + expect(__getActiveGenerationForTests(sessionId)).toBeUndefined(); + expect(dispose).toHaveBeenCalled(); + + unsubscribe(); + }); + + it("stops repeated planning output as a loop and retries to completion", async () => { + vi.useFakeTimers(); + + const dispose = vi.fn(); + let streamOptions: { onText?: (delta: string) => void } | undefined; + __setCreateFnAgent(async (options: { onText?: (delta: string) => void }) => { + streamOptions = options; + return { + session: { + state: { messages: [] }, + prompt: vi.fn(async () => { + await new Promise(() => undefined); + }), + dispose, + }, + }; + }); + + const sessionId = await createSessionWithAgent( + "127.0.0.152", + "Planning repeated output", + "/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(); + + for (let i = 0; i < GENERATION_LOOP_REPEAT_LIMIT + 1; i += 1) { + streamOptions?.onText?.("same repeated chunk"); + } + await vi.advanceTimersByTimeAsync(0); + + expect(aiSessionStore.get(sessionId)?.status).toBe("error"); + expect(aiSessionStore.get(sessionId)?.error).toMatch(/repeating the same output/i); + expect(errorEvents).toContainEqual(expect.stringMatching(/repeating the same output/i)); + expect(__getActiveGenerationForTests(sessionId)).toBeUndefined(); + expect(dispose).toHaveBeenCalled(); + + __setCreateFnAgent( + async () => + createMockAgent([ + JSON.stringify({ + type: "question", + data: { id: "q-loop-retry", type: "text", question: "Recovered after loop" }, + }), + ]), + ); + + await retrySession(sessionId, "/tmp/project", undefined, taskStore); + + expect(aiSessionStore.get(sessionId)?.status).toBe("awaiting_input"); + expect(aiSessionStore.get(sessionId)?.error).toBeNull(); + expect(getSession(sessionId)?.currentQuestion?.id).toBe("q-loop-retry"); + + unsubscribe(); + }); + + it("manual stop preserves the user-stopped Planning Mode error", 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.153", + "Planning manual stop", + "/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(stopGeneration(sessionId)).toBe(true); + await vi.advanceTimersByTimeAsync(0); + + expect(aiSessionStore.get(sessionId)?.status).toBe("error"); + expect(aiSessionStore.get(sessionId)?.error).toMatch(/stopped by user/i); + expect(errorEvents).toContainEqual(expect.stringMatching(/stopped by user/i)); expect(__getActiveGenerationForTests(sessionId)).toBeUndefined(); expect(dispose).toHaveBeenCalled(); diff --git a/packages/dashboard/src/planning.ts b/packages/dashboard/src/planning.ts index e1bc75ae54..32ede852bf 100644 --- a/packages/dashboard/src/planning.ts +++ b/packages/dashboard/src/planning.ts @@ -260,9 +260,19 @@ const MAX_SESSIONS_PER_IP_PER_HOUR = 1000; /** Rate limiting window in milliseconds (1 hour) */ const RATE_LIMIT_WINDOW_MS = 60 * 60 * 1000; -/** Generation timeout in milliseconds (120 seconds). */ +/* +FNXC:PlanningLiveness 2026-06-24-00:00: +Planning Mode must allow long-running reasoning when the agent is producing new thinking/text, because a fixed wall-clock cap incorrectly fails legitimate sessions. The watchdog is scoped to Planning Mode and treats this value as an inactivity window: non-empty, materially new output refreshes liveness; repeated identical output is counted separately and stopped as a loop so stuck sessions still fail deterministically without changing subtask, mission, milestone, or onboarding timeout semantics. +*/ export const GENERATION_TIMEOUT_MS = 120_000; +/** Repeated identical planning stream chunks before the generation is classified as looping. */ +export const GENERATION_LOOP_REPEAT_LIMIT = 8; + +const PLANNING_STUCK_ERROR_MESSAGE = "AI generation appears stuck with no new output. You can retry or start a new session."; +const PLANNING_LOOP_ERROR_MESSAGE = "AI generation appears stuck repeating the same output. You can retry or start a new session."; +const PLANNING_USER_STOP_ERROR_MESSAGE = "Generation stopped by user. You can retry or start a new session."; + export type PlanningDepth = "small" | "medium" | "large"; const PLANNING_DEPTH_PROMPT_SUFFIX: Record = { @@ -364,8 +374,17 @@ const sessions = new Map(); /** Rate limiting state indexed by IP */ const rateLimits = new Map(); +type PlanningGenerationAbortReason = "stuck" | "loop" | "user-stop" | "displaced"; + +interface ActivePlanningGeneration { + abortController: AbortController; + timer: NodeJS.Timeout; + abortReason?: PlanningGenerationAbortReason; + markProgress: (output: string) => void; +} + /** Active planning generations keyed by session ID. */ -const activeGenerations = new Map(); +const activeGenerations = new Map(); // ── AI Session Persistence ──────────────────────────────────────────────── @@ -1469,6 +1488,7 @@ async function createPlanningAgent( } : {}), onThinking: (delta: string) => { + markPlanningGenerationProgress(session.id, delta); session.thinkingOutput += delta; persistThinking(session.id, session.thinkingOutput); planningStreamManager.broadcast(session.id, { @@ -1480,6 +1500,7 @@ async function createPlanningAgent( // Capture AI response text — will be parsed at end of turn. Also // surface it through the same stream so non-thinking models (which // never emit thinking_delta) still show streaming output in the UI. + markPlanningGenerationProgress(session.id, delta); session.thinkingOutput += delta; persistThinking(session.id, session.thinkingOutput); planningStreamManager.broadcast(session.id, { @@ -1628,22 +1649,71 @@ function createAbortError(): Error { return error; } +function normalizeGenerationProgress(output: string): string { + return output.replace(/\s+/g, " ").trim(); +} + +function markPlanningGenerationProgress(sessionId: string, output: string): void { + activeGenerations.get(sessionId)?.markProgress(output); +} + async function runGenerationWithTimeout(session: Session, operation: (abortSignal: AbortSignal) => Promise): Promise { const existing = activeGenerations.get(session.id); if (existing) { clearTimeout(existing.timer); + existing.abortReason = "displaced"; existing.abortController.abort(); } const abortController = new AbortController(); - let timeoutTriggered = false; - const timer = setTimeout(() => { - timeoutTriggered = true; - setSessionError(session, "AI generation timed out. You can retry or start a new session."); - disposeSessionAgentForRetry(session); + let lastProgressSignature = ""; + let repeatedProgressCount = 0; + + const abortGeneration = (reason: PlanningGenerationAbortReason, message?: string): void => { + if (abortController.signal.aborted) { + return; + } + generationRecord.abortReason = reason; + clearTimeout(generationRecord.timer); + if (message) { + diagnostics.warn("Planning generation watchdog aborting session", { + sessionId: session.id, + reason, + operation: "planning-generation-watchdog", + }); + setSessionError(session, message); + disposeSessionAgentForRetry(session); + } abortController.abort(); + }; + + const scheduleInactivityTimer = (): NodeJS.Timeout => setTimeout(() => { + abortGeneration("stuck", PLANNING_STUCK_ERROR_MESSAGE); }, GENERATION_TIMEOUT_MS); - const generationRecord = { abortController, timer }; + + const generationRecord: ActivePlanningGeneration = { + abortController, + timer: scheduleInactivityTimer(), + markProgress: (output: string) => { + const signature = normalizeGenerationProgress(output); + if (!signature) { + return; + } + + if (signature === lastProgressSignature) { + repeatedProgressCount += 1; + if (repeatedProgressCount >= GENERATION_LOOP_REPEAT_LIMIT) { + abortGeneration("loop", PLANNING_LOOP_ERROR_MESSAGE); + } + return; + } + + lastProgressSignature = signature; + repeatedProgressCount = 0; + clearTimeout(generationRecord.timer); + generationRecord.timer = scheduleInactivityTimer(); + }, + }; activeGenerations.set(session.id, generationRecord); @@ -1659,13 +1729,14 @@ async function runGenerationWithTimeout(session: Session, operation: (abortSi return await Promise.race([operation(abortController.signal), abortPromise]); } catch (error) { if (error instanceof Error && error.name === "AbortError") { - if (!timeoutTriggered && !session.error) { - setSessionError(session, "Generation stopped by user. You can retry or start a new session."); + const reason = generationRecord.abortReason; + if (reason === "user-stop" && !session.error) { + setSessionError(session, PLANNING_USER_STOP_ERROR_MESSAGE); } } throw error; } finally { - clearTimeout(timer); + clearTimeout(generationRecord.timer); if (activeGenerations.get(session.id) === generationRecord) { activeGenerations.delete(session.id); } @@ -1714,6 +1785,8 @@ async function continueAgentConversation(session: Session, message: string): Pro } } + markPlanningGenerationProgress(session.id, responseText); + // Diagnostic: warn when response text is empty or very short if (!responseText || responseText.length < 10) { const contentBlockTypes = Array.isArray(lastMessage?.content) @@ -1777,6 +1850,7 @@ async function continueAgentConversation(session: Session, message: string): Pro } } responseText = retryText; + markPlanningGenerationProgress(session.id, responseText); } catch (retryErr) { // Retry prompt itself failed — give up diagnostics.errorFromException( @@ -2254,6 +2328,7 @@ export function stopGeneration(sessionId: string): boolean { return false; } + activeGeneration.abortReason = "user-stop"; activeGeneration.abortController.abort(); clearTimeout(activeGeneration.timer); activeGenerations.delete(sessionId); @@ -2268,7 +2343,7 @@ export function stopGeneration(sessionId: string): boolean { session.agent = undefined; } - setSessionError(session, "Generation stopped by user. You can retry or start a new session."); + setSessionError(session, PLANNING_USER_STOP_ERROR_MESSAGE); return true; }