/**
 * Chat Streaming Events (Optimized)
 *
 * High-performance event-driven chat streaming with:
 * - Zero polling (pure EventEmitter push)
 * - Automatic cleanup on disconnect
 * - Support for stream cancellation
 */

import { t } from 'elysia';
import { createRouter } from '$shared/utils/ws-server';
import { streamManager, type StreamEvent } from '../../chat/stream-manager';
import type { EngineType, UnifiedMessage } from '$shared/types/unified';
import { debug } from '$shared/utils/logger';
import { trimSubAgentForWire } from '$shared/utils/subagent-wire-trim';
import { ws } from '$backend/utils/ws';
import { broadcastPresence } from '../projects/status';
import { projectQueries, sessionQueries, messageQueries } from '../../database/queries';
import { sendPushToUser } from '../../push/sender';
import { chatStreamPushTag, waitingInputPushTag } from '../../push/tags';
import {
	chatNotificationMessage,
	type ChatNotificationContext
} from '$shared/constants/notification-messages';
import { requireSessionAccess } from '../access';
import { resolveSessionPath } from '../../worktrees';

/**
 * Where a notification-worthy event happened, resolved once on the server.
 *
 * The browser cannot do this itself: the sessions store only holds the project
 * currently open, and these notifications exist precisely for the chats the
 * user is *not* looking at. Shipping the resolved names in the event payload
 * also keeps the tab's local toast worded identically to the server push that
 * replaces it — they share a tag, so any difference would read as the
 * notification rewriting itself.
 *
 * Missing rows yield `undefined`, never a placeholder: the composer drops an
 * absent segment, which says more than the word "Unknown" did.
 */
function resolveNotificationContext(
	projectId: string,
	chatSessionId: string | undefined
): ChatNotificationContext {
	return {
		projectName: projectQueries.getById(projectId)?.name || undefined,
		sessionTitle: chatSessionId ? sessionQueries.getById(chatSessionId)?.title || undefined : undefined
	};
}

/**
 * Broadcast presence + notify project members when a chat message may flip the
 * stream's waiting-for-input state (an AskUserQuestion tool_use arrives, or a
 * tool_result answers it). `event.data.message` IS the UnifiedMessage, so its
 * blocks live at `.content` — reading `.message.content` (one level too deep)
 * silently yields `[]` and the status indicators never update until refresh.
 *
 * Shared by the initial-stream and reconnect message handlers so the two paths
 * cannot drift apart again.
 */
function handleWaitingInputChange(
	event: StreamEvent,
	chatSessionId: string,
	projectId: string | undefined,
): void {
	const message = event.data?.message;
	const msgContent = Array.isArray(message?.content) ? message.content : [];

	const askToolUse = msgContent.find(
		(item: any) => item.type === 'tool_use' && item.name === 'AskUserQuestion'
	);
	const hasToolResult = msgContent.some((item: any) => item.type === 'tool_result');

	// Recompute presence whenever the waiting state may have changed in either
	// direction (AskUserQuestion → amber, tool_result → green).
	if (askToolUse || hasToolResult) {
		broadcastPresence().catch((err) => {
			debug.warn('chat', 'Presence broadcast error on waiting-input state change:', err);
		});
	}

	// Notify all project members when AskUserQuestion arrives (sound + push)
	if (askToolUse && projectId) {
		const context = resolveNotificationContext(projectId, chatSessionId);

		ws.emit.projectMembers(projectId, 'chat:waiting-input', {
			projectId,
			chatSessionId,
			toolUseId: askToolUse.id,
			timestamp: event.data?.timestamp || new Date().toISOString(),
			...context
		});

		// Background route: reach the requester's devices even when Chrome is
		// closed. Deterministic tag replaces (not stacks with) the local toast
		// the open tab shows for the same question.
		const requestedByUserId = streamManager.getSessionStream(
			chatSessionId,
			projectId
		)?.requestedByUserId;
		if (requestedByUserId) {
			void sendPushToUser(requestedByUserId, {
				...chatNotificationMessage('waiting-input', context),
				tag: waitingInputPushTag(askToolUse.id)
			}).catch((error) => {
				debug.warn('notification', 'Background waiting-input push failed:', error);
			});
		}
	}
}

