/**
 * Stream Manager - Background Service for Chat Streams (Optimized)
 *
 * High-performance chat stream manager with:
 * - Event-driven push model (no polling)
 * - Background processing that continues even when browser is closed
 * - Reconnection to active streams after browser refresh
 * - Multiple concurrent streams per user/session
 */

import { EventEmitter } from 'events';
import { checkEngineSetup } from '../engine/engine-setup';
import type {
	EngineOutput,
	UnifiedMessage,
	UserMessage,
	AssistantMessage,
	ReasoningMessage,
	CompactBoundaryMessage,
	StreamEvent as UnifiedStreamEvent,
	TextDeltaEvent,
	StreamLifecycleEvent,
	SuccessResultEvent,
	ErrorResultEvent,
	SystemInitEvent,
	RateLimitEvent,
	RateLimitType,
	NotificationEvent,
	TokenUsage,
	StopReason,
	UserContentBlock,
	StreamRequest,
} from '$shared/types/unified';
import type { EngineType } from '$shared/types/unified';
import type { DatabaseMessage } from '$shared/types/database/schema';
import { initializeProjectEngine, type AIEngine } from '../engine';
import { messageQueries, projectQueries, sessionQueries } from '../database/queries';
import { snapshotService } from '../snapshot/snapshot-service';
import { snapshotQueries } from '../database/queries/snapshot-queries';
import { projectContextService, refreshExpiringExternalOAuth } from '../mcp';
import { resolveActiveProfileId } from '../profiles';
import { expandSlashInvocation } from '../skills';
import { browserMcpControl } from '../preview';
import { extractMessageText } from '../snapshot/helpers';
import { deferEpisodicIngest, ingestTurn } from '../memory/extract';
import { buildMemoryContext, withMemoryContext } from '../memory/context';
import { buildEngineHandoff, resolveBranchEngine, withHandoff } from './engine-handoff';
import { debug } from '$shared/utils/logger';
import { DEFAULT_MODEL_ID, DEFAULT_MODEL_NAME } from '$shared/constants/engines';

// ============================================================================
// Types
// ============================================================================

export interface StreamState {
	streamId: string;
	chatSessionId: string;
	projectId?: string;
	projectPath?: string;
	processId: string;
	engine: EngineType;
	accountId?: number;
	/**
	 * Server-trusted id of the requesting user (see StreamRequest). Carried
	 * through so terminal lifecycle events can fan out Web Push to the right
	 * devices without trusting any client-supplied field.
	 */
	requestedByUserId?: string;
	/** Reasoning/thinking level token for this run (native per engine; undefined = engine default). */
	reasoningEffort?: string;
	status: 'active' | 'completed' | 'error' | 'cancelled';
	startedAt: Date;
	completedAt?: Date;
	messages: unknown[];
	currentMessage?: UnifiedMessage;
	currentPartialText?: string;
	currentReasoningText?: string;
	error?: string;
	abortController?: AbortController;
	streamPromise?: Promise<void>;
	/**
	 * The engine instance actually running this stream.
	 *
	 * Pinned rather than looked up by (projectId, engineType), because the cache
	 * behind that lookup is now replaced whenever engine config changes. Looking
	 * it up again would hand back the REPLACEMENT: an instance with no query to
	 * cancel and no pending question to answer, so Stop would do nothing and an
	 * AskUserQuestion answer would go nowhere — and only for users unlucky enough
	 * to have changed a setting while a chat was running.
	 */
	engineInstance?: AIEngine;
	sdkSessionId?: string;
	preStreamSessionId?: string | null;
	hasCompactBoundary?: boolean;
	eventSeq: number;
}

// ============================================================================
// SDK → Unified Prompt Converter
// ============================================================================

/** Normalize raw UserMessage prompt from WS handler into a validated UserMessage */
function convertRawPromptToUserMessage(
	rawPrompt: any,
	senderId?: string,
	senderName?: string,
): UserMessage {
	const content: UserContentBlock[] = Array.isArray(rawPrompt?.content) && rawPrompt.content.length > 0
		? rawPrompt.content as UserContentBlock[]
		: [{ type: 'text', text: '' }];

	// Build engine object from raw prompt
	const rawEngine = rawPrompt?.engine;
	const engineObj = rawEngine && typeof rawEngine === 'object'
		? rawEngine
		: {
			type: rawEngine || 'claude-code',
			provider: '',
			model: {
				id: rawPrompt?.model?.id ?? rawPrompt?.modelId ?? '',
				name: rawPrompt?.model?.name ?? rawPrompt?.modelName ?? '',
			},
			account: { id: rawPrompt?.account?.id ?? 0, name: rawPrompt?.account?.name ?? '' },
		};

	return {
		type: 'user',
		createdAt: rawPrompt?.createdAt || new Date().toISOString(),
		messageId: rawPrompt?.messageId || rawPrompt?.id || crypto.randomUUID(),
		sessionId: null, // User messages from the frontend have no SDK session ID
		parent: {
			messageId: rawPrompt?.parent?.messageId ?? rawPrompt?.parentMessageId ?? null,
			sessionId: rawPrompt?.parent?.sessionId ?? rawPrompt?.parentSessionId ?? null,
			toolUseId: rawPrompt?.parent?.toolUseId ?? rawPrompt?.parentToolUseId ?? null,
		},
		engine: engineObj,
		sender: {
			id: senderId || rawPrompt?.sender?.id || '',
			name: senderName || rawPrompt?.sender?.name || '',
		},
		content,
		synthetic: rawPrompt?.synthetic ?? false,
	};
}

/** Request-level engine context injected into every SDK-emitted message */
interface RequestEngineContext {
	accountId: number;
	accountName: string;
	modelName: string;
	/** Reasoning/thinking level for this turn (native per engine; undefined = no knob). */
	reasoningEffort?: string;
}

/**
 * Enrich a message's engine block with request-level data that SDK adapters
 * cannot know (display-name for the selected account and the human-readable
 * model name). The adapter supplies type/provider/model.id; stream-manager
 * is the single source of truth for the rest.
 */
/** One file's hash pair inside a snapshot's `session_changes` map. */
interface SnapshotChange {
	oldHash?: string;
	newHash?: string;
}

/**
 * Parsed `session_changes`, which is stored as a JSON string. Returns an empty
 * map when the column is absent or unparseable — a snapshot with no recorded
 * changes simply gives the Memory Graph nothing to ingest.
 */
function parseSessionChanges(raw: string | null | undefined): Record<string, SnapshotChange> {
	if (!raw) return {};
	try {
		const parsed = JSON.parse(raw) as Record<string, SnapshotChange>;
		return parsed && typeof parsed === 'object' ? parsed : {};
	} catch {
		return {};
	}
}

/**
 * What actually changed on disk during THIS turn.
 *
 * `session_changes` is cumulative: it records every file that differs from the
 * SESSION's baseline, so it only ever grows as a session runs. Handing it to the
 * Memory Graph as "files changed this turn" was wrong in three ways at once — the
 * extraction prompt described a hundred files as belonging to one turn, the
 * sixty-file cap re-ingested whatever sorted first while genuinely new files fell
 * off the end, and every file in the session was re-upserted every turn, which
 * flattened the reinforcement signal it was supposed to produce.
 *
 * Differencing against the parent snapshot recovers the real delta: a path is in
 * it when its content hash differs from what the previous checkpoint recorded.
 */
function diffSnapshotAgainstParent(snapshot: {
	session_changes?: unknown;
	parent_snapshot_id?: string | null;
}): { changed: string[]; deleted: string[] } {
	const current = parseSessionChanges(snapshot?.session_changes as string | null | undefined);
	const parentId = snapshot?.parent_snapshot_id ?? null;
	const parent = parentId
		? parseSessionChanges(snapshotQueries.getById(parentId)?.session_changes as string | null | undefined)
		: {};

	const changed: string[] = [];
	const deleted: string[] = [];

	for (const [path, entry] of Object.entries(current)) {
		const before = parent[path];
		// Unchanged since the previous checkpoint — it belongs to an earlier turn.
		if (before && before.newHash === entry.newHash) continue;
		// An empty `newHash` is how the snapshot service records a deletion.
		if (!entry.newHash) deleted.push(path);
		else changed.push(path);
	}

	return { changed, deleted };
}

function enrichMessageEngine(
	message: UnifiedMessage,
	ctx: RequestEngineContext,
): UnifiedMessage {
	return {
		...message,
		engine: {
			...message.engine,
			model: {
				...message.engine.model,
				name: ctx.modelName || message.engine.model.name,
			},
			account: {
				id: ctx.accountId || message.engine.account.id,
				name: ctx.accountName || message.engine.account.name,
			},
			// Surface the turn's reasoning level (when the engine exposes one) so it's
			// visible in the Raw Message view. Omitted when there's no knob.
			...(ctx.reasoningEffort !== undefined && { reasoningEffort: ctx.reasoningEffort }),
		},
	};
}

