diff --git a/docs/assets/web-remote-harness-identity.png b/docs/assets/web-remote-harness-identity.png new file mode 100644 index 000000000..f8e70eb93 Binary files /dev/null and b/docs/assets/web-remote-harness-identity.png differ diff --git a/docs/product/standalone-web.md b/docs/product/standalone-web.md index ecfdd5b91..8bb25573a 100644 --- a/docs/product/standalone-web.md +++ b/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 与多浏览器同时响应的实机验收。 diff --git a/package-lock.json b/package-lock.json index ed2a1905a..ff6c5dfe0 100644 --- a/package-lock.json +++ b/package-lock.json @@ -16804,7 +16804,8 @@ "name": "@codexhost/desktop-control", "version": "0.0.0", "dependencies": { - "@codexhost/shared-contracts": "0.0.0" + "@codexhost/shared-contracts": "0.0.0", + "ws": "8.21.3" } }, "packages/harness-adapter": { diff --git a/packages/desktop-control/package.json b/packages/desktop-control/package.json index 5b2a00dc8..d0753fb84 100644 --- a/packages/desktop-control/package.json +++ b/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" } } diff --git a/packages/desktop-control/scripts/build-release.mjs b/packages/desktop-control/scripts/build-release.mjs index 75f812783..2204a78d1 100644 --- a/packages/desktop-control/scripts/build-release.mjs +++ b/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, diff --git a/packages/desktop-control/src/cdp-client.ts b/packages/desktop-control/src/cdp-client.ts index 67383584d..f2390e7e6 100644 --- a/packages/desktop-control/src/cdp-client.ts +++ b/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 { 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; diff --git a/packages/desktop-control/src/remote-session-client.ts b/packages/desktop-control/src/remote-session-client.ts index 8914cadab..7b7e7ce2c 100644 --- a/packages/desktop-control/src/remote-session-client.ts +++ b/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) => diff --git a/packages/desktop-control/src/renderer-remote-sessions.ts b/packages/desktop-control/src/renderer-remote-sessions.ts index a07f40778..db6afd738 100644 --- a/packages/desktop-control/src/renderer-remote-sessions.ts +++ b/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 => { 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", { diff --git a/packages/desktop-control/test/cdp-client-large-response.test.ts b/packages/desktop-control/test/cdp-client-large-response.test.ts new file mode 100644 index 000000000..a79d946b0 --- /dev/null +++ b/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((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("'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'&¶ms.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, +); diff --git a/packages/desktop-control/test/renderer-remote-sessions.test.ts b/packages/desktop-control/test/renderer-remote-sessions.test.ts index 50743b459..db01ab36b 100644 --- a/packages/desktop-control/test/renderer-remote-sessions.test.ts +++ b/packages/desktop-control/test/renderer-remote-sessions.test.ts @@ -10,7 +10,7 @@ function fixture() { { id: 7, method: "item/commandExecution/requestApproval", params: { threadId: "same" } }, ]; const manager = () => ({ - sendRequest: vi.fn(async (method: string) => + sendRequest: vi.fn(async (method: string): Promise => method === "thread/read" ? { thread: { id: "same" } } : {}, ), getConversation: () => ({ requests }), @@ -40,6 +40,57 @@ function fixture() { return { target, routes, read, requests, route }; } describe("Desktop existing remote session boundary", () => { + it("carries authoritative external Harness identity without reading or resuming history", async () => { + const f = fixture(); + const manager = f.route("remote-ssh-discovered:a").manager; + manager.sendRequest.mockImplementation(async (method: string) => { + if (method === "thread/list") + return { + data: [ + { id: "native", modelProvider: "custom" }, + { id: "external", modelProvider: "codexhost" }, + ], + nextCursor: "next", + }; + if (method === "codexhost/thread/ownership/list") + return { threads: [{ threadId: "external", owner: "external", harnessId: "claude-code" }] }; + if (method === "codexhost/harness/plugins/list") + return { plugins: [{ id: "claude-code", name: "Claude Code" }] }; + throw new Error("Unexpected history or mutation request"); + }); + const read = () => + rendererRemoteSession(f.target, { + kind: "list", + hostId: "remote-ssh-discovered:a", + cursor: null, + archived: false, + }); + expect(await read()).toEqual({ + data: [ + { id: "native", modelProvider: "custom" }, + { + id: "external", + modelProvider: "codexhost", + harnessIdentity: { id: "claude-code", name: "Claude Code" }, + }, + ], + nextCursor: "next", + }); + expect(manager.sendRequest.mock.calls.map(([method]) => method)).toEqual([ + "thread/list", + "codexhost/thread/ownership/list", + "codexhost/harness/plugins/list", + ]); + manager.sendRequest.mockImplementation(async (method: string) => { + if (method === "thread/list") + return { data: [{ id: "external", modelProvider: "codexhost" }], nextCursor: null }; + throw new Error("Older remote Host does not support ownership metadata"); + }); + expect(await read()).toEqual({ + data: [{ id: "external", modelProvider: "codexhost" }], + nextCursor: null, + }); + }); it("shows configured offline hosts and restores only already-enabled connections once", async () => { const f = fixture(); const writes: unknown[] = []; @@ -195,6 +246,76 @@ describe("Desktop existing remote session boundary", () => { "thread/resume", "turn/start", ]); + expect(send).toHaveBeenNthCalledWith(2, "thread/resume", { + threadId: "same", + excludeTurns: true, + }); + }); + it("keeps other Threads reachable after resuming a large native history", async () => { + const f = fixture(); + const send = f.route("remote-ssh-discovered:a").manager.sendRequest; + let connected = true; + const started: string[] = []; + send.mockImplementation(async (method: string, ...args: unknown[]) => { + const params = args[0] as { threadId: string; excludeTurns?: boolean }; + if (!connected) throw new Error("Native connection closed"); + if (method === "thread/resume" && params.excludeTurns !== true) { + connected = false; + throw new Error("Unbounded native resume exceeded transport limit"); + } + if (method === "turn/start") started.push(params.threadId); + return method === "thread/read" ? { thread: { id: params.threadId } } : {}; + }); + const { epoch } = await f.read(); + for (const threadId of ["large", "other"]) + await rendererRemoteSession(f.target, { + kind: "prompt", + hostId: "remote-ssh-discovered:a", + threadId, + epoch, + text: "hello", + }); + expect(started).toEqual(["large", "other"]); + await expect(f.read()).resolves.toHaveProperty("epoch", epoch); + }); + it("reads paginated external history without requesting all Turns", async () => { + const f = fixture(); + const send = f.route("remote-ssh-discovered:a").manager.sendRequest; + send.mockImplementation(async (method: string, ...args: unknown[]) => { + const params = args[0] as { includeTurns?: boolean }; + if (method === "thread/read" && params.includeTurns) + throw new Error("Paginated External Threads require thread/turns/list"); + return method === "thread/read" + ? { thread: { id: "same", historyMode: "paginated", turns: [] } } + : { data: [{ id: "newest" }], nextCursor: "older" }; + }); + const initial = await f.read(); + const page = await rendererRemoteSession(f.target, { + kind: "turns", + hostId: "remote-ssh-discovered:a", + threadId: "same", + epoch: initial.epoch, + cursor: null, + }); + expect(page).toEqual({ data: [{ id: "newest" }], nextCursor: "older" }); + expect(send).toHaveBeenLastCalledWith("thread/turns/list", { + threadId: "same", + cursor: null, + limit: 5, + sortDirection: "desc", + itemsView: "full", + }); + send.mockClear(); + await rendererRemoteSession(f.target, { + kind: "read", + hostId: "remote-ssh-discovered:a", + threadId: "same", + metadataOnly: true, + }); + expect(send).toHaveBeenCalledExactlyOnceWith("thread/read", { + threadId: "same", + includeTurns: false, + }); }); it("serializes data in the Controller expression and excludes local operations", async () => { const f = fixture(); diff --git a/packages/shared-contracts/src/remote-sessions.ts b/packages/shared-contracts/src/remote-sessions.ts index b2408555e..f17f87161 100644 --- a/packages/shared-contracts/src/remote-sessions.ts +++ b/packages/shared-contracts/src/remote-sessions.ts @@ -23,7 +23,13 @@ export const remoteSessionOperationSchema = z.discriminatedUnion("kind", [ cursor: z.string().nullable(), archived: z.boolean(), }), - z.strictObject({ kind: z.literal("read"), ...thread }), + z.strictObject({ kind: z.literal("read"), ...thread, metadataOnly: z.boolean().optional() }), + z.strictObject({ + kind: z.literal("turns"), + ...thread, + epoch: identity, + cursor: z.string().max(4096).nullable(), + }), z.strictObject({ kind: z.literal("prompt"), ...thread, diff --git a/packages/web-server/src/ch-thread-history.ts b/packages/web-server/src/ch-thread-history.ts index 9742d0437..f88209fd5 100644 --- a/packages/web-server/src/ch-thread-history.ts +++ b/packages/web-server/src/ch-thread-history.ts @@ -32,7 +32,7 @@ export class ChThreadHistory extends ChThreadView { private readonly pagination: { cursor: string | null }; constructor( - private readonly host: ChHostClient, + private readonly host: Pick, thread: ChThread, harnessId: string, onProjection: ProjectionListener, diff --git a/packages/web-server/src/ch-thread-view.ts b/packages/web-server/src/ch-thread-view.ts index 72bb24a09..13004e8ce 100644 --- a/packages/web-server/src/ch-thread-view.ts +++ b/packages/web-server/src/ch-thread-view.ts @@ -41,6 +41,7 @@ export interface ChThread { id: string; cwd: string; modelProvider: string; + harnessIdentity?: { id: string; name: string } | undefined; workspaceKind?: "projectless"; projectId?: string | null; name?: string | null | undefined; diff --git a/packages/web-server/src/remote-sessions.ts b/packages/web-server/src/remote-sessions.ts index 9f8eacd9e..2e26f9b54 100644 --- a/packages/web-server/src/remote-sessions.ts +++ b/packages/web-server/src/remote-sessions.ts @@ -9,6 +9,7 @@ import { type RemoteSessionOperation, } from "@codexhost/shared-contracts"; import { ChThreadView, type ChThread } from "./ch-thread-view.ts"; +import { ChThreadHistory } from "./ch-thread-history.ts"; import { defaultProjections } from "./session-log.ts"; import { requestOf } from "./store.ts"; import { @@ -50,6 +51,7 @@ const rowSchema = z id: z.string().min(1), cwd: z.string(), modelProvider: z.string(), + harnessIdentity: z.object({ id: z.string().min(1), name: z.string().min(1) }).optional(), createdAt: z.number(), updatedAt: z.number(), status: z.object({ type: z.string() }), @@ -169,8 +171,10 @@ export class RemoteSessions { ); try { return await this.call(parsed.data); - } catch { - const write = !["hosts", "list", "read"].includes(operation.kind); + } catch (error) { + if (error instanceof Error && "code" in error && error.code === -32094) + throw new RpcError("remote/too-large", error.message); + const write = !["hosts", "list", "read", "turns"].includes(operation.kind); throw new RpcError( write ? "remote/outcome-unknown" : "remote/offline", write @@ -308,7 +312,9 @@ export class RemoteSessions { }, ...(entry.row.modelProvider !== "codexhost" ? { harnessIdentity: { id: "codex", name: "Codex" } } - : {}), + : entry.row.harnessIdentity + ? { harnessIdentity: entry.row.harnessIdentity } + : {}), threadReference: this.reference(entry), nativeConnection: { state: entry.online ? "connected" : "reconnecting" }, nativeDiscovery: false, @@ -349,7 +355,13 @@ export class RemoteSessions { current.reading ??= (async () => { try { const snapshot = snapshotSchema.parse( - await this.invoke({ kind: "read", ...identity }), + await this.invoke({ + kind: "read", + ...identity, + ...(current.row.historyMode === "paginated" || current.view instanceof ChThreadHistory + ? { metadataOnly: true } + : {}), + }), ) as Snapshot; if (this.closed || snapshot.result.thread.id !== identity.threadId) throw new Error("Remote identity mismatch"); @@ -359,11 +371,20 @@ export class RemoteSessions { } current.epoch = snapshot.epoch; current.online = true; - current.row = snapshot.result.thread; + current.row = { + ...snapshot.result.thread, + ...(snapshot.result.thread.modelProvider === "codexhost" && current.row.harnessIdentity + ? { + harnessIdentity: + snapshot.result.thread.harnessIdentity ?? current.row.harnessIdentity, + } + : {}), + }; const row = { ...current.row, id, cwd: "" }; - current.view ??= this.createView(row, current.viewOrigin); + current.view ??= this.createView(row, current.viewOrigin, current); try { - current.view.update(row); + if (current.view instanceof ChThreadHistory) await current.view.refresh(); + else current.view.update(row); } catch (error) { if ( !(error instanceof RpcError) || @@ -371,9 +392,12 @@ export class RemoteSessions { ) throw error; this.retireView(current); - current.view = this.createView(row, current.viewOrigin); - current.view.update(row); + current.view = this.createView(row, current.viewOrigin, current); + if (current.view instanceof ChThreadHistory) await current.view.refresh(); + else current.view.update(row); } + if (current.view instanceof ChThreadHistory) + current.row = { ...current.row, turns: current.view.thread.turns }; for (const [key, value] of Object.entries(this.summary(id, current).projections.values)) if ( [ @@ -404,7 +428,8 @@ export class RemoteSessions { ); return current.view; } catch (error) { - this.offline(current); + if (!(error instanceof RpcError) || error.code !== "remote/too-large") + this.offline(current); throw error; } })().finally(() => { @@ -419,18 +444,38 @@ export class RemoteSessions { entry.view.log.reconnectFollowers(); delete entry.view; } - private createView(row: ChThread, origin = 0): ChThreadView { - const view = new ChThreadView( - row, - row.modelProvider === "codexhost" ? "remote" : "codex", - (sessionId, key, value, seq) => { - view.log.pushProjection(key, value, seq); - for (const sink of this.controls) - sink.push({ type: "projection", sessionId, key, value, seq }); - }, - 0, - origin, - ); + private createView(row: ChThread, origin: number | undefined, entry: Entry): ChThreadView { + const onProjection = (sessionId: string, key: string, value: unknown, seq: number) => { + view.log.pushProjection(key, value, seq); + for (const sink of this.controls) + sink.push({ type: "projection", sessionId, key, value, seq }); + }; + const harness = row.modelProvider === "codexhost" ? "remote" : "codex"; + const view = + row.historyMode === "paginated" + ? new ChThreadHistory( + { + request: async (method: string, params: Record): Promise => { + if (params.threadId !== row.id) + throw new RpcError("remote/identity-invalid", "Remote history identity mismatch"); + if (method === "thread/read") + return { thread: { ...entry.row, id: row.id, cwd: "", turns: [] } } as T; + if (method !== "thread/turns/list" || !entry.epoch) + throw new RpcError("remote/unsupported", "Remote history operation unavailable"); + return (await this.invoke({ + kind: "turns", + ...entry.identity, + epoch: entry.epoch, + cursor: typeof params.cursor === "string" ? params.cursor : null, + })) as T; + }, + }, + row, + harness, + onProjection, + origin, + ) + : new ChThreadView(row, harness, onProjection, 0, origin ?? 0); return view; } private interactions(id: string, entry: Entry, requests: PendingRequest[]): void { @@ -582,12 +627,19 @@ export class RemoteSessions { const cached = readOnly ? this.entries.get(id)?.view : undefined; const view = cached ?? (await this.read(id)); if (method === "session/projections") return view.log.projectionBaseline(); - if (method === "session/page") + if (method === "session/page") { + if (view instanceof ChThreadHistory) + return view.olderPage( + Number(request.throughSeq), + typeof request.beforeSeq === "number" ? request.beforeSeq : undefined, + typeof request.maxMessages === "number" ? request.maxMessages : 20, + ); return view.log.page( Number(request.throughSeq), typeof request.beforeSeq === "number" ? request.beforeSeq : undefined, { maxMessages: typeof request.maxMessages === "number" ? request.maxMessages : 20 }, ); + } if (method === "session/toolOutput") return readToolOutput(view.log, webToolOutputReferenceSchema.parse(request)); const entry = this.entries.get(id); diff --git a/packages/web-server/test/browser/remote-sessions.test.ts b/packages/web-server/test/browser/remote-sessions.test.ts index 5af17b764..9c9ffa9eb 100644 --- a/packages/web-server/test/browser/remote-sessions.test.ts +++ b/packages/web-server/test/browser/remote-sessions.test.ts @@ -1,11 +1,12 @@ import assert from "node:assert/strict"; import { it } from "node:test"; import { createServer } from "node:net"; -import { mkdtempSync, mkdirSync, writeFileSync, rmSync } from "node:fs"; +import { mkdtempSync, mkdirSync, writeFileSync, rmSync, readFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join, resolve } from "node:path"; import { chromium } from "playwright"; import { startServer } from "../support/server.ts"; +import { FakeChHost, startFakeChDebugger } from "../support/ch-host.ts"; it( "shows colliding remote sessions in the real Web shell and opens their own transcript", @@ -16,6 +17,7 @@ it( const nonce = "a".repeat(32); const calls: Array<{ kind: string; hostId?: string }> = []; const sent = new Map(); + const sentTurns = new Map>(); const archived = new Set(); const requestIds = new Map(); const timestamp = Math.floor(Date.now() / 1000); @@ -23,9 +25,13 @@ it( id: "same", cwd: "/remote/private", name: `Session ${hostId}`, - modelProvider: "openai", + modelProvider: hostId.startsWith("remote-ssh") ? "codexhost" : "openai", + ...(hostId.startsWith("remote-ssh") + ? { harnessIdentity: { id: "claude-code", name: "Claude Code" } } + : {}), + historyMode: hostId.startsWith("remote-ssh") ? "paginated" : "legacy", createdAt: timestamp, - updatedAt: timestamp + (sent.has(hostId) ? 1 : 0), + updatedAt: timestamp + (sentTurns.get(hostId)?.length ?? 0), status: { type: "idle" }, turns: [ { @@ -50,27 +56,23 @@ it( }, ], }, - ...(sent.has(hostId) - ? [ - { - id: "follow-up", - status: "completed", - items: [ - { - id: "follow-user", - type: "userMessage", - clientId: requestIds.get(hostId), - content: [{ type: "text", text: sent.get(hostId) }], - }, - { - id: "follow-answer", - type: "agentMessage", - text: `Follow-up accepted on ${hostId}`, - }, - ], - }, - ] - : []), + ...(sentTurns.get(hostId) ?? []).map((turn, index) => ({ + id: `follow-up-${index}`, + status: "completed", + items: [ + { + id: `follow-user-${index}`, + type: "userMessage", + clientId: turn.requestId, + content: [{ type: "text", text: turn.text }], + }, + { + id: `follow-answer-${index}`, + type: "agentMessage", + text: `Follow-up accepted on ${hostId}${index ? ` #${index + 1}` : ""}`, + }, + ], + })), ], }); const controller = createServer((socket) => { @@ -89,6 +91,9 @@ it( if (operation.kind === "prompt") { sent.set(operation.hostId, operation.text); requestIds.set(operation.hostId, operation.requestId); + const history = sentTurns.get(operation.hostId) ?? []; + history.push({ text: operation.text, requestId: operation.requestId }); + sentTurns.set(operation.hostId, history); } const result = operation.kind === "hosts" @@ -105,11 +110,17 @@ it( } : operation.kind === "read" ? { - result: { thread: row(operation.hostId) }, + result: { + thread: operation.metadataOnly + ? { ...row(operation.hostId), turns: [] } + : row(operation.hostId), + }, epoch: "fixture-epoch", requests: [], } - : {}; + : operation.kind === "turns" + ? { data: [...row(operation.hostId).turns].reverse(), nextCursor: null } + : {}; socket.end(JSON.stringify({ result }) + "\n"); }); }); @@ -156,12 +167,33 @@ it( ], }), ); + const localHost = new FakeChHost(); + const nativeRequest = localHost.request.bind(localHost); + localHost.request = async (method: string, params: Record): Promise => { + const result = await nativeRequest(method, params); + if (method !== "codexhost/harness/plugins/list") return result; + const manifest = JSON.parse( + readFileSync( + resolve(import.meta.dirname, "../../../adapters/claude-code/manifest.json"), + "utf8", + ), + ); + return { + plugins: [ + ...(result as { plugins: unknown[] }).plugins, + { id: "claude-code", name: "Claude Code", version: "1", iconStyle: manifest.iconStyle }, + ], + } as T; + }; + const debug = await startFakeChDebugger(localHost); + t.after(() => debug.close()); const url = await startServer( t, resolve(import.meta.dirname, "../.."), ["--import", "tsx", "src/main.ts"], - [], + ["--ch-cdp", debug.endpoint], { + authenticated: true, environment: { HOME: directory, USERPROFILE: directory, @@ -171,8 +203,49 @@ it( }, }, ); - const browser = await chromium.launch(); + const browser = await chromium.launch({ + ...(process.env.CODEXHOST_PLAYWRIGHT_EXECUTABLE_PATH + ? { executablePath: process.env.CODEXHOST_PLAYWRIGHT_EXECUTABLE_PATH } + : {}), + }); t.after(() => browser.close()); + const phone = await browser.newPage({ + viewport: { width: 390, height: 844 }, + isMobile: true, + hasTouch: true, + locale: "en-US", + }); + await phone.goto(url); + for (const hostId of ["remote-ssh-discovered:a", "remote-control:b"]) { + await phone.getByRole("button", { name: "Open sidebar", exact: true }).first().click(); + const group = phone + .locator('[data-row-key^="workspace:remote-workspace:"]') + .filter({ has: phone.locator(`[data-host-id="${hostId}"]`) }); + await group.waitFor(); + if ((await group.getAttribute("aria-expanded")) === "false") await group.click(); + const item = phone.getByText(`Session ${hostId}`, { exact: true }).first(); + await item.waitFor(); + if (hostId.startsWith("remote-ssh")) + await item + .locator("xpath=ancestor::*[@data-row-key][1]") + .locator('[data-harness-id="claude-code"] svg') + .waitFor(); + await item.click(); + await phone.getByText(`Answer from ${hostId}`, { exact: true }).waitFor(); + } + if (process.env.CODEXHOST_TEST_SCREENSHOTS) { + mkdirSync(process.env.CODEXHOST_TEST_SCREENSHOTS, { recursive: true }); + await phone.getByRole("button", { name: "Open sidebar", exact: true }).first().click(); + await phone.waitForFunction(() => { + const drawer = document.querySelector("[data-phone-drawer]"); + return drawer && drawer.getBoundingClientRect().left >= -1; + }); + await phone.screenshot({ + path: join(process.env.CODEXHOST_TEST_SCREENSHOTS, "remote-sessions-phone.png"), + animations: "disabled", + }); + } + await phone.close(); const page = await browser.newPage({ viewport: { width: 1280, height: 900 }, locale: "en-US" }); const localMediaRequests: string[] = []; await page.route("**/api/file?**", async (route) => { @@ -224,6 +297,15 @@ it( 0, "remote image paths must render as text, not local image elements", ); + const row = item.locator("xpath=ancestor::*[@data-row-key][1]"); + if (hostId.startsWith("remote-ssh")) { + const icon = row.locator('[data-harness-id="claude-code"]'); + await icon.waitFor(); + assert.equal(await icon.locator("svg").getAttribute("viewBox"), "0 0 24 24"); + assert.equal(await row.locator("[data-native-codex-icon]").count(), 0); + } else { + assert.equal(await row.locator("[data-native-codex-icon]").count(), 1); + } } assert.deepEqual(localMediaRequests, [], "remote paths must not request local image bytes"); assert.ok( @@ -258,6 +340,24 @@ it( assert.equal(sent.has("remote-ssh-discovered:a"), false); assert.equal(calls.filter((call) => call.kind === "prompt").length, 1); assert.equal(await page.getByText("remote follow-up", { exact: true }).count(), 1); + for (const [hostId, text] of [ + ["remote-ssh-discovered:a", "second conversation"], + ["remote-control:b", "back to first conversation"], + ] as const) { + const nextCount = (sentTurns.get(hostId)?.length ?? 0) + 1; + await page.getByText(`Session ${hostId}`, { exact: true }).first().click(); + await page.getByText(`Answer from ${hostId}`, { exact: true }).waitFor(); + await page.locator('[contenteditable="true"][role="textbox"]').fill(text); + await page.getByRole("button", { name: /Send message|发送消息/ }).click(); + await page + .getByText(`Follow-up accepted on ${hostId}${nextCount > 1 ? ` #${nextCount}` : ""}`, { + exact: true, + }) + .waitFor(); + } + assert.equal(calls.filter((call) => call.kind === "prompt").length, 3); + assert.equal(sent.get("remote-ssh-discovered:a"), "second conversation"); + assert.equal(sent.get("remote-control:b"), "back to first conversation"); const remoteId = "web-remote:" + Buffer.from(JSON.stringify(["remote-control:b", "same"])).toString("base64url"); @@ -271,7 +371,10 @@ it( await remoteRow.waitFor(); assert.equal(archived.size, 0); await remoteRow.click(); - await page.getByText("Follow-up accepted on remote-control:b", { exact: true }).waitFor(); + await page + .getByText("Follow-up accepted on remote-control:b", { exact: true }) + .last() + .waitFor(); assert.deepEqual( calls.filter((call) => call.kind === "archive").map((call) => call.hostId), ["remote-control:b", "remote-control:b"], diff --git a/packages/web-server/test/remote-history.test.ts b/packages/web-server/test/remote-history.test.ts new file mode 100644 index 000000000..631cf8b94 --- /dev/null +++ b/packages/web-server/test/remote-history.test.ts @@ -0,0 +1,142 @@ +import assert from "node:assert/strict"; +import { it } from "node:test"; +import { RemoteSessions, remoteSessionId } from "../src/remote-sessions.ts"; +import { ChThreadHistory } from "../src/ch-thread-history.ts"; +import { EventHub, RpcRegistry, StreamRegistry } from "../src/transport.ts"; +import type { RemoteSessionOperation } from "@codexhost/shared-contracts"; + +it("opens and pages a long remote external Thread without full-history reads", async (t) => { + const hostId = "remote-ssh-discovered:a", + threadId = "long-46"; + const row = { + id: threadId, + cwd: "/remote", + modelProvider: "codexhost", + createdAt: 1, + updatedAt: 2, + status: { type: "idle" }, + historyMode: "paginated", + turns: [], + }; + const calls: RemoteSessionOperation[] = []; + let losePage = false; + const turn = (id: string) => ({ + id, + status: "completed", + items: [{ id: id + "-answer", type: "agentMessage", text: id }], + }); + const remote = new RemoteSessions(new EventHub("history"), async (operation) => { + calls.push(operation); + if (operation.kind === "hosts") return [{ hostId, online: true }]; + if (operation.kind === "list") return { data: [row], nextCursor: null }; + if (operation.kind === "read") { + assert.equal(operation.metadataOnly, true); + return { result: { thread: row }, epoch: "connection-a", requests: [] }; + } + if (operation.kind === "turns") { + assert.equal(operation.threadId, threadId); + assert.equal(operation.epoch, "connection-a"); + if (losePage) throw new Error("Lost read reply"); + return operation.cursor === null + ? { data: [turn("newest"), turn("middle")], nextCursor: "older" } + : { data: [turn("oldest")], nextCursor: null }; + } + throw new Error("Unexpected mutation"); + }); + t.after(() => remote.close()); + const rpc = new RpcRegistry(); + rpc.register("session/page", () => { + throw new Error("Remote history must not route locally"); + }); + remote.install(rpc, new StreamRegistry()); + await remote.refresh(); + const id = remoteSessionId(hostId, threadId); + const view = await remote.read(id); + assert.ok(view instanceof ChThreadHistory); + assert.deepEqual( + view.thread.turns.map((turn) => turn.id), + ["middle", "newest"], + ); + losePage = true; + await assert.rejects( + view.loadOlder(), + (error) => + typeof error === "object" && + error !== null && + "code" in error && + error.code === "remote/offline", + ); + losePage = false; + const result = await rpc.dispatch("session/page", { + args: { + request: { + sessionId: id, + throughSeq: view.log.lastSeq, + beforeSeq: view.log.firstSeq, + maxMessages: 20, + }, + }, + }); + assert.equal(result.ok, true, JSON.stringify(result)); + assert.deepEqual( + view.thread.turns.map((turn) => turn.id), + ["oldest", "middle", "newest"], + ); + assert.equal( + calls.filter((operation) => operation.kind === "turns" && operation.cursor === "older").length, + 2, + ); +}); + +it("a rejected oversized history page does not mark the remote Host offline", async (t) => { + const hostId = "remote-ssh-discovered:fixture"; + let oversized = false; + const row = { + id: "thread", + cwd: "/fixture", + modelProvider: "openai", + historyMode: "paginated", + status: { type: "idle" }, + createdAt: 1, + updatedAt: 1, + turns: [], + }; + const remote = new RemoteSessions(new EventHub("size-fixture"), async (operation) => { + if (operation.kind === "hosts") return [{ hostId, online: true }]; + if (operation.kind === "list") return { data: [row], nextCursor: null }; + if (operation.kind === "read") + return { result: { thread: row }, epoch: "stable", requests: [] }; + if (operation.kind === "turns") { + if (oversized) + throw Object.assign(new Error("Remote history response exceeds the 16 MiB limit"), { + code: -32094, + }); + return { data: [{ id: "turn", status: "completed", items: [] }], nextCursor: null }; + } + throw new Error("Unexpected fixture operation"); + }); + t.after(() => remote.close()); + const rpc = new RpcRegistry(); + rpc.register("session/projections", () => null); + remote.install(rpc, new StreamRegistry()); + await remote.refresh(); + const id = remoteSessionId(hostId, "thread"); + await remote.read(id); + oversized = true; + row.updatedAt += 1; + await assert.rejects( + remote.read(id), + (error) => + typeof error === "object" && + error !== null && + "code" in error && + error.code === "remote/too-large", + ); + const result = await rpc.dispatch("session/projections", { + args: { request: { sessionId: id } }, + }); + assert.equal(result.ok, true); + if (!result.ok) throw Error("Projection fixture failed"); + const values = (result.value as { values: Record }).values; + assert.equal(values.nativeConnection?.state, "connected"); +}); diff --git a/packages/web-server/test/remote-sessions.test.ts b/packages/web-server/test/remote-sessions.test.ts index 8a7f21f28..d56dbc92b 100644 --- a/packages/web-server/test/remote-sessions.test.ts +++ b/packages/web-server/test/remote-sessions.test.ts @@ -11,6 +11,70 @@ import { join } from "node:path"; import { DataDir } from "../src/store.ts"; import { Workspaces } from "../src/workspaces.ts"; +it("keeps remote Harness identity in sidebar summaries and after opening metadata-only history", async (t) => { + const identity = { id: "claude-code", name: "Claude Code" }; + const rows = ["external", "native"].map((id) => ({ + id, + cwd: "/remote", + name: id, + modelProvider: id === "external" ? "codexhost" : "custom", + createdAt: 1, + updatedAt: 2, + status: { type: "idle" }, + turns: [], + })); + const events = new EventHub("icons"), + rpc = new RpcRegistry(); + rpc.register("session/list", () => ({ items: [] })); + rpc.register("session/projections", () => { + throw Error("Must not route locally"); + }); + const remote = new RemoteSessions(events, async (op) => { + if (op.kind === "hosts") return [{ hostId: "remote-control:a", online: true }]; + if (op.kind === "list") + return { + data: rows.map((row) => + row.id === "external" ? { ...row, harnessIdentity: identity } : row, + ), + nextCursor: null, + }; + if (op.kind === "read") + return { + result: { thread: rows.find((row) => row.id === op.threadId) }, + epoch: "epoch", + requests: [], + }; + throw Error("No model sends or mutations are expected"); + }); + t.after(() => remote.close()); + remote.install(rpc, new StreamRegistry()); + const call = (method: string, request: object = {}) => + rpc.dispatch(method, { args: { request } }); + await remote.refresh(); + const result = await call("session/list"); + assert.ok(result.ok); + const items = ( + result.value as { + items: Array<{ sessionId: string; projections: { values: Record } }>; + } + ).items; + const id = remoteSessionId("remote-control:a", "external"); + assert.deepEqual( + items.find((row) => row.sessionId === id)?.projections.values.harnessIdentity, + identity, + ); + assert.deepEqual(items.find((row) => row.sessionId !== id)?.projections.values.harnessIdentity, { + id: "codex", + name: "Codex", + }); + const opened = await call("session/projections", { sessionId: id }); + assert.ok(opened.ok); + assert.deepEqual( + (opened.value as { values: Record }).values.harnessIdentity, + identity, + ); +}); + it("archives and undoes through workspace RPCs without losing local archives or other hosts", async (t) => { const directory = mkdtempSync(join(tmpdir(), "remote-archive-")); t.after(() => rmSync(directory, { recursive: true, force: true })); diff --git a/tests/release/desktop-controller-bundle.test.mjs b/tests/release/desktop-controller-bundle.test.mjs index f1797d895..cab636c92 100644 --- a/tests/release/desktop-controller-bundle.test.mjs +++ b/tests/release/desktop-controller-bundle.test.mjs @@ -1,4 +1,5 @@ import { mkdtemp, readFile, rm } from "node:fs/promises"; +import { spawnSync } from "node:child_process"; import os from "node:os"; import path from "node:path"; @@ -59,6 +60,14 @@ describe("Desktop Controller release Bundle", () => { expect(source).not.toContain('process.stdout.write("ready\\n")'); expect(source).not.toContain("tools/renderer-binding"); expect(source).not.toContain("@anthropic-ai/"); + const startup = spawnSync(process.execPath, [outputPath, "--codexhost-bundle-check"], { + encoding: "utf8", + timeout: 5_000, + windowsHide: true, + }); + expect(startup.status).toBe(1); + expect(startup.stderr).toContain("unknown Desktop Controller option"); + expect(startup.stderr).not.toContain("Dynamic require"); } finally { await rm(directory, { recursive: true, force: true }); }