// ============================================================================
// Global stream lifecycle handler (module-level, not per-connection)
//
// This fires for ALL stream completions, even when no per-connection subscriber
// exists (e.g., after browser refresh when user is on a different project).
// Ensures cross-project notifications (presence update, sound, push) always work.
// ============================================================================
streamManager.on('stream:lifecycle', (event: { status: string; streamId: string; projectId?: string; chatSessionId?: string; requestedByUserId?: string; timestamp: string; reason?: string }) => {
	const { status, streamId, projectId, chatSessionId, requestedByUserId, timestamp, reason } = event;
	if (!projectId) return;

	debug.log('chat', `Stream lifecycle: ${status} for project ${projectId} session ${chatSessionId}${reason ? ` (reason: ${reason})` : ''}`);

	// Mark any tool_use blocks that never got a tool_result as interrupted (persisted to DB)
	if (chatSessionId) {
		try {
			messageQueries.markInterruptedMessages(chatSessionId);
		} catch (err) {
			debug.error('chat', 'Failed to mark interrupted messages:', err);
		}
	}

	// Resolved once and shared: the payload below carries it to the open tab,
	// and the background push below that reuses the very same values, so the
	// two copies of one event cannot word themselves differently.
	const context = resolveNotificationContext(projectId, chatSessionId);

	// Notify all project members (cross-project notification for sound + push)
	ws.emit.projectMembers(projectId, 'chat:stream-finished', {
		projectId,
		chatSessionId: chatSessionId || '',
		streamId,
		status: status as 'completed' | 'error' | 'cancelled',
		timestamp,
		reason,
		...context
	});

	// Background route: the open tab notifies locally via the WS event above;
	// this reaches the requester's devices when Chrome is closed. The shared
	// deterministic tag means an open tab's local toast is replaced, never
	// duplicated. Never throws — chat completion must not depend on push.
	if (requestedByUserId && reason !== 'session-deleted') {
		// `cancelled` gets its own title. Filing it under a "complete" heading
		// with an "interrupted" body made the notification contradict itself.
		const notificationEvent =
			status === 'error' ? 'error' : status === 'cancelled' ? 'cancelled' : 'completed';
		void sendPushToUser(requestedByUserId, {
			...chatNotificationMessage(notificationEvent, context),
			tag: chatStreamPushTag(streamId)
		}).catch((error) => {
			debug.warn('notification', 'Background stream-finished push failed:', error);
		});
	}

	// Broadcast updated presence (status indicators for all projects)
	broadcastPresence().catch((err) => {
		debug.warn('chat', 'Presence broadcast error after stream lifecycle:', err);
	});
});

// Notify project members when a snapshot is captured (so the timeline modal can refresh stats)
streamManager.on('snapshot:captured', (event: { projectId: string; chatSessionId: string }) => {
	const { projectId, chatSessionId } = event;
	if (!projectId) return;

	ws.emit.projectMembers(projectId, 'snapshot:captured', { projectId, chatSessionId });
});

// In-memory store for latest chat input state per chat session (keyed by chatSessionId)
const chatSessionInputState = new Map<string, { text: string; senderId: string; attachments?: any[] }>();

// In-memory store for edit mode state per chat session (keyed by chatSessionId)
const chatSessionEditMode = new Map<string, { isEditing: boolean; messageId: string | null; messageTimestamp: string | null }>();

// In-memory store for model state per chat session (keyed by chatSessionId)
const chatSessionModelState = new Map<string, { engine: string; provider: string; modelId: string; modelName: string; senderId: string }>();

// In-memory store for account state per chat session (keyed by chatSessionId)
const chatSessionAccountState = new Map<string, { accountId: number | null; senderId: string }>();

