refactor: split chat route responsibilities

src/routes/chatRunState.ts

@@ -0,0 +1,22 @@
+import {
+  type ActiveRun,
+  type RunStatus,
+} from "./chatUiState.js";
+
+export type ChatRunState = {
+  activeRuns: Map<string, ActiveRun>;
+  lastRunStatuses: Map<string, RunStatus>;
+  getStatus: (sessionId: string) => RunStatus | undefined;
+};
+
+export const createChatRunState = (): ChatRunState => {
+  const activeRuns = new Map<string, ActiveRun>();
+  const lastRunStatuses = new Map<string, RunStatus>();
+
+  return {
+    activeRuns,
+    lastRunStatuses,
+    getStatus: (sessionId) =>
+      activeRuns.get(sessionId)?.status ?? lastRunStatuses.get(sessionId),
+  };
+};

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<string, unknown>) =>
+  `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,
+      });
+    }
+  });
+};

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<string, unknown>[] = [];
+      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<string, unknown>) => {
+        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<string, unknown>) => {
+        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<string, unknown>))
+        : 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,
+      });
+    }
+  });
+
+};

src/server.ts

@@ -126,17 +126,17 @@ app.post("/internal/tools/session-search", async (req, res) => {
 
 app.use(
   "/api/v1/agent/chat",
-  buildChatRouter(
+  buildChatRouter({
-    sessionBridge,
-    opencodeRuntime,
-    sessionMetadataStore,
-    sessionUiStateStore,
-    memoryStore,
-    sessionTranscriptStore,
-    learningOrchestrator,
-    resultReferenceResolver,
     frontendActionCoordinator,
-  ),
+    learningOrchestrator,
+    memoryStore,
+    resultReferenceResolver,
+    runtime: opencodeRuntime,
+    sessionBridge,
+    sessionMetadataStore,
+    sessionTranscriptStore,
+    sessionUiStateStore,
+  }),
 );
 
 const bootstrap = async () => {