fix(chat): 修复跨端会话同步和搜索合并

Author: yyg-max

crates/agent-gateway/internal/proto/v1/gateway.pb.go

@@ -323,6 +323,8 @@ func (c *websocketConnection) dispatch(req websocketRequest) {
 		c.handleFsListDirs(req)
 	case "history.list":
 		c.handleHistoryList(req)
+	case "history.shared_list":
+		c.handleHistorySharedList(req)
 	case "history.get":
 		c.handleHistoryGet(req)
 	case "history.rename":
@@ -546,6 +548,79 @@ func (c *websocketConnection) handleHistoryList(req websocketRequest) {
 	})
 }
 
+func (c *websocketConnection) handleHistorySharedList(req websocketRequest) {
+	type payload struct {
+		Page     int `json:"page"`
+		PageSize int `json:"page_size"`
+	}
+
+	var body payload
+	if err := decodeWebSocketPayload(req.Payload, &body); err != nil {
+		_ = c.writeError(req.ID, "invalid history.shared_list payload")
+		return
+	}
+	page := body.Page
+	if page <= 0 {
+		_ = c.writeError(req.ID, "history.shared_list page must be greater than 0")
+		return
+	}
+	pageSize := body.PageSize
+	if pageSize <= 0 {
+		_ = c.writeError(req.ID, "history.shared_list page_size must be greater than 0")
+		return
+	} else if pageSize > maxHistoryListLimit {
+		pageSize = maxHistoryListLimit
+	}
+
+	argsJSON, err := json.Marshal(map[string]any{
+		"page":      page,
+		"page_size": pageSize,
+	})
+	if err != nil {
+		_ = c.writeError(req.ID, "invalid history.shared_list payload")
+		return
+	}
+
+	response, err := c.awaitAgentResponse(req.ID, &gatewayv1.GatewayEnvelope{
+		RequestId: req.ID,
+		Timestamp: time.Now().Unix(),
+		Payload: &gatewayv1.GatewayEnvelope_MemoryManage{
+			MemoryManage: &gatewayv1.MemoryManageRequest{
+				Command:  "history_shared_list",
+				ArgsJson: string(argsJSON),
+			},
+		},
+	})
+	if err != nil {
+		_ = c.writeError(req.ID, websocketErrorMessage(err))
+		return
+	}
+	if errResp := response.GetError(); errResp != nil {
+		_ = c.writeError(req.ID, errResp.GetMessage())
+		return
+	}
+
+	resp := response.GetMemoryManageResp()
+	if resp == nil {
+		_ = c.writeError(req.ID, "unexpected agent response")
+		return
+	}
+
+	var result struct {
+		Conversations []map[string]any `json:"conversations"`
+		TotalCount    int              `json:"total_count"`
+	}
+	if err := json.Unmarshal([]byte(resp.GetResultJson()), &result); err != nil {
+		_ = c.writeError(req.ID, "invalid history.shared_list response")
+		return
+	}
+
+	_ = c.writeResponse(req.ID, map[string]any{
+		"conversations": result.Conversations,
+		"total_count":   result.TotalCount,
+	})
+}
+
 func (c *websocketConnection) handleHistoryGet(req websocketRequest) {
 	type payload struct {
 		ConversationID string `json:"conversation_id"`

crates/agent-gateway/internal/proto/v1/gateway_grpc.pb.go

@@ -798,7 +798,9 @@ func (m *Manager) broadcastChatEvent(requestID string, event *gatewayv1.ChatEven
 	}
 
 	requestID = strings.TrimSpace(requestID)
+	conversationID := strings.TrimSpace(event.GetConversationId())
 	now := time.Now()
+	sessionEpoch := m.currentSessionEpoch()
 
 	m.chatMu.Lock()
 	m.pruneExpiredChatRunsLocked(now)
@@ -807,10 +809,24 @@ func (m *Manager) broadcastChatEvent(requestID string, event *gatewayv1.ChatEven
 		Event:     event,
 	}
 	var runSubscribers []*chatRunSubscriber
-	if run := m.chatRuns[requestID]; run != nil {
+	run := m.chatRuns[requestID]
+	if run == nil && requestID != "" {
+		run = &chatRun{
+			requestID:      requestID,
+			conversationID: conversationID,
+			sessionEpoch:   sessionEpoch,
+			updatedAt:      now,
+			subscribers:    make(map[int]*chatRunSubscriber),
+		}
+		m.chatRuns[requestID] = run
+		if conversationID != "" {
+			m.chatRunByConversation[conversationID] = requestID
+		}
+	}
+	if run != nil {
 		run.nextSeq += 1
 		run.updatedAt = now
-		if conversationID := strings.TrimSpace(event.GetConversationId()); conversationID != "" {
+		if conversationID != "" {
 			if run.conversationID != "" && run.conversationID != conversationID {
 				if m.chatRunByConversation[run.conversationID] == requestID {
 					delete(m.chatRunByConversation, run.conversationID)

crates/agent-gateway/internal/server/websocket.go

@@ -2,7 +2,7 @@ import fs from "node:fs";
 import path from "node:path";
 import vm from "node:vm";
 import { createRequire } from "node:module";
-import { pathToFileURL } from "node:url";
+import { fileURLToPath, pathToFileURL } from "node:url";
 
 const DEFAULT_EXTENSIONS = [".ts", ".tsx", ".js", ".jsx", ".mjs", ".cjs", ".json", ".css"];
 
@@ -106,7 +106,7 @@ function resolveAsFileOrDirectory(candidate) {
 export function createWebModuleLoader(options = {}) {
   const rootDir = options.rootDir
     ? path.resolve(options.rootDir)
-    : path.resolve(new URL("../../web", import.meta.url).pathname);
+    : path.resolve(fileURLToPath(new URL("../../web", import.meta.url)));
   const requireFromRoot = createRequire(path.join(rootDir, "package.json"));
   const ts = requireFromRoot("typescript");
   const cache = new Map();

crates/agent-gateway/internal/session/manager.go

@@ -252,6 +252,49 @@ func TestStartChatRunWithClientRequestReusesExistingRun(t *testing.T) {
 	}
 }
 
+func TestDesktopBroadcastChatEventCreatesAttachableRun(t *testing.T) {
+	t.Parallel()
+
+	sm := newTestSessionManager()
+	sm.SetSession(session.NewAgentSession(sm.LatestAuthSnapshot()))
+
+	sm.DispatchFromAgent(&gatewayv1.AgentEnvelope{
+		RequestId: "conversation-live-conversation-1",
+		Payload: &gatewayv1.AgentEnvelope_ChatEvent{
+			ChatEvent: &gatewayv1.ChatEvent{
+				Type:           gatewayv1.ChatEvent_TOKEN,
+				ConversationId: "conversation-1",
+				Data:           `{"text":"hello"}`,
+			},
+		},
+	})
+
+	ch, done, cleanup, snapshot, err := sm.SubscribeChatRun("", "conversation-1", 0)
+	if err != nil {
+		t.Fatalf("SubscribeChatRun: %v", err)
+	}
+	defer cleanup()
+	assertDoneOpen(t, done)
+	if snapshot.RequestID != "conversation-live-conversation-1" {
+		t.Fatalf("snapshot request id = %q, want conversation-live-conversation-1", snapshot.RequestID)
+	}
+
+	select {
+	case event := <-ch:
+		if event.Seq != 1 {
+			t.Fatalf("event seq = %d, want 1", event.Seq)
+		}
+		if event.Event.GetType() != gatewayv1.ChatEvent_TOKEN {
+			t.Fatalf("event type = %v, want TOKEN", event.Event.GetType())
+		}
+		if event.Event.GetConversationId() != "conversation-1" {
+			t.Fatalf("conversation id = %q, want conversation-1", event.Event.GetConversationId())
+		}
+	case <-time.After(time.Second):
+		t.Fatalf("timed out waiting for replayed desktop chat event")
+	}
+}
+
 func TestCompletedHistoryUpsertDoesNotPreemptTerminalChatEvent(t *testing.T) {
 	t.Parallel()
 

crates/agent-gateway/test/helpers/load-web-module.mjs

@@ -896,6 +896,49 @@ func TestWebSocketForwardsHistorySettingsAndFsRPCs(t *testing.T) {
 		t.Fatalf("history conversation share field = %#v", historyConversation)
 	}
 
+	sendEnvelope(t, conn, "history-shared-1", "history.shared_list", map[string]any{
+		"page":      1,
+		"page_size": 50,
+	})
+	sharedHistoryOutbound := readOutboundEnvelope(t, agentSession)
+	sharedHistoryReq := sharedHistoryOutbound.GetMemoryManage()
+	if sharedHistoryReq == nil {
+		t.Fatalf("shared history outbound payload = %T, want MemoryManageRequest", sharedHistoryOutbound.GetPayload())
+	}
+	if sharedHistoryReq.GetCommand() != "history_shared_list" {
+		t.Fatalf("shared history list request = %#v", sharedHistoryReq)
+	}
+	var sharedHistoryArgs map[string]any
+	if err := json.Unmarshal([]byte(sharedHistoryReq.GetArgsJson()), &sharedHistoryArgs); err != nil {
+		t.Fatalf("decode shared history args: %v", err)
+	}
+	if sharedHistoryArgs["page"] != float64(1) || sharedHistoryArgs["page_size"] != float64(50) {
+		t.Fatalf("shared history args = %#v", sharedHistoryArgs)
+	}
+	sm.DispatchFromAgent(&gatewayv1.AgentEnvelope{
+		RequestId: sharedHistoryOutbound.GetRequestId(),
+		Timestamp: time.Now().Unix(),
+		Payload: &gatewayv1.AgentEnvelope_MemoryManageResp{
+			MemoryManageResp: &gatewayv1.MemoryManageResponse{
+				ResultJson: `{"total_count":1,"conversations":[{"id":"conversation-1","title":"Gateway test","created_at":10,"updated_at":11,"message_count":3,"provider_id":"codex-provider","model":"gpt-test","session_id":"session-1","cwd":"/workspace","is_shared":true}]}`,
+			},
+		},
+	})
+	sharedHistoryResponse := receiveEnvelope(t, conn)
+	if sharedHistoryResponse.ID != "history-shared-1" || sharedHistoryResponse.Type != "response" {
+		t.Fatalf("shared history response = %#v", sharedHistoryResponse)
+	}
+	var sharedHistoryPayload map[string]any
+	if err := json.Unmarshal(sharedHistoryResponse.Payload, &sharedHistoryPayload); err != nil {
+		t.Fatalf("decode shared history response: %v", err)
+	}
+	if sharedHistoryPayload["total_count"] != float64(1) {
+		t.Fatalf("shared history payload = %#v", sharedHistoryPayload)
+	}
+	if _, ok := sharedHistoryPayload["running_conversation_ids"]; ok {
+		t.Fatalf("shared history response should not include running ids: %#v", sharedHistoryPayload)
+	}
+
 	sendEnvelope(t, conn, "history-get-1", "history.get", map[string]any{
 		"conversation_id": "conversation-1",
 		"max_messages":    360,

crates/agent-gateway/test/session/manager_test.go

@@ -581,6 +581,43 @@ test("GatewayWebSocketClient sends skill manage payloads", async () => {
   resetGatewayWebSocketClient();
 });
 
+test("GatewayWebSocketClient sends history list requests", async () => {
+  installBrowser();
+  const loader = createWebModuleLoader();
+  const { getGatewayWebSocketClient, resetGatewayWebSocketClient } = loader.loadModule("src/lib/gatewaySocket.ts");
+  resetGatewayWebSocketClient();
+
+  const client = getGatewayWebSocketClient("token");
+  const listPromise = client.listHistory(2, 50);
+  const socket = await connectAndAuth();
+  await waitFor(() => socket.sent.length >= 2, "history list envelope");
+  assert.equal(socket.sent[1].type, "history.list");
+  assert.deepEqual(socket.sent[1].payload, { page: 2, page_size: 50 });
+  socket.receive({
+    id: socket.sent[1].id,
+    type: "response",
+    payload: { conversations: [], total_count: 0, running_conversation_ids: [] },
+  });
+  assert.deepEqual(await listPromise, {
+    conversations: [],
+    total_count: 0,
+    running_conversation_ids: [],
+  });
+
+  const sharedListPromise = client.listSharedHistory(1, 25);
+  await waitFor(() => socket.sent.length >= 3, "shared history list envelope");
+  assert.equal(socket.sent[2].type, "history.shared_list");
+  assert.deepEqual(socket.sent[2].payload, { page: 1, page_size: 25 });
+  socket.receive({
+    id: socket.sent[2].id,
+    type: "response",
+    payload: { conversations: [], total_count: 0 },
+  });
+  assert.deepEqual(await sharedListPromise, { conversations: [], total_count: 0 });
+
+  resetGatewayWebSocketClient();
+});
+
 test("GatewayWebSocketClient sends history share requests", async () => {
   installBrowser();
   const loader = createWebModuleLoader();