export const streamHandler = createRouter()
	// Join a chat session room (subscribe to chat events for this session)
	.on('chat:join-session', {
		data: t.Object({
			chatSessionId: t.String()
		})
	}, ({ data, conn }) => {
		requireSessionAccess(conn, data.chatSessionId);
		// Leave all previous chat sessions first (1 session at a time per connection)
		ws.leaveAllChatSessions(conn);
		ws.joinChatSession(conn, data.chatSessionId);
		// Broadcast presence so all clients see updated chatSessionUsers
		broadcastPresence().catch((err) => {
			debug.warn('chat', 'Presence broadcast error after chat join-session:', err);
		});

		// Rehydrate rate-limit banners: replay only snapshots matching the
		// session's engine+account so banners are scoped to the active context.
		// In-memory state takes priority (most current); DB is the fallback for
		// sessions joined after a server restart.
		try {
			const userId = ws.getUserId(conn);
			const inMemoryModel = chatSessionModelState.get(data.chatSessionId);
			const inMemoryAccount = chatSessionAccountState.get(data.chatSessionId);
			const sessionEngine: string | undefined =
				inMemoryModel?.engine ?? sessionQueries.getById(data.chatSessionId)?.engine;
			const sessionAccountId: number | null =
				inMemoryAccount?.accountId ?? sessionQueries.getById(data.chatSessionId)?.account_id ?? null;

			const snapshots = streamManager.getAllActiveRateLimits().filter(snap => {
				if (sessionEngine && snap.engine !== sessionEngine) return false;
				if (sessionAccountId != null && snap.accountId !== sessionAccountId) return false;
				return true;
			});

			for (const snap of snapshots) {
				ws.emit.user(userId, 'chat:rate_limit', {
					chatSessionId: data.chatSessionId,
					engine: snap.engine,
					accountId: snap.accountId,
					status: snap.status,
					utilization: snap.utilization,
					resetsAt: snap.resetsAt,
					rateLimitType: snap.rateLimitType,
					timestamp: snap.receivedAt
				});
			}
		} catch (err) {
			debug.warn('chat', 'Failed to rehydrate rate-limit on join-session:', err);
		}
	})

	// Dismiss the active rate-limit snapshot for a given engine/account so the
	// banner stops re-hydrating on subsequent join-session calls.
	.on('chat:rate-limit-dismiss', {
		data: t.Object({
			engine: t.String(),
			accountId: t.Number()
		})
	}, ({ data }) => {
		if (!data.accountId) return;
		streamManager.dismissRateLimit(data.engine as EngineType, data.accountId);
	})

	// Leave a chat session room
	.on('chat:leave-session', {
		data: t.Object({
			chatSessionId: t.String()
		})
	}, ({ data, conn }) => {
		ws.leaveChatSession(conn, data.chatSessionId);
		// Broadcast presence so all clients see updated chatSessionUsers
		broadcastPresence().catch((err) => {
			debug.warn('chat', 'Presence broadcast error after chat leave-session:', err);
		});
	})

	// Start chat stream
	.on('chat:stream', {
		data: t.Object({
			sessionId: t.String(),
			chatSessionId: t.String(),
			projectPath: t.String(),
			prompt: t.Any(), // UserMessage object
			engine: t.Object({
				type: t.Union([t.Literal('claude-code'), t.Literal('opencode'), t.Literal('copilot'), t.Literal('codex'), t.Literal('qwen'), t.Literal('pi'), t.Literal('cline'), t.Literal('cursor')]),
				provider: t.String(),
				model: t.Object({
					id: t.String(),
					name: t.String()
				}),
				account: t.Object({
					id: t.Number(),
					name: t.String()
				})
			}),
			sender: t.Object({
				id: t.String(),
				name: t.String()
			}),
			// Active Profile for this session (null = explicit none; absent = use
			// the project default). Persisted like engine/model.
			profileId: t.Optional(t.Union([t.Number(), t.Null()])),
			// Reasoning/thinking level for this session (native per engine; null/
			// absent = engine default). Persisted like engine/model.
			reasoningEffort: t.Optional(t.Union([t.String(), t.Null()]))
		})
	}, async ({ data, conn }) => {
		requireSessionAccess(conn, data.chatSessionId);
		const projectId = ws.getProjectId(conn);

		// The client's projectPath is a view preference; the session's worktree is
		// the fact. Resolving server-side is what keeps an isolated session from
		// writing into the main tree when the two disagree.
		const workingRoot = resolveSessionPath(data.chatSessionId, data.projectPath);

		try {
			debug.log('chat', 'WS chat:stream received:', {
				chatSessionId: data.chatSessionId,
				projectId,
				workingRoot
			});

			// The WS user id is server-trusted (unlike the client-supplied sender)
			// and routes background Web Push to the requester's devices. Missing
			// context must not break the stream — push is best-effort.
			let requestedByUserId: string | undefined;
			try {
				requestedByUserId = ws.getUserId(conn);
			} catch {
				requestedByUserId = undefined;
			}

			// Start background stream
			const streamId = await streamManager.startStream({
				projectPath: workingRoot,
				projectId,
				prompt: data.prompt,
				chatSessionId: data.chatSessionId,
				engine: data.engine,
				sender: data.sender,
				profileId: data.profileId,
				reasoningEffort: data.reasoningEffort,
				requestedByUserId
			});

			debug.log('chat', 'Stream started with ID:', streamId);

			// Emit connection event to chat session room and broadcast presence immediately
			// (the connection event from startStream() fires before subscription, so we emit it manually)
			const stream = streamManager.getStream(streamId);
			if (stream) {
				ws.emit.chatSession(data.chatSessionId, 'chat:connection', {
					chatSessionId: data.chatSessionId,
					processId: stream.processId,
					timestamp: stream.startedAt.toISOString(),
					seq: 1
				});
				// User message is broadcast by stream-manager via event subscription below
				// (includes resume, sender info, and saved message ID)
			}
			broadcastPresence().catch((err) => {
				debug.warn('chat', 'Presence broadcast error on chat:stream start:', err);
			});

			// Subscribe to stream events (event-driven, no polling)
			// Use ws.emit.chatSession() for session-scoped chat events
			// Only users who joined this chat session room receive events
			const chatSessionId = data.chatSessionId;
			const handleStreamEvent = (event: StreamEvent) => {
				try {
					switch (event.type) {
						case 'connection':
							ws.emit.chatSession(chatSessionId, 'chat:connection', {
								chatSessionId,
								processId: event.data.processId,
								timestamp: event.data.timestamp,
								seq: event.seq
							});
							broadcastPresence().catch((err) => {
								debug.warn('chat', 'Presence broadcast error on stream connection event:', err);
							});
							break;

						case 'message': {
							ws.emit.chatSession(chatSessionId, 'chat:message', {
								chatSessionId,
								processId: event.processId,
								message: trimSubAgentForWire(event.data.message),
								usage: event.data.usage,
								timestamp: event.data.timestamp,
								message_id: event.data.message_id,
								parent_message_id: event.data.parent_message_id,
								sender_id: event.data.sender_id,
								sender_name: event.data.sender_name,
								engine: event.data.engine,
								seq: event.seq
							});
							// Broadcast presence + notify when waiting-input state may change
							handleWaitingInputChange(event, chatSessionId, projectId);
							break;
						}

						case 'partial':
							ws.emit.chatSession(chatSessionId, 'chat:partial', {
								chatSessionId,
								processId: event.processId,
								eventType: event.data.eventType as any,
								partialText: event.data.partialText || '',
								deltaText: event.data.deltaText || '',
								...(event.data.reasoning && { reasoning: true }),
								timestamp: event.data.timestamp,
								seq: event.seq
							});
							break;

						case 'notification':
							ws.emit.chatSession(chatSessionId, 'chat:notification', {
								notification: event.data.notification,
								timestamp: event.data.timestamp,
								seq: event.seq
							});
							break;

						case 'rate_limit':
							ws.emit.chatSession(chatSessionId, 'chat:rate_limit', {
								chatSessionId: event.data.chatSessionId,
								engine: event.data.engine,
								accountId: event.data.accountId,
								status: event.data.status,
								utilization: event.data.utilization,
								resetsAt: event.data.resetsAt,
								rateLimitType: event.data.rateLimitType ?? null,
								timestamp: event.data.timestamp,
								seq: event.seq
							});
							break;

						case 'complete':
							ws.emit.chatSession(chatSessionId, 'chat:complete', {
								chatSessionId,
								processId: event.processId,
								timestamp: event.data.timestamp,
								seq: event.seq
							});
							// Cross-project notifications (stream-finished, presence) are handled
							// by the global stream:lifecycle listener at the module level.
							unsubscribe();
							break;

						case 'error':
							ws.emit.chatSession(chatSessionId, 'chat:error', {
								chatSessionId,
								processId: event.processId,
								error: event.data.error,
								timestamp: event.data.timestamp,
								seq: event.seq
							});
							// Cross-project notifications handled by stream:lifecycle listener
							unsubscribe();
							break;

						case 'cancelled':
							ws.emit.chatSession(chatSessionId, 'chat:error', {
								chatSessionId,
								processId: event.processId,
								error: 'Stream cancelled',
								timestamp: event.data.timestamp,
								seq: event.seq
							});
							// Cross-project notifications handled by stream:lifecycle listener
							unsubscribe();
							break;
					}
				} catch (err) {
					// Log but do NOT unsubscribe — one bad event must not kill the
					// entire stream subscription. The bridge between StreamManager
					// and the WS room would be permanently broken, causing the UI
					// to stop receiving stream output while the WS stays connected.
					debug.error('chat', 'Error handling stream event:', err);
				}
			};

			// Subscribe to stream events
			const unsubscribe = streamManager.subscribeToStream(streamId, handleStreamEvent);

			// Register cleanup with WSServer (called automatically on connection close)
			ws.addCleanup(conn, unsubscribe);

		} catch (error) {
			const errorMessage = error instanceof Error ? error.message : 'Unknown error';
			debug.error('chat', 'WS chat:stream error:', errorMessage);

			ws.emit.chatSession(data.chatSessionId, 'chat:error', {
				chatSessionId: data.chatSessionId,
				processId: crypto.randomUUID(),
				error: errorMessage,
				timestamp: new Date().toISOString()
			});
		}
	})

	// Reconnect to an active stream (after browser refresh / project switch)
	// Re-subscribes the new connection to receive live stream events
	.on('chat:reconnect', {
		data: t.Object({
			chatSessionId: t.String()
		})
	}, async ({ data, conn }) => {
		requireSessionAccess(conn, data.chatSessionId);
		const projectId = ws.getProjectId(conn);

		try {
			const chatSessionId = data.chatSessionId;
			const streamState = streamManager.getSessionStream(chatSessionId, projectId);
			if (!streamState || streamState.status !== 'active') {
				// No active stream - nothing to reconnect
				return;
			}

			debug.log('chat', 'Reconnecting to active stream:', streamState.streamId);

			// Subscribe this connection to the stream's events (session-scoped)
			const handleStreamEvent = (event: StreamEvent) => {
				try {
					switch (event.type) {
						case 'connection':
							ws.emit.chatSession(chatSessionId, 'chat:connection', {
								chatSessionId,
								processId: event.data.processId,
								timestamp: event.data.timestamp,
								seq: event.seq
							});
							break;

						case 'message': {
							ws.emit.chatSession(chatSessionId, 'chat:message', {
								chatSessionId,
								processId: event.processId,
								message: trimSubAgentForWire(event.data.message),
								usage: event.data.usage,
								timestamp: event.data.timestamp,
								message_id: event.data.message_id,
								parent_message_id: event.data.parent_message_id,
								sender_id: event.data.sender_id,
								sender_name: event.data.sender_name,
								engine: event.data.engine,
								seq: event.seq
							});
							// Broadcast presence + notify when waiting-input state may change
							handleWaitingInputChange(event, chatSessionId, projectId);
							break;
						}

						case 'partial':
							ws.emit.chatSession(chatSessionId, 'chat:partial', {
								chatSessionId,
								processId: event.processId,
								eventType: event.data.eventType as any,
								partialText: event.data.partialText || '',
								deltaText: event.data.deltaText || '',
								...(event.data.reasoning && { reasoning: true }),
								timestamp: event.data.timestamp,
								seq: event.seq
							});
							break;

						case 'notification':
							ws.emit.chatSession(chatSessionId, 'chat:notification', {
								notification: event.data.notification,
								timestamp: event.data.timestamp,
								seq: event.seq
							});
							break;

						case 'rate_limit':
							ws.emit.chatSession(chatSessionId, 'chat:rate_limit', {
								chatSessionId: event.data.chatSessionId,
								engine: event.data.engine,
								accountId: event.data.accountId,
								status: event.data.status,
								utilization: event.data.utilization,
								resetsAt: event.data.resetsAt,
								rateLimitType: event.data.rateLimitType ?? null,
								timestamp: event.data.timestamp,
								seq: event.seq
							});
							break;

						case 'complete':
							ws.emit.chatSession(chatSessionId, 'chat:complete', {
								chatSessionId,
								processId: event.processId,
								timestamp: event.data.timestamp,
								seq: event.seq
							});
							// Cross-project notifications handled by stream:lifecycle listener
							unsubscribe();
							break;

						case 'error':
							ws.emit.chatSession(chatSessionId, 'chat:error', {
								chatSessionId,
								processId: event.processId,
								error: event.data.error,
								timestamp: event.data.timestamp,
								seq: event.seq
							});
							// Cross-project notifications handled by stream:lifecycle listener
							unsubscribe();
							break;

						case 'cancelled':
							ws.emit.chatSession(chatSessionId, 'chat:error', {
								chatSessionId,
								processId: event.processId,
								error: 'Stream cancelled',
								timestamp: event.data.timestamp,
								seq: event.seq
							});
							// Cross-project notifications handled by stream:lifecycle listener
							unsubscribe();
							break;
					}
				} catch (err) {
					// Log but do NOT unsubscribe — same rationale as the initial
					// stream handler: a transient error must not permanently break
					// the EventEmitter → WS room bridge.
					debug.error('chat', 'Error handling reconnected stream event:', err);
				}
			};

			const unsubscribe = streamManager.subscribeToStream(streamState.streamId, handleStreamEvent);
			ws.addCleanup(conn, unsubscribe);

			// Send current state snapshot to chat session room so frontend can catch up
			ws.emit.chatSession(chatSessionId, 'chat:connection', {
				chatSessionId,
				processId: streamState.processId,
				timestamp: streamState.startedAt.toISOString(),
				seq: streamState.eventSeq
			});

		} catch (error) {
			debug.error('chat', 'Error reconnecting to stream:', error);
		}
	})

	// Handle AskUserQuestion answer from user.
	//
	// Routed to the engine's built-in resolveUserAnswer (Claude: canUseTool,
	// OpenCode: event hook). Codex has no callback hook in the SDK — when
	// the SDK adds native support, route it here too instead of resurrecting
	// the previous MCP-based fallback.
	.on('chat:ask-user-answer', {
		data: t.Object({
			chatSessionId: t.String(),
			toolUseId: t.String(),
			answers: t.Record(t.String(), t.String())
		})
	}, async ({ data, conn }) => {
		requireSessionAccess(conn, data.chatSessionId);
		const projectId = ws.getProjectId(conn);

		try {
			debug.log('chat', 'WS chat:ask-user-answer received:', {
				chatSessionId: data.chatSessionId,
				toolUseId: data.toolUseId,
				answers: data.answers
			});

			const handled = streamManager.resolveUserAnswer(
				data.chatSessionId,
				projectId,
				data.toolUseId,
				data.answers
			);
			if (!handled) {
				debug.warn('chat', 'Failed to resolve user answer (no engine handler for this toolUseId)');
			}
		} catch (error) {
			const errorMessage = error instanceof Error ? error.message : 'Unknown error';
			debug.error('chat', 'WS chat:ask-user-answer error:', errorMessage);
		}
	})

	// Get active stream state (for reconnect after browser refresh / project switch)
	.http('chat:stream-state', {
		data: t.Object({
			chatSessionId: t.Optional(t.String())
		}),
		response: t.Object({
			streamId: t.String(),
			status: t.Union([
				t.Literal('active'),
				t.Literal('completed'),
				t.Literal('error'),
				t.Literal('cancelled')
			]),
			processId: t.String(),
			messages: t.Array(t.Any()),
			currentPartialText: t.Optional(t.String()),
			currentReasoningText: t.Optional(t.String()),
			error: t.Optional(t.String()),
			startedAt: t.String(),
			completedAt: t.Optional(t.String())
		})
	}, async ({ data, conn }) => {
		const projectId = ws.getProjectId(conn);

		if (data.chatSessionId) {
			requireSessionAccess(conn, data.chatSessionId);
		}

		const streamState = data.chatSessionId
			? streamManager.getSessionStream(data.chatSessionId, projectId)
			: undefined;

		// Return a "not found" response instead of throwing
		// This happens normally when a stream was cancelled/completed and cleaned up
		if (!streamState) {
			return {
				streamId: data.chatSessionId || '',
				status: 'completed' as const,
				processId: '',
				messages: [],
				currentPartialText: undefined,
				currentReasoningText: undefined,
				error: undefined,
				startedAt: new Date().toISOString(),
				completedAt: new Date().toISOString()
			};
		}

		return {
			streamId: streamState.streamId,
			status: streamState.status,
			processId: streamState.processId,
			// Trim sub-agent noise from the catch-up buffer too (see trimSubAgentForWire).
			// Buffer entries wrap the UnifiedMessage under `.message`; leave the rest intact.
			messages: streamState.messages.map(entry => {
				const wrapped = entry as { message?: UnifiedMessage };
				return wrapped?.message
					? { ...wrapped, message: trimSubAgentForWire(wrapped.message) }
					: entry;
			}),
			currentPartialText: streamState.currentPartialText,
			currentReasoningText: streamState.currentReasoningText,
			error: streamState.error,
			startedAt: streamState.startedAt.toISOString(),
			completedAt: streamState.completedAt?.toISOString()
		};
	})

	// Cancel stream
	.on('chat:cancel', {
		data: t.Object({
			sessionId: t.String(),
			chatSessionId: t.String()
		})
	}, async ({ data, conn }) => {
		requireSessionAccess(conn, data.chatSessionId);
		const projectId = ws.getProjectId(conn);
		const chatSessionId = data.chatSessionId;

		try {
			// Find stream by session
			const streamState = streamManager.getSessionStream(chatSessionId, projectId);
			if (!streamState) {
				// Stream not found - could already be completed/cleaned up
				// Send cancellation to chat session room so frontend stops loading
				ws.emit.chatSession(chatSessionId, 'chat:cancelled', {
					chatSessionId,
					status: 'cancelled',
					processId: ''
				});
				broadcastPresence().catch((err) => {
					debug.warn('chat', 'Presence broadcast error after cancel (stream not found):', err);
				});
				return;
			}

			await streamManager.cancelStream(streamState.streamId);
			// Always send cancelled to chat session room to clear UI
			ws.emit.chatSession(chatSessionId, 'chat:cancelled', {
				chatSessionId,
				status: 'cancelled',
				processId: streamState.processId
			});
			// Always broadcast presence after cancel attempt to update all clients
			broadcastPresence().catch((err) => {
				debug.warn('chat', 'Presence broadcast error after stream cancel:', err);
			});
		} catch (error) {
			const errorMessage = error instanceof Error ? error.message : 'Unknown error';
			ws.emit.chatSession(chatSessionId, 'chat:error', {
				chatSessionId,
				processId: '',
				error: errorMessage,
				timestamp: new Date().toISOString()
			});
			broadcastPresence().catch((err) => {
				debug.warn('chat', 'Presence broadcast error after cancel exception:', err);
			});
		}
	})

	// Collaborative edit mode - broadcast and store edit mode state per chat session
	.on('chat:edit-mode', {
		data: t.Object({
			senderId: t.String(),
			chatSessionId: t.String(),
			isEditing: t.Boolean(),
			messageId: t.Union([t.String(), t.Null()]),
			messageTimestamp: t.Union([t.String(), t.Null()])
		})
	}, ({ data, conn }) => {
		requireSessionAccess(conn, data.chatSessionId);
		const chatSessionId = data.chatSessionId;

		// Store on server for late joiners / refresh (keyed by chatSessionId)
		if (data.isEditing && data.messageId) {
			chatSessionEditMode.set(chatSessionId, {
				isEditing: true,
				messageId: data.messageId,
				messageTimestamp: data.messageTimestamp
			});
		} else {
			chatSessionEditMode.delete(chatSessionId);
		}

		ws.emit.chatSession(chatSessionId, 'chat:edit-mode', {
			senderId: data.senderId,
			isEditing: data.isEditing,
			messageId: data.messageId,
			messageTimestamp: data.messageTimestamp
		});
	})

	// Get current edit mode state for a chat session (for refresh / late joiners)
	.http('chat:get-edit-mode', {
		data: t.Object({
			chatSessionId: t.Optional(t.String())
		}),
		response: t.Object({
			isEditing: t.Boolean(),
			messageId: t.Union([t.String(), t.Null()]),
			messageTimestamp: t.Union([t.String(), t.Null()])
		})
	}, ({ data, conn }) => {
		const chatSessionId = data.chatSessionId || '';
		if (chatSessionId) {
			requireSessionAccess(conn, chatSessionId);
		}
		const editState = chatSessionEditMode.get(chatSessionId);
		return editState || { isEditing: false, messageId: null, messageTimestamp: null };
	})

	// Collaborative input sync - broadcast typing/attachments to other users in the same chat session
	.on('chat:input-sync', {
		data: t.Object({
			text: t.String(),
			senderId: t.String(),
			chatSessionId: t.String(),
			attachments: t.Optional(t.Array(t.Object({
				id: t.String(),
				fileName: t.String(),
				type: t.String(),
				mediaType: t.String(),
				base64: t.String()
			})))
		})
	}, ({ data, conn }) => {
		requireSessionAccess(conn, data.chatSessionId);
		const chatSessionId = data.chatSessionId;

		// Store latest input state on server for late-joining users (keyed by chatSessionId)
		chatSessionInputState.set(chatSessionId, {
			text: data.text,
			senderId: data.senderId,
			attachments: data.attachments
		});

		ws.emit.chatSession(chatSessionId, 'chat:input-sync', {
			text: data.text,
			senderId: data.senderId,
			attachments: data.attachments
		});
	})

	// Get latest input state for a chat session (for users switching sessions)
	.http('chat:get-input-state', {
		data: t.Object({
			chatSessionId: t.Optional(t.String())
		}),
		response: t.Object({
			text: t.String(),
			senderId: t.String(),
			attachments: t.Optional(t.Array(t.Object({
				id: t.String(),
				fileName: t.String(),
				type: t.String(),
				mediaType: t.String(),
				base64: t.String()
			})))
		})
	}, ({ data, conn }) => {
		const chatSessionId = data.chatSessionId || '';
		if (chatSessionId) {
			requireSessionAccess(conn, chatSessionId);
		}
		const state = chatSessionInputState.get(chatSessionId);
		return state || { text: '', senderId: '' };
	})

	// Collaborative model sync - broadcast model changes to other users in the same chat session
	.on('chat:model-sync', {
		data: t.Object({
			senderId: t.String(),
			chatSessionId: t.String(),
			engine: t.String(),
			provider: t.String(),
			modelId: t.String(),
			modelName: t.String()
		})
	}, ({ data, conn }) => {
		requireSessionAccess(conn, data.chatSessionId);
		const chatSessionId = data.chatSessionId;

		// Store latest model state on server for late joiners / refresh
		chatSessionModelState.set(chatSessionId, {
			engine: data.engine,
			provider: data.provider,
			modelId: data.modelId,
			modelName: data.modelName,
			senderId: data.senderId
		});

		// Persist engine/model to the session record in the database
		// so refreshes and late joiners get the correct model
		try {
			sessionQueries.updateEngineModel(chatSessionId, data.engine, data.provider, data.modelId, data.modelName);
		} catch (err) {
			debug.error('chat', 'Failed to persist model sync to DB:', err);
		}

		// Broadcast to all users in the same chat session
		ws.emit.chatSession(chatSessionId, 'chat:model-sync', {
			senderId: data.senderId,
			engine: data.engine,
			provider: data.provider,
			modelId: data.modelId,
			modelName: data.modelName
		});
	})

	// Collaborative account sync - broadcast account changes to other users in the same chat session
	.on('chat:account-sync', {
		data: t.Object({
			senderId: t.String(),
			chatSessionId: t.String(),
			accountId: t.Union([t.Number(), t.Null()]),
			accountName: t.Optional(t.Union([t.String(), t.Null()]))
		})
	}, ({ data, conn }) => {
		requireSessionAccess(conn, data.chatSessionId);
		const chatSessionId = data.chatSessionId;

		// Store latest account state on server for late joiners / refresh
		chatSessionAccountState.set(chatSessionId, {
			accountId: data.accountId,
			senderId: data.senderId
		});

		// Persist account to the session record in the database
		try {
			sessionQueries.updateAccount(chatSessionId, data.accountId, data.accountName ?? null);
		} catch (err) {
			debug.error('chat', 'Failed to persist account sync to DB:', err);
		}

		// Broadcast to all users in the same chat session
		ws.emit.chatSession(chatSessionId, 'chat:account-sync', {
			senderId: data.senderId,
			accountId: data.accountId,
			accountName: data.accountName ?? null
		});
	})

	// Collaborative reasoning-effort sync — broadcast + persist the per-session
	// reasoning/thinking level, mirroring chat:model-sync. Choosing a level is a
	// per-run choice like the model/account.
	.on('chat:reasoning-sync', {
		data: t.Object({
			senderId: t.String(),
			chatSessionId: t.String(),
			reasoningEffort: t.Union([t.String(), t.Null()])
		})
	}, ({ data, conn }) => {
		requireSessionAccess(conn, data.chatSessionId);
		const chatSessionId = data.chatSessionId;

		// Persist to the session record so refreshes and late joiners get it.
		try {
			sessionQueries.updateReasoning(chatSessionId, data.reasoningEffort);
		} catch (err) {
			debug.error('chat', 'Failed to persist reasoning sync to DB:', err);
		}

		// Broadcast to all users in the same chat session
		ws.emit.chatSession(chatSessionId, 'chat:reasoning-sync', {
			senderId: data.senderId,
			reasoningEffort: data.reasoningEffort
		});
	})

	// Collaborative profile sync — broadcast + persist the per-session active
	// profile, mirroring chat:model-sync. Non-admin: choosing a profile is a run
	// choice like the model, not an admin mutation of the profile itself.
	.on('chat:profile-sync', {
		data: t.Object({
			senderId: t.String(),
			chatSessionId: t.String(),
			profileId: t.Union([t.Number(), t.Null()])
		})
	}, ({ data, conn }) => {
		requireSessionAccess(conn, data.chatSessionId);
		try {
			sessionQueries.updateProfile(data.chatSessionId, data.profileId);
		} catch (err) {
			debug.error('chat', 'Failed to persist profile sync to DB:', err);
		}
		ws.emit.chatSession(data.chatSessionId, 'chat:profile-sync', {
			senderId: data.senderId,
			profileId: data.profileId
		});
	})

	// Event declarations
	.emit('chat:edit-mode', t.Object({
		senderId: t.String(),
		isEditing: t.Boolean(),
		messageId: t.Union([t.String(), t.Null()]),
		messageTimestamp: t.Union([t.String(), t.Null()])
	}))

	.emit('chat:input-sync', t.Object({
		text: t.String(),
		senderId: t.String(),
		attachments: t.Optional(t.Array(t.Object({
			id: t.String(),
			fileName: t.String(),
			type: t.String(),
			mediaType: t.String(),
			base64: t.String()
		}))),
		chatSessionId: t.Optional(t.String())
	}))

	.emit('chat:model-sync', t.Object({
		senderId: t.String(),
		engine: t.String(),
		provider: t.String(),
		modelId: t.String(),
		modelName: t.String()
	}))

	.emit('chat:account-sync', t.Object({
		senderId: t.String(),
		accountId: t.Union([t.Number(), t.Null()]),
		accountName: t.Union([t.String(), t.Null()])
	}))

	.emit('chat:profile-sync', t.Object({
		senderId: t.String(),
		profileId: t.Union([t.Number(), t.Null()])
	}))

	.emit('chat:reasoning-sync', t.Object({
		senderId: t.String(),
		reasoningEffort: t.Union([t.String(), t.Null()])
	}))

	.emit('chat:connection', t.Object({
		chatSessionId: t.String(),
		processId: t.String(),
		timestamp: t.String(),
		seq: t.Optional(t.Number())
	}))

	.emit('chat:message', t.Object({
		chatSessionId: t.String(),
		processId: t.String(),
		message: t.Any(), // SDKMessage
		usage: t.Optional(t.Any()),
		timestamp: t.String(),
		message_id: t.Optional(t.String()),
		parent_message_id: t.Optional(t.Union([t.String(), t.Null()])),
		sender_id: t.Optional(t.String()),
		sender_name: t.Optional(t.String()),
		engine: t.Optional(t.String()),
		seq: t.Optional(t.Number())
	}))

	.emit('chat:partial', t.Object({
		chatSessionId: t.String(),
		processId: t.String(),
		eventType: t.Union([
			t.Literal('start'),
			t.Literal('update'),
			t.Literal('end')
		]),
		partialText: t.String(),
		deltaText: t.String(),
		reasoning: t.Optional(t.Boolean()),
		timestamp: t.String(),
		seq: t.Optional(t.Number())
	}))

	.emit('chat:notification', t.Object({
		notification: t.Object({
			type: t.String(),
			title: t.String(),
			message: t.String(),
			icon: t.Optional(t.String())
		}),
		timestamp: t.String(),
		seq: t.Optional(t.Number())
	}))

	.emit('chat:rate_limit', t.Object({
		chatSessionId: t.String(),
		engine: t.String(),
		accountId: t.Number(),
		status: t.Union([t.Literal('allowed_warning'), t.Literal('rejected')]),
		utilization: t.Number(),
		resetsAt: t.Union([t.Number(), t.Null()]),
		rateLimitType: t.Union([
			t.Literal('five_hour'),
			t.Literal('seven_day'),
			t.Literal('seven_day_opus'),
			t.Literal('seven_day_sonnet'),
			t.Literal('overage'),
			t.Literal('seven_day_overage_included'),
			t.Null()
		]),
		timestamp: t.String(),
		seq: t.Optional(t.Number())
	}))

	.emit('chat:complete', t.Object({
		chatSessionId: t.String(),
		processId: t.String(),
		timestamp: t.String(),
		seq: t.Optional(t.Number())
	}))

	.emit('chat:error', t.Object({
		chatSessionId: t.String(),
		processId: t.String(),
		error: t.String(),
		timestamp: t.String(),
		seq: t.Optional(t.Number())
	}))

	.emit('chat:cancelled', t.Object({
		chatSessionId: t.String(),
		status: t.Literal('cancelled'),
		processId: t.Optional(t.String()),
		seq: t.Optional(t.Number())
	}))

	.emit('chat:messages-changed', t.Object({
		sessionId: t.String(),
		reason: t.String(),
		timestamp: t.String()
	}))

	// `projectName` / `sessionTitle` are resolved server-side and carried here
	// so the tab can word its local notification exactly like the background
	// push that replaces it. Optional because either row may be gone, and
	// because a tab left open across a server upgrade will not send them.
	.emit('chat:stream-finished', t.Object({
		projectId: t.String(),
		chatSessionId: t.String(),
		streamId: t.Optional(t.String()),
		status: t.Union([t.Literal('completed'), t.Literal('error'), t.Literal('cancelled')]),
		timestamp: t.String(),
		reason: t.Optional(t.String()),
		projectName: t.Optional(t.String()),
		sessionTitle: t.Optional(t.String())
	}))

	.emit('chat:waiting-input', t.Object({
		projectId: t.String(),
		chatSessionId: t.String(),
		toolUseId: t.String(),
		timestamp: t.String(),
		projectName: t.Optional(t.String()),
		sessionTitle: t.Optional(t.String())
	}))

	.emit('snapshot:captured', t.Object({
		projectId: t.String(),
		chatSessionId: t.String()
	}));