重构会话管理，简化上下文存储逻辑

2026-06-03 17:14:55 +08:00
parent 76d4b510f4
commit 04ded0ceb0
19 changed files with 420 additions and 235 deletions
@@ -11,6 +11,7 @@ import { type ResultReferenceResolver } from "../results/resolver.js";
 import { RESULT_REFERENCE_KIND } from "../results/store.js";
 import { type OpencodeRuntimeAdapter } from "../runtime/opencode.js";
 import { type ChatSessionBridge } from "../chat/sessionBridge.js";
+import { type ConversationRecord } from "../conversations/store.js";
 import { toActorKey, toProjectKey } from "../utils/fileStore.js";
 import {
  buildPromptWithLearningContext,
@@ -51,6 +52,12 @@ const conversationStateSchema = z.object({
  branch_groups: z.array(z.unknown()).default([]),
 });

+const toConversationStateContext = (conversation: ConversationRecord) => ({
+  actorKey: conversation.actorKey,
+  projectKey: conversation.projectKey,
+  sessionId: conversation.sessionId,
+});
+
 export const buildChatRouter = (
  sessionBridge: ChatSessionBridge,
  runtime: OpencodeRuntimeAdapter,
@@ -145,7 +152,9 @@ export const buildChatRouter = (
      return;
    }

-    const state = await conversationStateStore.read(conversation.sessionScopeKey);
+    const state = await conversationStateStore.read(
+      toConversationStateContext(conversation),
+    );
    res.json({
      id: conversation.sessionId,
      title: conversation.title ?? "新对话",
@@ -190,7 +199,7 @@ export const buildChatRouter = (
    const nextRecord = await conversationStore.touch(record, {
      ...(parsed.data.title ? { title: parsed.data.title } : {}),
    });
-    await conversationStateStore.write(nextRecord.sessionScopeKey, {
+    await conversationStateStore.write(toConversationStateContext(nextRecord), {
      sessionId: nextRecord.sessionId,
      isTitleManuallyEdited: parsed.data.is_title_manually_edited,
      messages: parsed.data.messages,
@@ -231,13 +240,18 @@ export const buildChatRouter = (
      return;
    }
    const nextConversation = await conversationStore.touch(conversation, { title });
-    const state = await conversationStateStore.read(nextConversation.sessionScopeKey);
+    const state = await conversationStateStore.read(
+      toConversationStateContext(nextConversation),
+    );
    if (state) {
-      await conversationStateStore.write(nextConversation.sessionScopeKey, {
+      await conversationStateStore.write(
+        toConversationStateContext(nextConversation),
+        {
        ...state,
        isTitleManuallyEdited:
          isTitleManuallyEdited ?? state.isTitleManuallyEdited,
-      });
+        },
+      );
    }
    res.json({
      id: nextConversation.sessionId,
@@ -264,7 +278,13 @@ export const buildChatRouter = (
      res.status(204).end();
      return;
    }
-    await conversationStateStore.remove(conversation.sessionScopeKey);
+    await conversationStateStore.remove(toConversationStateContext(conversation));
+    if (conversation.opencodeSessionId) {
+      await sessionBridge.deleteConversationSession({
+        clientSessionId: conversation.sessionId,
+        sessionId: conversation.opencodeSessionId,
+      });
+    }
    await conversationStore.remove(conversation);
    res.status(204).end();
  });
@@ -323,9 +343,20 @@ export const buildChatRouter = (
    }

    try {
-      const binding = await sessionBridge.abort({
-        clientSessionId: parsed.data.session_id,
-      });
+      const projectId = req.header("x-project-id") ?? undefined;
+      const userId = req.header("x-user-id") ?? undefined;
+      const actorKey = toActorKey(userId);
+      const projectKey = toProjectKey(projectId);
+      const conversation = await conversationStore.get(
+        { actorKey, projectId, projectKey, userId },
+        parsed.data.session_id,
+      );
+      const binding = conversation?.opencodeSessionId
+        ? await sessionBridge.abort({
+            clientSessionId: conversation.sessionId,
+            sessionId: conversation.opencodeSessionId,
+          })
+        : null;

      if (!binding) {
        res.status(204).end();
@@ -467,14 +498,22 @@ export const buildChatRouter = (
          userId,
        });
      const activeConversation = await conversationStore.touch(conversation);
+      const hadExistingRuntimeSession = Boolean(activeConversation.opencodeSessionId);

      const { binding, requestContext, created } = await sessionBridge.resolve({
        clientSessionId: activeConversation.sessionId,
+        sessionId: activeConversation.opencodeSessionId,
        accessToken,
        projectId,
        traceId,
        userId,
      });
+      const conversationWithRuntime =
+        created && binding.sessionId !== activeConversation.opencodeSessionId
+          ? await conversationStore.touch(activeConversation, {
+              opencodeSessionId: binding.sessionId,
+            })
+          : activeConversation;
      const historyContext = {
        actorKey: requestContext.actorKey,
        clientSessionId: requestContext.clientSessionId,
@@ -482,6 +521,9 @@ export const buildChatRouter = (
        sessionId: requestContext.clientSessionId,
      };
      const recentTurns = await sessionHistoryStore.getRecentTurns(historyContext, 8);
+      const initialConversationState = await conversationStateStore.read(
+        toConversationStateContext(conversationWithRuntime),
+      );

      logger.info(
        {
@@ -521,8 +563,12 @@ export const buildChatRouter = (
          memoryStore,
          requestContext.actorKey,
          requestContext.projectKey,
-          recentTurns,
-          parsed.data.message,
+          {
+            recentTurns,
+            persistedMessages: initialConversationState?.messages,
+            message: parsed.data.message,
+            restoreConversation: !hadExistingRuntimeSession,
+          },
        );
        const streamResult = await streamPromptResponse({
          runtime,
@@ -550,10 +596,10 @@ export const buildChatRouter = (
          const latestConversation =
            (await conversationStore.get(
              { actorKey, projectId, projectKey, userId },
-              activeConversation.sessionId,
-            )) ?? activeConversation;
+              conversationWithRuntime.sessionId,
+            )) ?? conversationWithRuntime;
          const latestConversationState = await conversationStateStore.read(
-            latestConversation.sessionScopeKey,
+            toConversationStateContext(latestConversation),
          );
          const existingSessionTitle = latestConversation.title;
          let sessionTitle = existingSessionTitle;
@@ -606,8 +652,7 @@ export const buildChatRouter = (
          }
        }
      } finally {
-        await sessionBridge.releaseRuntimeSession(clientSessionId, binding.sessionId);
-        sessionBridge.deleteAbortController(clientSessionId);
+        sessionBridge.finalizeRequest(clientSessionId);
        streamClosed = true;
        req.off("close", handleClientClose);
        res.off("close", handleClientClose);
@@ -192,15 +192,22 @@ export const buildPromptWithLearningContext = async (
  memoryStore: MemoryStore,
  actorKey: string,
  projectKey: string,
-  recentTurns: SessionTurnRecord[],
-  message: string,
+  options: {
+    recentTurns: SessionTurnRecord[];
+    persistedMessages?: unknown[];
+    message: string;
+    restoreConversation?: boolean;
+  },
 ) => {
  const snapshot = await memoryStore.buildPromptSnapshot({ actorKey, projectKey });
-  const restoredConversation = buildRestoredConversationContext(recentTurns);
+  const restoredConversation = options.restoreConversation === false
+    ? ""
+    : buildRestoredConversationFromMessages(options.persistedMessages) ||
+      buildRestoredConversationContext(options.recentTurns);
  if (!snapshot && !restoredConversation) {
-    return message;
+    return options.message;
  }
-  return [snapshot, restoredConversation, `[Current user request]\n${message}`]
+  return [snapshot, restoredConversation, `[Current user request]\n${options.message}`]
    .filter(Boolean)
    .join("\n\n");
 };
@@ -239,4 +246,57 @@ const compactMessage = (value: string) => {
  return normalized.length > RESTORE_MESSAGE_CHAR_LIMIT
    ? `${normalized.slice(0, RESTORE_MESSAGE_CHAR_LIMIT - 3)}...`
    : normalized;
+};
+
+const isObjectRecord = (value: unknown): value is Record<string, unknown> =>
+  typeof value === "object" && value !== null && !Array.isArray(value);
+
+const isSyntheticAssistantError = (content: string) =>
+  /^⚠️\s*\*\*(请求已中断|错误[:：]?)/.test(content);
+
+const buildRestoredConversationFromMessages = (messages: unknown[] | undefined) => {
+  if (!Array.isArray(messages) || messages.length === 0) {
+    return "";
+  }
+
+  const formattedMessages = messages
+    .slice(-(RESTORE_TURN_LIMIT * 2 + 2))
+    .flatMap((message) => {
+      if (!isObjectRecord(message)) {
+        return [];
+      }
+
+      const role = message.role;
+      const content = message.content;
+      if ((role !== "user" && role !== "assistant") || typeof content !== "string") {
+        return [];
+      }
+
+      const normalizedContent = compactMessage(content);
+      if (!normalizedContent) {
+        return [];
+      }
+
+      if (role === "assistant" && isSyntheticAssistantError(normalizedContent)) {
+        return [];
+      }
+
+      return [`${role === "user" ? "用户" : "助手"}：${normalizedContent}`];
+    });
+
+  if (formattedMessages.length === 0) {
+    return "";
+  }
+
+  const conversation = formattedMessages.join("\n");
+  const trimmedConversation =
+    conversation.length > RESTORE_CONTEXT_CHAR_LIMIT
+      ? `${conversation.slice(0, RESTORE_CONTEXT_CHAR_LIMIT - 3)}...`
+      : conversation;
+
+  return [
+    "[Previous conversation context]",
+    "以下为当前前端对话线程中最近的历史对话，请延续其中已确认的目标、约束、结论与引用结果。",
+    trimmedConversation,
+  ].join("\n");
 };