fix(serve): Keep restore request shapes distinct

docs/design/2026-08-10-transactional-webui-session-switching.md

@@ -12,9 +12,9 @@ Modern transactional behavior requires a successful capability snapshot that adv
 
 ## Coordinator
 
-Each provider owns one raw restore slot and one desired intent. Equivalent requests coalesce. A newer target rejects the prior public intent and replaces the queued intent, while an already-running SDK request continues to settlement because it is not cancellable. Its result is adopted only when it still matches the latest target; otherwise its attachment is detached once on a best-effort basis. The queued deadline begins when the caller requests the switch, so an expired target never starts a restore.
+Each provider owns one raw restore slot and one desired intent. Restore equivalence includes the normalized session and workspace plus the effective replay shape: `resume/none`, `load/all`, or `load/recent(N)`. The provider snapshots the effective page when admitting the intent, after applying daemon pagination capability, and uses that snapshot for the initial request, queued execution, and retries. Only exact shapes coalesce; a non-equivalent request rejects the prior public intent, permanently marks any different raw result as superseded, and replaces the queued intent, while the already-running SDK request continues to settlement because it is not cancellable. The superseded result is never adopted even if a later intent returns to its shape; its attachment is detached once on a best-effort basis. A timed-out raw request that has not been superseded by a different shape may still satisfy an exact-shape retry. The queued deadline begins when the caller requests the switch, so an expired target never starts a restore.
 
-Commit is guarded by the desired intent, absolute deadline, provider environment, local lifecycle, source logical identity, and restored target identity. Timeout, SDK failure, supersede, staging failure, and commit are explicit competing terminal states rather than an implicit `Promise.race`.
+Commit is guarded by the desired intent, absolute deadline, provider environment, local lifecycle, source logical identity, and restored target identity. A same-shape retry may adopt a late raw result only when an ordinary timeout left the lifecycle unchanged; an explicit lifecycle cancellation fences that result even if a later intent returns to the same shape. Timeout, SDK failure, supersede, staging failure, and commit are explicit competing terminal states rather than an implicit `Promise.race`.
 
 ## Staging and commit
 

integration-tests/cli/qwen-serve-webui-session-switching.test.ts

@@ -255,6 +255,125 @@ describe('qwen serve WebUI transactional session switching', () => {
     }
   }, 30_000);
 
