fix(web): page remote histories and preserve Harness identity

docs/product/standalone-web.md

@@ -66,7 +66,11 @@ Linux 的 Web 与 Desktop 环境不同时，发现过程还检查当前用户标
 
 支持读取会话、空闲时文本续聊、停止当前 Turn、改名、归档，以及 manager 提供原生 pending request 和 responder 时的命令/文件变更审批（允许一次、拒绝）。线程 resume 和 turn/start 使用原生 ID，owner 和鉴权由原生 manager 保持。缺少原生审批方法的旧 manager 或其他交互类型会提示回 Desktop 操作，不伪造权限能力。远程文件上传、终端、本机路径打开、模型/权限切换、fork、编辑、队列和 steering 不提供；这些请求不会落到本机 Thread handler。远程项目来自 Desktop 保存的 remote-projects，按 hostId+项目 ID 隔离；归属优先使用原生项目分配，其次在该主机已登记项目的路径范围内匹配，未匹配的会话保留在对应主机分组。分组仅用于只读导航，不把远程路径交给本机目录创建、终端或文件操作。
 
-列表每个 Host 最多读取 100 页，每页 100 条；超过限制不发布不完整目录。打开的会话每 1.5 秒读取不消费事件的绝对快照，多个浏览器独立可见；关闭跟随即停止读取。快照保留原生 Turn/Item ID，单次 Controller 响应上限 16 MiB，超大完整历史当前不可用，应在 Codex Desktop 打开；远程快照尚无端到端分段读取能力，调整浏览器消息页大小不能绕过这一上限。连接失败保留已见列表并标记离线；已打开的跟随保留订阅，读取按指数退避重试，最长间隔 30 秒，连接恢复后无需重新打开。关闭视图后停止重试。连接代际变化会使旧审批失效，重建浏览器投影。所有写请求只发送一次，超时或断线报告结果未知，不自动重发消息或审批；读后同步失败不否认已确认的写入。
+列表每个 Host 最多读取 100 页，每页 100 条；超过限制不发布不完整目录。远程外部会话的 Harness 标识和显示名来自同一 Host 的归属及插件目录，用于显示真实图标；查询不支持时保留目录，不根据模型名称猜测归属。
+
+打开的会话每 1.5 秒读取不消费事件的绝对状态，多个浏览器独立可见；关闭跟随即停止读取。分页会话先读取元数据，再通过同一原生 manager 的 `thread/turns/list` 每次加载五轮，复用已有历史分页、重连与滚动锚点；旧会话保留完整快照读取。发送前使用 `thread/resume {threadId, excludeTurns: true}` 建立订阅，不重新传输完整历史。快照保留原生 Turn/Item ID，单次 Controller 响应上限仍为 16 MiB；超大单页明确报错，不因此将整个 Host 标记为离线。
+
+连接失败保留已见列表并标记离线；已打开的跟随保留订阅，读取按指数退避重试，最长间隔 30 秒，连接恢复后无需重新打开。关闭视图后停止重试。连接代际变化会使旧审批失效，重建浏览器投影。所有写请求只发送一次，超时或断线报告结果未知，不自动重发消息或审批；读后同步失败不否认已确认的写入。
 
 本机 native Codex 目录及其共享 app-server 生命周期由本机来源负责，与此远程聚合独立。隔离 fixtures 覆盖复合身份、断线、代际、未知结果和审批路由；这些检查不代替 Windows/macOS 上真实 SSH、RC 活跃 Turn、Desktop 与多浏览器同时响应的实机验收。
 

packages/desktop-control/package.json

@@ -21,6 +21,7 @@
     "build:release": "node scripts/build-release.mjs"
   },
   "dependencies": {
-    "@codexhost/shared-contracts": "0.0.0"
+    "@codexhost/shared-contracts": "0.0.0",
+    "ws": "8.21.3"
   }
 }

packages/desktop-control/scripts/build-release.mjs