// ============================================================================
// Stream Event Types
// ============================================================================

export type StreamEventType =
	| 'connection'
	| 'message'
	| 'partial'
	| 'notification'
	| 'rate_limit'
	| 'complete'
	| 'error'
	| 'cancelled';

export interface StreamEvent {
	type: StreamEventType;
	streamId: string;
	processId: string;
	data: any;
	timestamp: string;
	seq: number; // Sequence number for deduplication
}

// ============================================================================
// Stream Manager
// ============================================================================

export interface RateLimitSnapshot {
	engine: EngineType;
	accountId: number;
	status: 'allowed_warning' | 'rejected';
	utilization: number;
	resetsAt: number | null;
	rateLimitType: RateLimitType | null;
	receivedAt: string;
}

class StreamManager extends EventEmitter {
	private activeStreams = new Map<string, StreamState>();
	private sessionStreams = new Map<string, string>(); // composite key -> streamId
	/** Guard against duplicate lifecycle events (e.g. if both inner and outer error paths fire) */
	private lifecycleEmitted = new Set<string>();
	/** Latest rate-limit snapshot per `${engine}:${accountId}`. Persists across refresh so the UI banner can rehydrate on join. */
	private activeRateLimits = new Map<string, RateLimitSnapshot>();

	private getAccountKey(engine: EngineType, accountId: number): string {
		return `${engine}:${accountId}`;
	}

	/** Return active rate limit for an account, purging it lazily if `resetsAt` has passed. */
	getRateLimitForAccount(engine: EngineType, accountId: number): RateLimitSnapshot | null {
		const key = this.getAccountKey(engine, accountId);
		const snap = this.activeRateLimits.get(key);
		if (!snap) return null;
		if (snap.resetsAt && snap.resetsAt * 1000 < Date.now()) {
			this.activeRateLimits.delete(key);
			return null;
		}
		return snap;
	}

	/** Return every active rate-limit snapshot, lazily purging expired entries. */
	getAllActiveRateLimits(): RateLimitSnapshot[] {
		const now = Date.now();
		const active: RateLimitSnapshot[] = [];
		for (const [key, snap] of this.activeRateLimits) {
			if (snap.resetsAt && snap.resetsAt * 1000 < now) {
				this.activeRateLimits.delete(key);
				continue;
			}
			active.push(snap);
		}
		return active;
	}

	/** Clear an active rate limit (called when the user dismisses the banner). */
	dismissRateLimit(engine: EngineType, accountId: number): void {
		this.activeRateLimits.delete(this.getAccountKey(engine, accountId));
	}

	constructor() {
		super();
		// Increase max listeners for high concurrency
		this.setMaxListeners(1000);

		// MCP browser locks are scoped to a chat session, and every release path
		// runs from this class. Teaching the lock manager how to check liveness
		// closes the gap those paths can't: a tool call that lands after its
		// stream ended used to be able to take a lock nothing would release.
		browserMcpControl.setSessionLivenessProbe((chatSessionId) =>
			this.isChatSessionStreaming(chatSessionId)
		);
	}

	/** Whether any stream for this chat session is still running. */
	private isChatSessionStreaming(chatSessionId: string): boolean {
		for (const stream of this.activeStreams.values()) {
			if (stream.chatSessionId === chatSessionId && stream.status === 'active') return true;
		}
		return false;
	}

	/**
	 * Emit a global lifecycle event when a stream reaches a terminal state.
	 * This event fires regardless of per-connection subscribers.
	 * Used by the WS layer to send cross-project notifications (presence, sound, push).
	 */
	private emitStreamLifecycle(streamState: StreamState, status: 'completed' | 'error' | 'cancelled', reason?: string): void {
		if (this.lifecycleEmitted.has(streamState.streamId)) return;
		this.lifecycleEmitted.add(streamState.streamId);

		this.emit('stream:lifecycle', {
			status,
			streamId: streamState.streamId,
			projectId: streamState.projectId,
			chatSessionId: streamState.chatSessionId,
			requestedByUserId: streamState.requestedByUserId,
			timestamp: (streamState.completedAt || new Date()).toISOString(),
			reason
		});

		// Clean up guard after 60s (no need to keep forever)
		setTimeout(() => this.lifecycleEmitted.delete(streamState.streamId), 60000);
	}

	/**
	 * Start a new background stream
	 */
	async startStream(request: StreamRequest): Promise<string> {
		const streamId = crypto.randomUUID();
		const processId = crypto.randomUUID();

		// Check if there's already an active stream for this chat session + project
		const sessionKey = this.getSessionKey(request.projectId, request.chatSessionId);
		const existingStreamId = this.sessionStreams.get(sessionKey);
		if (existingStreamId) {
			const existingStream = this.activeStreams.get(existingStreamId);
			if (existingStream && existingStream.status === 'active') {
				if (existingStream.projectId === request.projectId) {
					if (request.engine.type === 'claude-code') {
						// Claude Code: cancel existing stream to prevent message loss from race condition.
						// Claude Code SDK only returns session_id inside yielded messages, so a cancelled
						// stream may never have established a valid session — safe to cancel and restart.
						debug.log('chat', `Cancelling existing active stream ${existingStreamId} before starting new one`);
						await this.cancelStream(existingStreamId);
					} else {
						// Other engines (OpenCode): return existing stream ID (original behavior).
						// OpenCode creates sessions synchronously, so the existing stream is valid.
						return existingStreamId;
					}
				}
			}
		}

		// Initialize stream state
		const streamState: StreamState = {
			streamId,
			chatSessionId: request.chatSessionId,
			projectId: request.projectId,
			projectPath: request.projectPath,
			processId,
			engine: request.engine.type,
			accountId: request.engine.account?.id || undefined,
			requestedByUserId: request.requestedByUserId,
			reasoningEffort: request.reasoningEffort ?? undefined,
			status: 'active',
			startedAt: new Date(),
			messages: [],
			abortController: new AbortController(),
			eventSeq: 0 // Initialize sequence for deduplication
		};

		this.activeStreams.set(streamId, streamState);
		this.sessionStreams.set(sessionKey, streamId);

		// Save engine+model+account to session for persistence across refresh/switch
		if (request.chatSessionId) {
			try {
				sessionQueries.updateEngineModel(
					request.chatSessionId,
					request.engine.type,
					request.engine.provider,
					request.engine.model.id || DEFAULT_MODEL_ID,
					request.engine.model.name || DEFAULT_MODEL_NAME
				);
				if (request.engine.account.id) {
					sessionQueries.updateAccount(request.chatSessionId, request.engine.account.id, request.engine.account.name || null);
				}
				// Persist the session's explicit profile choice (number or null).
				// `undefined` (older client that doesn't send one) leaves it untouched
				// so the session keeps inheriting the project default.
				if (request.profileId !== undefined) {
					sessionQueries.updateProfile(request.chatSessionId, request.profileId);
				}
				// Persist the reasoning/thinking level (string or null). `undefined`
				// (client didn't send one) leaves the stored value untouched.
				if (request.reasoningEffort !== undefined) {
					sessionQueries.updateReasoning(request.chatSessionId, request.reasoningEffort);
				}
			} catch (error) {
				debug.error('chat', 'Failed to save engine/model to session:', error);
			}
		}

		// Register session -> projectId mapping for MCP context
		if (request.projectId) {
			projectContextService.registerSession(request.chatSessionId, request.projectId);
			// The engine goes with it: the remote HTTP MCP bridge cannot always
			// name the calling stream, and knowing which engine asked is what
			// keeps a fallback from reaching across into another engine's project.
			projectContextService.registerStream(streamId, request.projectId, request.chatSessionId, request.engine.type);
		}
		// Hand the AbortSignal to MCP so tool handlers can fast-fail on
		// cancellation. Without this, the engine subprocess dies on cancel
		// but the in-flight HTTP-MCP tool keeps issuing puppeteer ops —
		// surfacing as "preview keeps moving by itself" after interrupt.
		if (streamState.abortController) {
			projectContextService.registerStreamSignal(streamId, streamState.abortController.signal);
		}

		// Emit connection event immediately
		this.emitStreamEvent(streamState, 'connection', {
			processId,
			timestamp: streamState.startedAt.toISOString()
		});

		// Start background processing
		streamState.streamPromise = this.processStream(streamState, request).catch(error => {
			streamState.status = 'error';
			streamState.error = this.extractErrorDetail(error);
			streamState.completedAt = new Date();

			this.emitStreamEvent(streamState, 'error', {
				processId: streamState.processId,
				error: streamState.error,
				timestamp: streamState.completedAt.toISOString()
			});

			this.emitStreamLifecycle(streamState, 'error');
		});

		// Auto-cleanup after 5 minutes (fallback for completed/error streams)
		setTimeout(() => {
			if (streamState.status !== 'active') {
				this.cleanupStream(streamId);
			}
		}, 5 * 60 * 1000);

		return streamId;
	}

