import { Router } from "express";
import { z } from "zod";

import { getAgentAuthContext } from "../auth/agentAuth.js";
import { type CredentialRefreshCoordinator } from "../auth/credentialRefresh.js";
import {
  agentModelOptions,
  isSupportedModel,
  resolveDefaultModel,
  type SupportedModel,
} 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 } from "../runtime/sessionContext.js";
import { type ChatSessionBridge } from "../chat/sessionBridge.js";
import { type SessionRecord } from "../sessions/metadataStore.js";
import { toActorKey, toProjectKey } from "../utils/fileStore.js";
import {
  buildPromptWithLearningContext,
  extractLatestFrontendTurn,
  generateSessionTitle,
  shouldGenerateSessionTitle,
  shouldRestoreConversationForRuntime,
} from "./chatSession.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,
  cancelBackendTodos,
  completeBackendProgress,
  createInitialStreamingMessages,
  isObjectRecord,
  toFrontendPermission,
  toPermissionStatus,
  updateLastAssistantMessage,
  updateLastAssistantPermission,
  updateLastAssistantQuestion,
  upsertBackendProgress,
  upsertBackendQuestion,
  upsertBackendTodoUpdate,
} from "./chatUiState.js";

const payloadSchema = z.object({
  message: z.string().min(1).max(10000),
  session_id: z.string().max(128).optional(),
  model: z.string().refine(isSupportedModel, {
    message: "unsupported model",
  }).optional(),
  approval_mode: z
    .enum(["request", "auto", "always"])
    .optional()
    .default("request")
    .describe(
      "request forwards approval prompts; auto approves only low-risk allowlisted tools; always approves every prompt not explicitly denied by OpenCode",
    ),
});

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<string, ActiveRun>();
const lastRunStatuses = new Map<string, RunStatus>();

const toSessionUiStateContext = (sessionRecord: SessionRecord) => ({
  sessionId: sessionRecord.sessionId,
});

