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

fix(dashboard,core): bound AI session prompt() with timeout + abort

Browse Source 
Subtask, mission-interview, and milestone/slice-interview sessions could pin
their `generating` state forever when the underlying provider stream stalled
silently or a tool call hung. Wrap each `agent.session.prompt()` in a new
GenerationGuard helper (per-session AbortController + timer) so a stuck turn
becomes a bounded error users can retry. Adds matching `stop*Generation`
exports and threads abort through cleanup so dismissing a modal cancels the
in-flight call instead of leaking it.

Also closes the gh-cli tool hang vector: `runGhAsync` / `runGhJsonAsync` now
accept `{ signal, timeoutMs }` (default 30s). Github-touching extension tools
forward the AI tool's signal so an aborted agent kills the `gh` child instead
of orphaning it.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
gsxdsm
2026-04-28 08:05:12 -07:00
parent 596982dd75
commit 2f7ba29ead
11 changed files with 1164 additions and 266 deletions
Download Patch File Download Diff File Expand all files Collapse all files

132
packages/dashboard/src/__tests__/ai-session-timeout.test.ts Normal file
View File

@@ -0,0 +1,132 @@
// @vitest-environment node
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { GenerationGuard, isAbortError, createAbortError } from "../ai-session-timeout.js";
beforeEach(() => {
vi.useFakeTimers();
});
afterEach(() => {
vi.useRealTimers();
});
describe("GenerationGuard", () => {
it("resolves with the operation result when it completes before the timeout", async () => {
const guard = new GenerationGuard();
const onTimeout = vi.fn();
const result = await guard.run("s1", 1_000, { onTimeout }, async () => "ok");
expect(result).toBe("ok");
expect(onTimeout).not.toHaveBeenCalled();
expect(guard.has("s1")).toBe(false);
});
it("fires onTimeout, aborts the operation, and rejects with AbortError when the timer fires", async () => {
const guard = new GenerationGuard();
const onTimeout = vi.fn();
let opSettled = false;
const promise = guard.run("s1", 1_000, { onTimeout }, async () => {
// Simulate a hung prompt() that never resolves on its own.
await new Promise<void>(() => { /* intentionally never resolves */ });
opSettled = true;
return "should-not-reach";
});
// Catch promise rejection now so it doesn't become unhandled when timers advance.
const settled = promise.then(
(v) => ({ ok: true as const, v }),
(err) => ({ ok: false as const, err }),
);
await vi.advanceTimersByTimeAsync(1_000);
const outcome = await settled;
expect(outcome.ok).toBe(false);
expect(outcome.ok ? null : outcome.err).toSatisfy((err: unknown) => isAbortError(err));
expect(onTimeout).toHaveBeenCalledTimes(1);
expect(opSettled).toBe(false);
expect(guard.has("s1")).toBe(false);
});
it("fires onUserStop (not onTimeout) when stop() aborts before the timer", async () => {
const guard = new GenerationGuard();
const onTimeout = vi.fn();
const onUserStop = vi.fn();
const promise = guard.run("s1", 10_000, { onTimeout, onUserStop }, async () => {
await new Promise<void>(() => { /* hang */ });
});
const settled = promise.then(
() => ({ ok: true as const }),
(err) => ({ ok: false as const, err }),
);
// Stop before the timer fires.
expect(guard.stop("s1")).toBe(true);
const outcome = await settled;
expect(outcome.ok).toBe(false);
expect(onTimeout).not.toHaveBeenCalled();
expect(onUserStop).toHaveBeenCalledTimes(1);
// Subsequent stop is a no-op.
expect(guard.stop("s1")).toBe(false);
});
it("re-entrant run for the same id aborts the prior generation", async () => {
const guard = new GenerationGuard();
const firstUserStop = vi.fn();
const first = guard.run(
"s1",
10_000,
{ onTimeout: vi.fn(), onUserStop: firstUserStop },
async () => { await new Promise<void>(() => { /* hang */ }); },
);
const firstSettled = first.then(
() => ({ ok: true as const }),
(err) => ({ ok: false as const, err }),
);
const second = guard.run("s1", 10_000, { onTimeout: vi.fn() }, async () => "fresh");
await expect(second).resolves.toBe("fresh");
const firstOutcome = await firstSettled;
expect(firstOutcome.ok).toBe(false);
// Re-entrant cancellation goes through the internal cancel path, not the
// user-facing stop, so onUserStop is intentionally not fired for the
// displaced generation. The displaced caller still observes AbortError.
expect(firstUserStop).not.toHaveBeenCalled();
expect(guard.has("s1")).toBe(false);
});
it("reset() aborts every active generation", async () => {
const guard = new GenerationGuard();
const a = guard.run("a", 10_000, { onTimeout: vi.fn() }, async () => {
await new Promise<void>(() => { /* hang */ });
});
const b = guard.run("b", 10_000, { onTimeout: vi.fn() }, async () => {
await new Promise<void>(() => { /* hang */ });
});
const aSettled = a.catch((err) => err);
const bSettled = b.catch((err) => err);
guard.reset();
expect(isAbortError(await aSettled)).toBe(true);
expect(isAbortError(await bSettled)).toBe(true);
expect(guard.has("a")).toBe(false);
expect(guard.has("b")).toBe(false);
});
it("createAbortError is identifiable via isAbortError", () => {
expect(isAbortError(createAbortError())).toBe(true);
expect(isAbortError(new Error("other"))).toBe(false);
expect(isAbortError("not an error")).toBe(false);
});
});

81
packages/dashboard/src/__tests__/milestone-slice-interview.test.ts
View File

