From 1388ea52452b8a4e849e3c44b9535c5e00a617a0 Mon Sep 17 00:00:00 2001 From: Kieran Zhang Date: Fri, 2 Oct 2026 19:13:23 +0800 Subject: [PATCH] feat(spike): Pace dev page as the third durable client (P3) Pace's backend relays frames between a local Durable host and the renderer over the existing IPC as unsequenced events, without parsing them, so the renderer applies every op once. RemoteDurable gains a transport seam; the DEV-only #/durable-spike page reuses the web client's workbench over the relay. The runtime gateway client drops the new ephemeral event types. Includes a scripted three-client proof against an isolated electron-vite dev instance. --- .scratch/durable-multiview/HANDOFF.md | 35 ++++- apps/desktop/electron.vite.config.ts | 2 + apps/desktop/src/app/main.tsx | 11 ++ .../dev/durable-spike/backend-relay.test.ts | 49 ++++++ .../src/dev/durable-spike/backend-relay.ts | 40 +++++ .../dev/durable-spike/durable-spike-page.tsx | 16 ++ .../runtime/runtime-gateway-client.test.ts | 2 +- .../runtime/runtime-gateway-client.ts | 11 +- bun.lock | 3 + packages/backend/src/service.test.ts | 45 ++++++ packages/backend/src/service.ts | 48 ++++++ packages/backend/src/spikes/durable-bridge.ts | 88 +++++++++++ spikes/durable-multiview/package.json | 4 +- .../protocol/remote-durable.ts | 50 +++--- .../durable-multiview/protocol/transport.ts | 81 ++++++++++ .../scripts/three-clients.mjs | 144 ++++++++++++++++++ spikes/durable-multiview/test/relay.test.ts | 90 +++++++++++ spikes/durable-multiview/test/support.ts | 12 +- spikes/durable-multiview/web/App.tsx | 13 +- spikes/durable-multiview/web/use-remote.ts | 13 +- tsconfig.json | 4 +- vite.config.ts | 2 + 22 files changed, 726 insertions(+), 37 deletions(-) create mode 100644 apps/desktop/src/dev/durable-spike/backend-relay.test.ts create mode 100644 apps/desktop/src/dev/durable-spike/backend-relay.ts create mode 100644 apps/desktop/src/dev/durable-spike/durable-spike-page.tsx create mode 100644 packages/backend/src/spikes/durable-bridge.ts create mode 100644 spikes/durable-multiview/protocol/transport.ts create mode 100644 spikes/durable-multiview/scripts/three-clients.mjs create mode 100644 spikes/durable-multiview/test/relay.test.ts diff --git a/.scratch/durable-multiview/HANDOFF.md b/.scratch/durable-multiview/HANDOFF.md index a7e47759..a7f99996 100644 --- a/.scratch/durable-multiview/HANDOFF.md +++ b/.scratch/durable-multiview/HANDOFF.md @@ -1,7 +1,7 @@ # Handoff：Pi Durable 多端同屏 spike - 日期：2026-10-02 -- 状态：P0–P2 完成（见 §10–§12），下一步 P3（Pace dev 页面）。PR 栈：P0 [#426](https://github.com/bubbleptr/pace/pull/426) ← P1 [#427](https://github.com/bubbleptr/pace/pull/427) ← P2 `feat/durable-multiview-web`。前置的 Pi 1.0 升级 [PR #425](https://github.com/bubbleptr/pace/pull/425) 已合入，分支 `feat/durable-multiview-spike` 从 `origin/main` 开出。 +- 状态：P0–P3 完成（见 §10–§13），下一步 P4（演示扩展与剧本）。PR 栈：P0 [#426](https://github.com/bubbleptr/pace/pull/426) ← P1 [#427](https://github.com/bubbleptr/pace/pull/427) ← P2 [#429](https://github.com/bubbleptr/pace/pull/429) ← P3 `feat/durable-multiview-pace`。前置的 Pi 1.0 升级 [PR #425](https://github.com/bubbleptr/pace/pull/425) 已合入，分支 `feat/durable-multiview-spike` 从 `origin/main` 开出。 - 关联：[docs/research/pi-durable-analysis.md](../../docs/research/pi-durable-analysis.md)（框架分析，用户的写作素材，只在本地工作区，不提交）、[ADR-0040](../../docs/adr/0040-root-session-process-isolation.md)、[ADR-0042](../../docs/adr/0042-pace-as-chord-presentation-host.md)、[ADR-0044](../../docs/adr/0044-single-writer-session-projection-in-renderer.md) - 目的：把上一个会话里已核实的事实、已定的决策和待决问题交给新会话，避免重查。 @@ -234,3 +234,36 @@ Pace 页面放在 `apps/desktop/src/pages/` 下的 dev-only 路由，backend 侧 - **根会话的 ID 是 1，不是 0。** 用 `ROOT_CONVERSATION_ID`，不要写死。 - **修复一个上游带过来的不一致**：网关启动时扫描会话列表照搬了上游的写法，不给 main 标题，但运行中提交监听器会给 main 补上标题。结果宿主重启后，main 在列表里的标题就没了。现在启动时 main 也取第一条用户消息作为标题；有测试覆盖重启前后列表一致。 - Astryx 的 `ChatComposer` 在 `isStopShown` 时 Enter 照样调用 `onSubmit`，只有按钮变成停止，所以 Enter 在忙碌时就是 steer，不需要单独的 steer 按钮。 + +## 13. P3 进展（2026-10-02）：Pace dev 页面（三端同屏） + +### 13.1 落地的内容（§3.3 选了 a） + +- **传输抽象**（`protocol/transport.ts`）：`RemoteDurable` 不再直接 new WebSocket，而是依赖 `FrameTransport`。实现有两种：`webSocketTransport`（TUI、网页、测试用）和 `relayTransport`（经注入的 `FrameRelay` 收发帧，Pace 用）。中继只搬运帧，ops 仍由客户端应用。 +- **后端中继** `packages/backend/src/spikes/durable-bridge.ts`：按 `connectionId` 持有到宿主的 WebSocket（Electron utilityProcess 自带全局 `WebSocket`），帧原样转成 `seq: 0` 的临时事件 `durable_spike.frame` / `durable_spike.closed`，不写 journal，不分配序号，和终端流一样。`service.ts` 新增 `durable_spike_connect` / `_send` / `_disconnect` 三个命令。 + - token 从 `/experimental/durable-multiview/default/token` 读取，即宿主默认数据目录里的那个文件；没有 token 时报「先 `bun run host`」。 + - **只接受回环地址**：token 等于宿主 Agent 的完全控制权，不能发到别的机器。 + - 默认地址可以用 `PACE_DURABLE_SPIKE_URL` 覆盖，验证脚本靠它避开操作者自己的 7420。 +- **渲染进程**：`src/dev/durable-spike/` 不属于 FSD 的任何一层（同 `src/dev/ui-intent/`）。`backend-relay.ts` 把 `invoke` / `onBackendEvent` 适配成 `FrameRelay`。`durable-spike-page.tsx` 直接复用网页版的 `RemoteWorkbench`，**三端用的是同一份展示代码**，只有传输不同。路由 `#/durable-spike` 只在 `import.meta.env.DEV` 分支里懒加载注册，并加入了 preflight 豁免。 +- **`runtime-gateway-client.ts` 的临时事件忽略列表加入了两个新类型**。测试先失败过：不加的话，spike 帧会进入运行时状态和去重集合。 +- 根配置：`tsconfig.json` 加了 `allowImportingTsExtensions`（spike 在 Node 下运行的文件带 `.ts` 扩展名导入，根 `noEmit` 下无副作用）和路径别名 `@pace/durable-spike/*`；`vite.config.ts` 和 `electron.vite.config.ts` 的 renderer 也加了同名别名。 +- spike 和 `apps/desktop` 解析到同一份 Astryx、主题、StyleX、React，Pace 的 `` 可以覆盖复用进来的组件（已核对符号链接）。 + +### 13.2 验收 + +- `test/relay.test.ts`（spike，5 条）：用**真实的后端中继**连假模型宿主。经中继的客户端和直连客户端看到并驱动同一个对话，最终视图相等；宿主重启后经中继自动重连；可配置默认地址；没有 token 和非回环地址都给出明确错误。 +- `service.test.ts`：三个命令转发给中继；中继事件变成 `seq: 0` 的 envelope；`dispose` 时关闭中继。 +- `runtime-gateway-client.test.ts`：两个新事件类型在运行时状态和去重之前就被丢弃。 +- `backend-relay.test.ts`：命令映射；事件按类型和字段过滤。 +- **三端同屏实证** `scripts/three-clients.mjs`（`bun run verify:three`）：假模型宿主 + 隔离的 `electron-vite dev` Pace（独立 `PACE_DATA_DIR`、`PI_CODING_AGENT_DIR`、`--user-data-dir`，加 `PACE_E2E=1`，通过 CDP 驱动）+ 网页 + pty TUI。Pace 提问，网页和 TUI 都看到；网页和 TUI 提问，Pace 都看到；流式中途 `kill -9`，Pace 显示断线，重启后自动续上，出现 `interrupted` 片段和完整答案。截图和 TUI 屏幕转储写入证据目录。结束时按 `--user-data-dir` 精确结束 Electron（electron-vite 拉起的 Electron 不在它的进程组里），并删除临时目录。 +- 生产构建的 renderer 里查不到 spike 页面。后端 bundle 里有中继代码，但不被调用就什么都不做。 + +### 13.3 新发现（P5 结论文档要用） + +- **`electron-vite build` 总会去掉 DEV 分支**，`--mode development` 或 `NODE_ENV=development` 都不行。所以 verify-pace 的 Playwright 夹具（启动生产构建）够不着 dev-only 页面，要用 `electron-vite dev --remoteDebuggingPort` 加 CDP。 +- **Pace 接入 Durable 的最小改动面**： + - 后端：一个不解析帧的 WebSocket 中继，加三个命令。 + - 核心协议：不需要改，复用 `seq: 0` 的临时 envelope。 + - 渲染进程：一个传输适配器，加上把新事件类型列入忽略列表。 + - 视图状态由渲染进程持有，ops 只应用一次，没有经过 Session Projection 和 Journal。这一点和 ADR-0044「渲染进程单写者投影」一致，但**和 ADR-0021 的双持久化轨道是并列的另一条路**：Durable 自己的 SQLite 才是真相，Pace 不再需要 Journal。结论文档要讨论这一点。 +- 后端只做透传，所以 Durable 的背压（`watch()` 积压 100 帧后折叠）在宿主到后端这一段生效。后端到渲染进程这一段的 IPC 没有背压：慢渲染进程会在 Electron IPC 队列里积压。spike 里看不到问题，但真实接入要考虑。 diff --git a/apps/desktop/electron.vite.config.ts b/apps/desktop/electron.vite.config.ts index c48308f2..645e2c56 100644 --- a/apps/desktop/electron.vite.config.ts +++ b/apps/desktop/electron.vite.config.ts @@ -223,6 +223,8 @@ const reactPackage = dirname(requireFromRepo.resolve("react/package.json")); const reactDomPackage = dirname(requireFromRepo.resolve("react-dom/package.json")); const rendererReactAlias = { ...coreAlias, + // Dev-only Durable multiview spike page; removed with spikes/durable-multiview. + "@pace/durable-spike": resolve(__dirname, "../../spikes/durable-multiview"), react: reactPackage, "react-dom": reactDomPackage, }; diff --git a/apps/desktop/src/app/main.tsx b/apps/desktop/src/app/main.tsx index 78037b44..cf4d28cf 100644 --- a/apps/desktop/src/app/main.tsx +++ b/apps/desktop/src/app/main.tsx @@ -38,6 +38,7 @@ function isPreflightExemptPath(pathname: string) { return ( pathname === "/preflight" || pathname === "/design" || + pathname === "/durable-spike" || pathname === "/settings" || pathname.startsWith("/settings/") ); @@ -226,6 +227,16 @@ const devOnlyRoutes = import.meta.env.DEV default: (await import("@/pages/design")).DesignPage, })), }), + // Durable multiview spike; removed with spikes/durable-multiview. + createRoute({ + getParentRoute: () => rootRoute, + path: "/durable-spike", + wrapInSuspense: true, + errorComponent: DevRouteError, + component: React.lazy(async () => ({ + default: (await import("@/dev/durable-spike/durable-spike-page")).DurableSpikePage, + })), + }), ] : []; diff --git a/apps/desktop/src/dev/durable-spike/backend-relay.test.ts b/apps/desktop/src/dev/durable-spike/backend-relay.test.ts new file mode 100644 index 00000000..0db1e2a7 --- /dev/null +++ b/apps/desktop/src/dev/durable-spike/backend-relay.test.ts @@ -0,0 +1,49 @@ +import type { BackendRpcEvent } from "@pace/backend"; +import { describe, expect, it, vi } from "vitest"; +import { backendRelay } from "./backend-relay"; + +function envelope(type: string, payload: Record): BackendRpcEvent { + return { type: "event", event: { id: "evt-1", seq: 0, sessionId: "", piSessionId: "", type, ts: "2026-10-02T00:00:00Z", payload } }; +} + +describe("backendRelay", () => { + it("maps relay calls to backend commands", async () => { + const invoke = vi.fn(async () => null) as unknown as (command: string, args?: Record) => Promise; + const relay = backendRelay({ invoke, onBackendEvent: () => () => {} }); + + await relay.connect("c1"); + relay.send("c1", "frame"); + relay.disconnect("c1"); + + expect(vi.mocked(invoke).mock.calls).toEqual([ + ["durable_spike_connect", { connectionId: "c1" }], + ["durable_spike_send", { connectionId: "c1", data: "frame" }], + ["durable_spike_disconnect", { connectionId: "c1" }], + ]); + }); + + it("passes on frames and closes from backend events and drops everything else", () => { + let emit: (event: BackendRpcEvent) => void = () => {}; + const relay = backendRelay({ + invoke: vi.fn() as never, + onBackendEvent: (listener) => { + emit = listener; + return () => {}; + }, + }); + const received: unknown[] = []; + relay.subscribe((event) => received.push(event)); + + emit(envelope("durable_spike.frame", { connectionId: "c1", data: "{}" })); + emit(envelope("terminal_output", { terminalId: "t", data: "x" })); + emit(envelope("durable_spike.frame", { data: "no connection id" })); + emit(envelope("durable_spike.closed", { connectionId: "c1", code: 4401, reason: "unauthorized" })); + emit(envelope("durable_spike.closed", { connectionId: "c2", code: 1006 })); + + expect(received).toEqual([ + { kind: "frame", connectionId: "c1", data: "{}" }, + { kind: "closed", connectionId: "c1", code: 4401, reason: "unauthorized" }, + { kind: "closed", connectionId: "c2", code: 1006 }, + ]); + }); +}); diff --git a/apps/desktop/src/dev/durable-spike/backend-relay.ts b/apps/desktop/src/dev/durable-spike/backend-relay.ts new file mode 100644 index 00000000..65c179c8 --- /dev/null +++ b/apps/desktop/src/dev/durable-spike/backend-relay.ts @@ -0,0 +1,40 @@ +import type { BackendRpcEvent } from "@pace/backend"; +import type { FrameRelay, RelayEvent } from "@pace/durable-spike/protocol/transport.ts"; + +type Invoke = (command: string, args?: Record) => Promise; +type SubscribeBackendEvent = (listener: (event: BackendRpcEvent) => void) => () => void; + +/** + * The spike's relay over Pace's backend: the renderer cannot open a WebSocket + * under its CSP, so the backend holds it and frames cross the existing IPC. + */ +export function backendRelay({ invoke, onBackendEvent }: { invoke: Invoke; onBackendEvent: SubscribeBackendEvent }): FrameRelay { + return { + connect: (connectionId) => invoke("durable_spike_connect", { connectionId }).then(() => {}), + send: (connectionId, data) => void invoke("durable_spike_send", { connectionId, data }), + disconnect: (connectionId) => void invoke("durable_spike_disconnect", { connectionId }), + subscribe: (listener) => + onBackendEvent((event) => { + const relayed = relayEventOf(event); + if (relayed !== undefined) listener(relayed); + }), + }; +} + +function relayEventOf(event: BackendRpcEvent): RelayEvent | undefined { + if (event.type !== "event") return undefined; + const { type, payload } = event.event; + if (typeof payload.connectionId !== "string") return undefined; + if (type === "durable_spike.frame" && typeof payload.data === "string") { + return { kind: "frame", connectionId: payload.connectionId, data: payload.data }; + } + if (type === "durable_spike.closed" && typeof payload.code === "number") { + return { + kind: "closed", + connectionId: payload.connectionId, + code: payload.code, + ...(typeof payload.reason === "string" ? { reason: payload.reason } : {}), + }; + } + return undefined; +} diff --git a/apps/desktop/src/dev/durable-spike/durable-spike-page.tsx b/apps/desktop/src/dev/durable-spike/durable-spike-page.tsx new file mode 100644 index 00000000..09c94c2b --- /dev/null +++ b/apps/desktop/src/dev/durable-spike/durable-spike-page.tsx @@ -0,0 +1,16 @@ +import { relayTransport } from "@pace/durable-spike/protocol/transport.ts"; +import { RemoteWorkbench } from "@pace/durable-spike/web/App.tsx"; +import { invoke, onBackendEvent } from "@/shared/runtime"; +import { backendRelay } from "./backend-relay"; + +const label = "Pace backend → ws://127.0.0.1:7420"; +const transport = relayTransport(backendRelay({ invoke, onBackendEvent }), label); + +/** + * Dev-only third client of the Durable multiview spike (spikes/durable-multiview): + * the web client's workbench, reached through Pace's backend instead of a + * browser WebSocket. Start the host first (`bun run host` in the spike). + */ +export function DurableSpikePage() { + return ; +} diff --git a/apps/desktop/src/entities/runtime/runtime-gateway-client.test.ts b/apps/desktop/src/entities/runtime/runtime-gateway-client.test.ts index 168d5dd4..a4d48af7 100644 --- a/apps/desktop/src/entities/runtime/runtime-gateway-client.test.ts +++ b/apps/desktop/src/entities/runtime/runtime-gateway-client.test.ts @@ -1141,7 +1141,7 @@ describe("Runtime Gateway client", () => { ]); }); - it.each(["workspace.invalidated", "model_catalog.invalidated", "terminal_output", "terminal_exit"])("ignores ephemeral %s before runtime state and deduplication", async (type) => { + it.each(["workspace.invalidated", "model_catalog.invalidated", "terminal_output", "terminal_exit", "durable_spike.frame", "durable_spike.closed"])("ignores ephemeral %s before runtime state and deduplication", async (type) => { let receive: ((event: BackendRpcEvent) => void) | undefined; const snapshot: RuntimeGatewaySnapshot = { sessionId: "session-1", runtimeId: "pi-sdk:session-1", piSessionId: "pi-session-1", projectId: "pig", cwd: "/repo", status: "idle", events: [], updatedAt: "2026-09-05T00:00:00.000Z" }; const client = createRuntimeGatewayClient({ diff --git a/apps/desktop/src/entities/runtime/runtime-gateway-client.ts b/apps/desktop/src/entities/runtime/runtime-gateway-client.ts index dd0ceeae..212d264c 100644 --- a/apps/desktop/src/entities/runtime/runtime-gateway-client.ts +++ b/apps/desktop/src/entities/runtime/runtime-gateway-client.ts @@ -654,8 +654,15 @@ export function createRuntimeGatewayClient( return; } - // Ephemeral workspace and terminal signals never belong to runtime truth. - if (["workspace.invalidated", "model_catalog.invalidated", "terminal_output", "terminal_exit"].includes(event.event.type)) { + // Ephemeral workspace, terminal, and dev-spike signals never belong to runtime truth. + if ([ + "workspace.invalidated", + "model_catalog.invalidated", + "terminal_output", + "terminal_exit", + "durable_spike.frame", + "durable_spike.closed", + ].includes(event.event.type)) { return; } diff --git a/bun.lock b/bun.lock index 7887d7b9..d943c738 100644 --- a/bun.lock +++ b/bun.lock @@ -116,6 +116,7 @@ "@types/react-dom": "^19.2.3", "@types/ws": "8.18.2", "@vitejs/plugin-react": "^4.3.1", + "@xterm/headless": "^6.0.0", "vite": "^5.4.2", }, }, @@ -1089,6 +1090,8 @@ "@xterm/addon-fit": ["@xterm/addon-fit@0.11.0", "", {}, "sha512-jYcgT6xtVYhnhgxh3QgYDnnNMYTcf8ElbxxFzX0IZo+vabQqSPAjC3c1wJrKB5E19VwQei89QCiZZP86DCPF7g=="], + "@xterm/headless": ["@xterm/headless@6.0.0", "", {}, "sha512-5Yj1QINYCyzrZtf8OFIHi47iQtI+0qYFPHmouEfG8dHNxbZ9Tb9YGSuLcsEwj9Z+OL75GJqPyJbyoFer80a2Hw=="], + "@xterm/xterm": ["@xterm/xterm@6.0.0", "", {}, "sha512-TQwDdQGtwwDt+2cgKDLn0IRaSxYu1tSUjgKarSDkUM0ZNiSRXFpjxEsvc/Zgc5kq5omJ+V0a8/kIM2WD3sMOYg=="], "abbrev": ["abbrev@4.0.0", "", {}, "sha512-a1wflyaL0tHtJSmLSOVybYhy22vRih4eduhhrkcjgrWGnRfrZtovJ2FRjxuTtkkj47O/baf0R86QU5OuYpz8fA=="], diff --git a/packages/backend/src/service.test.ts b/packages/backend/src/service.test.ts index b1caa753..1b9374eb 100644 --- a/packages/backend/src/service.test.ts +++ b/packages/backend/src/service.test.ts @@ -12,6 +12,7 @@ import { import { createInMemorySessionProjectionStore } from "./persistence/session-projection-store"; import type { SessionSummary } from "@pace/core"; import type { PiRuntimeDriver } from "./gateway/runtime-gateway"; +import type { DurableSpikeBridge, DurableSpikeBridgeEvent } from "./spikes/durable-bridge"; import type { TerminalManager, TerminalManagerEvent, @@ -2243,3 +2244,47 @@ it("delivers external Git changes to linked checkout siblings without a runtime await rm(root, { recursive: true, force: true }); } }); + +it("relays the Durable spike's frames between the renderer and the bridge as unsequenced events", async () => { + const bridgeListeners = new Set<(event: DurableSpikeBridgeEvent) => void>(); + const bridge = { + connect: vi.fn(async () => {}), + send: vi.fn(), + disconnect: vi.fn(), + onEvent: vi.fn((listener: (event: DurableSpikeBridgeEvent) => void) => { + bridgeListeners.add(listener); + return () => bridgeListeners.delete(listener); + }), + dispose: vi.fn(), + } satisfies DurableSpikeBridge; + const service = createBackendService({ + agentDir: fixtureAgentDir(), + sessionProjectionStore: createInMemorySessionProjectionStore(), + runtimeJournal: createInMemorySessionEventJournal(), + runtimeDriver: { onEvent: () => () => {} } as unknown as PiRuntimeDriver, + durableSpikeBridge: bridge, + }); + const events: import("./service").BackendRpcEvent[] = []; + service.onEvent((event) => events.push(event)); + + const connect = { connectionId: "c1", url: "ws://127.0.0.1:7420" }; + await expect(service.handleRequest({ id: "r1", method: "durable_spike_connect", params: connect })).resolves.toEqual({ id: "r1", result: null }); + await service.handleRequest({ id: "r2", method: "durable_spike_send", params: { connectionId: "c1", data: "{\"type\":\"subscribe\"}" } }); + await service.handleRequest({ id: "r3", method: "durable_spike_disconnect", params: { connectionId: "c1" } }); + expect(bridge.connect).toHaveBeenCalledWith(connect); + expect(bridge.send).toHaveBeenCalledWith({ connectionId: "c1", data: "{\"type\":\"subscribe\"}" }); + expect(bridge.disconnect).toHaveBeenCalledWith({ connectionId: "c1" }); + + for (const listener of bridgeListeners) { + listener({ kind: "frame", connectionId: "c1", data: "{\"type\":\"hello\"}" }); + listener({ kind: "closed", connectionId: "c1", code: 4401, reason: "unauthorized" }); + } + const envelope = { id: expect.stringMatching(/^evt-/), seq: 0, sessionId: "", piSessionId: "", ts: expect.any(String) }; + expect(events).toEqual([ + { type: "event", event: { ...envelope, type: "durable_spike.frame", payload: { connectionId: "c1", data: "{\"type\":\"hello\"}" } } }, + { type: "event", event: { ...envelope, type: "durable_spike.closed", payload: { connectionId: "c1", code: 4401, reason: "unauthorized" } } }, + ]); + + await service.dispose(); + expect(bridge.dispose).toHaveBeenCalled(); +}); diff --git a/packages/backend/src/service.ts b/packages/backend/src/service.ts index 6ee91837..2de8481a 100644 --- a/packages/backend/src/service.ts +++ b/packages/backend/src/service.ts @@ -102,6 +102,7 @@ import { type SessionIndexCache, } from "./workspace/sessions"; import { resolveChatWorkspaceRoot } from "./workspace/chat-workspace"; +import { createDurableSpikeBridge, type DurableSpikeBridge } from "./spikes/durable-bridge"; export type BackendRpcRequest = { id: string; @@ -149,6 +150,7 @@ export type BackendServiceOptions = { mcpServers?: McpServersService; modelCatalog?: ModelCatalog; terminalManager?: TerminalManager; + durableSpikeBridge?: DurableSpikeBridge; /** Fetch the signed-in accounts' model lists once at startup (the app sets this; tests do not). */ refreshAccountModelsOnStart?: boolean; }; @@ -344,6 +346,31 @@ export function createBackendService(options: BackendServiceOptions = {}): Backe } }); + // Dev-only Durable spike relay; like terminal streams, neither journaled nor sequenced. + // Inert until the dev page connects. Deleted with spikes/durable-multiview. + const durableSpikeBridge = + options.durableSpikeBridge ?? + createDurableSpikeBridge({ agentDir, ...(process.env.PACE_DURABLE_SPIKE_URL ? { defaultUrl: process.env.PACE_DURABLE_SPIKE_URL } : {}) }); + durableSpikeBridge.onEvent((event) => { + for (const listener of listeners) { + listener({ + type: "event", + event: { + id: `evt-${crypto.randomUUID()}`, + seq: 0, + sessionId: "", + piSessionId: "", + type: event.kind === "frame" ? "durable_spike.frame" : "durable_spike.closed", + ts: new Date().toISOString(), + payload: + event.kind === "frame" + ? { connectionId: event.connectionId, data: event.data } + : { connectionId: event.connectionId, code: event.code, ...(event.reason ? { reason: event.reason } : {}) }, + }, + }); + } + }); + let disposal: Promise | undefined; let closing = false; return { @@ -356,6 +383,7 @@ export function createBackendService(options: BackendServiceOptions = {}): Backe await runtimeGateway.flush(); await runtimeJournal.flush?.(); terminalManager.disposeAll(); + durableSpikeBridge.dispose(); gitWatchers.dispose(); unsubscribeModelCatalog(); invalidation.dispose(); @@ -392,6 +420,7 @@ export function createBackendService(options: BackendServiceOptions = {}): Backe terminalManager, invalidation, dataDir, + durableSpikeBridge, }), }; } catch (error) { @@ -434,6 +463,7 @@ async function dispatchRequest(input: { terminalManager: TerminalManager; invalidation: ReturnType; dataDir: string; + durableSpikeBridge: DurableSpikeBridge; }) { const params = paramsRecord(input.request.params); @@ -695,6 +725,24 @@ async function dispatchRequest(input: { case "close_terminal": input.terminalManager.close(requiredString(params.terminalId, "terminalId")); return null; + case "durable_spike_connect": { + const url = optionalString(params.url); + const token = optionalString(params.token); + await input.durableSpikeBridge.connect({ + connectionId: requiredString(params.connectionId, "connectionId"), + ...(url ? { url } : {}), + ...(token ? { token } : {}), + }); + return null; + } + case "durable_spike_send": { + if (typeof params.data !== "string") throw new Error("data is required"); + input.durableSpikeBridge.send({ connectionId: requiredString(params.connectionId, "connectionId"), data: params.data }); + return null; + } + case "durable_spike_disconnect": + input.durableSpikeBridge.disconnect({ connectionId: requiredString(params.connectionId, "connectionId") }); + return null; default: throw new Error(`Unknown backend RPC method "${input.request.method}".`); } diff --git a/packages/backend/src/spikes/durable-bridge.ts b/packages/backend/src/spikes/durable-bridge.ts new file mode 100644 index 00000000..d4fa570a --- /dev/null +++ b/packages/backend/src/spikes/durable-bridge.ts @@ -0,0 +1,88 @@ +import { readFile } from "node:fs/promises"; +import { join } from "node:path"; + +/** + * Dev-only relay for the Durable multiview spike (spikes/durable-multiview): + * holds the WebSocket to a local Durable host for the renderer, which cannot + * open one under its CSP. It moves frames and never parses them, so the + * renderer stays the single writer of what it shows (ADR-0044) and the hop + * reuses process IPC (ADR-0042). Deleted with the spike. + */ +export type DurableSpikeBridgeEvent = + | { kind: "frame"; connectionId: string; data: string } + | { kind: "closed"; connectionId: string; code: number; reason?: string }; + +export type DurableSpikeBridge = { + /** Starts a connection; frames and its close arrive as events. Rejects only when it cannot try. */ + connect(input: { connectionId: string; url?: string; token?: string }): Promise; + send(input: { connectionId: string; data: string }): void; + disconnect(input: { connectionId: string }): void; + onEvent(listener: (event: DurableSpikeBridgeEvent) => void): () => void; + dispose(): void; +}; + +export const DURABLE_SPIKE_DEFAULT_URL = "ws://127.0.0.1:7420"; + +const LOOPBACK_HOSTS = new Set(["127.0.0.1", "localhost", "[::1]"]); + +/** Where the spike host keeps its token: `~/.pi/agent/experimental/durable-multiview/default/token`. */ +export function durableSpikeTokenPath(agentDir: string): string { + return join(agentDir, "experimental", "durable-multiview", "default", "token"); +} + +export function createDurableSpikeBridge(options: { + agentDir: string; + /** Host used when a connect names none; a verification run points this away from the operator's host. */ + defaultUrl?: string; +}): DurableSpikeBridge { + const sockets = new Map(); + const listeners = new Set<(event: DurableSpikeBridgeEvent) => void>(); + const emit = (event: DurableSpikeBridgeEvent) => { + for (const listener of listeners) listener(event); + }; + + async function tokenFor(given: string | undefined): Promise { + if (given) return given; + const path = durableSpikeTokenPath(options.agentDir); + try { + return (await readFile(path, "utf8")).trim(); + } catch { + throw new Error(`No host token at ${path}. Start the host first: bun run host (in spikes/durable-multiview).`); + } + } + + return { + async connect({ connectionId, url, token }) { + const address = new URL(url ?? options.defaultUrl ?? DURABLE_SPIKE_DEFAULT_URL); + // The token grants full control of the host's agent; never send it off this machine. + if (!LOOPBACK_HOSTS.has(address.hostname)) { + throw new Error(`Durable host must be on a loopback address, not ${address.hostname}.`); + } + address.searchParams.set("token", await tokenFor(token)); + const socket = new WebSocket(address); + sockets.set(connectionId, socket); + socket.onmessage = (event) => emit({ kind: "frame", connectionId, data: String(event.data) }); + socket.onclose = (event) => { + if (sockets.get(connectionId) === socket) sockets.delete(connectionId); + emit({ kind: "closed", connectionId, code: event.code, ...(event.reason ? { reason: event.reason } : {}) }); + }; + }, + send({ connectionId, data }) { + const socket = sockets.get(connectionId); + if (socket?.readyState === WebSocket.OPEN) socket.send(data); + }, + disconnect({ connectionId }) { + sockets.get(connectionId)?.close(); + sockets.delete(connectionId); + }, + onEvent(listener) { + listeners.add(listener); + return () => listeners.delete(listener); + }, + dispose() { + for (const socket of sockets.values()) socket.close(); + sockets.clear(); + listeners.clear(); + }, + }; +} diff --git a/spikes/durable-multiview/package.json b/spikes/durable-multiview/package.json index 196efae0..1d404f50 100644 --- a/spikes/durable-multiview/package.json +++ b/spikes/durable-multiview/package.json @@ -10,7 +10,8 @@ "test": "vitest run --config vitest.config.ts", "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": "vite --config web/vite.config.ts", + "verify:three": "node scripts/three-clients.mjs" }, "dependencies": { "@astryxdesign/core": "^0.3.0", @@ -35,6 +36,7 @@ "@types/react-dom": "^19.2.3", "@types/ws": "8.18.2", "@vitejs/plugin-react": "^4.3.1", + "@xterm/headless": "^6.0.0", "vite": "^5.4.2" } } diff --git a/spikes/durable-multiview/protocol/remote-durable.ts b/spikes/durable-multiview/protocol/remote-durable.ts index b9f02e86..096519b6 100644 --- a/spikes/durable-multiview/protocol/remote-durable.ts +++ b/spikes/durable-multiview/protocol/remote-durable.ts @@ -9,6 +9,7 @@ import { type StreamName, UNAUTHORIZED_CLOSE_CODE, } from "./frames.ts"; +import { type FrameConnection, type FrameTransport, webSocketTransport } from "./transport.ts"; import type { ConversationSummary, DurableController, @@ -17,12 +18,14 @@ import type { Notice, } from "./view.ts"; -export interface RemoteDurableOptions { - /** `ws://127.0.0.1:` of a host's gateway. */ - readonly url: string; - readonly token: string; - readonly reconnectDelayMs?: { readonly min: number; readonly max: number }; -} +export type RemoteDurableOptions = ( + | { + /** `ws://127.0.0.1:` of a host's gateway. */ + readonly url: string; + readonly token: string; + } + | { readonly transport: FrameTransport } +) & { readonly reconnectDelayMs?: { readonly min: number; readonly max: number } }; export interface RemoteDurable { readonly view: DurableViewSource; @@ -49,7 +52,8 @@ class RemoteClient { readonly #wanted = new Set(["conversations"]); readonly #snapshotWaiters = new Map(); readonly #pending = new Map(); - #socket: WebSocket | undefined; + readonly #transport: FrameTransport; + #connection: FrameConnection | undefined; #state: DurableView | undefined; #current: ConversationId | undefined; #nextCall = 1; @@ -63,6 +67,7 @@ class RemoteClient { constructor(options: RemoteDurableOptions) { this.#options = options; + this.#transport = "transport" in options ? options.transport : webSocketTransport(options.url, options.token); this.ready = new Promise((resolve, reject) => { this.#resolveReady = resolve; this.#rejectReady = reject; @@ -71,22 +76,27 @@ class RemoteClient { } #open(): void { - const url = new URL(this.#options.url); - url.searchParams.set("token", this.#options.token); - const socket = new WebSocket(url); - this.#socket = socket; - socket.onmessage = (event) => this.#receive(JSON.parse(String(event.data)) as ServerFrame); - socket.onclose = (event) => this.#lost(socket, event.code); + const connection: FrameConnection = this.#transport.open({ + message: (data) => { + if (this.#connection === connection) this.#receive(JSON.parse(data) as ServerFrame); + }, + closed: (code, reason) => this.#lost(connection, code, reason), + }); + this.#connection = connection; } - #lost(socket: WebSocket, code: number): void { - if (this.#socket !== socket) return; - this.#socket = undefined; + #lost(connection: FrameConnection, code: number, reason: string | undefined): void { + if (this.#connection !== connection) return; + this.#connection = undefined; for (const pending of this.#pending.values()) pending.reject(new Error("Disconnected from host")); this.#pending.clear(); if (this.#state === undefined) { this.#closed = true; - this.#rejectReady(new Error(code === UNAUTHORIZED_CLOSE_CODE ? "Unauthorized: wrong host token" : `Could not connect to ${this.#options.url}`)); + this.#rejectReady( + new Error( + code === UNAUTHORIZED_CLOSE_CODE ? "Unauthorized: wrong host token" : (reason ?? `Could not connect to ${this.#transport.label}`), + ), + ); return; } if (this.#closed) return; @@ -105,7 +115,7 @@ class RemoteClient { } #send(frame: ClientFrame): void { - this.#socket?.send(JSON.stringify(frame)); + this.#connection?.send(JSON.stringify(frame)); } #receive(frame: ServerFrame): void { @@ -205,7 +215,7 @@ class RemoteClient { } #call(method: M, args: CallMethods[M]["args"]): Promise { - if (this.#socket === undefined || this.#state?.connection !== "connected") { + if (this.#connection === undefined || this.#state?.connection !== "connected") { return Promise.reject(new Error("Not connected to the host")); } const id = this.#nextCall++; @@ -288,6 +298,6 @@ class RemoteClient { if (this.#closed) return; this.#closed = true; this.#update({ connection: "closed" }); - this.#socket?.close(); + this.#connection?.close(); } } diff --git a/spikes/durable-multiview/protocol/transport.ts b/spikes/durable-multiview/protocol/transport.ts new file mode 100644 index 00000000..1dfe038c --- /dev/null +++ b/spikes/durable-multiview/protocol/transport.ts @@ -0,0 +1,81 @@ +/** One open connection to a host's gateway, carrying serialized frames. */ +export interface FrameConnection { + send(data: string): void; + close(): void; +} + +export interface FrameHandlers { + message(data: string): void; + /** Called once, also when the connection never opened; `reason` explains a failure to connect. */ + closed(code: number, reason?: string): void; +} + +/** How a client reaches a host: a WebSocket, or a relay such as Pace's backend over IPC. */ +export interface FrameTransport { + /** Shown in errors, for example the host URL. */ + readonly label: string; + open(handlers: FrameHandlers): FrameConnection; +} + +export function webSocketTransport(url: string, token: string): FrameTransport { + return { + label: url, + open(handlers) { + const address = new URL(url); + address.searchParams.set("token", token); + const socket = new WebSocket(address); + socket.onmessage = (event) => handlers.message(String(event.data)); + socket.onclose = (event) => handlers.closed(event.code); + return { send: (data) => socket.send(data), close: () => socket.close() }; + }, + }; +} + +export type RelayEvent = + | { readonly kind: "frame"; readonly connectionId: string; readonly data: string } + | { readonly kind: "closed"; readonly connectionId: string; readonly code: number; readonly reason?: string }; + +/** + * A relay that holds the WebSocket for the client, as Pace's backend does for + * its renderer. It only moves frames; the client still applies every op. + */ +export interface FrameRelay { + /** Rejects when the relay cannot even try, for example with no token; the reason reaches the user. */ + connect(connectionId: string): Promise; + send(connectionId: string, data: string): void; + disconnect(connectionId: string): void; + subscribe(listener: (event: RelayEvent) => void): () => void; +} + +export function relayTransport(relay: FrameRelay, label: string): FrameTransport { + return { + label, + open(handlers) { + const connectionId = crypto.randomUUID(); + let closed = false; + const close = (code: number, reason?: string): void => { + if (closed) return; + closed = true; + unsubscribe(); + handlers.closed(code, reason); + }; + // Subscribed before connecting, so no frame of this connection is missed. + const unsubscribe = relay.subscribe((event) => { + if (event.connectionId !== connectionId) return; + if (event.kind === "frame") handlers.message(event.data); + else close(event.code, event.reason); + }); + relay.connect(connectionId).catch((error: unknown) => close(1006, error instanceof Error ? error.message : String(error))); + return { + send: (data) => { + if (!closed) relay.send(connectionId, data); + }, + close: () => { + if (closed) return; + relay.disconnect(connectionId); + close(1000); + }, + }; + }, + }; +} diff --git a/spikes/durable-multiview/scripts/three-clients.mjs b/spikes/durable-multiview/scripts/three-clients.mjs new file mode 100644 index 00000000..0054b220 --- /dev/null +++ b/spikes/durable-multiview/scripts/three-clients.mjs @@ -0,0 +1,144 @@ +// P3 proof: one faux host and three clients (Pace's dev page through its backend relay, the web +// page, the TUI), each driving the shared conversation; then a kill -9 mid-stream and a restart. +// Isolated from an operator's host (7420), web server (5199) and Pace profile. Pace runs under +// `electron-vite dev` because its page is DEV-only and builds always drop the DEV branch. +// +// 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 { tmpdir } from "node:os"; +import { join, resolve } from "node:path"; +import { fileURLToPath } from "node:url"; +import { spawn as spawnPty } from "@lydell/node-pty"; +import { chromium } from "@playwright/test"; +import xterm from "@xterm/headless"; + +const spike = fileURLToPath(new URL("..", import.meta.url)); +const repo = resolve(spike, "../.."); +const T = mkdtempSync(join(tmpdir(), "dmv-three-")); +const hostDir = join(T, "agent/experimental/durable-multiview/default"); +mkdirSync(join(T, "data"), { recursive: true }); +writeFileSync(join(T, "data/preflight-status.json"), JSON.stringify({ completedAt: "2026-10-02T00:00:00.000Z" })); +const PORT = 7466; +const out = resolve(process.argv[2] ?? mkdtempSync(join(tmpdir(), "dmv-evidence-"))); +mkdirSync(out, { recursive: true }); +const sleep = (ms) => new Promise((r) => setTimeout(r, ms)); +const children = []; +// Drained so Electron never blocks on a full pipe; printed when a run fails. +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"] }); + 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 }; +} +async function until(what, predicate, timeout = 30000) { + const deadline = Date.now() + timeout; + while (!(await predicate())) { + if (Date.now() > deadline) throw new Error(`timed out waiting for ${what}`); + await sleep(200); + } + console.log("ok:", what); +} + +try { + let { host, ready } = await startHost(); + + // Pace in dev mode (the route is DEV-only), isolated like the e2e fixture. + const pace = spawn(join(repo, "node_modules/.bin/electron-vite"), ["dev", "--remoteDebuggingPort", "9339", "--", `--user-data-dir=${join(T, "profile")}`], { + cwd: join(repo, "apps/desktop"), + env: { ...process.env, PACE_DATA_DIR: join(T, "data"), PI_CODING_AGENT_DIR: join(T, "agent"), PACE_DURABLE_SPIKE_URL: ready.url, PACE_E2E: "1" }, + stdio: ["ignore", "pipe", "pipe"], + detached: true, + }); + children.push(pace); + pace.stdout.on("data", (d) => (paceLog += d)); + pace.stderr.on("data", (d) => (paceLog += d)); + let cdp; + await until("Pace CDP", async () => { try { cdp = await chromium.connectOverCDP("http://127.0.0.1:9339"); return true; } catch { return false; } }, 90000); + let pacePage; + await until("Pace window", async () => { pacePage = cdp.contexts().flatMap((c) => c.pages()).find((p) => !p.url().startsWith("devtools")); return pacePage !== undefined; }); + await pacePage.waitForLoadState("domcontentloaded"); + await pacePage.evaluate(() => { window.location.hash = "#/durable-spike"; }); + await until("Pace page connected", async () => (await pacePage.getByText("Durable host").count()) > 0, 30000); + + // The web page on its own port. + const web = spawn(process.execPath, [join(repo, "node_modules/vite/bin/vite.js"), "--config", "web/vite.config.ts", "--port", "5299"], { cwd: spike, stdio: ["ignore", "pipe", "inherit"] }); + children.push(web); + await new Promise((resolve) => web.stdout.on("data", (d) => String(d).includes("5299") && resolve())); + const browser = await chromium.launch(); + const webPage = await browser.newPage({ viewport: { width: 1280, height: 720 } }); + await webPage.goto(ready.web.replace("5199", "5299")); + + // The TUI in a pty, rendered by a headless terminal. + const cols = 110, rows = 24; + const term = new xterm.Terminal({ cols, rows, allowProposedApi: true }); + mkdirSync(join(T, "tui-agent"), { recursive: true }); + const tui = spawnPty(process.execPath, ["tui/main.ts", "--url", ready.url, "--token", ready.token], { cols, rows, cwd: spike, env: { ...process.env, PI_CODING_AGENT_DIR: join(T, "tui-agent"), TERM: "xterm-256color" } }); + children.push({ kill: () => tui.kill() }); + let tuiText = ""; + tui.onData((d) => { term.write(d); tuiText += d; }); + const screen = () => Array.from({ length: rows }, (_, i) => term.buffer.active.getLine(i)?.translateToString(true).trimEnd() ?? "").join("\n"); + const tuiShows = (text) => screen().includes(text); + await until("TUI footer", () => tuiShows("faux/faux-1")); + + const ask = async (page, text) => { await page.getByRole("textbox").fill(text); await page.getByRole("textbox").press("Enter"); }; + await ask(pacePage, "question-from-pace"); + await until("web shows Pace's question", async () => (await webPage.getByText("question-from-pace").count()) > 0); + await until("TUI shows Pace's question", () => tuiShows("question-from-pace")); + await until("Pace shows the answer", async () => (await pacePage.getByText("Fix: run the backfill in batches.").count()) > 0); + await ask(webPage, "question-from-web"); + await until("Pace shows web's question", async () => (await pacePage.getByText("question-from-web").count()) > 0); + await until("idle after web turn", async () => (await pacePage.getByText("Fix: run the backfill in batches.").count()) >= 2); + for (const ch of "question-from-tui") { tui.write(ch); await sleep(5); } + tui.write("\r"); + await until("Pace shows TUI's question", async () => (await pacePage.getByText("question-from-tui").count()) > 0); + await sleep(2500); + await pacePage.screenshot({ path: join(out, "pace-streaming.png") }); + await webPage.screenshot({ path: join(out, "web-streaming.png") }); + writeFileSync(join(out, "tui-streaming.txt"), screen()); + + // Crash mid-stream, restart, and watch Pace resume. + host.kill("SIGKILL"); + await until("Pace shows the lost host", async () => (await pacePage.getByText("Host connection lost").count()) > 0); + await pacePage.screenshot({ path: join(out, "pace-host-lost.png") }); + ({ host } = await startHost()); + await until("Pace reconnected", async () => (await pacePage.getByText("Host connection lost").count()) === 0, 30000); + await until("Pace shows the interrupted partial", async () => (await pacePage.getByText("interrupted").count()) > 0, 30000); + await until("third answer complete", async () => (await pacePage.getByText("Fix: run the backfill in batches.").count()) >= 3, 60000); + await pacePage.screenshot({ path: join(out, "pace-recovered.png") }); + await webPage.screenshot({ path: join(out, "web-recovered.png") }); + writeFileSync(join(out, "tui-recovered.txt"), screen()); + console.log(`DONE; evidence in ${out}`); + await browser.close(); +} catch (error) { + console.error("FAILED:", error.message); + console.error(paceLog.slice(-3000)); + process.exitCode = 1; +} finally { + for (const child of children) { + try { + if (child.pid) process.kill(-child.pid, "SIGTERM"); + else child.kill(); + } catch { + try { + child.kill(); + } catch {} + } + } + // electron-vite starts Electron outside its process group; end only the one on this run's profile. + const profile = join(T, "profile"); + for (const line of execFileSync("ps", ["-axo", "pid=,command="], { encoding: "utf8" }).split("\n")) { + const [pid, ...command] = line.trim().split(/\s+/); + if (pid && command.join(" ").includes(`--user-data-dir=${profile}`)) { + try { + process.kill(Number(pid), "SIGTERM"); + } catch {} + } + } + await sleep(1000); + rmSync(T, { recursive: true, force: true }); + process.exit(); +} diff --git a/spikes/durable-multiview/test/relay.test.ts b/spikes/durable-multiview/test/relay.test.ts new file mode 100644 index 00000000..2b908ba5 --- /dev/null +++ b/spikes/durable-multiview/test/relay.test.ts @@ -0,0 +1,90 @@ +import { mkdir, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import { describe, expect, it } from "vitest"; +// Pace's backend relay, the hop between a host and Pace's renderer. +import { createDurableSpikeBridge, type DurableSpikeBridge } from "../../../packages/backend/src/spikes/durable-bridge.ts"; +import { connectRemoteDurable, type RemoteDurable } from "../protocol/remote-durable.ts"; +import { transcript } from "../protocol/transcript.ts"; +import { type FrameRelay, relayTransport } from "../protocol/transport.ts"; +import { connectTo, freePort, startFauxHost, tempDir, useCleanups, waitForView } from "./support.ts"; + +const defer = useCleanups(); + +/** An agent dir holding a host token where the bridge looks for it. */ +async function agentDirWithToken(token: string): Promise { + const dir = await tempDir(); + defer(dir.remove); + const tokenDir = join(dir.path, "experimental", "durable-multiview", "default"); + await mkdir(tokenDir, { recursive: true }); + await writeFile(join(tokenDir, "token"), `${token}\n`); + return dir.path; +} + +function relayOf(bridge: DurableSpikeBridge, url: string): FrameRelay { + return { + connect: (connectionId) => bridge.connect({ connectionId, url }), + send: (connectionId, data) => bridge.send({ connectionId, data }), + disconnect: (connectionId) => bridge.disconnect({ connectionId }), + subscribe: (listener) => bridge.onEvent(listener), + }; +} + +async function connectThroughBridge(bridge: DurableSpikeBridge, url: string): Promise { + const client = await connectRemoteDurable({ transport: relayTransport(relayOf(bridge, url), "pace"), reconnectDelayMs: { min: 100, max: 500 } }); + defer(() => client.close()); + return client; +} + +describe("Pace backend relay", () => { + it("carries a client that sees and drives the same conversation as a direct one", async () => { + const host = await startFauxHost(defer, { answers: ["answer-through-the-relay"] }); + const bridge = createDurableSpikeBridge({ agentDir: await agentDirWithToken(host.token) }); + defer(() => bridge.dispose()); + const relayed = await connectThroughBridge(bridge, host.url); + const direct = await connectTo(defer, host); + + await relayed.controller.submit("question-through-the-relay", "followUp"); + const answered = (client: RemoteDurable) => + waitForView(client.view, (view) => transcript(view.conversation).at(-1)?.text === "answer-through-the-relay"); + await Promise.all([answered(relayed), answered(direct)]); + expect(relayed.view.current().conversation).toEqual(direct.view.current().conversation); + }); + + it("reconnects through the relay after the host restarts", async () => { + const port = await freePort(); + const first = await startFauxHost(defer, { port }); + const bridge = createDurableSpikeBridge({ agentDir: await agentDirWithToken(first.token) }); + defer(() => bridge.dispose()); + const relayed = await connectThroughBridge(bridge, first.url); + const dataDir = relayed.view.current().session.directory; + + await first.close(); + await waitForView(relayed.view, (view) => view.connection === "reconnecting"); + await startFauxHost(defer, { port, dataDir }); + await waitForView(relayed.view, (view) => view.connection === "connected"); + }); + + it("connects to its configured default host when the client names none", async () => { + const host = await startFauxHost(defer); + const bridge = createDurableSpikeBridge({ agentDir: await agentDirWithToken(host.token), defaultUrl: host.url }); + defer(() => bridge.dispose()); + const relay: FrameRelay = { ...relayOf(bridge, host.url), connect: (connectionId) => bridge.connect({ connectionId }) }; + const client = await connectRemoteDurable({ transport: relayTransport(relay, "pace") }); + defer(() => client.close()); + expect(client.view.current().connection).toBe("connected"); + }); + + it("says to start the host when there is no token yet", async () => { + const dir = await tempDir(); + defer(dir.remove); + const bridge = createDurableSpikeBridge({ agentDir: dir.path }); + defer(() => bridge.dispose()); + await expect(connectThroughBridge(bridge, "ws://127.0.0.1:7420")).rejects.toThrow(/no host token.*bun run host/i); + }); + + it("refuses a host that is not on this machine", async () => { + const bridge = createDurableSpikeBridge({ agentDir: await agentDirWithToken("t") }); + defer(() => bridge.dispose()); + await expect(connectThroughBridge(bridge, "ws://example.com:7420")).rejects.toThrow(/loopback/i); + }); +}); diff --git a/spikes/durable-multiview/test/support.ts b/spikes/durable-multiview/test/support.ts index 374436fe..9fd2aa23 100644 --- a/spikes/durable-multiview/test/support.ts +++ b/spikes/durable-multiview/test/support.ts @@ -25,7 +25,15 @@ export async function startFauxHost( answers = [LONG_ANSWER], tokensPerSecond = 200, dataDir, - }: { answers?: string[]; tokensPerSecond?: number; /** Reopen an earlier host's store. */ dataDir?: string } = {}, + port = 0, + }: { + answers?: string[]; + tokensPerSecond?: number; + /** Reopen an earlier host's store. */ + dataDir?: string; + /** Fixed, so clients reconnect to a restarted host. */ + port?: number; + } = {}, ): Promise { const dir = dataDir === undefined ? await tempDir() : { path: dataDir, remove: () => {} }; defer(dir.remove); @@ -40,7 +48,7 @@ export async function startFauxHost( models, modelSummaries: () => [{ provider: model.provider, modelId: model.id, name: model.name, contextWindow: model.contextWindow }], initialModel: { provider: model.provider, modelId: model.id }, - port: 0, + port, }); defer(() => host.close()); return host; diff --git a/spikes/durable-multiview/web/App.tsx b/spikes/durable-multiview/web/App.tsx index 9a1d685f..36bf2467 100644 --- a/spikes/durable-multiview/web/App.tsx +++ b/spikes/durable-multiview/web/App.tsx @@ -21,7 +21,7 @@ import { Token } from "@astryxdesign/core/Token"; import type { AgentState } from "@earendil-works/pi-durable"; import { type CSSProperties, type ReactNode, useMemo, useState } from "react"; import { type ChatItem, chatItems, queueItems, statusText, taskRows } from "../presentation/chat.ts"; -import type { RemoteDurable } from "../protocol/remote-durable.ts"; +import type { RemoteDurable, RemoteDurableOptions } from "../protocol/remote-durable.ts"; import { isBusy } from "../protocol/transcript.ts"; import type { DurableView } from "../protocol/view.ts"; import { addressFromHash } from "./address.ts"; @@ -42,7 +42,7 @@ export function App() { return ; } -function Centered({ children }: { children: ReactNode }) { +export function Centered({ children }: { children: ReactNode }) { return ( {children} @@ -51,11 +51,16 @@ function Centered({ children }: { children: ReactNode }) { } function Connected({ address }: { address: { url: string; token: string } }) { - const state = useRemoteDurable(address); + return ; +} + +/** Connects with `options` and shows the workbench, or why it cannot; shared by this page and Pace's dev page. */ +export function RemoteWorkbench({ options, connectionKey, label }: { options: RemoteDurableOptions; connectionKey: string; label: string }) { + const state = useRemoteDurable(options, connectionKey); if (state.status === "connecting") { return ( - + ); } diff --git a/spikes/durable-multiview/web/use-remote.ts b/spikes/durable-multiview/web/use-remote.ts index 6f21a87f..5e770ef3 100644 --- a/spikes/durable-multiview/web/use-remote.ts +++ b/spikes/durable-multiview/web/use-remote.ts @@ -1,5 +1,5 @@ import { useEffect, useState, useSyncExternalStore } from "react"; -import { connectRemoteDurable, type RemoteDurable } from "../protocol/remote-durable.ts"; +import { connectRemoteDurable, type RemoteDurable, type RemoteDurableOptions } from "../protocol/remote-durable.ts"; import type { DurableView } from "../protocol/view.ts"; export type RemoteState = @@ -7,13 +7,16 @@ export type RemoteState = | { readonly status: "failed"; readonly error: string } | { readonly status: "ready"; readonly remote: RemoteDurable }; -/** One connection per address for the component's lifetime; later losses are the view's `connection`, not this state. */ -export function useRemoteDurable(address: { url: string; token: string }): RemoteState { +/** + * One connection per `key` for the component's lifetime; later losses are the + * view's `connection`, not this state. `options` is read when `key` changes. + */ +export function useRemoteDurable(options: RemoteDurableOptions, key: string): RemoteState { const [state, setState] = useState({ status: "connecting" }); useEffect(() => { let opened: RemoteDurable | undefined; let cancelled = false; - connectRemoteDurable({ ...address, reconnectDelayMs: { min: 200, max: 2000 } }).then( + connectRemoteDurable({ reconnectDelayMs: { min: 200, max: 2000 }, ...options }).then( (remote) => { if (cancelled) { remote.close(); @@ -33,7 +36,7 @@ export function useRemoteDurable(address: { url: string; token: string }): Remot cancelled = true; opened?.close(); }; - }, [address.url, address.token]); + }, [key]); return state; } diff --git a/tsconfig.json b/tsconfig.json index ac2a8873..a94a6511 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -14,6 +14,7 @@ "resolveJsonModule": true, "isolatedModules": true, "noEmit": true, + "allowImportingTsExtensions": true, "jsx": "react-jsx", "baseUrl": ".", "paths": { @@ -22,7 +23,8 @@ "@pace/pi-mcp/*": [ "packages/backend/node_modules/@earendil-works/pi-coding-agent/dist/extensions/mcp/*" ], - "@/*": ["apps/desktop/src/*"] + "@/*": ["apps/desktop/src/*"], + "@pace/durable-spike/*": ["spikes/durable-multiview/*"] } }, "include": ["apps", "packages"], diff --git a/vite.config.ts b/vite.config.ts index 28fb7040..3d436e4c 100644 --- a/vite.config.ts +++ b/vite.config.ts @@ -17,6 +17,8 @@ export default defineConfig({ "packages/backend/node_modules/@earendil-works/pi-coding-agent/dist/extensions/mcp", ), "@": resolve(__dirname, "apps/desktop/src"), + // Dev-only Durable multiview spike page; removed with spikes/durable-multiview. + "@pace/durable-spike": resolve(__dirname, "spikes/durable-multiview"), }, }, server: {