fix(streams): streaming correctness, reliability, and performance overhaul

.github/workflows/deploy-preview.yml

@@ -251,6 +251,7 @@ jobs:
           STRIPE_PRO_MONTHLY_PRICE_ID: ${{ secrets.STRIPE_PRO_MONTHLY_PRICE_ID }}
           STRIPE_PRO_YEARLY_PRICE_ID: ${{ secrets.STRIPE_PRO_YEARLY_PRICE_ID }}
           SLACK_BILLING_WEBHOOK_URL: ${{ secrets.SLACK_BILLING_WEBHOOK_URL }}
+          STREAMS_URL: ${{ env.STREAMS_URL }}
         run: |
           vercel pull --yes --environment=preview --token=$VERCEL_TOKEN
           vercel build --token=$VERCEL_TOKEN
@@ -295,7 +296,8 @@ jobs:
             --env STRIPE_WEBHOOK_SECRET=$STRIPE_WEBHOOK_SECRET \
             --env STRIPE_PRO_MONTHLY_PRICE_ID=$STRIPE_PRO_MONTHLY_PRICE_ID \
             --env STRIPE_PRO_YEARLY_PRICE_ID=$STRIPE_PRO_YEARLY_PRICE_ID \
-            --env SLACK_BILLING_WEBHOOK_URL=$SLACK_BILLING_WEBHOOK_URL)
+            --env SLACK_BILLING_WEBHOOK_URL=$SLACK_BILLING_WEBHOOK_URL \
+            --env STREAMS_URL=$STREAMS_URL)
           vercel alias $VERCEL_URL ${{ env.API_ALIAS }} --scope=$VERCEL_ORG_ID --token=$VERCEL_TOKEN
           echo "vercel_url=$VERCEL_URL" >> $GITHUB_OUTPUT
 

.github/workflows/deploy-production.yml

@@ -110,6 +110,7 @@ jobs:
           STRIPE_PRO_MONTHLY_PRICE_ID: ${{ secrets.STRIPE_PRO_MONTHLY_PRICE_ID }}
           STRIPE_PRO_YEARLY_PRICE_ID: ${{ secrets.STRIPE_PRO_YEARLY_PRICE_ID }}
           SLACK_BILLING_WEBHOOK_URL: ${{ secrets.SLACK_BILLING_WEBHOOK_URL }}
+          STREAMS_URL: ${{ secrets.STREAMS_URL }}
         run: |
           vercel pull --yes --environment=production --token=$VERCEL_TOKEN
           vercel build --prod --token=$VERCEL_TOKEN
@@ -154,7 +155,8 @@ jobs:
             --env STRIPE_WEBHOOK_SECRET=$STRIPE_WEBHOOK_SECRET \
             --env STRIPE_PRO_MONTHLY_PRICE_ID=$STRIPE_PRO_MONTHLY_PRICE_ID \
             --env STRIPE_PRO_YEARLY_PRICE_ID=$STRIPE_PRO_YEARLY_PRICE_ID \
-            --env SLACK_BILLING_WEBHOOK_URL=$SLACK_BILLING_WEBHOOK_URL
+            --env SLACK_BILLING_WEBHOOK_URL=$SLACK_BILLING_WEBHOOK_URL \
+            --env STREAMS_URL=$STREAMS_URL
 
   deploy-web:
     name: Deploy Web to Vercel

apps/desktop/src/lib/trpc/routers/ai-chat/index.ts

@@ -17,6 +17,12 @@ interface CommandEntry {
 	argumentHint: string;
 }
 