@@ -33,6 +33,8 @@ import {
skipTargetInterview,
MILESTONE_INTERVIEW_SYSTEM_PROMPT,
SLICE_INTERVIEW_SYSTEM_PROMPT,
stopMilestoneSliceInterviewGeneration,
GENERATION_TIMEOUT_MS,
type MilestoneInterviewSummary,
type SliceInterviewSummary,
} from "../milestone-slice-interview.js";
@@ -769,4 +771,83 @@ describe("milestone-slice-interview module", () => {
expect(SLICE_INTERVIEW_SYSTEM_PROMPT).toContain("acceptanceCriteria");
});
});
describe("generation timeout / abort", () => {
it("marks the session as error when initial generation exceeds GENERATION_TIMEOUT_MS", async () => {
vi.useFakeTimers();
let resolveHungPrompt: (() => void) | undefined;
mockCreateFnAgent.mockImplementationOnce(async () => ({
session: {
state: { messages: [] },
prompt: vi.fn(async () => {
await new Promise<void>((resolve) => { resolveHungPrompt = resolve; });
}),
dispose: vi.fn(),
},
}));
const sessionId = await createTargetInterviewSession(
"10.0.1.10",
"milestone",
"milestone-stuck",
"Hung milestone interview",
undefined,
"/tmp/project",
);
// Yield so the guard registration runs.
await Promise.resolve();
await Promise.resolve();
await vi.advanceTimersByTimeAsync(GENERATION_TIMEOUT_MS);
await vi.advanceTimersByTimeAsync(0);
const session = getTargetInterviewSession(sessionId);
expect(session?.error).toMatch(/timed out/i);
resolveHungPrompt?.();
await vi.advanceTimersByTimeAsync(0);
vi.useRealTimers();
});
it("stopMilestoneSliceInterviewGeneration aborts an in-flight session and marks it stopped", async () => {
let resolveHungPrompt: (() => void) | undefined;
mockCreateFnAgent.mockImplementationOnce(async () => ({
session: {
state: { messages: [] },
prompt: vi.fn(async () => {
await new Promise<void>((resolve) => { resolveHungPrompt = resolve; });
}),
dispose: vi.fn(),
},
}));
const sessionId = await createTargetInterviewSession(
"10.0.1.11",
"slice",
"slice-stoppable",
"Stoppable slice interview",
undefined,
"/tmp/project",
);
let stopped = false;
for (let i = 0; i < 50 && !stopped; i++) {
stopped = stopMilestoneSliceInterviewGeneration(sessionId);
if (!stopped) await new Promise((resolve) => setTimeout(resolve, 5));
}
expect(stopped).toBe(true);
await new Promise((resolve) => setTimeout(resolve, 0));
const session = getTargetInterviewSession(sessionId);
expect(session?.error).toMatch(/stopped by user/i);
expect(stopMilestoneSliceInterviewGeneration(sessionId)).toBe(false);
resolveHungPrompt?.();
await new Promise((resolve) => setTimeout(resolve, 0));
});
});
});

83
packages/dashboard/src/__tests__/mission-interview.test.ts
View File

@@ -27,7 +27,9 @@ import {
setAiSessionStore,
RateLimitError,
SessionNotFoundError,
stopMissionInterviewGeneration,
submitMissionInterviewResponse,
GENERATION_TIMEOUT_MS,
} from "../mission-interview.js";
import {
setDiagnosticsSink,
@@ -927,4 +929,85 @@ describe("mission-interview module", () => {
expect(lastCall[0].systemPrompt).toMatch(/^You are a mission planning assistant/);
});
});
describe("generation timeout / abort", () => {
it("marks the session as error when initial generation exceeds GENERATION_TIMEOUT_MS", async () => {
vi.useFakeTimers();
let resolveHungPrompt: (() => void) | undefined;
mockCreateFnAgent.mockImplementationOnce(async () => ({
session: {
state: { messages: [] },
prompt: vi.fn(async () => {
await new Promise<void>((resolve) => { resolveHungPrompt = resolve; });
}),
dispose: vi.fn(),
},
}));
const sessionId = await createMissionInterviewSession(
"10.0.0.10",
"Hung mission interview",
"/tmp/project",
);
// Yield so initializeAgent's async work registers the generation guard.
await Promise.resolve();
await Promise.resolve();
await vi.advanceTimersByTimeAsync(GENERATION_TIMEOUT_MS);
await vi.advanceTimersByTimeAsync(0);
const session = getMissionInterviewSession(sessionId);
expect(session?.error).toMatch(/timed out/i);
resolveHungPrompt?.();
await vi.advanceTimersByTimeAsync(0);
vi.useRealTimers();
});
it("stopMissionInterviewGeneration aborts an in-flight session and marks it stopped", async () => {
// Real timers — we want the guard.run() registration to actually happen
// through normal microtask scheduling without us racing it.
let resolveHungPrompt: (() => void) | undefined;
mockCreateFnAgent.mockImplementationOnce(async () => ({
session: {
state: { messages: [] },
prompt: vi.fn(async () => {
await new Promise<void>((resolve) => { resolveHungPrompt = resolve; });
}),
dispose: vi.fn(),
},
}));
const sessionId = await createMissionInterviewSession(
"10.0.0.11",
"Stoppable mission interview",
"/tmp/project",
);
// initializeAgent → createMissionInterviewAgent → continueAgentConversation
// → generationGuard.run is several awaits deep; poll until the guard is
// registered, then stop. Bounded so a regression doesn't hang the suite.
let stopped = false;
for (let i = 0; i < 50 && !stopped; i++) {
stopped = stopMissionInterviewGeneration(sessionId);
if (!stopped) await new Promise((resolve) => setTimeout(resolve, 5));
}
expect(stopped).toBe(true);
// Yield so the guard's catch/finally and onUserStop run.
await new Promise((resolve) => setTimeout(resolve, 0));
const session = getMissionInterviewSession(sessionId);
expect(session?.error).toMatch(/stopped by user/i);
// Stop is idempotent — no in-flight generation after first call.
expect(stopMissionInterviewGeneration(sessionId)).toBe(false);
resolveHungPrompt?.();
await new Promise((resolve) => setTimeout(resolve, 0));
});
});
});

