fix(spike): recover backend connections and preserve typed steering

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<string, unknown>): 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 <T>(command: string, args?: Record<string, unknown>) => Promise<T>;
+    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 <T>(command: string, args?: Record<string, unknown>) => Promise<T>;
     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 <T>(command: string, args?: Record<string, unknown>) => Promise<T>;
+    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({

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 = <T>(command: string, args?: Record<string, unknown>) => Promise<T>;
-type SubscribeBackendEvent = (listener: (event: BackendRpcEvent) => void) => () => void;
+type BackendEvent = {
+  type: "event";
+  event: { sessionId: string; type: string; payload: Record<string, unknown> };
+};
+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<string>();
+  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<null>("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<null>("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;

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

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

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

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<string, unknown>) {
+  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<string> {
   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<typeof envelope>) => void>();
+    const emit = (event: ReturnType<typeof envelope>) => {
+      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<typeof startBackend> | undefined = startBackend();
+    const invoke = (async (command: string, args: Record<string, unknown> = {}) => {
+      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 <T>(command: string, args?: Record<string, unknown>) => Promise<T>;
+    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) });

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

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