+const permissionModeSchema = z.enum([
+	"default",
+	"acceptEdits",
+	"bypassPermissions",
+]);
+
 function scanCustomCommands(cwd: string): CommandEntry[] {
 	const dirs = [
 		join(cwd, ".claude", "commands"),
@@ -79,7 +85,7 @@ export const createAiChatRouter = () => {
 					paneId: z.string().optional(),
 					tabId: z.string().optional(),
 					model: z.string().optional(),
-					permissionMode: z.string().optional(),
+					permissionMode: permissionModeSchema.optional(),
 				}),
 			)
 			.mutation(async ({ input }) => {
@@ -103,7 +109,7 @@ export const createAiChatRouter = () => {
 					paneId: z.string().optional(),
 					tabId: z.string().optional(),
 					model: z.string().optional(),
-					permissionMode: z.string().optional(),
+					permissionMode: permissionModeSchema.optional(),
 				}),
 			)
 			.mutation(async ({ input }) => {
@@ -151,10 +157,7 @@ export const createAiChatRouter = () => {
 					sessionId: z.string(),
 					maxThinkingTokens: z.number().nullable().optional(),
 					model: z.string().nullable().optional(),
-					permissionMode: z
-						.enum(["default", "acceptEdits", "bypassPermissions"])
-						.nullable()
-						.optional(),
+					permissionMode: permissionModeSchema.nullable().optional(),
 				}),
 			)
 			.mutation(async ({ input }) => {
@@ -175,8 +178,9 @@ export const createAiChatRouter = () => {
 				}),
 			)
 			.mutation(async ({ input }) => {
-				await chatSessionManager.updateSessionMeta(input.sessionId, {
-					title: input.title,
+				await chatSessionManager.updateSessionMeta({
+					sessionId: input.sessionId,
+					patch: { title: input.title },
 				});
 				return { success: true };
 			}),
@@ -229,10 +233,17 @@ export const createAiChatRouter = () => {
 			.input(z.object({ sessionId: z.string(), text: z.string() }))
 			.mutation(({ input }) => {
 				// Fire-and-forget: agent runs in background, errors surface via streamEvents
-				chatSessionManager.startAgent({
-					sessionId: input.sessionId,
-					prompt: input.text,
-				});
+				void chatSessionManager
+					.startAgent({
+						sessionId: input.sessionId,
+						prompt: input.text,
+					})
+					.catch((error: unknown) => {
+						console.error(
+							"[ai-chat/sendMessage] Failed to start agent:",
+							error,
+						);
+					});
 				return { success: true };
 			}),
 

apps/desktop/src/lib/trpc/routers/ai-chat/utils/session-manager/agent-execution.ts

@@ -0,0 +1,123 @@
+import { join } from "node:path";
+import {
+	createPermissionRequest,
+	executeAgent,
+	resolvePendingPermission,
+} from "@superset/agent";
+import { app } from "electron";
+import { buildClaudeEnv } from "../auth";
+import type { SessionStore } from "../session-store";
+import type { PermissionRequestEvent } from "./session-events";
+import type { ActiveSession } from "./session-types";
+
+function getClaudeBinaryPath(): string {
+	if (app.isPackaged) {
+		return join(process.resourcesPath, "bin", "claude");
+	}
+	const platform = process.platform;
+	const arch = process.arch;
+	return join(
+		app.getAppPath(),
+		"resources",
+		"bin",
+		`${platform}-${arch}`,
+		"claude",
+	);
+}
+
+export interface ResolvePermissionInput {
+	sessionId: string;
+	toolUseId: string;
+	approved: boolean;
+	updatedInput?: Record<string, unknown>;
+}
+
+export interface ExecuteAgentInput {
+	session: ActiveSession;
+	sessionId: string;
+	prompt: string;
+	abortController: AbortController;
+	onChunk: (chunk: unknown) => void;
+}
+
+interface AgentExecutionDeps {
+	store: SessionStore;
+	emitPermissionRequest: (event: PermissionRequestEvent) => void;
+}
+
+export class AgentExecution {
+	constructor(private readonly deps: AgentExecutionDeps) {}
+
+	async execute({
+		session,
+		sessionId,
+		prompt,
+		abortController,
+		onChunk,
+	}: ExecuteAgentInput): Promise<void> {
+		const agentEnv = buildClaudeEnv();
+
+		await executeAgent({
+			sessionId,
+			prompt,
+			cwd: session.cwd,
+			pathToClaudeCodeExecutable: getClaudeBinaryPath(),
+			env: agentEnv,
+			model: session.model,
+			permissionMode: session.permissionMode ?? "default",
+			maxThinkingTokens: session.maxThinkingTokens,
+			signal: abortController.signal,
+			onChunk,
+			onPermissionRequest: async (params) => {
+				this.deps.emitPermissionRequest({
+					type: "permission_request",
+					sessionId,
+					toolUseId: params.toolUseId,
+					toolName: params.toolName,
+					input: params.input,
+				});
+
+				return createPermissionRequest({
+					toolUseId: params.toolUseId,
+					signal: params.signal,
+				});
+			},
+			onEvent: (event) => {
+				if (event.type === "session_initialized") {
+					this.deps.store
+						.update(sessionId, {
+							providerSessionId: event.claudeSessionId,
+							lastActiveAt: Date.now(),
+						})
+						.catch((err: unknown) => {
+							console.error(
+								`[chat/session] Failed to update providerSessionId:`,
+								err,
+							);
+						});
+				}
+			},
+		});
+	}
+
+	resolvePermission({
+		sessionId,
+		toolUseId,
+		approved,
+		updatedInput,
+	}: ResolvePermissionInput): void {
+		const result = approved
+			? {
+					behavior: "allow" as const,
+					updatedInput: updatedInput ?? {},
+				}
+			: { behavior: "deny" as const, message: "User denied permission" };
+
+		const resolved = resolvePendingPermission({ toolUseId, result });
+		if (!resolved) {
+			console.warn(
+				`[chat/session] No pending permission for toolUseId=${toolUseId} in session ${sessionId}`,
+			);
+		}
+	}
+}

apps/desktop/src/lib/trpc/routers/ai-chat/utils/session-manager/agent-runner.ts

@@ -0,0 +1,137 @@
+import type { SessionStore } from "../session-store";
+import type { ResolvePermissionInput } from "./agent-execution";
+import { AgentExecution } from "./agent-execution";
+import { AgentStreamWriter } from "./agent-stream-writer";
+import type { ChunkBatcher } from "./chunk-batcher";
+import type { GenerationWatchdog } from "./generation-watchdog";
+import type { PermissionRequestEvent } from "./session-events";
+import type { ActiveSession, EnsureSessionReadyInput } from "./session-types";
+
+export type { ResolvePermissionInput } from "./agent-execution";
+
+export interface StartAgentInput {
+	sessionId: string;
+	prompt: string;
+}
+
+interface AgentRunnerDeps {
+	store: SessionStore;
+	sessions: Map<string, ActiveSession>;
+	runningAgents: Map<string, AbortController>;
+	proxyUrl: string;
+	emitSessionError: (params: { sessionId: string; error: string }) => void;
+	emitPermissionRequest: (event: PermissionRequestEvent) => void;
+	ensureSessionReady: (input: EnsureSessionReadyInput) => Promise<void>;
+}
+
+export class AgentRunner {
+	private readonly execution: AgentExecution;
+	private readonly streamWriter: AgentStreamWriter;
+
+	constructor(private readonly deps: AgentRunnerDeps) {
+		this.execution = new AgentExecution({
+			store: deps.store,
+			emitPermissionRequest: deps.emitPermissionRequest,
+		});
+		this.streamWriter = new AgentStreamWriter({
+			proxyUrl: deps.proxyUrl,
+			emitSessionError: deps.emitSessionError,
+			ensureSessionReady: deps.ensureSessionReady,
+			isSessionActive: (sessionId) => this.deps.sessions.has(sessionId),
+		});
+	}
+
+	private abortExistingAgent({ sessionId }: { sessionId: string }): void {
+		const existingController = this.deps.runningAgents.get(sessionId);
+		if (!existingController) return;
+		console.warn(`[chat/session] Aborting previous agent run for ${sessionId}`);
+		existingController.abort();
+		if (this.deps.runningAgents.get(sessionId) === existingController) {
+			this.deps.runningAgents.delete(sessionId);
+		}
+	}
+
+	async startAgent({ sessionId, prompt }: StartAgentInput): Promise<void> {
+		const session = this.deps.sessions.get(sessionId);
+		if (!session) {
+			console.error(
+				`[chat/session] Session ${sessionId} not found for startAgent`,
+			);
+			this.deps.emitSessionError({
+				sessionId,
+				error: "Session not active",
+			});
+			return;
+		}
+
+		this.abortExistingAgent({ sessionId });
+
+		const abortController = new AbortController();
+		this.deps.runningAgents.set(sessionId, abortController);
+
+		const messageId = crypto.randomUUID();
+		let headers: Record<string, string> | null = null;
+		let batcher: ChunkBatcher | null = null;
+		let watchdog: GenerationWatchdog | null = null;
+
+		try {
+			const prepared = await this.streamWriter.prepareStream({
+				sessionId,
+				session,
+				abortController,
+			});
+			headers = prepared.headers;
+			batcher = prepared.batcher;
+			watchdog = prepared.watchdog;
+
+			await this.execution.execute({
+				session,
+				sessionId,
+				prompt,
+				abortController,
+				onChunk: (chunk) => {
+					this.streamWriter.onAssistantChunk({
+						watchdog: watchdog as GenerationWatchdog,
+						batcher: batcher as ChunkBatcher,
+						messageId,
+						chunk,
+					});
+				},
+			});
+		} catch (error) {
+			const message = error instanceof Error ? error.message : String(error);
+			if (!abortController.signal.aborted) {
+				console.error(
+					`[chat/session] Agent execution failed for ${sessionId}:`,
+					message,
+				);
+				this.deps.emitSessionError({ sessionId, error: message });
+			} else if (watchdog?.wasTriggered) {
+				console.warn(
+					`[chat/session] Agent aborted by watchdog for ${sessionId}:`,
+					message,
+				);
+			}
+		} finally {
+			watchdog?.clear();
+			await this.streamWriter.drainChunkBatcher({
+				sessionId,
+				batcher,
+				abortController,
+			});
+			await this.streamWriter.finalizeGeneration({
+				sessionId,
+				session,
+				messageId,
+				headers,
+			});
+			if (this.deps.runningAgents.get(sessionId) === abortController) {
+				this.deps.runningAgents.delete(sessionId);
+			}
+		}
+	}
+
+	resolvePermission(input: ResolvePermissionInput): void {
+		this.execution.resolvePermission(input);
+	}
+}