	/**
	 * Emit a stream event to all subscribers
	 */
	private emitStreamEvent(streamState: StreamState, type: StreamEventType, data: any): void {
		// Increment sequence number for deduplication
		streamState.eventSeq++;

		// Attach engine type string to event data for frontend routing metadata
		// (stream-level metadata, distinct from message.engine which is the full MessageEngine object)
		if (data && typeof data === 'object') {
			data.engine = streamState.engine;
		}

		const event: StreamEvent = {
			type,
			streamId: streamState.streamId,
			processId: streamState.processId,
			data,
			timestamp: new Date().toISOString(),
			seq: streamState.eventSeq
		};

		// Emit to stream-specific channel only
		// (session channel removed to prevent duplicate dispatches)
		this.emit(`stream:${streamState.streamId}`, event);
	}

	/**
	 * Subscribe to a stream's events
	 */
	subscribeToStream(streamId: string, handler: (event: StreamEvent) => void): () => void {
		const eventName = `stream:${streamId}`;
		this.on(eventName, handler);

		// Return unsubscribe function
		return () => {
			this.off(eventName, handler);
		};
	}

	/**
	 * Subscribe to a session's events (for reconnection)
	 */
	subscribeToSession(projectId: string | undefined, chatSessionId: string, handler: (event: StreamEvent) => void): () => void {
		const sessionKey = this.getSessionKey(projectId, chatSessionId);
		const eventName = `session:${sessionKey}`;
		this.on(eventName, handler);

		return () => {
			this.off(eventName, handler);
		};
	}

