import { credentialRefreshCoordinator } from "../product/credential-refresh.js";
import { type Router } from "express";
import { randomUUID } from "node:crypto";
import {
  UI_MESSAGE_STREAM_HEADERS,
  type UIMessage,
  type UIMessageChunk,
} from "ai";
import { z } from "zod";
import { createExecutionPlanSnapshot, executionPlanNeedsTaskPublish, readExecutionPlans } from "./executionPlan.js";
import { executionPlanSchema } from "./executionPlan.js";

import { isSupportedModel } from "../chat/models.js";
import { type ChatSessionBridge } from "../chat/sessionBridge.js";
import {
  getLocalAgentContext,
  getTrustedProductCredential,
} 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,
  readPurpose,
} from "../sessions/metadataStore.js";
import {
  bindOperationalTaskCredential,
  clearOperationalTaskLease,
  operationalTaskClient,
  releaseOperationalTaskCredential,
} from "../product/operational-task-client.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";
import {
  agentTriggerContextSchema,
  buildAgentTriggerPrompt,
} from "./agentTrigger.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([]),
  trigger_context: agentTriggerContextSchema.optional(),
});

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) => {
  const finalPart = message?.parts.slice().reverse().find((part) => isObjectRecord(part) && part.type === "data-final_answer" && isObjectRecord(part.data) && typeof part.data.content === "string");
  if (finalPart && "data" in finalPart && isObjectRecord(finalPart.data)) return String(finalPart.data.content).trim();
  return 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) : [];

export const toCollaborationMessages = (messages: UIMessage[]) =>
  messages.flatMap((message) => {
    if (message.role !== "user" && message.role !== "assistant") return [];
    const content = readUiMessageText(message);
    return content ? [{ ...message, parts: [{ type: "text" as const, text: content }] } as UIMessage] : [];
  });

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 }],
});

