root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

FN-6978: fix Planning Mode liveness watchdog

Browse Source 
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
This commit is contained in:
gsxdsm
2026-06-25 00:38:41 -07:00
parent 7e7b0c6818
commit f1b3bd8854
3 changed files with 281 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
.changeset/fn-6978-planning-liveness.md Normal file
View File

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

193
packages/dashboard/src/__tests__/session-error-recovery.test.ts
View File

@@ -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(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<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();

99
packages/dashboard/src/planning.ts
View File

@@ -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;
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<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) {
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;
}