@@ -49,6 +49,9 @@ export async function buildDesktopControllerBundle({ repositoryRoot, outputPath
     bundle: true,
     platform: "node",
     format: "esm",
+    banner: {
+      js: 'import { createRequire } from "node:module"; const require = createRequire(import.meta.url);',
+    },
     target: "node22",
     packages: "bundle",
     sourcemap: false,

packages/desktop-control/src/cdp-client.ts

@@ -1,3 +1,5 @@
+import WebSocket from "ws";
+
 export interface CdpTarget {
   id: string;
   type: string;
@@ -138,7 +140,7 @@ function defaultFetch(url: string, init?: { signal?: AbortSignal }): Promise<Cdp
 }
 
 function defaultSocketFactory(url: string): CdpSocket {
-  return new WebSocket(url) as unknown as CdpSocket;
+  return new WebSocket(url, { maxPayload: 32 * 1024 * 1024 }) as unknown as CdpSocket;
 }
 
 function messageText(value: unknown): string {

packages/desktop-control/src/remote-connections-control.ts

@@ -14,8 +14,16 @@ const MAX_REPLY_BYTES = 1024 * 1024;
 /** Fixed renderer API; user input is data, never executable JavaScript. */
 export function remoteConnectionsExpression(input: RemoteConnectionsRequest): string {
   const request = remoteConnectionsRequestSchema.parse(input);
-  if (request.action === "session")
-    return `(async () => { try { return {result:await (${rendererRemoteSession.toString()})(window,${JSON.stringify(request.operation)})}; } catch(error) {return {error:{code:-32090,message:String(error?.message ?? error)}};} })()`;
+  if (request.action === "session") {
+    const readOnly = ["hosts", "list", "read", "turns"].includes(request.operation.kind);
+    // Reject oversized history before it can close the shared CDP transport.
+    return `(async () => { try {
+      const reply = {result:await (${rendererRemoteSession.toString()})(window,${JSON.stringify(request.operation)})};
+      if (${readOnly} && new TextEncoder().encode(JSON.stringify(reply)).byteLength + 1 > ${REMOTE_THREAD_REPLY_MAX_BYTES})
+        return {error:{code:-32094,message:"Remote history response exceeds the 16 MiB limit; open this Thread in Codex Desktop."}};
+      return reply;
+    } catch(error) {return {error:{code:-32090,message:String(error?.message ?? error)}};} })()`;
+  }
   return `(async () => {
     try {
       const api = window.__codexhostRendererBindingProbeV1;

packages/desktop-control/src/remote-session-client.ts

@@ -79,7 +79,8 @@ export async function requestRemoteSession(
     },
     { action: "session", operation: input },
   );
-  if ("error" in reply) throw new Error(reply.error.message);
+  if ("error" in reply)
+    throw Object.assign(new Error(reply.error.message), { code: reply.error.code });
   if (input.kind === "hosts" && Array.isArray(reply.result)) {
     const projects = await readRemoteProjectMetadata().catch(() => []);
     return reply.result.map((host) =>

packages/desktop-control/src/renderer-remote-sessions.ts

@@ -95,13 +95,60 @@ export async function rendererRemoteSession(
   }
   if ("epoch" in operation && operation.epoch !== generation.epoch)
     throw new Error("Remote connection changed; refresh before operating");
-  if (operation.kind === "list")
-    return manager.sendRequest("thread/list", {
+  if (operation.kind === "list") {
+    const page = (await manager.sendRequest("thread/list", {
       cursor: operation.cursor,
       archived: operation.archived,
       limit: 100,
-    });
+    })) as { data: Array<{ id: string; modelProvider: string }>; nextCursor: string | null };
+    const threadIds = page.data
+      .filter((row) => row.modelProvider === "codexhost")
+      .map((row) => row.id);
+    if (!threadIds.length) return page;
+    // Ownership is independent of model/provider names, and never requires history reads.
+    const [ownership, catalog] = await Promise.all([
+      Promise.resolve(manager.sendRequest("codexhost/thread/ownership/list", { threadIds })).catch(
+        () => null,
+      ),
+      Promise.resolve(manager.sendRequest("codexhost/harness/plugins/list", {})).catch(() => null),
+    ]);
+    if (routing.forHost(operation.hostId) !== route)
+      throw new Error("Remote connection changed during ownership read");
+    const owners = (
+      ownership as {
+        threads?: Array<{ threadId: string; owner: string; harnessId?: string }>;
+      } | null
+    )?.threads;
+    const plugins = (catalog as { plugins?: Array<{ id: string; name: string }> } | null)?.plugins;
+    return {
+      ...page,
+      data: page.data.map((row) => {
+        const owner = Array.isArray(owners)
+          ? owners.find((entry) => entry?.threadId === row.id && entry.owner === "external")
+          : undefined;
+        const id = owner?.harnessId;
+        if (row.modelProvider !== "codexhost" || typeof id !== "string" || !id) return row;
+        const plugin = Array.isArray(plugins)
+          ? plugins.find((entry) => entry?.id === id)
+          : undefined;
+        const name = typeof plugin?.name === "string" && plugin.name ? plugin.name : id;
+        return { ...row, harnessIdentity: { id, name } };
+      }),
+    };
+  }
   const threadId = operation.threadId;
+  if (operation.kind === "turns") {
+    const page = await manager.sendRequest("thread/turns/list", {
+      threadId,
+      cursor: operation.cursor,
+      limit: 5,
+      sortDirection: "desc",
+      itemsView: "full",
+    });
+    if (routing.forHost(operation.hostId) !== route)
+      throw new Error("Remote connection changed during history read");
+    return page;
+  }
   const pendingRequests = async (): Promise<Request[]> => {
     const conversation = (await manager.getConversation?.(threadId)) as
       { requests?: Request[] } | undefined;
@@ -110,7 +157,21 @@ export async function rendererRemoteSession(
     return Array.isArray(conversation?.requests) ? conversation.requests : [];
   };
   if (operation.kind === "read") {
-    const result = await manager.sendRequest("thread/read", { threadId, includeTurns: true });
+    let result;
+    try {
+      result = await manager.sendRequest("thread/read", {
+        threadId,
+        includeTurns: operation.metadataOnly !== true,
+      });
+    } catch (error) {
+      if (
+        operation.metadataOnly ||
+        !(error instanceof Error) ||
+        !error.message.includes("Paginated External Threads require thread/turns/list")
+      )
+        throw error;
+      result = await manager.sendRequest("thread/read", { threadId, includeTurns: false });
+    }
     if (routing.forHost(operation.hostId) !== route)
       throw new Error("Remote connection changed during read");
     const requests = await pendingRequests();
@@ -161,7 +222,10 @@ export async function rendererRemoteSession(
   }
   switch (operation.kind) {
     case "prompt":
-      await manager.sendRequest("thread/resume", { threadId });
+      // Resuming is required before a native turn, but Web already reads history
+      // in pages. Returning every old Turn here can overwhelm the shared native
+      // connection before the new message is even dispatched.
+      await manager.sendRequest("thread/resume", { threadId, excludeTurns: true });
       if (routing.forHost(operation.hostId) !== route)
         throw new Error("Remote connection changed during resume");
       return manager.sendRequest("turn/start", {

packages/desktop-control/test/cdp-client-large-response.test.ts

@@ -0,0 +1,163 @@
+import { once } from "node:events";
+import { mkdtemp, readFile, rm } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import path from "node:path";
+import { chromium } from "@playwright/test";
+import type { AddressInfo } from "node:net";
+import { WebSocketServer } from "ws";
+import { expect, it } from "vitest";
+import { CdpClient } from "../src/cdp-client.js";
+import { remoteConnectionsExpression } from "../src/remote-connections-control.js";
+import type { RemoteSessionOperation } from "@codexhost/shared-contracts";
+
+it("receives a fragmented multi-megabyte history response and keeps the CDP connection usable", async () => {
+  const server = new WebSocketServer({ host: "127.0.0.1", port: 0, perMessageDeflate: false });
+  await once(server, "listening");
+  const history = "history-历史-".repeat(600_000);
+  server.on("connection", (socket) => {
+    socket.on("message", (data) => {
+      const request = JSON.parse(String(data)) as { id: number; method: string };
+      if (request.method !== "Runtime.evaluate") {
+        socket.send(JSON.stringify({ id: request.id, result: { accepted: true } }));
+        return;
+      }
+      const reply = Buffer.from(
+        JSON.stringify({
+          id: request.id,
+          result: { result: { type: "object", value: { history } } },
+        }),
+      );
+      // CDP can fragment a large response, including across a UTF-8 code point.
+      for (let offset = 0; offset < reply.length; offset += 1_048_576) {
+        socket.send(reply.subarray(offset, offset + 1_048_576), {
+          binary: false,
+          fin: offset + 1_048_576 >= reply.length,
+        });
+        if (offset === 0) socket.ping();
+      }
+    });
+  });
+  let client: CdpClient | undefined;
+  try {
+    const address = server.address() as AddressInfo;
+    client = await CdpClient.connect(`ws://127.0.0.1:${address.port}/devtools/page/history`, {
+      commandTimeoutMs: 10_000,
+    });
+    expect((await client.evaluate<{ history: string }>("history")).history).toBe(history);
+    await expect(client.command("Runtime.enable")).resolves.toEqual({ accepted: true });
+  } finally {
+    client?.close();
+    for (const socket of server.clients) socket.terminate();
+    await new Promise<void>((resolve, reject) =>
+      server.close((error) => (error ? reject(error) : resolve())),
+    );
+  }
+}, 15_000);
+
+it.skipIf(!process.env.CODEXHOST_TEST_CDP_BROWSER)(
+  "receives a large real Chromium CDP result",
+  async () => {
+    const directory = await mkdtemp(path.join(tmpdir(), "codexhost-cdp-history-"));
+    const executablePath = process.env.CODEXHOST_TEST_CDP_BROWSER;
+    const browser = await chromium.launchPersistentContext(directory, {
+      ...(executablePath ? { executablePath } : {}),
+      headless: true,
+      args: ["--remote-debugging-port=0", "--remote-debugging-address=127.0.0.1"],
+    });
+    let client: CdpClient | undefined;
+    try {
+      const port = (await readFile(path.join(directory, "DevToolsActivePort"), "utf8")).split(
+        "\n",
+      )[0];
+      await browser.pages()[0]?.goto(`http://127.0.0.1:${port}/json/version`);
+      const targets = (await (await fetch(`http://127.0.0.1:${port}/json/list`)).json()) as Array<{
+        type: string;
+        webSocketDebuggerUrl: string;
+      }>;
+      const target = targets.find((candidate) => candidate.type === "page");
+      if (!target) throw new Error("Missing isolated Chromium page");
+      client = await CdpClient.connect(target.webSocketDebuggerUrl);
+      const value = await client.evaluate<string>("'history-\\u5386\\u53f2-'.repeat(600000)");
+      expect(value.length).toBe("history-历史-".length * 600_000);
+      await expect(client.evaluate("6*7")).resolves.toBe(42);
+      await client.evaluate(`(()=>{
+        window.fixtureSends=[];
+        const manager={
+          getConversation:()=>({requests:[]}),
+          async sendRequest(method,params){
+            if(method==='thread/resume'&&params.excludeTurns!==true)
+              throw Error('Native resume must not transfer unbounded history');
+            if(method==='thread/read') return {thread:{id:params.threadId,turns:[{
+              id:'history',items:[{type:'agentMessage',text:params.threadId==='oversized'
+                ? 'x'.repeat(17*1024*1024)
+                : window.fixtureSends.length?'history-\\u5386\\u53f2-'.repeat(600000):'small'}]
+            }]}};
+            if(method==='turn/start'){window.fixtureSends.push(params.threadId);return {accepted:true}}
+            return {};
+          }
+        };
+        const route={manager};
+        window.__codexhostHostRoutingV1={knownHostIds:()=>['remote-ssh-discovered:fixture'],forHost:()=>route};
+        return true;
+      })()`);
+      const activeClient = client;
+      const call = (operation: RemoteSessionOperation) =>
+        activeClient.evaluate<{
+          result?: { epoch: string; result: { thread: { id: string } } };
+          error?: { code: number };
+        }>(remoteConnectionsExpression({ action: "session", operation }));
+      for (const threadId of ["a", "b", "c", "a"]) {
+        const snapshot = await call({
+          kind: "read",
+          hostId: "remote-ssh-discovered:fixture",
+          threadId,
+        });
+        expect(snapshot.error).toBeUndefined();
+        expect(snapshot.result?.result.thread.id).toBe(threadId);
+        if (!snapshot.result) throw new Error("Missing fixture snapshot");
+        const sent = await call({
+          kind: "prompt",
+          hostId: "remote-ssh-discovered:fixture",
+          threadId,
+          epoch: snapshot.result.epoch,
+          text: "isolated regression",
+        });
+        expect(sent.error).toBeUndefined();
+      }
+      expect(
+        (
+          await call({
+            kind: "read",
+            hostId: "remote-ssh-discovered:fixture",
+            threadId: "oversized",
+          })
+        ).error?.code,
+      ).toBe(-32094);
+      const recovered = await call({
+        kind: "read",
+        hostId: "remote-ssh-discovered:fixture",
+        threadId: "b",
+      });
+      if (!recovered.result) throw new Error("Unrelated fixture Thread became unavailable");
+      expect(
+        (
+          await call({
+            kind: "prompt",
+            hostId: "remote-ssh-discovered:fixture",
+            threadId: "b",
+            epoch: recovered.result.epoch,
+            text: "after oversized history",
+          })
+        ).error,
+      ).toBeUndefined();
+      expect(await client.evaluate("window.fixtureSends")).toEqual(["a", "b", "c", "a", "b"]);
+    } finally {
+      client?.close();
+      await browser.close();
+      if (!path.resolve(directory).startsWith(path.resolve(tmpdir()) + path.sep))
+        throw new Error("Unexpected test directory");
+      await rm(directory, { recursive: true, force: true });
+    }
+  },
+  20_000,
+);