export const buildFailedTaskExecutionPlan = (history: Record<string, unknown>[]) => {
  const previous = history.at(-1);
  const parsed = previous ? executionPlanSchema.safeParse(previous) : null;
  if (!parsed?.success) {
    return {
      title: "工况任务执行失败",
      reason: "Agent 模型或工具执行失败，已记录失败并停止本次执行。",
      nodes: [{
        id: "AGENT_RUN_FAILURE",
        title: "Agent 执行失败",
        status: "failed" as const,
        owner: "agent" as const,
        detail: "模型或工具执行失败，等待人工检查或重新发起。",
        evidence: [],
        action_refs: [],
      }],
      edges: [],
    };
  }
  let changed = false;
  const nodes = parsed.data.nodes.map((node) => {
    if (node.status === "running") {
      changed = true;
      return {
        ...node,
        status: "failed" as const,
        detail: "模型或工具执行失败，当前节点未完成，等待人工检查或重新发起。",
      };
    }
    if (node.status === "pending" || node.status === "waiting") {
      return {
        ...node,
        status: "cancelled" as const,
        detail: `${node.detail ? `${node.detail}；` : ""}本次 Agent 执行失败，后续工作未启动。`,
      };
    }
    return node;
  });
  if (!changed && !nodes.some((node) => node.status === "failed")) {
    const known = new Set(nodes.map((node) => node.id));
    let index = 1;
    let id = "AGENT_RUN_FAILURE";
    while (known.has(id)) id = `AGENT_RUN_FAILURE_${index++}`;
    nodes.push({
      id,
      title: "Agent 执行失败",
      status: "failed",
      owner: "agent",
      detail: "模型或工具执行失败，等待人工检查或重新发起。",
      evidence: [],
      action_refs: [],
    });
  }
  return {
    ...parsed.data,
    reason: "Agent 模型或工具执行失败，已记录失败并停止本次执行。",
    nodes,
  };
};

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;
  const startingSessions = new Set<string>();
  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;
    }

    let boundOperationalSessionId: string | undefined;
    let startingSessionId: string | undefined;
    try {
      const authContext = getLocalAgentContext(req);
      const projectId = authContext.projectId;
      const traceId = req.header("x-trace-id") ?? undefined;
      const userId = authContext.userId;
      const actorKey = authContext.actorKey;
      const projectKey = authContext.projectKey;
      const trigger = parsed.data.trigger_context;
      const requestUiMessages = toUiMessages(parsed.data.messages);
      const requestPromptText =
        parsed.data.message?.trim() || readLatestUserText(requestUiMessages);
      const promptText = trigger
        ? buildAgentTriggerPrompt(trigger)
        : requestPromptText;
      if (!promptText) {
        res.status(400).json({
          message: "message or messages with latest user text is required",
        });
        return;
      }
      if (trigger && !parsed.data.session_id?.trim()) {
        res.status(400).json({
          message: "event trigger requires a reserved session_id",
        });
        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;
      if (requestedSessionId && !existingSessionRecord) {
        res.status(404).json({ message: "session not found" });
        return;
      }
      if (trigger && existingSessionRecord?.triggerId !== trigger.trigger_id) {
        res.status(409).json({ message: "trigger does not match session" });
        return;
      }
      if (
        existingSessionRecord &&
        readPurpose(existingSessionRecord) === "task_execution" &&
        existingSessionRecord.modelId &&
        parsed.data.model !== existingSessionRecord.modelId
      ) {
        res.status(409).json({ message: "task execution model does not match the reserved session" });
        return;
      }
      const hadExistingRuntimeSession = Boolean(existingSessionRecord);

      const { binding, requestContext, created } = await sessionBridge.resolve({
        sessionId: requestedSessionId,
        network: authContext.network,
        projectId,
        traceId,
        userId,
        actorKey,
        projectKey,
      });
      setRuntimeSessionContext({
        ...(getRuntimeSessionContext(binding.sessionId)!),
        capabilities: parsed.data.capabilities,
        purpose: existingSessionRecord ? readPurpose(existingSessionRecord) : "ordinary_chat",
        taskId: existingSessionRecord?.taskId,
      });
      const { record: ensuredSessionRecord, created: sessionCreated } =
        await sessionMetadataStore.ensure({
          actorKey,
          projectId,
          projectKey,
          sessionId: binding.sessionId,
          userId,
        });
      const activeSessionRecord = await sessionMetadataStore.touch(ensuredSessionRecord);
      const sessionPurpose = readPurpose(activeSessionRecord);
      if (
        activeRuns.get(activeSessionRecord.sessionId)?.status === "running" ||
        startingSessions.has(activeSessionRecord.sessionId)
      ) {
        res.status(409).json({
          message: "session is already streaming",
          session_id: activeSessionRecord.sessionId,
        });
        return;
      }
      startingSessions.add(activeSessionRecord.sessionId);
      startingSessionId = activeSessionRecord.sessionId;
      bindOperationalTaskCredential(
        binding.sessionId,
        getTrustedProductCredential(req),
      );
      boundOperationalSessionId = binding.sessionId;
      const resolvedTaskContext = activeSessionRecord.taskId
        && (sessionPurpose === "task_collaboration" || sessionPurpose === "task_execution")
        ? await operationalTaskClient.getAgentContext(binding.sessionId, activeSessionRecord.taskId)
        : null;
      const taskContext = sessionPurpose === "task_collaboration"
        ? resolvedTaskContext
        : null;
      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);
      let executionPlanHistory = initialSessionState?.executionPlans ?? readExecutionPlans(initialSessionState?.messages ?? []);
      const visiblePersistedMessages = sessionPurpose === "task_collaboration"
        ? toCollaborationMessages(persistedUiMessages)
        : persistedUiMessages;
      const visibleRequestMessages = sessionPurpose === "task_collaboration"
        ? toCollaborationMessages(requestUiMessages)
        : requestUiMessages;
      const requestMessages =
        visibleRequestMessages.length > 0
          ? visibleRequestMessages
          : [
              ...visiblePersistedMessages,
              createFallbackUserUiMessage(promptText),
            ];
      let latestUiMessages = requestMessages;
      const baseMessages = toLegacyMessages(requestMessages);
      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: updateLastAssistantMessage(initialMessages, (message) => ({
          ...message, executionPlans: executionPlanHistory,
        })),
        pendingPermissions: new Map(),
        pendingQuestions: new Map(),
        status: "running",
        subscribers: new Set(),
      };
      activeRuns.set(clientSessionId, activeRun);
      startingSessions.delete(activeSessionRecord.sessionId);
      startingSessionId = undefined;
      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, {
        executionPlans: executionPlanHistory,
        sessionId: activeSessionRecord.sessionId,
        isTitleManuallyEdited: initialSessionState?.isTitleManuallyEdited ?? false,
        messages: sessionPurpose === "task_collaboration" ? toCollaborationMessages(latestUiMessages) : latestUiMessages,
      });
      const queueSessionUiStatePersist = () => {
        const snapshot = {
          executionPlans: executionPlanHistory,
          sessionId: activeSessionRecord.sessionId,
          isTitleManuallyEdited: initialSessionState?.isTitleManuallyEdited ?? false,
          messages: sessionPurpose === "task_collaboration" ? toCollaborationMessages(latestUiMessages) : 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);
      let taskArtifactPublishQueue: Promise<unknown> = Promise.resolve();

      const publish = (event: string, data: Record<string, unknown>) => {
        let 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 === "final_answer") {
          activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
            ...message, content: typeof data.content === "string" ? data.content : message.content,
          }));
        } 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 === "execution_plan") {
          try {
            const snapshot = createExecutionPlanSnapshot(data, [{ executionPlans: executionPlanHistory }], clientSessionId);
            executionPlanHistory = [...executionPlanHistory, snapshot];
            subscriberData = snapshot;
            activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
              ...message,
              executionPlans: [...(Array.isArray(message.executionPlans) ? message.executionPlans : []), snapshot],
            }));
            if (activeSessionRecord.taskId && executionPlanNeedsTaskPublish(data)) {
              taskArtifactPublishQueue = taskArtifactPublishQueue
                .catch(() => undefined)
                .then(() => operationalTaskClient.publishArtifacts(binding.sessionId, activeSessionRecord.taskId!, {
                  requestId: randomUUID(),
                  sourceSessionId: clientSessionId,
                  executionPlan: snapshot,
                }))
                .catch((error) => {
                  logger.warn({ err: error, sessionId: clientSessionId, taskId: activeSessionRecord.taskId }, "failed to publish execution plan to task record");
                });
            }
          } catch (error) {
            logger.warn({ err: error, sessionId: clientSessionId }, "rejected invalid execution plan");
            publish("progress", {
              id: "execution-plan-rejected", phase: "execution_plan", status: "error",
              title: "计划更新未通过校验，保留上一版",
              detail: error instanceof Error ? error.message : "计划结构无效",
            });
            return;
          }
        } 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),
          }));
        }

        const hiddenCollaborationEvent = sessionPurpose === "task_collaboration" && [
          "progress",
          "todo_update",
          "tool_call",
          "ui_envelope",
          "execution_plan",
        ].includes(event);
        if (!hiddenCollaborationEvent) {
          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;
      let credentialFailure: Error | undefined;
      const unsubscribeCredentialRefresh = credentialRefreshCoordinator.subscribe(binding.sessionId, (event) => {
        publish(event.type, { ...event, session_id: clientSessionId, request_id: event.requestId });
        if (event.type === "credential_refresh_failed") {
          credentialFailure = new Error(event.message);
          abortController.abort();
        }
      });
      if (sessionPurpose === "task_execution") credentialRefreshCoordinator.enableBackground(binding.sessionId);
      const cancelCredentialRefresh = () => credentialRefreshCoordinator.cancelSession(binding.sessionId);
      abortController.signal.addEventListener("abort", cancelCredentialRefresh, { once: true });
      let taskHeartbeat: ReturnType<typeof setInterval> | undefined;
      let taskHeartbeatActive = true;
      let taskHeartbeatPromise: Promise<void> | undefined;
      let taskLeaseFailure: Error | undefined;
      let taskRunFinalized = false;
      const stopTaskHeartbeat = async () => {
        taskHeartbeatActive = false;
        if (taskHeartbeat) clearInterval(taskHeartbeat);
        await taskHeartbeatPromise?.catch(() => undefined);
      };
      if (sessionPurpose === "task_execution" && activeSessionRecord.taskId) {
        taskHeartbeat = setInterval(() => {
          if (!taskHeartbeatActive || taskHeartbeatPromise) return;
          const heartbeat = operationalTaskClient.heartbeatTask(
            binding.sessionId,
            activeSessionRecord.taskId!,
            activeSessionRecord.runId!,
            clientSessionId,
          ).then(() => undefined).catch((error) => {
            if (!taskHeartbeatActive || abortController.signal.aborted) return;
            taskLeaseFailure = error instanceof Error
              ? error
              : new Error("task lease heartbeat failed");
            taskHeartbeatActive = false;
            if (taskHeartbeat) clearInterval(taskHeartbeat);
            logger.warn(
              { err: error, sessionId: clientSessionId, taskId: activeSessionRecord.taskId },
              "task lease heartbeat failed; stopping stale execution",
            );
            abortController.abort();
          }).finally(() => {
            if (taskHeartbeatPromise === heartbeat) taskHeartbeatPromise = undefined;
          });
          taskHeartbeatPromise = heartbeat;
        }, 20_000);
      }
      taskHeartbeat?.unref();

      try {
        const preparedMessage = taskContext
          ? [
              "你正在与用户协作理解一个工况任务。只根据下面由后端授权返回的任务、计划、报告和证据快照解释判断；不要展示后台执行会话、工具轨迹或内部推理。若资料不足，明确说明缺口。用户要求变更任务时使用结构化工具写回。",
              `任务快照：${JSON.stringify(taskContext)}`,
              `用户问题：${promptText}`,
            ].join("\n\n")
          : 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);
          },
        });
        if (credentialFailure) {
          publish("error", { session_id: clientSessionId, message: "会话凭据续期失败，本次执行已停止。" });
        } else if (taskLeaseFailure) {
          publish("error", {
            session_id: clientSessionId,
            message: "工况任务租约已失效，本次执行已停止，等待任务重新领取。",
            detail: taskLeaseFailure.message,
          });
        } else if (streamResult.failed && sessionPurpose === "task_execution" && activeSessionRecord.taskId) {
          publish("execution_plan", buildFailedTaskExecutionPlan(executionPlanHistory));
        }
        await taskArtifactPublishQueue;
        await stopTaskHeartbeat();
        if (
          sessionPurpose === "task_execution"
          && activeSessionRecord.taskId
          && activeSessionRecord.runId
          && !taskLeaseFailure
        ) {
          if (streamResult.failed || streamResult.aborted) {
            await operationalTaskClient.failTaskRun(
              binding.sessionId,
              activeSessionRecord.taskId,
              activeSessionRecord.runId,
              clientSessionId,
              credentialFailure ? "CREDENTIAL_REFRESH_FAILED" : streamResult.aborted ? "AGENT_ABORTED" : "AGENT_STREAM_FAILED",
              streamResult.aborted ? "Agent execution was aborted" : "Agent stream failed",
            );
            taskRunFinalized = true;
          } else {
            try {
              const completedTask = await operationalTaskClient.completeTaskRun(
                binding.sessionId,
                activeSessionRecord.taskId,
                activeSessionRecord.runId,
                clientSessionId,
              );
              taskRunFinalized = true;
              if (completedTask.lifecycle_status === "failed") {
                throw new Error("Server rejected the run result because required artifacts were incomplete");
              }
            } catch (error) {
              if (!taskRunFinalized) {
                const code = error && typeof error === "object" && "code" in error
                  && typeof error.code === "string" ? error.code : "AGENT_COMPLETION_FAILED";
                await operationalTaskClient.failTaskRun(
                  binding.sessionId,
                  activeSessionRecord.taskId,
                  activeSessionRecord.runId,
                  clientSessionId,
                  code,
                  error instanceof Error ? error.message : "Agent run completion failed",
                );
                taskRunFinalized = true;
              }
              throw error;
            }
          }
        }
        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 = readUiMessageText(latestUiMessages.at(-1)) || 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 && sessionPurpose === "ordinary_chat") {
            void learningOrchestrator.onTurnCompleted({
              ...latestTurn,
              requestContext,
              sessionId: clientSessionId,
            }).catch((error) => {
              logger.warn(
                { err: error, sessionId: clientSessionId },
                "stream-completed learning failed",
              );
            });
          }
        }
      } catch (error) {
        await stopTaskHeartbeat();
        if (
          sessionPurpose === "task_execution" &&
          activeSessionRecord.taskId
        ) {
          publish("execution_plan", buildFailedTaskExecutionPlan(executionPlanHistory));
          await taskArtifactPublishQueue;
          if (activeSessionRecord.runId && !taskLeaseFailure && !taskRunFinalized) {
            await operationalTaskClient.failTaskRun(
              binding.sessionId,
              activeSessionRecord.taskId,
              activeSessionRecord.runId,
              clientSessionId,
              abortController.signal.aborted ? "AGENT_ABORTED" : "AGENT_EXECUTION_ERROR",
              abortController.signal.aborted
                ? "Agent execution was aborted"
                : error instanceof Error ? error.message : "Agent execution failed",
            ).then(() => {
              taskRunFinalized = true;
            }).catch((finishError) => {
              logger.warn({ err: finishError, sessionId: clientSessionId }, "failed to record task run failure");
            });
          }
        }
        throw error;
      } finally {
        unsubscribeCredentialRefresh();
        abortController.signal.removeEventListener("abort", cancelCredentialRefresh);
        await stopTaskHeartbeat();
        await taskArtifactPublishQueue.catch(() => undefined);
        unregisterFrontendActionSink?.();
        if (abortController.signal.aborted && !taskLeaseFailure) {
          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);
        releaseOperationalTaskCredential(binding.sessionId);
        activeRun.status = (taskLeaseFailure || credentialFailure)
          ? "error"
          : abortController.signal.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) {
      if (startingSessionId) {
        startingSessions.delete(startingSessionId);
      }
      if (boundOperationalSessionId) {
        releaseOperationalTaskCredential(boundOperationalSessionId);
      }
      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,
      });
    }
  });

};