export const buildForkedSessionUiState = (
  sourceState: { messages?: unknown[] } | null | undefined,
  input: {
    keepMessageCount: number;
    targetSessionId: string;
  },
) => ({
  sessionId: input.targetSessionId,
  isTitleManuallyEdited: false,
  messages: Array.isArray(sourceState?.messages)
    ? sourceState.messages.slice(0, input.keepMessageCount)
    : [],
});

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 const buildChatRouter = (
  sessionBridge: ChatSessionBridge,
  runtime: OpencodeRuntimeAdapter,
  sessionMetadataStore: SessionMetadataStore,
  sessionUiStateStore: SessionUiStateStore,
  memoryStore: MemoryStore,
  sessionTranscriptStore: SessionTranscriptStore,
  learningOrchestrator: LearningOrchestrator,
  resultReferenceResolver: ResultReferenceResolver,
  credentialRefreshCoordinator: CredentialRefreshCoordinator,
) => {
  const chatRouter = Router();

  chatRouter.get("/models", (_req, res) => {
    res.json({
      default_model: resolveDefaultModel(config.OPENCODE_MODEL),
      models: agentModelOptions,
    });
  });

  chatRouter.post("/sessions", 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 = getAgentAuthContext(req);
    const projectId = authContext.projectId;
    const userId = authContext.userId;
    const actorKey = toActorKey(userId);
    const projectKey = toProjectKey(projectId);
    const requestedSessionId = parsed.data.session_id?.trim();
    const sessionId =
      requestedSessionId ||
      (await runtime.createSession(undefined, { conversationWorkspace: true })).id;

    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 = getAgentAuthContext(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("/sessions/:session_id", async (req, res) => {
    const sessionId = req.params.session_id?.trim();
    const authContext = getAgentAuthContext(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("/sessions/:session_id/runs/current/events", async (req, res) => {
    const sessionId = req.params.session_id?.trim();
    const authContext = getAgentAuthContext(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 pendingCredentialRefresh =
      credentialRefreshCoordinator.getPendingEvent(sessionRecord.sessionId);
    if (pendingCredentialRefresh) {
      subscriber.write(pendingCredentialRefresh.type, {
        session_id: sessionRecord.sessionId,
        request_id: pendingCredentialRefresh.requestId,
        reason: pendingCredentialRefresh.reason,
        timeout_ms: pendingCredentialRefresh.timeoutMs,
      });
    }

    const cleanup = () => {
      run.subscribers.delete(subscriber);
    };
    req.on("close", cleanup);
    res.on("close", cleanup);
  });

  chatRouter.patch("/sessions/:session_id", 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 = getAgentAuthContext(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("/sessions/:session_id", async (req, res) => {
    const sessionId = req.params.session_id?.trim();
    const authContext = getAgentAuthContext(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,
    });
    activeRuns.delete(sessionRecord.sessionId);
    lastRunStatuses.delete(sessionRecord.sessionId);
    await sessionMetadataStore.remove(sessionRecord);
    res.status(204).end();
  });

  registerChatAuxiliaryRoutes(chatRouter, {
    activeRuns,
    lastRunStatuses,
    resultReferenceResolver,
    sessionBridge,
    sessionMetadataStore,
    sessionUiStateStore,
  });

  registerChatInteractionRoutes(chatRouter, {
    activeRuns,
    credentialRefreshCoordinator,
    runtime,
    sessionMetadataStore,
    sessionUiStateStore,
  });

  chatRouter.post("/sessions/:session_id/forks", async (req, res) => {
    const parsed = forkPayloadSchema.safeParse({
      ...req.body,
      session_id: req.params.session_id,
    });
    if (!parsed.success) {
      res.status(400).json({
        message: "invalid request payload",
        detail: parsed.error.flatten(),
      });
      return;
    }

    try {
      const authContext = getAgentAuthContext(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;
      if (!sourceSessionId || !sourceSessionRecord) {
        res.status(404).json({ message: "source session not found" });
        return;
      }
      const forkSession = await runtime.createSession(undefined, {
        conversationWorkspace: true,
      });
      const { record: targetSessionRecord } = await sessionMetadataStore.ensure({
        actorKey,
        parentSessionId: sourceSessionId,
        projectId,
        projectKey,
        sessionId: forkSession.id,
        userId,
      });
      const nextSessionId = targetSessionRecord.sessionId;

      await sessionTranscriptStore.cloneThread(
        {
          actorKey,
          clientSessionId: sourceSessionId,
          projectKey,
          sessionId: sourceSessionId,
        },
        {
          actorKey,
          clientSessionId: nextSessionId,
          projectKey,
          sessionId: nextSessionId,
        },
        parsed.data.keep_message_count,
      );
      const sourceState = await sessionUiStateStore.read(
        toSessionUiStateContext(sourceSessionRecord),
      );
      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("/sessions/:session_id/runs", async (req, res) => {
    const parsed = payloadSchema.safeParse({
      ...req.body,
      session_id: req.params.session_id,
    });
    if (!parsed.success) {
      res.status(400).json({
        message: "invalid request payload",
        detail: parsed.error.flatten(),
      });
      return;
    }

    try {
      const authContext = getAgentAuthContext(req);
      const accessToken = authContext.accessToken;
      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 requestedSessionId = parsed.data.session_id?.trim();
      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,
        accessToken,
        network: authContext.network,
        projectId,
        traceId,
        tokenExpiresAt: authContext.tokenExpiresAt,
        userId,
      });
      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 persistedMessages = initialSessionState?.messages ?? [];
      const baseMessages = persistedMessages;
      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);
      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 clientSessionId = requestContext.clientSessionId;
      let streamClosed = false;
      const abortController = new AbortController();
      sessionBridge.registerAbortController(clientSessionId, abortController);
      const initialMessages = createInitialStreamingMessages(
        baseMessages,
        parsed.data.message,
      );
      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);
      let persistQueue = sessionUiStateStore.write(sessionUiStateContext, {
        sessionId: activeSessionRecord.sessionId,
        isTitleManuallyEdited: initialSessionState?.isTitleManuallyEdited ?? false,
        messages: initialMessages,
      });
      const queueSessionUiStatePersist = () => {
        const snapshot = {
          sessionId: activeSessionRecord.sessionId,
          isTitleManuallyEdited: initialSessionState?.isTitleManuallyEdited ?? false,
          messages: activeRun.messages,
        };
        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) {
            res.write(toSse(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>) => {
        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 === "auth_required") {
          activeRun.status = "error";
          lastRunStatuses.set(clientSessionId, "error");
          activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
            ...message,
            content:
              typeof data.message === "string"
                ? `⚠️ **${data.message}**`
                : "⚠️ **登录态已过期，请刷新登录后重试**",
            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),
          }));
        }

        for (const subscriber of activeRun.subscribers) {
          subscriber.write(event, data);
        }
        void queueSessionUiStatePersist().catch((error) => {
          logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist chat stream state");
        });
      };
      const unsubscribeCredentialRefresh = credentialRefreshCoordinator.subscribe(
        binding.sessionId,
        (event) => {
          publish(event.type, {
            session_id: clientSessionId,
            request_id: event.requestId,
            ...(event.type === "credential_refresh_required"
              ? {
                  reason: event.reason,
                  timeout_ms: event.timeoutMs,
                }
              : {}),
            ...(event.type === "credential_refresh_failed"
              ? { message: event.message }
              : {}),
          });
        },
      );
      const cancelCredentialRefreshOnAbort = () => {
        credentialRefreshCoordinator.cancelSession(
          binding.sessionId,
          "credential refresh cancelled because the agent run was aborted",
        );
      };
      abortController.signal.addEventListener(
        "abort",
        cancelCredentialRefreshOnAbort,
        { once: true },
      );

      try {
        const preparedMessage = await buildPromptWithLearningContext(
          memoryStore,
          requestContext.actorKey,
          requestContext.projectKey,
          {
            recentTurns,
            persistedMessages: baseMessages,
            message: parsed.data.message,
            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,
          workspaceRoot: binding.workspaceDirectory,
          write: (event, data) => {
            publish(event, data);
          },
        });
        await persistQueue.catch((error) => {
          logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist chat stream state");
        });

        const latestRuntimeContext = getRuntimeSessionContext(binding.sessionId);
        if (latestRuntimeContext?.authExpired) {
          publish("auth_required", {
            session_id: clientSessionId,
            reason: latestRuntimeContext.authExpired.reason,
            message: latestRuntimeContext.authExpired.message,
          });
          return;
        }

        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: parsed.data.message,
              fallbackTitle: existingSessionTitle,
            });
          }
          const nextSessionRecord = 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 {
        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),
          }));
          void queueSessionUiStatePersist().catch((error) => {
            logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist aborted chat stream state");
          });
        }
        await persistQueue.catch((error) => {
          logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist chat stream state");
        });
        sessionBridge.finalizeRequest(clientSessionId);
        abortController.signal.removeEventListener(
          "abort",
          cancelCredentialRefreshOnAbort,
        );
        credentialRefreshCoordinator.cancelSession(binding.sessionId);
        unsubscribeCredentialRefresh();
        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) {
          res.write(toSse("error", {
            message: "chat stream failed",
            detail,
          }));
          res.end();
        }
        return;
      }
      res.status(500).json({
        message: "chat stream failed",
        detail,
      });
    }
  });

  return chatRouter;
};

const toSse = (event: string, data: Record<string, unknown>) =>
  `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`;