+  it('serializes non-equivalent load and resume requests for one target', async () => {
+    const originalFetch = globalThis.fetch;
+    const state = await setup();
+    const loadResponseReady = deferred();
+    const releaseLoadResponse = deferred();
+    let loadRequests = 0;
+    let resumeRequests = 0;
+    let targetDetachRequests = 0;
+    let staleClientId: string | undefined;
+    const detachedClientIds: Array<string | null> = [];
+    let loadOutcome: Promise<unknown> | undefined;
+    let resumeOutcome: Promise<void> | undefined;
+    try {
+      globalThis.fetch = async (input, init) => {
+        const request =
+          input instanceof Request ? input : new Request(input, init);
+        const pathname = new URL(request.url).pathname;
+        const targetPath = `/session/${encodeURIComponent(state.target.sessionId)}`;
+        if (
+          request.method === 'POST' &&
+          pathname.endsWith(`${targetPath}/resume`)
+        ) {
+          resumeRequests += 1;
+        }
+        if (
+          request.method === 'POST' &&
+          pathname.endsWith(`${targetPath}/detach`)
+        ) {
+          targetDetachRequests += 1;
+          detachedClientIds.push(request.headers.get('X-Qwen-Client-Id'));
+        }
+        const response = await originalFetch(request);
+        if (
+          request.method === 'POST' &&
+          pathname.endsWith(`${targetPath}/load`)
+        ) {
+          loadRequests += 1;
+          const payload = (await response.clone().json()) as {
+            clientId?: unknown;
+          };
+          staleClientId =
+            typeof payload.clientId === 'string' ? payload.clientId : undefined;
+          loadResponseReady.resolve();
+          await releaseLoadResponse.promise;
+        }
+        return response;
+      };
+      act(() => {
+        loadOutcome = state
+          .getActions()
+          .loadSession(state.target.sessionId, {
+            workspaceCwd: state.workspace,
+          })
+          .then(
+            () => undefined,
+            (error: unknown) => error,
+          );
+      });
+      await loadResponseReady.promise;
+
+      act(() => {
+        resumeOutcome = state
+          .getActions()
+          .resumeSession(state.target.sessionId, {
+            workspaceCwd: state.workspace,
+          });
+      });
+      expect(await loadOutcome).toMatchObject({ name: 'AbortError' });
+      expect(loadRequests).toBe(1);
+      expect(resumeRequests).toBe(0);
+      expect(state.getConnection()).toMatchObject({
+        status: 'connected',
+        sessionId: state.source.sessionId,
+        sessionTransition: { phase: 'queued', operation: 'resume' },
+      });
+
+      await activeDaemon!.client.prompt(state.source.sessionId, {
+        prompt: [{ type: 'text', text: 'source remains live while queued' }],
+      });
+      await waitFor(
+        () =>
+          JSON.stringify(state.getBlocks()).includes(
+            'source remains live while queued',
+          ),
+        'source event while resume is queued',
+      );
+
+      await act(async () => {
+        releaseLoadResponse.resolve();
+        await resumeOutcome;
+      });
+      expect(resumeRequests).toBe(1);
+      await waitFor(
+        () => targetDetachRequests === 1,
+        'stale target attachment cleanup',
+      );
+      expect(staleClientId).toBeTruthy();
+      expect(detachedClientIds).toEqual([staleClientId]);
+      expect(state.getConnection()).toMatchObject({
+        status: 'connected',
+        sessionId: state.target.sessionId,
+      });
+      expect(state.getConnection()?.clientId).toBeTruthy();
+      expect(state.getConnection()?.clientId).not.toBe(staleClientId);
+    } finally {
+      globalThis.fetch = originalFetch;
+      releaseLoadResponse.resolve();
+      await loadOutcome?.catch(() => undefined);
+      await resumeOutcome?.catch(() => undefined);
+      if (root) {
+        await act(async () => root?.unmount());
+        root = undefined;
+      }
+      await activeDaemon?.dispose();
+      activeDaemon = undefined;
+      fs.rmSync(state.workspace, { recursive: true, force: true });
+    }
+  }, 30_000);
+
   it('preserves the source after a structured target timeout', async () => {
     const originalFetch = globalThis.fetch;
     const state = await setup();

packages/acp-bridge/src/bridge.test.ts

@@ -5294,6 +5294,204 @@ describe('createAcpSessionBridge', () => {
     await bridge.shutdown();
   });
 
+  it('coalesces response restores with the same history page', async () => {
+    const load = deferred<LoadSessionResponse>();
+    const handle = makeChannel({ loadSessionImpl: () => load.promise });
+    const bridge = makeBridge({ channelFactory: async () => handle.channel });
+
+    const first = bridge.loadSession({
+      sessionId: 'coalesce-page',
+      workspaceCwd: WS_A,
+      historyReplay: 'response',
+      historyPageSize: 100,
+    });
+    for (let i = 0; i < 50 && handle.agent.loadSessionCalls.length !== 1; i++) {
+      await new Promise((resolve) => setTimeout(resolve, 10));
+    }
+    const second = bridge.loadSession({
+      sessionId: 'coalesce-page',
+      workspaceCwd: WS_A,
+      historyReplay: 'response',
+      historyPageSize: 100,
+    });
+
+    load.resolve({ _meta: { tag: 'same-page' } });
+    const [owner, waiter] = await Promise.all([first, second]);
+
+    expect(handle.agent.loadSessionCalls).toHaveLength(1);
+    expect(handle.agent.loadSessionCalls[0]?._meta).toMatchObject({
+      'qwen.session.loadReplayMode': 'bulk',
+      'qwen.session.loadReplayPageSize': 100,
+    });
+    expect(owner).toMatchObject({
+      attached: false,
+      state: { _meta: { tag: 'same-page' } },
+    });
+    expect(waiter).toMatchObject({
+      attached: true,
+      state: { _meta: { tag: 'same-page' } },
+    });
+    expect(waiter.clientId).not.toBe(owner.clientId);
+
+    await bridge.shutdown();
+  });
+
+  it.each([0, 501, 1.5])(
+    'rejects invalid response history page %s before restore',
+    async (historyPageSize) => {
+      const handle = makeChannel();
+      const bridge = makeBridge({
+        channelFactory: async () => handle.channel,
+      });
+
+      await expect(
+        bridge.loadSession({
+          sessionId: 'invalid-page',
+          workspaceCwd: WS_A,
+          historyReplay: 'response',
+          historyPageSize,
+        }),
+      ).rejects.toThrow('Invalid historyPageSize');
+      expect(handle.agent.loadSessionCalls).toHaveLength(0);
+      await bridge.shutdown();
+    },
+  );
+
+  it.each([
+    ['a different explicit page', 500],
+    ['an omitted page', undefined],
+  ] as const)(
+    'rejects response restore coalescing with %s',
+    async (_label, historyPageSize) => {
+      const load = deferred<LoadSessionResponse>();
+      const handle = makeChannel({ loadSessionImpl: () => load.promise });
+      const bridge = makeBridge({
+        channelFactory: async () => handle.channel,
+      });
+
+      const first = bridge.loadSession({
+        sessionId: 'mismatched-page',
+        workspaceCwd: WS_A,
+        historyReplay: 'response',
+        historyPageSize: 100,
+      });
+      for (
+        let i = 0;
+        i < 50 && handle.agent.loadSessionCalls.length !== 1;
+        i++
+      ) {
+        await new Promise((resolve) => setTimeout(resolve, 10));
+      }
+
+      await expect(
+        bridge.loadSession({
+          sessionId: 'mismatched-page',
+          workspaceCwd: WS_A,
+          historyReplay: 'response',
+          ...(historyPageSize !== undefined ? { historyPageSize } : {}),
+        }),
+      ).rejects.toBeInstanceOf(RestoreInProgressError);
+
+      load.resolve({});
+      const restored = await first;
+      expect(handle.agent.loadSessionCalls).toHaveLength(1);
+      await bridge.killSession(restored.sessionId, {
+        requireZeroAttaches: true,
+      });
+      expect(bridge.sessionCount).toBe(0);
+      await bridge.shutdown();
+    },
+  );
+
+  it('ignores history pages when coalescing streamed loads', async () => {
+    const load = deferred<LoadSessionResponse>();
+    const handle = makeChannel({ loadSessionImpl: () => load.promise });
+    const bridge = makeBridge({ channelFactory: async () => handle.channel });
+
+    const first = bridge.loadSession({
+      sessionId: 'stream-pages-ignored',
+      workspaceCwd: WS_A,
+      historyReplay: 'stream',
+      historyPageSize: 100,
+    });
+    for (let i = 0; i < 50 && handle.agent.loadSessionCalls.length !== 1; i++) {
+      await new Promise((resolve) => setTimeout(resolve, 10));
+    }
+    const second = bridge.loadSession({
+      sessionId: 'stream-pages-ignored',
+      workspaceCwd: WS_A,
+      historyReplay: 'stream',
+      historyPageSize: 500,
+    });
+
+    load.resolve({});
+    await Promise.all([first, second]);
+    expect(handle.agent.loadSessionCalls).toHaveLength(1);
+    expect(handle.agent.loadSessionCalls[0]?._meta).not.toHaveProperty(
+      'qwen.session.loadReplayPageSize',
+    );
+    await bridge.shutdown();
+  });
+
+  it('ignores history pages when coalescing resumes', async () => {
+    const resume = deferred<ResumeSessionResponse>();
+    const handle = makeChannel({
+      resumeSessionImpl: () => resume.promise,
+    });
+    const bridge = makeBridge({ channelFactory: async () => handle.channel });
+
+    const first = bridge.resumeSession({
+      sessionId: 'resume-pages-ignored',
+      workspaceCwd: WS_A,
+      historyPageSize: 100,
+    });
+    for (
+      let i = 0;
+      i < 50 && handle.agent.resumeSessionCalls.length !== 1;
+      i++
+    ) {
+      await new Promise((resolve) => setTimeout(resolve, 10));
+    }
+    const second = bridge.resumeSession({
+      sessionId: 'resume-pages-ignored',
+      workspaceCwd: WS_A,
+      historyPageSize: 500,
+    });
+
+    resume.resolve({});
+    await Promise.all([first, second]);
+    expect(handle.agent.resumeSessionCalls).toHaveLength(1);
+    await bridge.shutdown();
+  });
+
+  it('rejects restore coalescing with different inherited-history policies', async () => {
+    const load = deferred<LoadSessionResponse>();
+    const handle = makeChannel({ loadSessionImpl: () => load.promise });
+    const bridge = makeBridge({ channelFactory: async () => handle.channel });
+
+    const first = bridge.loadSession({
+      sessionId: 'coalesce-inherited-policy',
+      workspaceCwd: WS_A,
+      hideInheritedHistory: true,
+    });
+    for (let i = 0; i < 50 && handle.agent.loadSessionCalls.length !== 1; i++) {
+      await new Promise((resolve) => setTimeout(resolve, 10));
+    }
+
+    await expect(
+      bridge.loadSession({
+        sessionId: 'coalesce-inherited-policy',
+        workspaceCwd: WS_A,
+        hideInheritedHistory: false,
+      }),
+    ).rejects.toBeInstanceOf(RestoreInProgressError);
+
+    load.resolve({});
+    await first;
+    expect(handle.agent.loadSessionCalls).toHaveLength(1);
+    await bridge.shutdown();
+  });
+
   it('rejects coalescing load requests with incompatible replay modes', async () => {
     let releaseLoad: ((value: LoadSessionResponse) => void) | undefined;
     const factory: ChannelFactory = async () =>