fix(spike): enforce durable host and approval boundaries

.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` 下面嵌了第二份）。
 

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 文件取凭证，不再从日志读取。

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);

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);

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:<kind>:<conversationId>` streams. */
   readonly docs?: readonly ConversationDocToken<JsonObject>[];
@@ -68,8 +71,10 @@ export async function startGateway(options: GatewayOptions): Promise<Gateway> {
   });
   const clients = new Set<GatewayClient>();
   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<unknown> | void };
+type SubscriptionState = { subscription?: Subscription };
+type SendFrame = (frame: ServerFrame) => Promise<void>;
 
 class GatewayClient {
   readonly #socket: WebSocket;
   readonly #options: GatewayOptions;
   readonly #conversations: ConversationList;
-  readonly #subscriptions = new Map<StreamName, Subscription | "pending">();
+  readonly #subscriptions = new Map<StreamName, SubscriptionState>();
   #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<void> {
-    // 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<Subscription> {
+  async #open(stream: StreamName, send: SendFrame): Promise<Subscription> {
     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<Subscription> {
+  async #openDoc(stream: StreamName, kind: string, id: ConversationId, send: SendFrame): Promise<Subscription> {
     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<void>((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<T>(stream: StreamName, watch: WatchHandle<T>): Promise<Subscription> {
-    await this.send({ type: "snapshot", stream, value: watch.value });
-    watch.start((_value, ops) => this.send({ type: "ops", stream, ops }));
+  async #forward<T>(stream: StreamName, watch: WatchHandle<T>, send: SendFrame): Promise<Subscription> {
+    await send({ type: "snapshot", stream, value: watch.value });
+    watch.start((_value, ops) => send({ type: "ops", stream, ops }));
     return watch;
   }
 
   async #unsubscribe(stream: StreamName): Promise<void> {
     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<Conversation> {
@@ -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();
   }
 }

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<OpenedHost> {
       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 }),
     });

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.

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"
   },