fix(webui): Make same-session refresh transactional

docs/design/2026-08-11-transactional-same-session-refresh.md

@@ -0,0 +1,47 @@
+# Transactional same-session refresh
+
+## Problem
+
+Cross-session restore is transactional, but refreshing the current logical session still used the legacy handoff: it stopped the source event runner and could detach or clear the source before `load` or `resume` settled. A slow, failed, partial, or stale refresh could therefore interrupt an otherwise healthy transcript, prompt, and attachment. Changing an explicit client ID had the same problem.
+
+## Scope
+
+This change covers `loadSession`, ordinary or configured `reloadSession`, `resumeSession`, and explicit non-empty client-ID replacement when the normalized `(sessionId, workspaceCwd)` remains unchanged. It reuses the provider-local restore coordinator introduced for cross-session switching. Epoch or ring resync, memory repair, branch adoption, selective JSONL reading, and daemon-side resource scheduling remain separate work.
+
+Modern transactional behavior requires a successful capability snapshot advertising `client_identity` and concrete source and candidate client IDs. A daemon that explicitly lacks the feature retains the legacy destructive path. Unknown capabilities, incomplete modern responses, missing cursor or epoch state, and malformed ownership fail closed and preserve the source. An explicit client ID changing to `undefined` keeps the current attachment.
+
+## Scheduling and request identity
+
+Restore identity includes the normalized session and workspace, the effective replay shape (`load/all`, `load/recent(N)`, or `resume/none`), and the requested client ID. Only identical signal-free requests coalesce. Different same-session intents are latest-wins, while a cross-session target supersedes a refresh and a pending cross-session target cannot be silently cancelled by reloading its source. One ordinary restore RPC runs at a time; a compatible same-shape retry may adopt a late result, while every stale result is detached once on a best-effort basis.
+
+A same-session request waits for the source runner to be ready and free of local, restored, or observed work before it starts. This wait does not consume the restore budget. The budget starts with the raw RPC, and signal, lifecycle, navigation, resync, or environment changes can still cancel the intent. Resync remains authoritative and continues through its existing destructive recovery path for this change.
+
+Source-bound branch, create, attach, and legacy restore operations exclude ordinary restores. The exclusion follows the raw operation rather than an outer action timeout: a timed-out create keeps restores blocked until its raw request settles, and a late successful create is detached once. A blocked controlled target is still routed through the coordinator so it publishes a terminal failed transition and does not leave the host waiting indefinitely.
+
+## Cursor capture and integrity
+
+The source runner tracks a processed cursor separately from the SDK read cursor. It advances the processed cursor only after transcript normalization, notices, side channels, workspace signals, prompt settlement, and connection side effects for an event have completed.
+
+When a full load starts, the runner captures the exact source object, client ID, event epoch, and processed cursor. Subsequent raw event references are retained only while their IDs are contiguous and increasing. The capture is bounded by the configured event queue and 8 MiB of serialized UTF-8 data; id-less non-sentinel frames, gaps, serialization failures, overflow, epoch changes, or in-place source client-ID changes invalidate the candidate.
+
+A load candidate must carry both replay arrays, a matching epoch, a valid watermark at or after capture start, complete non-degraded replay, and no partial-replay diagnostic. A resume candidate must carry a matching epoch and valid watermark. If the candidate watermark is ahead, the source remains live until the processed cursor catches up. A candidate claiming active prompt work cannot commit until the source processes a later terminal or cancellation and no runner-owned turn remains.
+
+## Staging and commit
+
+Full-load replay is normalized into an unsubscribed shadow store in batches of at most 512 events. Replay arrays are traversed directly rather than concatenated. At commit, the bounded source tail after the candidate watermark and through the final processed cursor is applied to the shadow store. Staging does not publish notices, side channels, workspace signals, prompt state, transcript, history, or connection updates; malformed or repair-requiring replay invalidates the candidate.
+
+One synchronous commit rechecks the desired intent, lifecycle and environment, exact source object and client ID, epoch, deadline, runner readiness, turn state, and processed cursor. It then flushes and stops the source runner, installs the candidate attachment and connection, and either replaces the visible replay page for `load` or preserves the existing transcript for `resume`. Resume creates a new history owner so stale pagination cannot write through. The candidate cursor is advanced to the source's final processed cursor before its metadata and SSE runner starts. The public promise resolves only after visible owners agree; source detach happens afterward and never blocks the result.
+
+Same-session notices and settled-prompt bookkeeping are preserved. Candidate replay and captured tail side effects are not republished because the source already processed them through the final cursor. Connection metadata is based on the connection current at commit and refreshed by the new runner, avoiding rollback to metadata captured when the request began.
+
+## Client-ID reconciliation and failure behavior
+
+Raw `clientId` props are desired input rather than committed owner state. A modern explicit client-ID change performs transactional resume. A change while another target is preparing updates that target rather than rebinding the source. Legacy daemons use a full destructive load so the transcript is not replaced by an empty resume replay.
+
+The commit CAS includes the source object's current client ID. If SDK prompt-admission self-heal updates that ID in place, the prepared candidate is discarded and the healed source remains active. Failures publish one recoverable transition failure while leaving source connection, transcript, prompt, metadata, and controls usable; they never rewrite the source as missing or disconnected.
+
+## Verification and risks
+
+Unit coverage checks delayed success and failure, local and observer prompt gating, response completeness, partial or degraded replay, epoch and tail gaps, cursor catch-up, client-ID rebind, in-place self-heal, late cleanup, and cross-session arbitration. SDK tests cover epoch and replay-integrity propagation. A real-daemon JSDOM test withholds an already-completed same-session load response, sends live source work during the hold, and verifies atomic replay-plus-tail commit without loss or duplication; structured timeout and client-ID rebind paths verify source preservation and transcript continuity.
+
+Staging temporarily retains the visible transcript, the candidate replay, and up to 8 MiB of source tail. CPU-heavy restore in the same ACP child may still delay source events. Detach is deliberately single-attempt and best effort, so a failed cleanup can leave an invisible client reference until the existing reaper runs.

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

