From c45ec2ca64aadc446c7b508a4708581fe38af213 Mon Sep 17 00:00:00 2001 From: Kieran Zhang Date: Fri, 2 Oct 2026 23:27:38 +0800 Subject: [PATCH] fix(spike): recover backend connections and preserve typed steering --- .../dev/durable-spike/backend-relay.test.ts | 54 +++++++++++++++ .../src/dev/durable-spike/backend-relay.ts | 66 +++++++++++++++--- .../durable-multiview-review-fixes.md | 18 ++++- packages/backend/src/spikes/durable-bridge.ts | 3 +- .../scripts/demo-scenario.mjs | 5 +- spikes/durable-multiview/test/relay.test.ts | 68 +++++++++++++++++++ spikes/durable-multiview/test/web-ui.test.ts | 29 +++++++- spikes/durable-multiview/web/App.tsx | 2 +- 8 files changed, 229 insertions(+), 16 deletions(-) diff --git a/apps/desktop/src/dev/durable-spike/backend-relay.test.ts b/apps/desktop/src/dev/durable-spike/backend-relay.test.ts index 0db1e2a7..15b06308 100644 --- a/apps/desktop/src/dev/durable-spike/backend-relay.test.ts +++ b/apps/desktop/src/dev/durable-spike/backend-relay.test.ts @@ -1,4 +1,5 @@ import type { BackendRpcEvent } from "@pace/backend"; +import { relayTransport } from "@pace/durable-spike/protocol/transport.ts"; import { describe, expect, it, vi } from "vitest"; import { backendRelay } from "./backend-relay"; @@ -7,6 +8,44 @@ function envelope(type: string, payload: Record): BackendRpcEve } describe("backendRelay", () => { + it("closes the old transport on backend exit and carries frames on a new connection", async () => { + const listeners = new Set<(event: BackendRpcEvent) => void>(); + const emit = (event: BackendRpcEvent) => { + for (const listener of listeners) listener(event); + }; + const invoke = vi.fn(async () => null) as unknown as (command: string, args?: Record) => Promise; + const transport = relayTransport(backendRelay({ + invoke, + onBackendEvent: (listener) => { + listeners.add(listener); + return () => { listeners.delete(listener); }; + }, + }), "pace"); + const closed = vi.fn(); + const first = transport.open({ message: vi.fn(), closed }); + const firstId = vi.mocked(invoke).mock.calls[0]![1]!.connectionId; + + const disconnected = envelope("error", { lifecycle: "disconnected", generation: 1 }); + disconnected.event.sessionId = "__backend__"; + emit(disconnected); + expect(closed).toHaveBeenCalledOnce(); + expect(closed).toHaveBeenCalledWith(1006, expect.stringMatching(/backend/i)); + first.send("must-not-reach-the-old-connection"); + expect(vi.mocked(invoke).mock.calls.filter(([command]) => command === "durable_spike_send")).toEqual([]); + + const message = vi.fn(); + const second = transport.open({ message, closed: vi.fn() }); + const calls = vi.mocked(invoke).mock.calls; + const secondId = calls[calls.length - 1]![1]!.connectionId; + expect(secondId).not.toBe(firstId); + second.send("new-command"); + emit(envelope("durable_spike.frame", { connectionId: firstId, data: "stale" })); + emit(envelope("durable_spike.frame", { connectionId: secondId, data: "new-reply" })); + expect(invoke).toHaveBeenCalledWith("durable_spike_send", { connectionId: secondId, data: "new-command" }); + expect(message).toHaveBeenCalledExactlyOnceWith("new-reply"); + second.close(); + }); + 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: () => () => {} }); @@ -22,6 +61,21 @@ describe("backendRelay", () => { ]); }); + it("closes the transport when a frame cannot reach the backend", async () => { + const invoke = (async (command: string) => { + if (command === "durable_spike_send") throw new Error("Backend is not connected"); + return null; + }) as (command: string, args?: Record) => Promise; + const transport = relayTransport(backendRelay({ invoke, onBackendEvent: () => () => {} }), "pace"); + const closed = vi.fn(); + const connection = transport.open({ message: vi.fn(), closed }); + connection.send("pending-command"); + await Promise.resolve(); + + expect(closed).toHaveBeenCalledExactlyOnceWith(1006, "Backend is not connected"); + connection.close(); + }); + it("passes on frames and closes from backend events and drops everything else", () => { let emit: (event: BackendRpcEvent) => void = () => {}; const relay = backendRelay({ diff --git a/apps/desktop/src/dev/durable-spike/backend-relay.ts b/apps/desktop/src/dev/durable-spike/backend-relay.ts index 65c179c8..175893ba 100644 --- a/apps/desktop/src/dev/durable-spike/backend-relay.ts +++ b/apps/desktop/src/dev/durable-spike/backend-relay.ts @@ -1,27 +1,71 @@ -import type { BackendRpcEvent } from "@pace/backend"; -import type { FrameRelay, RelayEvent } from "@pace/durable-spike/protocol/transport.ts"; +import type { FrameRelay, RelayEvent } from "../../../../../spikes/durable-multiview/protocol/transport.ts"; type Invoke = (command: string, args?: Record) => Promise; -type SubscribeBackendEvent = (listener: (event: BackendRpcEvent) => void) => () => void; +type BackendEvent = { + type: "event"; + event: { sessionId: string; type: string; payload: Record }; +}; +type SubscribeBackendEvent = (listener: (event: BackendEvent) => 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 { + const connections = new Set(); + const listeners = new Set<(event: RelayEvent) => void>(); + let unsubscribe: (() => void) | undefined; + const emit = (event: RelayEvent) => { + for (const listener of listeners) listener(event); + }; 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) => { + connect: async (connectionId) => { + connections.add(connectionId); + try { + await invoke("durable_spike_connect", { connectionId }); + } catch (error) { + connections.delete(connectionId); + throw error; + } + }, + send: (connectionId, data) => { + void invoke("durable_spike_send", { connectionId, data }).catch((error: unknown) => { + if (!connections.delete(connectionId)) return; + emit({ kind: "closed", connectionId, code: 1006, reason: error instanceof Error ? error.message : String(error) }); + }); + }, + disconnect: (connectionId) => { + connections.delete(connectionId); + // The transport is already closed locally; a dead backend has nothing left to close. + void invoke("durable_spike_disconnect", { connectionId }).catch(() => {}); + }, + subscribe: (listener) => { + listeners.add(listener); + unsubscribe ??= onBackendEvent((event) => { + if (event.event.sessionId === "__backend__" && event.event.payload.lifecycle === "disconnected") { + for (const connectionId of connections) { + connections.delete(connectionId); + emit({ kind: "closed", connectionId, code: 1006, reason: "Pace backend disconnected" }); + } + return; + } const relayed = relayEventOf(event); - if (relayed !== undefined) listener(relayed); - }), + if (relayed === undefined) return; + if (relayed.kind === "closed") connections.delete(relayed.connectionId); + emit(relayed); + }); + return () => { + listeners.delete(listener); + if (listeners.size === 0) { + unsubscribe?.(); + unsubscribe = undefined; + } + }; + }, }; } -function relayEventOf(event: BackendRpcEvent): RelayEvent | undefined { +function relayEventOf(event: BackendEvent): RelayEvent | undefined { if (event.type !== "event") return undefined; const { type, payload } = event.event; if (typeof payload.connectionId !== "string") return undefined; diff --git a/docs/research/durable-multiview-review-fixes.md b/docs/research/durable-multiview-review-fixes.md index 2fb08301..8f5c38a5 100644 --- a/docs/research/durable-multiview-review-fixes.md +++ b/docs/research/durable-multiview-review-fixes.md @@ -1,6 +1,6 @@ # Durable 多端 spike：高优先检视修复 -本轮针对已合并 spike 的审批归属、订阅竞态和消息边界问题补充回归保护。其余历史检视不在本轮范围内；本轮也不代表生产迁移或安全验收完成。 +本轮针对已合并 spike 的审批归属、订阅竞态、消息边界、backend 重连及 steer 按钮问题补充回归保护。其余历史检视不在本轮范围内；本轮也不代表生产迁移或安全验收完成。 ## 宿主与审批边界 @@ -32,3 +32,19 @@ bun run web:link - `test/host-startup.test.ts`：从进程输出确认凭证未泄露，再从本地 token 文件连接真实宿主。 原有宿主恢复、工具热替换及完整三端脚本已改为从本地 token 文件取凭证，不再从日志读取。 + +## 客户端恢复与发送 + +backend 的 `__backend__/disconnected` 生命周期事件会关闭现存的 relay transport，交给原有重连路径建立新连接。即使该事件丢失，向不存在或未打开的 socket 发送也会明确失败，renderer 将 IPC 拒绝转换为断线；提示不会再静默消失。未完成的调用会显示错误，结果未知的命令不会自动重放。 + +运行中的 composer 只有在输入为空时显示 Stop。有文本时显示 Send，点击后提交 steer，并保留正在等待审批的工具；发送后输入清空，Stop 恢复可用。 + +- `backend-relay.test.ts`：使用实际 backend 事件结构验证断线通知及 IPC 发送失败。 +- `test/relay.test.ts`：真实宿主与 backend bridge，在 backend 替换和生命周期事件丢失两种情况下重新连接，后续提示均得到模型回答。 +- `test/web-ui.test.ts`：真实浏览器点击 Send 后，steer 入队且工具仍待审批；输入清空后点击 Stop 终止任务。 + +## 本轮验证 + +2026-10-02：宿主层独立通过 spike typecheck 和 60 项测试；完整修复栈通过根 typecheck、lint、build，根测试 168 个文件 / 1977 项，spike typecheck 与测试 15 个文件 / 64 项。每个具体缺陷均先运行失败用例，再实现修复。 + +完整 `bun run verify:demo` 通过 12 幕，包含 SIGKILL 恢复、三端审批、子会话 steer、补偿、提醒、fork、三端压缩、热替换及窄屏布局。Pace 的 steer 步骤改为实际点击 Send；已检查 [运行中发送按钮截图](../../output/durable-multiview-boundaries-20261002/05-steer-ready-pace.png)。完整本机证据目录：[output/durable-multiview-boundaries-20261002](../../output/durable-multiview-boundaries-20261002)。 diff --git a/packages/backend/src/spikes/durable-bridge.ts b/packages/backend/src/spikes/durable-bridge.ts index d4fa570a..2d2d355e 100644 --- a/packages/backend/src/spikes/durable-bridge.ts +++ b/packages/backend/src/spikes/durable-bridge.ts @@ -69,7 +69,8 @@ export function createDurableSpikeBridge(options: { }, send({ connectionId, data }) { const socket = sockets.get(connectionId); - if (socket?.readyState === WebSocket.OPEN) socket.send(data); + if (socket?.readyState !== WebSocket.OPEN) throw new Error("Durable host connection is not open"); + socket.send(data); }, disconnect({ connectionId }) { sockets.get(connectionId)?.close(); diff --git a/spikes/durable-multiview/scripts/demo-scenario.mjs b/spikes/durable-multiview/scripts/demo-scenario.mjs index 004cca82..585a0ca0 100644 --- a/spikes/durable-multiview/scripts/demo-scenario.mjs +++ b/spikes/durable-multiview/scripts/demo-scenario.mjs @@ -178,7 +178,10 @@ try { await ask(p, "also look at ap-south"); const logsId = subagents().find((summary) => summary.title.startsWith("logs:")).id; await until("the logs subagent is running a tool", () => Object.values(seen().tasks?.tasks ?? {}).some((node) => node.kind === "pi.tool" && node.conversationId === logsId)); - await ask(p, "focus on the connection pool limit"); + await p.getByRole("textbox").first().fill("focus on the connection pool limit"); + await p.getByRole("button", { name: "Send", exact: true }).waitFor(); + await shoot("05-steer-ready", { pace: p }); + await p.getByRole("button", { name: "Send", exact: true }).click(); await until("Pace steers the running subagent", shows(p, "focus on the connection pool limit")); await until("the subagent answers Pace", async () => (await p.getByText("Found in search_logs").count()) >= 2); await shoot("05-pace-subagent", { pace: p }); diff --git a/spikes/durable-multiview/test/relay.test.ts b/spikes/durable-multiview/test/relay.test.ts index 2b908ba5..7f4ce0fd 100644 --- a/spikes/durable-multiview/test/relay.test.ts +++ b/spikes/durable-multiview/test/relay.test.ts @@ -1,5 +1,6 @@ import { mkdir, writeFile } from "node:fs/promises"; import { join } from "node:path"; +import { backendRelay } from "../../../apps/desktop/src/dev/durable-spike/backend-relay.ts"; 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"; @@ -10,6 +11,10 @@ import { connectTo, freePort, startFauxHost, tempDir, useCleanups, waitForView } const defer = useCleanups(); +function envelope(type: string, payload: Record) { + return { type: "event" as const, event: { id: "evt-1", seq: 0, sessionId: "", piSessionId: "", type, ts: "2026-10-02T00:00:00Z", payload } }; +} + /** An agent dir holding a host token where the bridge looks for it. */ async function agentDirWithToken(token: string): Promise { const dir = await tempDir(); @@ -36,6 +41,69 @@ async function connectThroughBridge(bridge: DurableSpikeBridge, url: string): Pr } describe("Pace backend relay", () => { + it.each(["lifecycle event", "lost socket"])("recovers a live client after a backend restart (%s) and delivers its next command", async (failure) => { + const host = await startFauxHost(defer, { answers: ["after-restart-answer"] }); + const listeners = new Set<(event: ReturnType) => void>(); + const emit = (event: ReturnType) => { + for (const listener of listeners) listener(event); + }; + const startBackend = () => { + const bridge = createDurableSpikeBridge({ agentDir: "/unused", defaultUrl: host.url }); + const detach = bridge.onEvent((event) => emit(envelope(`durable_spike.${event.kind}`, event))); + defer(() => bridge.dispose()); + return { bridge, detach }; + }; + let backend: ReturnType | undefined = startBackend(); + const invoke = (async (command: string, args: Record = {}) => { + if (backend === undefined) throw new Error("Pace backend is not connected"); + const connectionId = args.connectionId as string; + if (command === "durable_spike_connect") await backend.bridge.connect({ connectionId, token: host.token }); + else if (command === "durable_spike_send") backend.bridge.send({ connectionId, data: args.data as string }); + else if (command === "durable_spike_disconnect") backend.bridge.disconnect({ connectionId }); + else throw new Error(`Unexpected command: ${command}`); + return null; + }) as (command: string, args?: Record) => Promise; + const relay = backendRelay({ invoke, onBackendEvent: (listener) => { + listeners.add(listener); + return () => { listeners.delete(listener); }; + } }); + const client = await connectRemoteDurable({ transport: relayTransport(relay, "pace"), reconnectDelayMs: { min: 1, max: 5 } }); + defer(() => client.close()); + + // A crashed process cannot forward the sockets' close events. + backend.detach(); + backend.bridge.dispose(); + backend = undefined; + if (failure === "lifecycle event") { + const disconnected = envelope("error", { lifecycle: "disconnected", generation: 1 }); + disconnected.event.sessionId = "__backend__"; + emit(disconnected); + expect(client.view.current().connection).toBe("reconnecting"); + } else { + backend = startBackend(); + } + await client.controller.submit("offline-command", "followUp"); + expect(client.view.current().connection).toBe("reconnecting"); + expect(client.view.current().notices.at(-1)?.message).toMatch(/not connected|disconnected/i); + + backend ??= startBackend(); + const connected = envelope("status", { lifecycle: "connected", generation: 2 }); + connected.event.sessionId = "__backend__"; + emit(connected); + await waitForView(client.view, (view) => view.connection === "connected"); + await client.controller.submit("after-restart-command", "followUp"); + await waitForView(client.view, (view) => transcript(view.conversation).at(-1)?.text === "after-restart-answer"); + expect(transcript(client.view.current().conversation).map((item) => item.text)).toEqual([ + "after-restart-command", "after-restart-answer", + ]); + }); + + it("rejects a command for a connection the restarted backend no longer owns", () => { + const bridge = createDurableSpikeBridge({ agentDir: "/unused" }); + defer(() => bridge.dispose()); + expect(() => bridge.send({ connectionId: "before-restart", data: "command" })).toThrow(/connection.*not open/i); + }); + 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) }); diff --git a/spikes/durable-multiview/test/web-ui.test.ts b/spikes/durable-multiview/test/web-ui.test.ts index 2a28c1ea..c1895c9a 100644 --- a/spikes/durable-multiview/test/web-ui.test.ts +++ b/spikes/durable-multiview/test/web-ui.test.ts @@ -9,7 +9,7 @@ import { expect, it } from "vitest"; import { openHost, type OpenedHost } from "../host/host.ts"; import { chatItems } from "../presentation/chat.ts"; import { isBusy, transcript } from "../protocol/transcript.ts"; -import { connectTo, freePort, startFauxHost, tempDir, useCleanups, waitForView } from "./support.ts"; +import { connectTo, freePort, startDemoHost, startFauxHost, tempDir, useCleanups, waitForView } from "./support.ts"; const defer = useCleanups(); const webConfig = fileURLToPath(new URL("../web/vite.config.ts", import.meta.url)); @@ -67,6 +67,33 @@ it("preserves an unsent follow-up and disables write controls while reconnecting expect(await composer.textContent()).toBe("keep this draft"); }); +it("sends a typed steer from the button without stopping the pending tool, and still stops with empty input", async () => { + const webOrigin = await startWeb(); + const { host } = await startDemoHost(defer, { browserOrigins: [webOrigin] }); + const observer = await connectTo(defer, host); + const page = await openPage(host, 1280, webOrigin); + const composer = page.getByRole("textbox"); + await composer.fill("roll back to v2.2"); + await composer.press("Enter"); + await waitForView(observer.view, (view) => view.approvals.length === 1); + await page.getByRole("button", { name: /stop/i, exact: true }).waitFor(); + + await composer.fill("keep eu-west drained until the fix ships"); + const send = page.getByRole("button", { name: /send/i, exact: true }); + expect(await send.count()).toBe(1); + await send.click(); + await waitForView(observer.view, (view) => { + const inbox = view.conversation.docs["pi.inbox"] as { items?: { mode: string }[] } | undefined; + return inbox?.items?.some((item) => item.mode === "steer") === true; + }); + expect(observer.view.current().approvals).toHaveLength(1); + expect(isBusy(observer.view.current().conversation)).toBe(true); + await expect.poll(() => composer.textContent()).toBe(""); + + await page.getByRole("button", { name: /stop/i, exact: true }).click(); + await waitForView(observer.view, (view) => !isBusy(view.conversation) && view.approvals.length === 0); +}); + it.each([ ["thinking-only", fauxAssistantMessage(fauxThinking("Compare the release evidence."))], ["tool-only", fauxAssistantMessage(fauxToolCall("search_logs", { release: "v2.3" }), { stopReason: "toolUse" })], diff --git a/spikes/durable-multiview/web/App.tsx b/spikes/durable-multiview/web/App.tsx index bf6ab9e2..4d44449f 100644 --- a/spikes/durable-multiview/web/App.tsx +++ b/spikes/durable-multiview/web/App.tsx @@ -353,7 +353,7 @@ function Composer({ onChange={setValue} // Enter prompts when idle and steers when busy, as in the TUI. onSubmit={(text) => void remote.controller.submit(text, "steer")} - isStopShown={busy} + isStopShown={busy && value.trim() === ""} onStop={() => void remote.controller.abort()} isDisabled={disconnected} placeholder={busy ? "Steer the running turn…" : "Ask the agent…"}