fix(dashboard): stop Planning Mode duplicate generations, 'AI returned no valid JSON' errors, and context-overflow failures (#2417)
## Symptom Users reported Planning Mode regularly failing with **"AI returned no valid JSON.. Retry this planning session or start a new one."**, and that leaving and returning to the interface mid-generation **duplicates the generation — infinitely**. Every app-tab switch unmounts the Planning view, so the leave/return path is the *normal* path, not an edge case. ## Root causes 1. **Check-then-act turn admission.** `activeGenerations.has()` was checked at turn entry, but the record was only created inside `runGenerationWithTimeout`, after several awaits. Overlapping turn entries (remounted view re-submitting, racing auto-retries, duplicate start of an existing session) both passed the guard; the second displaced the first, and the displaced teardown disposed the **session-shared agent the surviving turn was actively prompting** — which then read an empty assistant message and failed parse with "AI returned no valid JSON". 2. **Per-mount auto-retry budget.** The client reset its 3-attempt auto-retry budget on every mount, so each return to an errored session re-ran a full-turn regeneration (agent rebuild + complete history replay) — forever. 3. **SSE replay appended onto existing output.** Fresh stream connections replay buffered thinking; the client pre-seeded from persisted `thinkingOutput` (or kept prior output on silent reconnect) and then appended the replay — visibly doubling the generation on every reconnect. The 100-event buffer only held a suffix of a turn, forcing that pre-seed. 4. **Raw `session.prompt()` at context limits.** A long interview that overflowed the model's context window errored terminally, and auto-retry replayed the full history into a fresh agent — overflowing again, unrecoverably. ## Fix - Synchronous per-session **turn reservation** shared by `submitResponse`, `retrySession`, `startExistingSession`, and the initial turn; losers get `GenerationInProgressError` instead of displacing the winner. Duplicate starts of a generating session are no-ops. Rewind aborts an active generation through its own teardown first. - Client auto-retry budget is **module-scoped per session** (survives remounts); exhausted budget shows the error view instead of a stuck spinner. Retry rejections for "already in progress" rejoin the live run. - Fresh SSE connections **clear streamed output before the buffered replay**; buffer deepened to a full turn (2000 events); rejoin paths reconnect cleanly instead of seeding persisted thinking. - All six planning prompt sites route through the engine's **`promptWithFallback`**, recovering context-window overflows via prompt/memory compaction and `session.compact()`. - Cosmetic: no more doubled period in the retryable parse error message. ## Symptom Verification - **Original symptom:** "AI returned no valid JSON" after answering questions; generations duplicating on leave/return. - **Exact reproduction:** concurrent turn entries on one session (submit×2, retry×2, start-while-generating) — previously displaced/disposed the live agent mid-prompt. - **Assertion it is gone:** `planning-turn-admission.test.ts` asserts exactly one turn is admitted per race, the winner completes with a question and no session error, and the shared agent is never disposed; `planning-context-compaction.test.ts` asserts every planning prompt routes through `promptWithFallback` (signal forwarded) and that a recovered context overflow leaves the turn healthy. ## Verification - `vitest run` on all 7 planning server test files + the 2 new regression files: **40/40 pass**. - `routes-planning*` failures at main tip are pre-existing (identical 103/126 + 3/6 counts with and without this diff; main is mid-refactor on route wiring). `planning-answered-question-reemit` 3 timeouts also reproduce on clean main. - `pnpm verify:fast` green; dashboard `tsconfig.json` + `tsconfig.app.json` typechecks clean; eslint clean on touched files. 🤖 Generated with [Claude Code](https://claude.com/claude-code) <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit * **Bug Fixes** * Prevented Planning Mode from duplicating generations and triggering “AI returned no valid JSON” errors when leaving and re-entering mid-run. * Made planning turn handling concurrency-safe and idempotent across submit, retry, rewind, and duplicate start actions. * Improved SSE reconnect recovery: clearer replay after reconnect, no duplicated “thinking” output, and preserved auto-retry limits across remounts. * Improved long-context recovery via fallback prompting and cleaned up retry error formatting. * **Tests** * Expanded coverage for concurrent Planning actions, reconnect replay, context compaction, rewind behavior, and retry formatting; improved parallel test-harness reliability. <!-- end of auto-generated comment: release notes by coderabbit.ai --> --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
gsxdsm
7
.changeset/planning-turn-admission.md
Normal file
7
.changeset/planning-turn-admission.md
Normal file
@@ -0,0 +1,7 @@
|
||||
---
|
||||
"@runfusion/fusion": patch
|
||||
---
|
||||
|
||||
summary: Fix Planning Mode duplicating generations and "AI returned no valid JSON" errors after leaving and returning mid-run.
|
||||
category: fix
|
||||
dev: Planning turns are admitted through a synchronous per-session reservation across submitResponse/retrySession/startExistingSession and the initial turn, so a racing entry is rejected instead of displacing the in-flight generation and disposing its agent mid-prompt. Duplicate starts of a generating session are no-ops, the client auto-retry budget survives view remounts (module-scoped per-session map), and SSE reconnects rebuild thinking output from a full-turn replay buffer (2000 events) instead of appending onto existing output. Planning prompts also route through the engine's promptWithFallback so context-window overflows recover via compaction instead of erroring the session.
|
||||
@@ -79,6 +79,16 @@ const PLANNING_SIDEBAR_MAX_WIDTH = 560;
|
||||
const PLANNING_SIDEBAR_STORAGE_KEY = "fusion:planning-sidebar-width";
|
||||
|
||||
const MAX_PLANNING_AUTO_RETRIES = 3;
|
||||
|
||||
/*
|
||||
FNXC:PlanningRetry 2026-07-22-21:00:
|
||||
The auto-retry budget must survive remounts. Every app-tab switch unmounts the Planning view,
|
||||
and a ref-scoped budget was re-granted on each return, so a session stuck in a persisted error
|
||||
state regenerated its full turn (agent rebuild + history replay) on every visit — the reported
|
||||
"leave and come back duplicates the generation infinitely". Attempts are tracked per session in
|
||||
module scope; success paths (question/summary/new session) still clear the entry.
|
||||
*/
|
||||
const planningAutoRetryAttemptsBySession = new Map<string, number>();
|
||||
const MAX_PLANNING_CREATE_CLAIM_RETRIES = 20;
|
||||
|
||||
function isPlanningCreateClaimConflict(error: unknown): boolean {
|
||||
@@ -516,6 +526,9 @@ export function PlanningModeModal({ isOpen, onClose, onTaskCreated, onTasksCreat
|
||||
|
||||
const resetPlanningAutoRetryAttempts = useCallback(() => {
|
||||
planningAutoRetryAttemptRef.current = 0;
|
||||
// Clear the remount-durable budget too, so successful progress re-arms auto-retry.
|
||||
const sessionId = currentSessionIdRef.current;
|
||||
if (sessionId) planningAutoRetryAttemptsBySession.delete(sessionId);
|
||||
setAutoRetryAttempt(0);
|
||||
setIsAutoRetrying(false);
|
||||
}, []);
|
||||
@@ -1054,6 +1067,16 @@ export function PlanningModeModal({ isOpen, onClose, onTaskCreated, onTasksCreat
|
||||
(sessionId: string) => {
|
||||
const streamEpoch = ++streamConnectionEpochRef.current;
|
||||
streamConnectionRef.current?.close();
|
||||
/*
|
||||
FNXC:PlanningStreamCatchup 2026-07-22-21:00:
|
||||
A brand-new stream connection replays the session's buffered thinking from the start
|
||||
(the buffer is sized to hold a full turn). Whatever is currently on screen is a subset
|
||||
of that replay, so it must be cleared first — appending the replay onto existing output
|
||||
duplicated the visible generation on every reconnect (mobile tab switches unmount this
|
||||
view, so this happened constantly).
|
||||
*/
|
||||
setStreamingOutput("");
|
||||
streamingOutputRef.current = "";
|
||||
// Guard handlers against late events from a connection the user has
|
||||
// already navigated away from (e.g. clicked "New Session" while the
|
||||
// previous SSE flushed a buffered question). currentSessionIdRef is
|
||||
@@ -1243,7 +1266,10 @@ export function PlanningModeModal({ isOpen, onClose, onTaskCreated, onTasksCreat
|
||||
let retryError: unknown = err;
|
||||
const retryErrorMessage = getErrorMessage(err) || "";
|
||||
|
||||
if (retryErrorMessage.includes("not in an error state")) {
|
||||
// FNXC:PlanningTurnAdmission 2026-07-22-21:00: a retry rejected because another turn
|
||||
// already holds the session's turn slot means work is in progress — rejoin it via the
|
||||
// same session-refresh path instead of surfacing a terminal error.
|
||||
if (retryErrorMessage.includes("not in an error state") || retryErrorMessage.includes("already in progress")) {
|
||||
try {
|
||||
const session = await fetchAiSession(retryTarget.sessionId);
|
||||
if (!session) {
|
||||
@@ -1254,7 +1280,10 @@ export function PlanningModeModal({ isOpen, onClose, onTaskCreated, onTasksCreat
|
||||
currentSessionIdRef.current = session.id;
|
||||
|
||||
if (session.status === "generating") {
|
||||
setStreamingOutput(session.thinkingOutput ?? "");
|
||||
// FNXC:PlanningStreamCatchup 2026-07-22-21:00: rejoin through a clean stream
|
||||
// (clear + buffered replay) instead of seeding persisted thinking alongside the
|
||||
// in-flight replay, which raced and duplicated the visible output.
|
||||
connectToPlanningStream(session.id);
|
||||
setView({ type: "loading" });
|
||||
} else if (session.status === "awaiting_input") {
|
||||
if (!session.currentQuestion) {
|
||||
@@ -1322,7 +1351,7 @@ export function PlanningModeModal({ isOpen, onClose, onTaskCreated, onTasksCreat
|
||||
if (!retryStillOwnsSession()) return;
|
||||
streamConnectionRef.current?.close();
|
||||
streamConnectionRef.current = null;
|
||||
if (options.auto && planningAutoRetryAttemptRef.current < MAX_PLANNING_AUTO_RETRIES) {
|
||||
if (options.auto && (planningAutoRetryAttemptsBySession.get(retryTarget.sessionId) ?? 0) < MAX_PLANNING_AUTO_RETRIES) {
|
||||
viewRef.current = { type: "loading" };
|
||||
setView({ type: "loading" });
|
||||
setIsAutoRetrying(true);
|
||||
@@ -1359,13 +1388,16 @@ export function PlanningModeModal({ isOpen, onClose, onTaskCreated, onTasksCreat
|
||||
return true;
|
||||
}
|
||||
if (viewRef.current.type === "error") return false;
|
||||
if (planningAutoRetryAttemptRef.current >= MAX_PLANNING_AUTO_RETRIES) {
|
||||
// FNXC:PlanningRetry 2026-07-22-21:00: budget is per-session and survives remounts.
|
||||
const priorAttempts = planningAutoRetryAttemptsBySession.get(sessionId) ?? 0;
|
||||
if (priorAttempts >= MAX_PLANNING_AUTO_RETRIES) {
|
||||
setIsAutoRetrying(false);
|
||||
return false;
|
||||
}
|
||||
|
||||
const attempt = planningAutoRetryAttemptRef.current + 1;
|
||||
const attempt = priorAttempts + 1;
|
||||
const retryToken = Symbol(`planning-auto-retry:${sessionId}:${attempt}`);
|
||||
planningAutoRetryAttemptsBySession.set(sessionId, attempt);
|
||||
planningAutoRetryAttemptRef.current = attempt;
|
||||
planningAutoRetryOwnerRef.current = { sessionId, token: retryToken };
|
||||
setAutoRetryAttempt(attempt);
|
||||
@@ -1622,10 +1654,26 @@ export function PlanningModeModal({ isOpen, onClose, onTaskCreated, onTasksCreat
|
||||
through the same generation retry path used by live stream failures while preserving the
|
||||
hydrated running plan and the existing bounded single-flight protection.
|
||||
*/
|
||||
/*
|
||||
FNXC:PlanningRetry 2026-07-22-21:00:
|
||||
Do NOT reset the auto-retry budget here. Loading an errored session happens on every
|
||||
remount (each app-tab switch unmounts this view), and a per-mount reset turned the
|
||||
bounded three-attempt budget into an unbounded regeneration loop. The module-scoped
|
||||
per-session budget carries across remounts; only real progress clears it.
|
||||
*/
|
||||
if (planningAutoRetryOwnerRef.current?.sessionId !== sessionId) {
|
||||
resetPlanningAutoRetryAttempts();
|
||||
planningAutoRetryAttemptRef.current = planningAutoRetryAttemptsBySession.get(sessionId) ?? 0;
|
||||
setAutoRetryAttempt(planningAutoRetryAttemptRef.current);
|
||||
}
|
||||
const autoRetryStarted = await startPlanningAutoRetry(sessionId);
|
||||
if (!autoRetryStarted) {
|
||||
// Budget exhausted: surface the persisted error with a manual Retry affordance.
|
||||
setView({
|
||||
type: "error",
|
||||
session: { sessionId, currentQuestion: null, summary: persistedRunningSummary },
|
||||
errorMessage: session.error || t("planning.sessionFailed", "Session failed while contacting the AI."),
|
||||
});
|
||||
}
|
||||
await startPlanningAutoRetry(sessionId);
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -1708,7 +1756,9 @@ export function PlanningModeModal({ isOpen, onClose, onTaskCreated, onTasksCreat
|
||||
}
|
||||
} else if (session.status === "generating") {
|
||||
setView({ type: "loading" });
|
||||
if (session.thinkingOutput) setStreamingOutput(session.thinkingOutput);
|
||||
// FNXC:PlanningStreamCatchup 2026-07-22-21:00: no pre-seed from persisted
|
||||
// thinkingOutput — the stream replay reconstructs the loading view exactly once;
|
||||
// seeding here and then replaying doubled the visible output on every reload.
|
||||
connectToPlanningStream(sessionId);
|
||||
}
|
||||
} catch (err) {
|
||||
|
||||
@@ -0,0 +1,155 @@
|
||||
// @vitest-environment node
|
||||
|
||||
/*
|
||||
FNXC:PlanningContextCompaction 2026-07-22-22:40:
|
||||
Planning prompts must route through the engine's promptWithFallback so context-window
|
||||
overflows recover via compaction instead of surfacing "prompt is too long" as a terminal
|
||||
session error (whose auto-retry replays the FULL history and overflows again). These tests
|
||||
pin the invariant: when the engine exposes promptWithFallback, every planning agent prompt —
|
||||
initial turn, answer turns, reformat retries — goes through it, with the generation
|
||||
AbortSignal forwarded; and a context-limit error recovered inside promptWithFallback leaves
|
||||
the planning turn healthy.
|
||||
*/
|
||||
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { EventEmitter } from "node:events";
|
||||
import type { TaskStore } from "@fusion/core";
|
||||
|
||||
const promptWithFallbackCalls: Array<{ prompt: string; options?: { signal?: AbortSignal } }> = [];
|
||||
let simulateContextRecoveryOnce = false;
|
||||
|
||||
vi.mock("@fusion/engine", () => ({
|
||||
listCliAdapterDescriptors: () => [],
|
||||
resolveMcpServersForStore: async () => ({ servers: [] }),
|
||||
buildSessionSkillContextSync: () => ({
|
||||
skillSelectionContext: undefined,
|
||||
resolvedSkillNames: ["fusion"],
|
||||
skillSource: "role-fallback" as const,
|
||||
}),
|
||||
createFnAgent: vi.fn(),
|
||||
createWorkflowAuthoringTools: () => [],
|
||||
createChatTaskDocumentTools: () => [],
|
||||
createChatTaskLogsReadTool: () => ({}),
|
||||
promptWithFallback: vi.fn(async (
|
||||
agentSession: { prompt: (input: string, options?: { signal?: AbortSignal }) => Promise<void> },
|
||||
prompt: string,
|
||||
options?: { signal?: AbortSignal },
|
||||
) => {
|
||||
promptWithFallbackCalls.push({ prompt, options });
|
||||
if (simulateContextRecoveryOnce) {
|
||||
// Mirror the engine contract: the raw prompt overflows, compaction recovers,
|
||||
// and the retried prompt succeeds — the caller sees one successful await.
|
||||
simulateContextRecoveryOnce = false;
|
||||
try {
|
||||
throw new Error("prompt is too long: 210000 tokens > 200000 maximum");
|
||||
} catch {
|
||||
// compacted; fall through to the retried prompt below
|
||||
}
|
||||
}
|
||||
await agentSession.prompt(prompt, options);
|
||||
}),
|
||||
}));
|
||||
|
||||
import {
|
||||
__resetPlanningState,
|
||||
__setCreateFnAgent,
|
||||
createSessionWithAgent,
|
||||
getSession,
|
||||
planningStreamManager,
|
||||
setAiSessionStore,
|
||||
submitResponse,
|
||||
} from "../planning.js";
|
||||
|
||||
const MOCK_TASK_STORE = {
|
||||
listTasks: vi.fn(async () => []),
|
||||
getSettings: vi.fn(async () => ({})),
|
||||
getTask: vi.fn(async () => {
|
||||
throw new Error("not found");
|
||||
}),
|
||||
} as unknown as TaskStore;
|
||||
|
||||
const QUESTION_JSON = JSON.stringify({
|
||||
type: "question",
|
||||
data: { id: "q-next", type: "single_select", question: "What next?" },
|
||||
});
|
||||
|
||||
function createFakeAgent() {
|
||||
const messages: Array<{ role: string; content: string }> = [];
|
||||
const prompt = vi.fn(async () => {
|
||||
messages.push({ role: "assistant", content: QUESTION_JSON });
|
||||
});
|
||||
return { session: { state: { messages }, prompt, dispose: vi.fn() } };
|
||||
}
|
||||
|
||||
async function waitFor(predicate: () => Promise<boolean> | boolean, attempts = 50): Promise<void> {
|
||||
for (let i = 0; i < attempts; i++) {
|
||||
if (await predicate()) return;
|
||||
await new Promise((resolve) => setTimeout(resolve, 5));
|
||||
}
|
||||
throw new Error("condition not reached");
|
||||
}
|
||||
|
||||
describe("planning context-window compaction routing", () => {
|
||||
beforeEach(() => {
|
||||
__resetPlanningState();
|
||||
promptWithFallbackCalls.length = 0;
|
||||
simulateContextRecoveryOnce = false;
|
||||
setAiSessionStore(Object.assign(new EventEmitter(), {
|
||||
upsert: vi.fn(async () => {}),
|
||||
get: vi.fn(async () => null),
|
||||
updateThinking: vi.fn(),
|
||||
}) as never);
|
||||
});
|
||||
|
||||
it("routes every planning prompt through promptWithFallback and forwards the abort signal", async () => {
|
||||
const agent = createFakeAgent();
|
||||
__setCreateFnAgent(vi.fn(async () => agent) as never);
|
||||
|
||||
const sessionId = await createSessionWithAgent(
|
||||
"10.0.7.7",
|
||||
"Plan a long feature",
|
||||
"/tmp/project",
|
||||
MOCK_TASK_STORE,
|
||||
);
|
||||
planningStreamManager.consumeInitialTurn(sessionId)?.();
|
||||
await waitFor(async () => Boolean((await getSession(sessionId))?.currentQuestion));
|
||||
|
||||
expect(promptWithFallbackCalls.length).toBeGreaterThan(0);
|
||||
const initialCall = promptWithFallbackCalls[0];
|
||||
expect(initialCall.options?.signal).toBeInstanceOf(AbortSignal);
|
||||
|
||||
const question = (await getSession(sessionId))!.currentQuestion!;
|
||||
const beforeTurn = promptWithFallbackCalls.length;
|
||||
await submitResponse(sessionId, { [question.id]: "option-1" });
|
||||
|
||||
// The answer turn also went through the context-limit-aware path, signal included.
|
||||
expect(promptWithFallbackCalls.length).toBeGreaterThan(beforeTurn);
|
||||
const turnCall = promptWithFallbackCalls[beforeTurn];
|
||||
expect(turnCall.options?.signal).toBeInstanceOf(AbortSignal);
|
||||
// Planning never bypasses the wrapper: every raw prompt was issued by the wrapper itself.
|
||||
expect(agent.session.prompt).toHaveBeenCalledTimes(promptWithFallbackCalls.length);
|
||||
});
|
||||
|
||||
it("keeps the turn healthy when promptWithFallback recovers from a context overflow", async () => {
|
||||
const agent = createFakeAgent();
|
||||
__setCreateFnAgent(vi.fn(async () => agent) as never);
|
||||
|
||||
const sessionId = await createSessionWithAgent(
|
||||
"10.0.7.7",
|
||||
"Plan a long feature",
|
||||
"/tmp/project",
|
||||
MOCK_TASK_STORE,
|
||||
);
|
||||
planningStreamManager.consumeInitialTurn(sessionId)?.();
|
||||
await waitFor(async () => Boolean((await getSession(sessionId))?.currentQuestion));
|
||||
|
||||
const question = (await getSession(sessionId))!.currentQuestion!;
|
||||
simulateContextRecoveryOnce = true;
|
||||
const response = await submitResponse(sessionId, { [question.id]: "option-1" });
|
||||
|
||||
expect(response.type).toBe("question");
|
||||
const session = await getSession(sessionId);
|
||||
expect(session?.error).toBeUndefined();
|
||||
expect(session?.currentQuestion).toBeDefined();
|
||||
});
|
||||
});
|
||||
328
packages/dashboard/src/__tests__/planning-turn-admission.test.ts
Normal file
328
packages/dashboard/src/__tests__/planning-turn-admission.test.ts
Normal file
@@ -0,0 +1,328 @@
|
||||
// @vitest-environment node
|
||||
|
||||
/*
|
||||
FNXC:PlanningTurnAdmission 2026-07-22-21:00:
|
||||
Regression tests for the single-turn admission invariant. Reported bug: leaving and
|
||||
re-entering Planning Mode mid-generation (every app-tab switch unmounts the view) raced a
|
||||
second turn entry against the in-flight one; the loser displaced the winner and disposed the
|
||||
session-shared agent mid-prompt, surfacing "AI returned no valid JSON" and visibly duplicated
|
||||
generations. The invariant: at most one turn per session is ever admitted, across ALL entry
|
||||
points (submitResponse, retrySession, startExistingSession/initial turn), and the losing entry
|
||||
is rejected or ignored without touching the winner's agent.
|
||||
*/
|
||||
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { EventEmitter } from "node:events";
|
||||
import type { TaskStore } from "@fusion/core";
|
||||
|
||||
vi.mock("@fusion/engine", () => ({
|
||||
listCliAdapterDescriptors: () => [],
|
||||
resolveMcpServersForStore: async () => ({ servers: [] }),
|
||||
buildSessionSkillContextSync: () => ({
|
||||
skillSelectionContext: undefined,
|
||||
resolvedSkillNames: ["fusion"],
|
||||
skillSource: "role-fallback" as const,
|
||||
}),
|
||||
createFnAgent: vi.fn(),
|
||||
createWorkflowAuthoringTools: () => [],
|
||||
createChatTaskDocumentTools: () => [],
|
||||
createChatTaskLogsReadTool: () => ({}),
|
||||
}));
|
||||
|
||||
import {
|
||||
__resetPlanningState,
|
||||
__setCreateFnAgent,
|
||||
createSessionWithAgent,
|
||||
GenerationInProgressError,
|
||||
getSession,
|
||||
planningStreamManager,
|
||||
retrySession,
|
||||
rewindSession,
|
||||
setAiSessionStore,
|
||||
startExistingSession,
|
||||
submitResponse,
|
||||
} from "../planning.js";
|
||||
|
||||
const MOCK_TASK_STORE = {
|
||||
listTasks: vi.fn(async () => []),
|
||||
getSettings: vi.fn(async () => ({})),
|
||||
getTask: vi.fn(async () => {
|
||||
throw new Error("not found");
|
||||
}),
|
||||
} as unknown as TaskStore;
|
||||
|
||||
const QUESTION_JSON = JSON.stringify({
|
||||
type: "question",
|
||||
data: { id: "q-next", type: "single_select", question: "What next?" },
|
||||
});
|
||||
|
||||
interface ScriptedAgent {
|
||||
agent: { session: { state: { messages: Array<{ role: string; content: string }> }; prompt: ReturnType<typeof vi.fn>; dispose: ReturnType<typeof vi.fn> } };
|
||||
holdNextPrompt: () => void;
|
||||
releasePrompt: () => void;
|
||||
setResponder: (fn: () => string) => void;
|
||||
}
|
||||
|
||||
function createScriptedAgent(): ScriptedAgent {
|
||||
const messages: Array<{ role: string; content: string }> = [];
|
||||
let gate: Promise<void> | null = null;
|
||||
let releaseGate: (() => void) | null = null;
|
||||
let responder: () => string = () => QUESTION_JSON;
|
||||
|
||||
const prompt = vi.fn(async () => {
|
||||
if (gate) {
|
||||
const pending = gate;
|
||||
gate = null;
|
||||
await pending;
|
||||
}
|
||||
messages.push({ role: "assistant", content: responder() });
|
||||
});
|
||||
|
||||
return {
|
||||
agent: { session: { state: { messages }, prompt, dispose: vi.fn() } },
|
||||
holdNextPrompt: () => {
|
||||
gate = new Promise<void>((resolve) => {
|
||||
releaseGate = resolve;
|
||||
});
|
||||
},
|
||||
releasePrompt: () => {
|
||||
releaseGate?.();
|
||||
releaseGate = null;
|
||||
},
|
||||
setResponder: (fn: () => string) => {
|
||||
responder = fn;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
async function waitFor(predicate: () => Promise<boolean> | boolean, attempts = 50): Promise<void> {
|
||||
for (let i = 0; i < attempts; i++) {
|
||||
if (await predicate()) return;
|
||||
await new Promise((resolve) => setTimeout(resolve, 5));
|
||||
}
|
||||
throw new Error("condition not reached");
|
||||
}
|
||||
|
||||
async function startSessionAwaitingInput(scripted: ScriptedAgent): Promise<string> {
|
||||
__setCreateFnAgent(vi.fn(async () => scripted.agent) as never);
|
||||
const sessionId = await createSessionWithAgent(
|
||||
"10.0.9.9",
|
||||
"Plan something small",
|
||||
"/tmp/project",
|
||||
MOCK_TASK_STORE,
|
||||
);
|
||||
planningStreamManager.consumeInitialTurn(sessionId)?.();
|
||||
await waitFor(async () => Boolean((await getSession(sessionId))?.currentQuestion));
|
||||
return sessionId;
|
||||
}
|
||||
|
||||
describe("planning single-turn admission", () => {
|
||||
let storeGet: ReturnType<typeof vi.fn>;
|
||||
|
||||
beforeEach(() => {
|
||||
__resetPlanningState();
|
||||
storeGet = vi.fn(async () => null);
|
||||
setAiSessionStore(Object.assign(new EventEmitter(), {
|
||||
upsert: vi.fn(async () => {}),
|
||||
get: storeGet,
|
||||
updateThinking: vi.fn(),
|
||||
}) as never);
|
||||
});
|
||||
|
||||
it("rejects a concurrent submitResponse instead of displacing the in-flight turn", async () => {
|
||||
const scripted = createScriptedAgent();
|
||||
const sessionId = await startSessionAwaitingInput(scripted);
|
||||
const question = (await getSession(sessionId))!.currentQuestion!;
|
||||
|
||||
scripted.holdNextPrompt();
|
||||
const first = submitResponse(sessionId, { [question.id]: "option-1" });
|
||||
const second = submitResponse(sessionId, { [question.id]: "option-2" });
|
||||
|
||||
const [firstResult, secondResult] = await Promise.allSettled([
|
||||
first,
|
||||
second.finally(() => scripted.releasePrompt()),
|
||||
]);
|
||||
scripted.releasePrompt();
|
||||
|
||||
expect(secondResult.status).toBe("rejected");
|
||||
expect((secondResult as PromiseRejectedResult).reason).toBeInstanceOf(GenerationInProgressError);
|
||||
|
||||
// The admitted turn must complete cleanly: its agent was never disposed mid-prompt and
|
||||
// it produced the next question rather than "AI returned no valid JSON".
|
||||
expect(firstResult.status).toBe("fulfilled");
|
||||
expect((firstResult as PromiseFulfilledResult<{ type: string }>).value.type).toBe("question");
|
||||
const session = await getSession(sessionId);
|
||||
expect(session?.error).toBeUndefined();
|
||||
expect(session?.currentQuestion).toBeDefined();
|
||||
expect(scripted.agent.session.dispose).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("admits exactly one of two racing retries", async () => {
|
||||
const scripted = createScriptedAgent();
|
||||
const sessionId = await startSessionAwaitingInput(scripted);
|
||||
const session = (await getSession(sessionId))!;
|
||||
session.error = "AI returned no valid JSON. Retry this planning session or start a new one.";
|
||||
// Both retries must pass the persisted error-state check so the synchronous turn
|
||||
// reservation — not the winner's error-clearing side effect — is what rejects the loser.
|
||||
storeGet.mockImplementation(async () => ({ id: sessionId, type: "planning", status: "error" }));
|
||||
|
||||
const results = await Promise.allSettled([
|
||||
retrySession(sessionId, "/tmp/project", undefined, MOCK_TASK_STORE),
|
||||
retrySession(sessionId, "/tmp/project", undefined, MOCK_TASK_STORE),
|
||||
]);
|
||||
|
||||
const rejected = results.filter((result) => result.status === "rejected");
|
||||
expect(rejected).toHaveLength(1);
|
||||
expect((rejected[0] as PromiseRejectedResult).reason).toBeInstanceOf(GenerationInProgressError);
|
||||
|
||||
const after = await getSession(sessionId);
|
||||
expect(after?.error).toBeUndefined();
|
||||
expect(after?.currentQuestion).toBeDefined();
|
||||
});
|
||||
|
||||
it("treats a duplicate start of a generating session as a no-op", async () => {
|
||||
const scripted = createScriptedAgent();
|
||||
__setCreateFnAgent(vi.fn(async () => scripted.agent) as never);
|
||||
|
||||
scripted.holdNextPrompt();
|
||||
const sessionId = await createSessionWithAgent(
|
||||
"10.0.9.9",
|
||||
"Plan something small",
|
||||
"/tmp/project",
|
||||
MOCK_TASK_STORE,
|
||||
);
|
||||
planningStreamManager.consumeInitialTurn(sessionId)?.();
|
||||
await waitFor(() => scripted.agent.session.prompt.mock.calls.length > 0);
|
||||
|
||||
// A remounted client re-issuing start-streaming must not displace the live generation.
|
||||
await startExistingSession(sessionId, "/tmp/project", MOCK_TASK_STORE);
|
||||
expect(planningStreamManager.hasPendingInitialTurn(sessionId)).toBe(false);
|
||||
expect(scripted.agent.session.dispose).not.toHaveBeenCalled();
|
||||
|
||||
scripted.releasePrompt();
|
||||
await waitFor(async () => Boolean((await getSession(sessionId))?.currentQuestion));
|
||||
const session = await getSession(sessionId);
|
||||
expect(session?.error).toBeUndefined();
|
||||
expect(scripted.agent.session.dispose).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("resolves two concurrent duplicate starts without an initial-turn registration error", async () => {
|
||||
const scripted = createScriptedAgent();
|
||||
__setCreateFnAgent(vi.fn(async () => scripted.agent) as never);
|
||||
|
||||
scripted.holdNextPrompt();
|
||||
const sessionId = await createSessionWithAgent(
|
||||
"10.0.9.9",
|
||||
"Plan something small",
|
||||
"/tmp/project",
|
||||
MOCK_TASK_STORE,
|
||||
);
|
||||
planningStreamManager.consumeInitialTurn(sessionId)?.();
|
||||
await waitFor(() => scripted.agent.session.prompt.mock.calls.length > 0);
|
||||
|
||||
// FNXC:PlanningTurnAdmission 2026-07-23-10:10:
|
||||
// Two racing duplicate starts must BOTH be quiet no-ops — the pre-fix code let the
|
||||
// second reach registerInitialTurn and throw "Initial planning turn already registered".
|
||||
const results = await Promise.allSettled([
|
||||
startExistingSession(sessionId, "/tmp/project", MOCK_TASK_STORE),
|
||||
startExistingSession(sessionId, "/tmp/project", MOCK_TASK_STORE),
|
||||
]);
|
||||
expect(results.every((result) => result.status === "fulfilled")).toBe(true);
|
||||
expect(planningStreamManager.hasPendingInitialTurn(sessionId)).toBe(false);
|
||||
|
||||
scripted.releasePrompt();
|
||||
await waitFor(async () => Boolean((await getSession(sessionId))?.currentQuestion));
|
||||
expect((await getSession(sessionId))?.error).toBeUndefined();
|
||||
});
|
||||
|
||||
it("rewind waits for the cancelled turn to release before mutating state", async () => {
|
||||
const scripted = createScriptedAgent();
|
||||
const sessionId = await startSessionAwaitingInput(scripted);
|
||||
const question = (await getSession(sessionId))!.currentQuestion!;
|
||||
|
||||
scripted.holdNextPrompt();
|
||||
const submit = submitResponse(sessionId, { [question.id]: "option-1" });
|
||||
await waitFor(() => scripted.agent.session.prompt.mock.calls.length > 1);
|
||||
|
||||
/*
|
||||
FNXC:PlanningTurnAdmission 2026-07-23-10:10:
|
||||
Rewind aborts the in-flight turn, waits for its owner to release the turn slot AND for
|
||||
the cancelled operation (the held provider prompt) to settle, then republishes the
|
||||
answered question as awaiting input. The prompt is released after rewind starts so the
|
||||
settle-wait resolves deterministically; awaiting `submit` afterwards proves the cancelled
|
||||
turn's post-prompt abort checks bail out without corrupting the rewound state.
|
||||
*/
|
||||
const rewindPromise = rewindSession(sessionId, undefined, "/tmp/project", undefined, MOCK_TASK_STORE);
|
||||
scripted.releasePrompt();
|
||||
const rewound = await rewindPromise;
|
||||
expect(rewound.currentQuestion.id).toBe(question.id);
|
||||
|
||||
await submit;
|
||||
|
||||
const session = await getSession(sessionId);
|
||||
expect(session?.currentQuestion?.id).toBe(question.id);
|
||||
expect(session?.error).toBeUndefined();
|
||||
});
|
||||
|
||||
it("ignores late streaming callbacks from a disposed agent after rewind", async () => {
|
||||
/*
|
||||
FNXC:PlanningTurnAdmission 2026-07-23-10:40:
|
||||
A provider that ignores cancellation beyond rewind's bounded settle-wait can keep
|
||||
streaming into its onThinking/onText callbacks after the agent was disposed. Those late
|
||||
deltas must be inert — no thinkingOutput mutation, no broadcast — or they would corrupt
|
||||
the state published after the rewound question (PR #2417 review finding).
|
||||
*/
|
||||
const scripted = createScriptedAgent();
|
||||
const capturedOnThinking: Array<(delta: string) => void> = [];
|
||||
__setCreateFnAgent(vi.fn(async (options: { onThinking?: (delta: string) => void }) => {
|
||||
if (options?.onThinking) capturedOnThinking.push(options.onThinking);
|
||||
return scripted.agent;
|
||||
}) as never);
|
||||
|
||||
const sessionId = await createSessionWithAgent(
|
||||
"10.0.9.9",
|
||||
"Plan something small",
|
||||
"/tmp/project",
|
||||
MOCK_TASK_STORE,
|
||||
);
|
||||
planningStreamManager.consumeInitialTurn(sessionId)?.();
|
||||
await waitFor(async () => Boolean((await getSession(sessionId))?.currentQuestion));
|
||||
const question = (await getSession(sessionId))!.currentQuestion!;
|
||||
|
||||
scripted.holdNextPrompt();
|
||||
const submit = submitResponse(sessionId, { [question.id]: "option-1" });
|
||||
await waitFor(() => scripted.agent.session.prompt.mock.calls.length > 1);
|
||||
|
||||
const rewindPromise = rewindSession(sessionId, undefined, "/tmp/project", undefined, MOCK_TASK_STORE);
|
||||
scripted.releasePrompt();
|
||||
await rewindPromise;
|
||||
await submit;
|
||||
|
||||
// The first (disposed) agent's streaming callback fires late — it must be a no-op.
|
||||
const staleEvents: unknown[] = [];
|
||||
const unsubscribe = planningStreamManager.subscribe(sessionId, (event) => staleEvents.push(event));
|
||||
expect(capturedOnThinking.length).toBeGreaterThan(0);
|
||||
capturedOnThinking[0]("stale delta after dispose");
|
||||
unsubscribe();
|
||||
|
||||
const session = await getSession(sessionId);
|
||||
expect(session?.thinkingOutput).toBe("");
|
||||
expect(staleEvents).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("emits the retryable parse error without a doubled period", async () => {
|
||||
const scripted = createScriptedAgent();
|
||||
const sessionId = await startSessionAwaitingInput(scripted);
|
||||
const question = (await getSession(sessionId))!.currentQuestion!;
|
||||
|
||||
// Both the original turn and the single reformat retry return JSON-free prose.
|
||||
scripted.setResponder(() => "je ne peux pas produire de JSON ici");
|
||||
await submitResponse(sessionId, { [question.id]: "option-1" });
|
||||
|
||||
const session = await getSession(sessionId);
|
||||
expect(session?.error).toBe(
|
||||
"AI returned no valid JSON. Retry this planning session or start a new one.",
|
||||
);
|
||||
expect(session?.error).not.toContain("..");
|
||||
});
|
||||
});
|
||||
@@ -371,6 +371,15 @@ interface Session {
|
||||
ip: string;
|
||||
initialPlan: string;
|
||||
title: string;
|
||||
/*
|
||||
FNXC:PlanningTurnAdmission 2026-07-23-10:40:
|
||||
Monotonic in-memory epoch for streaming-callback invalidation. Bumped every time the
|
||||
session's agent is disposed/replaced; agent onThinking/onText closures capture the epoch at
|
||||
agent creation and no-op when it has moved on. Closes the residual PR #2417 window where a
|
||||
provider that ignores cancellation beyond rewind's bounded settle-wait could persist or
|
||||
broadcast stale thinking deltas after the rewound state was published. Never persisted.
|
||||
*/
|
||||
agentCallbackEpoch?: number;
|
||||
projectId?: string;
|
||||
/** Workflow selected at session start, retained for agent reconstruction. */
|
||||
workflowId?: string;
|
||||
@@ -457,6 +466,125 @@ interface ActivePlanningGeneration {
|
||||
/** Active planning generations keyed by session ID. */
|
||||
const activeGenerations = new Map<string, ActivePlanningGeneration>();
|
||||
|
||||
/*
|
||||
FNXC:PlanningTurnAdmission 2026-07-22-21:00:
|
||||
Reported bug: "AI returned no valid JSON" recurred whenever the Planning UI was left and
|
||||
re-entered mid-generation (mobile tab switches unmount the view), and generations visibly
|
||||
duplicated. Root cause: the `activeGenerations.has()` guard is check-then-act across several
|
||||
awaits (persistSession/ensureSessionAgent) — the ActivePlanningGeneration record only exists
|
||||
once runGenerationWithTimeout runs. Two overlapping turn entries (re-submitted answer from a
|
||||
remounted view, racing auto-retries, duplicate start of an existing session) both passed the
|
||||
guard; the second displaced the first, and the displaced teardown disposed the session-shared
|
||||
agent that the surviving turn was actively prompting, so the surviving turn read an empty
|
||||
assistant message and failed parse with "AI returned no valid JSON".
|
||||
Invariant: at most one turn may be admitted per session at any time, enforced SYNCHRONOUSLY
|
||||
(no await between check and reservation). Every generation entry point (submitResponse,
|
||||
retrySession, startExistingSession, initializeAgent) must hold a reservation for its full span,
|
||||
so a concurrent entry is rejected with GenerationInProgressError instead of displacing a
|
||||
healthy in-flight generation.
|
||||
*/
|
||||
interface PlanningTurnReservation {
|
||||
done: Promise<void>;
|
||||
resolveDone: () => void;
|
||||
}
|
||||
|
||||
const pendingTurnReservations = new Map<string, PlanningTurnReservation>();
|
||||
|
||||
function isPlanningTurnActive(sessionId: string): boolean {
|
||||
return activeGenerations.has(sessionId) || pendingTurnReservations.has(sessionId);
|
||||
}
|
||||
|
||||
/** Synchronously reserve the session's single turn slot; returns the release fn. */
|
||||
function reservePlanningTurn(sessionId: string): () => void {
|
||||
if (isPlanningTurnActive(sessionId)) {
|
||||
throw new GenerationInProgressError("Generation already in progress");
|
||||
}
|
||||
let resolveDone!: () => void;
|
||||
const done = new Promise<void>((resolve) => {
|
||||
resolveDone = resolve;
|
||||
});
|
||||
const reservation: PlanningTurnReservation = { done, resolveDone };
|
||||
pendingTurnReservations.set(sessionId, reservation);
|
||||
return () => {
|
||||
if (pendingTurnReservations.get(sessionId) === reservation) {
|
||||
pendingTurnReservations.delete(sessionId);
|
||||
}
|
||||
reservation.resolveDone();
|
||||
};
|
||||
}
|
||||
|
||||
/*
|
||||
FNXC:PlanningTurnAdmission 2026-07-23-08:30:
|
||||
User-takes-control actions (rewind/edit) abort the in-flight generation and must then WAIT for
|
||||
that turn's owner to unwind and release its reservation before mutating session state.
|
||||
Deleting the active-generation record without waiting left both admission sets empty while the
|
||||
aborted turn was still unwinding, so a concurrent submit/retry could interleave with rewind's
|
||||
own awaits and corrupt question/history/agent state (review finding on PR #2417).
|
||||
*/
|
||||
/*
|
||||
FNXC:PlanningTurnAdmission 2026-07-23-10:10:
|
||||
The turn reservation is released as soon as the abort wins runGenerationWithTimeout's race,
|
||||
but the generation OPERATION (which holds the raw provider prompt) can still be pending if the
|
||||
provider ignores the AbortSignal. User-takes-control paths (rewind) must also wait — bounded —
|
||||
for that operation to settle before disposing/replacing the agent, or a slow provider callback
|
||||
could still be running against the mutable session while the rewound state is published
|
||||
(review finding on PR #2417). Tracked separately from reservations because settlement can
|
||||
outlive the owner's release.
|
||||
*/
|
||||
const settlingTurnOperations = new Map<string, Promise<void>>();
|
||||
|
||||
function trackTurnOperationSettled(sessionId: string, operationPromise: Promise<unknown>): void {
|
||||
const settled = operationPromise.then(
|
||||
() => {},
|
||||
() => {},
|
||||
);
|
||||
settlingTurnOperations.set(sessionId, settled);
|
||||
void settled.then(() => {
|
||||
if (settlingTurnOperations.get(sessionId) === settled) {
|
||||
settlingTurnOperations.delete(sessionId);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Bounded wait for a cancelled turn's operation (including its provider prompt) to settle.
|
||||
* Returns false on timeout — callers proceed anyway: agent disposal is the backstop, the
|
||||
* cancelled closure's post-prompt abort checks prevent state writes, and blocking a user's
|
||||
* rewind forever on an unresponsive provider would be worse.
|
||||
*/
|
||||
async function waitForTurnOperationSettled(sessionId: string, timeoutMs = 2000): Promise<boolean> {
|
||||
const settling = settlingTurnOperations.get(sessionId);
|
||||
if (!settling) return true;
|
||||
let timer: NodeJS.Timeout | undefined;
|
||||
const timedOut = await Promise.race([
|
||||
settling.then(() => false),
|
||||
new Promise<boolean>((resolve) => {
|
||||
timer = setTimeout(() => resolve(true), timeoutMs);
|
||||
}),
|
||||
]);
|
||||
clearTimeout(timer);
|
||||
return !timedOut;
|
||||
}
|
||||
|
||||
async function waitForPlanningTurnRelease(sessionId: string, timeoutMs = 2000): Promise<boolean> {
|
||||
const deadline = Date.now() + timeoutMs;
|
||||
while (isPlanningTurnActive(sessionId)) {
|
||||
const remaining = deadline - Date.now();
|
||||
if (remaining <= 0) return false;
|
||||
const reservation = pendingTurnReservations.get(sessionId);
|
||||
if (reservation) {
|
||||
await Promise.race([
|
||||
reservation.done,
|
||||
new Promise((resolve) => setTimeout(resolve, Math.min(remaining, 50))),
|
||||
]);
|
||||
} else {
|
||||
// Active generation whose owner has not yet reached its release — brief yield.
|
||||
await new Promise((resolve) => setTimeout(resolve, 10));
|
||||
}
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
// ── AI Session Persistence ────────────────────────────────────────────────
|
||||
|
||||
/** Optional store for persisting session state across reloads/browsers. */
|
||||
@@ -953,6 +1081,11 @@ export class PlanningStreamManager extends EventEmitter {
|
||||
this.pendingInitialTurns.set(sessionId, start);
|
||||
}
|
||||
|
||||
/** True while a registered initial turn has not been consumed by a stream connection yet. */
|
||||
hasPendingInitialTurn(sessionId: string): boolean {
|
||||
return this.pendingInitialTurns.has(sessionId);
|
||||
}
|
||||
|
||||
consumeInitialTurn(sessionId: string): (() => void) | undefined {
|
||||
const start = this.pendingInitialTurns.get(sessionId);
|
||||
if (!start) {
|
||||
@@ -998,7 +1131,16 @@ export class PlanningStreamManager extends EventEmitter {
|
||||
}
|
||||
|
||||
/** Singleton instance of the planning stream manager */
|
||||
export const planningStreamManager = new PlanningStreamManager();
|
||||
/*
|
||||
FNXC:PlanningStreamCatchup 2026-07-22-21:00:
|
||||
Reconnecting clients (mobile tab switches unmount the Planning view, so every return opens a
|
||||
fresh SSE connection) rebuild the loading view exclusively from the buffered-event replay.
|
||||
The default 100-event buffer only held a suffix of a turn's thinking deltas, which forced the
|
||||
client to pre-seed from persisted thinkingOutput and then receive the replay again — the
|
||||
"generation duplicates every time I come back" report. The buffer must be deep enough to hold
|
||||
a full turn of thinking deltas so replay alone reconstructs the view exactly once.
|
||||
*/
|
||||
export const planningStreamManager = new PlanningStreamManager(2000);
|
||||
|
||||
// ── Rate Limiting ───────────────────────────────────────────────────────────
|
||||
|
||||
@@ -1181,8 +1323,8 @@ async function getFirstQuestionFromAgent(
|
||||
throw new InvalidSessionStateError("AI agent not initialized");
|
||||
}
|
||||
|
||||
// Send message to agent
|
||||
await session.agent.session.prompt(message);
|
||||
// Send message to agent (context-limit aware — see promptPlanningAgent)
|
||||
await promptPlanningAgent(session.agent.session, message);
|
||||
|
||||
// Extract response text
|
||||
interface AgentMessage {
|
||||
@@ -1246,7 +1388,8 @@ async function getFirstQuestionFromAgent(
|
||||
|
||||
if (attempt < MAX_PARSE_RETRIES) {
|
||||
try {
|
||||
await session.agent.session.prompt(
|
||||
await promptPlanningAgent(
|
||||
session.agent.session,
|
||||
"Your previous response could not be parsed as JSON. " +
|
||||
'Please respond with ONLY a valid JSON object: {"type":"question","data":{"runningPlan":{...},...}}. ' +
|
||||
"No markdown, no explanation, just the JSON."
|
||||
@@ -1332,7 +1475,8 @@ async function requestMandatoryFirstPlanningQuestion(
|
||||
abortSignal?: AbortSignal,
|
||||
): Promise<{ type: "question"; data: PlanningQuestion }> {
|
||||
try {
|
||||
await (session.agent!.session.prompt as (input: string, options?: { signal?: AbortSignal }) => Promise<void>)(
|
||||
await promptPlanningAgent(
|
||||
session.agent!.session,
|
||||
'Before producing a plan, ask one clarifying question. Return ONLY valid JSON: {"type":"question","data":{...}}.',
|
||||
{ signal: abortSignal },
|
||||
);
|
||||
@@ -1610,8 +1754,32 @@ export async function startExistingSession(
|
||||
session.clarificationEnabled = runtimeOptions.clarificationEnabled === true;
|
||||
session.ntfyConfig = runtimeOptions.ntfyConfig;
|
||||
}
|
||||
|
||||
/*
|
||||
FNXC:PlanningTurnAdmission 2026-07-22-21:00:
|
||||
Starting an existing session must be idempotent while its generation is in flight. A
|
||||
remounted client (mobile tab switches unmount the Planning view) could re-issue
|
||||
start-streaming for a session that is already generating; previously this displaced the
|
||||
live generation and disposed its agent mid-prompt. Now the duplicate start is a no-op and
|
||||
the client simply reconnects to the stream of the run already in progress.
|
||||
*/
|
||||
if (isPlanningTurnActive(sessionId) || planningStreamManager.hasPendingInitialTurn(sessionId)) {
|
||||
diagnostics.warn("Ignoring duplicate start for planning session with an active generation", {
|
||||
sessionId,
|
||||
operation: "start-existing-duplicate",
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
/*
|
||||
FNXC:PlanningTurnAdmission 2026-07-23-08:30:
|
||||
The duplicate-start guard and the initial-turn registration must be one synchronous block —
|
||||
no await between them. With persistSession in between, two concurrent duplicate starts could
|
||||
both pass the guard and the second registerInitialTurn threw "Initial planning turn already
|
||||
registered" for a normal duplicate start (review finding on PR #2417). Registration IS the
|
||||
claim; persistence follows it.
|
||||
*/
|
||||
beginPlanningGeneration(session, "initial_plan");
|
||||
await persistSession(session, "generating");
|
||||
planningStreamManager.registerInitialTurn(sessionId, () => {
|
||||
session.pluginRunner = pluginRunner;
|
||||
initializeAgent(session, rootDir, store, modelProvider, modelId, session.draftThinkingLevel, promptOverrides, pluginRunner).catch((err) => {
|
||||
@@ -1623,6 +1791,7 @@ export async function startExistingSession(
|
||||
});
|
||||
});
|
||||
});
|
||||
await persistSession(session, "generating");
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -1732,6 +1901,25 @@ async function initializeAgent(
|
||||
promptOverrides?: PromptOverrideMap,
|
||||
pluginRunner?: SkillPluginRunner,
|
||||
): Promise<void> {
|
||||
/*
|
||||
FNXC:PlanningTurnAdmission 2026-07-22-21:00:
|
||||
The initial turn reserves the session's single turn slot synchronously (this runs inside
|
||||
consumeInitialTurn's synchronous callback). A duplicate initial turn racing an admitted
|
||||
generation exits quietly instead of displacing it and disposing its agent mid-prompt.
|
||||
*/
|
||||
let releaseTurn: () => void;
|
||||
try {
|
||||
releaseTurn = reservePlanningTurn(session.id);
|
||||
} catch (err) {
|
||||
if (err instanceof GenerationInProgressError) {
|
||||
diagnostics.warn("Skipping duplicate initial planning turn — generation already active", {
|
||||
sessionId: session.id,
|
||||
operation: "initialize-agent-duplicate",
|
||||
});
|
||||
return;
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
try {
|
||||
await runGenerationWithTimeout(session, async (abortSignal) => {
|
||||
/*
|
||||
@@ -1789,6 +1977,8 @@ async function initializeAgent(
|
||||
type: "error",
|
||||
data: errorMessage,
|
||||
});
|
||||
} finally {
|
||||
releaseTurn();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1809,6 +1999,16 @@ async function createPlanningAgent(
|
||||
|
||||
const skillContext = buildSessionSkillContextSync(null, "executor", rootDir, pluginRunner);
|
||||
|
||||
/*
|
||||
FNXC:PlanningTurnAdmission 2026-07-23-10:40:
|
||||
Capture the callback epoch at agent creation. A provider that ignores cancellation beyond
|
||||
rewind's bounded settle-wait can keep streaming after this agent is disposed; the epoch
|
||||
check makes those late deltas inert (no thinkingOutput mutation, no persist, no broadcast)
|
||||
instead of letting them corrupt the state published after the agent was replaced.
|
||||
*/
|
||||
const callbackEpoch = session.agentCallbackEpoch ?? 0;
|
||||
const callbacksInvalidated = (): boolean => (session.agentCallbackEpoch ?? 0) !== callbackEpoch;
|
||||
|
||||
/*
|
||||
FNXC:PlanningSkills 2026-06-17-19:33:
|
||||
Streaming planning sessions share the executor skill contract because custom planning/workflow tools can benefit from agent-declared skills and enabled plugin skills exactly like task execution.
|
||||
@@ -1838,6 +2038,7 @@ async function createPlanningAgent(
|
||||
: {}),
|
||||
...(thinkingLevel ? { defaultThinkingLevel: thinkingLevel } : {}),
|
||||
onThinking: (delta: string) => {
|
||||
if (callbacksInvalidated()) return;
|
||||
markPlanningGenerationProgress(session.id, delta);
|
||||
session.thinkingOutput += delta;
|
||||
persistThinking(session.id, session.thinkingOutput);
|
||||
@@ -1850,6 +2051,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.
|
||||
if (callbacksInvalidated()) return;
|
||||
markPlanningGenerationProgress(session.id, delta);
|
||||
session.thinkingOutput += delta;
|
||||
persistThinking(session.id, session.thinkingOutput);
|
||||
@@ -1919,7 +2121,7 @@ async function ensureSessionAgent(
|
||||
if (abortSignal.aborted) {
|
||||
throw createAbortError();
|
||||
}
|
||||
await (session.agent!.session.prompt as (input: string, options?: { signal?: AbortSignal }) => Promise<void>)(contextMessage, {
|
||||
await promptPlanningAgent(session.agent!.session, contextMessage, {
|
||||
signal: abortSignal,
|
||||
});
|
||||
if (abortSignal.aborted) {
|
||||
@@ -1968,8 +2170,11 @@ Planning Mode malformed AI output must either recover through the bounded reform
|
||||
*/
|
||||
|
||||
function buildRetryableParseErrorMessage(error: Error | undefined): string {
|
||||
// Strip any trailing "Please try again." suffix AND trailing periods so the appended
|
||||
// call to action cannot render as "…no valid JSON.. Retry this planning session…".
|
||||
const baseMessage = (error?.message || "Failed to parse AI response")
|
||||
.replace(/\s*Please try again\.?\s*$/i, "")
|
||||
.replace(/[.\s]+$/, "")
|
||||
.trim();
|
||||
return `${baseMessage}. Retry this planning session or start a new one.`;
|
||||
}
|
||||
@@ -1997,6 +2202,37 @@ function createAbortError(): Error {
|
||||
return error;
|
||||
}
|
||||
|
||||
/*
|
||||
FNXC:PlanningContextCompaction 2026-07-22-22:40:
|
||||
Long planning interviews accumulate the whole Q/A history in one agent session and can hit the
|
||||
model's context window mid-interview. Planning previously called session.prompt() raw, so a
|
||||
context overflow surfaced as a terminal session error — and the auto-retry then replayed the
|
||||
FULL history into a fresh agent, overflowing again, unrecoverably. Every planning prompt now
|
||||
routes through the engine's promptWithFallback, which classifies context-limit errors and
|
||||
recovers via prompt/memory compaction and session.compact() before retrying. Test fakes that
|
||||
mock @fusion/engine without promptWithFallback fall back to the raw prompt unchanged.
|
||||
*/
|
||||
async function promptPlanningAgent(
|
||||
agentSession: { prompt: (input: string, options?: { signal?: AbortSignal }) => Promise<void> },
|
||||
message: string,
|
||||
options?: { signal?: AbortSignal },
|
||||
): Promise<void> {
|
||||
let promptWithFallback: ((session: unknown, prompt: string, options?: unknown) => Promise<void>) | undefined;
|
||||
try {
|
||||
promptWithFallback = (engineModule as {
|
||||
promptWithFallback?: (session: unknown, prompt: string, options?: unknown) => Promise<void>;
|
||||
}).promptWithFallback;
|
||||
} catch {
|
||||
// vi.mock("@fusion/engine") proxies throw on undeclared exports; treat as unavailable.
|
||||
promptWithFallback = undefined;
|
||||
}
|
||||
if (typeof promptWithFallback === "function") {
|
||||
await promptWithFallback(agentSession, message, options);
|
||||
return;
|
||||
}
|
||||
await agentSession.prompt(message, options);
|
||||
}
|
||||
|
||||
function normalizeGenerationProgress(output: string): string {
|
||||
return output.replace(/\s+/g, " ").trim();
|
||||
}
|
||||
@@ -2087,7 +2323,11 @@ async function runGenerationWithTimeout<T>(session: Session, operation: (abortSi
|
||||
});
|
||||
|
||||
try {
|
||||
return await Promise.race([operation(abortController.signal), abortPromise]);
|
||||
// Track the operation's own settlement: on abort the race settles immediately while the
|
||||
// operation (and its provider prompt) may still be pending — rewind waits on this.
|
||||
const operationPromise = operation(abortController.signal);
|
||||
trackTurnOperationSettled(session.id, operationPromise);
|
||||
return await Promise.race([operationPromise, abortPromise]);
|
||||
} finally {
|
||||
clearTimeout(generationRecord.timer);
|
||||
if (activeGenerations.get(session.id) === generationRecord) {
|
||||
@@ -2319,7 +2559,7 @@ async function continueAgentConversation(session: Session, message: string): Pro
|
||||
if (abortSignal.aborted) {
|
||||
throw createAbortError();
|
||||
}
|
||||
await (session.agent.session.prompt as (input: string, options?: { signal?: AbortSignal }) => Promise<void>)(message, {
|
||||
await promptPlanningAgent(session.agent.session, message, {
|
||||
signal: abortSignal,
|
||||
});
|
||||
if (abortSignal.aborted) {
|
||||
@@ -2393,7 +2633,8 @@ async function continueAgentConversation(session: Session, message: string): Pro
|
||||
if (abortSignal.aborted) {
|
||||
throw createAbortError();
|
||||
}
|
||||
await (session.agent.session.prompt as (input: string, options?: { signal?: AbortSignal }) => Promise<void>)(
|
||||
await promptPlanningAgent(
|
||||
session.agent.session,
|
||||
"Your previous response could not be parsed as JSON. " +
|
||||
'Please respond with ONLY a valid JSON object: {"type":"question","data":{"runningPlan":{...},...}}. ' +
|
||||
'No markdown, no explanation, just the JSON.',
|
||||
@@ -2788,12 +3029,14 @@ export async function submitResponse(
|
||||
if (store && !session.store) session.store = store;
|
||||
if (rootDir && !session.rootDir) session.rootDir = rootDir;
|
||||
|
||||
if (activeGenerations.has(session.id)) {
|
||||
// FNXC:PlanningTurnAdmission 2026-07-22-21:00: synchronous single-turn admission — see reservePlanningTurn.
|
||||
if (isPlanningTurnActive(session.id)) {
|
||||
if (didSubmitSameAnswer(session, responses)) {
|
||||
throw new GenerationInProgressError("Generation already in progress for this response");
|
||||
}
|
||||
throw new GenerationInProgressError("Generation already in progress");
|
||||
}
|
||||
const releaseTurn = reservePlanningTurn(session.id);
|
||||
|
||||
/*
|
||||
FNXC:PlanningRetry 2026-07-14-00:00:
|
||||
@@ -2810,69 +3053,73 @@ export async function submitResponse(
|
||||
*/
|
||||
let answeredQuestion: PlanningQuestion | undefined;
|
||||
|
||||
if (isRefineRequest(responses) && session.summary) {
|
||||
// Refinement steers which question comes next; it is never an answer to the
|
||||
// currently displayed question and therefore must not create a history entry.
|
||||
beginPlanningGeneration(session, "question");
|
||||
session.currentQuestion = undefined;
|
||||
session.error = undefined;
|
||||
await persistSession(session, "generating");
|
||||
try {
|
||||
if (isRefineRequest(responses) && session.summary) {
|
||||
// Refinement steers which question comes next; it is never an answer to the
|
||||
// currently displayed question and therefore must not create a history entry.
|
||||
beginPlanningGeneration(session, "question");
|
||||
session.currentQuestion = undefined;
|
||||
session.error = undefined;
|
||||
await persistSession(session, "generating");
|
||||
|
||||
await ensureSessionAgent(session, rootDir, session.history, promptOverrides, store);
|
||||
const focus = typeof responses.focus === "string" ? responses.focus.trim() : undefined;
|
||||
const refineMessage = formatRefineRequestForAgent(session.summary, focus);
|
||||
await continueAgentConversation(session, refineMessage);
|
||||
} else if (!session.currentQuestion) {
|
||||
throw new InvalidSessionStateError("No active question in session");
|
||||
} else {
|
||||
const currentQuestion = captureOtherCustomText(session.currentQuestion, responses);
|
||||
const historyEntry = {
|
||||
question: currentQuestion,
|
||||
response: responses,
|
||||
thinkingOutput: session.lastGeneratedThinking || "",
|
||||
};
|
||||
|
||||
session.error = undefined;
|
||||
/*
|
||||
FNXC:DashboardSessionPersistence 2026-06-14-09:09:
|
||||
Persist the user's answered planning turn before the agent generates the next question or errors. AiSessionStore snapshots happen inside continueAgentConversation, so history must already include the submitted answer for retry replay and SQLite round-trip tests to observe durable state.
|
||||
*/
|
||||
const editIndex = session.editingQuestionId
|
||||
? session.history.findIndex((entry) => entry.question.id === session.editingQuestionId)
|
||||
: -1;
|
||||
const isEditingPriorAnswer = editIndex >= 0;
|
||||
if (isEditingPriorAnswer) {
|
||||
session.history[editIndex] = historyEntry;
|
||||
session.editingQuestionId = undefined;
|
||||
// Rebuild from history before the next turn so stale pre-edit plan prose cannot survive.
|
||||
session.summary = buildRunningSummary(session.initialPlan, session.history);
|
||||
// Existing agent context contains the old answer; rebuild it from the preserved history.
|
||||
disposeSessionAgentForRetry(session);
|
||||
await ensureSessionAgent(session, rootDir, session.history, promptOverrides, store);
|
||||
const focus = typeof responses.focus === "string" ? responses.focus.trim() : undefined;
|
||||
const refineMessage = formatRefineRequestForAgent(session.summary, focus);
|
||||
await continueAgentConversation(session, refineMessage);
|
||||
} else if (!session.currentQuestion) {
|
||||
throw new InvalidSessionStateError("No active question in session");
|
||||
} else {
|
||||
session.history.push(historyEntry);
|
||||
}
|
||||
answeredQuestion = currentQuestion;
|
||||
const currentQuestion = captureOtherCustomText(session.currentQuestion, responses);
|
||||
const historyEntry = {
|
||||
question: currentQuestion,
|
||||
response: responses,
|
||||
thinkingOutput: session.lastGeneratedThinking || "",
|
||||
};
|
||||
|
||||
// Clear the answered question while generation is active so reconnects cannot replay it.
|
||||
// The completed turn persists and broadcasts exactly one newly generated question.
|
||||
beginPlanningGeneration(session, "plan_update");
|
||||
session.currentQuestion = undefined;
|
||||
await persistSession(session, "generating");
|
||||
if (!session.agent) {
|
||||
// An edited older answer must be replayed in its original position with every
|
||||
// later answer retained; only a newly appended answer is sent after replay.
|
||||
await ensureSessionAgent(
|
||||
session,
|
||||
rootDir,
|
||||
isEditingPriorAnswer ? session.history : session.history.slice(0, -1),
|
||||
promptOverrides,
|
||||
store,
|
||||
);
|
||||
session.error = undefined;
|
||||
/*
|
||||
FNXC:DashboardSessionPersistence 2026-06-14-09:09:
|
||||
Persist the user's answered planning turn before the agent generates the next question or errors. AiSessionStore snapshots happen inside continueAgentConversation, so history must already include the submitted answer for retry replay and SQLite round-trip tests to observe durable state.
|
||||
*/
|
||||
const editIndex = session.editingQuestionId
|
||||
? session.history.findIndex((entry) => entry.question.id === session.editingQuestionId)
|
||||
: -1;
|
||||
const isEditingPriorAnswer = editIndex >= 0;
|
||||
if (isEditingPriorAnswer) {
|
||||
session.history[editIndex] = historyEntry;
|
||||
session.editingQuestionId = undefined;
|
||||
// Rebuild from history before the next turn so stale pre-edit plan prose cannot survive.
|
||||
session.summary = buildRunningSummary(session.initialPlan, session.history);
|
||||
// Existing agent context contains the old answer; rebuild it from the preserved history.
|
||||
disposeSessionAgentForRetry(session);
|
||||
} else {
|
||||
session.history.push(historyEntry);
|
||||
}
|
||||
answeredQuestion = currentQuestion;
|
||||
|
||||
// Clear the answered question while generation is active so reconnects cannot replay it.
|
||||
// The completed turn persists and broadcasts exactly one newly generated question.
|
||||
beginPlanningGeneration(session, "plan_update");
|
||||
session.currentQuestion = undefined;
|
||||
await persistSession(session, "generating");
|
||||
if (!session.agent) {
|
||||
// An edited older answer must be replayed in its original position with every
|
||||
// later answer retained; only a newly appended answer is sent after replay.
|
||||
await ensureSessionAgent(
|
||||
session,
|
||||
rootDir,
|
||||
isEditingPriorAnswer ? session.history : session.history.slice(0, -1),
|
||||
promptOverrides,
|
||||
store,
|
||||
);
|
||||
}
|
||||
const message = isEditingPriorAnswer
|
||||
? "An earlier answer was edited. Use the complete preserved interview context above, regenerate the running plan, and ask exactly one next question."
|
||||
: formatResponseForAgent(currentQuestion, responses);
|
||||
await continueAgentConversation(session, message);
|
||||
}
|
||||
const message = isEditingPriorAnswer
|
||||
? "An earlier answer was edited. Use the complete preserved interview context above, regenerate the running plan, and ask exactly one next question."
|
||||
: formatResponseForAgent(currentQuestion, responses);
|
||||
await continueAgentConversation(session, message);
|
||||
} finally {
|
||||
releaseTurn();
|
||||
}
|
||||
|
||||
// Return the current state (will be updated via SSE)
|
||||
@@ -2927,37 +3174,49 @@ export async function retrySession(
|
||||
throw new InvalidSessionStateError(`Planning session ${sessionId} is not in an error state`);
|
||||
}
|
||||
|
||||
disposeSessionAgentForRetry(session);
|
||||
|
||||
session.error = undefined;
|
||||
session.summary = undefined;
|
||||
/*
|
||||
FNXC:PlanningRetry 2026-07-14-00:00:
|
||||
A retry regenerates the last turn, so no question is awaiting input. Clearing here also
|
||||
scrubs stale answered questions persisted by pre-fix builds; without this, the fresh SSE
|
||||
connection the retry path opens would be handed the answered question by the stream route's
|
||||
catch-up emit, resetting the FN-7946 auto-retry budget and looping forever.
|
||||
FNXC:PlanningTurnAdmission 2026-07-22-21:00:
|
||||
Two racing retries (e.g. auto-retry from a remounted Planning view plus a second tab) could
|
||||
both read status "error" before either persisted "generating"; the loser then disposed the
|
||||
agent the winner was actively prompting, producing the empty-response "AI returned no valid
|
||||
JSON" failure. Admission is reserved synchronously before any turn state is touched.
|
||||
*/
|
||||
session.currentQuestion = undefined;
|
||||
session.updatedAt = new Date();
|
||||
beginPlanningGeneration(session, session.history.length === 0 ? "initial_plan" : "plan_update");
|
||||
await persistSession(session, "generating");
|
||||
const releaseTurn = reservePlanningTurn(session.id);
|
||||
try {
|
||||
disposeSessionAgentForRetry(session);
|
||||
|
||||
if (session.history.length === 0) {
|
||||
await ensureSessionAgent(session, rootDir, [], promptOverrides, store);
|
||||
await continueAgentConversation(session, formatInitialRunningPlanRequestForAgent(session.initialPlan));
|
||||
return;
|
||||
session.error = undefined;
|
||||
session.summary = undefined;
|
||||
/*
|
||||
FNXC:PlanningRetry 2026-07-14-00:00:
|
||||
A retry regenerates the last turn, so no question is awaiting input. Clearing here also
|
||||
scrubs stale answered questions persisted by pre-fix builds; without this, the fresh SSE
|
||||
connection the retry path opens would be handed the answered question by the stream route's
|
||||
catch-up emit, resetting the FN-7946 auto-retry budget and looping forever.
|
||||
*/
|
||||
session.currentQuestion = undefined;
|
||||
session.updatedAt = new Date();
|
||||
beginPlanningGeneration(session, session.history.length === 0 ? "initial_plan" : "plan_update");
|
||||
await persistSession(session, "generating");
|
||||
|
||||
if (session.history.length === 0) {
|
||||
await ensureSessionAgent(session, rootDir, [], promptOverrides, store);
|
||||
await continueAgentConversation(session, formatInitialRunningPlanRequestForAgent(session.initialPlan));
|
||||
return;
|
||||
}
|
||||
|
||||
const replayHistory = session.history.slice(0, -1);
|
||||
const lastEntry = session.history[session.history.length - 1];
|
||||
|
||||
await ensureSessionAgent(session, rootDir, replayHistory, promptOverrides, store);
|
||||
const replayMessage = formatResponseForAgent(
|
||||
lastEntry.question,
|
||||
coerceResponseRecord(lastEntry.question, lastEntry.response),
|
||||
);
|
||||
await continueAgentConversation(session, replayMessage);
|
||||
} finally {
|
||||
releaseTurn();
|
||||
}
|
||||
|
||||
const replayHistory = session.history.slice(0, -1);
|
||||
const lastEntry = session.history[session.history.length - 1];
|
||||
|
||||
await ensureSessionAgent(session, rootDir, replayHistory, promptOverrides, store);
|
||||
const replayMessage = formatResponseForAgent(
|
||||
lastEntry.question,
|
||||
coerceResponseRecord(lastEntry.question, lastEntry.response),
|
||||
);
|
||||
await continueAgentConversation(session, replayMessage);
|
||||
}
|
||||
|
||||
export interface PlanningRewindResult {
|
||||
@@ -2988,38 +3247,87 @@ export async function rewindSession(
|
||||
throw new InvalidSessionStateError("Planning session has no previous question to rewind to");
|
||||
}
|
||||
|
||||
const rewindIndex = questionId
|
||||
? session.history.findIndex((entry) => entry.question.id === questionId)
|
||||
: session.history.length - 1;
|
||||
if (rewindIndex < 0) {
|
||||
throw new InvalidSessionStateError("Planning question to edit was not found");
|
||||
/*
|
||||
FNXC:PlanningTurnAdmission 2026-07-22-21:00:
|
||||
Rewind/edit is a user-takes-control action. Cancel any in-flight generation through its own
|
||||
abort/teardown path (like validateSession) before disposing the agent, so the turn cannot
|
||||
keep running against a disposed session and surface "AI returned no valid JSON".
|
||||
|
||||
FNXC:PlanningTurnAdmission 2026-07-23-08:30:
|
||||
After the abort, WAIT for the cancelled turn's owner to unwind and release its reservation,
|
||||
then hold the reservation for rewind's own span. Without this, both admission sets were
|
||||
empty during rewind's awaits and a concurrent submit/retry could interleave and corrupt
|
||||
question/history/agent state; the cancelled turn's post-prompt abort checks plus this
|
||||
serialization keep a slow provider prompt from outliving the rewound state (PR #2417).
|
||||
All state mutation (including history.pop) happens only after admission succeeds.
|
||||
*/
|
||||
const activeGeneration = activeGenerations.get(session.id);
|
||||
if (activeGeneration) {
|
||||
activeGeneration.abortReason = "user-stop";
|
||||
clearTimeout(activeGeneration.timer);
|
||||
activeGeneration.abortTeardown();
|
||||
activeGeneration.abortController.abort();
|
||||
activeGenerations.delete(session.id);
|
||||
}
|
||||
const rewindEntry = session.history[rewindIndex]!;
|
||||
if (!questionId) session.history.pop();
|
||||
|
||||
disposeSessionAgentForRetry(session);
|
||||
|
||||
session.currentQuestion = rewindEntry.question;
|
||||
session.editingQuestionId = questionId ? questionId : undefined;
|
||||
// Re-derive from retained answers so an edit cannot revive a prior question as a deliverable.
|
||||
session.summary = buildRunningSummary(session.initialPlan, session.history);
|
||||
session.error = undefined;
|
||||
session.lastGeneratedThinking = session.history[session.history.length - 1]?.thinkingOutput ?? "";
|
||||
session.thinkingOutput = "";
|
||||
session.updatedAt = new Date();
|
||||
|
||||
if (!session.agent && rootDir) {
|
||||
await ensureSessionAgent(session, rootDir, session.history, promptOverrides, store);
|
||||
if (!(await waitForPlanningTurnRelease(session.id))) {
|
||||
throw new GenerationInProgressError("Generation already in progress");
|
||||
}
|
||||
const releaseTurn = reservePlanningTurn(session.id);
|
||||
|
||||
persistSession(session, "awaiting_input");
|
||||
planningStreamManager.broadcast(session.id, { type: "summary", data: session.summary });
|
||||
planningStreamManager.broadcast(session.id, { type: "question", data: rewindEntry.question });
|
||||
try {
|
||||
/*
|
||||
FNXC:PlanningTurnAdmission 2026-07-23-10:10:
|
||||
Also wait — bounded — for the cancelled turn's OPERATION (including its raw provider
|
||||
prompt) to settle before disposing/replacing the agent. The reservation releases as soon
|
||||
as the abort wins the race, but a provider that ignores the signal can leave its prompt
|
||||
callback live; publishing the rewound state under it was the residual PR #2417 finding.
|
||||
On timeout we proceed anyway: disposal is the backstop and the cancelled closure's
|
||||
post-prompt abort checks prevent state writes.
|
||||
*/
|
||||
if (!(await waitForTurnOperationSettled(session.id))) {
|
||||
diagnostics.warn("Rewinding past a provider prompt that has not settled after abort", {
|
||||
sessionId: session.id,
|
||||
operation: "rewind-unsettled-prompt",
|
||||
});
|
||||
}
|
||||
|
||||
return {
|
||||
currentQuestion: rewindEntry.question,
|
||||
history: [...session.history],
|
||||
};
|
||||
// Resolve the rewind target only after admission: a turn that completed while we
|
||||
// waited may have appended to history, so an index computed earlier would be stale.
|
||||
const rewindIndex = questionId
|
||||
? session.history.findIndex((entry) => entry.question.id === questionId)
|
||||
: session.history.length - 1;
|
||||
if (rewindIndex < 0) {
|
||||
throw new InvalidSessionStateError("Planning question to edit was not found");
|
||||
}
|
||||
const rewindEntry = session.history[rewindIndex]!;
|
||||
if (!questionId) session.history.pop();
|
||||
|
||||
disposeSessionAgentForRetry(session);
|
||||
|
||||
session.currentQuestion = rewindEntry.question;
|
||||
session.editingQuestionId = questionId ? questionId : undefined;
|
||||
// Re-derive from retained answers so an edit cannot revive a prior question as a deliverable.
|
||||
session.summary = buildRunningSummary(session.initialPlan, session.history);
|
||||
session.error = undefined;
|
||||
session.lastGeneratedThinking = session.history[session.history.length - 1]?.thinkingOutput ?? "";
|
||||
session.thinkingOutput = "";
|
||||
session.updatedAt = new Date();
|
||||
|
||||
if (!session.agent && rootDir) {
|
||||
await ensureSessionAgent(session, rootDir, session.history, promptOverrides, store);
|
||||
}
|
||||
|
||||
persistSession(session, "awaiting_input");
|
||||
planningStreamManager.broadcast(session.id, { type: "summary", data: session.summary });
|
||||
planningStreamManager.broadcast(session.id, { type: "question", data: rewindEntry.question });
|
||||
|
||||
return {
|
||||
currentQuestion: rewindEntry.question,
|
||||
history: [...session.history],
|
||||
};
|
||||
} finally {
|
||||
releaseTurn();
|
||||
}
|
||||
}
|
||||
|
||||
export function stopGeneration(sessionId: string): boolean {
|
||||
@@ -3157,6 +3465,9 @@ function coerceResponseRecord(question: PlanningQuestion, response: unknown): Re
|
||||
}
|
||||
|
||||
function disposeSessionAgentForRetry(session: Session): void {
|
||||
// FNXC:PlanningTurnAdmission 2026-07-23-10:40: invalidate the disposed agent's streaming
|
||||
// callbacks even when the provider keeps running past dispose — see Session.agentCallbackEpoch.
|
||||
session.agentCallbackEpoch = (session.agentCallbackEpoch ?? 0) + 1;
|
||||
if (!session.agent) {
|
||||
return;
|
||||
}
|
||||
@@ -3570,6 +3881,8 @@ export function __resetPlanningState(): void {
|
||||
rateLimits.clear();
|
||||
planningStreamManager.reset();
|
||||
activeGenerations.clear();
|
||||
pendingTurnReservations.clear();
|
||||
settlingTurnOperations.clear();
|
||||
|
||||
if (_aiSessionStore && _aiSessionDeletedListener) {
|
||||
_aiSessionStore.off("ai_session:deleted", _aiSessionDeletedListener);
|
||||
|
||||
Reference in New Issue
Block a user