diff --git a/docs/architecture/web-thread-tree.md b/docs/architecture/web-thread-tree.md index d714b615a..1bc36e32c 100644 --- a/docs/architecture/web-thread-tree.md +++ b/docs/architecture/web-thread-tree.md @@ -20,6 +20,7 @@ Web 侧栏把本机 Host 的 Thread 显示成可以展开、收起的树。树 - 每次刷新列表都读取完整的 `parents`。本机 CH 会话摘要的投影 `threadTreeParent` 为显示父 Thread ID 或 `null`;草稿和远程 Host 会话没有该字段。 - 已加载行的祖先不在当前页时,Web Server 按 ID 单独读取并一起返回,最多向上 64 层。遇到环或无法确认的祖先时停止,子 Thread 显示在顶层。 - 侧栏的归档与恢复(`workspace/archiveSession`、`workspace/unarchiveSession`、`session/archive`、`session/unarchive`)级联到全部后代,包括未加载的后代:先父后子,跳过已处于目标状态的后代,最后只刷新一次列表。用户直接操作的 Thread 写入失败时返回它自己的错误,不处理后代;只有后代失败时返回 `session/archive-partial`,`details.failedSessionIds` 列出失败的 Thread,已成功的不回滚。 +- 恢复时,已知归档的后代直接恢复;未出现在 Web 已加载归档集合中的后代,先通过包含归档的 Host 目录确认状态,并跟随 `officialCursor.archived` 直到归档分支结束。整个级联共用一次确认,不读取官方活动分支;确认未归档的后代才会跳过。确认读取失败或游标重复时,相应后代按恢复失败报告,不把未知状态当作已恢复。 ## Web 侧栏 diff --git a/packages/web-server/src/ch-thread-tree.ts b/packages/web-server/src/ch-thread-tree.ts index 6a803e38e..053534a72 100644 --- a/packages/web-server/src/ch-thread-tree.ts +++ b/packages/web-server/src/ch-thread-tree.ts @@ -1,4 +1,8 @@ -import { clientThreadTreeSchema } from "@codexhost/shared-contracts"; +import { + clientThreadTreeSchema, + THREAD_CATALOG_READ_METHOD, + type ThreadCatalogReadResult, +} from "@codexhost/shared-contracts"; import type { ChHostClient } from "./ch-host-client.ts"; import type { ChThread } from "./ch-thread-view.ts"; import { RpcError } from "./transport.ts"; @@ -50,6 +54,25 @@ export class ChThreadTree { } } + private async readArchivedIds(): Promise> { + const ids = new Set(); + const seen = new Set(); + let archived: string | undefined; + for (;;) { + const page = await this.host.request(THREAD_CATALOG_READ_METHOD, { + includeArchived: true, + officialCursor: { live: null, ...(archived === undefined ? {} : { archived }) }, + }); + for (const id of page.archivedThreadIds) ids.add(id); + const next = page.officialCursor?.archived; + if (next == null) return ids; + if (seen.has(next)) + throw new RpcError("host/catalog-invalid", "Host archive cursor did not advance."); + seen.add(next); + archived = next; + } + } + async archive( id: string, archived: boolean, @@ -66,12 +89,18 @@ export class ChThreadTree { const queue = [id]; const visited = new Set(); const failedSessionIds: string[] = []; + let confirmedArchiveIds: Promise> | undefined; for (const next of queue) { if (visited.has(next)) continue; visited.add(next); queue.push(...(children.get(next) ?? [])); - if (next !== id && archivedIds.has(next) === archived) continue; + if (next !== id && archived && archivedIds.has(next)) continue; try { + if (next !== id && !archived && !archivedIds.has(next)) { + // Native archive IDs cover only loaded pages; absence needs a complete archive read. + confirmedArchiveIds ??= this.readArchivedIds(); + if (!(await confirmedArchiveIds).has(next)) continue; + } await setArchived(next); } catch (error) { if (next === id) throw error; diff --git a/packages/web-server/test/ch-thread-tree-restore.test.ts b/packages/web-server/test/ch-thread-tree-restore.test.ts new file mode 100644 index 000000000..baeb130e8 --- /dev/null +++ b/packages/web-server/test/ch-thread-tree-restore.test.ts @@ -0,0 +1,174 @@ +import assert from "node:assert/strict"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { it, type TestContext } from "node:test"; +import { hostThreadIdSchema } from "@codexhost/shared-contracts"; +import { ChSessions } from "../src/ch-sessions.ts"; +import type { CatalogPage } from "../src/ch-session-page.ts"; +import { DataDir } from "../src/store.ts"; +import { EventHub, RpcError, RpcRegistry, StreamRegistry } from "../src/transport.ts"; +import { Workspaces } from "../src/workspaces.ts"; +import { FakeChHost } from "./support/ch-host.ts"; +import { nativeCodex } from "./support/native-codex.ts"; + +function fixture(t: TestContext) { + const directory = mkdtempSync(join(tmpdir(), "ch-tree-restore-")); + const host = nativeCodex(new FakeChHost()); + for (const id of ["native-parent", "native-unrelated", "native-child", "native-live"]) { + host.add(id, directory).modelProvider = "openai"; + if (id !== "native-live") host.archivedIds.add(id); + } + const parents: Record = { "native-child": "native-parent" }; + const archivePages = [["native-parent", "native-unrelated"], ["native-child"]]; + const request = host.request.bind(host); + host.request = async (method: string, params: Record): Promise => { + if (method === "codexhost/thread/tree/read") { + host.requests.push({ method, params: structuredClone(params) }); + return { parents: { ...parents } } as T; + } + if (method === "codexhost/thread/catalog/read") { + host.requests.push({ method, params: structuredClone(params) }); + const cursor = params.officialCursor as CatalogPage["officialCursor"]; + const index = cursor?.archived === undefined ? 0 : Number(cursor.archived); + const archived = + params.includeArchived === true && cursor?.archived !== null + ? (archivePages[index] ?? []).filter((id) => host.archivedIds.has(id)) + : []; + const live = + cursor?.live === null + ? [] + : [...host.threads.keys()].filter((id) => !host.archivedIds.has(id)); + const ids = [...live, ...archived]; + const page: CatalogPage = { + threads: ids.map((id) => { + const thread = host.threads.get(id); + assert.ok(thread); + return { ...structuredClone(thread), turns: [] }; + }), + ownership: ids.map((id) => ({ threadId: hostThreadIdSchema.parse(id), owner: "codex" })), + archivedThreadIds: archived.map((id) => hostThreadIdSchema.parse(id)), + officialCursor: { + live: null, + ...(params.includeArchived === true + ? { archived: index + 1 < archivePages.length ? String(index + 1) : null } + : {}), + }, + }; + return page as T; + } + if (method === "thread/unarchive" && !host.archivedIds.has(String(params.threadId))) { + host.requests.push({ method, params: structuredClone(params) }); + throw new RpcError("host/rejected", "no archived rollout found", { hostCode: -32600 }); + } + return request(method, params); + }; + const workspaces = new Workspaces(new DataDir(directory), directory, true); + const sessions = new ChSessions(host, workspaces, new EventHub(directory)); + const rpc = new RpcRegistry(); + sessions.register(rpc, new StreamRegistry()); + t.after(async () => { + await sessions.close(); + rmSync(directory, { recursive: true, force: true }); + }); + const dispatch = (method: string, request: object = {}) => + rpc.dispatch(method, { args: { request } }); + const load = async () => { + const result = await dispatch("session/list", { includeArchived: true }); + assert.ok(result.ok); + assert.deepEqual(workspaces.archivedSessionIds(), ["native-parent", "native-unrelated"]); + assert.equal(sessions["rows"].has("native-child"), false); + host.requests.length = 0; + }; + const writes = () => + host.requests + .filter((entry) => entry.method === "thread/unarchive") + .map((entry) => entry.params.threadId); + const confirmations = () => + host.requests.filter( + (entry) => + entry.method === "codexhost/thread/catalog/read" && entry.params.includeArchived === true, + ); + return { host, parents, dispatch, load, writes, confirmations, workspaces }; +} + +for (const method of ["workspace/unarchiveSession", "session/unarchive"]) { + it(`${method} restores a native child on a later archive page in a fresh Web instance`, async (t) => { + const f = fixture(t); + await f.load(); + const result = await f.dispatch(method, { sessionId: "native-parent" }); + assert.ok(result.ok, JSON.stringify(result)); + assert.equal( + f.host.archivedIds.has("native-child"), + false, + "child must be restored with parent", + ); + assert.deepEqual(f.writes(), ["native-parent", "native-child"]); + assert.deepEqual([...f.host.archivedIds], ["native-unrelated"]); + assert.deepEqual(f.workspaces.archivedSessionIds(), ["native-unrelated"]); + assert.deepEqual( + f.confirmations().map((entry) => entry.params), + [ + { includeArchived: true, officialCursor: { live: null } }, + { includeArchived: true, officialCursor: { live: null, archived: "1" } }, + ], + ); + }); +} + +it("confirms live descendants before skipping their native writes and shares the archive read", async (t) => { + const f = fixture(t); + f.parents["native-live"] = "native-parent"; + await f.load(); + const result = await f.dispatch("workspace/unarchiveSession", { sessionId: "native-parent" }); + assert.ok(result.ok, JSON.stringify(result)); + assert.deepEqual(f.writes(), ["native-parent", "native-child"]); + assert.deepEqual([...f.host.archivedIds], ["native-unrelated"]); + assert.equal(f.confirmations().length, 2, "one archive traversal serves both descendants"); +}); + +for (const failure of ["offline", "repeated cursor"] as const) { + it(`reports an unknown descendant when archive confirmation has a ${failure} failure and restores known archives`, async (t) => { + const f = fixture(t); + f.parents["native-unrelated"] = "native-parent"; + await f.load(); + const request = f.host.request.bind(f.host); + t.mock.method( + f.host, + "request", + async (method: string, params: Record): Promise => { + const secondPage = + method === "codexhost/thread/catalog/read" && + (params.officialCursor as CatalogPage["officialCursor"])?.archived === "1"; + if (secondPage && failure === "offline") { + f.host.requests.push({ method, params: structuredClone(params) }); + throw new RpcError("host/offline", "Archive catalog unavailable"); + } + const result = await request(method, params); + if (secondPage && failure === "repeated cursor") + (result as CatalogPage).officialCursor = { live: null, archived: "1" }; + return result; + }, + ); + const result = await f.dispatch("workspace/unarchiveSession", { sessionId: "native-parent" }); + assert.ok(!result.ok); + assert.equal(result.error.code, "session/archive-partial"); + assert.deepEqual(result.error.details, { failedSessionIds: ["native-child"] }); + assert.deepEqual(f.writes(), ["native-parent", "native-unrelated"]); + assert.deepEqual([...f.host.archivedIds], ["native-child"]); + assert.equal(f.confirmations().length, 2); + }); +} + +it("preserves the selected Thread's native restore error without confirming or mutating descendants", async (t) => { + const f = fixture(t); + await f.load(); + f.host.archivedIds.delete("native-parent"); + const result = await f.dispatch("workspace/unarchiveSession", { sessionId: "native-parent" }); + assert.ok(!result.ok); + assert.equal(result.error.code, "host/rejected"); + assert.deepEqual(result.error.details, { hostCode: -32600 }); + assert.deepEqual(f.writes(), ["native-parent"]); + assert.equal(f.confirmations().length, 0); + assert.equal(f.host.archivedIds.has("native-child"), true); +});