Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions docs/architecture/web-thread-tree.md
Original file line number Diff line number Diff line change
Expand Up @@ -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 侧栏

Expand Down
33 changes: 31 additions & 2 deletions packages/web-server/src/ch-thread-tree.ts
Original file line number Diff line number Diff line change
@@ -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";
Expand Down Expand Up @@ -50,6 +54,25 @@ export class ChThreadTree {
}
}

private async readArchivedIds(): Promise<ReadonlySet<string>> {
const ids = new Set<string>();
const seen = new Set<string>();
let archived: string | undefined;
for (;;) {
const page = await this.host.request<ThreadCatalogReadResult>(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,
Expand All @@ -66,12 +89,18 @@ export class ChThreadTree {
const queue = [id];
const visited = new Set<string>();
const failedSessionIds: string[] = [];
let confirmedArchiveIds: Promise<ReadonlySet<string>> | 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;
Expand Down
174 changes: 174 additions & 0 deletions packages/web-server/test/ch-thread-tree-restore.test.ts
Original file line number Diff line number Diff line change
@@ -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<string, string> = { "native-child": "native-parent" };
const archivePages = [["native-parent", "native-unrelated"], ["native-child"]];
const request = host.request.bind(host);
host.request = async <T>(method: string, params: Record<string, unknown>): Promise<T> => {
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<T>(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 <T>(method: string, params: Record<string, unknown>): Promise<T> => {
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<T>(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);
});
Loading