/** * Planning Mode Session Management * * Manages AI-guided planning sessions for interactive task creation. * Sessions are stored in-memory with TTL cleanup. * * Features: * - AI agent integration via createFnAgent for real-time planning conversations * - Streaming via SSE (createSessionWithAgent) and non-streaming (createSession) * - Rate limiting per IP * - Session expiration and cleanup * - JSON response parsing with robust extraction and repair */ import type { PlanningQuestion, PlanningSummary, PlanningResponse, TaskStore, NtfyNotificationEvent, } from "@fusion/core"; import { resolvePrompt, summarizeTitle, type PromptOverrideMap } from "@fusion/core"; import type { SubtaskItem } from "./subtask-breakdown.js"; import { randomUUID } from "node:crypto"; import { EventEmitter } from "node:events"; import type { AiSessionStore, AiSessionRow } from "./ai-session-store.js"; import { SessionEventBuffer, type SessionBufferedEvent } from "./sse-buffer.js"; import { createSessionDiagnostics, resetDiagnosticsSink, nonfatal, } from "./ai-session-diagnostics.js"; import { createFnAgent as engineCreateFnAgent } from "@fusion/engine"; import * as engineModule from "@fusion/engine"; // eslint-disable-next-line @typescript-eslint/no-explicit-any type AgentResult = any; // eslint-disable-next-line @typescript-eslint/no-explicit-any let createFnAgent: any = engineCreateFnAgent; // ── Notification Integration ──────────────────────────────────────────── // // The planning module sends "planning-awaiting-input" notifications when an // AI planning session needs user input. Currently this uses ntfy-specific // helpers loaded dynamically from @fusion/engine. // // The engine now exposes a pluggable NotificationService abstraction // (NotificationService + NotificationProvider interface) that dispatches // events to registered providers (ntfy, webhook, etc.). The planning module // will migrate to using NotificationService.dispatch() for a provider- // agnostic flow once the broader notification integration is complete. // // For now, the ntfy-specific path remains active because "planning-awaiting-input" // is only supported by the ntfy provider and is not dispatched through the // NotificationService event listeners (which handle task:moved, task:updated, // task:merged, settings:updated). /** * Configuration for planning session notifications. * * Currently drives ntfy-specific notifications for the "planning-awaiting-input" event. * This will be generalized to support the pluggable NotificationService abstraction * (from @fusion/engine) once additional providers support planning events. * Kept as "Ntfy" in the name for backward compatibility with existing call sites. */ interface PlanningNtfyConfig { enabled: boolean; topic?: string; dashboardHost?: string; events?: NtfyNotificationEvent[]; ntfyBaseUrl?: string; } /** * Ntfy-specific helper functions loaded from @fusion/engine at runtime. * * These wrap the engine's exported notification helpers. In the future, this * will be replaced by direct use of NotificationService.dispatch() for a * provider-agnostic notification flow. */ interface PlanningNtfyHelpers { isNtfyEventEnabled: (events: NtfyNotificationEvent[] | undefined, event: NtfyNotificationEvent) => boolean; buildNtfyClickUrl: (options: { dashboardHost?: string; projectId?: string; taskId?: string }) => string | undefined; sendNtfyNotification: (input: { ntfyBaseUrl?: string; topic: string; title: string; message: string; priority?: "low" | "default" | "high" | "urgent"; clickUrl?: string; }) => Promise; } /** Cached notification helpers. Loaded lazily by ensureNtfyHelpersReady(). */ let planningNtfyHelpers: PlanningNtfyHelpers | undefined; /** * Shared diagnostics helper for the planning module. * Uses the shared ai-session-diagnostics helper for consistent scoped logging. * @see ai-session-diagnostics.ts for the shared contract */ const diagnostics = createSessionDiagnostics("planning"); /** * Get the current diagnostics logger (for backward compatibility). * @internal - exposed for test hook */ export function __getPlanningDiagnostics() { return diagnostics; } /** * Inject a diagnostics sink (test-only). * Delegates to the shared ai-session-diagnostics sink. * When a sink is injected, all planning module diagnostics route through it. * This allows tests to assert on diagnostics without global console spies. */ export function __setPlanningDiagnostics(_logger: unknown): void { // For backward compatibility, we keep this function but it now delegates // to the shared helper's sink mechanism. The actual sink injection // should use setDiagnosticsSink() from ai-session-diagnostics. // This function is kept for backward compatibility with existing tests. if (_logger === null) { resetDiagnosticsSink(); } } function ensureEngineReady(): Promise { return Promise.resolve(); } async function ensureNtfyHelpersReady(): Promise { if (planningNtfyHelpers) { return; } const hasNotificationService = "NotificationService" in engineModule && typeof engineModule.NotificationService === "function"; const hasAllHelpers = "isNtfyEventEnabled" in engineModule && "buildNtfyClickUrl" in engineModule && "sendNtfyNotification" in engineModule && typeof engineModule.isNtfyEventEnabled === "function" && typeof engineModule.buildNtfyClickUrl === "function" && typeof engineModule.sendNtfyNotification === "function"; if (!hasAllHelpers) { return; } planningNtfyHelpers = { isNtfyEventEnabled: engineModule.isNtfyEventEnabled, buildNtfyClickUrl: engineModule.buildNtfyClickUrl, sendNtfyNotification: engineModule.sendNtfyNotification, }; if (hasNotificationService) { diagnostics.info( "NotificationService abstraction detected in engine", { operation: "notification-service-detection" }, ); } } // ── Constants ─────────────────────────────────────────────────────────────── /** Planning system prompt for the AI agent */ export const PLANNING_SYSTEM_PROMPT = `You are a planning assistant for the fn task board system. Your job: help users transform vague, high-level ideas into well-defined, actionable tasks. ## Conversation Flow 1. User provides a high-level plan (e.g., "Build a user auth system") 2. You ask clarifying questions to understand scope, requirements, and constraints 3. You present UI-friendly selection options when appropriate 4. Once you have enough information, generate a structured summary ## Question Types to Use - "text": Open-ended follow-up questions for detailed input - "single_select": When user must choose one option (e.g., tech stack preference) - "multi_select": When multiple options can apply (e.g., features to include) - "confirm": Yes/No questions for quick decisions ## Guidelines - Ask 3-7 questions depending on complexity - Start broad, then narrow down specifics - Suggest sensible defaults based on project context - Keep questions focused and actionable - When asking about file scope, reference actual project structure ## Summary Generation When ready to complete, generate: - A concise but descriptive title (max 80 chars) - A detailed description with context gathered - Size estimate (S/M/L) based on scope - Any suggested dependencies on existing tasks - Key deliverables as a checklist ## Response Format Always respond with valid JSON in one of these formats: For questions: {\n "type": "question",\n "data": {\n "id": "unique-id",\n "type": "text|single_select|multi_select|confirm",\n "question": "The question text",\n "description": "Helpful context",\n "options": [{"id": "opt1", "label": "Option 1", "description": "Details"}]\n }\n} For completion: {\n "type": "complete",\n "data": {\n "title": "Task title",\n "description": "Detailed description",\n "suggestedSize": "S|M|L",\n "suggestedDependencies": [],\n "keyDeliverables": ["Item 1", "Item 2"]\n }\n}`; /** Placeholder title for draft sessions before the user starts planning. */ export const DRAFT_PLACEHOLDER_TITLE = "New planning session"; /** * Shape of the JSON blob persisted in `ai_sessions.inputPayload` for draft * planning sessions. Carries the in-progress plan text plus an optional * model override so reopening a draft restores the model selection the user * picked at create time, and so summarizeDraftTitle calls hit that same * model rather than silently falling back to project defaults. * * `summarizedFor` records the exact `initialPlan` string the current * persisted title was summarized from. The start-existing path uses it to * skip re-summarizing when blur/close already produced a title for the * final text, avoiding a redundant model call. */ export interface DraftInputPayload { initialPlan?: string; modelProvider?: string; modelId?: string; summarizedFor?: string; } /** Session TTL in milliseconds (7 days) */ export const SESSION_TTL_MS = 7 * 24 * 60 * 60 * 1000; /** Cleanup interval in milliseconds (5 minutes) */ const CLEANUP_INTERVAL_MS = 5 * 60 * 1000; /** Max planning sessions per IP per hour */ const MAX_SESSIONS_PER_IP_PER_HOUR = 1000; /** Rate limiting window in milliseconds (1 hour) */ const RATE_LIMIT_WINDOW_MS = 60 * 60 * 1000; /** Generation timeout in milliseconds (120 seconds). */ export const GENERATION_TIMEOUT_MS = 120_000; export type PlanningDepth = "small" | "medium" | "large"; const PLANNING_DEPTH_PROMPT_SUFFIX: Record = { small: "Ask exactly 1-2 focused questions. Prioritize speed and getting to a summary quickly. Skip optional clarification.", medium: "Ask 3-5 well-rounded questions. Balance breadth and depth. This is the default behavior.", large: "Ask 5-8 thorough questions. Deeply explore scope, edge cases, dependencies, and implementation details. Be comprehensive.", }; export function buildDepthPromptSuffix( depth?: PlanningDepth, customQuestionCount?: number, ): string { if (Number.isInteger(customQuestionCount) && (customQuestionCount ?? 0) > 0) { return `Ask exactly ${customQuestionCount} questions. Adjust depth and breadth to fit within that count.`; } if (!depth) { return ""; } return PLANNING_DEPTH_PROMPT_SUFFIX[depth]; } // ── Types ─────────────────────────────────────────────────────────────────── /** SSE event types for planning session streaming */ export type PlanningStreamEvent = | { type: "thinking"; data: string } | { type: "question"; data: PlanningQuestion } | { type: "summary"; data: PlanningSummary } | { type: "error"; data: string } | { type: "complete" }; /** Callback function for streaming events */ export type PlanningStreamCallback = (event: PlanningStreamEvent, eventId?: number) => void; interface PlanningHistoryEntry { question: PlanningQuestion; response: unknown; thinkingOutput?: string; } interface Session { id: string; ip: string; initialPlan: string; title: string; projectId?: string; /** Model override the user picked at draft-create time. Persisted in inputPayload so reopen restores it. */ draftModelProvider?: string; draftModelId?: string; /** Plan text the current title was summarized from; lets startExistingSession skip a redundant re-summarize when blur/close already covered the final text. */ draftSummarizedFor?: string; ntfyConfig?: PlanningNtfyConfig; /** Last planning question notified via ntfy, keyed as `${sessionId}:${questionId}` for dedupe across reconnect/replay. */ lastNotifiedQuestionKey?: string; history: PlanningHistoryEntry[]; currentQuestion?: PlanningQuestion; summary?: PlanningSummary; /** Last terminal error for retry UX */ error?: string; /** AI agent session for real-time interaction */ agent?: AgentResult; /** Callback for streaming events to SSE clients */ streamCallback?: PlanningStreamCallback; /** Accumulated thinking output for display */ thinkingOutput: string; /** Thinking output generated while producing currentQuestion */ lastGeneratedThinking: string; createdAt: Date; updatedAt: Date; } interface RateLimitEntry { count: number; firstRequestAt: Date; } // ── In-Memory Storage ─────────────────────────────────────────────────────── /** Active planning sessions indexed by session ID */ const sessions = new Map(); /** Rate limiting state indexed by IP */ const rateLimits = new Map(); /** Active planning generations keyed by session ID. */ const activeGenerations = new Map(); // ── AI Session Persistence ──────────────────────────────────────────────── /** Optional store for persisting session state across reloads/browsers. */ let _aiSessionStore: AiSessionStore | undefined; let _aiSessionDeletedListener: ((sessionId: string) => void) | undefined; function safeParseJson( text: string | null, fallback: T, options?: { throwOnError?: boolean; fieldName?: string }, ): T { if (!text) { return fallback; } try { return JSON.parse(text) as T; } catch (error) { if (options?.throwOnError) { const fieldSuffix = options.fieldName ? ` in ${options.fieldName}` : ""; throw new Error(`Invalid JSON${fieldSuffix}: ${(error as Error).message}`); } return fallback; } } /** Wire up the AI session persistence store. Called once from server.ts. */ export function setAiSessionStore(store: AiSessionStore): void { if (_aiSessionStore && _aiSessionDeletedListener) { _aiSessionStore.off("ai_session:deleted", _aiSessionDeletedListener); } _aiSessionStore = store; _aiSessionDeletedListener = (sessionId: string) => { cleanupInMemorySession(sessionId); }; _aiSessionStore.on("ai_session:deleted", _aiSessionDeletedListener); } function cleanupInMemorySession(sessionId: string): boolean { const session = sessions.get(sessionId); if (!session) { return false; } const activeGeneration = activeGenerations.get(sessionId); if (activeGeneration) { clearTimeout(activeGeneration.timer); activeGenerations.delete(sessionId); } if (session.agent) { try { session.agent.session.dispose?.(); } catch (err) { diagnostics.errorFromException("Error disposing agent for session", err, { sessionId, operation: "dispose-session" }); } session.agent = undefined; } planningStreamManager.cleanupSession(sessionId); sessions.delete(sessionId); return true; } /** Persist the current session state to SQLite (no-op if store not wired). */ function persistSession(session: Session, status: "generating" | "awaiting_input" | "complete" | "error" | "draft", error?: string): void { if (!_aiSessionStore) return; const row: AiSessionRow = { id: session.id, type: "planning", status, title: session.title || session.initialPlan.slice(0, 120), inputPayload: JSON.stringify({ ip: session.ip, initialPlan: session.initialPlan, ...(session.draftModelProvider ? { modelProvider: session.draftModelProvider } : {}), ...(session.draftModelId ? { modelId: session.draftModelId } : {}), ...(session.draftSummarizedFor ? { summarizedFor: session.draftSummarizedFor } : {}), }), conversationHistory: JSON.stringify(session.history), currentQuestion: session.currentQuestion ? JSON.stringify(session.currentQuestion) : null, result: session.summary ? JSON.stringify(session.summary) : null, thinkingOutput: session.thinkingOutput, error: error ?? null, projectId: session.projectId ?? null, createdAt: session.createdAt.toISOString(), updatedAt: new Date().toISOString(), lockedByTab: null, lockedAt: null, }; _aiSessionStore.upsert(row); } /** Persist only thinking output (debounced). */ function persistThinking(sessionId: string, thinkingOutput: string): void { if (!_aiSessionStore) return; _aiSessionStore.updateThinking(sessionId, thinkingOutput); } /** Remove session from persistence. */ function unpersistSession(sessionId: string): void { if (!_aiSessionStore) return; _aiSessionStore.delete(sessionId); } function buildSessionFromRow(row: AiSessionRow): Session { const payload = safeParseJson( row.inputPayload, {}, { throwOnError: true, fieldName: "inputPayload" }, ); const createdAt = new Date(row.createdAt); const updatedAt = new Date(row.updatedAt); if (Number.isNaN(createdAt.getTime()) || Number.isNaN(updatedAt.getTime())) { throw new Error("Invalid session timestamps"); } const currentQuestion = row.currentQuestion ? (safeParseJson(row.currentQuestion, null, { throwOnError: true, fieldName: "currentQuestion", }) ?? undefined) : undefined; return { id: row.id, ip: payload.ip ?? "", initialPlan: payload.initialPlan ?? row.title, title: row.title, projectId: row.projectId ?? undefined, draftModelProvider: payload.modelProvider, draftModelId: payload.modelId, draftSummarizedFor: payload.summarizedFor, history: safeParseJson( row.conversationHistory, [], { throwOnError: true, fieldName: "conversationHistory" }, ), currentQuestion, lastNotifiedQuestionKey: currentQuestion ? `${row.id}:${currentQuestion.id}` : undefined, summary: row.result ? (safeParseJson(row.result, null, { throwOnError: true, fieldName: "result", }) ?? undefined) : undefined, thinkingOutput: row.thinkingOutput, lastGeneratedThinking: row.thinkingOutput || "", error: row.error ?? undefined, createdAt, updatedAt, agent: undefined, }; } export function rehydrateFromStore(store: AiSessionStore): number { let rows: AiSessionRow[] = []; try { rows = store.listRecoverable().filter((row) => row.type === "planning"); } catch (error) { diagnostics.errorFromException("Failed to list recoverable sessions", error, { operation: "list-recoverable" }); return 0; } let rehydrated = 0; for (const row of rows) { try { const session = buildSessionFromRow(row); sessions.set(session.id, session); rehydrated += 1; } catch (error) { diagnostics.errorFromException("Failed to rehydrate session", error, { sessionId: row.id, operation: "rehydrate" }); } } return rehydrated; } // ── Cleanup Interval ──────────────────────────────────────────────────────── /** * Remove expired sessions and stale rate limit entries. * Runs periodically via setInterval. */ function cleanupExpiredSessions(): void { const now = Date.now(); let cleanedSessions = 0; let cleanedRateLimits = 0; // Clean up expired sessions for (const [id, session] of sessions) { if (now - session.updatedAt.getTime() > SESSION_TTL_MS) { if (cleanupInMemorySession(id)) { cleanedSessions++; } } } // Clean up stale rate limit entries for (const [ip, entry] of rateLimits) { if (now - entry.firstRequestAt.getTime() > RATE_LIMIT_WINDOW_MS) { rateLimits.delete(ip); cleanedRateLimits++; } } if (cleanedSessions > 0 || cleanedRateLimits > 0) { diagnostics.info( "Cleanup completed", { cleanedSessions, cleanedRateLimits, operation: "cleanup-expired" } ); } } // Start cleanup interval const cleanupInterval = setInterval(cleanupExpiredSessions, CLEANUP_INTERVAL_MS); cleanupInterval.unref?.(); // Handle graceful shutdown process.on("beforeExit", () => { clearInterval(cleanupInterval); }); // ── Planning Stream Manager ───────────────────────────────────────────────── /** * Manages SSE connections for active planning sessions. * Each session can have multiple connected clients receiving streaming updates. */ export class PlanningStreamManager extends EventEmitter { private readonly sessions = new Map>(); private readonly buffers = new Map(); constructor(private readonly bufferSize = 100) { super(); } /** * Register a client callback for a planning session. * Returns a function to unsubscribe. */ subscribe(sessionId: string, callback: PlanningStreamCallback): () => void { if (!this.sessions.has(sessionId)) { this.sessions.set(sessionId, new Set()); } const callbacks = this.sessions.get(sessionId)!; callbacks.add(callback); return () => { callbacks.delete(callback); if (callbacks.size === 0) { this.sessions.delete(sessionId); } }; } private getBuffer(sessionId: string): SessionEventBuffer { let buffer = this.buffers.get(sessionId); if (!buffer) { buffer = new SessionEventBuffer(this.bufferSize); this.buffers.set(sessionId, buffer); } return buffer; } /** * Broadcast an event to all clients subscribed to a session. * Every event is buffered and assigned a monotonically increasing id. */ broadcast(sessionId: string, event: PlanningStreamEvent): number { const serialized = JSON.stringify((event as { data?: unknown }).data ?? {}); const eventData = typeof serialized === "string" ? serialized : "{}"; const eventId = this.getBuffer(sessionId).push(event.type, eventData); const callbacks = this.sessions.get(sessionId); if (!callbacks) return eventId; for (const callback of callbacks) { nonfatal( () => callback(event, eventId), diagnostics, "Error broadcasting to client", { sessionId, operation: "broadcast" } ); } return eventId; } /** * Get buffered events with id > sinceId for the session. */ getBufferedEvents(sessionId: string, sinceId: number): SessionBufferedEvent[] { const buffer = this.buffers.get(sessionId); if (!buffer) return []; return buffer.getEventsSince(sinceId); } /** * Check if a session has active subscribers. */ hasSubscribers(sessionId: string): boolean { const callbacks = this.sessions.get(sessionId); return callbacks !== undefined && callbacks.size > 0; } /** * Get the number of subscribers for a session. */ getSubscriberCount(sessionId: string): number { return this.sessions.get(sessionId)?.size ?? 0; } /** * Clean up all subscriptions and buffered events for a session. */ cleanupSession(sessionId: string): void { this.sessions.delete(sessionId); this.buffers.delete(sessionId); } /** * Reset all subscriptions and buffers (test helper). */ reset(): void { this.sessions.clear(); this.buffers.clear(); this.removeAllListeners(); } } /** Singleton instance of the planning stream manager */ export const planningStreamManager = new PlanningStreamManager(); // ── Rate Limiting ─────────────────────────────────────────────────────────── /** * Check if IP can create a new planning session. * Returns true if allowed, false if rate limited. */ export function checkRateLimit(ip: string): boolean { const now = Date.now(); const entry = rateLimits.get(ip); if (!entry) { // First request from this IP rateLimits.set(ip, { count: 1, firstRequestAt: new Date(), }); return true; } // Check if window has expired if (now - entry.firstRequestAt.getTime() > RATE_LIMIT_WINDOW_MS) { // Reset window rateLimits.set(ip, { count: 1, firstRequestAt: new Date(), }); return true; } // Within window - check limit if (entry.count >= MAX_SESSIONS_PER_IP_PER_HOUR) { return false; } // Increment count entry.count++; return true; } /** * Get rate limit reset time for an IP. * Returns null if no rate limit entry exists. */ export function getRateLimitResetTime(ip: string): Date | null { const entry = rateLimits.get(ip); if (!entry) return null; return new Date(entry.firstRequestAt.getTime() + RATE_LIMIT_WINDOW_MS); } // ── Session Management ─────────────────────────────────────────────────────── /** * Create a new planning session. * Uses stubbed AI logic for immediate response (no streaming). * For streaming AI responses, use createSessionWithAgent. */ export async function createSession( ip: string, initialPlan: string, _store?: TaskStore, rootDir?: string, promptOverrides?: PromptOverrideMap, planningDepth?: PlanningDepth, customQuestionCount?: number, ): Promise<{ sessionId: string; firstQuestion: PlanningQuestion }> { // Check rate limit if (!checkRateLimit(ip)) { const resetTime = getRateLimitResetTime(ip); throw new RateLimitError( `Rate limit exceeded. Maximum ${MAX_SESSIONS_PER_IP_PER_HOUR} planning sessions per hour. ` + `Reset at ${resetTime?.toISOString() || "unknown"}` ); } if (!rootDir) { throw new Error("rootDir is required for AI-powered planning sessions"); } const sessionId = randomUUID(); const session: Session = { id: sessionId, ip, initialPlan, title: initialPlan.slice(0, 120), history: [], thinkingOutput: "", lastGeneratedThinking: "", createdAt: new Date(), updatedAt: new Date(), }; sessions.set(sessionId, session); persistSession(session, "generating"); // Resolve the effective system prompt (override or default) const baseSystemPrompt = resolvePrompt("planning-system", promptOverrides) || PLANNING_SYSTEM_PROMPT; const depthPromptSuffix = buildDepthPromptSuffix(planningDepth, customQuestionCount); const systemPrompt = depthPromptSuffix ? `${baseSystemPrompt}\n\n${depthPromptSuffix}` : baseSystemPrompt; // Create AI agent and get the first question // Only await engineReady if createFnAgent hasn't been set externally (e.g., via __setCreateFnAgent) if (!createFnAgent) { await ensureEngineReady(); } const agentResult = await createFnAgent({ cwd: rootDir, systemPrompt, tools: "readonly", onThinking: () => { // Non-streaming path ignores thinking output }, onText: () => { // Non-streaming path ignores incremental text }, }); session.agent = agentResult; session.updatedAt = new Date(); // Send initial plan to get first question from AI const firstQuestion = await getFirstQuestionFromAgent(session, initialPlan); session.currentQuestion = firstQuestion; session.updatedAt = new Date(); persistSession(session, "awaiting_input"); return { sessionId, firstQuestion }; } /** * Get the first question from the AI agent by sending the initial plan. * Waits for the agent response and parses it as a PlanningQuestion. * Throws if the agent returns a summary instead of a question. */ async function getFirstQuestionFromAgent( session: Session, message: string ): Promise { if (!session.agent) { throw new InvalidSessionStateError("AI agent not initialized"); } // Send message to agent await session.agent.session.prompt(message); // Extract response text interface AgentMessage { role: string; content?: string | Array<{ type: string; text?: string; thinking?: string }>; } const lastMessage = (session.agent.session.state.messages as AgentMessage[]) .filter((m: AgentMessage) => m.role === "assistant") .pop(); let responseText = ""; if (lastMessage?.content) { if (typeof lastMessage.content === "string") { responseText = lastMessage.content; } else if (Array.isArray(lastMessage.content)) { // Try text blocks first const textContent = lastMessage.content .filter((c): c is { type: "text"; text: string } => c.type === "text" && typeof c.text === "string") .map((c) => c.text) .join(""); if (textContent) { responseText = textContent; } else { // Fallback: extract thinking blocks when no text blocks are present const thinkingContent = lastMessage.content .filter((c): c is { type: "thinking"; thinking: string } => c.type === "thinking" && typeof c.thinking === "string") .map((c) => c.thinking) .join(""); responseText = thinkingContent; } } } // Diagnostic: warn when response text is empty or very short if (!responseText || responseText.length < 10) { const contentBlockTypes = Array.isArray(lastMessage?.content) ? lastMessage.content.map((c: { type: string }) => c.type) : typeof lastMessage?.content === "string" ? ["string"] : []; diagnostics.warn( "Response text is empty or very short before parse", { sessionId: session.id, responseTextLength: responseText.length, contentBlockTypes, usedThinkingBlocksFallback: !Array.isArray(lastMessage?.content) ? false : !lastMessage.content.some((c: { type: string }) => c.type === "text"), operation: "response-extraction", } ); } // Parse response with retry let parsed: PlanningResponse | undefined; let lastError: Error | undefined; for (let attempt = 0; attempt <= MAX_PARSE_RETRIES; attempt++) { try { parsed = parseAgentResponse(responseText); break; } catch (err) { lastError = err instanceof Error ? err : new Error(String(err)); if (attempt < MAX_PARSE_RETRIES) { try { await session.agent.session.prompt( "Your previous response could not be parsed as JSON. " + 'Please respond with ONLY a valid JSON object: {"type":"question","data":{...}}. ' + "No markdown, no explanation, just the JSON." ); const retryMessage = (session.agent.session.state.messages as AgentMessage[]) .filter((m: AgentMessage) => m.role === "assistant") .pop(); if (retryMessage?.content) { if (typeof retryMessage.content === "string") { responseText = retryMessage.content; } else if (Array.isArray(retryMessage.content)) { const textContent = retryMessage.content .filter((c): c is { type: "text"; text: string } => c.type === "text" && typeof c.text === "string") .map((c) => c.text) .join(""); if (textContent) { responseText = textContent; } else { const thinkingContent = retryMessage.content .filter((c): c is { type: "thinking"; thinking: string } => c.type === "thinking" && typeof c.thinking === "string") .map((c) => c.thinking) .join(""); responseText = thinkingContent; } } } } catch { break; } } } } if (!parsed) { // Clean up the session on failure sessions.delete(session.id); unpersistSession(session.id); throw new Error( `Failed to get first question from AI: ${lastError?.message || "Unknown error"}` ); } if (parsed.type === "complete") { // AI returned a summary instead of a question — return a minimal question // so the caller can present the summary const summary = parsed.data; session.summary = summary; persistSession(session, "complete"); return { id: "q-direct-summary", type: "confirm", question: `The AI has generated a plan: "${summary.title}". Proceed with this?`, description: summary.description, }; } return parsed.data; } export async function createDraftSession( ip: string, initialPlan: string, _rootDir: string, modelProvider?: string, modelId?: string, _promptOverrides?: PromptOverrideMap, options?: { projectId?: string }, ): Promise<{ sessionId: string; title: string }> { if (!checkRateLimit(ip)) { const resetTime = getRateLimitResetTime(ip); throw new RateLimitError( `Rate limit exceeded. Maximum ${MAX_SESSIONS_PER_IP_PER_HOUR} planning sessions per hour. ` + `Reset at ${resetTime?.toISOString() || "unknown"}`, ); } const sessionId = randomUUID(); const title = DRAFT_PLACEHOLDER_TITLE; // Pair modelProvider+modelId — the runtime treats half-set overrides as // invalid (resolveTaskPlanningModel etc.), so persist nothing rather than // a half-configured override that would mislead reopen. const hasModelOverride = Boolean(modelProvider && modelId); const session: Session = { id: sessionId, ip, initialPlan, title, projectId: options?.projectId, draftModelProvider: hasModelOverride ? modelProvider : undefined, draftModelId: hasModelOverride ? modelId : undefined, history: [], thinkingOutput: "", lastGeneratedThinking: "", createdAt: new Date(), updatedAt: new Date(), }; sessions.set(sessionId, session); persistSession(session, "draft"); return { sessionId, title }; } /** * Replace a draft session's placeholder sidebar title with an AI-summarized * one derived from the latest persisted initialPlan. Bounded by status, not * by title content, so a user who blurs once then keeps editing still gets * the title refreshed on subsequent blurs/close (otherwise the title would * lock to the first summary and silently diverge from what they typed). * - Only runs against rows still in `draft` status — once a session has been * started, its title is owned by the start path / final summary and must * not be overwritten by a stale draft summarize call that arrives late. * - Reads the current initialPlan and any persisted model override from * SQLite so the summary reflects whatever the latest debounced PATCH * /draft persisted, and uses the model the draft was created under. * - Re-checks status (not title) after the model call to detect a * concurrent start and avoid clobbering the generating/awaiting_input * title with a stale draft summary. * * Returns the resolved title (existing or freshly generated) or null if the * session was not eligible for summarization. */ export async function summarizeDraftTitle( sessionId: string, rootDir: string, modelProvider?: string, modelId?: string, ): Promise { if (!_aiSessionStore) return null; const row = _aiSessionStore.get(sessionId); if (!row || row.type !== "planning" || row.status !== "draft") { return null; } const payload = safeParseJson(row.inputPayload, {}); const trimmed = (payload.initialPlan ?? "").trim(); if (!trimmed) return null; // Prefer the model the draft was created under; fall back to the caller- // supplied override (e.g. project/global planning settings). const effectiveProvider = payload.modelProvider ?? modelProvider; const effectiveModelId = payload.modelId ?? modelId; let finalTitle = trimmed.slice(0, 60).trim(); try { const generated = await summarizeTitle(trimmed, rootDir, effectiveProvider, effectiveModelId); finalTitle = generated?.trim() || finalTitle; } catch (error) { diagnostics.errorFromException( "summarizeDraftTitle: model call failed, falling back to truncated text", error, { sessionId, operation: "summarize-draft-title" }, ); } if (!finalTitle) return null; // Re-check status (not title) so a concurrent Start Planning or a later // edit-then-blur cycle doesn't overwrite a real generating/complete title. const latest = _aiSessionStore.get(sessionId); if (!latest || latest.status !== "draft") { return latest?.title ?? null; } _aiSessionStore.markDraftSummarized(sessionId, finalTitle, trimmed); const session = sessions.get(sessionId); if (session) { session.title = finalTitle; session.draftSummarizedFor = trimmed; } return finalTitle; } export async function startExistingSession( sessionId: string, rootDir: string, modelProvider?: string, modelId?: string, promptOverrides?: PromptOverrideMap, ): Promise { let session = sessions.get(sessionId); // Draft sessions aren't included in rehydrateFromStore (which only loads // recoverable in-flight sessions), and a backend restart drops the in-memory // map entirely. Rebuild lazily from SQLite so persisted drafts can still be // started after a restart, and so updateDraft-only state survives. if (!session && _aiSessionStore) { const row = _aiSessionStore.get(sessionId); if (row && row.type === "planning") { try { session = buildSessionFromRow(row); sessions.set(sessionId, session); } catch (error) { diagnostics.errorFromException( "Failed to rebuild planning session from store", error, { sessionId, operation: "start-existing-rebuild" }, ); } } } if (!session) { throw new SessionNotFoundError(`Planning session ${sessionId} not found or expired`); } // Drafts are sync'd via aiSessionStore.updateDraft, which only writes // SQLite. Pull the latest initialPlan + persisted model override + the // text the current title was already summarized from. Lets the agent // see everything the user typed, lets summarize use the original model, // and lets us skip a redundant model call when blur/close already // summarized this exact text. let persistedProvider: string | undefined; let persistedModelId: string | undefined; let persistedSummarizedFor: string | undefined; let cameFromDraft = false; if (_aiSessionStore) { const row = _aiSessionStore.get(sessionId); if (row) { cameFromDraft = row.status === "draft"; const payload = safeParseJson(row.inputPayload, {}); if (payload.initialPlan) { session.initialPlan = payload.initialPlan; } persistedProvider = payload.modelProvider; persistedModelId = payload.modelId; persistedSummarizedFor = payload.summarizedFor; } } // Re-summarize when transitioning out of draft so the title reflects the // FINAL text — but skip the model call when blur/close already produced a // summary for this exact text and the user hasn't edited since. Saves // tokens when the user blurs the textarea then immediately clicks Start. if (cameFromDraft) { const trimmed = session.initialPlan.trim(); const alreadySummarized = session.title !== DRAFT_PLACEHOLDER_TITLE && persistedSummarizedFor === trimmed; if (!alreadySummarized) { const fallback = trimmed.slice(0, 60).trim(); if (session.title === DRAFT_PLACEHOLDER_TITLE) { session.title = fallback || DRAFT_PLACEHOLDER_TITLE; } const summarizeProvider = modelProvider ?? persistedProvider; const summarizeModelId = modelId ?? persistedModelId; void (async () => { try { const generated = await summarizeTitle(trimmed, rootDir, summarizeProvider, summarizeModelId); const finalTitle = generated?.trim() || fallback; if (!finalTitle) return; session.title = finalTitle; _aiSessionStore?.updateTitle(sessionId, finalTitle); } catch { // Keep fallback title } })(); } } persistSession(session, "generating"); await initializeAgent(session, rootDir, modelProvider, modelId, promptOverrides); } /** * Create a new planning session with AI agent streaming. * This initializes an AI agent that will stream thinking output via SSE. * * @param ip - Client IP for rate limiting * @param initialPlan - The user's initial plan description * @param rootDir - Project root directory for AI agent context * @param modelProvider - Optional AI model provider override * @param modelId - Optional AI model ID override * @param promptOverrides - Optional prompt override map for system prompt customization * @returns Session ID (use with planningStreamManager to receive events) */ export async function createSessionWithAgent( ip: string, initialPlan: string, rootDir: string, modelProvider?: string, modelId?: string, promptOverrides?: PromptOverrideMap, options?: { projectId?: string; ntfyConfig?: PlanningNtfyConfig; planningDepth?: PlanningDepth; customQuestionCount?: number; }, ): Promise { // Check rate limit if (!checkRateLimit(ip)) { const resetTime = getRateLimitResetTime(ip); throw new RateLimitError( `Rate limit exceeded. Maximum ${MAX_SESSIONS_PER_IP_PER_HOUR} planning sessions per hour. ` + `Reset at ${resetTime?.toISOString() || "unknown"}` ); } const sessionId = randomUUID(); const session: Session = { id: sessionId, ip, initialPlan, title: initialPlan.slice(0, 120), projectId: options?.projectId, ntfyConfig: options?.ntfyConfig ? { enabled: options.ntfyConfig.enabled, topic: options.ntfyConfig.topic, dashboardHost: options.ntfyConfig.dashboardHost, events: options.ntfyConfig.events ? [...options.ntfyConfig.events] : undefined, ntfyBaseUrl: options.ntfyConfig.ntfyBaseUrl, } : undefined, history: [], thinkingOutput: "", lastGeneratedThinking: "", createdAt: new Date(), updatedAt: new Date(), }; sessions.set(sessionId, session); persistSession(session, "generating"); // Initialize AI agent in background - it will stream via planningStreamManager initializeAgent( session, rootDir, modelProvider, modelId, promptOverrides, options?.planningDepth, options?.customQuestionCount, ).catch((err) => { diagnostics.errorFromException("Failed to initialize agent for session", err, { sessionId, operation: "initialize-agent" }); persistSession(session, "error", err.message || "Failed to initialize AI agent"); planningStreamManager.broadcast(sessionId, { type: "error", data: err.message || "Failed to initialize AI agent", }); }); return sessionId; } /** * Initialize the AI agent for a session and start the first turn. */ async function initializeAgent( session: Session, rootDir: string, modelProvider?: string, modelId?: string, promptOverrides?: PromptOverrideMap, planningDepth?: PlanningDepth, customQuestionCount?: number, ): Promise { try { session.agent = await createPlanningAgent( session, rootDir, modelProvider, modelId, promptOverrides, planningDepth, customQuestionCount, ); session.updatedAt = new Date(); // Send initial message to get first question await continueAgentConversation(session, session.initialPlan); } catch (err) { const errorMessage = err instanceof Error ? err.message : "Failed to initialize AI agent"; diagnostics.errorFromException("Agent initialization error for session", err, { sessionId: session.id, operation: "initialize-agent" }); session.error = errorMessage; session.updatedAt = new Date(); persistSession(session, "error", errorMessage); planningStreamManager.broadcast(session.id, { type: "error", data: errorMessage, }); } } async function createPlanningAgent( session: Session, rootDir: string, modelProvider?: string, modelId?: string, promptOverrides?: PromptOverrideMap, planningDepth?: PlanningDepth, customQuestionCount?: number, ): Promise { // Ensure engine is loaded before using createFnAgent await ensureEngineReady(); // Resolve the effective system prompt (override or default) const baseSystemPrompt = resolvePrompt("planning-system", promptOverrides) || PLANNING_SYSTEM_PROMPT; const depthPromptSuffix = buildDepthPromptSuffix(planningDepth, customQuestionCount); const systemPrompt = depthPromptSuffix ? `${baseSystemPrompt}\n\n${depthPromptSuffix}` : baseSystemPrompt; return createFnAgent({ cwd: rootDir, systemPrompt, tools: "readonly", ...(modelProvider && modelId ? { defaultProvider: modelProvider, defaultModelId: modelId, } : {}), onThinking: (delta: string) => { session.thinkingOutput += delta; persistThinking(session.id, session.thinkingOutput); planningStreamManager.broadcast(session.id, { type: "thinking", data: delta, }); }, onText: (delta: string) => { // 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. session.thinkingOutput += delta; persistThinking(session.id, session.thinkingOutput); planningStreamManager.broadcast(session.id, { type: "thinking", data: delta, }); }, }); } function buildHistoryReplayPrompt( history: Array<{ question: PlanningQuestion; response: unknown }>, ): string { const interviewSummary = formatInterviewQA(history); if (!interviewSummary) { return "No prior planning interview context is available."; } return [ "Previous conversation summary:", interviewSummary, "Use this as context for the next response. Do not repeat prior questions unless necessary.", ].join("\n\n"); } async function ensureSessionAgent( session: Session, rootDir: string | undefined, historyForReplay: Array<{ question: PlanningQuestion; response: unknown }>, promptOverrides?: PromptOverrideMap, ): Promise { if (session.agent) { return; } if (!rootDir) { throw new InvalidSessionStateError( "Planning session has no AI agent and cannot be resumed without project context", ); } session.agent = await createPlanningAgent(session, rootDir, undefined, undefined, promptOverrides); if (historyForReplay.length === 0) { return; } const contextMessage = buildHistoryReplayPrompt(historyForReplay); await session.agent.session.prompt(contextMessage); } async function maybeNotifyPlanningAwaitingInput(session: Session, question: PlanningQuestion): Promise { const config = session.ntfyConfig; if (!config?.enabled || !config.topic) { return; } await ensureNtfyHelpersReady(); const eventEnabled = planningNtfyHelpers?.isNtfyEventEnabled ? planningNtfyHelpers.isNtfyEventEnabled(config.events, "planning-awaiting-input") : (config.events ? config.events.includes("planning-awaiting-input") : true); if (!eventEnabled) { return; } const questionKey = `${session.id}:${question.id}`; if (session.lastNotifiedQuestionKey === questionKey) { return; } session.lastNotifiedQuestionKey = questionKey; if (!planningNtfyHelpers) { return; } try { const clickUrl = planningNtfyHelpers.buildNtfyClickUrl({ dashboardHost: config.dashboardHost, projectId: session.projectId, }); await planningNtfyHelpers.sendNtfyNotification({ ntfyBaseUrl: config.ntfyBaseUrl, topic: config.topic, title: "Planning needs your input", message: `Planning mode is waiting for input: ${question.question}`, priority: "high", clickUrl, }); } catch (error) { diagnostics.warn("Failed to deliver planning awaiting-input ntfy notification", { sessionId: session.id, questionId: question.id, error: error instanceof Error ? error.message : String(error), operation: "planning-notify-awaiting-input", }); } } /** Max number of retry attempts when AI returns unparseable output */ const MAX_PARSE_RETRIES = 1; /** * Continue the AI conversation with a user message. * * Includes a bounded recovery path: if the AI response cannot be parsed, * one retry attempt is made with a reformat prompt before emitting a * terminal session error. */ function setSessionError(session: Session, message: string): void { session.error = message; session.updatedAt = new Date(); persistSession(session, "error", message); planningStreamManager.broadcast(session.id, { type: "error", data: message, }); } function createAbortError(): Error { const error = new Error("Generation aborted"); error.name = "AbortError"; return error; } async function runGenerationWithTimeout(session: Session, operation: () => Promise): Promise { const existing = activeGenerations.get(session.id); if (existing) { clearTimeout(existing.timer); existing.abortController.abort(); } const abortController = new AbortController(); let timeoutTriggered = false; const timer = setTimeout(() => { timeoutTriggered = true; setSessionError(session, "AI generation timed out. You can retry or start a new session."); abortController.abort(); }, GENERATION_TIMEOUT_MS); activeGenerations.set(session.id, { abortController, timer }); const abortPromise = new Promise((_, reject) => { abortController.signal.addEventListener( "abort", () => reject(createAbortError()), { once: true }, ); }); try { return await Promise.race([operation(), abortPromise]); } catch (error) { if (error instanceof Error && error.name === "AbortError") { if (!timeoutTriggered && !session.error) { setSessionError(session, "Generation stopped by user. You can retry or start a new session."); } } throw error; } finally { clearTimeout(timer); activeGenerations.delete(session.id); } } async function continueAgentConversation(session: Session, message: string): Promise { if (!session.agent) { throw new InvalidSessionStateError("AI agent not initialized"); } try { await runGenerationWithTimeout(session, async () => { // Clear thinking output for this turn session.thinkingOutput = ""; // Send message to agent using .prompt() - it will stream thinking via onThinking callback await session.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(); let responseText = session.thinkingOutput; if (lastMessage?.content) { // Handle both string and array content types if (typeof lastMessage.content === "string") { responseText = lastMessage.content; } else if (Array.isArray(lastMessage.content)) { // Extract text from content blocks; only overwrite fallback if non-empty const extracted = 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 (extracted) { responseText = extracted; } } } // Diagnostic: warn when response text is empty or very short if (!responseText || responseText.length < 10) { const contentBlockTypes = Array.isArray(lastMessage?.content) ? lastMessage.content.map((c: { type: string }) => c.type) : typeof lastMessage?.content === "string" ? ["string"] : []; diagnostics.warn( "Response text is empty or very short before parse", { sessionId: session.id, responseTextLength: responseText.length, contentBlockTypes, usedThinkingOutputFallback: responseText === session.thinkingOutput, operation: "response-extraction", } ); } // Parse the JSON response with retry let parsed: PlanningResponse | undefined; let lastError: Error | undefined; for (let attempt = 0; attempt <= MAX_PARSE_RETRIES; attempt++) { try { parsed = parseAgentResponse(responseText); break; // success } catch (err) { lastError = err instanceof Error ? err : new Error(String(err)); if (attempt < MAX_PARSE_RETRIES) { // Retry: ask the AI to reformat as clean JSON 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":{...}}. No markdown, no explanation, just the JSON.' ); // Get the new response text 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)) { const extracted = 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(""); if (extracted) { retryText = extracted; } } } responseText = retryText; } catch (retryErr) { // Retry prompt itself failed — give up diagnostics.errorFromException( "Retry prompt failed for session", retryErr, { sessionId: session.id, operation: "retry-prompt" } ); break; } } } } if (!parsed) { // All attempts exhausted — emit actionable error const errorMsg = `${lastError?.message || "Failed to parse AI response"} You can try responding again or start a new planning 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); planningStreamManager.broadcast(session.id, { type: "error", data: errorMsg, }); return; } if (parsed.type === "question") { session.currentQuestion = parsed.data; session.summary = undefined; session.error = undefined; session.lastGeneratedThinking = session.thinkingOutput; session.updatedAt = new Date(); persistSession(session, "awaiting_input"); void maybeNotifyPlanningAwaitingInput(session, parsed.data); planningStreamManager.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"); planningStreamManager.broadcast(session.id, { type: "summary", data: parsed.data, }); planningStreamManager.broadcast(session.id, { type: "complete" }); } }); } catch (err) { if (err instanceof Error && err.name === "AbortError") { return; } const errorMessage = err instanceof Error ? err.message : "AI processing failed"; diagnostics.errorFromException("Agent conversation error for session", err, { sessionId: session.id, operation: "conversation" }); setSessionError(session, errorMessage); } } /** * Extract the best JSON candidate from AI response text. * * Handles: * - Markdown-wrapped JSON (```json ... ```) * - JSON embedded in leading/trailing prose * - Multiple JSON objects (picks the largest balanced one) * * Returns the extracted JSON string or null if nothing usable is found. */ function extractJsonCandidate(text: string): string | null { if (!text || !text.trim()) return null; // 1. Try markdown code blocks first (most reliable) const codeBlockMatch = text.match(/```(?:json)?\s*([\s\S]*?)\s*```/); if (codeBlockMatch?.[1]) { const candidate = codeBlockMatch[1].trim(); if (candidate.startsWith("{")) return candidate; } // 2. Find all top-level brace-delimited objects using balanced brace counting const candidates: Array<{ start: number; end: number; text: string }> = []; for (let i = 0; i < text.length; i++) { if (text[i] === "{") { let depth = 0; let inString = false; let escape = false; for (let j = i; j < text.length; j++) { const ch = text[j]; if (escape) { escape = false; continue; } if (ch === "\\") { escape = true; continue; } if (ch === '"') { inString = !inString; continue; } if (inString) continue; if (ch === "{") depth++; if (ch === "}") depth--; if (depth === 0) { const candidate = text.slice(i, j + 1).trim(); // Only accept candidates that parse as valid JSON try { JSON.parse(candidate); candidates.push({ start: i, end: j, text: candidate }); } catch { // Not valid JSON, skip } break; } } } } // Pick the largest valid candidate (most likely the full response) if (candidates.length > 0) { candidates.sort((a, b) => b.text.length - a.text.length); return candidates[0].text; } // 3. Last resort: try the full trimmed text const trimmed = text.trim(); if (trimmed.startsWith("{")) return trimmed; return null; } /** * Attempt to repair common JSON issues: * - Truncated JSON (missing closing braces) * - Trailing commas before closing braces * - Missing closing quotes * * Returns the repaired string, or the original if no repair was possible. */ function repairJson(text: string): string { let repaired = text; // Fix trailing commas before } or ] repaired = repaired.replace(/,\s*([}\]])/g, "$1"); // Count open/close braces and brackets let openBraces = 0; let openBrackets = 0; let inString = false; let escape = false; for (const ch of repaired) { if (escape) { escape = false; continue; } if (ch === "\\") { escape = true; continue; } if (ch === '"') { inString = !inString; continue; } if (inString) continue; if (ch === "{") openBraces++; if (ch === "}") openBraces--; if (ch === "[") openBrackets++; if (ch === "]") openBrackets--; } // If we're in an unclosed string, close it if (inString) { repaired += '"'; } // Re-count after potential string fix openBraces = 0; openBrackets = 0; inString = false; escape = false; for (const ch of repaired) { if (escape) { escape = false; continue; } if (ch === "\\") { escape = true; continue; } if (ch === '"') { inString = !inString; continue; } if (inString) continue; if (ch === "{") openBraces++; if (ch === "}") openBraces--; if (ch === "[") openBrackets++; if (ch === "]") openBrackets--; } // Close unclosed brackets and braces repaired += "]".repeat(Math.max(0, openBrackets)); repaired += "}".repeat(Math.max(0, openBraces)); return repaired; } /** * Parse agent response JSON with robust extraction and recovery. * * Strategy: * 1. Extract JSON candidate from text (handles markdown wrapping, prose) * 2. Try parsing directly * 3. If parse fails, attempt repair (truncated JSON, trailing commas) * 4. Validate the resulting structure */ export function parseAgentResponse(text: string): PlanningResponse { const candidate = extractJsonCandidate(text); if (!candidate) { diagnostics.error("No JSON candidate found in agent response", { inputSnippet: text.slice(0, 500), operation: "parse-json" }); throw new Error("AI returned no valid JSON. Please try again."); } let parsed: unknown; try { parsed = JSON.parse(candidate); } catch { // Attempt repair for truncated/malformed JSON try { const repaired = repairJson(candidate); parsed = JSON.parse(repaired); } catch (repairErr) { diagnostics.error( "Failed to parse agent response (repair also failed)", { inputSnippet: candidate.slice(0, 500), operation: "parse-json-repair" } ); throw new Error( `Failed to parse AI response: ${repairErr instanceof Error ? repairErr.message : "Unknown error"}. Please try again.` ); } } // Validate structure if ( typeof parsed === "object" && parsed !== null && "type" in parsed && "data" in parsed ) { const typed = parsed as { type: string; data: unknown }; if ( (typed.type === "question" || typed.type === "complete") && typed.data !== null && typed.data !== undefined ) { return parsed as PlanningResponse; } } diagnostics.error("Invalid response structure from AI", { parsedSnippet: JSON.stringify(parsed).slice(0, 500), operation: "parse-validate" }); throw new Error("AI returned an invalid response structure. Please try again."); } /** * Submit a response to the current question and get the next question or summary. * Supports both stubbed mode and AI agent mode. */ function isRefineRequest(responses: Record): boolean { return responses.refine === true; } function formatRefineRequestForAgent(summary: PlanningSummary): string { return [ "The user clicked Refine Further on the planning summary.", "Continue the planning interview from the existing context.", "Either ask one focused follow-up question or return an updated completion summary if sufficient.", "Current summary:", JSON.stringify(summary), ].join("\n\n"); } export async function submitResponse( sessionId: string, responses: Record, rootDir?: string, promptOverrides?: PromptOverrideMap, ): Promise { const session = getSession(sessionId); if (!session) { throw new SessionNotFoundError(`Planning session ${sessionId} not found or expired`); } if (!session.currentQuestion) { if (!isRefineRequest(responses) || !session.summary) { throw new InvalidSessionStateError("No active question in session"); } session.error = undefined; persistSession(session, "generating"); await ensureSessionAgent(session, rootDir, session.history, promptOverrides); const refineMessage = formatRefineRequestForAgent(session.summary); await continueAgentConversation(session, refineMessage); } else { // Record the response session.history.push({ question: session.currentQuestion, response: responses, thinkingOutput: session.lastGeneratedThinking || "", }); session.error = undefined; persistSession(session, "generating"); if (!session.agent) { const replayHistory = session.history.slice(0, -1); await ensureSessionAgent(session, rootDir, replayHistory, promptOverrides); } const message = formatResponseForAgent(session.currentQuestion, responses); await continueAgentConversation(session, message); } // Return the current state (will be updated via SSE) if (session.summary) { return { type: "complete", data: session.summary }; } if (session.currentQuestion) { return { type: "question", data: session.currentQuestion }; } // Should not reach here, but handle gracefully throw new InvalidSessionStateError("AI agent did not return a question or summary"); } export async function retrySession( sessionId: string, rootDir: string, promptOverrides?: PromptOverrideMap, ): Promise { const session = getSession(sessionId); if (!session) { throw new SessionNotFoundError(`Planning session ${sessionId} not found or expired`); } const persisted = _aiSessionStore?.get(sessionId); if (persisted && persisted.type !== "planning") { throw new SessionNotFoundError(`Planning session ${sessionId} not found or expired`); } const inErrorState = persisted ? persisted.status === "error" : Boolean(session.error); if (!inErrorState) { throw new InvalidSessionStateError(`Planning session ${sessionId} is not in an error state`); } disposeSessionAgentForRetry(session); session.error = undefined; session.summary = undefined; session.updatedAt = new Date(); persistSession(session, "generating"); if (session.history.length === 0) { await ensureSessionAgent(session, rootDir, [], promptOverrides); await continueAgentConversation(session, session.initialPlan); return; } const replayHistory = session.history.slice(0, -1); const lastEntry = session.history[session.history.length - 1]; await ensureSessionAgent(session, rootDir, replayHistory, promptOverrides); const replayMessage = formatResponseForAgent( lastEntry.question, coerceResponseRecord(lastEntry.question, lastEntry.response), ); await continueAgentConversation(session, replayMessage); } export interface PlanningRewindResult { currentQuestion: PlanningQuestion; history: PlanningHistoryEntry[]; } export async function rewindSession( sessionId: string, rootDir?: string, promptOverrides?: PromptOverrideMap, ): Promise { const session = getSession(sessionId); if (!session) { throw new SessionNotFoundError(`Planning session ${sessionId} not found or expired`); } if (session.history.length === 0) { throw new InvalidSessionStateError("Planning session has no previous question to rewind to"); } const rewindEntry = session.history.pop(); if (!rewindEntry) { throw new InvalidSessionStateError("Planning session has no previous question to rewind to"); } disposeSessionAgentForRetry(session); session.currentQuestion = rewindEntry.question; session.summary = undefined; 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); } persistSession(session, "awaiting_input"); planningStreamManager.broadcast(session.id, { type: "question", data: rewindEntry.question }); return { currentQuestion: rewindEntry.question, history: [...session.history], }; } export function stopGeneration(sessionId: string): boolean { const session = sessions.get(sessionId); const activeGeneration = activeGenerations.get(sessionId); if (!session || !activeGeneration) { return false; } activeGeneration.abortController.abort(); clearTimeout(activeGeneration.timer); activeGenerations.delete(sessionId); if (session.agent) { nonfatal( () => session.agent?.session.dispose?.(), diagnostics, "Error disposing agent for stop-generation", { sessionId, operation: "stop-generation-dispose" }, ); session.agent = undefined; } setSessionError(session, "Generation stopped by user. You can retry or start a new session."); return true; } /** * Format user response as a message for the AI agent. */ function formatResponseForAgent( question: PlanningQuestion, responses: Record ): string { const responseValue = responses[question.id]; const comment = typeof responses._comment === "string" ? responses._comment.trim() : ""; let formatted: string; switch (question.type) { case "text": formatted = `Question: ${question.question}\n\nAnswer: ${responseValue}`; break; case "single_select": if (typeof responseValue === "string") { const option = question.options?.find((o) => o.id === responseValue); formatted = `Question: ${question.question}\n\nSelected: ${option?.label || responseValue}`; break; } formatted = `Question: ${question.question}\n\nAnswer: ${responseValue}`; break; case "multi_select": if (Array.isArray(responseValue)) { const selected = responseValue.map((id) => { const option = question.options?.find((o) => o.id === id); return option?.label || id; }); formatted = `Question: ${question.question}\n\nSelected: ${selected.join(", ")}`; break; } formatted = `Question: ${question.question}\n\nAnswer: ${responseValue}`; break; case "confirm": formatted = `Question: ${question.question}\n\nAnswer: ${responseValue === true ? "Yes" : "No"}`; break; default: formatted = `Question: ${question.question}\n\nAnswer: ${JSON.stringify(responseValue)}`; break; } return comment.length > 0 ? `${formatted}\n\nAdditional context: ${comment}` : formatted; } function coerceResponseRecord(question: PlanningQuestion, response: unknown): Record { if (response && typeof response === "object" && !Array.isArray(response)) { return response as Record; } return { [question.id]: response, }; } function disposeSessionAgentForRetry(session: Session): void { if (!session.agent) { return; } nonfatal( () => session.agent.session.dispose?.(), diagnostics, "Error disposing agent for retry", { sessionId: session.id, operation: "dispose-retry" } ); session.agent = undefined; } function formatInterviewAnswer(question: PlanningQuestion, responseValue: unknown): string { switch (question.type) { case "text": return typeof responseValue === "string" ? responseValue : String(responseValue ?? ""); case "single_select": if (typeof responseValue === "string") { const option = question.options?.find((candidate) => candidate.id === responseValue); return option?.label || responseValue; } return String(responseValue ?? ""); case "multi_select": if (Array.isArray(responseValue)) { const selected = responseValue.map((id) => { if (typeof id !== "string") { return String(id); } const option = question.options?.find((candidate) => candidate.id === id); return option?.label || id; }); return selected.join(", "); } return String(responseValue ?? ""); case "confirm": return responseValue === true ? "Yes" : "No"; default: return JSON.stringify(responseValue); } } /** * Format planning interview Q&A history for task descriptions and logs. */ export function formatInterviewQA( history: Array<{ question: PlanningQuestion; response: unknown }> ): string { if (history.length === 0) { return ""; } const entries = history.map(({ question, response }) => { const responseRecord = response && typeof response === "object" && !Array.isArray(response) ? (response as Record) : undefined; const responseValue = responseRecord ? responseRecord[question.id] : response; const comment = typeof responseRecord?._comment === "string" ? responseRecord._comment.trim() : ""; const answerLine = `**Q: ${question.question}**\nA: ${formatInterviewAnswer(question, responseValue)}`; return comment.length > 0 ? `${answerLine}\nComment: ${comment}` : answerLine; }); return `## Planning Interview Context\n\n${entries.join("\n\n")}`; } /** * Cancel and cleanup a planning session. */ export async function cancelSession(sessionId: string): Promise { const removed = cleanupInMemorySession(sessionId); if (!removed) { throw new SessionNotFoundError(`Planning session ${sessionId} not found or expired`); } unpersistSession(sessionId); } /** * Get session details. */ export function getSession(sessionId: string): Session | undefined { const inMemory = sessions.get(sessionId); if (inMemory) { return inMemory; } if (!_aiSessionStore) { return undefined; } const row = _aiSessionStore.get(sessionId); if (!row || row.type !== "planning") { return undefined; } try { const restored = buildSessionFromRow(row); sessions.set(restored.id, restored); return restored; } catch (error) { diagnostics.errorFromException("Failed to restore session from SQLite", error, { sessionId, operation: "restore" }); return undefined; } } /** * Get the current question for a session. */ export function getCurrentQuestion(sessionId: string): PlanningQuestion | undefined { return sessions.get(sessionId)?.currentQuestion; } /** * Get the summary for a completed session. */ export function getSummary(sessionId: string): PlanningSummary | undefined { return sessions.get(sessionId)?.summary; } /** * Generate subtasks from a completed planning summary. * Uses the planning session's summary to create a SubtaskItem[] for multi-task creation. * * @param sessionId - The planning session ID * @returns Array of SubtaskItem with titles derived from keyDeliverables, or fallback */ function buildPlanningSubtaskDescription(input: { taskGuidance: string; summaryDescription: string; qaSection: string; }): string { const contextSections = [ "## Larger Plan Context", input.summaryDescription, ]; if (input.qaSection) { contextSections.push(input.qaSection); } return `${input.taskGuidance}\n\n${contextSections.join("\n\n")}`; } export function generateSubtasksFromPlanning(sessionId: string): SubtaskItem[] { const session = sessions.get(sessionId); if (!session) return []; if (!session.summary) return []; const { summary } = session; const qaSection = formatInterviewQA(session.history); // If key deliverables exist, create one subtask per deliverable if (summary.keyDeliverables.length > 0) { return summary.keyDeliverables.map((deliverable, index) => { const id = `subtask-${index + 1}`; const dependsOn = index > 0 ? [`subtask-${index}`] : [] as string[]; return { id, title: deliverable, description: buildPlanningSubtaskDescription({ taskGuidance: `Implement "${deliverable}" as this subtask's primary outcome. Focus only on the concrete changes needed to deliver this item.`, summaryDescription: summary.description, qaSection, }), suggestedSize: index === 0 ? "S" as const : index === summary.keyDeliverables.length - 1 ? "S" as const : "M" as const, dependsOn, }; }); } // Fallback: 3 subtasks return [ { id: "subtask-1", title: "Define implementation approach", description: buildPlanningSubtaskDescription({ taskGuidance: "Define the implementation approach for the plan, including architecture and sequencing decisions needed before coding.", summaryDescription: summary.description, qaSection, }), suggestedSize: "S" as const, dependsOn: [], }, { id: "subtask-2", title: "Implement core changes", description: buildPlanningSubtaskDescription({ taskGuidance: "Implement the core code changes described by the plan, using the agreed approach from the prior subtask.", summaryDescription: summary.description, qaSection, }), suggestedSize: "M" as const, dependsOn: ["subtask-1"], }, { id: "subtask-3", title: "Verify and polish", description: buildPlanningSubtaskDescription({ taskGuidance: "Verify the implementation end-to-end, then polish quality items like tests, docs, and edge-case handling.", summaryDescription: summary.description, qaSection, }), suggestedSize: "S" as const, dependsOn: ["subtask-2"], }, ]; } /** * Cleanup a session (used after task creation). */ export function cleanupSession(sessionId: string): void { cleanupInMemorySession(sessionId); unpersistSession(sessionId); } /** * Reset all planning state. Used for testing only. */ export function __resetPlanningState(): void { // Cleanup all agent sessions for (const [id] of sessions) { cleanupInMemorySession(id); } sessions.clear(); rateLimits.clear(); planningStreamManager.reset(); activeGenerations.clear(); if (_aiSessionStore && _aiSessionDeletedListener) { _aiSessionStore.off("ai_session:deleted", _aiSessionDeletedListener); } _aiSessionDeletedListener = undefined; _aiSessionStore = undefined; planningNtfyHelpers = undefined; // Reset diagnostics sink to default resetDiagnosticsSink(); } /** * Inject a mock createFnAgent function. Used for testing only. */ export function __setCreateFnAgent(mock: typeof createFnAgent): void { createFnAgent = mock; } /** Inject ntfy helper implementations (test-only). */ export function __setPlanningNtfyHelpers(mock: PlanningNtfyHelpers | undefined): void { planningNtfyHelpers = mock; } // ── Custom Errors ─────────────────────────────────────────────────────────── export class RateLimitError extends Error { constructor(message: string) { super(message); this.name = "RateLimitError"; } } export class SessionNotFoundError extends Error { constructor(message: string) { super(message); this.name = "SessionNotFoundError"; } } export class InvalidSessionStateError extends Error { constructor(message: string) { super(message); this.name = "InvalidSessionStateError"; } }