refactor: consolidate chat state and task bridges

Centralize session scope, chat projections and typed task bridges.
Keep historical interaction replies authoritative and publish terminal state to subscribers. Validate Server contracts before replacing generated artifacts.

src/routes/chatStreamRoute.ts

@@ -1,3 +1,4 @@
+import { toSessionScope } from "../sessions/session-scope.js";
 import type { TaskCalculationWaiter } from "../mcp/task-calculation-waiter.js";
 import { createResponsesProjection, responseRequestSchema, responsePrompt } from "../chat/responses-protocol.js";
 import { ResponseInputTracker, WaitingResponseInputQueue } from "../chat/response-inputs.js";
@@ -5,7 +6,7 @@ import { reconcileInteractionDelivery } from "../sessions/interaction-delivery-r
 import type { InteractionRequestStore } from "../sessions/interaction-request-store.js";
 import { renderInteractionAnswer, type InteractionRequest } from "../chat/interaction-request-contract.js";
 import { createInteractionReplyMessage } from "../chat/interaction-message-presentation.js";
-import { createAnswerStreamProjection } from "../chat/answer-stream-projection.js";
+import { createChatRunAccumulator, publishRunEvent } from "./chat-run-accumulator.js";
 import type { AgentMessageCoordinator } from "../chat/agent-message-coordinator.js";
 import { credentialRefreshCoordinator } from "../product/credential-refresh.js";
 import { type Router } from "express";
@@ -47,7 +48,6 @@ import {
 } 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,
@@ -59,29 +59,14 @@ import {
 } 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,
@@ -106,11 +91,6 @@ const payloadSchema = z.object({
   }).strict().optional(),
 });
 