84
packages/dashboard/src/__tests__/subtask-breakdown.test.ts
View File

@@ -31,7 +31,9 @@ import {
SessionNotFoundError,
InvalidSessionStateError,
setAiSessionStore,
stopSubtaskGeneration,
SubtaskStreamManager,
GENERATION_TIMEOUT_MS,
} from "../subtask-breakdown.js";
const UUID_REGEX =
@@ -913,3 +915,85 @@ describe("SessionNotFoundError", () => {
expect(error.message).toBe("Missing session");
});
});
describe("subtask generation timeout / abort", () => {
it("marks the session as error and stops the prompt() promise when generation exceeds GENERATION_TIMEOUT_MS", async () => {
vi.useFakeTimers();
let resolveHungPrompt: (() => void) | undefined;
const hungPromptCallable = vi.fn(async () => {
// Simulate a stalled provider stream that never terminates on its own.
await new Promise<void>((resolve) => { resolveHungPrompt = resolve; });
return undefined;
});
mockCreateFnAgent.mockImplementation(async () => ({
session: {
state: { messages: [] },
prompt: hungPromptCallable,
dispose: vi.fn(),
},
}));
const created = await createSubtaskSession(
"Hung subtask generation",
undefined,
"/tmp/project",
);
// Yield once so startSubtaskGeneration's microtasks run before we advance time.
await Promise.resolve();
expect(getSubtaskSession(created.sessionId)?.status).toBe("generating");
await vi.advanceTimersByTimeAsync(GENERATION_TIMEOUT_MS);
// Drain any remaining microtasks (the abort propagates through Promise.race).
await vi.advanceTimersByTimeAsync(0);
const after = getSubtaskSession(created.sessionId);
expect(after?.status).toBe("error");
expect(after?.error).toMatch(/timed out/i);
// The hung prompt is still pending; release it so its microtask completes.
resolveHungPrompt?.();
await vi.advanceTimersByTimeAsync(0);
vi.useRealTimers();
});
it("stopSubtaskGeneration aborts an in-flight session and marks it stopped", async () => {
vi.useFakeTimers();
let resolveHungPrompt: (() => void) | undefined;
mockCreateFnAgent.mockImplementation(async () => ({
session: {
state: { messages: [] },
prompt: vi.fn(async () => {
await new Promise<void>((resolve) => { resolveHungPrompt = resolve; });
}),
dispose: vi.fn(),
},
}));
const created = await createSubtaskSession(
"Stoppable subtask generation",
undefined,
"/tmp/project",
);
await Promise.resolve();
expect(stopSubtaskGeneration(created.sessionId)).toBe(true);
await vi.advanceTimersByTimeAsync(0);
const after = getSubtaskSession(created.sessionId);
expect(after?.status).toBe("error");
expect(after?.error).toMatch(/stopped by user/i);
// Stop is idempotent — no in-flight generation after first call.
expect(stopSubtaskGeneration(created.sessionId)).toBe(false);
resolveHungPrompt?.();
await vi.advanceTimersByTimeAsync(0);
vi.useRealTimers();
});
});

138
packages/dashboard/src/ai-session-timeout.ts Normal file
View File

@@ -0,0 +1,138 @@
/**
* Per-session generation guard: bounds the runtime of an in-flight AI
* `prompt()` call with a timeout and an `AbortController`, so a silently-
* stalled model stream or hung tool call cannot leave a session pinned in
* `generating` forever.
*
* The guard is module-scoped: planning, subtask-breakdown, mission-interview,
* and milestone-slice-interview each instantiate their own. The session ID is
* the key, so concurrent generations across modules don't collide.
*/
export interface TimeoutHandlers {
/** Fired exactly once when the timeout elapses, before the abort propagates. */
onTimeout: () => void;
/** Fired when abort happens for a non-timeout reason (e.g. manual stop). */
onUserStop?: () => void;
}
interface ActiveEntry {
abort: AbortController;
timer: ReturnType<typeof setTimeout>;
}
type AbortCause = "timeout" | "user-stop" | "displaced";
export class GenerationGuard {
private readonly active = new Map<string, ActiveEntry>();
/**
* Tracks the cause of a pending abort so the original `run()` can
* distinguish a user-initiated stop from re-entrant displacement. The flag
* is consumed (deleted) by the catch block of the displaced `run()`.
*/
private readonly abortCause = new Map<AbortController, AbortCause>();
/**
* Wrap `op` with a timeout + abort. If a previous generation is still
* registered for the same id, it is aborted first (the prior `run()`
* rejects with `AbortError`, marked as `displaced` so its `onUserStop`
* handler does NOT fire).
*/
async run<T>(
sessionId: string,
timeoutMs: number,
handlers: TimeoutHandlers,
op: () => Promise<T>,
): Promise<T> {
this.cancelInternal(sessionId, "displaced");
const abort = new AbortController();
const timer = setTimeout(() => {
this.abortCause.set(abort, "timeout");
try {
handlers.onTimeout();
} catch {
// swallow — handler errors must not prevent abort
}
abort.abort();
}, timeoutMs);
const entry: ActiveEntry = { abort, timer };
this.active.set(sessionId, entry);
const abortPromise = new Promise<never>((_, reject) => {
abort.signal.addEventListener(
"abort",
() => reject(createAbortError()),
{ once: true },
);
});
try {
return await Promise.race([op(), abortPromise]);
} catch (err) {
if (isAbortError(err)) {
const cause = this.abortCause.get(abort) ?? "user-stop";
if (cause === "user-stop") {
try {
handlers.onUserStop?.();
} catch {
// swallow
}
}
}
throw err;
} finally {
clearTimeout(timer);
this.abortCause.delete(abort);
if (this.active.get(sessionId) === entry) {
this.active.delete(sessionId);
}
}
}
/** AbortSignal of the in-flight generation, if any — for tools that honor it. */
signal(sessionId: string): AbortSignal | undefined {
return this.active.get(sessionId)?.abort.signal;
}
has(sessionId: string): boolean {
return this.active.has(sessionId);
}
/**
* Manually abort the active generation. Returns true if there was one.
* The wrapped operation will reject with `AbortError`, and the `onUserStop`
* handler from the original `run()` call will fire.
*/
stop(sessionId: string): boolean {
return this.cancelInternal(sessionId, "user-stop");
}
/** Reset all in-flight generations (test/shutdown only). */
reset(): void {
for (const sessionId of [...this.active.keys()]) {
this.cancelInternal(sessionId, "user-stop");
}
}
private cancelInternal(sessionId: string, cause: AbortCause): boolean {
const entry = this.active.get(sessionId);
if (!entry) return false;
clearTimeout(entry.timer);
this.abortCause.set(entry.abort, cause);
entry.abort.abort();
this.active.delete(sessionId);
return true;
}
}
export function createAbortError(): Error {
const error = new Error("Generation aborted");
error.name = "AbortError";
return error;
}
export function isAbortError(err: unknown): boolean {
return err instanceof Error && err.name === "AbortError";
}

