fix(webui): Make cross-session switching transactional

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

@@ -0,0 +1,37 @@
+# Transactional cross-session switching
+
+## Problem
+
+The WebUI historically detached the current session, stopped its event stream, and cleared its transcript before a target `loadSession` or `resumeSession` completed. A slow or failed restore therefore left the user without the still-healthy source session. The WebShell also keyed its main provider by the requested session, so controlled navigation remounted the provider before the target was usable.
+
+## Scope
+
+This change makes only cross-logical-session load and resume transactional. A logical target is the normalized `(sessionId, workspaceCwd)` pair. Initial bootstrap, same-logical reload, client-id replacement, full resync, memory repair, and branch adoption retain their existing behavior and are follow-up work.
+
+Modern transactional behavior requires a successful capability snapshot that advertises `client_identity` and concrete client IDs for both attachments. A daemon that explicitly lacks the feature retains the legacy destructive path. Unknown capabilities or malformed modern responses fail closed and preserve the source.
+
+## 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.
+
+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`.
+
+## Staging and commit
+
+Replay is normalized into an unsubscribed shadow transcript store in batches of at most 512 events. The compacted replay and live journal arrays are traversed directly and are not concatenated. Only bounded summaries of notices and side-channel events are retained. Staging never writes the visible transcript, connection, prompt maps, notices, or workspace signals.
+
+After the final guard succeeds, one synchronous commit flushes the source runner's legal buffered events, stops its stream, installs the target transcript/history/session/workspace/client and connection ref, notifies the WebShell wrapper, publishes staged side effects, and settles source-local prompt waiters. The public load promise resolves only after those synchronous owners agree. Target metadata and SSE start afterward without a second restore. Source detach is asynchronous, single-attempt, and never blocks the public result or the next restore.
+
+## WebShell ownership
+
+For modern daemons, the main workspace wrapper keeps one provider instance and separates the desired target from the committed target. Workspace resolution and restore failures continue rendering the committed source. A synchronous commit callback advances wrapper ownership before the public promise resolves. Stable failed targets are latched so unrelated renders do not retry them; a controlled failure rolls the host back only while the failed desired generation is still current.
+
+Session transition state gates new prompt and mutation entry points while preserving the source event stream, existing prompt completion, cancellation, permissions, and read-only controls. UI navigation uses an invocation token plus an attachment-identity snapshot so stale completion handlers cannot clear or focus a newer request. Session-owned worktree, branch, git intent, and recap state are not cleared until ownership commits.
+
+## Compatibility and risks
+
+Legacy daemons keep the old keyed/destructive behavior. Cleanup is deliberately best effort: a failed detach can leave an invisible client reference until the existing reaper runs. Staging temporarily holds the source transcript and target replay at once, and CPU-heavy restore work in a shared ACP child can still delay source events. This change does not optimize JSONL reading, selective replay, or daemon capacity.
+
+## Verification
+
+Unit coverage exercises delayed success/failure, exact-target coalescing, latest-only serialization, controlled switching, malformed ownership, write gating, synchronous commit ownership, source events during preparation, wrapper remount compatibility, workspace resolution failure, invocation fencing, and post-commit catch-up timeout behavior. A focused JSDOM/real-daemon test delays delivery of an already-completed target restore response and verifies that the source remains usable until atomic commit; a structured 504 must leave the source intact.

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

@@ -0,0 +1,334 @@
+/**
+ * @license
+ * Copyright 2025 Qwen Team
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import * as fs from 'node:fs';
+import * as path from 'node:path';
+import { fileURLToPath } from 'node:url';
+import { act, createElement } from 'react';
+import type { Root } from 'react-dom/client';
+import { JSDOM } from 'jsdom';
+import { afterAll, afterEach, beforeAll, describe, expect, it } from 'vitest';
+import {
+  DaemonHttpError,
+  type DaemonTranscriptBlock,
+} from '@qwen-code/sdk/daemon';
+import {
+  makeTempWorkspace,
+  spawnDaemon,
+  type SpawnedDaemon,
+} from './_daemon-harness.js';
+
+const __dirname = path.dirname(fileURLToPath(import.meta.url));
+const MOCK_AGENT_PATH = path.resolve(
+  __dirname,
+  '../fixtures/mock-acp-child/agent.mjs',
+);
+
+let activeDaemon: SpawnedDaemon | undefined;
+let root: Root | undefined;
+let dom: JSDOM;
+let createRoot: typeof import('react-dom/client').createRoot;
+let DaemonSessionProvider: typeof import('@qwen-code/webui/daemon-react-sdk').DaemonSessionProvider;
+let useActions: typeof import('@qwen-code/webui/daemon-react-sdk').useActions;
+let useConnection: typeof import('@qwen-code/webui/daemon-react-sdk').useConnection;
+let useTranscriptBlocks: typeof import('@qwen-code/webui/daemon-react-sdk').useTranscriptBlocks;
+const originalGlobalDescriptors = new Map(
+  ['window', 'document', 'navigator', 'IS_REACT_ACT_ENVIRONMENT'].map(
+    (key) => [key, Object.getOwnPropertyDescriptor(globalThis, key)] as const,
+  ),
+);
+
+beforeAll(async () => {
+  dom = new JSDOM('<!doctype html><html><body></body></html>', {
+    url: 'http://localhost',
+  });
+  Object.defineProperty(globalThis, 'window', {
+    configurable: true,
+    value: dom.window,
+  });
+  Object.defineProperty(globalThis, 'document', {
+    configurable: true,
+    value: dom.window.document,
+  });
+  Object.defineProperty(globalThis, 'navigator', {
+    configurable: true,
+    value: dom.window.navigator,
+  });
+  Object.defineProperty(globalThis, 'IS_REACT_ACT_ENVIRONMENT', {
+    configurable: true,
+    value: true,
+  });
+  ({ createRoot } = await import('react-dom/client'));
+  ({ DaemonSessionProvider, useActions, useConnection, useTranscriptBlocks } =
+    await import('@qwen-code/webui/daemon-react-sdk'));
+});
+
+afterAll(() => {
+  dom.window.close();
+  for (const [key, descriptor] of originalGlobalDescriptors) {
+    if (descriptor) Object.defineProperty(globalThis, key, descriptor);
+    else Reflect.deleteProperty(globalThis, key);
+  }
+});
+
+afterEach(async () => {
+  if (root) {
+    await act(async () => root?.unmount());
+    root = undefined;
+  }
+  await activeDaemon?.dispose();
+  activeDaemon = undefined;
+});
+
+function deferred<T = void>() {
+  let resolve!: (value: T | PromiseLike<T>) => void;
+  let reject!: (reason?: unknown) => void;
+  const promise = new Promise<T>((res, rej) => {
+    resolve = res;
+    reject = rej;
+  });
+  return { promise, resolve, reject };
+}
+
+async function waitFor(
+  condition: () => boolean,
+  description: string,
+  timeoutMs = 8_000,
+): Promise<void> {
+  const deadline = Date.now() + timeoutMs;
+  while (Date.now() < deadline) {
+    if (condition()) return;
+    await act(async () => {
+      await new Promise((resolve) => setTimeout(resolve, 20));
+    });
+  }
+  throw new Error(`Timed out waiting for ${description}`);
+}
+
+describe('qwen serve WebUI transactional session switching', () => {
+  async function setup() {
+    const workspace = makeTempWorkspace('webui-session-switching');
+    activeDaemon = await spawnDaemon({
+      workspaceCwd: workspace,
+      env: {
+        QWEN_CLI_ENTRY: MOCK_AGENT_PATH,
+        MOCK_ACP_MODE: 'echo',
+      },
+    });
+    const source = await activeDaemon.client.createOrAttachSession({
+      sessionScope: 'thread',
+    });
+    const resolvedWorkspace = source.workspaceCwd ?? workspace;
+    await activeDaemon.client.prompt(source.sessionId, {
+      prompt: [{ type: 'text', text: 'source transcript' }],
+    });
+    const target = await activeDaemon.client.createOrAttachSession({
+      sessionScope: 'thread',
+    });
+    await activeDaemon.client.prompt(target.sessionId, {
+      prompt: [{ type: 'text', text: 'target transcript' }],
+    });
+    let actions: ReturnType<typeof useActions> | undefined;
+    let connection: ReturnType<typeof useConnection> | undefined;
+    let blocks: readonly DaemonTranscriptBlock[] = [];
+    function Harness() {
+      actions = useActions();
+      connection = useConnection();
+      blocks = useTranscriptBlocks();
+      return null;
+    }
+    const container = document.createElement('div');
+    document.body.appendChild(container);
+    root = createRoot(container);
+    await act(async () => {
+      root?.render(
+        createElement(
+          DaemonSessionProvider,
+          {
+            autoConnect: true,
+            baseUrl: activeDaemon!.base,
+            token: activeDaemon!.token,
+            sessionId: source.sessionId,
+            workspaceCwd: resolvedWorkspace,
+          },
+          createElement(Harness),
+        ),
+      );
+    });
+    await waitFor(
+      () =>
+        connection?.status === 'connected' &&
+        connection.sessionId === source.sessionId &&
+        connection.capabilities?.features.includes('client_identity') ===
+          true &&
+        JSON.stringify(blocks).includes('source transcript'),
+      'source session bootstrap',
+    );
+    return {
+      workspace: resolvedWorkspace,
+      source,
+      target,
+      getActions: () => {
+        if (!actions) throw new Error('session actions unavailable');
+        return actions;
+      },
+      getConnection: () => connection,
+      getBlocks: () => blocks,
+    };
+  }
+
+  it('keeps the source usable until a completed target response is released', async () => {
+    const originalFetch = globalThis.fetch;
+    const state = await setup();
+    const responseReady = deferred();
+    const releaseResponse = deferred();
+    let loadOutcome: Promise<unknown> | undefined;
+    try {
+      globalThis.fetch = async (input, init) => {
+        const request =
+          input instanceof Request ? input : new Request(input, init);
+        const response = await originalFetch(request);
+        if (
+          request.method === 'POST' &&
+          new URL(request.url).pathname.endsWith(
+            `/session/${encodeURIComponent(state.target.sessionId)}/load`,
+          )
+        ) {
+          responseReady.resolve();
+          await releaseResponse.promise;
+        }
+        return response;
+      };
+      act(() => {
+        loadOutcome = state
+          .getActions()
+          .loadSession(state.target.sessionId, {
+            workspaceCwd: state.workspace,
+          })
+          .then(
+            () => undefined,
+            (error: unknown) => error,
+          );
+      });
+      await responseReady.promise;
+
+      expect(state.getConnection()).toMatchObject({
+        status: 'connected',
+        sessionId: state.source.sessionId,
+        sessionTransition: { phase: 'preparing' },
+      });
+      await expect(state.getActions().cancel()).resolves.toBeUndefined();
+      await activeDaemon!.client.prompt(state.source.sessionId, {
+        prompt: [{ type: 'text', text: 'source remains live' }],
+      });
+      await waitFor(
+        () => JSON.stringify(state.getBlocks()).includes('source remains live'),
+        'source event while target response is held',
+      );
+
+      await act(async () => {
+        releaseResponse.resolve();
+        expect(await loadOutcome).toBeUndefined();
+      });
+      expect(state.getConnection()).toMatchObject({
+        status: 'connected',
+        sessionId: state.target.sessionId,
+      });
+      expect(JSON.stringify(state.getBlocks())).toContain('target transcript');
+      expect(JSON.stringify(state.getBlocks())).not.toContain(
+        'source remains live',
+      );
+    } finally {
+      globalThis.fetch = originalFetch;
+      releaseResponse.resolve();
+      await loadOutcome?.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();
+    try {
+      globalThis.fetch = async (input, init) => {
+        const request =
+          input instanceof Request ? input : new Request(input, init);
+        if (
+          request.method === 'POST' &&
+          new URL(request.url).pathname.endsWith(
+            `/session/${encodeURIComponent(state.target.sessionId)}/load`,
+          )
+        ) {
+          return new Response(
+            JSON.stringify({
+              code: 'session_restore_timeout',
+              error: 'Session restore timed out',
+              retryable: true,
+            }),
+            {
+              status: 504,
+              headers: {
+                'Content-Type': 'application/json',
+                'Retry-After': '5',
+              },
+            },
+          );
+        }
+        return originalFetch(request);
+      };
+      let restoreError: unknown;
+      await act(async () => {
+        try {
+          await state.getActions().loadSession(state.target.sessionId, {
+            workspaceCwd: state.workspace,
+          });
+        } catch (error) {
+          restoreError = error;
+        }
+      });
+      expect(restoreError).toBeInstanceOf(DaemonHttpError);
+      expect(restoreError).toMatchObject({
+        status: 504,
+        body: {
+          code: 'session_restore_timeout',
+          retryable: true,
+        },
+      });
+      expect(state.getConnection()).toMatchObject({
+        status: 'connected',
+        sessionId: state.source.sessionId,
+        sessionTransition: {
+          phase: 'failed',
+          error: { code: 'session_restore_timeout', status: 504 },
+        },
+      });
+      expect(JSON.stringify(state.getBlocks())).toContain('source transcript');
+      await activeDaemon!.client.prompt(state.source.sessionId, {
+        prompt: [{ type: 'text', text: 'source after timeout' }],
+      });
+      await waitFor(
+        () =>
+          JSON.stringify(state.getBlocks()).includes('source after timeout'),
+        'source event after target timeout',
+      );
+    } finally {
+      globalThis.fetch = originalFetch;
+      if (root) {
+        await act(async () => root?.unmount());
+        root = undefined;
+      }
+      await activeDaemon?.dispose();
+      activeDaemon = undefined;
+      fs.rmSync(state.workspace, { recursive: true, force: true });
+    }
+  }, 30_000);
+});