	/**
	 * Process stream in background — routes EngineOutput events by type discriminant
	 */
	private async processStream(streamState: StreamState, requestData: StreamRequest): Promise<void> {
		let userMessageId: string | undefined;

		// Launch the pre-edit baseline scan as early as possible so it overlaps with
		// engine initialization. It MUST finish before the engine can write any files
		// (awaited just before streamQuery below) — otherwise early writes would be
		// folded into the baseline and silently dropped from this turn's checkpoint.
		let baselineInitPromise: Promise<void> | null = null;
		if (requestData.projectPath && requestData.chatSessionId) {
			baselineInitPromise = snapshotService.initializeSessionBaseline(
				requestData.projectPath,
				requestData.chatSessionId
			).catch(err => debug.error('snapshot', 'Failed to initialize session baseline:', err));
		}

		try {
			const { projectPath, prompt: rawPrompt, chatSessionId, engine: requestEngine, sender: requestSender } = requestData;

			// Engine context that stream-manager injects into every SDK-emitted message.
			// (SDK adapters cannot know these values; only the request layer can.)
			const engineCtx: RequestEngineContext = {
				accountId: requestEngine.account.id,
				accountName: requestEngine.account.name,
				modelName: requestEngine.model.name,
				reasoningEffort: streamState.reasoningEffort,
			};

			const projectPathExists = projectPath ? await this.existsSync(projectPath) : false;
			if (!projectPath) throw new Error('Project path is required. Please select a valid project directory.');
			if (!projectPathExists) throw new Error(`Project path does not exist: ${projectPath}. Please select a valid project directory.`);

			// Which engine produced the trailing part of this branch. When it differs
			// from the engine the user just picked, the branch's SDK session id
			// belongs to a foreign store and MUST NOT be used as a resume target —
			// the conversation is carried over as prompt content instead (see
			// buildEngineHandoff below). Null means a fresh branch: nothing to hand
			// over, nothing to resume.
			const branchEngine = chatSessionId ? resolveBranchEngine(chatSessionId) : null;
			const engineSwitched = branchEngine !== null && branchEngine !== requestEngine.type;

			// Get resume session ID (branch-aware).
			// Primary: use parentSessionId carried by the UserMessage — set by the
			// frontend from the last assistant/reasoning sessionId in the current branch.
			// Fallback: walk the HEAD chain in the DB (handles messages sent from
			// older clients or tool-result user messages that lack parentSessionId).
			let resumeSessionId: string | undefined = undefined;
			if (chatSessionId && !engineSwitched) {
				// Primary source: parent.sessionId on the raw prompt.
				// Note this is client-supplied and engine-blind, which is why the
				// engineSwitched guard above is authoritative — an older client will
				// happily send the previous engine's id.
				const promptParentSessionId = rawPrompt?.parent?.sessionId;
				if (promptParentSessionId && promptParentSessionId !== chatSessionId) {
					resumeSessionId = promptParentSessionId;
				} else {
					// Fallback: walk HEAD chain
					try {
						const head = sessionQueries.getHead(chatSessionId);
						if (head) {
							const chain = messageQueries.getPathToRoot(head);
							for (let i = chain.length - 1; i >= 0; i--) {
								try {
									const msg = JSON.parse(chain[i].data) as UnifiedMessage;
									if (msg.type === 'user') continue;
									if (msg.sessionId && msg.sessionId !== chatSessionId) {
										resumeSessionId = msg.sessionId;
										break;
									}
								} catch { /* skip unparseable blobs */ }
							}
						}
					} catch (error) {
						debug.error('chat', 'Failed to get resume session ID from HEAD chain:', error);
					}
				}
			}
			streamState.preStreamSessionId = resumeSessionId ?? null;

			// Convert raw SDK prompt → UserMessage (unified)
			const userMessage = convertRawPromptToUserMessage(rawPrompt, requestSender.id, requestSender.name);

			// Save user message to DB
			const userMessageTimestamp = new Date().toISOString();
			const savedMessage = await this.saveMessage(
				userMessage,
				chatSessionId,
				userMessageTimestamp
			);
			userMessageId = savedMessage?.id;

			streamState.messages.push({
				processId: streamState.processId,
				message: userMessage as any,
				timestamp: userMessageTimestamp,
				message_id: savedMessage?.id,
				parent_message_id: savedMessage?.parent_message_id || null,
				sender_id: requestSender.id,
				sender_name: requestSender.name
			});

			this.emitStreamEvent(streamState, 'message', {
				processId: streamState.processId,
				message: userMessage,
				timestamp: userMessageTimestamp,
				message_id: savedMessage?.id,
				parent_message_id: savedMessage?.parent_message_id || null,
				sender_id: requestSender.id,
				sender_name: requestSender.name
			});

			if ((streamState.status as string) === 'cancelled' || streamState.abortController?.signal.aborted) {
				debug.log('chat', 'Stream cancelled after saving user message, skipping query');
				return;
			}

			let sdkSessionId: string | undefined;
			let lastAssistantTextContent: string | null = null;
			const projectId = streamState.projectId || 'default';

			// Resolve the effective Profile once per stream: the session's explicit
			// choice, else the project's shared default (see backend/profiles). Passed
			// down via mcpContext so the existing per-engine sync + MCP config path
			// scopes artifacts/connectors to the profile's bundle — no new sync path.
			const activeProfileId = resolveActiveProfileId(requestData.profileId, streamState.projectId);

			if ((streamState.status as string) === 'cancelled' || streamState.abortController?.signal.aborted) {
				debug.log('chat', 'Stream cancelled before engine initialization, skipping query');
				return;
			}

			// Push back any memory summary parked for this session. Extraction talks to
			// the same engine this stream is about to use, and on a shared-process
			// engine that request would queue against the user's reply.
			if (chatSessionId) deferEpisodicIngest(chatSessionId);

			const engine = await initializeProjectEngine(projectId, requestEngine.type);
			streamState.engineInstance = engine;

			if ((streamState.status as string) === 'cancelled' || streamState.abortController?.signal.aborted) {
				debug.log('chat', 'Stream cancelled during engine initialization, skipping query');
				return;
			}

			// ── Cross-engine handoff ──
			// The user switched engine mid-conversation, so the new engine has no
			// native session to resume. Replay the branch as prompt content. This
			// is prepended to the ENGINE prompt only — `userMessage` (already saved
			// above) stays clean, so the transcript never reaches the timeline.
			let enginePrompt = userMessage;

			// ── Slash skill expansion ──
			// `/review-pr 123` is resolved HERE, not by the engine. Four of the eight
			// engines have no native command directory at all, so the old per-engine
			// materialization delivered nothing but a name to them; expanding once,
			// centrally, is what makes a slash skill behave the same everywhere and
			// lets `uses:` pull in several skills deterministically. The SAVED user
			// message keeps the `/slug` the user typed — only the engine prompt is
			// rewritten, exactly like the handoff and memory blocks below.
			try {
				const firstText = enginePrompt.content.find(block => block.type === 'text');
				if (firstText?.type === 'text') {
					const expanded = await expandSlashInvocation(firstText.text, { profileId: activeProfileId });
					if (expanded) {
						enginePrompt = {
							...enginePrompt,
							content: enginePrompt.content.map(block =>
								block === firstText ? { ...block, text: expanded.text } : block
							)
						};
						debug.log(
							'chat',
							`Expanded /${expanded.slug} into the engine prompt` +
							(expanded.used.length ? ` (requires: ${expanded.used.join(', ')})` : '')
						);
					}
				}
			} catch (error) {
				// A failed expansion must not block the turn — the engine simply
				// receives the message exactly as the user typed it.
				debug.warn('chat', 'Slash skill expansion failed, sending the message as typed:', error);
			}

			if (engineSwitched && chatSessionId) {
				try {
					const handoff = buildEngineHandoff(
						chatSessionId,
						requestEngine.type,
						requestEngine.model.id || '',
						branchEngine,
						userMessageId
					);
					if (handoff) {
						enginePrompt = withHandoff(userMessage, handoff.blocks);
						const { turns, clearedToolResults, droppedAttachments, estimatedTokens } = handoff.stats;
						debug.log(
							'chat',
							`Engine handoff ${branchEngine} → ${requestEngine.type}: ${turns} turn(s), ~${estimatedTokens} tokens` +
							(clearedToolResults ? `, ${clearedToolResults} tool result(s) cleared` : '') +
							(droppedAttachments ? `, ${droppedAttachments} attachment(s) omitted (unsupported by target)` : '')
						);
					}
				} catch (error) {
					// A failed handoff must not block the turn — the engine simply
					// starts without carried context.
					debug.error('chat', 'Engine handoff failed, continuing without carried context:', error);
				}
			}

			// ── Memory Graph injection ──
			// Relevant memories from earlier sessions, retrieved from the user's own
			// words. Done here rather than per adapter so every engine behaves the
			// same, and prepended to the ENGINE prompt only — like the handoff above,
			// `userMessage` stays clean so this never reaches the timeline.
			try {
				// The files this session has been working in are seeds too. A turn whose
				// text carries no retrievable signal — "continue", "fix it" — still has a
				// working set, and that is often the only thing pointing at the memory
				// that matters.
				const anchorPaths = chatSessionId
					? Object.keys(
							parseSessionChanges(
								snapshotQueries.getLatestBySessionId(chatSessionId)?.session_changes as string | undefined
							)
						).slice(-40)
					: [];

				const memory = buildMemoryContext({
					query: extractMessageText(userMessage),
					projectId: projectId ?? null,
					sessionId: chatSessionId ?? null,
					anchorPaths,
					// So a memory learned in another repository can say which one. Without
					// the name, a travelling insight reads as a claim about the project in
					// front of the agent — which is worse than not surfacing it at all.
					projectNames: new Map(projectQueries.getAll().map(project => [project.id, project.name]))
				});
				if (memory) {
					enginePrompt = withMemoryContext(enginePrompt, memory.text);
					debug.log(
						'memory',
						`Injected ${memory.text.length} chars of memory context (${memory.nodeIds.length} memories):\n${memory.text}`
					);
				}
			} catch (error) {
				debug.warn('memory', 'Memory injection failed, continuing without it:', error);
			}

			// Detect orphaned user messages and prepend context (claude-code only).
			// Same idea as the handoff above but for the in-engine case: messages the
			// engine's own session never saw because an earlier turn was cancelled.
			if (!engineSwitched && requestEngine.type === 'claude-code' && chatSessionId) {
				try {
					const head = sessionQueries.getHead(chatSessionId);
					if (head) {
						const chain = messageQueries.getPathToRoot(head);
						const previousChain = chain.slice(0, -1);

						if (previousChain.length > 0) {
							let boundaryIndex = -1;
							if (resumeSessionId) {
								for (let i = previousChain.length - 1; i >= 0; i--) {
									try {
										const msg = JSON.parse(previousChain[i].data) as UnifiedMessage;
										if (msg.sessionId === resumeSessionId) {
											boundaryIndex = i;
											break;
										}
									} catch { /* skip unparseable */ }
								}
							}

							const orphanedUserTexts: string[] = [];
							for (let i = boundaryIndex + 1; i < previousChain.length; i++) {
								try {
									const msg = JSON.parse(previousChain[i].data) as UnifiedMessage;
									if (msg.type === 'user') {
											const text = msg.content
												.filter(block => block.type === 'text')
												.map(block => block.text)
												.join('\n');
											if (text.trim()) orphanedUserTexts.push(text.trim());
									}
								} catch { /* skip unparseable */ }
							}

							if (orphanedUserTexts.length > 0) {
								debug.log('chat', `Prepending ${orphanedUserTexts.length} orphaned user message(s) as context`);
								const contextPrefix = [
									'[Previous unprocessed messages from the user:]',
									...orphanedUserTexts.map((text, i) => `${i + 1}. "${text}"`),
									'',
									'[Current message:]'
								].join('\n');

								enginePrompt = {
									...userMessage,
									content: [
										{ type: 'text' as const, text: contextPrefix },
										...userMessage.content,
									],
								};
							}
						}
					}
				} catch (error) {
					debug.error('chat', 'Failed to detect orphaned messages:', error);
				}
			}

			// Guarantee the pre-edit baseline is captured before the engine can write
			// any files (see processStream top). Without this await a fast engine on a
			// large repo could begin editing while the baseline scan is still running.
			if (baselineInitPromise) await baselineInitPromise;

			// Refresh any near-expiry managed MCP OAuth tokens so the engine's
			// (synchronous) MCP config builders below inject a still-valid bearer.
			await refreshExpiringExternalOAuth().catch(error =>
				debug.warn('chat', 'MCP OAuth refresh failed:', error)
			);

			// Refuse to stream an engine that isn't ready and point the user at the
			// exact fix (Stack to install/update, Engines to sign in). Thrown here
			// so the processStream catch surfaces it as a chat error, which the
			// ErrorMessage renderer turns into a one-click action.
			const setupIssue = await checkEngineSetup(requestEngine.type, requestEngine.account.id);
			if (setupIssue) throw new Error(setupIssue.message);

			// Stream EngineOutput events through the engine adapter
			const streamIterable = engine.streamQuery({
				projectPath,
				prompt: enginePrompt,
				resume: resumeSessionId,
				providerSlug: requestEngine.provider,
				modelId: requestEngine.model.id,
				...(streamState.reasoningEffort && { reasoningEffort: streamState.reasoningEffort }),
				includePartialMessages: true,
				abortController: streamState.abortController,
				...(requestEngine.account.id !== 0 && { accountId: requestEngine.account.id }),
				...(projectId && chatSessionId && {
					mcpContext: { projectId, chatSessionId, streamId: streamState.streamId, ...(activeProfileId != null && { profileId: activeProfileId }) }
				}),
			});

			await projectContextService.runWithContextAsync(
				{ chatSessionId, projectId, streamId: streamState.streamId },
				async () => { for await (const output of streamIterable) {
				if ((streamState.status as string) === 'cancelled' || streamState.abortController?.signal.aborted) {
					break;
				}

				// ── Route by type discriminant ──────────────────────────────

				// Extract session ID from first event that has one
				if (!sdkSessionId) {
					const eventSessionId = 'sessionId' in output ? (output as any).sessionId : null;
					if (eventSessionId) {
						sdkSessionId = eventSessionId;
						streamState.sdkSessionId = sdkSessionId;
						if (chatSessionId) {
							try { sessionQueries.updateSessionId(chatSessionId, sdkSessionId!); }
							catch { /* ignore */ }
						}
					}
				}

				switch (output.type) {
					// ── System Init ────────────────────────────────────────
					case 'system_init': {
						const initEvent = output as SystemInitEvent;
						// Suppress the toast for "soft" statuses: `pending` (still
						// connecting) and `needs-auth` (server expects OAuth/credentials
						// configured in Settings → MCP). These are not errors — a scary
						// toast for them is confusing. (Status is the raw SDK string,
						// wider than the EngineOutput union, hence the cast.)
						const SOFT_STATUSES = new Set(['pending', 'needs-auth']);
						const failedServers = initEvent.mcpServers.filter(
							s => s.status !== 'connected' && !SOFT_STATUSES.has(s.status as string)
						);
						initEvent.mcpServers
							.filter(s => SOFT_STATUSES.has(s.status as string))
							.forEach(s => debug.log('mcp', `MCP server "${s.name}" not ready (${s.status}) — suppressing toast`));
						failedServers.forEach(server => {
							debug.warn('mcp', `MCP server connection failed: ${server.name} (${server.status})`);
							this.emitStreamEvent(streamState, 'notification', {
								notification: {
									type: 'warning',
									title: 'MCP Server Connection Failed',
									message: `Failed to connect to MCP server "${server.name}".`,
								},
								timestamp: new Date().toISOString()
							});
						});
						const connected = initEvent.mcpServers.filter(s => s.status === 'connected');
						if (connected.length > 0) {
							debug.log('mcp', `✓ Connected MCP servers: ${connected.map(s => s.name).join(', ')}`);
						}
						continue; // Don't save to DB
					}

					// ── Compact Boundary ───────────────────────────────────
					case 'compact_boundary': {
						const boundary = enrichMessageEngine(output, engineCtx) as CompactBoundaryMessage;
						streamState.hasCompactBoundary = true;
						const compactTimestamp = boundary.createdAt;

						let savedCompactId: string | undefined;
						let savedCompactParentId: string | null = null;
						if (chatSessionId) {
							const saved = await this.saveMessage(
								boundary,
								chatSessionId,
								compactTimestamp
							);
							savedCompactId = saved?.id;
							savedCompactParentId = saved?.parent_message_id || null;
						}

						streamState.messages.push({
							processId: streamState.processId,
							message: boundary as any,
							timestamp: compactTimestamp,
							message_id: savedCompactId,
							parent_message_id: savedCompactParentId,
							compactBoundary: {
								trigger: boundary.trigger,
								preTokens: boundary.preTokens
							}
						});

						this.emitStreamEvent(streamState, 'message', {
							processId: streamState.processId,
							message: boundary,
							timestamp: compactTimestamp,
							message_id: savedCompactId,
							parent_message_id: savedCompactParentId,
							sender_id: requestSender.id,
							sender_name: requestSender.name
						});
						continue;
					}

					// ── Rate Limit ─────────────────────────────────────────
					case 'rate_limit': {
						const rl = output as RateLimitEvent;
						const accountId = streamState.accountId;
						const nowIso = new Date().toISOString();
						if (accountId) {
							this.activeRateLimits.set(
								this.getAccountKey(streamState.engine, accountId),
								{
									engine: streamState.engine,
									accountId,
									status: rl.status,
									utilization: rl.utilization,
									resetsAt: rl.resetsAt,
									rateLimitType: rl.rateLimitType,
									receivedAt: nowIso
								}
							);
						}
						this.emitStreamEvent(streamState, 'rate_limit', {
							chatSessionId: streamState.chatSessionId,
							engine: streamState.engine,
							accountId: accountId ?? 0,
							status: rl.status,
							utilization: rl.utilization,
							resetsAt: rl.resetsAt,
							rateLimitType: rl.rateLimitType,
							timestamp: nowIso
						});
						continue;
					}

					// ── Notification (e.g. background-task completion) ─────
					case 'notification': {
						const note = output as NotificationEvent;
						this.emitStreamEvent(streamState, 'notification', {
							notification: {
								type: note.level,
								title: note.title,
								message: note.message,
							},
							timestamp: new Date().toISOString()
						});
						continue; // transient — don't save to DB
					}

					// ── Result ─────────────────────────────────────────────
					case 'result': {
						if (output.subtype === 'success') {
							const successResult = output as SuccessResultEvent;
							// Backfill stopReason to last assistant message if SDK left it null
							if (successResult.stopReason && chatSessionId) {
								const mapped = this.backfillStopReason(chatSessionId, successResult.stopReason);
								// Re-emit patched assistant message so frontend updates live data
								if (mapped && streamState.currentMessage?.type === 'assistant' && !streamState.currentMessage.stopReason) {
									const patched = { ...streamState.currentMessage, stopReason: mapped } as AssistantMessage;
									streamState.currentMessage = patched;
									// Find the saved message ID from the last emitted assistant
									const lastEntry = [...streamState.messages].reverse()
										.find((m: any) => m.message?.type === 'assistant') as any;
									this.emitStreamEvent(streamState, 'message', {
										processId: streamState.processId,
										message: patched,
										timestamp: patched.createdAt,
										message_id: lastEntry?.message_id ?? undefined,
										parent_message_id: lastEntry?.parent_message_id ?? null,
										sender_id: requestSender.id,
										sender_name: requestSender.name
									});
								}
							}
							// Codex and Cursor emit usage only once per turn (after all
							// items streamed live), so saved assistant rows have
							// usage:null. Backfill the turn's aggregate to every saved
							// assistant without usage so it survives a refresh.
							if (successResult.usage && (streamState.engine === 'codex' || streamState.engine === 'cursor')) {
								this.backfillUsageForStream(streamState, successResult.usage, requestSender);
							}
						} else {
							const errResult = output as ErrorResultEvent;
							if (errResult.errors?.length) {
								debug.warn('chat', `SDK result error: ${errResult.subtype}`, errResult.errors);
							}
						}
						continue; // Don't save to DB
					}

					// ── Stream Events (deltas, start/stop) ────────────────
					case 'stream_event': {
						const streamEvt = output as UnifiedStreamEvent;

						if (streamEvt.event === 'start') {
							const lifecycle = streamEvt as StreamLifecycleEvent;
							if (lifecycle.reasoning) {
								streamState.currentReasoningText = '';
							} else {
								streamState.currentPartialText = '';
							}
							this.emitStreamEvent(streamState, 'partial', {
								processId: streamState.processId,
								eventType: 'start',
								partialText: '',
								deltaText: '',
								...(lifecycle.reasoning && { reasoning: true }),
								timestamp: new Date().toISOString()
							});
						} else if (streamEvt.event === 'delta') {
							const delta = streamEvt as TextDeltaEvent;
							if (delta.reasoning) {
								streamState.currentReasoningText = (streamState.currentReasoningText || '') + delta.text;
								this.emitStreamEvent(streamState, 'partial', {
									processId: streamState.processId,
									eventType: 'update',
									partialText: streamState.currentReasoningText,
									deltaText: delta.text,
									reasoning: true,
									timestamp: new Date().toISOString()
								});
							} else {
								streamState.currentPartialText = (streamState.currentPartialText || '') + delta.text;
								this.emitStreamEvent(streamState, 'partial', {
									processId: streamState.processId,
									eventType: 'update',
									partialText: streamState.currentPartialText,
									deltaText: delta.text,
									timestamp: new Date().toISOString()
								});
							}
						} else if (streamEvt.event === 'stop') {
							const lifecycle = streamEvt as StreamLifecycleEvent;
							if (lifecycle.reasoning && streamState.currentReasoningText) {
								this.emitStreamEvent(streamState, 'partial', {
									processId: streamState.processId,
									eventType: 'end',
									partialText: streamState.currentReasoningText,
									deltaText: '',
									reasoning: true,
									timestamp: new Date().toISOString()
								});
								streamState.currentReasoningText = '';
							} else {
								this.emitStreamEvent(streamState, 'partial', {
									processId: streamState.processId,
									eventType: 'end',
									partialText: streamState.currentPartialText || '',
									deltaText: '',
									timestamp: new Date().toISOString()
								});
							}
						}
						continue;
					}

					// ── Reasoning Message ──────────────────────────────────
					case 'reasoning': {
						const reasoning = enrichMessageEngine(output, engineCtx) as ReasoningMessage;
						streamState.currentReasoningText = undefined;

						const reasoningTimestamp = reasoning.createdAt;
						let savedReasoningId: string | undefined;
						let savedReasoningParentId: string | null = null;
						if (chatSessionId) {
							const saved = await this.saveMessage(
								reasoning,
								chatSessionId,
								reasoningTimestamp
							);
							savedReasoningId = saved?.id;
							savedReasoningParentId = saved?.parent_message_id || null;
						}

						this.emitStreamEvent(streamState, 'message', {
							processId: streamState.processId,
							message: reasoning,
							timestamp: reasoningTimestamp,
							message_id: savedReasoningId,
							parent_message_id: savedReasoningParentId,
							sender_id: requestSender.id,
							sender_name: requestSender.name
						});
						continue;
					}

					// ── Assistant Message ──────────────────────────────────
					case 'assistant': {
						const assistantMsg = enrichMessageEngine(output, engineCtx) as AssistantMessage;

						// Deduplicate consecutive identical assistant messages
						const currentText = assistantMsg.content
							.filter(c => c.type === 'text')
							.map(c => (c as any).text as string)
							.join('');
						if (currentText && currentText === lastAssistantTextContent) {
							debug.warn('chat', 'Skipping duplicate consecutive assistant message');
							continue;
						}
						if (currentText) lastAssistantTextContent = currentText;

						const usage = assistantMsg.usage;
						const messageTimestamp = assistantMsg.createdAt;

						let savedMsgId: string | undefined;
						let savedParentId: string | null = null;
						if (chatSessionId) {
							const saved = await this.saveMessage(
								assistantMsg,
								chatSessionId,
								messageTimestamp
							);
							savedMsgId = saved?.id;
							savedParentId = saved?.parent_message_id || null;
						}

						streamState.currentPartialText = undefined;

						streamState.messages.push({
							processId: streamState.processId,
							message: assistantMsg as any,
							usage: usage as any,
							timestamp: messageTimestamp,
							message_id: savedMsgId,
							parent_message_id: savedParentId,
							sender_id: requestSender.id,
							sender_name: requestSender.name
						});

						streamState.currentMessage = assistantMsg;

						this.emitStreamEvent(streamState, 'message', {
							processId: streamState.processId,
							message: assistantMsg,
							usage,
							timestamp: messageTimestamp,
							message_id: savedMsgId,
							parent_message_id: savedParentId,
							sender_id: requestSender.id,
							sender_name: requestSender.name
						});
						continue;
					}

					// ── User Message (tool results from SDK) ───────────────
					case 'user': {
						const userMsg = enrichMessageEngine(output, engineCtx) as UserMessage;
						lastAssistantTextContent = null; // Reset dedup tracker

						const messageTimestamp = userMsg.createdAt;
						let savedMsgId: string | undefined;
						let savedParentId: string | null = null;
						if (chatSessionId) {
							const saved = await this.saveMessage(
								userMsg,
								chatSessionId,
								messageTimestamp
							);
							savedMsgId = saved?.id;
							savedParentId = saved?.parent_message_id || null;
						}

						streamState.messages.push({
							processId: streamState.processId,
							message: userMsg as any,
							timestamp: messageTimestamp,
							message_id: savedMsgId,
							parent_message_id: savedParentId,
							sender_id: requestSender.id,
							sender_name: requestSender.name
						});

						streamState.currentMessage = userMsg;

						this.emitStreamEvent(streamState, 'message', {
							processId: streamState.processId,
							message: userMsg,
							timestamp: messageTimestamp,
							message_id: savedMsgId,
							parent_message_id: savedParentId,
							sender_id: requestSender.id,
							sender_name: requestSender.name
						});
						continue;
					}

					default:
						debug.log('chat', `[SM] Skipping unknown EngineOutput type: ${(output as any).type}`);
						continue;
				}
			} }); // end runWithContextAsync + for await

			if (streamState.status === 'active') {
				streamState.status = 'completed';
				streamState.completedAt = new Date();

				this.emitStreamEvent(streamState, 'complete', {
					processId: streamState.processId,
					timestamp: streamState.completedAt.toISOString()
				});
				this.emitStreamLifecycle(streamState, 'completed');
			}

			if (chatSessionId) {
				browserMcpControl.releaseSession(chatSessionId);
				debug.log('mcp', `✅ Auto-released MCP tabs for session ${chatSessionId.slice(0, 8)} on stream completion`);
			}

		} catch (error) {
			if (streamState.status !== 'cancelled') {
				streamState.status = 'error';
				streamState.error = this.extractErrorDetail(error);
				streamState.completedAt = new Date();

				const errorTimestamp = streamState.completedAt.toISOString();

				// Build a synthetic error assistant message (unified type)
				const errorAssistantMsg: AssistantMessage = {
					type: 'assistant',
					createdAt: errorTimestamp,
					messageId: crypto.randomUUID(),
					sessionId: streamState.sdkSessionId || null,
					parent: { messageId: null, sessionId: null, toolUseId: null },
					engine: { type: streamState.engine, provider: '', model: { id: '', name: '' }, account: { id: 0, name: '' } },
					content: [{ type: 'text', text: `**Error:** ${streamState.error}` }],
					stopReason: null,
					usage: null,
				};

				let savedErrorMsgId: string | undefined;
				let savedErrorParentId: string | null = null;
				if (requestData.chatSessionId) {
					const saved = await this.saveMessage(
						errorAssistantMsg,
						requestData.chatSessionId,
						errorTimestamp
					);
					savedErrorMsgId = saved?.id;
					savedErrorParentId = saved?.parent_message_id || null;
				}

				this.emitStreamEvent(streamState, 'message', {
					processId: streamState.processId,
					message: errorAssistantMsg,
					timestamp: errorTimestamp,
					message_id: savedErrorMsgId,
					parent_message_id: savedErrorParentId,
					sender_id: requestData.sender.id,
					sender_name: requestData.sender.name,
				});

				this.emitStreamEvent(streamState, 'error', {
					processId: streamState.processId,
					error: streamState.error,
					timestamp: errorTimestamp
				});

				this.emitStreamLifecycle(streamState, 'error');
			}

			if (requestData.chatSessionId) {
				browserMcpControl.releaseSession(requestData.chatSessionId);
				debug.log('mcp', `✅ Auto-released MCP tabs for session ${requestData.chatSessionId.slice(0, 8)} on stream error`);
			}
		} finally {
			const { projectPath, projectId, chatSessionId } = requestData;
			// Captured into a const: `userMessageId` is reassigned during the stream,
			// so its narrowing does not survive into the async callback below.
			const snapshotMessageId = userMessageId;
			if (projectPath && projectId && chatSessionId && snapshotMessageId) {
				// The Memory Graph is fed from the SNAPSHOT rather than from tool calls.
				// The snapshot is a gitignore-aware disk diff, so it also catches files
				// rewritten through Bash, a codemod or a formatter — none of which appear
				// as an Edit/Write tool call. Passing the snapshot straight in leaves the
				// `snapshot:captured` broadcast payload (a WS schema) untouched.
				//
				// Ingestion is attached with `finally`, NOT with `then`, and that is the
				// difference between losing a conversation and losing a file list. A
				// snapshot can fail for reasons the transcript has no stake in — a
				// permission error, a path that vanished mid-turn, a repository that grew
				// past a limit — and hanging the only write path in the feature off its
				// success meant one of those quietly cost the turn's memories entirely.
				// The code can be re-read next turn; the conversation happened once.
				snapshotService.captureSnapshot(projectPath, projectId, chatSessionId, snapshotMessageId)
					.then(snapshot => {
						debug.log('chat', `Stream-end snapshot captured for message: ${snapshotMessageId}`);
						this.emit('snapshot:captured', { projectId, chatSessionId });
						return snapshot ? diffSnapshotAgainstParent(snapshot) : null;
					})
					.catch(err => {
						debug.error('chat', 'Failed to capture stream-end snapshot:', err);
						return null;
					})
					.then(delta => {
						ingestTurn({
							projectId,
							projectPath,
							sessionId: chatSessionId,
							userMessageId: snapshotMessageId,
							changedPaths: delta?.changed ?? [],
							deletedPaths: delta?.deleted ?? []
						});
					});
			}
		}
	}

