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

@@ -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.

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<T = void>() {
+  let resolve!: (value: T | PromiseLike<T>) => void;
+  let reject!: (reason?: unknown) => void;
+  const promise = new Promise<T>((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(aiSessionStore.get(sessionId)?.error).toMatch(/stuck with no new output/i);
-    expect(errorEvents).toContainEqual(expect.stringMatching(/timed out/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(aiSessionStore.get(sessionId)?.error).toMatch(/stuck with no new output/i);
-    expect(errorEvents).toContainEqual(expect.stringMatching(/timed out/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<never>(() => 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<never>(() => 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();
 

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<PlanningDepth, string> = {
@@ -364,8 +374,17 @@ const sessions = new Map<string, Session>();
 /** Rate limiting state indexed by IP */
 const rateLimits = new Map<string, RateLimitEntry>();
 
+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<string, { abortController: AbortController; timer: NodeJS.Timeout }>();
+const activeGenerations = new Map<string, ActivePlanningGeneration>();
 
 // ── 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<T>(session: Session, operation: (abortSignal: AbortSignal) => Promise<T>): Promise<T> {
   const existing = activeGenerations.get(session.id);
   if (existing) {
     clearTimeout(existing.timer);
+    existing.abortReason = "displaced";
     existing.abortController.abort();
   }
 
   const abortController = new AbortController();
-  let timeoutTriggered = false;
+  let lastProgressSignature = "";
-  const timer = setTimeout(() => {
+  let repeatedProgressCount = 0;
-    timeoutTriggered = true;
+
-    setSessionError(session, "AI generation timed out. You can retry or start a new session.");
+  const abortGeneration = (reason: PlanningGenerationAbortReason, message?: string): void => {
-    disposeSessionAgentForRetry(session);
+    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<T>(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) {
+      const reason = generationRecord.abortReason;
-        setSessionError(session, "Generation stopped by user. You can retry or start a new session.");
+      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;
 }