@@ -0,0 +1,323 @@
+/**
+ * @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 activeWorkspace: string | 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;
+  if (activeWorkspace) {
+    fs.rmSync(activeWorkspace, { recursive: true, force: true });
+    activeWorkspace = undefined;
+  }
+});
+
+function deferred() {
+  let resolve!: () => void;
+  const promise = new Promise<void>((res) => {
+    resolve = res;
+  });
+  return { promise, resolve };
+}
+
+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 same-session refresh', () => {
+  async function setup() {
+    const workspace = makeTempWorkspace('webui-same-session-refresh');
+    activeWorkspace = workspace;
+    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' }],
+    });
+    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);
+    const render = async (clientId?: string) => {
+      await act(async () => {
+        root?.render(
+          createElement(
+            DaemonSessionProvider,
+            {
+              autoConnect: true,
+              baseUrl: activeDaemon!.base,
+              token: activeDaemon!.token,
+              sessionId: source.sessionId,
+              workspaceCwd: resolvedWorkspace,
+              ...(clientId ? { clientId } : {}),
+            },
+            createElement(Harness),
+          ),
+        );
+      });
+    };
+    await render();
+    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,
+      render,
+      getActions: () => {
+        if (!actions) throw new Error('session actions unavailable');
+        return actions;
+      },
+      getConnection: () => connection,
+      getBlocks: () => blocks,
+    };
+  }
+
+  it('merges the live source tail before committing a held load response', async () => {
+    const originalFetch = globalThis.fetch;
+    const state = await setup();
+    const responseReady = deferred();
+    const releaseResponse = deferred();
+    let refresh: Promise<void> | 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.source.sessionId)}/load`,
+          )
+        ) {
+          responseReady.resolve();
+          await releaseResponse.promise;
+        }
+        return response;
+      };
+      act(() => {
+        refresh = state.getActions().loadSession(state.source.sessionId, {
+          workspaceCwd: state.workspace,
+        });
+      });
+      await responseReady.promise;
+      expect(state.getConnection()).toMatchObject({
+        status: 'connected',
+        sessionId: state.source.sessionId,
+        sessionTransition: { phase: 'preparing' },
+      });
+
+      await activeDaemon!.client.prompt(state.source.sessionId, {
+        prompt: [{ type: 'text', text: 'live during refresh' }],
+      });
+      await waitFor(
+        () => JSON.stringify(state.getBlocks()).includes('live during refresh'),
+        'live source tail',
+      );
+      await act(async () => {
+        releaseResponse.resolve();
+        await refresh;
+      });
+
+      const transcript = JSON.stringify(state.getBlocks());
+      expect(transcript).toContain('source transcript');
+      expect(transcript.match(/live during refresh/g)).toHaveLength(1);
+      expect(state.getConnection()).toMatchObject({
+        status: 'connected',
+        sessionId: state.source.sessionId,
+      });
+    } finally {
+      globalThis.fetch = originalFetch;
+      releaseResponse.resolve();
+      await refresh?.catch(() => undefined);
+    }
+  }, 30_000);
+
+  it('preserves the source after a structured same-session timeout', async () => {
+    const originalFetch = globalThis.fetch;
+    const state = await setup();
+    const sourceClientId = state.getConnection()?.clientId;
+    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.source.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' },
+            },
+          );
+        }
+        return originalFetch(request);
+      };
+      let restoreError: unknown;
+      await act(async () => {
+        try {
+          await state.getActions().reloadSession(new AbortController().signal);
+        } 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,
+        clientId: sourceClientId,
+        sessionTransition: {
+          phase: 'failed',
+          error: { code: 'session_restore_timeout', status: 504 },
+        },
+      });
+      expect(JSON.stringify(state.getBlocks())).toContain('source transcript');
+    } finally {
+      globalThis.fetch = originalFetch;
+    }
+  }, 30_000);
+
+  it('preserves the transcript across a controlled clientId rebind', async () => {
+    const state = await setup();
+    const rebound = await activeDaemon!.client.resumeSession(
+      state.source.sessionId,
+      { workspaceCwd: state.workspace },
+    );
+    const reboundClientId = rebound.clientId;
+    expect(reboundClientId).toBeTruthy();
+    await state.render(reboundClientId!);
+    await waitFor(
+      () => state.getConnection()?.clientId === reboundClientId,
+      'clientId rebind',
+    );
+    expect(JSON.stringify(state.getBlocks())).toContain('source transcript');
+    await act(async () => {
+      await state.getActions().sendPrompt('prompt after rebind');
+    });
+    await waitFor(
+      () => JSON.stringify(state.getBlocks()).includes('prompt after rebind'),
+      'prompt after clientId rebind',
+    );
+  }, 30_000);
+});