	/**
	 * Get stream state by ID
	 */
	getStream(streamId: string): StreamState | undefined {
		return this.activeStreams.get(streamId);
	}

	/**
	 * Get active stream for a chat session
	 */
	getSessionStream(chatSessionId: string, projectId?: string): StreamState | undefined {
		if (projectId) {
			const sessionKey = this.getSessionKey(projectId, chatSessionId);
			const streamIdWithProject = this.sessionStreams.get(sessionKey);
			if (streamIdWithProject) {
				const stream = this.activeStreams.get(streamIdWithProject);
				if (stream) return stream;
			}
		}

		// Legacy fallback
		const streamId = this.sessionStreams.get(chatSessionId);
		if (!streamId) return undefined;
		return this.activeStreams.get(streamId);
	}

	/**
	 * Generate session key
	 */
	private getSessionKey(projectId: string | undefined, chatSessionId: string): string {
		return projectId ? `${projectId}-${chatSessionId}` : chatSessionId;
	}

	/**
	 * Cancel an active stream
	 */
	/**
	 * Resolve a pending AskUserQuestion for an active stream.
	 * Unblocks the engine's canUseTool callback so the SDK can continue.
	 */
	resolveUserAnswer(chatSessionId: string, projectId: string | undefined, toolUseId: string, answers: Record<string, string>): boolean {
		// Find the active stream for this session
		const sessionKey = this.getSessionKey(projectId, chatSessionId);
		const streamId = this.sessionStreams.get(sessionKey);
		if (!streamId) {
			debug.warn('chat', 'resolveUserAnswer: No stream found for session');
			return false;
		}

		const streamState = this.activeStreams.get(streamId);
		if (!streamState || streamState.status !== 'active') {
			debug.warn('chat', 'resolveUserAnswer: Stream not active');
			return false;
		}

		// The stream is live, so it is holding the engine that owns the pending
		// question. Use that one — a replacement instance holds no question to
		// answer, and a fresh one would not even be the same process.
		const engine = streamState.engineInstance;

		if (!engine?.resolveUserAnswer) {
			debug.warn('chat', 'resolveUserAnswer: Engine does not support resolveUserAnswer');
			return false;
		}

		debug.log('chat', 'Resolving AskUserQuestion answer:', { toolUseId, answers });
		return engine.resolveUserAnswer(toolUseId, answers);
	}

