From 9286ca67e96dc22b464b954dfb7983c90c6f1b7a Mon Sep 17 00:00:00 2001 From: Huarch Date: Wed, 22 Jul 2026 18:16:43 +0800 Subject: [PATCH] refactor: split chat route responsibilities --- src/routes/chat.ts | 1242 ++----------------------------- src/routes/chatRunState.ts | 22 + src/routes/chatSessionRoutes.ts | 426 +++++++++++ src/routes/chatStreamRoute.ts | 811 ++++++++++++++++++++ src/server.ts | 20 +- 5 files changed, 1341 insertions(+), 1180 deletions(-) create mode 100644 src/routes/chatRunState.ts create mode 100644 src/routes/chatSessionRoutes.ts create mode 100644 src/routes/chatStreamRoute.ts diff --git a/src/routes/chat.ts b/src/routes/chat.ts index d3d7f69..a945064 100644 --- a/src/routes/chat.ts +++ b/src/routes/chat.ts @@ -1,196 +1,52 @@ import { Router } from "express"; -import { - UI_MESSAGE_STREAM_HEADERS, - type UIMessage, - type UIMessageChunk, -} from "ai"; -import { z } from "zod"; -import { getLocalAgentContext } from "../context/localContext.js"; -import { - agentModelOptions, - isSupportedModel, - resolveDefaultModel, -} from "../chat/models.js"; -import { config } from "../config.js"; -import { type LearningOrchestrator } from "../learning/orchestrator.js"; -import { type SessionTranscriptStore } from "../sessions/transcriptStore.js"; -import { logger } from "../logger.js"; -import { MemoryStore } from "../memory/store.js"; -import { type SessionUiStateStore } from "../sessions/uiStateStore.js"; -import { type SessionMetadataStore } from "../sessions/metadataStore.js"; -import { type ResultReferenceResolver } from "../results/resolver.js"; -import { - type OpencodeRuntimeAdapter, -} from "../runtime/opencode.js"; -import { getRuntimeSessionContext, setRuntimeSessionContext } from "../runtime/sessionContext.js"; +import { agentModelOptions, resolveDefaultModel } from "../chat/models.js"; import { type ChatSessionBridge } from "../chat/sessionBridge.js"; -import { type SessionRecord } from "../sessions/metadataStore.js"; -import { uiRegistryResponse } from "../uiEnvelope/registry.js"; -import { FrontendActionCoordinator } from "../frontendAction/coordinator.js"; +import { config } from "../config.js"; +import { type FrontendActionCoordinator } from "../frontendAction/coordinator.js"; import { frontendActionRegistryResponse } from "../frontendAction/registry.js"; -import type { UIEnvelopePayload } from "../uiEnvelope/types.js"; -import { toActorKey, toProjectKey } from "../utils/fileStore.js"; -import { - buildForkedSessionUiState, - buildPromptWithLearningContext, - extractLatestFrontendTurn, - generateSessionTitle, - resolveRequestedStreamSessionId, - shouldGenerateSessionTitle, - shouldRestoreConversationForRuntime, -} from "./chatSession.js"; +import { type LearningOrchestrator } from "../learning/orchestrator.js"; +import { type MemoryStore } from "../memory/store.js"; +import { type ResultReferenceResolver } from "../results/resolver.js"; +import { type OpencodeRuntimeAdapter } from "../runtime/opencode.js"; +import { type SessionMetadataStore } from "../sessions/metadataStore.js"; +import { type SessionTranscriptStore } from "../sessions/transcriptStore.js"; +import { type SessionUiStateStore } from "../sessions/uiStateStore.js"; +import { uiRegistryResponse } from "../uiEnvelope/registry.js"; import { registerChatAuxiliaryRoutes } from "./chatAuxiliaryRoutes.js"; import { registerChatInteractionRoutes } from "./chatInteractionRoutes.js"; -import { - collectTextContent, - type PermissionRequestPayload, - type QuestionRequestPayload, - streamPromptResponse, - type TodoUpdatePayload, -} from "./chatStream.js"; -import { - type ActiveRun, - type RunStatus, - type StreamSubscriber, - appendBackendToolArtifact, - appendBackendUiEnvelope, - cancelBackendTodos, - completeBackendProgress, - createInitialStreamingMessages, - isObjectRecord, - toFrontendPermission, - toPermissionStatus, - updateLastAssistantMessage, - updateLastAssistantPermission, - updateLastAssistantQuestion, - upsertBackendProgress, - upsertBackendQuestion, - upsertBackendTodoUpdate, -} from "./chatUiState.js"; +import { createChatRunState } from "./chatRunState.js"; +import { registerChatSessionRoutes } from "./chatSessionRoutes.js"; +import { registerChatStreamRoute } from "./chatStreamRoute.js"; -const payloadSchema = z.object({ - id: z.string().max(128).optional(), - messages: z.array(z.unknown()).max(200).optional(), - message: z.string().min(1).max(10000).optional(), - session_id: z.string().max(128).optional(), - model: z.string().refine(isSupportedModel, { - message: "unsupported model", - }).optional(), - approval_mode: z.enum(["request", "always"]).optional().default("request"), - capabilities: z.array(z.string().max(64)).max(20).optional().default([]), -}); - -const createSessionPayloadSchema = z.object({ - session_id: z.string().max(128).optional(), - parent_session_id: z.string().max(128).optional(), -}); - -const forkPayloadSchema = z.object({ - session_id: z.string().max(128).optional(), - keep_message_count: z.coerce.number().int().min(0), -}); - -const activeRuns = new Map(); -const lastRunStatuses = new Map(); - -const toSessionUiStateContext = (sessionRecord: SessionRecord) => ({ - sessionId: sessionRecord.sessionId, -}); - -const getSessionRunStatus = (sessionId: string) => - activeRuns.get(sessionId)?.status ?? lastRunStatuses.get(sessionId); - -const runtimeHasConversation = async ( - runtime: OpencodeRuntimeAdapter, - sessionId: string, -) => { - const messages = await runtime.messages(sessionId, 1); - return messages.some( - (message) => - message.info.role === "user" || message.info.role === "assistant", - ); +export type BuildChatRouterOptions = { + frontendActionCoordinator: FrontendActionCoordinator; + learningOrchestrator: LearningOrchestrator; + memoryStore: MemoryStore; + resultReferenceResolver: ResultReferenceResolver; + runtime: OpencodeRuntimeAdapter; + sessionBridge: ChatSessionBridge; + sessionMetadataStore: SessionMetadataStore; + sessionTranscriptStore: SessionTranscriptStore; + sessionUiStateStore: SessionUiStateStore; }; -const isUiMessage = (value: unknown): value is UIMessage => - isObjectRecord(value) && - typeof value.id === "string" && - (value.role === "system" || value.role === "user" || value.role === "assistant") && - Array.isArray(value.parts); +const runState = createChatRunState(); -const readUiMessageText = (message: UIMessage | undefined) => - message?.parts - .flatMap((part) => - isObjectRecord(part) && part.type === "text" && typeof part.text === "string" - ? [part.text] - : [], - ) - .join("") - .trim() ?? ""; - -const readLatestUserText = (messages: UIMessage[]) => { - for (let index = messages.length - 1; index >= 0; index -= 1) { - const message = messages[index]; - if (message.role !== "user") { - continue; - } - const text = readUiMessageText(message); - if (text) { - return text; - } - } - return ""; -}; - -const toUiMessages = (messages: unknown[] | undefined) => - Array.isArray(messages) ? messages.filter(isUiMessage) : []; - -const toLegacyMessages = (messages: UIMessage[]) => - messages.flatMap((message) => { - const content = readUiMessageText(message); - if (!content || (message.role !== "user" && message.role !== "assistant")) { - return []; - } - return [ - { - id: message.id, - role: message.role, - content, - }, - ]; - }); - -const createFallbackUserUiMessage = (content: string): UIMessage => ({ - id: `user-${Date.now().toString(36)}`, - role: "user", - parts: [{ type: "text", text: content }], -}); - -const pipeUiChunk = ( - res: { - write: (chunk: string) => void; - writableEnded?: boolean; - destroyed?: boolean; - }, - chunk: UIMessageChunk, -) => { - if (!res.writableEnded && !res.destroyed) { - res.write(`data: ${JSON.stringify(chunk)}\n\n`); - } -}; - -export const buildChatRouter = ( - sessionBridge: ChatSessionBridge, - runtime: OpencodeRuntimeAdapter, - sessionMetadataStore: SessionMetadataStore, - sessionUiStateStore: SessionUiStateStore, - memoryStore: MemoryStore, - sessionTranscriptStore: SessionTranscriptStore, - learningOrchestrator: LearningOrchestrator, - resultReferenceResolver: ResultReferenceResolver, - frontendActionCoordinator: FrontendActionCoordinator, -) => { +export const buildChatRouter = (options: BuildChatRouterOptions) => { + const { + frontendActionCoordinator, + learningOrchestrator, + memoryStore, + resultReferenceResolver, + runtime, + sessionBridge, + sessionMetadataStore, + sessionTranscriptStore, + sessionUiStateStore, + } = options; const chatRouter = Router(); + const { activeRuns, lastRunStatuses } = runState; chatRouter.get("/models", (_req, res) => { res.json({ @@ -215,272 +71,35 @@ export const buildChatRouter = ( } try { const outcome = frontendActionCoordinator.submitResult(sessionId, req.body); - res.json({ accepted: true, duplicate: outcome.duplicate, result: outcome.result }); + res.json({ + accepted: true, + duplicate: outcome.duplicate, + result: outcome.result, + }); } catch (error) { - const message = error instanceof Error ? error.message : "frontend action result rejected"; - const status = typeof error === "object" && error && "status" in error && typeof error.status === "number" ? error.status : 400; + const message = + error instanceof Error + ? error.message + : "frontend action result rejected"; + const status = + typeof error === "object" && + error && + "status" in error && + typeof error.status === "number" + ? error.status + : 400; res.status(status).json({ message }); } }); - chatRouter.post("/session", async (req, res) => { - const parsed = createSessionPayloadSchema.safeParse(req.body ?? {}); - if (!parsed.success) { - res.status(400).json({ - message: "invalid request payload", - detail: parsed.error.flatten(), - }); - return; - } - - const authContext = getLocalAgentContext(req); - const projectId = authContext.projectId; - const userId = authContext.userId; - const actorKey = toActorKey(userId); - const projectKey = toProjectKey(projectId); - const requestedSessionId = parsed.data.session_id?.trim(); - let sessionId = requestedSessionId; - if (!sessionId) { - try { - sessionId = (await runtime.createSession()).id; - } catch (error) { - sessionId = sessionBridge.createClientSessionId(); - logger.warn( - { err: error, sessionId }, - "runtime unavailable while creating chat session; using local degraded session", - ); - } - } - - const { record, created } = await sessionMetadataStore.ensure({ - actorKey, - parentSessionId: parsed.data.parent_session_id, - projectId, - projectKey, - sessionId, - userId, - }); - - res.status(created ? 201 : 200).json({ - session_id: record.sessionId, - created_at: record.createdAt, - updated_at: record.updatedAt, - status: record.status, - title: record.title, - parent_session_id: record.parentSessionId, - }); - }); - - chatRouter.get("/sessions", async (req, res) => { - const authContext = getLocalAgentContext(req); - const projectId = authContext.projectId; - const userId = authContext.userId; - const actorKey = toActorKey(userId); - const projectKey = toProjectKey(projectId); - const records = await sessionMetadataStore.list({ - actorKey, - projectId, - projectKey, - userId, - }); - res.json({ - sessions: records.map((record) => ({ - id: record.sessionId, - title: record.title ?? "新对话", - created_at: record.createdAt, - updated_at: record.updatedAt, - status: record.status, - parent_session_id: record.parentSessionId, - is_streaming: activeRuns.get(record.sessionId)?.status === "running", - run_status: getSessionRunStatus(record.sessionId), - })), - }); - }); - - chatRouter.get("/session/:session_id", async (req, res) => { - const sessionId = req.params.session_id?.trim(); - const authContext = getLocalAgentContext(req); - const projectId = authContext.projectId; - const userId = authContext.userId; - const actorKey = toActorKey(userId); - const projectKey = toProjectKey(projectId); - if (!sessionId) { - res.status(400).json({ message: "session_id is required" }); - return; - } - - const sessionRecord = await sessionMetadataStore.get( - { - actorKey, - projectId, - projectKey, - userId, - }, - sessionId, - ); - if (!sessionRecord) { - res.status(404).json({ message: "session not found" }); - return; - } - - const state = await sessionUiStateStore.read( - toSessionUiStateContext(sessionRecord), - ); - res.json({ - id: sessionRecord.sessionId, - title: sessionRecord.title ?? "新对话", - is_title_manually_edited: state?.isTitleManuallyEdited ?? false, - created_at: sessionRecord.createdAt, - updated_at: sessionRecord.updatedAt, - status: sessionRecord.status, - session_id: sessionRecord.sessionId, - messages: state?.messages ?? [], - parent_session_id: sessionRecord.parentSessionId, - is_streaming: activeRuns.get(sessionRecord.sessionId)?.status === "running", - run_status: getSessionRunStatus(sessionRecord.sessionId), - }); - }); - - chatRouter.get("/session/:session_id/stream", async (req, res) => { - const sessionId = req.params.session_id?.trim(); - const authContext = getLocalAgentContext(req); - const projectId = authContext.projectId; - const userId = authContext.userId; - const actorKey = toActorKey(userId); - const projectKey = toProjectKey(projectId); - if (!sessionId) { - res.status(400).json({ message: "session_id is required" }); - return; - } - - const sessionRecord = await sessionMetadataStore.get( - { actorKey, projectId, projectKey, userId }, - sessionId, - ); - if (!sessionRecord) { - res.status(404).json({ message: "session not found" }); - return; - } - - res.status(200); - res.setHeader("Content-Type", "text/event-stream; charset=utf-8"); - res.setHeader("Cache-Control", "no-cache"); - res.setHeader("Connection", "keep-alive"); - res.setHeader("X-Accel-Buffering", "no"); - res.flushHeaders?.(); - - const run = activeRuns.get(sessionRecord.sessionId); - const state = await sessionUiStateStore.read(toSessionUiStateContext(sessionRecord)); - res.write( - toSse("state", { - session_id: sessionRecord.sessionId, - messages: state?.messages ?? run?.messages ?? [], - is_streaming: run?.status === "running", - run_status: getSessionRunStatus(sessionRecord.sessionId) ?? "completed", - }), - ); - - if (!run || run.status !== "running") { - res.end(); - return; - } - - const subscriber: StreamSubscriber = { - write: (event, data) => { - if (!res.writableEnded && !res.destroyed) { - res.write(toSse(event, data)); - } - }, - close: () => { - if (!res.writableEnded && !res.destroyed) { - res.end(); - } - }, - }; - run.subscribers.add(subscriber); - - const cleanup = () => { - run.subscribers.delete(subscriber); - }; - req.on("close", cleanup); - res.on("close", cleanup); - }); - - chatRouter.patch("/session/:session_id/title", async (req, res) => { - const sessionId = req.params.session_id?.trim(); - const title = - typeof req.body?.title === "string" ? req.body.title.trim() : ""; - const isTitleManuallyEdited = - typeof req.body?.is_title_manually_edited === "boolean" - ? req.body.is_title_manually_edited - : undefined; - const authContext = getLocalAgentContext(req); - const projectId = authContext.projectId; - const userId = authContext.userId; - const actorKey = toActorKey(userId); - const projectKey = toProjectKey(projectId); - if (!sessionId || !title) { - res.status(400).json({ message: "session_id and title are required" }); - return; - } - const sessionRecord = await sessionMetadataStore.get( - { actorKey, projectId, projectKey, userId }, - sessionId, - ); - if (!sessionRecord) { - res.status(404).json({ message: "session not found" }); - return; - } - const nextSessionRecord = await sessionMetadataStore.touch(sessionRecord, { title }); - const state = await sessionUiStateStore.read( - toSessionUiStateContext(nextSessionRecord), - ); - if (state) { - await sessionUiStateStore.write( - toSessionUiStateContext(nextSessionRecord), - { - ...state, - isTitleManuallyEdited: - isTitleManuallyEdited ?? state.isTitleManuallyEdited, - }, - ); - } - res.json({ - id: nextSessionRecord.sessionId, - title: nextSessionRecord.title, - updated_at: nextSessionRecord.updatedAt, - }); - }); - - chatRouter.delete("/session/:session_id", async (req, res) => { - const sessionId = req.params.session_id?.trim(); - const authContext = getLocalAgentContext(req); - const projectId = authContext.projectId; - const userId = authContext.userId; - const actorKey = toActorKey(userId); - const projectKey = toProjectKey(projectId); - if (!sessionId) { - res.status(400).json({ message: "session_id is required" }); - return; - } - const sessionRecord = await sessionMetadataStore.get( - { actorKey, projectId, projectKey, userId }, - sessionId, - ); - if (!sessionRecord) { - res.status(204).end(); - return; - } - await sessionUiStateStore.remove(toSessionUiStateContext(sessionRecord)); - await sessionBridge.deleteSession({ - clientSessionId: sessionRecord.sessionId, - sessionId: sessionRecord.sessionId, - }); - frontendActionCoordinator.cancelSession(sessionRecord.sessionId); - activeRuns.delete(sessionRecord.sessionId); - lastRunStatuses.delete(sessionRecord.sessionId); - await sessionMetadataStore.remove(sessionRecord); - res.status(204).end(); + registerChatSessionRoutes(chatRouter, { + frontendActionCoordinator, + runState, + runtime, + sessionBridge, + sessionMetadataStore, + sessionTranscriptStore, + sessionUiStateStore, }); registerChatAuxiliaryRoutes(chatRouter, { @@ -500,734 +119,17 @@ export const buildChatRouter = ( sessionUiStateStore, }); - chatRouter.post("/fork", async (req, res) => { - const parsed = forkPayloadSchema.safeParse(req.body); - if (!parsed.success) { - res.status(400).json({ - message: "invalid request payload", - detail: parsed.error.flatten(), - }); - return; - } - - try { - const authContext = getLocalAgentContext(req); - const projectId = authContext.projectId; - const traceId = req.header("x-trace-id") ?? undefined; - const userId = authContext.userId; - - const actorKey = toActorKey(userId); - const projectKey = toProjectKey(projectId); - const sourceSessionId = parsed.data.session_id?.trim(); - const sourceSessionRecord = sourceSessionId - ? await sessionMetadataStore.get( - { - actorKey, - projectId, - projectKey, - userId, - }, - sourceSessionId, - ) - : null; - const forkSession = await runtime.createSession(); - const { record: targetSessionRecord } = await sessionMetadataStore.ensure({ - actorKey, - parentSessionId: sourceSessionId, - projectId, - projectKey, - sessionId: forkSession.id, - userId, - }); - const nextSessionId = targetSessionRecord.sessionId; - - if (sourceSessionId) { - await sessionTranscriptStore.cloneThread( - { - actorKey, - clientSessionId: sourceSessionId, - projectKey, - sessionId: sourceSessionId, - }, - { - actorKey, - clientSessionId: nextSessionId, - projectKey, - sessionId: nextSessionId, - }, - parsed.data.keep_message_count, - ); - } - const sourceState = sourceSessionRecord - ? await sessionUiStateStore.read(toSessionUiStateContext(sourceSessionRecord)) - : null; - const forkTitle = sourceSessionRecord?.title - ? `${sourceSessionRecord.title} 副本` - : "新对话副本"; - const titledTargetSessionRecord = await sessionMetadataStore.touch( - targetSessionRecord, - { title: forkTitle }, - ); - await sessionUiStateStore.write( - toSessionUiStateContext(titledTargetSessionRecord), - buildForkedSessionUiState(sourceState, { - keepMessageCount: parsed.data.keep_message_count, - targetSessionId: nextSessionId, - }), - ); - - logger.info( - { - sourceSessionId: parsed.data.session_id, - sessionId: nextSessionId, - traceId, - projectId, - keepMessageCount: parsed.data.keep_message_count, - }, - "forked chat session", - ); - - res.status(200).json({ - session_id: nextSessionId, - }); - } catch (error) { - const detail = error instanceof Error ? error.message : String(error); - logger.error({ err: error }, "chat fork failed"); - res.status(500).json({ - message: "chat fork failed", - detail, - }); - } - }); - - chatRouter.post("/stream", async (req, res) => { - const parsed = payloadSchema.safeParse(req.body); - if (!parsed.success) { - res.status(400).json({ - message: "invalid request payload", - detail: parsed.error.flatten(), - }); - return; - } - - try { - const authContext = getLocalAgentContext(req); - const projectId = authContext.projectId; - const traceId = req.header("x-trace-id") ?? undefined; - const userId = authContext.userId; - const actorKey = toActorKey(userId); - const projectKey = toProjectKey(projectId); - const requestUiMessages = toUiMessages(parsed.data.messages); - const promptText = - parsed.data.message?.trim() || readLatestUserText(requestUiMessages); - if (!promptText) { - res.status(400).json({ - message: "message or messages with latest user text is required", - }); - return; - } - const requestedSessionId = resolveRequestedStreamSessionId({ - sessionId: parsed.data.session_id, - promptText, - createClientSessionId: () => sessionBridge.createClientSessionId(), - }); - const existingSessionRecord = requestedSessionId - ? await sessionMetadataStore.get( - { actorKey, projectId, projectKey, userId }, - requestedSessionId, - ) - : null; - const hadExistingRuntimeSession = Boolean(existingSessionRecord); - - const { binding, requestContext, created } = await sessionBridge.resolve({ - sessionId: requestedSessionId, - network: authContext.network, - projectId, - traceId, - userId, - }); - setRuntimeSessionContext({ - ...(getRuntimeSessionContext(binding.sessionId)!), - capabilities: parsed.data.capabilities, - }); - const { record: ensuredSessionRecord, created: sessionCreated } = - await sessionMetadataStore.ensure({ - actorKey, - projectId, - projectKey, - sessionId: binding.sessionId, - userId, - }); - const activeSessionRecord = await sessionMetadataStore.touch(ensuredSessionRecord); - const hasRuntimeConversation = hadExistingRuntimeSession - ? await runtimeHasConversation(runtime, binding.sessionId) - : false; - const shouldRestoreConversation = shouldRestoreConversationForRuntime({ - hadExistingSessionRecord: hadExistingRuntimeSession, - runtimeHasConversation: hasRuntimeConversation, - }); - const historyContext = { - actorKey: requestContext.actorKey, - clientSessionId: requestContext.clientSessionId, - projectKey: requestContext.projectKey, - sessionId: requestContext.clientSessionId, - }; - const initialSessionState = await sessionUiStateStore.read( - toSessionUiStateContext(activeSessionRecord), - ); - const persistedUiMessages = toUiMessages(initialSessionState?.messages); - const requestMessages = - requestUiMessages.length > 0 - ? requestUiMessages - : [ - ...persistedUiMessages, - createFallbackUserUiMessage(promptText), - ]; - let latestUiMessages = requestMessages; - const baseMessages = toLegacyMessages(requestMessages); - if (activeRuns.get(activeSessionRecord.sessionId)?.status === "running") { - res.status(409).json({ - message: "session is already streaming", - session_id: activeSessionRecord.sessionId, - }); - return; - } - const recentTurns = await sessionTranscriptStore.getRecentTurns(historyContext, 8); - - logger.info( - { - clientSessionId: requestContext.clientSessionId, - sessionId: binding.sessionId, - created: created || sessionCreated, - model: parsed.data.model, - approvalMode: parsed.data.approval_mode, - traceId: requestContext.traceId, - projectId: requestContext.projectId, - }, - "processing chat request", - ); - - res.status(200); - for (const [header, value] of Object.entries(UI_MESSAGE_STREAM_HEADERS)) { - res.setHeader(header, value); - } - res.setHeader("X-Accel-Buffering", "no"); - res.flushHeaders?.(); - - const clientSessionId = requestContext.clientSessionId; - let streamClosed = false; - const abortController = new AbortController(); - sessionBridge.registerAbortController(clientSessionId, abortController); - const initialMessages = createInitialStreamingMessages( - baseMessages, - promptText, - ); - const activeRun: ActiveRun = { - clientSessionId, - controller: abortController, - messages: initialMessages, - pendingPermissions: new Map(), - pendingQuestions: new Map(), - status: "running", - subscribers: new Set(), - }; - activeRuns.set(clientSessionId, activeRun); - lastRunStatuses.set(clientSessionId, "running"); - const sessionUiStateContext = toSessionUiStateContext(activeSessionRecord); - const assistantUiMessageId = `assistant-${Date.now().toString(36)}`; - const textPartId = "answer"; - const assistantParts: Record[] = []; - let responseText = ""; - let textStarted = false; - let textEnded = false; - let uiFinished = false; - const syncLatestUiMessages = () => { - latestUiMessages = [ - ...requestMessages, - { - id: assistantUiMessageId, - role: "assistant", - parts: assistantParts.map((part) => ({ ...part })), - } as UIMessage, - ]; - }; - const writeUiChunk = (chunk: UIMessageChunk) => { - pipeUiChunk(res, chunk); - }; - const writeTextDelta = (delta: string) => { - if (!delta || uiFinished) { - return; - } - if (!textStarted) { - textStarted = true; - assistantParts.push({ - type: "text", - text: "", - state: "streaming", - }); - writeUiChunk({ type: "text-start", id: textPartId }); - } - responseText += delta; - const textPart = assistantParts.find( - (part) => part.type === "text", - ); - if (textPart) { - textPart.text = responseText; - textPart.state = "streaming"; - } - writeUiChunk({ type: "text-delta", id: textPartId, delta }); - writeUiChunk({ - type: "data-stream_token", - data: { - session_id: clientSessionId, - message_id: assistantUiMessageId, - content: delta, - }, - transient: true, - } as UIMessageChunk); - syncLatestUiMessages(); - }; - const writeDataPart = (event: string, data: Record) => { - if (uiFinished) { - return; - } - const type = `data-${event}`; - const id = - typeof data.id === "string" - ? data.id - : typeof data.request_id === "string" - ? data.request_id - : typeof data.envelope_id === "string" - ? data.envelope_id - : undefined; - const existingIndex = - id === undefined - ? -1 - : assistantParts.findIndex( - (part) => part.type === type && part.id === id, - ); - const part = id ? { type, id, data } : { type, data }; - if (existingIndex >= 0) { - assistantParts[existingIndex] = part; - } else { - assistantParts.push(part); - } - writeUiChunk(part as UIMessageChunk); - syncLatestUiMessages(); - }; - const finishUiMessage = (finishReason: "stop" | "error" = "stop") => { - if (uiFinished) { - return; - } - if (!textStarted && assistantParts.length === 0) { - writeTextDelta("Agent 已完成处理。"); - } - if (textStarted && !textEnded) { - textEnded = true; - const textPart = assistantParts.find( - (part) => part.type === "text", - ); - if (textPart) { - textPart.state = "done"; - } - writeUiChunk({ type: "text-end", id: textPartId }); - } - syncLatestUiMessages(); - writeUiChunk({ type: "finish", finishReason }); - uiFinished = true; - }; - writeUiChunk({ type: "start", messageId: assistantUiMessageId }); - let persistQueue = sessionUiStateStore.write(sessionUiStateContext, { - sessionId: activeSessionRecord.sessionId, - isTitleManuallyEdited: initialSessionState?.isTitleManuallyEdited ?? false, - messages: latestUiMessages, - }); - const queueSessionUiStatePersist = () => { - const snapshot = { - sessionId: activeSessionRecord.sessionId, - isTitleManuallyEdited: initialSessionState?.isTitleManuallyEdited ?? false, - messages: latestUiMessages, - }; - persistQueue = persistQueue - .catch((error) => { - logger.warn( - { err: error, sessionId: clientSessionId }, - "failed to persist previous chat stream state", - ); - }) - .then(() => sessionUiStateStore.write(sessionUiStateContext, snapshot)); - return persistQueue; - }; - const primarySubscriber: StreamSubscriber = { - write: (event, data) => { - if (streamClosed || res.writableEnded || res.destroyed) { - return; - } - if (event === "token") { - writeTextDelta(typeof data.content === "string" ? data.content : ""); - return; - } - if (event === "done") { - finishUiMessage("stop"); - return; - } - if (event === "error") { - writeUiChunk({ - type: "error", - errorText: - typeof data.message === "string" - ? data.message - : "Agent stream failed", - }); - finishUiMessage("error"); - return; - } - writeDataPart(event, data); - }, - close: () => { - if (!res.writableEnded && !res.destroyed) { - res.end(); - } - }, - }; - activeRun.subscribers.add(primarySubscriber); - const handleClientClose = () => { - streamClosed = true; - activeRun.subscribers.delete(primarySubscriber); - }; - - req.on("close", handleClientClose); - res.on("close", handleClientClose); - - const publish = (event: string, data: Record) => { - const subscriberData = - event === "token" - ? { - ...data, - message_id: - typeof data.message_id === "string" - ? data.message_id - : assistantUiMessageId, - } - : data; - if (event === "token") { - activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ - ...message, - content: `${typeof message.content === "string" ? message.content : ""}${typeof data.content === "string" ? data.content : ""}`, - isError: false, - })); - } else if (event === "progress") { - activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ - ...message, - progress: upsertBackendProgress(message.progress, data), - })); - } else if (event === "done") { - activeRun.status = "completed"; - lastRunStatuses.set(clientSessionId, "completed"); - activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ - ...message, - content: - typeof message.content === "string" && message.content.trim() - ? message.content - : "Agent 已完成处理，但没有生成文本回答。请查看过程记录，或换个更具体的问题重试。", - progress: completeBackendProgress(message.progress), - })); - } else if (event === "error") { - activeRun.status = activeRun.status === "aborted" ? "aborted" : "error"; - lastRunStatuses.set(clientSessionId, activeRun.status); - activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ - ...message, - content: - typeof message.content === "string" && message.content.trim() - ? message.content - : `⚠️ **错误：** ${typeof data.message === "string" ? data.message : "unknown error"}`, - isError: true, - progress: completeBackendProgress(message.progress), - todos: cancelBackendTodos(message.todos), - })); - } else if (event === "permission_request") { - const payload = data as PermissionRequestPayload; - activeRun.pendingPermissions.set(payload.request_id, payload); - activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ - ...message, - permissions: [ - ...(Array.isArray(message.permissions) ? message.permissions : []), - toFrontendPermission(payload), - ], - })); - } else if (event === "permission_response") { - const requestId = - typeof data.request_id === "string" ? data.request_id : undefined; - const reply = - data.reply === "once" || data.reply === "always" || data.reply === "reject" - ? data.reply - : undefined; - if (requestId && reply) { - activeRun.pendingPermissions.delete(requestId); - activeRun.messages = updateLastAssistantPermission( - activeRun.messages, - requestId, - (permission) => ({ - ...permission, - status: toPermissionStatus(reply), - repliedAt: Date.now(), - }), - ); - } - } else if (event === "question_request") { - const payload = data as QuestionRequestPayload; - let shouldTrackQuestion = true; - if (payload.tool?.callID) { - if (payload.request_id !== payload.tool.callID) { - activeRun.pendingQuestions.delete(payload.tool.callID); - } else { - const hasActionableQuestion = [...activeRun.pendingQuestions.values()].some( - (question) => - question.tool?.callID === payload.tool?.callID && - question.request_id !== payload.tool?.callID, - ); - if (hasActionableQuestion) { - activeRun.messages = updateLastAssistantMessage( - activeRun.messages, - (message) => ({ - ...message, - questions: upsertBackendQuestion(message.questions, payload), - }), - ); - shouldTrackQuestion = false; - } - } - } - if (shouldTrackQuestion) { - activeRun.pendingQuestions.set(payload.request_id, payload); - activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ - ...message, - questions: upsertBackendQuestion(message.questions, payload), - })); - } - } else if (event === "question_response") { - const requestId = - typeof data.request_id === "string" ? data.request_id : undefined; - if (requestId) { - activeRun.pendingQuestions.delete(requestId); - activeRun.messages = updateLastAssistantQuestion( - activeRun.messages, - requestId, - (question) => ({ - ...question, - status: data.rejected === true ? "rejected" : "answered", - repliedAt: Date.now(), - answers: Array.isArray(data.answers) ? data.answers : question.answers, - error: undefined, - }), - ); - } - } else if (event === "todo_update") { - const payload = data as TodoUpdatePayload; - activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ - ...message, - todos: upsertBackendTodoUpdate(message.todos, payload), - })); - } else if (event === "tool_call") { - activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ - ...message, - artifacts: appendBackendToolArtifact(message.artifacts, data), - })); - } else if (event === "ui_envelope") { - const payload = data as UIEnvelopePayload; - activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ - ...message, - uiEnvelopes: appendBackendUiEnvelope(message.uiEnvelopes, payload), - })); - } - - for (const subscriber of activeRun.subscribers) { - subscriber.write(event, subscriberData); - } - void queueSessionUiStatePersist().catch((error) => { - logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist chat stream state"); - }); - }; - - const unregisterFrontendActionSink = parsed.data.capabilities.includes("frontend-action@1") - ? frontendActionCoordinator.registerSink(clientSessionId, (request) => publish("frontend_action", request as unknown as Record)) - : undefined; - - try { - const preparedMessage = await buildPromptWithLearningContext( - memoryStore, - requestContext.actorKey, - requestContext.projectKey, - { - recentTurns, - persistedMessages: baseMessages, - message: promptText, - restoreConversation: shouldRestoreConversation, - }, - ); - const streamResult = await streamPromptResponse({ - runtime, - sessionId: binding.sessionId, - clientSessionId, - message: preparedMessage, - model: parsed.data.model, - approvalMode: parsed.data.approval_mode, - traceId: requestContext.traceId, - projectId: requestContext.projectId, - signal: abortController.signal, - suppressLegacyFrontendActions: parsed.data.capabilities.includes("frontend-action@1"), - write: (event, data) => { - publish(event, data); - }, - }); - await persistQueue.catch((error) => { - logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist chat stream state"); - }); - - if (!streamResult.aborted && !streamResult.failed) { - const messages = await runtime.messages(binding.sessionId, 60); - const assistantMessage = [...messages] - .reverse() - .find((message) => message.info.role === "assistant"); - const assistantText = collectTextContent(assistantMessage?.parts ?? []); - const latestSessionRecord = - (await sessionMetadataStore.get( - { actorKey, projectId, projectKey, userId }, - activeSessionRecord.sessionId, - )) ?? activeSessionRecord; - const latestSessionState = await sessionUiStateStore.read( - toSessionUiStateContext(latestSessionRecord), - ); - const existingSessionTitle = latestSessionRecord.title; - let sessionTitle = existingSessionTitle; - const shouldGenerateTitle = shouldGenerateSessionTitle({ - recentTurnCount: recentTurns.length, - isTitleManuallyEdited: - latestSessionState?.isTitleManuallyEdited ?? false, - }); - if (shouldGenerateTitle) { - sessionTitle = await generateSessionTitle(runtime, { - sessionId: binding.sessionId, - latestAssistantMessage: assistantText, - latestUserMessage: promptText, - fallbackTitle: existingSessionTitle, - }); - } - await sessionMetadataStore.touch(latestSessionRecord, { - ...(sessionTitle && sessionTitle !== existingSessionTitle - ? { title: sessionTitle } - : {}), - }); - if ( - shouldGenerateTitle && - sessionTitle && - sessionTitle !== existingSessionTitle - ) { - publish("session_title", { - session_id: clientSessionId, - title: sessionTitle, - }); - await persistQueue.catch((error) => { - logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist chat stream state"); - }); - } - const latestTurn = extractLatestFrontendTurn(activeRun.messages); - if (latestTurn) { - void learningOrchestrator.onTurnCompleted({ - ...latestTurn, - requestContext, - sessionId: clientSessionId, - }).catch((error) => { - logger.warn( - { err: error, sessionId: clientSessionId }, - "stream-completed learning failed", - ); - }); - } - } - } finally { - unregisterFrontendActionSink?.(); - if (abortController.signal.aborted) { - activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ - ...message, - content: - typeof message.content === "string" && message.content.trim() - ? message.content - : "⚠️ **请求已中断**", - isError: true, - progress: completeBackendProgress(message.progress), - todos: cancelBackendTodos(message.todos), - })); - if (!uiFinished && !streamClosed && !res.writableEnded && !res.destroyed) { - writeUiChunk({ - type: "error", - errorText: "请求已中断", - }); - finishUiMessage("error"); - } - void queueSessionUiStatePersist().catch((error) => { - logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist aborted chat stream state"); - }); - } - if (!uiFinished && !streamClosed && !res.writableEnded && !res.destroyed) { - finishUiMessage(activeRun.status === "error" ? "error" : "stop"); - } - await persistQueue.catch((error) => { - logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist chat stream state"); - }); - sessionBridge.finalizeRequest(clientSessionId); - activeRun.status = abortController.signal.aborted - ? activeRun.status === "aborted" - ? "aborted" - : "aborted" - : activeRun.status === "running" - ? "completed" - : activeRun.status; - lastRunStatuses.set(clientSessionId, activeRun.status); - for (const subscriber of activeRun.subscribers) { - subscriber.close(); - } - activeRun.subscribers.clear(); - if ( - activeRun.pendingPermissions.size === 0 && - activeRun.pendingQuestions.size === 0 - ) { - activeRuns.delete(clientSessionId); - } - streamClosed = true; - req.off("close", handleClientClose); - res.off("close", handleClientClose); - } - - if (!res.writableEnded && !res.destroyed) { - res.end(); - } - } catch (error) { - const detail = error instanceof Error ? error.message : String(error); - logger.error({ err: error }, "chat stream failed"); - if (res.headersSent) { - if (!res.writableEnded && !res.destroyed) { - pipeUiChunk(res, { - type: "error", - errorText: detail, - }); - pipeUiChunk(res, { - type: "finish", - finishReason: "error", - }); - res.end(); - } - return; - } - res.status(500).json({ - message: "chat stream failed", - detail, - }); - } + registerChatStreamRoute(chatRouter, { + frontendActionCoordinator, + learningOrchestrator, + memoryStore, + runState, + runtime, + sessionBridge, + sessionMetadataStore, + sessionTranscriptStore, + sessionUiStateStore, }); return chatRouter; }; - -const toSse = (event: string, data: Record) => - `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`; diff --git a/src/routes/chatRunState.ts b/src/routes/chatRunState.ts new file mode 100644 index 0000000..0deb4bd --- /dev/null +++ b/src/routes/chatRunState.ts @@ -0,0 +1,22 @@ +import { + type ActiveRun, + type RunStatus, +} from "./chatUiState.js"; + +export type ChatRunState = { + activeRuns: Map; + lastRunStatuses: Map; + getStatus: (sessionId: string) => RunStatus | undefined; +}; + +export const createChatRunState = (): ChatRunState => { + const activeRuns = new Map(); + const lastRunStatuses = new Map(); + + return { + activeRuns, + lastRunStatuses, + getStatus: (sessionId) => + activeRuns.get(sessionId)?.status ?? lastRunStatuses.get(sessionId), + }; +}; diff --git a/src/routes/chatSessionRoutes.ts b/src/routes/chatSessionRoutes.ts new file mode 100644 index 0000000..9594adb --- /dev/null +++ b/src/routes/chatSessionRoutes.ts @@ -0,0 +1,426 @@ +import { type Router } from "express"; +import { z } from "zod"; + +import { type ChatSessionBridge } from "../chat/sessionBridge.js"; +import { getLocalAgentContext } from "../context/localContext.js"; +import { type FrontendActionCoordinator } from "../frontendAction/coordinator.js"; +import { logger } from "../logger.js"; +import { type OpencodeRuntimeAdapter } from "../runtime/opencode.js"; +import { + type SessionMetadataStore, + type SessionRecord, +} from "../sessions/metadataStore.js"; +import { type SessionTranscriptStore } from "../sessions/transcriptStore.js"; +import { type SessionUiStateStore } from "../sessions/uiStateStore.js"; +import { toActorKey, toProjectKey } from "../utils/fileStore.js"; +import { type ChatRunState } from "./chatRunState.js"; +import { buildForkedSessionUiState } from "./chatSession.js"; +import { type StreamSubscriber } from "./chatUiState.js"; + +const createSessionPayloadSchema = z.object({ + session_id: z.string().max(128).optional(), + parent_session_id: z.string().max(128).optional(), +}); + +const forkPayloadSchema = z.object({ + session_id: z.string().max(128).optional(), + keep_message_count: z.coerce.number().int().min(0), +}); + +type RegisterChatSessionRoutesOptions = { + frontendActionCoordinator: FrontendActionCoordinator; + runState: ChatRunState; + runtime: OpencodeRuntimeAdapter; + sessionBridge: ChatSessionBridge; + sessionMetadataStore: SessionMetadataStore; + sessionTranscriptStore: SessionTranscriptStore; + sessionUiStateStore: SessionUiStateStore; +}; + +const toSessionUiStateContext = (sessionRecord: SessionRecord) => ({ + sessionId: sessionRecord.sessionId, +}); + +const toSse = (event: string, data: Record) => + `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`; + +export const registerChatSessionRoutes = ( + chatRouter: Router, + options: RegisterChatSessionRoutesOptions, +) => { + const { + frontendActionCoordinator, + runState, + runtime, + sessionBridge, + sessionMetadataStore, + sessionTranscriptStore, + sessionUiStateStore, + } = options; + const { activeRuns, lastRunStatuses } = runState; + const getSessionRunStatus = runState.getStatus; + + chatRouter.post("/session", async (req, res) => { + const parsed = createSessionPayloadSchema.safeParse(req.body ?? {}); + if (!parsed.success) { + res.status(400).json({ + message: "invalid request payload", + detail: parsed.error.flatten(), + }); + return; + } + + const authContext = getLocalAgentContext(req); + const projectId = authContext.projectId; + const userId = authContext.userId; + const actorKey = toActorKey(userId); + const projectKey = toProjectKey(projectId); + const requestedSessionId = parsed.data.session_id?.trim(); + let sessionId = requestedSessionId; + if (!sessionId) { + try { + sessionId = (await runtime.createSession()).id; + } catch (error) { + sessionId = sessionBridge.createClientSessionId(); + logger.warn( + { err: error, sessionId }, + "runtime unavailable while creating chat session; using local degraded session", + ); + } + } + + const { record, created } = await sessionMetadataStore.ensure({ + actorKey, + parentSessionId: parsed.data.parent_session_id, + projectId, + projectKey, + sessionId, + userId, + }); + + res.status(created ? 201 : 200).json({ + session_id: record.sessionId, + created_at: record.createdAt, + updated_at: record.updatedAt, + status: record.status, + title: record.title, + parent_session_id: record.parentSessionId, + }); + }); + + chatRouter.get("/sessions", async (req, res) => { + const authContext = getLocalAgentContext(req); + const projectId = authContext.projectId; + const userId = authContext.userId; + const actorKey = toActorKey(userId); + const projectKey = toProjectKey(projectId); + const records = await sessionMetadataStore.list({ + actorKey, + projectId, + projectKey, + userId, + }); + res.json({ + sessions: records.map((record) => ({ + id: record.sessionId, + title: record.title ?? "新对话", + created_at: record.createdAt, + updated_at: record.updatedAt, + status: record.status, + parent_session_id: record.parentSessionId, + is_streaming: activeRuns.get(record.sessionId)?.status === "running", + run_status: getSessionRunStatus(record.sessionId), + })), + }); + }); + + chatRouter.get("/session/:session_id", async (req, res) => { + const sessionId = req.params.session_id?.trim(); + const authContext = getLocalAgentContext(req); + const projectId = authContext.projectId; + const userId = authContext.userId; + const actorKey = toActorKey(userId); + const projectKey = toProjectKey(projectId); + if (!sessionId) { + res.status(400).json({ message: "session_id is required" }); + return; + } + + const sessionRecord = await sessionMetadataStore.get( + { + actorKey, + projectId, + projectKey, + userId, + }, + sessionId, + ); + if (!sessionRecord) { + res.status(404).json({ message: "session not found" }); + return; + } + + const state = await sessionUiStateStore.read( + toSessionUiStateContext(sessionRecord), + ); + res.json({ + id: sessionRecord.sessionId, + title: sessionRecord.title ?? "新对话", + is_title_manually_edited: state?.isTitleManuallyEdited ?? false, + created_at: sessionRecord.createdAt, + updated_at: sessionRecord.updatedAt, + status: sessionRecord.status, + session_id: sessionRecord.sessionId, + messages: state?.messages ?? [], + parent_session_id: sessionRecord.parentSessionId, + is_streaming: activeRuns.get(sessionRecord.sessionId)?.status === "running", + run_status: getSessionRunStatus(sessionRecord.sessionId), + }); + }); + + chatRouter.get("/session/:session_id/stream", async (req, res) => { + const sessionId = req.params.session_id?.trim(); + const authContext = getLocalAgentContext(req); + const projectId = authContext.projectId; + const userId = authContext.userId; + const actorKey = toActorKey(userId); + const projectKey = toProjectKey(projectId); + if (!sessionId) { + res.status(400).json({ message: "session_id is required" }); + return; + } + + const sessionRecord = await sessionMetadataStore.get( + { actorKey, projectId, projectKey, userId }, + sessionId, + ); + if (!sessionRecord) { + res.status(404).json({ message: "session not found" }); + return; + } + + res.status(200); + res.setHeader("Content-Type", "text/event-stream; charset=utf-8"); + res.setHeader("Cache-Control", "no-cache"); + res.setHeader("Connection", "keep-alive"); + res.setHeader("X-Accel-Buffering", "no"); + res.flushHeaders?.(); + + const run = activeRuns.get(sessionRecord.sessionId); + const state = await sessionUiStateStore.read(toSessionUiStateContext(sessionRecord)); + res.write( + toSse("state", { + session_id: sessionRecord.sessionId, + messages: state?.messages ?? run?.messages ?? [], + is_streaming: run?.status === "running", + run_status: getSessionRunStatus(sessionRecord.sessionId) ?? "completed", + }), + ); + + if (!run || run.status !== "running") { + res.end(); + return; + } + + const subscriber: StreamSubscriber = { + write: (event, data) => { + if (!res.writableEnded && !res.destroyed) { + res.write(toSse(event, data)); + } + }, + close: () => { + if (!res.writableEnded && !res.destroyed) { + res.end(); + } + }, + }; + run.subscribers.add(subscriber); + + const cleanup = () => { + run.subscribers.delete(subscriber); + }; + req.on("close", cleanup); + res.on("close", cleanup); + }); + + chatRouter.patch("/session/:session_id/title", async (req, res) => { + const sessionId = req.params.session_id?.trim(); + const title = + typeof req.body?.title === "string" ? req.body.title.trim() : ""; + const isTitleManuallyEdited = + typeof req.body?.is_title_manually_edited === "boolean" + ? req.body.is_title_manually_edited + : undefined; + const authContext = getLocalAgentContext(req); + const projectId = authContext.projectId; + const userId = authContext.userId; + const actorKey = toActorKey(userId); + const projectKey = toProjectKey(projectId); + if (!sessionId || !title) { + res.status(400).json({ message: "session_id and title are required" }); + return; + } + const sessionRecord = await sessionMetadataStore.get( + { actorKey, projectId, projectKey, userId }, + sessionId, + ); + if (!sessionRecord) { + res.status(404).json({ message: "session not found" }); + return; + } + const nextSessionRecord = await sessionMetadataStore.touch(sessionRecord, { + title, + }); + const state = await sessionUiStateStore.read( + toSessionUiStateContext(nextSessionRecord), + ); + if (state) { + await sessionUiStateStore.write( + toSessionUiStateContext(nextSessionRecord), + { + ...state, + isTitleManuallyEdited: + isTitleManuallyEdited ?? state.isTitleManuallyEdited, + }, + ); + } + res.json({ + id: nextSessionRecord.sessionId, + title: nextSessionRecord.title, + updated_at: nextSessionRecord.updatedAt, + }); + }); + + chatRouter.delete("/session/:session_id", async (req, res) => { + const sessionId = req.params.session_id?.trim(); + const authContext = getLocalAgentContext(req); + const projectId = authContext.projectId; + const userId = authContext.userId; + const actorKey = toActorKey(userId); + const projectKey = toProjectKey(projectId); + if (!sessionId) { + res.status(400).json({ message: "session_id is required" }); + return; + } + const sessionRecord = await sessionMetadataStore.get( + { actorKey, projectId, projectKey, userId }, + sessionId, + ); + if (!sessionRecord) { + res.status(204).end(); + return; + } + await sessionUiStateStore.remove(toSessionUiStateContext(sessionRecord)); + await sessionBridge.deleteSession({ + clientSessionId: sessionRecord.sessionId, + sessionId: sessionRecord.sessionId, + }); + frontendActionCoordinator.cancelSession(sessionRecord.sessionId); + activeRuns.delete(sessionRecord.sessionId); + lastRunStatuses.delete(sessionRecord.sessionId); + await sessionMetadataStore.remove(sessionRecord); + res.status(204).end(); + }); + + chatRouter.post("/fork", async (req, res) => { + const parsed = forkPayloadSchema.safeParse(req.body); + if (!parsed.success) { + res.status(400).json({ + message: "invalid request payload", + detail: parsed.error.flatten(), + }); + return; + } + + try { + const authContext = getLocalAgentContext(req); + const projectId = authContext.projectId; + const traceId = req.header("x-trace-id") ?? undefined; + const userId = authContext.userId; + + const actorKey = toActorKey(userId); + const projectKey = toProjectKey(projectId); + const sourceSessionId = parsed.data.session_id?.trim(); + const sourceSessionRecord = sourceSessionId + ? await sessionMetadataStore.get( + { + actorKey, + projectId, + projectKey, + userId, + }, + sourceSessionId, + ) + : null; + const forkSession = await runtime.createSession(); + const { record: targetSessionRecord } = await sessionMetadataStore.ensure({ + actorKey, + parentSessionId: sourceSessionId, + projectId, + projectKey, + sessionId: forkSession.id, + userId, + }); + const nextSessionId = targetSessionRecord.sessionId; + + if (sourceSessionId) { + await sessionTranscriptStore.cloneThread( + { + actorKey, + clientSessionId: sourceSessionId, + projectKey, + sessionId: sourceSessionId, + }, + { + actorKey, + clientSessionId: nextSessionId, + projectKey, + sessionId: nextSessionId, + }, + parsed.data.keep_message_count, + ); + } + const sourceState = sourceSessionRecord + ? await sessionUiStateStore.read( + toSessionUiStateContext(sourceSessionRecord), + ) + : null; + const forkTitle = sourceSessionRecord?.title + ? `${sourceSessionRecord.title} 副本` + : "新对话副本"; + const titledTargetSessionRecord = await sessionMetadataStore.touch( + targetSessionRecord, + { title: forkTitle }, + ); + await sessionUiStateStore.write( + toSessionUiStateContext(titledTargetSessionRecord), + buildForkedSessionUiState(sourceState, { + keepMessageCount: parsed.data.keep_message_count, + targetSessionId: nextSessionId, + }), + ); + + logger.info( + { + sourceSessionId: parsed.data.session_id, + sessionId: nextSessionId, + traceId, + projectId, + keepMessageCount: parsed.data.keep_message_count, + }, + "forked chat session", + ); + + res.status(200).json({ + session_id: nextSessionId, + }); + } catch (error) { + const detail = error instanceof Error ? error.message : String(error); + logger.error({ err: error }, "chat fork failed"); + res.status(500).json({ + message: "chat fork failed", + detail, + }); + } + }); +}; diff --git a/src/routes/chatStreamRoute.ts b/src/routes/chatStreamRoute.ts new file mode 100644 index 0000000..a16810c --- /dev/null +++ b/src/routes/chatStreamRoute.ts @@ -0,0 +1,811 @@ +import { type Router } from "express"; +import { + UI_MESSAGE_STREAM_HEADERS, + type UIMessage, + type UIMessageChunk, +} from "ai"; +import { z } from "zod"; + +import { isSupportedModel } from "../chat/models.js"; +import { type ChatSessionBridge } from "../chat/sessionBridge.js"; +import { getLocalAgentContext } from "../context/localContext.js"; +import { type FrontendActionCoordinator } from "../frontendAction/coordinator.js"; +import { type LearningOrchestrator } from "../learning/orchestrator.js"; +import { logger } from "../logger.js"; +import { type MemoryStore } from "../memory/store.js"; +import { type OpencodeRuntimeAdapter } from "../runtime/opencode.js"; +import { + getRuntimeSessionContext, + setRuntimeSessionContext, +} from "../runtime/sessionContext.js"; +import { + type SessionMetadataStore, + type SessionRecord, +} from "../sessions/metadataStore.js"; +import { type SessionTranscriptStore } from "../sessions/transcriptStore.js"; +import { type SessionUiStateStore } from "../sessions/uiStateStore.js"; +import type { UIEnvelopePayload } from "../uiEnvelope/types.js"; +import { toActorKey, toProjectKey } from "../utils/fileStore.js"; +import { + buildPromptWithLearningContext, + extractLatestFrontendTurn, + generateSessionTitle, + resolveRequestedStreamSessionId, + shouldGenerateSessionTitle, + shouldRestoreConversationForRuntime, +} from "./chatSession.js"; +import { + collectTextContent, + type PermissionRequestPayload, + type QuestionRequestPayload, + streamPromptResponse, + type TodoUpdatePayload, +} from "./chatStream.js"; +import { type ChatRunState } from "./chatRunState.js"; +import { + type ActiveRun, + type StreamSubscriber, + appendBackendToolArtifact, + appendBackendUiEnvelope, + cancelBackendTodos, + completeBackendProgress, + createInitialStreamingMessages, + isObjectRecord, + toFrontendPermission, + toPermissionStatus, + updateLastAssistantMessage, + updateLastAssistantPermission, + updateLastAssistantQuestion, + upsertBackendProgress, + upsertBackendQuestion, + upsertBackendTodoUpdate, +} from "./chatUiState.js"; + +const payloadSchema = z.object({ + id: z.string().max(128).optional(), + messages: z.array(z.unknown()).max(200).optional(), + message: z.string().min(1).max(10000).optional(), + session_id: z.string().max(128).optional(), + model: z.string().refine(isSupportedModel, { + message: "unsupported model", + }).optional(), + approval_mode: z.enum(["request", "always"]).optional().default("request"), + capabilities: z.array(z.string().max(64)).max(20).optional().default([]), +}); + +const toSessionUiStateContext = (sessionRecord: SessionRecord) => ({ + sessionId: sessionRecord.sessionId, +}); + +const runtimeHasConversation = async ( + runtime: OpencodeRuntimeAdapter, + sessionId: string, +) => { + const messages = await runtime.messages(sessionId, 1); + return messages.some( + (message) => + message.info.role === "user" || message.info.role === "assistant", + ); +}; + +const isUiMessage = (value: unknown): value is UIMessage => + isObjectRecord(value) && + typeof value.id === "string" && + (value.role === "system" || value.role === "user" || value.role === "assistant") && + Array.isArray(value.parts); + +const readUiMessageText = (message: UIMessage | undefined) => + message?.parts + .flatMap((part) => + isObjectRecord(part) && part.type === "text" && typeof part.text === "string" + ? [part.text] + : [], + ) + .join("") + .trim() ?? ""; + +const readLatestUserText = (messages: UIMessage[]) => { + for (let index = messages.length - 1; index >= 0; index -= 1) { + const message = messages[index]; + if (message.role !== "user") { + continue; + } + const text = readUiMessageText(message); + if (text) { + return text; + } + } + return ""; +}; + +const toUiMessages = (messages: unknown[] | undefined) => + Array.isArray(messages) ? messages.filter(isUiMessage) : []; + +const toLegacyMessages = (messages: UIMessage[]) => + messages.flatMap((message) => { + const content = readUiMessageText(message); + if (!content || (message.role !== "user" && message.role !== "assistant")) { + return []; + } + return [ + { + id: message.id, + role: message.role, + content, + }, + ]; + }); + +const createFallbackUserUiMessage = (content: string): UIMessage => ({ + id: `user-${Date.now().toString(36)}`, + role: "user", + parts: [{ type: "text", text: content }], +}); + +const pipeUiChunk = ( + res: { + write: (chunk: string) => void; + writableEnded?: boolean; + destroyed?: boolean; + }, + chunk: UIMessageChunk, +) => { + if (!res.writableEnded && !res.destroyed) { + res.write(`data: ${JSON.stringify(chunk)}\n\n`); + } +}; + +type RegisterChatStreamRouteOptions = { + frontendActionCoordinator: FrontendActionCoordinator; + learningOrchestrator: LearningOrchestrator; + memoryStore: MemoryStore; + runState: ChatRunState; + runtime: OpencodeRuntimeAdapter; + sessionBridge: ChatSessionBridge; + sessionMetadataStore: SessionMetadataStore; + sessionTranscriptStore: SessionTranscriptStore; + sessionUiStateStore: SessionUiStateStore; +}; + +export const registerChatStreamRoute = ( + chatRouter: Router, + { + frontendActionCoordinator, + learningOrchestrator, + memoryStore, + runState, + runtime, + sessionBridge, + sessionMetadataStore, + sessionTranscriptStore, + sessionUiStateStore, + }: RegisterChatStreamRouteOptions, +) => { + const { activeRuns, lastRunStatuses } = runState; + chatRouter.post("/stream", async (req, res) => { + const parsed = payloadSchema.safeParse(req.body); + if (!parsed.success) { + res.status(400).json({ + message: "invalid request payload", + detail: parsed.error.flatten(), + }); + return; + } + + try { + const authContext = getLocalAgentContext(req); + const projectId = authContext.projectId; + const traceId = req.header("x-trace-id") ?? undefined; + const userId = authContext.userId; + const actorKey = toActorKey(userId); + const projectKey = toProjectKey(projectId); + const requestUiMessages = toUiMessages(parsed.data.messages); + const promptText = + parsed.data.message?.trim() || readLatestUserText(requestUiMessages); + if (!promptText) { + res.status(400).json({ + message: "message or messages with latest user text is required", + }); + return; + } + const requestedSessionId = resolveRequestedStreamSessionId({ + sessionId: parsed.data.session_id, + promptText, + createClientSessionId: () => sessionBridge.createClientSessionId(), + }); + const existingSessionRecord = requestedSessionId + ? await sessionMetadataStore.get( + { actorKey, projectId, projectKey, userId }, + requestedSessionId, + ) + : null; + const hadExistingRuntimeSession = Boolean(existingSessionRecord); + + const { binding, requestContext, created } = await sessionBridge.resolve({ + sessionId: requestedSessionId, + network: authContext.network, + projectId, + traceId, + userId, + }); + setRuntimeSessionContext({ + ...(getRuntimeSessionContext(binding.sessionId)!), + capabilities: parsed.data.capabilities, + }); + const { record: ensuredSessionRecord, created: sessionCreated } = + await sessionMetadataStore.ensure({ + actorKey, + projectId, + projectKey, + sessionId: binding.sessionId, + userId, + }); + const activeSessionRecord = await sessionMetadataStore.touch(ensuredSessionRecord); + const hasRuntimeConversation = hadExistingRuntimeSession + ? await runtimeHasConversation(runtime, binding.sessionId) + : false; + const shouldRestoreConversation = shouldRestoreConversationForRuntime({ + hadExistingSessionRecord: hadExistingRuntimeSession, + runtimeHasConversation: hasRuntimeConversation, + }); + const historyContext = { + actorKey: requestContext.actorKey, + clientSessionId: requestContext.clientSessionId, + projectKey: requestContext.projectKey, + sessionId: requestContext.clientSessionId, + }; + const initialSessionState = await sessionUiStateStore.read( + toSessionUiStateContext(activeSessionRecord), + ); + const persistedUiMessages = toUiMessages(initialSessionState?.messages); + const requestMessages = + requestUiMessages.length > 0 + ? requestUiMessages + : [ + ...persistedUiMessages, + createFallbackUserUiMessage(promptText), + ]; + let latestUiMessages = requestMessages; + const baseMessages = toLegacyMessages(requestMessages); + if (activeRuns.get(activeSessionRecord.sessionId)?.status === "running") { + res.status(409).json({ + message: "session is already streaming", + session_id: activeSessionRecord.sessionId, + }); + return; + } + const recentTurns = await sessionTranscriptStore.getRecentTurns(historyContext, 8); + + logger.info( + { + clientSessionId: requestContext.clientSessionId, + sessionId: binding.sessionId, + created: created || sessionCreated, + model: parsed.data.model, + approvalMode: parsed.data.approval_mode, + traceId: requestContext.traceId, + projectId: requestContext.projectId, + }, + "processing chat request", + ); + + res.status(200); + for (const [header, value] of Object.entries(UI_MESSAGE_STREAM_HEADERS)) { + res.setHeader(header, value); + } + res.setHeader("X-Accel-Buffering", "no"); + res.flushHeaders?.(); + + const clientSessionId = requestContext.clientSessionId; + let streamClosed = false; + const abortController = new AbortController(); + sessionBridge.registerAbortController(clientSessionId, abortController); + const initialMessages = createInitialStreamingMessages( + baseMessages, + promptText, + ); + const activeRun: ActiveRun = { + clientSessionId, + controller: abortController, + messages: initialMessages, + pendingPermissions: new Map(), + pendingQuestions: new Map(), + status: "running", + subscribers: new Set(), + }; + activeRuns.set(clientSessionId, activeRun); + lastRunStatuses.set(clientSessionId, "running"); + const sessionUiStateContext = toSessionUiStateContext(activeSessionRecord); + const assistantUiMessageId = `assistant-${Date.now().toString(36)}`; + const textPartId = "answer"; + const assistantParts: Record[] = []; + let responseText = ""; + let textStarted = false; + let textEnded = false; + let uiFinished = false; + const syncLatestUiMessages = () => { + latestUiMessages = [ + ...requestMessages, + { + id: assistantUiMessageId, + role: "assistant", + parts: assistantParts.map((part) => ({ ...part })), + } as UIMessage, + ]; + }; + const writeUiChunk = (chunk: UIMessageChunk) => { + pipeUiChunk(res, chunk); + }; + const writeTextDelta = (delta: string) => { + if (!delta || uiFinished) { + return; + } + if (!textStarted) { + textStarted = true; + assistantParts.push({ + type: "text", + text: "", + state: "streaming", + }); + writeUiChunk({ type: "text-start", id: textPartId }); + } + responseText += delta; + const textPart = assistantParts.find( + (part) => part.type === "text", + ); + if (textPart) { + textPart.text = responseText; + textPart.state = "streaming"; + } + writeUiChunk({ type: "text-delta", id: textPartId, delta }); + writeUiChunk({ + type: "data-stream_token", + data: { + session_id: clientSessionId, + message_id: assistantUiMessageId, + content: delta, + }, + transient: true, + } as UIMessageChunk); + syncLatestUiMessages(); + }; + const writeDataPart = (event: string, data: Record) => { + if (uiFinished) { + return; + } + const type = `data-${event}`; + const id = + typeof data.id === "string" + ? data.id + : typeof data.request_id === "string" + ? data.request_id + : typeof data.envelope_id === "string" + ? data.envelope_id + : undefined; + const existingIndex = + id === undefined + ? -1 + : assistantParts.findIndex( + (part) => part.type === type && part.id === id, + ); + const part = id ? { type, id, data } : { type, data }; + if (existingIndex >= 0) { + assistantParts[existingIndex] = part; + } else { + assistantParts.push(part); + } + writeUiChunk(part as UIMessageChunk); + syncLatestUiMessages(); + }; + const finishUiMessage = (finishReason: "stop" | "error" = "stop") => { + if (uiFinished) { + return; + } + if (!textStarted && assistantParts.length === 0) { + writeTextDelta("Agent 已完成处理。"); + } + if (textStarted && !textEnded) { + textEnded = true; + const textPart = assistantParts.find( + (part) => part.type === "text", + ); + if (textPart) { + textPart.state = "done"; + } + writeUiChunk({ type: "text-end", id: textPartId }); + } + syncLatestUiMessages(); + writeUiChunk({ type: "finish", finishReason }); + uiFinished = true; + }; + writeUiChunk({ type: "start", messageId: assistantUiMessageId }); + let persistQueue = sessionUiStateStore.write(sessionUiStateContext, { + sessionId: activeSessionRecord.sessionId, + isTitleManuallyEdited: initialSessionState?.isTitleManuallyEdited ?? false, + messages: latestUiMessages, + }); + const queueSessionUiStatePersist = () => { + const snapshot = { + sessionId: activeSessionRecord.sessionId, + isTitleManuallyEdited: initialSessionState?.isTitleManuallyEdited ?? false, + messages: latestUiMessages, + }; + persistQueue = persistQueue + .catch((error) => { + logger.warn( + { err: error, sessionId: clientSessionId }, + "failed to persist previous chat stream state", + ); + }) + .then(() => sessionUiStateStore.write(sessionUiStateContext, snapshot)); + return persistQueue; + }; + const primarySubscriber: StreamSubscriber = { + write: (event, data) => { + if (streamClosed || res.writableEnded || res.destroyed) { + return; + } + if (event === "token") { + writeTextDelta(typeof data.content === "string" ? data.content : ""); + return; + } + if (event === "done") { + finishUiMessage("stop"); + return; + } + if (event === "error") { + writeUiChunk({ + type: "error", + errorText: + typeof data.message === "string" + ? data.message + : "Agent stream failed", + }); + finishUiMessage("error"); + return; + } + writeDataPart(event, data); + }, + close: () => { + if (!res.writableEnded && !res.destroyed) { + res.end(); + } + }, + }; + activeRun.subscribers.add(primarySubscriber); + const handleClientClose = () => { + streamClosed = true; + activeRun.subscribers.delete(primarySubscriber); + }; + + req.on("close", handleClientClose); + res.on("close", handleClientClose); + + const publish = (event: string, data: Record) => { + const subscriberData = + event === "token" + ? { + ...data, + message_id: + typeof data.message_id === "string" + ? data.message_id + : assistantUiMessageId, + } + : data; + if (event === "token") { + activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ + ...message, + content: `${typeof message.content === "string" ? message.content : ""}${typeof data.content === "string" ? data.content : ""}`, + isError: false, + })); + } else if (event === "progress") { + activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ + ...message, + progress: upsertBackendProgress(message.progress, data), + })); + } else if (event === "done") { + activeRun.status = "completed"; + lastRunStatuses.set(clientSessionId, "completed"); + activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ + ...message, + content: + typeof message.content === "string" && message.content.trim() + ? message.content + : "Agent 已完成处理，但没有生成文本回答。请查看过程记录，或换个更具体的问题重试。", + progress: completeBackendProgress(message.progress), + })); + } else if (event === "error") { + activeRun.status = activeRun.status === "aborted" ? "aborted" : "error"; + lastRunStatuses.set(clientSessionId, activeRun.status); + activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ + ...message, + content: + typeof message.content === "string" && message.content.trim() + ? message.content + : `⚠️ **错误：** ${typeof data.message === "string" ? data.message : "unknown error"}`, + isError: true, + progress: completeBackendProgress(message.progress), + todos: cancelBackendTodos(message.todos), + })); + } else if (event === "permission_request") { + const payload = data as PermissionRequestPayload; + activeRun.pendingPermissions.set(payload.request_id, payload); + activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ + ...message, + permissions: [ + ...(Array.isArray(message.permissions) ? message.permissions : []), + toFrontendPermission(payload), + ], + })); + } else if (event === "permission_response") { + const requestId = + typeof data.request_id === "string" ? data.request_id : undefined; + const reply = + data.reply === "once" || data.reply === "always" || data.reply === "reject" + ? data.reply + : undefined; + if (requestId && reply) { + activeRun.pendingPermissions.delete(requestId); + activeRun.messages = updateLastAssistantPermission( + activeRun.messages, + requestId, + (permission) => ({ + ...permission, + status: toPermissionStatus(reply), + repliedAt: Date.now(), + }), + ); + } + } else if (event === "question_request") { + const payload = data as QuestionRequestPayload; + let shouldTrackQuestion = true; + if (payload.tool?.callID) { + if (payload.request_id !== payload.tool.callID) { + activeRun.pendingQuestions.delete(payload.tool.callID); + } else { + const hasActionableQuestion = [...activeRun.pendingQuestions.values()].some( + (question) => + question.tool?.callID === payload.tool?.callID && + question.request_id !== payload.tool?.callID, + ); + if (hasActionableQuestion) { + activeRun.messages = updateLastAssistantMessage( + activeRun.messages, + (message) => ({ + ...message, + questions: upsertBackendQuestion(message.questions, payload), + }), + ); + shouldTrackQuestion = false; + } + } + } + if (shouldTrackQuestion) { + activeRun.pendingQuestions.set(payload.request_id, payload); + activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ + ...message, + questions: upsertBackendQuestion(message.questions, payload), + })); + } + } else if (event === "question_response") { + const requestId = + typeof data.request_id === "string" ? data.request_id : undefined; + if (requestId) { + activeRun.pendingQuestions.delete(requestId); + activeRun.messages = updateLastAssistantQuestion( + activeRun.messages, + requestId, + (question) => ({ + ...question, + status: data.rejected === true ? "rejected" : "answered", + repliedAt: Date.now(), + answers: Array.isArray(data.answers) ? data.answers : question.answers, + error: undefined, + }), + ); + } + } else if (event === "todo_update") { + const payload = data as TodoUpdatePayload; + activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ + ...message, + todos: upsertBackendTodoUpdate(message.todos, payload), + })); + } else if (event === "tool_call") { + activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ + ...message, + artifacts: appendBackendToolArtifact(message.artifacts, data), + })); + } else if (event === "ui_envelope") { + const payload = data as UIEnvelopePayload; + activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ + ...message, + uiEnvelopes: appendBackendUiEnvelope(message.uiEnvelopes, payload), + })); + } + + for (const subscriber of activeRun.subscribers) { + subscriber.write(event, subscriberData); + } + void queueSessionUiStatePersist().catch((error) => { + logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist chat stream state"); + }); + }; + + const unregisterFrontendActionSink = parsed.data.capabilities.includes("frontend-action@1") + ? frontendActionCoordinator.registerSink(clientSessionId, (request) => publish("frontend_action", request as unknown as Record)) + : undefined; + + try { + const preparedMessage = await buildPromptWithLearningContext( + memoryStore, + requestContext.actorKey, + requestContext.projectKey, + { + recentTurns, + persistedMessages: baseMessages, + message: promptText, + restoreConversation: shouldRestoreConversation, + }, + ); + const streamResult = await streamPromptResponse({ + runtime, + sessionId: binding.sessionId, + clientSessionId, + message: preparedMessage, + model: parsed.data.model, + approvalMode: parsed.data.approval_mode, + traceId: requestContext.traceId, + projectId: requestContext.projectId, + signal: abortController.signal, + suppressLegacyFrontendActions: parsed.data.capabilities.includes("frontend-action@1"), + write: (event, data) => { + publish(event, data); + }, + }); + await persistQueue.catch((error) => { + logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist chat stream state"); + }); + + if (!streamResult.aborted && !streamResult.failed) { + const messages = await runtime.messages(binding.sessionId, 60); + const assistantMessage = [...messages] + .reverse() + .find((message) => message.info.role === "assistant"); + const assistantText = collectTextContent(assistantMessage?.parts ?? []); + const latestSessionRecord = + (await sessionMetadataStore.get( + { actorKey, projectId, projectKey, userId }, + activeSessionRecord.sessionId, + )) ?? activeSessionRecord; + const latestSessionState = await sessionUiStateStore.read( + toSessionUiStateContext(latestSessionRecord), + ); + const existingSessionTitle = latestSessionRecord.title; + let sessionTitle = existingSessionTitle; + const shouldGenerateTitle = shouldGenerateSessionTitle({ + recentTurnCount: recentTurns.length, + isTitleManuallyEdited: + latestSessionState?.isTitleManuallyEdited ?? false, + }); + if (shouldGenerateTitle) { + sessionTitle = await generateSessionTitle(runtime, { + sessionId: binding.sessionId, + latestAssistantMessage: assistantText, + latestUserMessage: promptText, + fallbackTitle: existingSessionTitle, + }); + } + await sessionMetadataStore.touch(latestSessionRecord, { + ...(sessionTitle && sessionTitle !== existingSessionTitle + ? { title: sessionTitle } + : {}), + }); + if ( + shouldGenerateTitle && + sessionTitle && + sessionTitle !== existingSessionTitle + ) { + publish("session_title", { + session_id: clientSessionId, + title: sessionTitle, + }); + await persistQueue.catch((error) => { + logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist chat stream state"); + }); + } + const latestTurn = extractLatestFrontendTurn(activeRun.messages); + if (latestTurn) { + void learningOrchestrator.onTurnCompleted({ + ...latestTurn, + requestContext, + sessionId: clientSessionId, + }).catch((error) => { + logger.warn( + { err: error, sessionId: clientSessionId }, + "stream-completed learning failed", + ); + }); + } + } + } finally { + unregisterFrontendActionSink?.(); + if (abortController.signal.aborted) { + activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ + ...message, + content: + typeof message.content === "string" && message.content.trim() + ? message.content + : "⚠️ **请求已中断**", + isError: true, + progress: completeBackendProgress(message.progress), + todos: cancelBackendTodos(message.todos), + })); + if (!uiFinished && !streamClosed && !res.writableEnded && !res.destroyed) { + writeUiChunk({ + type: "error", + errorText: "请求已中断", + }); + finishUiMessage("error"); + } + void queueSessionUiStatePersist().catch((error) => { + logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist aborted chat stream state"); + }); + } + if (!uiFinished && !streamClosed && !res.writableEnded && !res.destroyed) { + finishUiMessage(activeRun.status === "error" ? "error" : "stop"); + } + await persistQueue.catch((error) => { + logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist chat stream state"); + }); + sessionBridge.finalizeRequest(clientSessionId); + activeRun.status = abortController.signal.aborted + ? activeRun.status === "aborted" + ? "aborted" + : "aborted" + : activeRun.status === "running" + ? "completed" + : activeRun.status; + lastRunStatuses.set(clientSessionId, activeRun.status); + for (const subscriber of activeRun.subscribers) { + subscriber.close(); + } + activeRun.subscribers.clear(); + if ( + activeRun.pendingPermissions.size === 0 && + activeRun.pendingQuestions.size === 0 + ) { + activeRuns.delete(clientSessionId); + } + streamClosed = true; + req.off("close", handleClientClose); + res.off("close", handleClientClose); + } + + if (!res.writableEnded && !res.destroyed) { + res.end(); + } + } catch (error) { + const detail = error instanceof Error ? error.message : String(error); + logger.error({ err: error }, "chat stream failed"); + if (res.headersSent) { + if (!res.writableEnded && !res.destroyed) { + pipeUiChunk(res, { + type: "error", + errorText: detail, + }); + pipeUiChunk(res, { + type: "finish", + finishReason: "error", + }); + res.end(); + } + return; + } + res.status(500).json({ + message: "chat stream failed", + detail, + }); + } + }); + +}; diff --git a/src/server.ts b/src/server.ts index 0bb0e09..a5c5305 100644 --- a/src/server.ts +++ b/src/server.ts @@ -126,17 +126,17 @@ app.post("/internal/tools/session-search", async (req, res) => { app.use( "/api/v1/agent/chat", - buildChatRouter( - sessionBridge, - opencodeRuntime, - sessionMetadataStore, - sessionUiStateStore, - memoryStore, - sessionTranscriptStore, - learningOrchestrator, - resultReferenceResolver, + buildChatRouter({ frontendActionCoordinator, - ), + learningOrchestrator, + memoryStore, + resultReferenceResolver, + runtime: opencodeRuntime, + sessionBridge, + sessionMetadataStore, + sessionTranscriptStore, + sessionUiStateStore, + }), ); const bootstrap = async () => {