Skip to content

Commit 13f95ca

Browse files
committed
fix(spike): recover backend connections and preserve typed steering
1 parent 2b8a017 commit 13f95ca

8 files changed

Lines changed: 229 additions & 16 deletions

File tree

‎apps/desktop/src/dev/durable-spike/backend-relay.test.ts‎

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
import type { BackendRpcEvent } from "@pace/backend";
2+
import { relayTransport } from "@pace/durable-spike/protocol/transport.ts";
23
import { describe, expect, it, vi } from "vitest";
34
import { backendRelay } from "./backend-relay";
45

@@ -7,6 +8,44 @@ function envelope(type: string, payload: Record<string, unknown>): BackendRpcEve
78
}
89

910
describe("backendRelay", () => {
11+
it("closes the old transport on backend exit and carries frames on a new connection", async () => {
12+
const listeners = new Set<(event: BackendRpcEvent) => void>();
13+
const emit = (event: BackendRpcEvent) => {
14+
for (const listener of listeners) listener(event);
15+
};
16+
const invoke = vi.fn(async () => null) as unknown as <T>(command: string, args?: Record<string, unknown>) => Promise<T>;
17+
const transport = relayTransport(backendRelay({
18+
invoke,
19+
onBackendEvent: (listener) => {
20+
listeners.add(listener);
21+
return () => { listeners.delete(listener); };
22+
},
23+
}), "pace");
24+
const closed = vi.fn();
25+
const first = transport.open({ message: vi.fn(), closed });
26+
const firstId = vi.mocked(invoke).mock.calls[0]![1]!.connectionId;
27+
28+
const disconnected = envelope("error", { lifecycle: "disconnected", generation: 1 });
29+
disconnected.event.sessionId = "__backend__";
30+
emit(disconnected);
31+
expect(closed).toHaveBeenCalledOnce();
32+
expect(closed).toHaveBeenCalledWith(1006, expect.stringMatching(/backend/i));
33+
first.send("must-not-reach-the-old-connection");
34+
expect(vi.mocked(invoke).mock.calls.filter(([command]) => command === "durable_spike_send")).toEqual([]);
35+
36+
const message = vi.fn();
37+
const second = transport.open({ message, closed: vi.fn() });
38+
const calls = vi.mocked(invoke).mock.calls;
39+
const secondId = calls[calls.length - 1]![1]!.connectionId;
40+
expect(secondId).not.toBe(firstId);
41+
second.send("new-command");
42+
emit(envelope("durable_spike.frame", { connectionId: firstId, data: "stale" }));
43+
emit(envelope("durable_spike.frame", { connectionId: secondId, data: "new-reply" }));
44+
expect(invoke).toHaveBeenCalledWith("durable_spike_send", { connectionId: secondId, data: "new-command" });
45+
expect(message).toHaveBeenCalledExactlyOnceWith("new-reply");
46+
second.close();
47+
});
48+
1049
it("maps relay calls to backend commands", async () => {
1150
const invoke = vi.fn(async () => null) as unknown as <T>(command: string, args?: Record<string, unknown>) => Promise<T>;
1251
const relay = backendRelay({ invoke, onBackendEvent: () => () => {} });
@@ -22,6 +61,21 @@ describe("backendRelay", () => {
2261
]);
2362
});
2463

64+
it("closes the transport when a frame cannot reach the backend", async () => {
65+
const invoke = (async (command: string) => {
66+
if (command === "durable_spike_send") throw new Error("Backend is not connected");
67+
return null;
68+
}) as <T>(command: string, args?: Record<string, unknown>) => Promise<T>;
69+
const transport = relayTransport(backendRelay({ invoke, onBackendEvent: () => () => {} }), "pace");
70+
const closed = vi.fn();
71+
const connection = transport.open({ message: vi.fn(), closed });
72+
connection.send("pending-command");
73+
await Promise.resolve();
74+
75+
expect(closed).toHaveBeenCalledExactlyOnceWith(1006, "Backend is not connected");
76+
connection.close();
77+
});
78+
2579
it("passes on frames and closes from backend events and drops everything else", () => {
2680
let emit: (event: BackendRpcEvent) => void = () => {};
2781
const relay = backendRelay({

‎apps/desktop/src/dev/durable-spike/backend-relay.ts‎

Lines changed: 55 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -1,27 +1,71 @@
1-
import type { BackendRpcEvent } from "@pace/backend";
2-
import type { FrameRelay, RelayEvent } from "@pace/durable-spike/protocol/transport.ts";
1+
import type { FrameRelay, RelayEvent } from "../../../../../spikes/durable-multiview/protocol/transport.ts";
32

43
type Invoke = <T>(command: string, args?: Record<string, unknown>) => Promise<T>;
5-
type SubscribeBackendEvent = (listener: (event: BackendRpcEvent) => void) => () => void;
4+
type BackendEvent = {
5+
type: "event";
6+
event: { sessionId: string; type: string; payload: Record<string, unknown> };
7+
};
8+
type SubscribeBackendEvent = (listener: (event: BackendEvent) => void) => () => void;
69

710
/**
811
* The spike's relay over Pace's backend: the renderer cannot open a WebSocket
912
* under its CSP, so the backend holds it and frames cross the existing IPC.
1013
*/
1114
export function backendRelay({ invoke, onBackendEvent }: { invoke: Invoke; onBackendEvent: SubscribeBackendEvent }): FrameRelay {
15+
const connections = new Set<string>();
16+
const listeners = new Set<(event: RelayEvent) => void>();
17+
let unsubscribe: (() => void) | undefined;
18+
const emit = (event: RelayEvent) => {
19+
for (const listener of listeners) listener(event);
20+
};
1221
return {
13-
connect: (connectionId) => invoke<null>("durable_spike_connect", { connectionId }).then(() => {}),
14-
send: (connectionId, data) => void invoke("durable_spike_send", { connectionId, data }),
15-
disconnect: (connectionId) => void invoke("durable_spike_disconnect", { connectionId }),
16-
subscribe: (listener) =>
17-
onBackendEvent((event) => {
22+
connect: async (connectionId) => {
23+
connections.add(connectionId);
24+
try {
25+
await invoke<null>("durable_spike_connect", { connectionId });
26+
} catch (error) {
27+
connections.delete(connectionId);
28+
throw error;
29+
}
30+
},
31+
send: (connectionId, data) => {
32+
void invoke("durable_spike_send", { connectionId, data }).catch((error: unknown) => {
33+
if (!connections.delete(connectionId)) return;
34+
emit({ kind: "closed", connectionId, code: 1006, reason: error instanceof Error ? error.message : String(error) });
35+
});
36+
},
37+
disconnect: (connectionId) => {
38+
connections.delete(connectionId);
39+
// The transport is already closed locally; a dead backend has nothing left to close.
40+
void invoke("durable_spike_disconnect", { connectionId }).catch(() => {});
41+
},
42+
subscribe: (listener) => {
43+
listeners.add(listener);
44+
unsubscribe ??= onBackendEvent((event) => {
45+
if (event.event.sessionId === "__backend__" && event.event.payload.lifecycle === "disconnected") {
46+
for (const connectionId of connections) {
47+
connections.delete(connectionId);
48+
emit({ kind: "closed", connectionId, code: 1006, reason: "Pace backend disconnected" });
49+
}
50+
return;
51+
}
1852
const relayed = relayEventOf(event);
19-
if (relayed !== undefined) listener(relayed);
20-
}),
53+
if (relayed === undefined) return;
54+
if (relayed.kind === "closed") connections.delete(relayed.connectionId);
55+
emit(relayed);
56+
});
57+
return () => {
58+
listeners.delete(listener);
59+
if (listeners.size === 0) {
60+
unsubscribe?.();
61+
unsubscribe = undefined;
62+
}
63+
};
64+
},
2165
};
2266
}
2367

24-
function relayEventOf(event: BackendRpcEvent): RelayEvent | undefined {
68+
function relayEventOf(event: BackendEvent): RelayEvent | undefined {
2569
if (event.type !== "event") return undefined;
2670
const { type, payload } = event.event;
2771
if (typeof payload.connectionId !== "string") return undefined;

‎docs/research/durable-multiview-review-fixes.md‎

Lines changed: 17 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
# Durable 多端 spike:高优先检视修复
22

3-
本轮针对已合并 spike 的审批归属、订阅竞态和消息边界问题补充回归保护。其余历史检视不在本轮范围内;本轮也不代表生产迁移或安全验收完成。
3+
本轮针对已合并 spike 的审批归属、订阅竞态、消息边界、backend 重连及 steer 按钮问题补充回归保护。其余历史检视不在本轮范围内;本轮也不代表生产迁移或安全验收完成。
44

55
## 宿主与审批边界
66

@@ -32,3 +32,19 @@ bun run web:link
3232
- `test/host-startup.test.ts`:从进程输出确认凭证未泄露,再从本地 token 文件连接真实宿主。
3333

3434
原有宿主恢复、工具热替换及完整三端脚本已改为从本地 token 文件取凭证,不再从日志读取。
35+
36+
## 客户端恢复与发送
37+
38+
backend 的 `__backend__/disconnected` 生命周期事件会关闭现存的 relay transport,交给原有重连路径建立新连接。即使该事件丢失,向不存在或未打开的 socket 发送也会明确失败,renderer 将 IPC 拒绝转换为断线;提示不会再静默消失。未完成的调用会显示错误,结果未知的命令不会自动重放。
39+
40+
运行中的 composer 只有在输入为空时显示 Stop。有文本时显示 Send,点击后提交 steer,并保留正在等待审批的工具;发送后输入清空,Stop 恢复可用。
41+
42+
- `backend-relay.test.ts`:使用实际 backend 事件结构验证断线通知及 IPC 发送失败。
43+
- `test/relay.test.ts`:真实宿主与 backend bridge,在 backend 替换和生命周期事件丢失两种情况下重新连接,后续提示均得到模型回答。
44+
- `test/web-ui.test.ts`:真实浏览器点击 Send 后,steer 入队且工具仍待审批;输入清空后点击 Stop 终止任务。
45+
46+
## 本轮验证
47+
48+
2026-10-02:宿主层独立通过 spike typecheck 和 60 项测试;完整修复栈通过根 typecheck、lint、build,根测试 168 个文件 / 1977 项,spike typecheck 与测试 15 个文件 / 64 项。每个具体缺陷均先运行失败用例,再实现修复。
49+
50+
完整 `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)。

‎packages/backend/src/spikes/durable-bridge.ts‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -69,7 +69,8 @@ export function createDurableSpikeBridge(options: {
6969
},
7070
send({ connectionId, data }) {
7171
const socket = sockets.get(connectionId);
72-
if (socket?.readyState === WebSocket.OPEN) socket.send(data);
72+
if (socket?.readyState !== WebSocket.OPEN) throw new Error("Durable host connection is not open");
73+
socket.send(data);
7374
},
7475
disconnect({ connectionId }) {
7576
sockets.get(connectionId)?.close();

‎spikes/durable-multiview/scripts/demo-scenario.mjs‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -178,7 +178,10 @@ try {
178178
await ask(p, "also look at ap-south");
179179
const logsId = subagents().find((summary) => summary.title.startsWith("logs:")).id;
180180
await until("the logs subagent is running a tool", () => Object.values(seen().tasks?.tasks ?? {}).some((node) => node.kind === "pi.tool" && node.conversationId === logsId));
181-
await ask(p, "focus on the connection pool limit");
181+
await p.getByRole("textbox").first().fill("focus on the connection pool limit");
182+
await p.getByRole("button", { name: "Send", exact: true }).waitFor();
183+
await shoot("05-steer-ready", { pace: p });
184+
await p.getByRole("button", { name: "Send", exact: true }).click();
182185
await until("Pace steers the running subagent", shows(p, "focus on the connection pool limit"));
183186
await until("the subagent answers Pace", async () => (await p.getByText("Found in search_logs").count()) >= 2);
184187
await shoot("05-pace-subagent", { pace: p });

‎spikes/durable-multiview/test/relay.test.ts‎

Lines changed: 68 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
import { mkdir, writeFile } from "node:fs/promises";
22
import { join } from "node:path";
3+
import { backendRelay } from "../../../apps/desktop/src/dev/durable-spike/backend-relay.ts";
34
import { describe, expect, it } from "vitest";
45
// Pace's backend relay, the hop between a host and Pace's renderer.
56
import { createDurableSpikeBridge, type DurableSpikeBridge } from "../../../packages/backend/src/spikes/durable-bridge.ts";
@@ -10,6 +11,10 @@ import { connectTo, freePort, startFauxHost, tempDir, useCleanups, waitForView }
1011

1112
const defer = useCleanups();
1213

14+
function envelope(type: string, payload: Record<string, unknown>) {
15+
return { type: "event" as const, event: { id: "evt-1", seq: 0, sessionId: "", piSessionId: "", type, ts: "2026-10-02T00:00:00Z", payload } };
16+
}
17+
1318
/** An agent dir holding a host token where the bridge looks for it. */
1419
async function agentDirWithToken(token: string): Promise<string> {
1520
const dir = await tempDir();
@@ -36,6 +41,69 @@ async function connectThroughBridge(bridge: DurableSpikeBridge, url: string): Pr
3641
}
3742

3843
describe("Pace backend relay", () => {
44+
it.each(["lifecycle event", "lost socket"])("recovers a live client after a backend restart (%s) and delivers its next command", async (failure) => {
45+
const host = await startFauxHost(defer, { answers: ["after-restart-answer"] });
46+
const listeners = new Set<(event: ReturnType<typeof envelope>) => void>();
47+
const emit = (event: ReturnType<typeof envelope>) => {
48+
for (const listener of listeners) listener(event);
49+
};
50+
const startBackend = () => {
51+
const bridge = createDurableSpikeBridge({ agentDir: "/unused", defaultUrl: host.url });
52+
const detach = bridge.onEvent((event) => emit(envelope(`durable_spike.${event.kind}`, event)));
53+
defer(() => bridge.dispose());
54+
return { bridge, detach };
55+
};
56+
let backend: ReturnType<typeof startBackend> | undefined = startBackend();
57+
const invoke = (async (command: string, args: Record<string, unknown> = {}) => {
58+
if (backend === undefined) throw new Error("Pace backend is not connected");
59+
const connectionId = args.connectionId as string;
60+
if (command === "durable_spike_connect") await backend.bridge.connect({ connectionId, token: host.token });
61+
else if (command === "durable_spike_send") backend.bridge.send({ connectionId, data: args.data as string });
62+
else if (command === "durable_spike_disconnect") backend.bridge.disconnect({ connectionId });
63+
else throw new Error(`Unexpected command: ${command}`);
64+
return null;
65+
}) as <T>(command: string, args?: Record<string, unknown>) => Promise<T>;
66+
const relay = backendRelay({ invoke, onBackendEvent: (listener) => {
67+
listeners.add(listener);
68+
return () => { listeners.delete(listener); };
69+
} });
70+
const client = await connectRemoteDurable({ transport: relayTransport(relay, "pace"), reconnectDelayMs: { min: 1, max: 5 } });
71+
defer(() => client.close());
72+
73+
// A crashed process cannot forward the sockets' close events.
74+
backend.detach();
75+
backend.bridge.dispose();
76+
backend = undefined;
77+
if (failure === "lifecycle event") {
78+
const disconnected = envelope("error", { lifecycle: "disconnected", generation: 1 });
79+
disconnected.event.sessionId = "__backend__";
80+
emit(disconnected);
81+
expect(client.view.current().connection).toBe("reconnecting");
82+
} else {
83+
backend = startBackend();
84+
}
85+
await client.controller.submit("offline-command", "followUp");
86+
expect(client.view.current().connection).toBe("reconnecting");
87+
expect(client.view.current().notices.at(-1)?.message).toMatch(/not connected|disconnected/i);
88+
89+
backend ??= startBackend();
90+
const connected = envelope("status", { lifecycle: "connected", generation: 2 });
91+
connected.event.sessionId = "__backend__";
92+
emit(connected);
93+
await waitForView(client.view, (view) => view.connection === "connected");
94+
await client.controller.submit("after-restart-command", "followUp");
95+
await waitForView(client.view, (view) => transcript(view.conversation).at(-1)?.text === "after-restart-answer");
96+
expect(transcript(client.view.current().conversation).map((item) => item.text)).toEqual([
97+
"after-restart-command", "after-restart-answer",
98+
]);
99+
});
100+
101+
it("rejects a command for a connection the restarted backend no longer owns", () => {
102+
const bridge = createDurableSpikeBridge({ agentDir: "/unused" });
103+
defer(() => bridge.dispose());
104+
expect(() => bridge.send({ connectionId: "before-restart", data: "command" })).toThrow(/connection.*not open/i);
105+
});
106+
39107
it("carries a client that sees and drives the same conversation as a direct one", async () => {
40108
const host = await startFauxHost(defer, { answers: ["answer-through-the-relay"] });
41109
const bridge = createDurableSpikeBridge({ agentDir: await agentDirWithToken(host.token) });

‎spikes/durable-multiview/test/web-ui.test.ts‎

Lines changed: 28 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,7 @@ import { expect, it } from "vitest";
99
import { openHost, type OpenedHost } from "../host/host.ts";
1010
import { chatItems } from "../presentation/chat.ts";
1111
import { isBusy, transcript } from "../protocol/transcript.ts";
12-
import { connectTo, freePort, startFauxHost, tempDir, useCleanups, waitForView } from "./support.ts";
12+
import { connectTo, freePort, startDemoHost, startFauxHost, tempDir, useCleanups, waitForView } from "./support.ts";
1313

1414
const defer = useCleanups();
1515
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
6767
expect(await composer.textContent()).toBe("keep this draft");
6868
});
6969

70+
it("sends a typed steer from the button without stopping the pending tool, and still stops with empty input", async () => {
71+
const webOrigin = await startWeb();
72+
const { host } = await startDemoHost(defer, { browserOrigins: [webOrigin] });
73+
const observer = await connectTo(defer, host);
74+
const page = await openPage(host, 1280, webOrigin);
75+
const composer = page.getByRole("textbox");
76+
await composer.fill("roll back to v2.2");
77+
await composer.press("Enter");
78+
await waitForView(observer.view, (view) => view.approvals.length === 1);
79+
await page.getByRole("button", { name: /stop/i, exact: true }).waitFor();
80+
81+
await composer.fill("keep eu-west drained until the fix ships");
82+
const send = page.getByRole("button", { name: /send/i, exact: true });
83+
expect(await send.count()).toBe(1);
84+
await send.click();
85+
await waitForView(observer.view, (view) => {
86+
const inbox = view.conversation.docs["pi.inbox"] as { items?: { mode: string }[] } | undefined;
87+
return inbox?.items?.some((item) => item.mode === "steer") === true;
88+
});
89+
expect(observer.view.current().approvals).toHaveLength(1);
90+
expect(isBusy(observer.view.current().conversation)).toBe(true);
91+
await expect.poll(() => composer.textContent()).toBe("");
92+
93+
await page.getByRole("button", { name: /stop/i, exact: true }).click();
94+
await waitForView(observer.view, (view) => !isBusy(view.conversation) && view.approvals.length === 0);
95+
});
96+
7097
it.each([
7198
["thinking-only", fauxAssistantMessage(fauxThinking("Compare the release evidence."))],
7299
["tool-only", fauxAssistantMessage(fauxToolCall("search_logs", { release: "v2.3" }), { stopReason: "toolUse" })],

‎spikes/durable-multiview/web/App.tsx‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -353,7 +353,7 @@ function Composer({
353353
onChange={setValue}
354354
// Enter prompts when idle and steers when busy, as in the TUI.
355355
onSubmit={(text) => void remote.controller.submit(text, "steer")}
356-
isStopShown={busy}
356+
isStopShown={busy && value.trim() === ""}
357357
onStop={() => void remote.controller.abort()}
358358
isDisabled={disconnected}
359359
placeholder={busy ? "Steer the running turn…" : "Ask the agent…"}

0 commit comments

Comments
 (0)