	async cancelStream(streamId: string, reason?: string): Promise<boolean> {
		const streamState = this.activeStreams.get(streamId);
		if (!streamState || streamState.status !== 'active') {
			return false;
		}

		// Mark as cancelled FIRST to prevent further message processing
		streamState.status = 'cancelled';
		streamState.completedAt = new Date();

		// Save partial reasoning text to DB before cancelling (persists across refresh/project switch)
		if (streamState.currentReasoningText && streamState.chatSessionId) {
			try {
				const timestamp = new Date().toISOString();
				const currentHead = sessionQueries.getHead(streamState.chatSessionId);

				const reasoningMessage: ReasoningMessage = {
					type: 'reasoning',
					createdAt: timestamp,
					messageId: crypto.randomUUID(),
					sessionId: null, // Partial cancel saves are not valid resume targets
					parent: { messageId: currentHead || null, sessionId: null, toolUseId: null },
					engine: { type: streamState.engine, provider: '', model: { id: '', name: '' }, account: { id: 0, name: '' } },
					text: streamState.currentReasoningText,
				};

				const savedMessage = messageQueries.create({
					session_id: streamState.chatSessionId,
					message: reasoningMessage,
					timestamp,
					parent_message_id: currentHead || undefined
				});

				sessionQueries.updateHead(streamState.chatSessionId, savedMessage.id);
				sessionQueries.updateOnMessage(streamState.chatSessionId, {
					messageType: 'reasoning', timestamp,
				});
				debug.log('chat', 'Saved partial reasoning on cancel:', savedMessage.id);
			} catch (error) {
				debug.error('chat', 'Failed to save partial reasoning on cancel:', error);
			}
		}

		// Save partial text to DB before cancelling (persists across refresh/project switch)
		if (streamState.currentPartialText && streamState.chatSessionId) {
			try {
				const timestamp = new Date().toISOString();
				const currentHead = sessionQueries.getHead(streamState.chatSessionId);

				const partialMessage: AssistantMessage = {
					type: 'assistant',
					createdAt: timestamp,
					messageId: crypto.randomUUID(),
					sessionId: null, // Partial cancel saves are not valid resume targets
					parent: { messageId: currentHead || null, sessionId: null, toolUseId: null },
					engine: { type: streamState.engine, provider: '', model: { id: '', name: '' }, account: { id: 0, name: '' } },
					content: [{ type: 'text', text: streamState.currentPartialText }],
					stopReason: 'interrupted',
					usage: null,
				};

				const savedMessage = messageQueries.create({
					session_id: streamState.chatSessionId,
					message: partialMessage,
					timestamp,
					parent_message_id: currentHead || undefined
				});

				sessionQueries.updateHead(streamState.chatSessionId, savedMessage.id);
				const clean = streamState.currentPartialText.replace(/```[\s\S]*?```/g, '').trim();
				sessionQueries.updateOnMessage(streamState.chatSessionId, {
					messageType: 'assistant', timestamp,
					headSummary: clean ? clean.slice(0, 200) + (clean.length > 200 ? '...' : '') : undefined,
				});
				debug.log('chat', 'Saved partial text on cancel:', savedMessage.id);
			} catch (error) {
				debug.error('chat', 'Failed to save partial text on cancel:', error);
			}
		}

		// Claude Code only: restore head_session_id to pre-stream value.
		// Claude Code SDK only returns session_id inside yielded messages, so a cancelled
		// stream's fork session_id is not a valid resume target. OpenCode creates sessions
		// synchronously, so its head_session_id is always valid — no restoration needed.
		if (streamState.engine === 'claude-code' && streamState.chatSessionId && streamState.preStreamSessionId !== undefined) {
			try {
				if (streamState.preStreamSessionId) {
					sessionQueries.updateSessionId(streamState.chatSessionId, streamState.preStreamSessionId);
				} else {
					sessionQueries.clearSessionId(streamState.chatSessionId);
				}
				debug.log('chat', `Restored head_session_id to: ${streamState.preStreamSessionId || 'null'}`);
			} catch (error) {
				debug.error('chat', 'Failed to restore head_session_id:', error);
			}
		}

		// Abort first, then release the browser locks, and only then ask the
		// engine to stop.
		//
		// The order matters and used to be the other way round. `engine.cancel()`
		// is awaited for up to five seconds, and a browser-automation batch that
		// was mid-flight kept resolving its target tab throughout — re-acquiring
		// control of a tab moments after (or before) it was handed back, under a
		// chat session whose release had already run. That is the interrupt that
		// left the tab locked with nobody to unlock it. Aborting up front makes
		// the batch stop between actions, and releasing before the wait means
		// anything that slips through is refused rather than orphaned.
		if (!streamState.abortController?.signal.aborted) {
			streamState.abortController?.abort();
		}

		// Auto-release all MCP-controlled tabs for this chat session
		if (streamState.chatSessionId) {
			browserMcpControl.releaseSession(streamState.chatSessionId);
			debug.log('mcp', `✅ Auto-released MCP tabs for session ${streamState.chatSessionId.slice(0, 8)} on stream cancellation`);
		}

		// Stop this stream's run on the engine, with a bounded timeout.
		// engine.cancel() stops the SDK work (Claude Code: close() kills subprocess,
		// OpenCode: aborts controller + HTTP abort to server). If cancel() hangs
		// (e.g. unresponsive SDK), the timeout ensures we always proceed to emit
		// events and update presence — preventing infinite loader on the frontend.
		//
		// The abortController identifies WHICH run to stop. One engine instance is
		// shared by every chat session of a project, so an untargeted cancel stops
		// whichever run the instance happened to have on hand — cancelling one chat
		// would kill another chat of the same project mid-answer.
		try {
			const engine = streamState.engineInstance;
			if (engine && streamState.abortController) {
				await Promise.race([
					engine.cancel(streamState.abortController),
					new Promise<void>(resolve => setTimeout(resolve, 5000))
				]);
			}
		} catch (error) {
			debug.error('chat', 'Error cancelling engine (non-fatal):', error);
		}

		this.emitStreamEvent(streamState, 'cancelled', {
			processId: streamState.processId,
			timestamp: streamState.completedAt.toISOString()
		});

		this.emitStreamLifecycle(streamState, 'cancelled', reason);

		// Sweep again after the engine has actually stopped: a tool call still
		// in flight during the wait above may have taken a lock that its own
		// release path will never reach.
		browserMcpControl.releaseOrphans();

		return true;
	}

