diff --git a/.scratch/durable-multiview/HANDOFF.md b/.scratch/durable-multiview/HANDOFF.md index 3bc8143a..b4d066ff 100644 --- a/.scratch/durable-multiview/HANDOFF.md +++ b/.scratch/durable-multiview/HANDOFF.md @@ -219,7 +219,7 @@ Pace 页面放在 `apps/desktop/src/pages/` 下的 dev-only 路由,backend 侧 - 右侧面板:任务树、队列、通知。 - 断线时顶部显示 `Banner`,输入框禁用。 - aborted 条目带橙色 `interrupted` 标签。 -- token 放在 URL 的 `#fragment` 里,不会发给 dev server。宿主的 ready 行里多了一个 `web` 字段,就是完整链接。 +- token 放在 URL 的 `#fragment` 里,不会发给 dev server。宿主 ready 行的 `web` 字段只有无凭证地址;显式运行 `bun run web:link` 获取完整链接,原生客户端从本地 token 文件读取。浏览器来源必须匹配宿主的 Origin 允许列表,见 [检视修复与运行说明](../../docs/research/durable-multiview-review-fixes.md)。 - `web/tsconfig.json` 用 `Bundler` 解析:Astryx 的 `.d.ts` 是不带扩展名的相对再导出,`NodeNext` 下类型会全部退化。spike 的 `typecheck` 脚本两份配置都会检查。 - Vite 配置照 Pace 的做法固定一份 `react`(bun 在 `@astryxdesign/core` 下面嵌了第二份)。 diff --git a/docs/research/durable-multiview-review-fixes.md b/docs/research/durable-multiview-review-fixes.md new file mode 100644 index 00000000..2fb08301 --- /dev/null +++ b/docs/research/durable-multiview-review-fixes.md @@ -0,0 +1,34 @@ +# Durable 多端 spike:高优先检视修复 + +本轮针对已合并 spike 的审批归属、订阅竞态和消息边界问题补充回归保护。其余历史检视不在本轮范围内;本轮也不代表生产迁移或安全验收完成。 + +## 宿主与审批边界 + +- 审批决定必须属于请求所在的会话。错误会话的调用在写入审批文档前失败,不释放原工具;同会话重复决定仍返回先前决定。 +- 每次订阅拥有独立身份。过期订阅的快照、增量和结束消息都不能覆盖新订阅;迟到的 watch 会被关闭。 +- 文档尚不存在时,先监听提交,再检查文档是否存在。创建发生在这两步之间也会触发订阅,不依赖下一次更新来补救。 +- WebSocket 消息在入口解析和校验。无效 JSON、消息类型或参数只关闭发送方连接,不让异常逃逸到宿主进程;健康客户端仍可操作。 +- 有 Origin 的浏览器连接必须精确匹配允许列表,并通过 token 鉴权。无 Origin 的原生 TUI、观察器和 Pace backend 中继仍必须提供 token。 +- 常规启动日志只给出 token 文件路径,不输出 token 或带凭证的浏览器链接。token 仍由本地权限为 `0600` 的文件持久保存。 + +## 本地运行 + +在 `spikes/durable-multiview` 下运行: + +```sh +bun run demo:faux +bun run web +bun run web:link +``` + +前两条需要各自运行在一个终端中。第三条显式生成浏览器连接链接;该链接含宿主控制凭证,不应写入共享日志。自定义宿主目录或地址可传 `--data-dir`、`--url`,与 TUI 的参数一致。 + +宿主 CLI 默认只允许 `http://127.0.0.1:5199`。其他浏览器来源用 `--browser-origin http://127.0.0.1:端口` 显式指定;此参数可重复,指定后替换默认列表。`openHost()` 默认不允许任何浏览器来源,调用方需传 `browserOrigins`。测试和验收脚本为各自的 Vite 端口配置精确来源。 + +## 回归边界 + +- `test/approval-boundaries.test.ts`:真实 Harness 上拒绝跨会话审批,检查两边文档和待审批工具,保留合法决定及重复调用。 +- `test/gateway-boundaries.test.ts`:真实 WebSocket 与宿主进程,覆盖错误来源、畸形消息、订阅先后顺序、取消和文档创建间隙。并发通过显式屏障协调。 +- `test/host-startup.test.ts`:从进程输出确认凭证未泄露,再从本地 token 文件连接真实宿主。 + +原有宿主恢复、工具热替换及完整三端脚本已改为从本地 token 文件取凭证,不再从日志读取。 diff --git a/spikes/durable-multiview/cli/web-link.ts b/spikes/durable-multiview/cli/web-link.ts new file mode 100644 index 00000000..cb40f425 --- /dev/null +++ b/spikes/durable-multiview/cli/web-link.ts @@ -0,0 +1,7 @@ +import { hostAddress } from "./host-address.ts"; + +const { url, token } = await hostAddress(process.argv.slice(2)); +const web = new URL("http://127.0.0.1:5199/"); +web.hash = new URLSearchParams({ token, url }).toString(); +// Deliberately requested by the operator, never emitted by the host's routine logs. +console.log(web.href); diff --git a/spikes/durable-multiview/host/demo/approvals.ts b/spikes/durable-multiview/host/demo/approvals.ts index df30462a..1e6435ab 100644 --- a/spikes/durable-multiview/host/demo/approvals.ts +++ b/spikes/durable-multiview/host/demo/approvals.ts @@ -1,6 +1,6 @@ import type { Context } from "@earendil-works/chord"; import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; -import type { ConversationId, Harness } from "@earendil-works/pi-durable"; +import type { ConversationId, Harness, TaskId } from "@earendil-works/pi-durable"; import type { ApprovalDecision, PendingApproval } from "../../protocol/demo.ts"; import { ApprovalsDoc } from "./state.ts"; @@ -65,8 +65,17 @@ export class ApprovalBoard { if (harness === undefined) throw new Error("Approvals are not attached to a Harness"); const candidate: ApprovalDecision = { approved, by, at: Date.now() }; const decision = await harness.commit(async (tx) => { + const taskId = Number(id); + if (!Number.isSafeInteger(taskId) || taskId < 1 || String(taskId) !== id) { + throw new Error("Invalid approval task ID"); + } + const task = await tx.task(taskId as TaskId); + if (task === undefined || task.conversationId !== conversationId) { + throw new Error("Approval belongs to another conversation"); + } + const pending = this.#waiters.get(id); const doc = await tx.doc(ApprovalsDoc, conversationId); - if (doc.decisions[id] === undefined && !this.#waiters.has(id)) throw new Error(`No pending approval ${id}`); + if (doc.decisions[id] === undefined && pending === undefined) throw new Error(`No pending approval ${id}`); doc.decisions[id] ??= { ...candidate }; return { ...doc.decisions[id] } as ApprovalDecision; }, BACKGROUND_CONTEXT); diff --git a/spikes/durable-multiview/host/gateway.ts b/spikes/durable-multiview/host/gateway.ts index e2b0dbc1..a511317d 100644 --- a/spikes/durable-multiview/host/gateway.ts +++ b/spikes/durable-multiview/host/gateway.ts @@ -25,6 +25,7 @@ import { type CallMethod, type CallMethods, type ClientFrame, + isClientFrame, type ServerFrame, type StreamName, UNAUTHORIZED_CLOSE_CODE, @@ -47,6 +48,8 @@ export interface GatewayOptions { readonly session: SessionInfo; readonly token: string; readonly port: number; + /** Exact browser origins permitted to connect. Native clients send no Origin. */ + readonly browserOrigins?: readonly string[]; readonly approvals?: ApprovalSource; /** Conversation documents offered as `doc::` streams. */ readonly docs?: readonly ConversationDocToken[]; @@ -68,8 +71,10 @@ export async function startGateway(options: GatewayOptions): Promise { }); const clients = new Set(); server.on("connection", (socket: WebSocket, request: IncomingMessage) => { + socket.on("error", () => socket.terminate()); const token = new URL(request.url ?? "/", "ws://127.0.0.1").searchParams.get("token"); - if (token !== options.token) { + const origin = request.headers.origin; + if (token !== options.token || (origin !== undefined && !options.browserOrigins?.includes(origin))) { socket.close(UNAUTHORIZED_CLOSE_CODE, "unauthorized"); return; } @@ -190,19 +195,34 @@ async function firstInput(harness: Harness, id: ConversationId): Promise<{ title } type Subscription = { stop(): Promise | void }; +type SubscriptionState = { subscription?: Subscription }; +type SendFrame = (frame: ServerFrame) => Promise; class GatewayClient { readonly #socket: WebSocket; readonly #options: GatewayOptions; readonly #conversations: ConversationList; - readonly #subscriptions = new Map(); + readonly #subscriptions = new Map(); #disposed = false; constructor(socket: WebSocket, options: GatewayOptions, conversations: ConversationList) { this.#socket = socket; this.#options = options; this.#conversations = conversations; - socket.on("message", (data) => void this.#receive(JSON.parse(String(data)) as ClientFrame)); + socket.on("message", (data) => { + let frame: unknown; + try { + frame = JSON.parse(String(data)); + } catch { + socket.close(1008, "Invalid client frame"); + return; + } + if (!isClientFrame(frame)) { + socket.close(1008, "Invalid client frame"); + return; + } + void this.#receive(frame).catch(() => socket.close(1011, "Gateway request failed")); + }); this.send({ type: "hello", session: options.session, @@ -225,91 +245,120 @@ class GatewayClient { } async #subscribe(stream: StreamName): Promise { - // A resubscribe restarts the stream from a fresh snapshot. - await this.#unsubscribe(stream); - this.#subscriptions.set(stream, "pending"); - let subscription: Subscription | undefined; + const previous = this.#subscriptions.get(stream); + const state: SubscriptionState = {}; + this.#subscriptions.set(stream, state); + const current = () => !this.#disposed && this.#subscriptions.get(stream) === state; + const send: SendFrame = (frame) => (current() ? this.send(frame) : Promise.resolve()); + let subscription: Subscription; try { - subscription = await this.#open(stream); + await previous?.subscription?.stop(); + if (!current()) return; + subscription = await this.#open(stream, send); } catch (error) { - this.#subscriptions.delete(stream); - await this.send({ type: "ended", stream, reason: error instanceof Error ? error.message : String(error) }); + if (current()) { + await send({ type: "ended", stream, reason: error instanceof Error ? error.message : String(error) }); + if (current()) this.#subscriptions.delete(stream); + } return; } - // Unsubscribed, or the socket went away, while the watch was being acquired. - if (this.#disposed || this.#subscriptions.get(stream) !== "pending") { + // An obsolete watch can finish acquiring after its replacement has already opened. + if (!current()) { await subscription.stop(); return; } - this.#subscriptions.set(stream, subscription); + state.subscription = subscription; } - async #open(stream: StreamName): Promise { + async #open(stream: StreamName, send: SendFrame): Promise { if (stream === "conversations") { - void this.send({ type: "snapshot", stream, value: this.#conversations.value }); - const unsubscribe = this.#conversations.subscribe((value) => void this.send({ type: "snapshot", stream, value })); + void send({ type: "snapshot", stream, value: this.#conversations.value }); + const unsubscribe = this.#conversations.subscribe((value) => void send({ type: "snapshot", stream, value })); return { stop: unsubscribe }; } if (stream === "approvals") { const approvals = this.#options.approvals; - void this.send({ type: "snapshot", stream, value: approvals?.value ?? [] }); - const unsubscribe = approvals?.subscribe((value) => void this.send({ type: "snapshot", stream, value })) ?? (() => {}); + void send({ type: "snapshot", stream, value: approvals?.value ?? [] }); + const unsubscribe = approvals?.subscribe((value) => void send({ type: "snapshot", stream, value })) ?? (() => {}); return { stop: unsubscribe }; } - if (stream === "tasks") return this.#forward(stream, await this.#options.harness.watchTaskGraph(context)); + if (stream === "tasks") return this.#forward(stream, await this.#options.harness.watchTaskGraph(context), send); if (stream.startsWith("doc:")) { const at = stream.lastIndexOf(":"); - return this.#openDoc(stream, stream.slice("doc:".length, at), Number(stream.slice(at + 1)) as ConversationId); + return this.#openDoc(stream, stream.slice("doc:".length, at), Number(stream.slice(at + 1)) as ConversationId, send); } const conversation = await this.#conversation(Number(stream.slice("conversation:".length)) as ConversationId); - return this.#forward(stream, await conversation.watch(context)); + return this.#forward(stream, await conversation.watch(context), send); } /** A document that does not exist yet streams `null`, then its value from the commit that creates it. */ - async #openDoc(stream: StreamName, kind: string, id: ConversationId): Promise { + async #openDoc(stream: StreamName, kind: string, id: ConversationId, send: SendFrame): Promise { const token = this.#options.docs?.find((candidate) => candidate.definition.kind === kind); if (token === undefined) throw new Error(`Unknown document ${kind}`); await this.#conversation(id); const { harness } = this.#options; - const existing = await harness.watchDoc(token, id, context); - if (existing !== undefined) return this.#forward(stream, existing); - await this.send({ type: "snapshot", stream, value: null }); + let signalCreation!: () => void; + const created = new Promise((resolve) => { + signalCreation = resolve; + }); let inner: Subscription | undefined; let stopped = false; + // Observe creation before probing absence, so a commit cannot fall between the two. const unsubscribe = harness.subscribeCommits((publication) => { - const created = publication.changes.some( + const matches = publication.changes.some( (change) => (change.type === "document" || change.type === "document.copy") && change.record.kind === kind && change.conversationId === id, ); - if (!created) return; + if (!matches) return; unsubscribe(); // A commit listener must not call Session APIs; attach after it returns. - setImmediate(() => { - void harness.watchDoc(token, id, context).then(async (watch) => { - if (watch === undefined) return; - if (stopped) await watch.stop(); - else inner = await this.#forward(stream, watch); - }); - }); + setImmediate(() => signalCreation()); }); + try { + const existing = await harness.watchDoc(token, id, context); + if (existing !== undefined) { + unsubscribe(); + return this.#forward(stream, existing, send); + } + await send({ type: "snapshot", stream, value: null }); + } catch (error) { + unsubscribe(); + throw error; + } + void created + .then(async () => { + if (stopped) return; + const watch = await harness.watchDoc(token, id, context); + if (watch === undefined) return; + if (stopped) await watch.stop(); + else { + inner = await this.#forward(stream, watch, send); + if (stopped) await inner.stop(); + } + }) + .catch((error: unknown) => { + unsubscribe(); + return send({ type: "ended", stream, reason: error instanceof Error ? error.message : String(error) }); + }); return { stop: async () => { stopped = true; unsubscribe(); + signalCreation(); await inner?.stop(); }, }; } - async #forward(stream: StreamName, watch: WatchHandle): Promise { - await this.send({ type: "snapshot", stream, value: watch.value }); - watch.start((_value, ops) => this.send({ type: "ops", stream, ops })); + async #forward(stream: StreamName, watch: WatchHandle, send: SendFrame): Promise { + await send({ type: "snapshot", stream, value: watch.value }); + watch.start((_value, ops) => send({ type: "ops", stream, ops })); return watch; } async #unsubscribe(stream: StreamName): Promise { const subscription = this.#subscriptions.get(stream); this.#subscriptions.delete(stream); - if (subscription !== undefined && subscription !== "pending") await subscription.stop(); + await subscription?.subscription?.stop(); } async #conversation(id: ConversationId): Promise { @@ -401,6 +450,6 @@ class GatewayClient { this.#disposed = true; const subscriptions = [...this.#subscriptions.values()]; this.#subscriptions.clear(); - for (const subscription of subscriptions) if (subscription !== "pending") await subscription.stop(); + for (const state of subscriptions) await state.subscription?.stop(); } } diff --git a/spikes/durable-multiview/host/host.ts b/spikes/durable-multiview/host/host.ts index ee449ce0..b1420278 100644 --- a/spikes/durable-multiview/host/host.ts +++ b/spikes/durable-multiview/host/host.ts @@ -31,6 +31,8 @@ export interface OpenHostOptions { readonly initialModel?: ModelRef & { readonly thinkingLevel?: ModelThinkingLevel }; /** 0 picks a free port. A fixed port lets clients reconnect to a restarted host. */ readonly port: number; + /** Exact browser origins allowed to use the gateway; native clients send no Origin. */ + readonly browserOrigins?: readonly string[]; /** How long a lock left by a killed host blocks the next one. proper-lockfile's minimum is 2000. */ readonly lockStaleMs?: number; /** Tool calls held for a human decision, offered to every client. */ @@ -107,6 +109,7 @@ export async function openHost(options: OpenHostOptions): Promise { session: { id: basename(options.dataDir), directory: options.dataDir, cwd: options.cwd }, token, port: options.port, + ...(options.browserOrigins === undefined ? {} : { browserOrigins: options.browserOrigins }), ...(options.approvals === undefined ? {} : { approvals: options.approvals }), ...(options.docs === undefined ? {} : { docs: options.docs }), }); diff --git a/spikes/durable-multiview/host/main.ts b/spikes/durable-multiview/host/main.ts index 7836cb40..141e4d85 100644 --- a/spikes/durable-multiview/host/main.ts +++ b/spikes/durable-multiview/host/main.ts @@ -31,6 +31,7 @@ const { values } = parseArgs({ "step-ms": { type: "string" }, "reminder-seconds": { type: "string" }, "investigation-module": { type: "string" }, + "browser-origin": { type: "string", multiple: true }, }, }); @@ -49,6 +50,7 @@ const common = { dataDir, cwd, port: Number(values.port), + browserOrigins: values["browser-origin"] ?? ["http://127.0.0.1:5199"], ...(values["lock-stale-ms"] === undefined ? {} : { lockStaleMs: Number(values["lock-stale-ms"]) }), ...demo?.hostOptions, }; @@ -93,9 +95,9 @@ const options = : await piOptions(); const host = await openHost(options); const model = options.initialModel === undefined ? null : `${options.initialModel.provider}/${options.initialModel.modelId}`; -// The web client (`bun run web`) reads the host from the fragment, which never leaves the browser. -const web = `http://127.0.0.1:5199/#token=${encodeURIComponent(host.token)}&url=${encodeURIComponent(host.url)}`; -console.log(JSON.stringify({ event: "ready", url: host.url, token: host.token, dataDir, model, web, demo: demo !== undefined })); +// Credentials stay in the local token file; generating a browser link is an explicit CLI action. +const web = "http://127.0.0.1:5199/"; +console.log(JSON.stringify({ event: "ready", url: host.url, tokenFile: join(dataDir, "token"), dataDir, model, web, demo: demo !== undefined })); if (demo !== undefined) { // Editors save in bursts; reload once the burst settles. diff --git a/spikes/durable-multiview/package.json b/spikes/durable-multiview/package.json index d06ef6c5..153aed4c 100644 --- a/spikes/durable-multiview/package.json +++ b/spikes/durable-multiview/package.json @@ -13,6 +13,7 @@ "typecheck": "tsc --noEmit -p tsconfig.json && tsc --noEmit -p web/tsconfig.json", "tui": "node tui/main.ts", "web": "vite --config web/vite.config.ts", + "web:link": "node cli/web-link.ts", "verify:three": "node scripts/three-clients.mjs", "verify:demo": "node scripts/demo-scenario.mjs" }, diff --git a/spikes/durable-multiview/protocol/frames.ts b/spikes/durable-multiview/protocol/frames.ts index 0d963772..645e4c4a 100644 --- a/spikes/durable-multiview/protocol/frames.ts +++ b/spikes/durable-multiview/protocol/frames.ts @@ -79,3 +79,46 @@ export type ClientFrame = readonly args: CallMethods[M]["args"]; }; }[CallMethod]; + +/** JSON from a socket has no TypeScript guarantees; reject it before dispatch. */ +export function isClientFrame(value: unknown): value is ClientFrame { + if (!record(value)) return false; + if (value.type === "subscribe" || value.type === "unsubscribe") { + const stream = value.stream; + return ( + typeof stream === "string" + && (stream === "tasks" || stream === "conversations" || stream === "approvals" || /^(?:conversation:|doc:.+:)[1-9]\d*$/.test(stream)) + ); + } + if (value.type !== "call" || !Number.isSafeInteger(value.id) || !record(value.args)) return false; + const args = value.args; + if (typeof args.conversationId !== "number" || !Number.isSafeInteger(args.conversationId) || args.conversationId < 1) return false; + switch (value.method) { + case "submit": + return typeof args.text === "string" && (args.whenBusy === "steer" || args.whenBusy === "followUp") && nonempty(args.requestId); + case "abort": + case "cycleThinking": + return true; + case "compact": + return args.instructions === undefined || typeof args.instructions === "string"; + case "setModel": + return record(args.model) && nonempty(args.model.provider) && nonempty(args.model.modelId); + case "approve": + return nonempty(args.approvalId) && typeof args.approved === "boolean" && nonempty(args.by); + case "fork": + return ( + typeof args.entryId === "string" && /^[1-9]\d*$/.test(args.entryId) + && (args.removeTools === undefined || (Array.isArray(args.removeTools) && args.removeTools.every(nonempty))) + ); + default: + return false; + } +} + +function record(value: unknown): value is Record { + return value !== null && typeof value === "object" && !Array.isArray(value); +} + +function nonempty(value: unknown): value is string { + return typeof value === "string" && value.length > 0; +} diff --git a/spikes/durable-multiview/scripts/demo-scenario.mjs b/spikes/durable-multiview/scripts/demo-scenario.mjs index 9a361454..004cca82 100644 --- a/spikes/durable-multiview/scripts/demo-scenario.mjs +++ b/spikes/durable-multiview/scripts/demo-scenario.mjs @@ -68,11 +68,14 @@ async function readyLine(child, predicate, label) { async function startHost() { const args = ["host/main.ts", "--faux-demo", "--data-dir", hostDir, "--cwd", T, "--port", String(PORT), "--lock-stale-ms", "2000"]; - args.push("--faux-tps", "60", "--pace-ms", "300", "--step-ms", "1500", "--reminder-seconds", "8", "--investigation-module", investigation); + args.push("--browser-origin", `http://127.0.0.1:${webPort}`, "--faux-tps", "60", "--pace-ms", "300", "--step-ms", "1500", "--reminder-seconds", "8", "--investigation-module", investigation); const host = spawn(process.execPath, args, { cwd: spike, stdio: ["ignore", "pipe", "inherit"] }); children.push(host); const ready = JSON.parse(await readyLine(host, (line) => line.startsWith("{") && JSON.parse(line).event === "ready", "host")); - return { host, ready }; + const token = readFileSync(ready.tokenFile, "utf8").trim(); + const web = new URL(ready.web); + web.hash = new URLSearchParams({ token, url: ready.url }).toString(); + return { host, ready: { ...ready, token, web: web.href } }; } async function until(what, predicate, timeout = 30000) { const deadline = Date.now() + timeout; diff --git a/spikes/durable-multiview/scripts/three-clients.mjs b/spikes/durable-multiview/scripts/three-clients.mjs index 0054b220..1b76938d 100644 --- a/spikes/durable-multiview/scripts/three-clients.mjs +++ b/spikes/durable-multiview/scripts/three-clients.mjs @@ -5,7 +5,7 @@ // // node scripts/three-clients.mjs [evidence-dir] # default: a new temp dir, printed at the end import { execFileSync, spawn } from "node:child_process"; -import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join, resolve } from "node:path"; import { fileURLToPath } from "node:url"; @@ -29,10 +29,13 @@ let paceLog = ""; const answer = "Root cause: the v2.3 migration locks the orders table for 40 s; deploys time out at 30 s. Fix: run the backfill in batches."; async function startHost() { - const host = spawn(process.execPath, ["host/main.ts", "--data-dir", hostDir, "--cwd", T, "--port", String(PORT), "--faux", answer, "--faux-tps", "5", "--lock-stale-ms", "2000"], { cwd: spike, stdio: ["ignore", "pipe", "inherit"] }); + const host = spawn(process.execPath, ["host/main.ts", "--data-dir", hostDir, "--cwd", T, "--port", String(PORT), "--browser-origin", "http://127.0.0.1:5299", "--faux", answer, "--faux-tps", "5", "--lock-stale-ms", "2000"], { cwd: spike, stdio: ["ignore", "pipe", "inherit"] }); children.push(host); const ready = await new Promise((resolve) => host.stdout.on("data", (d) => { const s = String(d); if (s.includes("ready")) resolve(JSON.parse(s)); })); - return { host, ready }; + const token = readFileSync(ready.tokenFile, "utf8").trim(); + const web = new URL(ready.web); + web.hash = new URLSearchParams({ token, url: ready.url }).toString(); + return { host, ready: { ...ready, token, web: web.href } }; } async function until(what, predicate, timeout = 30000) { const deadline = Date.now() + timeout; diff --git a/spikes/durable-multiview/test/approval-boundaries.test.ts b/spikes/durable-multiview/test/approval-boundaries.test.ts new file mode 100644 index 00000000..4e4d0fe6 --- /dev/null +++ b/spikes/durable-multiview/test/approval-boundaries.test.ts @@ -0,0 +1,58 @@ +import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; +import { expect, it } from "vitest"; +import { ApprovalsDoc } from "../host/demo/state.ts"; +import { connectTo, startDemoHost, useCleanups, waitForView } from "./support.ts"; + +const defer = useCleanups(); +const context = BACKGROUND_CONTEXT; + +it("rejects an approval for another conversation without releasing the original tool", async () => { + const { host, demo } = await startDemoHost(defer); + const client = await connectTo(defer, host); + const other = await host.harness.createConversation({ ownership: { kind: "ownerless" } }, context); + await client.controller.submit("roll back to v2.2", "followUp"); + await waitForView(client.view, (view) => view.approvals.length === 1); + const approval = client.view.current().approvals[0]!; + const board = demo.hostOptions.approvals!; + + await expect(board.decide(other.id, approval.id, true, "web")).rejects.toThrow(/conversation/i); + expect(board.value).toContainEqual(approval); + expect((await host.harness.snapshot(ApprovalsDoc, other.id, context))?.decisions[approval.id]).toBeUndefined(); + expect((await host.harness.snapshot(ApprovalsDoc, approval.conversationId, context))?.decisions[approval.id]).toBeUndefined(); + + const denied = await board.decide(approval.conversationId, approval.id, false, "tui"); + expect(denied).toMatchObject({ first: true, decision: { approved: false, by: "tui" } }); + expect(await board.decide(approval.conversationId, approval.id, true, "web")).toMatchObject({ + first: false, + decision: { approved: false, by: "tui" }, + }); + await expect(board.decide(other.id, approval.id, true, "web")).rejects.toThrow(); + expect((await host.harness.snapshot(ApprovalsDoc, other.id, context))?.decisions[approval.id]).toBeUndefined(); +}); + +it.each([false, true])("rejects a foreign stored decision after cancellation (restart: %s)", async (restart) => { + let { host, demo } = await startDemoHost(defer); + const client = await connectTo(defer, host); + const other = await host.harness.createConversation({ ownership: { kind: "ownerless" } }, context); + await client.controller.submit("roll back to v2.2", "followUp"); + await waitForView(client.view, (view) => view.approvals.length === 1); + const approval = client.view.current().approvals[0]!; + await client.controller.abort(); + await waitForView(client.view, (view) => view.approvals.length === 0); + + // Earlier hosts could persist a decision under the wrong conversation before checking ownership. + await host.harness.commit(async (tx) => { + (await tx.doc(ApprovalsDoc, other.id)).decisions[approval.id] = { approved: true, by: "other", at: 1 }; + }, context); + if (restart) { + const dataDir = client.view.current().session.directory; + client.close(); + await host.close(); + ({ host, demo } = await startDemoHost(defer, { dataDir })); + } + + const board = demo.hostOptions.approvals!; + await expect(board.decide(other.id, approval.id, true, "retry")).rejects.toThrow(/conversation/i); + await expect(board.decide(approval.conversationId, approval.id, true, "owner")).rejects.toThrow(/No pending approval/); + expect((await host.harness.snapshot(ApprovalsDoc, approval.conversationId, context))?.decisions[approval.id]).toBeUndefined(); +}); diff --git a/spikes/durable-multiview/test/crash-recovery.test.ts b/spikes/durable-multiview/test/crash-recovery.test.ts index 0a364985..161a3a4c 100644 --- a/spikes/durable-multiview/test/crash-recovery.test.ts +++ b/spikes/durable-multiview/test/crash-recovery.test.ts @@ -1,5 +1,6 @@ import { type ChildProcess, spawn } from "node:child_process"; import { once } from "node:events"; +import { readFile } from "node:fs/promises"; import { createInterface } from "node:readline"; import { fileURLToPath } from "node:url"; import { afterEach, expect, it } from "vitest"; @@ -30,8 +31,8 @@ async function spawnHost(dataDir: string, port: number, model: readonly string[] }); const lines = createInterface({ input: child.stdout! }); for await (const line of lines) { - const event = JSON.parse(line) as { event: string; url: string; token: string }; - if (event.event === "ready") return { child, url: event.url, token: event.token }; + const event = JSON.parse(line) as { event: string; url: string; tokenFile: string }; + if (event.event === "ready") return { child, url: event.url, token: (await readFile(event.tokenFile, "utf8")).trim() }; } throw new Error("host exited before it was ready"); } diff --git a/spikes/durable-multiview/test/gateway-boundaries.test.ts b/spikes/durable-multiview/test/gateway-boundaries.test.ts new file mode 100644 index 00000000..c050a363 --- /dev/null +++ b/spikes/durable-multiview/test/gateway-boundaries.test.ts @@ -0,0 +1,211 @@ +import { spawn } from "node:child_process"; +import { once } from "node:events"; +import { readFile } from "node:fs/promises"; +import { join } from "node:path"; +import { createInterface } from "node:readline"; +import { fileURLToPath } from "node:url"; +import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; +import { ROOT_CONVERSATION_ID } from "@earendil-works/pi-durable"; +import { PlanDoc } from "../host/demo/state.ts"; +import { createModels } from "@earendil-works/pi-ai/models"; +import { describe, expect, it, vi } from "vitest"; +import { WebSocket } from "ws"; +import { startGateway, type GatewayOptions } from "../host/gateway.ts"; +import type { ServerFrame } from "../protocol/frames.ts"; +import { startDemoHost, startFauxHost, tempDir, useCleanups } from "./support.ts"; + +const defer = useCleanups(); +const spikeDir = fileURLToPath(new URL("..", import.meta.url)); + +function barrier() { + let resolve!: () => void; + const promise = new Promise((done) => { resolve = done; }); + return { promise, resolve }; +} + +async function socketTo(url: string, token: string, origin?: string) { + const socket = new WebSocket(`${url}?token=${encodeURIComponent(token)}`, origin === undefined ? {} : { origin }); + defer(() => socket.terminate()); + const frames: ServerFrame[] = []; + socket.on("message", (data) => frames.push(JSON.parse(String(data)) as ServerFrame)); + await once(socket, "open"); + return { socket, frames }; +} + +async function fixture(options: Partial = {}) { + const host = await startFauxHost(defer); + const gateway = await startGateway({ + harness: host.harness, + token: host.token, + models: createModels(), + modelSummaries: () => [], + session: { id: "boundary-test", directory: "", cwd: "" }, + port: 0, + ...options, + }); + defer(() => gateway.close()); + return { host, gateway }; +} + +describe("gateway boundaries", () => { + it.each([false, true])("keeps the latest subscription when an earlier acquisition settles late (failure: %s)", async (fails) => { + const { host, gateway } = await fixture(); + const acquire = barrier(); + const release = barrier(); + const finished = barrier(); + const original = host.harness.watchTaskGraph.bind(host.harness); + const watch = vi.spyOn(host.harness, "watchTaskGraph").mockImplementationOnce(async (context) => { + const first = await original(context); + const start = first.start.bind(first); + const stop = first.stop.bind(first); + vi.spyOn(first, "start").mockImplementation((listener) => { start(listener); finished.resolve(); }); + vi.spyOn(first, "stop").mockImplementation(async () => { const result = await stop(); finished.resolve(); return result; }); + acquire.resolve(); + await release.promise; + if (fails) { + await first.stop(); + throw new Error("obsolete acquisition failed"); + } + return first; + }); + defer(() => watch.mockRestore()); + const client = await socketTo(gateway.url, host.token); + const subscribe = () => client.socket.send(JSON.stringify({ type: "subscribe", stream: "tasks" })); + subscribe(); + await acquire.promise; + subscribe(); + while (!client.frames.some((frame) => frame.type === "snapshot" && frame.stream === "tasks")) await once(client.socket, "message"); + release.resolve(); + await finished.promise; + client.socket.send(JSON.stringify({ type: "subscribe", stream: "conversations" })); + while (!client.frames.some((frame) => frame.type === "snapshot" && frame.stream === "conversations")) await once(client.socket, "message"); + expect(client.frames.filter((frame) => frame.type === "snapshot" && frame.stream === "tasks")).toHaveLength(1); + expect(client.frames.filter((frame) => frame.type === "ended" && frame.stream === "tasks")).toEqual([]); + }); + + it("drops an in-flight watch when the client unsubscribes", async () => { + const { host, gateway } = await fixture(); + const acquire = barrier(); + const release = barrier(); + const stopped = barrier(); + const original = host.harness.watchTaskGraph.bind(host.harness); + const watch = vi.spyOn(host.harness, "watchTaskGraph").mockImplementationOnce(async (context) => { + const pending = await original(context); + const stop = pending.stop.bind(pending); + vi.spyOn(pending, "stop").mockImplementation(async () => { + const result = await stop(); + stopped.resolve(); + return result; + }); + acquire.resolve(); + await release.promise; + return pending; + }); + defer(() => watch.mockRestore()); + const client = await socketTo(gateway.url, host.token); + client.socket.send(JSON.stringify({ type: "subscribe", stream: "tasks" })); + await acquire.promise; + client.socket.send(JSON.stringify({ type: "unsubscribe", stream: "tasks" })); + client.socket.send(JSON.stringify({ type: "subscribe", stream: "conversations" })); + while (!client.frames.some((frame) => frame.type === "snapshot" && frame.stream === "conversations")) await once(client.socket, "message"); + release.resolve(); + await stopped.promise; + client.socket.send(JSON.stringify({ type: "subscribe", stream: "approvals" })); + while (!client.frames.some((frame) => frame.type === "snapshot" && frame.stream === "approvals")) await once(client.socket, "message"); + expect(client.frames.filter((frame) => "stream" in frame && frame.stream === "tasks")).toEqual([]); + }); + + it("streams a document created while its initial absence is being read", async () => { + const { host } = await startDemoHost(defer); + const absent = barrier(); + const release = barrier(); + const original = host.harness.watchDoc.bind(host.harness); + const watch = vi.spyOn(host.harness, "watchDoc").mockImplementationOnce(async (...args) => { + const result = await original(...args); + expect(result).toBeUndefined(); + absent.resolve(); + await release.promise; + return result; + }); + defer(() => watch.mockRestore()); + const client = await socketTo(host.url, host.token); + const stream = `doc:demo.plan:${ROOT_CONVERSATION_ID}`; + client.socket.send(JSON.stringify({ type: "subscribe", stream })); + await absent.promise; + await host.harness.commit(async (tx) => { + (await tx.doc(PlanDoc, ROOT_CONVERSATION_ID)).items = [{ text: "Investigate release", status: "doing" }]; + }, BACKGROUND_CONTEXT); + release.resolve(); + const received = barrier(); + const check = () => { + if (client.frames.some((frame) => frame.type === "snapshot" && frame.stream === stream && frame.value !== null)) received.resolve(); + }; + client.socket.on("message", check); + defer(() => { client.socket.off("message", check); }); + // The timeout is only a failure bound; the commit/acquisition barrier triggers the race. + await Promise.race([received.promise, new Promise((_, reject) => setTimeout(() => reject(new Error("created document never arrived")), 1000))]); + expect(client.frames.findLast((frame) => frame.type === "snapshot" && frame.stream === stream)).toMatchObject({ + value: { items: [{ text: "Investigate release", status: "doing" }] }, + }); + }); + + it("rejects an untrusted browser even when it has the token", async () => { + const { host, gateway } = await fixture(); + const { socket, frames } = await socketTo(gateway.url, host.token, "https://untrusted.example"); + const result = await Promise.race([ + once(socket, "close").then(([code]) => code), + once(socket, "message").then(() => "accepted"), + ]); + expect(result).toBe(4401); + expect(frames).toEqual([]); + }); + + it("admits only the configured browser origin while retaining native access", async () => { + const origin = "http://127.0.0.1:5199"; + const { host, gateway } = await fixture({ browserOrigins: [origin] }); + for (const [candidate, accepted] of [[undefined, true], [origin, true], ["http://127.0.0.1:5200", false], ["null", false]] as const) { + const { socket, frames } = await socketTo(gateway.url, host.token, candidate); + const result = frames.length > 0 ? "hello" : await Promise.race([ + once(socket, "message").then(() => "hello"), + once(socket, "close").then(([code]) => code), + ]); + expect(result).toBe(accepted ? "hello" : 4401); + } + }); + + it("closes malformed clients while keeping the host available", async () => { + const dir = await tempDir(); + defer(dir.remove); + const child = spawn(process.execPath, ["host/main.ts", "--data-dir", dir.path, "--cwd", dir.path, "--port", "0", "--faux", "done"], { + cwd: spikeDir, + stdio: ["ignore", "pipe", "pipe"], + }); + defer(async () => { + if (child.exitCode !== null || child.signalCode !== null) return; + child.kill("SIGKILL"); + await once(child, "exit"); + }); + const lines = createInterface({ input: child.stdout! }); + let url: string | undefined; + for await (const line of lines) { + const ready = JSON.parse(line) as { event: string; url: string }; + if (ready.event === "ready") { + url = ready.url; + break; + } + } + expect(url).toBeDefined(); + const token = (await readFile(join(dir.path, "token"), "utf8")).trim(); + for (const malformed of ["{", "null", '{"type":"subscribe","stream":null}', '{"type":"call","id":1,"method":"submit","args":{}}']) { + const { socket } = await socketTo(url!, token); + const closed = once(socket, "close"); + socket.send(malformed); + expect((await closed)[0]).toBe(1008); + expect(child.exitCode).toBeNull(); + } + const healthy = await socketTo(url!, token); + healthy.socket.send(JSON.stringify({ type: "subscribe", stream: "conversations" })); + while (!healthy.frames.some((frame) => frame.type === "snapshot")) await once(healthy.socket, "message"); + expect(healthy.frames.some((frame) => frame.type === "snapshot")).toBe(true); + }); +}); diff --git a/spikes/durable-multiview/test/host-reload.test.ts b/spikes/durable-multiview/test/host-reload.test.ts index 0b2582eb..34f393b6 100644 --- a/spikes/durable-multiview/test/host-reload.test.ts +++ b/spikes/durable-multiview/test/host-reload.test.ts @@ -37,9 +37,9 @@ it("reloads the investigation tools after consecutive atomic editor saves", asyn }); let address: { url: string; token: string } | undefined; for await (const line of createInterface({ input: child.stdout! })) { - const event = JSON.parse(line) as { event: string; url: string; token: string }; + const event = JSON.parse(line) as { event: string; url: string; tokenFile: string }; if (event.event !== "ready") continue; - address = event; + address = { url: event.url, token: (await readFile(event.tokenFile, "utf8")).trim() }; break; } if (address === undefined) throw new Error("host exited before it was ready"); diff --git a/spikes/durable-multiview/test/host-startup.test.ts b/spikes/durable-multiview/test/host-startup.test.ts new file mode 100644 index 00000000..259b0004 --- /dev/null +++ b/spikes/durable-multiview/test/host-startup.test.ts @@ -0,0 +1,43 @@ +import { spawn } from "node:child_process"; +import { once } from "node:events"; +import { readFile } from "node:fs/promises"; +import { join } from "node:path"; +import { createInterface } from "node:readline"; +import { fileURLToPath } from "node:url"; +import { expect, it } from "vitest"; +import { connectRemoteDurable } from "../protocol/remote-durable.ts"; +import { tempDir, useCleanups } from "./support.ts"; + +const defer = useCleanups(); + +it("keeps the host credential out of startup logs while local clients can connect", async () => { + const data = await tempDir(); + defer(data.remove); + const child = spawn(process.execPath, ["host/main.ts", "--faux", "ready", "--data-dir", data.path, "--port", "0"], { + cwd: fileURLToPath(new URL("..", import.meta.url)), + stdio: ["ignore", "pipe", "pipe"], + }); + defer(async () => { + if (child.exitCode !== null || child.signalCode !== null) return; + const exited = once(child, "exit"); + child.kill("SIGTERM"); + await exited; + }); + let output = ""; + child.stdout.on("data", (chunk) => { output += String(chunk); }); + child.stderr.on("data", (chunk) => { output += String(chunk); }); + let ready: { url: string; tokenFile: string; web: string } | undefined; + for await (const line of createInterface({ input: child.stdout })) { + const event = JSON.parse(line); + if (event.event === "ready") { ready = event; break; } + } + if (ready === undefined) throw new Error("host exited before ready"); + const token = (await readFile(join(data.path, "token"), "utf8")).trim(); + // A boolean assertion keeps the credential out of failed-test diagnostics too. + expect(output.includes(token), "startup logs contain the host credential").toBe(false); + expect(ready.tokenFile).toBe(join(data.path, "token")); + expect(new URL(ready.web).hash).toBe(""); + const client = await connectRemoteDurable({ url: ready.url, token }); + defer(() => client.close()); + expect(client.view.current().connection).toBe("connected"); +}); diff --git a/spikes/durable-multiview/test/support.ts b/spikes/durable-multiview/test/support.ts index 138689af..79565953 100644 --- a/spikes/durable-multiview/test/support.ts +++ b/spikes/durable-multiview/test/support.ts @@ -28,6 +28,7 @@ export async function startFauxHost( tokensPerSecond = 200, dataDir, port = 0, + browserOrigins, }: { answers?: string[]; tokensPerSecond?: number; @@ -35,6 +36,7 @@ export async function startFauxHost( dataDir?: string; /** Fixed, so clients reconnect to a restarted host. */ port?: number; + browserOrigins?: readonly string[]; } = {}, ): Promise { const dir = dataDir === undefined ? await tempDir() : { path: dataDir, remove: () => {} }; @@ -51,6 +53,7 @@ export async function startFauxHost( modelSummaries: () => [{ provider: model.provider, modelId: model.id, name: model.name, contextWindow: model.contextWindow }], initialModel: { provider: model.provider, modelId: model.id }, port, + ...(browserOrigins === undefined ? {} : { browserOrigins }), }); defer(() => host.close()); return host; @@ -62,9 +65,10 @@ export async function startDemoHost( { dataDir, port = 0, + browserOrigins, stepMs = 150, investigationModule, - }: { dataDir?: string; port?: number; stepMs?: number; investigationModule?: string } = {}, + }: { dataDir?: string; port?: number; browserOrigins?: readonly string[]; stepMs?: number; investigationModule?: string } = {}, ): Promise<{ host: OpenedHost; demo: Demo }> { const dir = dataDir === undefined ? await tempDir() : { path: dataDir, remove: () => {} }; defer(dir.remove); @@ -81,6 +85,7 @@ export async function startDemoHost( modelSummaries: () => [{ provider: model.provider, modelId: model.id, name: model.name, contextWindow: model.contextWindow }], initialModel: { provider: model.provider, modelId: model.id }, port, + ...(browserOrigins === undefined ? {} : { browserOrigins }), ...demo.hostOptions, settings: demo.settings(), }); diff --git a/spikes/durable-multiview/test/web-ui.test.ts b/spikes/durable-multiview/test/web-ui.test.ts index f69a22e9..2a28c1ea 100644 --- a/spikes/durable-multiview/test/web-ui.test.ts +++ b/spikes/durable-multiview/test/web-ui.test.ts @@ -14,21 +14,26 @@ import { connectTo, freePort, startFauxHost, tempDir, useCleanups, waitForView } const defer = useCleanups(); const webConfig = fileURLToPath(new URL("../web/vite.config.ts", import.meta.url)); -async function openPage(host: OpenedHost, width: number) { +async function startWeb(): Promise { const vite = await createServer({ configFile: webConfig, server: { port: await freePort() }, logLevel: "error" }); await vite.listen(); defer(() => vite.close()); + return new URL(vite.resolvedUrls!.local[0]!).origin; +} + +async function openPage(host: OpenedHost, width: number, webOrigin: string) { const browser = await chromium.launch(); defer(() => browser.close()); const page = await browser.newPage({ viewport: { width, height: 844 } }); - await page.goto(`${vite.resolvedUrls!.local[0]}#token=${encodeURIComponent(host.token)}&url=${encodeURIComponent(host.url)}`); + await page.goto(`${webOrigin}/#token=${encodeURIComponent(host.token)}&url=${encodeURIComponent(host.url)}`); await page.getByRole("textbox").waitFor(); return page; } it("keeps the chat usable on a phone with navigation and live state still reachable", async () => { - const host = await startFauxHost(defer); - const page = await openPage(host, 390); + const webOrigin = await startWeb(); + const host = await startFauxHost(defer, { browserOrigins: [webOrigin] }); + const page = await openPage(host, 390, webOrigin); const composer = page.getByRole("textbox"); expect((await composer.boundingBox())!.width).toBeGreaterThan(300); @@ -45,8 +50,9 @@ it("keeps the chat usable on a phone with navigation and live state still reacha }); it("preserves an unsent follow-up and disables write controls while reconnecting", async () => { - const host = await startFauxHost(defer, { answers: ["An answer to fork."] }); - const page = await openPage(host, 1280); + const webOrigin = await startWeb(); + const host = await startFauxHost(defer, { browserOrigins: [webOrigin], answers: ["An answer to fork."] }); + const page = await openPage(host, 1280, webOrigin); const composer = page.getByRole("textbox"); await composer.fill("hello"); await composer.press("Enter"); @@ -65,11 +71,12 @@ it.each([ ["thinking-only", fauxAssistantMessage(fauxThinking("Compare the release evidence."))], ["tool-only", fauxAssistantMessage(fauxToolCall("search_logs", { release: "v2.3" }), { stopReason: "toolUse" })], ])("forks a persisted %s answer from its message action", async (_, message) => { - const host = await startFauxHost(defer, { answers: ["The fork continued."] }); + const webOrigin = await startWeb(); + const host = await startFauxHost(defer, { browserOrigins: [webOrigin], answers: ["The fork continued."] }); const root = await host.harness.root(BACKGROUND_CONTEXT); const source = await root.commit((tx) => tx.appendEntry(AssistantEntry, root.id, { model: [message] }), BACKGROUND_CONTEXT); const observer = await connectTo(defer, host); - const page = await openPage(host, 1280); + const page = await openPage(host, 1280, webOrigin); const fork = page.getByRole("button", { name: "Fork", exact: true }); await expect.poll(() => fork.count()).toBe(1); @@ -87,6 +94,7 @@ it.each([ }); it("shows interruption without a text bubble and never offers a fork for streaming or aborted thinking", async () => { + const webOrigin = await startWeb(); const directory = await tempDir(); defer(directory.remove); const faux = fauxProvider({ tokensPerSecond: 20, tokenSize: { min: 1, max: 1 } }); @@ -101,10 +109,11 @@ it("shows interruption without a text bubble and never offers a fork for streami modelSummaries: () => [], initialModel: { provider: model.provider, modelId: model.id }, port: 0, + browserOrigins: [webOrigin], }); defer(() => host.close()); const observer = await connectTo(defer, host); - const page = await openPage(host, 390); + const page = await openPage(host, 390, webOrigin); await page.getByRole("textbox").fill("Investigate the release."); await page.getByRole("textbox").press("Enter"); await page.getByText("Thinking…", { exact: true }).waitFor(); diff --git a/spikes/durable-multiview/test/web.test.ts b/spikes/durable-multiview/test/web.test.ts index 861366cb..1347a067 100644 --- a/spikes/durable-multiview/test/web.test.ts +++ b/spikes/durable-multiview/test/web.test.ts @@ -10,12 +10,13 @@ const defer = useCleanups(); const webConfig = fileURLToPath(new URL("../web/vite.config.ts", import.meta.url)); it("shows the same conversation in the web page and the TUI, driven from either", async () => { - const host = await startFauxHost(defer, { answers: ["answer-for-the-web", "answer-for-the-tui"] }); - const observer = await connectTo(defer, host); - const vite = await createServer({ configFile: webConfig, server: { port: await freePort() }, logLevel: "error" }); await vite.listen(); defer(() => vite.close()); + const webOrigin = new URL(vite.resolvedUrls!.local[0]!).origin; + const host = await startFauxHost(defer, { answers: ["answer-for-the-web", "answer-for-the-tui"], browserOrigins: [webOrigin] }); + const observer = await connectTo(defer, host); + const browser = await chromium.launch(); defer(() => browser.close()); const page = await browser.newPage();