291
packages/dashboard/src/milestone-slice-interview.ts
View File

@@ -29,6 +29,7 @@ import {
resetDiagnosticsSink,
nonfatal,
} from "./ai-session-diagnostics.js";
import { GenerationGuard, isAbortError } from "./ai-session-timeout.js";
// Re-export JSON parsing utilities from mission-interview for external consumers
export {
@@ -122,6 +123,14 @@ const RATE_LIMIT_WINDOW_MS = 60 * 60 * 1000;
/** Max number of retry attempts when AI returns unparseable output */
const MAX_PARSE_RETRIES = 1;
/**
* Per-turn generation timeout. Bounds a stalled model stream or hung tool
* call so the session cannot stay pinned in `generating` indefinitely.
*/
export const GENERATION_TIMEOUT_MS = 120_000;
const generationGuard = new GenerationGuard();
/** Milestone interview system prompt */
export const MILESTONE_INTERVIEW_SYSTEM_PROMPT = `You are a milestone planning assistant for a project management system.
@@ -346,6 +355,9 @@ function cleanupInMemorySession(sessionId: string): boolean {
return false;
}
// Abort any in-flight generation so prompt() rejects promptly.
generationGuard.stop(sessionId);
if (session.agent) {
try { session.agent.session.dispose?.(); } catch { /* ignore */ }
session.agent = undefined;
@@ -356,6 +368,24 @@ function cleanupInMemorySession(sessionId: string): boolean {
return true;
}
function setTargetSessionError(session: TargetInterviewSession, message: string): void {
session.error = message;
session.updatedAt = new Date();
persistSession(session, "error", message);
milestoneSliceInterviewStreamManager.broadcast(session.id, {
type: "error",
data: message,
});
}
/**
* Manually abort an in-flight milestone/slice interview generation.
* Returns true if a generation was active and got aborted.
*/
export function stopMilestoneSliceInterviewGeneration(sessionId: string): boolean {
return generationGuard.stop(sessionId);
}
function getSessionType(targetType: TargetType): "milestone_interview" | "slice_interview" {
return targetType === "milestone" ? "milestone_interview" : "slice_interview";
}
@@ -744,12 +774,26 @@ async function ensureInterviewAgent(
return;
}
await session.agent.session.prompt(
[
"Previous conversation summary:",
historySummary,
"Use this context when handling the next user response.",
].join("\n\n"),
await generationGuard.run(
session.id,
GENERATION_TIMEOUT_MS,
{
onTimeout: () => setTargetSessionError(
session,
"AI generation timed out while restoring context. You can retry or start a new session.",
),
onUserStop: () => setTargetSessionError(
session,
"Generation stopped by user. You can retry or start a new session.",
),
},
() => session.agent!.session.prompt(
[
"Previous conversation summary:",
historySummary,
"Use this context when handling the next user response.",
].join("\n\n"),
),
);
}
@@ -791,128 +835,138 @@ async function continueAgentConversation(session: TargetInterviewSession, messag
}
try {
session.thinkingOutput = "";
await generationGuard.run(
session.id,
GENERATION_TIMEOUT_MS,
{
onTimeout: () => setTargetSessionError(
session,
"AI generation timed out. You can retry or start a new session.",
),
onUserStop: () => setTargetSessionError(
session,
"Generation stopped by user. You can retry or start a new session.",
),
},
async () => {
const agent = session.agent!;
session.thinkingOutput = "";
await session.agent.session.prompt(message);
await agent.session.prompt(message);
// Get the response text from the agent's state
interface AgentMessage {
role: string;
content?: string | Array<{ type: string; text: string }>;
}
const lastMessage = (session.agent.session.state.messages as AgentMessage[])
.filter((m: AgentMessage) => m.role === "assistant")
.pop();
// Get the response text from the agent's state
interface AgentMessage {
role: string;
content?: string | Array<{ type: string; text: string }>;
}
const lastMessage = (agent.session.state.messages as AgentMessage[])
.filter((m: AgentMessage) => m.role === "assistant")
.pop();
let responseText = session.thinkingOutput;
if (lastMessage?.content) {
if (typeof lastMessage.content === "string") {
responseText = lastMessage.content;
} else if (Array.isArray(lastMessage.content)) {
responseText = lastMessage.content
.filter((c: { type: string; text: string }): c is { type: "text"; text: string } => c.type === "text")
.map((c: { type: string; text: string }) => c.text)
.join("");
}
}
// Parse with retry using the target interview parser
let parsed: TargetInterviewResponse | undefined;
let lastError: Error | undefined;
for (let attempt = 0; attempt <= MAX_PARSE_RETRIES; attempt++) {
try {
parsed = parseTargetInterviewResponseImpl(responseText);
break;
} catch (err) {
lastError = err instanceof Error ? err : new Error(String(err));
if (attempt < MAX_PARSE_RETRIES) {
diagnostics.warn(
"Parse attempt failed, requesting reformat",
{ sessionId: session.id, attempt: attempt + 1, operation: "parse-retry" }
);
try {
session.thinkingOutput = "";
await session.agent.session.prompt(
"Your previous response could not be parsed as JSON. " +
'Please respond with ONLY a valid JSON object: either {"type":"question","data":{...}} ' +
'or {"type":"complete","data":{"title":"...","description":"...","planningNotes":"...","verification":"..."}}' +
". No markdown, no explanation, just the JSON."
);
const retryMessage = (session.agent.session.state.messages as AgentMessage[])
.filter((m: AgentMessage) => m.role === "assistant")
.pop();
let retryText = session.thinkingOutput;
if (retryMessage?.content) {
if (typeof retryMessage.content === "string") {
retryText = retryMessage.content;
} else if (Array.isArray(retryMessage.content)) {
retryText = retryMessage.content
.filter((c: { type: string; text: string }): c is { type: "text"; text: string } => c.type === "text")
.map((c: { type: string; text: string }) => c.text)
.join("");
}
}
responseText = retryText;
} catch (retryErr) {
diagnostics.errorFromException("Retry prompt failed for session", retryErr, { sessionId: session.id, operation: "retry-prompt" });
break;
let responseText = session.thinkingOutput;
if (lastMessage?.content) {
if (typeof lastMessage.content === "string") {
responseText = lastMessage.content;
} else if (Array.isArray(lastMessage.content)) {
responseText = lastMessage.content
.filter((c: { type: string; text: string }): c is { type: "text"; text: string } => c.type === "text")
.map((c: { type: string; text: string }) => c.text)
.join("");
}
}
}
}
if (!parsed) {
const errorMsg = `${lastError?.message || "Failed to parse AI response"} You can try responding again or start a new session.`;
diagnostics.error(
"All parse attempts exhausted for session",
{ sessionId: session.id, message: errorMsg, operation: "parse-exhausted" }
);
session.error = errorMsg;
session.updatedAt = new Date();
persistSession(session, "error", errorMsg);
milestoneSliceInterviewStreamManager.broadcast(session.id, {
type: "error",
data: errorMsg,
});
// Parse with retry using the target interview parser
let parsed: TargetInterviewResponse | undefined;
let lastError: Error | undefined;
for (let attempt = 0; attempt <= MAX_PARSE_RETRIES; attempt++) {
try {
parsed = parseTargetInterviewResponseImpl(responseText);
break;
} catch (err) {
lastError = err instanceof Error ? err : new Error(String(err));
if (attempt < MAX_PARSE_RETRIES) {
diagnostics.warn(
"Parse attempt failed, requesting reformat",
{ sessionId: session.id, attempt: attempt + 1, operation: "parse-retry" }
);
try {
session.thinkingOutput = "";
await agent.session.prompt(
"Your previous response could not be parsed as JSON. " +
'Please respond with ONLY a valid JSON object: either {"type":"question","data":{...}} ' +
'or {"type":"complete","data":{"title":"...","description":"...","planningNotes":"...","verification":"..."}}' +
". No markdown, no explanation, just the JSON."
);
const retryMessage = (agent.session.state.messages as AgentMessage[])
.filter((m: AgentMessage) => m.role === "assistant")
.pop();
let retryText = session.thinkingOutput;
if (retryMessage?.content) {
if (typeof retryMessage.content === "string") {
retryText = retryMessage.content;
} else if (Array.isArray(retryMessage.content)) {
retryText = retryMessage.content
.filter((c: { type: string; text: string }): c is { type: "text"; text: string } => c.type === "text")
.map((c: { type: string; text: string }) => c.text)
.join("");
}
}
responseText = retryText;
} catch (retryErr) {
diagnostics.errorFromException("Retry prompt failed for session", retryErr, { sessionId: session.id, operation: "retry-prompt" });
break;
}
}
}
}
if (!parsed) {
const errorMsg = `${lastError?.message || "Failed to parse AI response"} You can try responding again or start a new session.`;
diagnostics.error(
"All parse attempts exhausted for session",
{ sessionId: session.id, message: errorMsg, operation: "parse-exhausted" }
);
setTargetSessionError(session, errorMsg);
return;
}
if (parsed.type === "question") {
session.currentQuestion = parsed.data;
session.error = undefined;
session.lastGeneratedThinking = session.thinkingOutput;
session.updatedAt = new Date();
persistSession(session, "awaiting_input");
milestoneSliceInterviewStreamManager.broadcast(session.id, {
type: "question",
data: parsed.data,
});
} else if (parsed.type === "complete") {
session.summary = parsed.data;
session.currentQuestion = undefined;
session.error = undefined;
session.updatedAt = new Date();
persistSession(session, "complete");
milestoneSliceInterviewStreamManager.broadcast(session.id, {
type: "summary",
data: parsed.data,
});
milestoneSliceInterviewStreamManager.broadcast(session.id, { type: "complete" });
}
},
);
} catch (err) {
// Timeout / user-stop already published an error state via the guard
// handlers. Don't double-broadcast a generic AbortError.
if (isAbortError(err)) {
return;
}
if (parsed.type === "question") {
session.currentQuestion = parsed.data;
session.error = undefined;
session.lastGeneratedThinking = session.thinkingOutput;
session.updatedAt = new Date();
persistSession(session, "awaiting_input");
milestoneSliceInterviewStreamManager.broadcast(session.id, {
type: "question",
data: parsed.data,
});
} else if (parsed.type === "complete") {
session.summary = parsed.data;
session.currentQuestion = undefined;
session.error = undefined;
session.updatedAt = new Date();
persistSession(session, "complete");
milestoneSliceInterviewStreamManager.broadcast(session.id, {
type: "summary",
data: parsed.data,
});
milestoneSliceInterviewStreamManager.broadcast(session.id, { type: "complete" });
}
} catch (err) {
const errorMessage = err instanceof Error ? err.message : "AI processing failed";
diagnostics.errorFromException("Agent conversation error for session", err, { sessionId: session.id, operation: "conversation" });
session.error = errorMessage;
session.updatedAt = new Date();
persistSession(session, "error", errorMessage);
milestoneSliceInterviewStreamManager.broadcast(session.id, {
type: "error",
data: errorMessage,
});
setTargetSessionError(session, errorMessage);
}
}
@@ -1265,6 +1319,7 @@ export function __resetMilestoneSliceInterviewState(): void {
sessions.clear();
rateLimits.clear();
milestoneSliceInterviewStreamManager.reset();
generationGuard.reset();
if (_aiSessionStore && _aiSessionDeletedListener) {
_aiSessionStore.off("ai_session:deleted", _aiSessionDeletedListener);

296
packages/dashboard/src/mission-interview.ts
View File

@@ -26,6 +26,7 @@ import {
resetDiagnosticsSink,
nonfatal,
} from "./ai-session-diagnostics.js";
import { GenerationGuard, isAbortError } from "./ai-session-timeout.js";
import { createFnAgent as engineCreateFnAgent } from "@fusion/engine";
@@ -86,6 +87,15 @@ const RATE_LIMIT_WINDOW_MS = 60 * 60 * 1000;
/** Max number of retry attempts when AI returns unparseable output */
const MAX_PARSE_RETRIES = 1;
/**
* Per-turn generation timeout. Mission interview turns produce larger plans
* than planning, so this is more generous than planning's 120 s. Bounds the
* worst case for a silently-stalled model stream or hung tool call.
*/
export const GENERATION_TIMEOUT_MS = 180_000;
const generationGuard = new GenerationGuard();
/** Mission interview system prompt */
export const MISSION_INTERVIEW_SYSTEM_PROMPT = `You are a mission planning assistant for a project management system.
@@ -283,6 +293,9 @@ function cleanupInMemoryMissionSession(sessionId: string): boolean {
return false;
}
// Abort any in-flight generation so prompt() rejects promptly.
generationGuard.stop(sessionId);
if (session.agent) {
try { session.agent.session.dispose?.(); } catch { /* ignore */ }
session.agent = undefined;
@@ -859,18 +872,54 @@ async function ensureMissionInterviewAgent(
return;
}
await session.agent.session.prompt(
[
"Previous conversation summary:",
historySummary,
"Use this context when handling the next user response.",
].join("\n\n"),
await generationGuard.run(
session.id,
GENERATION_TIMEOUT_MS,
{
onTimeout: () => setMissionSessionError(
session,
"AI generation timed out while restoring context. You can retry or start a new session.",
),
onUserStop: () => setMissionSessionError(
session,
"Generation stopped by user. You can retry or start a new session.",
),
},
() => session.agent!.session.prompt(
[
"Previous conversation summary:",
historySummary,
"Use this context when handling the next user response.",
].join("\n\n"),
),
);
}
function setMissionSessionError(session: MissionInterviewSession, message: string): void {
session.error = message;
session.updatedAt = new Date();
persistMissionSession(session, "error", message);
missionInterviewStreamManager.broadcast(session.id, {
type: "error",
data: message,
});
}
/**
* Manually abort an in-flight mission interview generation. Returns true if
* a generation was active and got aborted.
*/
export function stopMissionInterviewGeneration(sessionId: string): boolean {
return generationGuard.stop(sessionId);
}
/**
* Continue the AI conversation with a user message.
* Includes bounded recovery: one retry on parse failure.
*
* The entire body — including the parse-retry inner `prompt()` calls — runs
* inside `generationGuard.run`, so a stalled stream or hung tool call on
* either the first or retry attempt is bounded by `GENERATION_TIMEOUT_MS`.
*/
async function continueAgentConversation(session: MissionInterviewSession, message: string): Promise<void> {
if (!session.agent) {
@@ -878,128 +927,138 @@ async function continueAgentConversation(session: MissionInterviewSession, messa
}
try {
session.thinkingOutput = "";
await generationGuard.run(
session.id,
GENERATION_TIMEOUT_MS,
{
onTimeout: () => setMissionSessionError(
session,
"AI generation timed out. You can retry or start a new session.",
),
onUserStop: () => setMissionSessionError(
session,
"Generation stopped by user. You can retry or start a new session.",
),
},
async () => {
const agent = session.agent!;
session.thinkingOutput = "";
await session.agent.session.prompt(message);
await agent.session.prompt(message);
// Get the response text from the agent's state
interface AgentMessage {
role: string;
content?: string | Array<{ type: string; text: string }>;
}
const lastMessage = (session.agent.session.state.messages as AgentMessage[])
.filter((m: AgentMessage) => m.role === "assistant")
.pop();
// Get the response text from the agent's state
interface AgentMessage {
role: string;
content?: string | Array<{ type: string; text: string }>;
}
const lastMessage = (agent.session.state.messages as AgentMessage[])
.filter((m: AgentMessage) => m.role === "assistant")
.pop();
let responseText = session.thinkingOutput;
if (lastMessage?.content) {
if (typeof lastMessage.content === "string") {
responseText = lastMessage.content;
} else if (Array.isArray(lastMessage.content)) {
responseText = lastMessage.content
.filter((c: { type: string; text: string }): c is { type: "text"; text: string } => c.type === "text")
.map((c: { type: string; text: string }) => c.text)
.join("");
}
}
// Parse with retry
let parsed: MissionInterviewResponse | undefined;
let lastError: Error | undefined;
for (let attempt = 0; attempt <= MAX_PARSE_RETRIES; attempt++) {
try {
parsed = parseMissionAgentResponse(responseText);
break;
} catch (err) {
lastError = err instanceof Error ? err : new Error(String(err));
if (attempt < MAX_PARSE_RETRIES) {
diagnostics.warn(
"Parse attempt failed, requesting reformat",
{ sessionId: session.id, attempt: attempt + 1, operation: "parse-retry" }
);
try {
session.thinkingOutput = "";
await session.agent.session.prompt(
"Your previous response could not be parsed as JSON. " +
'Please respond with ONLY a valid JSON object: either {"type":"question","data":{...}} ' +
'or {"type":"complete","data":{"missionTitle":"...","missionDescription":"...","milestones":[...]}}. ' +
"No markdown, no explanation, just the JSON."
);
const retryMessage = (session.agent.session.state.messages as AgentMessage[])
.filter((m: AgentMessage) => m.role === "assistant")
.pop();
let retryText = session.thinkingOutput;
if (retryMessage?.content) {
if (typeof retryMessage.content === "string") {
retryText = retryMessage.content;
} else if (Array.isArray(retryMessage.content)) {
retryText = retryMessage.content
.filter((c: { type: string; text: string }): c is { type: "text"; text: string } => c.type === "text")
.map((c: { type: string; text: string }) => c.text)
.join("");
}
}
responseText = retryText;
} catch (retryErr) {
diagnostics.errorFromException("Retry prompt failed for session", retryErr, { sessionId: session.id, operation: "retry-prompt" });
break;
let responseText = session.thinkingOutput;
if (lastMessage?.content) {
if (typeof lastMessage.content === "string") {
responseText = lastMessage.content;
} else if (Array.isArray(lastMessage.content)) {
responseText = lastMessage.content
.filter((c: { type: string; text: string }): c is { type: "text"; text: string } => c.type === "text")
.map((c: { type: string; text: string }) => c.text)
.join("");
}
}
}
}
if (!parsed) {
const errorMsg = `${lastError?.message || "Failed to parse AI response"} You can try responding again or start a new session.`;
diagnostics.error(
"All parse attempts exhausted for session",
{ sessionId: session.id, message: errorMsg, operation: "parse-exhausted" }
);
session.error = errorMsg;
session.updatedAt = new Date();
persistMissionSession(session, "error", errorMsg);
missionInterviewStreamManager.broadcast(session.id, {
type: "error",
data: errorMsg,
});
// Parse with retry
let parsed: MissionInterviewResponse | undefined;
let lastError: Error | undefined;
for (let attempt = 0; attempt <= MAX_PARSE_RETRIES; attempt++) {
try {
parsed = parseMissionAgentResponse(responseText);
break;
} catch (err) {
lastError = err instanceof Error ? err : new Error(String(err));
if (attempt < MAX_PARSE_RETRIES) {
diagnostics.warn(
"Parse attempt failed, requesting reformat",
{ sessionId: session.id, attempt: attempt + 1, operation: "parse-retry" }
);
try {
session.thinkingOutput = "";
await agent.session.prompt(
"Your previous response could not be parsed as JSON. " +
'Please respond with ONLY a valid JSON object: either {"type":"question","data":{...}} ' +
'or {"type":"complete","data":{"missionTitle":"...","missionDescription":"...","milestones":[...]}}. ' +
"No markdown, no explanation, just the JSON."
);
const retryMessage = (agent.session.state.messages as AgentMessage[])
.filter((m: AgentMessage) => m.role === "assistant")
.pop();
let retryText = session.thinkingOutput;
if (retryMessage?.content) {
if (typeof retryMessage.content === "string") {
retryText = retryMessage.content;
} else if (Array.isArray(retryMessage.content)) {
retryText = retryMessage.content
.filter((c: { type: string; text: string }): c is { type: "text"; text: string } => c.type === "text")
.map((c: { type: string; text: string }) => c.text)
.join("");
}
}
responseText = retryText;
} catch (retryErr) {
diagnostics.errorFromException("Retry prompt failed for session", retryErr, { sessionId: session.id, operation: "retry-prompt" });
break;
}
}
}
}
if (!parsed) {
const errorMsg = `${lastError?.message || "Failed to parse AI response"} You can try responding again or start a new session.`;
diagnostics.error(
"All parse attempts exhausted for session",
{ sessionId: session.id, message: errorMsg, operation: "parse-exhausted" }
);
setMissionSessionError(session, errorMsg);
return;
}
if (parsed.type === "question") {
session.currentQuestion = parsed.data;
session.error = undefined;
session.lastGeneratedThinking = session.thinkingOutput;
session.updatedAt = new Date();
persistMissionSession(session, "awaiting_input");
missionInterviewStreamManager.broadcast(session.id, {
type: "question",
data: parsed.data,
});
} else if (parsed.type === "complete") {
session.summary = parsed.data;
session.currentQuestion = undefined;
session.error = undefined;
session.updatedAt = new Date();
persistMissionSession(session, "complete");
missionInterviewStreamManager.broadcast(session.id, {
type: "summary",
data: parsed.data,
});
missionInterviewStreamManager.broadcast(session.id, { type: "complete" });
}
},
);
} catch (err) {
// Timeout / user-stop already published an error state via the guard
// handlers. Don't double-broadcast a generic AbortError.
if (isAbortError(err)) {
return;
}
if (parsed.type === "question") {
session.currentQuestion = parsed.data;
session.error = undefined;
session.lastGeneratedThinking = session.thinkingOutput;
session.updatedAt = new Date();
persistMissionSession(session, "awaiting_input");
missionInterviewStreamManager.broadcast(session.id, {
type: "question",
data: parsed.data,
});
} else if (parsed.type === "complete") {
session.summary = parsed.data;
session.currentQuestion = undefined;
session.error = undefined;
session.updatedAt = new Date();
persistMissionSession(session, "complete");
missionInterviewStreamManager.broadcast(session.id, {
type: "summary",
data: parsed.data,
});
missionInterviewStreamManager.broadcast(session.id, { type: "complete" });
}
} catch (err) {
const errorMessage = err instanceof Error ? err.message : "AI processing failed";
diagnostics.errorFromException("Agent conversation error for session", err, { sessionId: session.id, operation: "conversation" });
session.error = errorMessage;
session.updatedAt = new Date();
persistMissionSession(session, "error", errorMessage);
missionInterviewStreamManager.broadcast(session.id, {
type: "error",
data: errorMessage,
});
setMissionSessionError(session, errorMessage);
}
}
@@ -1211,6 +1270,7 @@ export function __resetMissionInterviewState(): void {
sessions.clear();
rateLimits.clear();
missionInterviewStreamManager.reset();
generationGuard.reset();
if (_aiSessionStore && _aiSessionDeletedListener) {
_aiSessionStore.off("ai_session:deleted", _aiSessionDeletedListener);

82
packages/dashboard/src/subtask-breakdown.ts
View File

@@ -8,6 +8,7 @@ import {
createSessionDiagnostics,
resetDiagnosticsSink,
} from "./ai-session-diagnostics.js";
import { GenerationGuard, isAbortError } from "./ai-session-timeout.js";
import { createFnAgent as engineCreateFnAgent } from "@fusion/engine";
@@ -77,6 +78,14 @@ export type SubtaskStreamCallback = (event: SubtaskStreamEvent, eventId?: number
const SESSION_TTL_MS = 7 * 24 * 60 * 60 * 1000;
const CLEANUP_INTERVAL_MS = 5 * 60 * 1000;
/**
* Subtask breakdown is a single-turn flow with no follow-up questions, so
* a stuck `prompt()` (silent stream stall, hung tool call) blocks the whole
* UI. 90 s is generous for a small JSON response and bounds the worst case.
*/
export const GENERATION_TIMEOUT_MS = 90_000;
const generationGuard = new GenerationGuard();
/** Minimal interface for the agent object created by createFnAgent */
interface SubtaskAgent {
@@ -208,6 +217,10 @@ function cleanupInMemorySubtaskSession(sessionId: string): boolean {
return false;
}
// Abort any in-flight generation so the agent.session.prompt() rejects
// promptly and we don't leak the timer / AbortController.
generationGuard.stop(sessionId);
try {
session.agent?.session?.dispose?.();
} catch {
@@ -407,6 +420,11 @@ async function startSubtaskGeneration(
try {
await generateSubtasks(sessionId, cwd, promptOverrides);
} catch (err) {
// Timeout / user-stop already published an error state via the guard
// handlers. Don't overwrite it with a generic AbortError message.
if (isAbortError(err)) {
return;
}
const existing = sessions.get(sessionId);
if (!existing) return;
existing.status = "error";
@@ -451,22 +469,39 @@ async function generateSubtasks(
});
session.agent = agent;
await agent.session.prompt(session.initialDescription);
const messages = agent.session.state.messages as Array<{ role: string; content?: string | Array<{ type: string; text: string }> }>;
const lastAssistant = messages.filter((m) => m.role === "assistant").pop();
let responseText = session.thinkingOutput;
if (typeof lastAssistant?.content === "string") {
responseText = lastAssistant.content;
} else if (Array.isArray(lastAssistant?.content)) {
responseText = lastAssistant.content
.filter((item): item is { type: "text"; text: string } => item.type === "text")
.map((item) => item.text)
.join("");
}
await generationGuard.run(
sessionId,
GENERATION_TIMEOUT_MS,
{
onTimeout: () => setSubtaskError(
sessionId,
"AI generation timed out. You can retry or start a new session.",
),
onUserStop: () => setSubtaskError(
sessionId,
"Generation stopped by user. You can retry or start a new session.",
),
},
async () => {
await agent.session.prompt(session.initialDescription);
const subtasks = parseSubtasks(responseText);
completeSession(sessionId, subtasks);
const messages = agent.session.state.messages as Array<{ role: string; content?: string | Array<{ type: string; text: string }> }>;
const lastAssistant = messages.filter((m) => m.role === "assistant").pop();
let responseText = session.thinkingOutput;
if (typeof lastAssistant?.content === "string") {
responseText = lastAssistant.content;
} else if (Array.isArray(lastAssistant?.content)) {
responseText = lastAssistant.content
.filter((item): item is { type: "text"; text: string } => item.type === "text")
.map((item) => item.text)
.join("");
}
const subtasks = parseSubtasks(responseText);
completeSession(sessionId, subtasks);
},
);
return;
}
@@ -520,6 +555,24 @@ function generateFallbackSubtasks(initialDescription: string): SubtaskItem[] {
];
}
function setSubtaskError(sessionId: string, message: string): void {
const session = sessions.get(sessionId);
if (!session) return;
session.status = "error";
session.error = message;
session.updatedAt = new Date();
persistSubtaskSession(session, "error", message);
subtaskStreamManager.broadcast(sessionId, { type: "error", data: message });
}
/**
* Manually abort an in-flight subtask generation (UI "stop" button).
* Returns true if a generation was active and got aborted.
*/
export function stopSubtaskGeneration(sessionId: string): boolean {
return generationGuard.stop(sessionId);
}
function completeSession(sessionId: string, subtasks: SubtaskItem[]): void {
const session = sessions.get(sessionId);
if (!session) return;
@@ -622,6 +675,7 @@ export function __resetSubtaskBreakdownState(): void {
}
sessions.clear();
subtaskStreamManager.reset();
generationGuard.reset();
if (_aiSessionStore && _aiSessionDeletedListener) {
_aiSessionStore.off("ai_session:deleted", _aiSessionDeletedListener);