	/**
	 * Clean up a stream
	 */
	private cleanupStream(streamId: string): void {
		const streamState = this.activeStreams.get(streamId);
		if (streamState) {
			const sessionKey = this.getSessionKey(streamState.projectId, streamState.chatSessionId);
			// Only delete session key if it still points to THIS stream.
			// A newer stream for the same session may have overridden the key;
			// blindly deleting it would orphan the active stream — making it
			// unfindable by getSessionStream() and breaking cancel/reconnect.
			if (this.sessionStreams.get(sessionKey) === streamId) {
				this.sessionStreams.delete(sessionKey);
			}
			if (this.sessionStreams.get(streamState.chatSessionId) === streamId) {
				this.sessionStreams.delete(streamState.chatSessionId);
			}
			this.activeStreams.delete(streamId);

			// Cleanup project context service
			projectContextService.unregisterStream(streamId);

			// Remove all listeners for this stream
			this.removeAllListeners(`stream:${streamId}`);
			this.removeAllListeners(`session:${sessionKey}`);
		}
	}

	/**
	 * Get all active streams
	 */
	getActiveStreams(): StreamState[] {
		return Array.from(this.activeStreams.values())
			.filter(stream => stream.status === 'active');
	}

	/**
	 * Save message to database and update session metadata.
	 */
	private async saveMessage(
		message: UnifiedMessage,
		sessionId: string,
		timestamp: string
	): Promise<DatabaseMessage | null> {
		try {
			const currentHead = sessionQueries.getHead(sessionId);

			const savedMessage = messageQueries.create({
				session_id: sessionId,
				message: message,
				timestamp,
				parent_message_id: currentHead || undefined
			});

			sessionQueries.updateHead(sessionId, savedMessage.id);

			// ── Update session metadata ──
			try {
				const text = extractMessageText(message);
				const updateOpts: Parameters<typeof sessionQueries.updateOnMessage>[1] = {
					messageType: message.type,
					senderId: message.type === 'user' ? message.sender?.id : undefined,
					senderName: message.type === 'user' ? message.sender?.name : undefined,
					timestamp,
				};

				if (message.type === 'user' && text.trim()) {
					updateOpts.headTitle = text.slice(0, 80) + (text.length > 80 ? '...' : '');
					// Auto-set title if this is the first user message
					if (!currentHead) {
						updateOpts.isFirstUserMessage = true;
						updateOpts.title = updateOpts.headTitle;
					}
				} else if (message.type === 'assistant' && text.trim()) {
					const clean = text.replace(/```[\s\S]*?```/g, '').trim();
					if (clean) {
						updateOpts.headSummary = clean.slice(0, 200) + (clean.length > 200 ? '...' : '');
					}
				}

				sessionQueries.updateOnMessage(sessionId, updateOpts);
			} catch (err) {
				debug.error('chat', 'Failed to update session metadata:', err);
			}

			// Update checkpoint_tree_state when saving a new checkpoint (real user message)
			if (message.type === 'user') {
				try {
					const { isCheckpointMessage, buildCheckpointTree, getCheckpointPathToRoot } = await import('../snapshot/helpers');
					const { checkpointQueries } = await import('../database/queries');

					// Check if this new message is a checkpoint
					if (isCheckpointMessage(savedMessage)) {
						const allMessages = messageQueries.getAllBySessionId(sessionId);
						const { parentMap } = buildCheckpointTree(allMessages);

						// Update active children along path from root to this new checkpoint
						const checkpointPath = getCheckpointPathToRoot(savedMessage.id, parentMap);
						if (checkpointPath.length > 1) {
							checkpointQueries.updateActiveChildrenAlongPath(sessionId, checkpointPath);
						}

						debug.log('snapshot', `Checkpoint tree state updated for new checkpoint ${savedMessage.id.slice(0, 8)}`);
					}
				} catch (err) {
					debug.error('snapshot', 'Failed to update checkpoint tree state:', err);
				}
			}

			return savedMessage;
		} catch (error) {
			debug.error('chat', 'Failed to save message to database:', error);
			return null;
		}
	}