-const toSessionUiStateContext = (sessionRecord: SessionRecord) => ({
-  sessionId: sessionRecord.sessionId,
-  actorKey: sessionRecord.actorKey,
-  projectKey: sessionRecord.projectKey,
-});
 
 const runtimeHasConversation = async (
   runtime: OpencodeRuntimeAdapter,
@@ -311,7 +291,7 @@ export const registerChatStreamRoute = (
     let boundOperationalSessionId: string | undefined;
     let startingSessionId: string | undefined;
     let continuation: InteractionRequest | null = null;
-    let continuationScope: ReturnType<typeof toSessionUiStateContext> | undefined;
+    let continuationScope: ReturnType<typeof toSessionScope> | undefined;
     let waitingInteraction: InteractionRequest | null = null;
     try {
       const authContext = getLocalAgentContext(req);
@@ -365,7 +345,7 @@ export const registerChatStreamRoute = (
         res.status(400).json({ message: "Interaction continuation requires an existing conversation" }); return;
       }
       if (existingSessionRecord && interactionRequestStore) {
-        continuationScope = toSessionUiStateContext(existingSessionRecord);
+        continuationScope = toSessionScope(existingSessionRecord);
         if (input.tjwater.interaction_id) {
           continuation = await interactionRequestStore.get(continuationScope, input.tjwater.interaction_id);
           if (!continuation?.delivery || !["answered", "skipped"].includes(continuation.status) || continuation.delivery.status === "cancelled") {
@@ -523,7 +503,7 @@ export const registerChatStreamRoute = (
         sessionId: requestContext.clientSessionId,
       };
       const initialSessionState = await sessionUiStateStore.read(
-        toSessionUiStateContext(activeSessionRecord),
+        toSessionScope(activeSessionRecord),
       );
       const persistedUiMessages = toUiMessages(initialSessionState?.messages);
       if (!continuation && input.tjwater.client_message_id && persistedUiMessages.some((message) => message.id === input.tjwater.client_message_id)) {
@@ -551,7 +531,6 @@ export const registerChatStreamRoute = (
               ...visiblePersistedMessages,
               createFallbackUserUiMessage(promptText, input.tjwater.client_message_id),
             ];
-      let latestUiMessages = requestMessages;
       const baseMessages = toLegacyMessages(requestMessages);
       const recentTurns = await sessionTranscriptStore.getRecentTurns(historyContext, 8);
 
@@ -605,44 +584,28 @@ export const registerChatStreamRoute = (
       startingSessions.delete(activeSessionRecord.sessionId);
       startingSessionId = undefined;
       lastRunStatuses.set(clientSessionId, "running");
-      const sessionUiStateContext = toSessionUiStateContext(activeSessionRecord);
+      const sessionUiStateContext = toSessionScope(activeSessionRecord);
       const assistantUiMessageId = continuationAssistantId ?? `msg_${randomUUID().replaceAll("-", "")}`;
       setRuntimeSessionContext({ ...getRuntimeSessionContext(binding.sessionId)!, hostTurnId: wire.response.id, hostMessageId: assistantUiMessageId });
-      const syncLatestUiMessages = () => {
-        latestUiMessages = [
-          ...requestMessages,
-          {
-            id: assistantUiMessageId,
-            role: "assistant",
-            parts: assistantParts.map((part) => ({ ...part })),
-          } as UIMessage,
-        ];
-      };
-      const writeUiChunk = (chunk: UIMessageChunk) => {
-        // Persist canonical parts after disconnect; the wire emitter checks the socket.
-        wire!.write(chunk);
-      };
-      const projection = createAnswerStreamProjection({
-        messageId: assistantUiMessageId, sessionId: clientSessionId,
-        write: writeUiChunk, onChange: syncLatestUiMessages,
+      const accumulator = createChatRunAccumulator({
+        run: activeRun, messages: requestMessages, messageId: assistantUiMessageId, sessionId: clientSessionId,
+        collaboration: sessionPurpose === "task_collaboration", write: (chunk) => wire!.write(chunk),
+        onStatus: (status) => lastRunStatuses.set(clientSessionId, status),
       });
-      const assistantParts = projection.parts;
-      const writeTextDelta = projection.appendText;
-      const writeDataPart = projection.appendData;
-      const finishUiMessage = projection.finish;
+      const writeUiChunk = (chunk: UIMessageChunk) => wire!.write(chunk);
       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,
+        messages: sessionPurpose === "task_collaboration" ? toCollaborationMessages(accumulator.uiMessages()) : accumulator.uiMessages(),
       });
       const queueSessionUiStatePersist = () => {
         const snapshot = {
           executionPlans: executionPlanHistory,
           sessionId: activeSessionRecord.sessionId,
           isTitleManuallyEdited: initialSessionState?.isTitleManuallyEdited ?? false,
-          messages: sessionPurpose === "task_collaboration" ? toCollaborationMessages(latestUiMessages) : latestUiMessages,
+          messages: sessionPurpose === "task_collaboration" ? toCollaborationMessages(accumulator.uiMessages()) : accumulator.uiMessages(),
         };
         persistQueue = persistQueue
           .catch((error) => {
@@ -654,36 +617,6 @@ export const registerChatStreamRoute = (
           .then(() => sessionUiStateStore.write(sessionUiStateContext, snapshot));
         return persistQueue;
       };
-      const primarySubscriber: StreamSubscriber = {
-        write: (event, data) => {
-          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);
       let taskArtifactPublishQueue: Promise<unknown> = Promise.resolve();
 
       const publish = (event: string, data: Record<string, unknown>) => {
@@ -697,131 +630,11 @@ export const registerChatStreamRoute = (
                     : 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 = data.reason === "waiting_input" ? "waiting_input" : "completed";
-          lastRunStatuses.set(clientSessionId, activeRun.status);
-          activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({
-            ...message,
-            content:
-              typeof message.content === "string" && message.content.trim()
-                ? message.content
-                : data.reason === "waiting_input" ? "等待你的回答。" : "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") {
+        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)
@@ -843,23 +656,6 @@ export const registerChatStreamRoute = (
             });
             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" && [
@@ -869,12 +665,7 @@ export const registerChatStreamRoute = (
           "ui_envelope",
           "execution_plan",
         ].includes(event);
-        if (!hiddenCollaborationEvent) {
-          for (const subscriber of activeRun.subscribers) {
-            subscriber.write(event, subscriberData);
-          }
-        }
-        activeRun.messages = updateLastAssistantMessage(activeRun.messages, (message) => ({ ...message, id: assistantUiMessageId, parts: assistantParts.map((part) => ({ ...part })) }));
+        publishRunEvent(activeRun, event, subscriberData, !hiddenCollaborationEvent);
         void queueSessionUiStatePersist().catch((error) => {
           logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist chat stream state");
         });
@@ -944,10 +735,7 @@ export const registerChatStreamRoute = (
               : typeof generation === "number" && generation > 0 ? generation : null,
             report_revision: reportRef.success ? reportRef.data.revision : null,
           };
-          const part = { type: "data-task_context" as const, data: reference };
-          assistantParts.push(part);
-          writeUiChunk(part);
-          syncLatestUiMessages();
+          accumulator.apply("task_context", reference);
         }
         const preparedMessage = taskContext
           ? [
@@ -970,11 +758,8 @@ export const registerChatStreamRoute = (
         const messaging = getTrustedProductCredential(req) ? agentMessages : undefined;
         messaging?.start(activeSessionRecord);
         const inputTracker = new ResponseInputTracker(runtime, binding.sessionId, (receipt) => {
-          if (receipt.status === "delivered" && !requestMessages.some((message) => message.id === receipt.id)) {
-            requestMessages.push({ id: receipt.id, role: "user", parts: [{ type: "text", text: receipt.text }] } as UIMessage);
-            const tail = activeRun.messages.at(-1);
-            activeRun.messages = [...activeRun.messages.slice(0, -1), { id: receipt.id, role: "user", content: receipt.text }, ...(tail ? [tail] : [])];
-            syncLatestUiMessages();
+          if (receipt.status === "delivered") {
+            accumulator.appendInput({ id: receipt.id, role: "user", parts: [{ type: "text", text: receipt.text }] } as UIMessage, receipt.text);
           }
           publish("user_input", { session_id: clientSessionId, ...receipt });
         });
@@ -1125,14 +910,14 @@ export const registerChatStreamRoute = (
           const assistantMessage = [...messages]
             .reverse()
             .find((message) => message.info.role === "assistant");
-          const assistantText = readUiMessageText(latestUiMessages.at(-1)) || collectTextContent(assistantMessage?.parts ?? []);
+          const assistantText = readUiMessageText(accumulator.uiMessages().at(-1)) || collectTextContent(assistantMessage?.parts ?? []);
           const latestSessionRecord =
             (await sessionMetadataStore.get(
               { actorKey, projectId, projectKey, userId },
               activeSessionRecord.sessionId,
             )) ?? activeSessionRecord;
           const latestSessionState = await sessionUiStateStore.read(
-            toSessionUiStateContext(latestSessionRecord),
+            toSessionScope(latestSessionRecord),
           );
           const existingSessionTitle = latestSessionRecord.title;
           let sessionTitle = existingSessionTitle;
@@ -1229,29 +1014,13 @@ export const registerChatStreamRoute = (
         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 (!projection.finished) {
-            writeUiChunk({
-              type: "error",
-              errorText: "请求已中断",
-            });
-            finishUiMessage("error");
-          }
+          publishRunEvent(activeRun, "aborted", { session_id: clientSessionId, message: "请求已中断" });
           void queueSessionUiStatePersist().catch((error) => {
             logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist aborted chat stream state");
           });
         }
-        if (!projection.finished) {
-          finishUiMessage(activeRun.status === "error" ? "error" : "stop");
+        if (!accumulator.finished) {
+          publishRunEvent(activeRun, activeRun.status === "error" ? "error" : "done", { session_id: clientSessionId });
         }
         await queueSessionUiStatePersist().catch((error) => {
           logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist chat stream state");