	/**
	 * Backfill stopReason from a result event to the last assistant message in DB.
	 * Claude Code SDK may yield assistant messages with stop_reason: null during streaming;
	 * the actual stop_reason only arrives in the result event after the turn completes.
	 * Returns the mapped StopReason if backfill succeeded, null otherwise.
	 */
	private backfillStopReason(chatSessionId: string, rawStopReason: string): StopReason | null {
		try {
			const mapped: StopReason = (['end_turn', 'tool_use', 'max_tokens', 'interrupted'] as StopReason[])
				.find(r => r === rawStopReason) || 'end_turn';

			const currentHead = sessionQueries.getHead(chatSessionId);
			if (!currentHead) return null;

			const headRow = messageQueries.getById(currentHead);
			if (!headRow) return null;

			const headMsg = JSON.parse(headRow.data) as UnifiedMessage;
			if (headMsg.type !== 'assistant' || headMsg.stopReason) return null;

			const updated = { ...headMsg, stopReason: mapped };
			const { getDatabase } = require('../database');
			getDatabase().prepare('UPDATE messages SET data = ? WHERE id = ?')
				.run(JSON.stringify(updated), headRow.id);

			return mapped;
		} catch (err) {
			debug.error('chat', 'Failed to backfill stopReason:', err);
			return null;
		}
	}

	/**
	 * Backfill `usage` onto every saved assistant row for a stream.
	 *
	 * Codex SDK emits its `Usage` aggregate exactly once per turn (in
	 * `turn.completed`), well after the individual assistant items have
	 * streamed and been persisted live with `usage: null`. Without this
	 * backfill the persisted tool_use rows would carry no usage data, and
	 * after a browser refresh the frontend would lose the per-message usage
	 * that the live stream had attached in memory.
	 *
	 * Engine-gated to Codex by the caller: Claude Code and Copilot already
	 * attach per-call usage on each assistant message, so applying a
	 * stream-wide aggregate to them would overwrite that finer-grained data.
	 */
	private backfillUsageForStream(
		streamState: StreamState,
		usage: TokenUsage,
		requestSender: StreamRequest['sender'],
	): void {
		try {
			const { getDatabase } = require('../database');
			const db = getDatabase();
			const updateStmt = db.prepare('UPDATE messages SET data = ? WHERE id = ?');

			for (const entry of streamState.messages as any[]) {
				const msg = entry?.message as UnifiedMessage | undefined;
				if (!msg || msg.type !== 'assistant') continue;
				if (msg.usage) continue;
				if (!entry.message_id) continue;

				const patched: AssistantMessage = { ...(msg as AssistantMessage), usage };
				try {
					updateStmt.run(JSON.stringify(patched), entry.message_id);
				} catch (err) {
					debug.error('chat', 'Failed to UPDATE usage row:', err);
					continue;
				}

				entry.message = patched;
				entry.usage = usage;

				this.emitStreamEvent(streamState, 'message', {
					processId: streamState.processId,
					message: patched,
					usage,
					timestamp: patched.createdAt,
					message_id: entry.message_id,
					parent_message_id: entry.parent_message_id ?? null,
					sender_id: requestSender.id,
					sender_name: requestSender.name,
				});
			}
		} catch (err) {
			debug.error('chat', 'Failed to backfill usage for stream:', err);
		}
	}

	/**
	 * Extract detailed error info from an error object.
	 * Handles engine-specific error shapes (Anthropic APIError, OpenCode SDK errors).
	 */
	private extractErrorDetail(error: unknown): string {
		if (!error) return 'Unknown error';
		if (typeof error === 'string') return this.normalizeErrorText(error);
		if (!(error instanceof Error)) return this.normalizeErrorText(String(error));

		const err = error as Record<string, any>;
		let message = err.message || 'Unknown error';

		// Enrich with status code if available (Anthropic APIError has .status)
		if (err.status && !message.includes(String(err.status))) {
			message += ` (status ${err.status})`;
		}

		// Enrich with nested error body (Anthropic APIError has .error.message)
		if (err.error?.message && !message.includes(err.error.message)) {
			message += ` - ${err.error.message}`;
		}

		return this.normalizeErrorText(message);
	}

	/**
	 * Strip redundant error class name prefixes from error text.
	 *
	 * Error text often passes through multiple wrapping layers (SDK → adapter → stream-manager),
	 * each potentially adding class names and "Error:" prefixes. This strips them:
	 *
	 *  "APIError: Bad Request: ..."                → "Bad Request: ..."
	 *  "UnknownError: Error: Unable to connect..." → "Unable to connect..."
	 *  "Claude Code process exited with code 1"    → unchanged
	 */
	private normalizeErrorText(text: string): string {
		let cleaned = text;

		// Strip error class name prefixes (e.g. "APIError: ", "UnknownError: ", "BadRequestError: ")
		cleaned = cleaned.replace(/^[A-Za-z]*Error:\s*/, '');

		// Strip a second redundant "Error: " that may remain (e.g. "UnknownError: Error: ...")
		cleaned = cleaned.replace(/^Error:\s*/, '');

		return cleaned.trim() || text;
	}

	/**
	 * Check if file exists
	 */
	private async existsSync(filePath: string): Promise<boolean> {
		try {
			const file = Bun.file(filePath);
			await file.stat();
			return true;
		} catch {
			return false;
		}
	}

	/**
	 * Get all streams for a specific project
	 */
	getProjectStreams(projectId: string): StreamState[] {
		const projectStreams: StreamState[] = [];
		this.activeStreams.forEach(stream => {
			if (stream.projectId === projectId) {
				projectStreams.push(stream);
			}
		});
		return projectStreams;
	}

	/**
	 * Get all streams
	 */
	getAllStreams(): StreamState[] {
		return Array.from(this.activeStreams.values());
	}

	/**
	 * Check if project has active streams
	 */
	hasActiveStreams(projectId: string): boolean {
		for (const stream of this.activeStreams.values()) {
			if (stream.projectId === projectId && stream.status === 'active') {
				return true;
			}
		}
		return false;
	}

	/**
	 * Clean up completed streams for a project
	 */
	cleanupProjectStreams(projectId: string): void {
		const streamsToClean: string[] = [];

		this.activeStreams.forEach((stream, streamId) => {
			if (stream.projectId === projectId &&
					(stream.status === 'completed' || stream.status === 'error' || stream.status === 'cancelled')) {
				streamsToClean.push(streamId);
			}
		});

		streamsToClean.forEach(streamId => {
			this.cleanupStream(streamId);
		});
	}

	/**
	 * Cancel and clean up all streams for a specific chat session.
	 * Used when a session is deleted to remove green/amber status indicators.
	 */
	async cleanupSessionStreams(chatSessionId: string): Promise<void> {
		const streamsToCancel: string[] = [];
		const streamsToClean: string[] = [];

		this.activeStreams.forEach((stream, streamId) => {
			if (stream.chatSessionId === chatSessionId) {
				if (stream.status === 'active') {
					streamsToCancel.push(streamId);
				} else {
					streamsToClean.push(streamId);
				}
			}
		});

		// Cancel active streams and await their processStream promise so the
		// finally block (snapshot capture) completes before the caller deletes
		// the session — preventing FOREIGN KEY constraint failures.
		for (const streamId of streamsToCancel) {
			await this.cancelStream(streamId, 'session-deleted');
			const stream = this.activeStreams.get(streamId);
			if (stream?.streamPromise) {
				await stream.streamPromise.catch(() => {});
			}
		}

		// Clean up non-active streams
		for (const streamId of streamsToClean) {
			this.cleanupStream(streamId);
		}
	}

	/**
	 * Clean up all completed streams
	 */
	cleanupAllCompletedStreams(): void {
		const streamsToClean: string[] = [];

		this.activeStreams.forEach((stream, streamId) => {
			if (stream.status !== 'active') {
				streamsToClean.push(streamId);
			}
		});

		streamsToClean.forEach(streamId => {
			this.cleanupStream(streamId);
		});
	}
}

// Export singleton instance
export const streamManager = new StreamManager();