diff --git a/docs/design/2026-08-11-transactional-same-session-refresh.md b/docs/design/2026-08-11-transactional-same-session-refresh.md new file mode 100644 index 00000000000..6dda31705f0 --- /dev/null +++ b/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. diff --git a/integration-tests/cli/qwen-serve-webui-same-session-refresh.test.ts b/integration-tests/cli/qwen-serve-webui-same-session-refresh.test.ts new file mode 100644 index 00000000000..080f6c6b457 --- /dev/null +++ b/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('', { + 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((res) => { + resolve = res; + }); + return { promise, resolve }; +} + +async function waitFor( + condition: () => boolean, + description: string, + timeoutMs = 8_000, +): Promise { + 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 | undefined; + let connection: ReturnType | 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 | 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); +}); diff --git a/packages/sdk-typescript/src/daemon/DaemonSessionClient.ts b/packages/sdk-typescript/src/daemon/DaemonSessionClient.ts index 0b57cf601e6..6cdba2a431a 100644 --- a/packages/sdk-typescript/src/daemon/DaemonSessionClient.ts +++ b/packages/sdk-typescript/src/daemon/DaemonSessionClient.ts @@ -82,6 +82,12 @@ export interface DaemonSessionClientOptions { eventEpoch?: string; /** Compacted replay snapshot from daemon load response. */ replaySnapshot?: DaemonReplaySnapshot; + /** True when the load response explicitly carried both replay arrays. */ + replaySnapshotComplete?: boolean; + /** True when persisted replay was only partially reconstructed. */ + replayPartial?: boolean; + /** Diagnostic for a partial persisted replay. */ + replayError?: string; /** True when older persisted records precede the replay snapshot. */ historyHasMore?: boolean; /** @@ -138,6 +144,9 @@ export class DaemonSessionClient { readonly session: DaemonSession; readonly state: DaemonSessionState; readonly replaySnapshot: DaemonReplaySnapshot; + readonly replaySnapshotComplete: boolean; + readonly replayPartial: boolean; + readonly replayError: string | undefined; readonly hasActivePrompt: boolean; readonly historyHasMore: boolean; /** @@ -188,6 +197,9 @@ export class DaemonSessionClient { compactedReplay: [], liveJournal: [], }; + this.replaySnapshotComplete = opts.replaySnapshotComplete ?? false; + this.replayPartial = opts.replayPartial ?? false; + this.replayError = opts.replayError; this.lastSeenEventId = validateLastEventId(opts.lastEventId); this.lastSeenEpoch = opts.eventEpoch; this.promptLimit = @@ -259,6 +271,10 @@ export class DaemonSessionClient { req: RestoreSessionRequest = {}, clientId?: string, ): Promise { + const restored = await client.loadSession(sessionId, req, clientId); + const replaySnapshotComplete = + Array.isArray(restored.compactedReplay) && + Array.isArray(restored.liveJournal); const { state, hasActivePrompt, @@ -267,10 +283,12 @@ export class DaemonSessionClient { historyHasMore, historyAnchorRecordId, replayDegraded, + partial, + replayError, lastEventId: serverLastEventId, eventEpoch, ...session - } = await client.loadSession(sessionId, req, clientId); + } = restored; return new DaemonSessionClient({ client, session, @@ -282,6 +300,9 @@ export class DaemonSessionClient { compactedReplay: compactedReplay ?? [], liveJournal: liveJournal ?? [], }, + replaySnapshotComplete, + replayPartial: partial === true, + replayError, historyHasMore, historyAnchorRecordId, replayDegraded, @@ -345,6 +366,10 @@ export class DaemonSessionClient { return this.lastSeenEventId; } + get eventEpoch(): string | undefined { + return this.lastSeenEpoch; + } + setLastEventId(lastEventId: number | undefined): void { this.lastSeenEventId = validateLastEventId(lastEventId); } diff --git a/packages/sdk-typescript/src/daemon/types.ts b/packages/sdk-typescript/src/daemon/types.ts index 02758b2e40b..f78d3d0ccb5 100644 --- a/packages/sdk-typescript/src/daemon/types.ts +++ b/packages/sdk-typescript/src/daemon/types.ts @@ -930,6 +930,10 @@ export interface DaemonSessionState { export interface DaemonRestoredSession extends DaemonSession { state: DaemonSessionState; artifactWarnings?: string[]; + /** True when persisted replay could only be reconstructed partially. */ + partial?: true; + /** Diagnostic for a partial persisted replay. */ + replayError?: string; /** Compacted events for completed turns (load only). */ compactedReplay?: DaemonEvent[]; /** Bounded replay events for the current incomplete turn (load only). */ diff --git a/packages/sdk-typescript/test/unit/DaemonSessionClient.test.ts b/packages/sdk-typescript/test/unit/DaemonSessionClient.test.ts index 7ad56080180..41b99b9484a 100644 --- a/packages/sdk-typescript/test/unit/DaemonSessionClient.test.ts +++ b/packages/sdk-typescript/test/unit/DaemonSessionClient.test.ts @@ -300,6 +300,7 @@ describe('DaemonSessionClient', () => { state: { configOptions: [] }, hasActivePrompt: true, lastEventId: 42, + eventEpoch: 'epoch-42', compactedReplay: [{ id: 1, v: 1, type: 'session_update', data: {} }], liveJournal: [{ id: 42, v: 1, type: 'session_update', data: {} }], }); @@ -319,6 +320,10 @@ describe('DaemonSessionClient', () => { expect(session.clientId).toBe('client-1'); expect(session.hasActivePrompt).toBe(true); expect(session.state).toEqual({ configOptions: [] }); + expect(session.eventEpoch).toBe('epoch-42'); + expect(session.replaySnapshotComplete).toBe(true); + expect(session.replayPartial).toBe(false); + expect(session.replayError).toBeUndefined(); expect(session.replaySnapshot.compactedReplay).toHaveLength(1); expect(session.replaySnapshot.liveJournal).toHaveLength(1); expect(JSON.parse(calls[0]!.body!)).toEqual({ cwd: '/work/a' }); @@ -381,6 +386,31 @@ describe('DaemonSessionClient', () => { expect(session.replayDegraded).toBe(true); }); + it('reports incomplete and partial load replay snapshots', async () => { + const { fetch } = recordingFetch((req) => { + if (req.url.endsWith('/session/s-1/load')) { + return jsonResponse(200, { + sessionId: 's-1', + workspaceCwd: '/work/a', + attached: true, + clientId: 'client-1', + state: {}, + compactedReplay: [], + partial: true, + replayError: 'journal read failed', + }); + } + return jsonResponse(500, { error: `unexpected ${req.url}` }); + }); + const client = new DaemonClient({ baseUrl: 'http://daemon', fetch }); + + const session = await DaemonSessionClient.load(client, 's-1'); + + expect(session.replaySnapshotComplete).toBe(false); + expect(session.replayPartial).toBe(true); + expect(session.replayError).toBe('journal read failed'); + }); + it('resumes an existing daemon session using server watermark', async () => { const { fetch, calls } = recordingFetch((req) => { if (req.url.endsWith('/session/s-1/resume')) { @@ -409,6 +439,7 @@ describe('DaemonSessionClient', () => { expect(session.state).toEqual({ modes: null }); expect(session.replaySnapshot.compactedReplay).toHaveLength(0); expect(session.replaySnapshot.liveJournal).toHaveLength(0); + expect(session.replaySnapshotComplete).toBe(false); for await (const _event of session.events()) { /* empty */ } diff --git a/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx b/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx index f3cf96d7811..e1947bd7666 100644 --- a/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx +++ b/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx @@ -64,6 +64,10 @@ interface MockSession { historyHasMore?: boolean; historyAnchorRecordId?: string; replayDegraded?: boolean; + replaySnapshotComplete?: boolean; + replayPartial?: boolean; + replayError?: string; + eventEpoch?: string; client?: MockClient; lastEventId?: number; setLastEventId: (lastEventId: number | undefined) => void; @@ -10907,6 +10911,68 @@ describe('DaemonSessionProvider', () => { expect(connection?.sessionTransition).toBeUndefined(); }); + it('publishes a controlled failure while session creation is in flight', async () => { + vi.spyOn(console, 'warn').mockImplementation(() => undefined); + sdkMocks.capabilities.mockResolvedValue({ + workspaceCwd: '/mock-workspace', + features: ['client_identity'], + }); + const source = createMockSession({ + sessionId: 'session-a', + clientId: 'client-a', + }); + sdkMocks.sessions.push(source); + const created = createDeferred(); + let actions: DaemonSessionActions | undefined; + let connection: DaemonConnectionState | undefined; + + function Harness() { + actions = useDaemonActions(); + connection = useDaemonConnection(); + return null; + } + + await renderWithProvider(, { + autoConnect: true, + sessionId: 'session-a', + }); + source.client!.createOrAttachSession = vi.fn(() => created.promise); + sdkMocks.MockDaemonSessionClient.load.mockClear(); + let create!: Promise; + act(() => { + create = requireActions(actions).createSession(); + root?.render( + + + , + ); + }); + await act(async () => flushPromises()); + + expect(connection).toMatchObject({ + status: 'connected', + sessionId: 'session-a', + clientId: 'client-a', + sessionTransition: { + phase: 'failed', + origin: 'controlled', + targetSessionId: 'session-b', + }, + }); + expect(sdkMocks.MockDaemonSessionClient.load).not.toHaveBeenCalled(); + + created.resolve( + createMockSession({ sessionId: 'created-session', clientId: 'created' }), + ); + await expect(create).resolves.toMatchObject({ + sessionId: 'created-session', + }); + }); + it('cancels a pending controlled target when props return to A', async () => { const detachFetch = vi.fn( async (_input: RequestInfo | URL, _init?: RequestInit) => @@ -14327,6 +14393,1231 @@ describe('DaemonSessionProvider', () => { expect(blocks).toMatchObject([{ kind: 'user', text: 'recent prompt' }]); }); + it.each(['load', 'reload'] as const)( + 'keeps the current attachment live while a same-session %s fails', + async (mode) => { + vi.stubGlobal( + 'fetch', + vi.fn(async () => new Response(null, { status: 204 })), + ); + sdkMocks.capabilities.mockResolvedValue({ + workspaceCwd: '/mock-workspace', + features: ['client_identity'], + }); + const sourceReady = createDeferred(); + const sourceEvent = createDeferred(); + let sourceSignal: AbortSignal | undefined; + let subscriptions = 0; + const source = createMockSession({ + sessionId: 'session-a', + clientId: 'client-a', + lastEventId: 2, + replaySnapshot: createTextReplaySnapshot('A transcript'), + events: async function* sourceEvents( + opts: { signal?: AbortSignal } = {}, + ) { + subscriptions += 1; + sourceSignal = opts.signal; + yield { v: 1, type: 'replay_complete', data: {} }; + sourceReady.resolve(); + await Promise.race([ + sourceEvent.promise, + new Promise((resolve) => + opts.signal?.addEventListener('abort', () => resolve(), { + once: true, + }), + ), + ]); + if (opts.signal?.aborted) return; + yield { + id: 3, + v: 1, + type: 'session_update', + data: { + update: { + sessionUpdate: 'agent_message_chunk', + content: { type: 'text', text: ' still live' }, + }, + }, + }; + await new Promise((resolve) => + opts.signal?.addEventListener('abort', () => resolve(), { + once: true, + }), + ); + }, + }); + sdkMocks.sessions.push(source); + const target = createDeferred(); + let actions: DaemonSessionActions | undefined; + let blocks: readonly DaemonTranscriptBlock[] = []; + let connection: DaemonConnectionState | undefined; + + function Harness() { + actions = useDaemonActions(); + blocks = useDaemonTranscriptBlocks(); + connection = useDaemonConnection(); + return null; + } + + await renderWithProvider(, { autoConnect: true }); + await sourceReady.promise; + sdkMocks.MockDaemonSessionClient.load.mockImplementationOnce( + async () => target.promise, + ); + let refresh!: Promise; + act(() => { + refresh = + mode === 'load' + ? requireActions(actions).loadSession('session-a') + : requireActions(actions).reloadSession( + new AbortController().signal, + ); + }); + const outcome = refresh.catch((error: unknown) => error); + await act(async () => flushPromises()); + + expect(sourceSignal?.aborted).toBe(false); + expect(source.detach).not.toHaveBeenCalled(); + expect(connection).toMatchObject({ + status: 'connected', + sessionId: 'session-a', + clientId: 'client-a', + }); + await act(async () => { + sourceEvent.resolve(); + await flushPromises(); + await flushTranscriptDispatch(); + }); + expect( + blocks.map((block) => ('text' in block ? block.text : undefined)), + ).toEqual(['A transcript', ' still live']); + + await act(async () => { + target.reject(new Error(`${mode} refresh failed`)); + await flushPromises(); + }); + await expect(outcome).resolves.toMatchObject({ + message: `${mode} refresh failed`, + }); + expect(sourceSignal?.aborted).toBe(false); + expect(subscriptions).toBe(1); + expect(connection).toMatchObject({ + status: 'connected', + sessionId: 'session-a', + clientId: 'client-a', + }); + }, + ); + + it('waits for an admitted prompt before starting a same-session refresh', async () => { + vi.stubGlobal( + 'fetch', + vi.fn(async () => new Response(null, { status: 204 })), + ); + sdkMocks.capabilities.mockResolvedValue({ + workspaceCwd: '/mock-workspace', + features: ['client_identity'], + }); + const terminal = createDeferred(); + const source = createMockSession({ + sessionId: 'session-a', + clientId: 'client-a', + lastEventId: 0, + submitPrompt: vi.fn(async () => ({ + promptId: 'prompt-a', + lastEventId: 0, + })), + events: async function* sourceEvents() { + yield { v: 1, type: 'replay_complete', data: {} }; + await terminal.promise; + yield { + id: 1, + v: 1, + type: 'turn_complete', + data: { promptId: 'prompt-a', stopReason: 'end_turn' }, + }; + }, + }); + sdkMocks.sessions.push(source); + const target = createDeferred(); + let actions: DaemonSessionActions | undefined; + + function Harness() { + actions = useDaemonActions(); + return null; + } + + await renderWithProvider(, { autoConnect: true }); + let prompt!: Promise; + act(() => { + prompt = requireActions(actions).sendPrompt('active prompt'); + }); + await act(async () => flushPromises()); + sdkMocks.MockDaemonSessionClient.load.mockImplementationOnce( + async () => target.promise, + ); + let refresh!: Promise; + act(() => { + refresh = requireActions(actions).loadSession('session-a'); + }); + const refreshOutcome = refresh.catch((error: unknown) => error); + await act(async () => flushPromises()); + expect(sdkMocks.MockDaemonSessionClient.load).toHaveBeenCalledTimes(1); + + await act(async () => { + terminal.resolve(); + await prompt; + await flushPromises(); + }); + expect(sdkMocks.MockDaemonSessionClient.load).toHaveBeenCalledTimes(2); + await act(async () => { + target.reject(new Error('refresh failed')); + await flushPromises(); + }); + await expect(refreshOutcome).resolves.toMatchObject({ + message: 'refresh failed', + }); + await expect(prompt).resolves.toMatchObject({ stopReason: 'end_turn' }); + }); + + it('starts a queued refresh after prompt admission fails without a terminal event', async () => { + vi.stubGlobal( + 'fetch', + vi.fn(async () => new Response(null, { status: 204 })), + ); + sdkMocks.capabilities.mockResolvedValue({ + workspaceCwd: '/mock-workspace', + features: ['client_identity'], + }); + const admission = createDeferred<{ + promptId: string; + lastEventId: number; + }>(); + const source = createMockSession({ + sessionId: 'session-a', + clientId: 'client-a', + lastEventId: 2, + replaySnapshot: createTextReplaySnapshot('A transcript'), + submitPrompt: vi.fn(async () => admission.promise), + events: async function* sourceEvents(opts = {}) { + yield { v: 1, type: 'replay_complete', data: {} }; + await new Promise((resolve) => + opts.signal?.addEventListener('abort', () => resolve(), { + once: true, + }), + ); + }, + }); + sdkMocks.sessions.push(source); + let actions: DaemonSessionActions | undefined; + + function Harness() { + actions = useDaemonActions(); + return null; + } + + await renderWithProvider(, { autoConnect: true }); + sdkMocks.MockDaemonSessionClient.load.mockClear(); + sdkMocks.MockDaemonSessionClient.load.mockResolvedValueOnce( + createMockSession({ + sessionId: 'session-a', + clientId: 'client-a', + lastEventId: 2, + replaySnapshot: createTextReplaySnapshot('refreshed transcript'), + }), + ); + let promptOutcome!: Promise; + act(() => { + promptOutcome = requireActions(actions) + .sendPrompt('prompt that fails admission') + .catch((error: unknown) => error); + }); + await act(async () => flushPromises()); + let refresh!: Promise; + act(() => { + refresh = requireActions(actions).loadSession('session-a'); + }); + await act(async () => flushPromises()); + expect(sdkMocks.MockDaemonSessionClient.load).not.toHaveBeenCalled(); + + await act(async () => { + admission.reject(new Error('admission failed')); + await promptOutcome; + await flushPromises(); + }); + await expect(promptOutcome).resolves.toMatchObject({ + message: 'admission failed', + }); + expect(sdkMocks.MockDaemonSessionClient.load).toHaveBeenCalledOnce(); + await act(async () => { + await refresh; + await flushPromises(); + }); + }); + + it('waits for an in-flight shell command before refreshing', async () => { + vi.stubGlobal( + 'fetch', + vi.fn(async () => new Response(null, { status: 204 })), + ); + sdkMocks.capabilities.mockResolvedValue({ + workspaceCwd: '/mock-workspace', + features: ['client_identity'], + }); + const shellResult = createDeferred(); + const source = createMockSession({ + sessionId: 'session-a', + clientId: 'client-a', + lastEventId: 2, + replaySnapshot: createTextReplaySnapshot('A transcript'), + shellCommand: vi.fn(async () => shellResult.promise), + events: async function* sourceEvents(opts = {}) { + yield { v: 1, type: 'replay_complete', data: {} }; + await new Promise((resolve) => + opts.signal?.addEventListener('abort', () => resolve(), { + once: true, + }), + ); + }, + }); + sdkMocks.sessions.push(source); + let actions: DaemonSessionActions | undefined; + + function Harness() { + actions = useDaemonActions(); + return null; + } + + await renderWithProvider(, { autoConnect: true }); + sdkMocks.MockDaemonSessionClient.load.mockClear(); + sdkMocks.MockDaemonSessionClient.load.mockResolvedValueOnce( + createMockSession({ + sessionId: 'session-a', + clientId: 'client-a', + lastEventId: 2, + replaySnapshot: createTextReplaySnapshot('refreshed transcript'), + }), + ); + let shell!: Promise; + act(() => { + shell = requireActions(actions).sendShellCommand('echo held'); + }); + await act(async () => flushPromises()); + let refresh!: Promise; + act(() => { + refresh = requireActions(actions).loadSession('session-a'); + }); + await act(async () => flushPromises()); + expect(sdkMocks.MockDaemonSessionClient.load).not.toHaveBeenCalled(); + + await act(async () => { + shellResult.resolve(undefined); + await shell; + await flushPromises(); + }); + expect(sdkMocks.MockDaemonSessionClient.load).toHaveBeenCalledOnce(); + await act(async () => { + await refresh; + await flushPromises(); + }); + }); + + it('keeps the transcript while a same-session resume fails', async () => { + vi.stubGlobal( + 'fetch', + vi.fn(async () => new Response(null, { status: 204 })), + ); + sdkMocks.capabilities.mockResolvedValue({ + workspaceCwd: '/mock-workspace', + features: ['client_identity'], + }); + const source = createMockSession({ + sessionId: 'session-a', + clientId: 'client-a', + lastEventId: 2, + replaySnapshot: createTextReplaySnapshot('A transcript'), + events: async function* sourceEvents(opts = {}) { + yield { v: 1, type: 'replay_complete', data: {} }; + await new Promise((resolve) => + opts.signal?.addEventListener('abort', () => resolve(), { + once: true, + }), + ); + }, + }); + sdkMocks.sessions.push(source); + const target = createDeferred(); + let actions: DaemonSessionActions | undefined; + let blocks: readonly DaemonTranscriptBlock[] = []; + let connection: DaemonConnectionState | undefined; + + function Harness() { + actions = useDaemonActions(); + blocks = useDaemonTranscriptBlocks(); + connection = useDaemonConnection(); + return null; + } + + await renderWithProvider(, { autoConnect: true }); + sdkMocks.MockDaemonSessionClient.resume.mockImplementationOnce( + async () => target.promise, + ); + let resume!: Promise; + act(() => { + resume = requireActions(actions).resumeSession('session-a'); + }); + const outcome = resume.catch((error: unknown) => error); + await act(async () => flushPromises()); + expect(blocks).toMatchObject([{ text: 'A transcript' }]); + expect(source.detach).not.toHaveBeenCalled(); + + await act(async () => { + target.reject(new Error('resume failed')); + await flushPromises(); + }); + await expect(outcome).resolves.toMatchObject({ message: 'resume failed' }); + expect(blocks).toMatchObject([{ text: 'A transcript' }]); + expect(connection).toMatchObject({ + status: 'connected', + sessionId: 'session-a', + clientId: 'client-a', + }); + }); + + it('commits a same-session load only after the source terminal tail is processed', async () => { + const detachFetch = vi.fn( + async (_input: RequestInfo | URL, _init?: RequestInit) => + new Response(null, { status: 204 }), + ); + vi.stubGlobal('fetch', detachFetch); + sdkMocks.capabilities.mockResolvedValue({ + workspaceCwd: '/mock-workspace', + features: ['client_identity'], + }); + const terminal = createDeferred(); + let sourceSignal: AbortSignal | undefined; + const source = createMockSession({ + sessionId: 'session-a', + clientId: 'client-a', + lastEventId: 2, + replaySnapshot: createTextReplaySnapshot('A transcript'), + events: async function* sourceEvents(opts = {}) { + sourceSignal = opts.signal; + yield { v: 1, type: 'replay_complete', data: {} }; + await terminal.promise; + yield { + id: 3, + v: 1, + type: 'model_switched', + data: { modelId: 'source-tail-model' }, + }; + yield { + id: 4, + v: 1, + type: 'turn_complete', + data: { promptId: 'remote-prompt', stopReason: 'end_turn' }, + }; + await new Promise((resolve) => + opts.signal?.addEventListener('abort', () => resolve(), { + once: true, + }), + ); + }, + }); + sdkMocks.sessions.push(source); + const candidate = createMockSession({ + sessionId: 'session-a', + clientId: 'client-a', + eventEpoch: 'epoch-1', + lastEventId: 2, + hasActivePrompt: true, + replaySnapshot: createTextReplaySnapshot('refreshed transcript'), + }); + let actions: DaemonSessionActions | undefined; + let blocks: readonly DaemonTranscriptBlock[] = []; + let connection: DaemonConnectionState | undefined; + + function Harness() { + actions = useDaemonActions(); + blocks = useDaemonTranscriptBlocks(); + connection = useDaemonConnection(); + return null; + } + + await renderWithProvider(, { autoConnect: true }); + sdkMocks.MockDaemonSessionClient.load.mockResolvedValueOnce(candidate); + let refresh!: Promise; + act(() => { + refresh = requireActions(actions).reloadSession( + new AbortController().signal, + ); + }); + let settled = false; + void refresh.then(() => { + settled = true; + }); + await act(async () => flushPromises()); + expect(settled).toBe(false); + expect(sourceSignal?.aborted).toBe(false); + expect(blocks).toMatchObject([{ text: 'A transcript' }]); + + await act(async () => { + terminal.resolve(); + await refresh; + await flushPromises(); + }); + expect( + blocks.map((block) => ('text' in block ? block.text : undefined)), + ).toContain('refreshed transcript'); + expect(sourceSignal?.aborted).toBe(true); + expect(candidate.setLastEventId).toHaveBeenLastCalledWith(4); + expect(connection?.currentModel).toBe('source-tail-model'); + const [url, init] = detachFetch.mock.calls[0] ?? []; + expect(String(url)).toContain('/session/session-a/detach'); + expect(new Headers(init?.headers).get('X-Qwen-Client-Id')).toBe('client-a'); + }); + + it('retires a prepared same-session candidate when a cross-session target supersedes it', async () => { + const detachFetch = vi.fn( + async (_input: RequestInfo | URL, _init?: RequestInit) => + new Response(null, { status: 204 }), + ); + vi.stubGlobal('fetch', detachFetch); + sdkMocks.capabilities.mockResolvedValue({ + workspaceCwd: '/mock-workspace', + features: ['client_identity'], + }); + const source = createMockSession({ + sessionId: 'session-a', + clientId: 'client-a', + lastEventId: 2, + replaySnapshot: createTextReplaySnapshot('A transcript'), + events: async function* sourceEvents(opts = {}) { + yield { v: 1, type: 'replay_complete', data: {} }; + await new Promise((resolve) => + opts.signal?.addEventListener('abort', () => resolve(), { + once: true, + }), + ); + }, + }); + const waitingCandidate = createMockSession({ + sessionId: 'session-a', + clientId: 'client-a', + lastEventId: 3, + replaySnapshot: createTextReplaySnapshot('waiting refresh'), + }); + const target = createDeferred(); + sdkMocks.sessions.push(source); + let actions: DaemonSessionActions | undefined; + let connection: DaemonConnectionState | undefined; + + function Harness() { + actions = useDaemonActions(); + connection = useDaemonConnection(); + return null; + } + + await renderWithProvider(, { autoConnect: true }); + sdkMocks.MockDaemonSessionClient.load.mockClear(); + sdkMocks.MockDaemonSessionClient.load + .mockResolvedValueOnce(waitingCandidate) + .mockImplementationOnce(async () => target.promise); + let refresh!: Promise; + act(() => { + refresh = requireActions(actions).loadSession('session-a'); + }); + const refreshOutcome = refresh.catch((error: unknown) => error); + await act(async () => flushPromises()); + expect(connection?.sessionTransition).toMatchObject({ + targetSessionId: 'session-a', + }); + + let switchToTarget!: Promise; + act(() => { + switchToTarget = requireActions(actions).loadSession('session-b'); + }); + await act(async () => flushPromises()); + + await expect(refreshOutcome).resolves.toMatchObject({ name: 'AbortError' }); + expect(sdkMocks.MockDaemonSessionClient.load).toHaveBeenCalledTimes(2); + expect(detachFetch).toHaveBeenCalledOnce(); + expect(String(detachFetch.mock.calls[0]?.[0])).toContain( + '/session/session-a/detach', + ); + expect(connection).toMatchObject({ + status: 'connected', + sessionId: 'session-a', + sessionTransition: { targetSessionId: 'session-b' }, + }); + + await act(async () => { + target.resolve( + createMockSession({ sessionId: 'session-b', clientId: 'client-b' }), + ); + await switchToTarget; + await flushPromises(); + }); + expect(connection).toMatchObject({ + status: 'connected', + sessionId: 'session-b', + clientId: 'client-b', + }); + }); + + it.each([ + ['epoch mismatch', { eventEpoch: 'epoch-2' }], + ['incomplete replay', { replaySnapshotComplete: false }], + ['partial replay', { replayPartial: true }], + ['replay error', { replayError: 'journal read failed' }], + ['degraded replay', { replayDegraded: true }], + ] as const)( + 'rejects a same-session load with %s without replacing the source', + async (_label, candidateOverrides) => { + const detachFetch = vi.fn( + async (_input: RequestInfo | URL, _init?: RequestInit) => + new Response(null, { status: 204 }), + ); + vi.stubGlobal('fetch', detachFetch); + sdkMocks.capabilities.mockResolvedValue({ + workspaceCwd: '/mock-workspace', + features: ['client_identity'], + }); + const source = createMockSession({ + sessionId: 'session-a', + clientId: 'client-a', + lastEventId: 2, + replaySnapshot: createTextReplaySnapshot('A transcript'), + events: async function* sourceEvents(opts = {}) { + yield { v: 1, type: 'replay_complete', data: {} }; + await new Promise((resolve) => + opts.signal?.addEventListener('abort', () => resolve(), { + once: true, + }), + ); + }, + }); + sdkMocks.sessions.push(source); + const candidate = createMockSession({ + sessionId: 'session-a', + clientId: 'client-a', + lastEventId: 2, + replaySnapshot: createTextReplaySnapshot('invalid replacement'), + ...candidateOverrides, + }); + let actions: DaemonSessionActions | undefined; + let blocks: readonly DaemonTranscriptBlock[] = []; + let connection: DaemonConnectionState | undefined; + + function Harness() { + actions = useDaemonActions(); + blocks = useDaemonTranscriptBlocks(); + connection = useDaemonConnection(); + return null; + } + + await renderWithProvider(, { autoConnect: true }); + sdkMocks.MockDaemonSessionClient.load.mockResolvedValueOnce(candidate); + let refresh!: Promise; + act(() => { + refresh = requireActions(actions).loadSession('session-a'); + }); + let refreshError: unknown; + await act(async () => { + try { + await refresh; + } catch (error) { + refreshError = error; + } + await flushPromises(); + }); + + expect(refreshError).toMatchObject({ + message: 'Session refresh returned an incomplete snapshot', + }); + expect(blocks).toMatchObject([{ text: 'A transcript' }]); + expect(connection).toMatchObject({ + status: 'connected', + sessionId: 'session-a', + clientId: 'client-a', + }); + expect(detachFetch).toHaveBeenCalledOnce(); + expect(source.detach).not.toHaveBeenCalled(); + }, + ); + + it('keeps a remote observer turn live before starting a same-session refresh', async () => { + sdkMocks.capabilities.mockResolvedValue({ + workspaceCwd: '/mock-workspace', + features: ['client_identity'], + }); + const terminal = createDeferred(); + const observed = createDeferred(); + const source = createMockSession({ + sessionId: 'session-a', + clientId: 'client-a', + lastEventId: 0, + events: async function* sourceEvents() { + yield { v: 1, type: 'replay_complete', data: {} }; + yield { + id: 1, + v: 1, + type: 'session_update', + data: { + update: { + sessionUpdate: 'user_message_chunk', + content: { type: 'text', text: 'remote prompt' }, + }, + }, + }; + observed.resolve(); + await terminal.promise; + yield { + id: 2, + v: 1, + type: 'turn_complete', + data: { promptId: 'remote-prompt', stopReason: 'end_turn' }, + }; + }, + }); + sdkMocks.sessions.push(source); + const target = createDeferred(); + let actions: DaemonSessionActions | undefined; + + function Harness() { + actions = useDaemonActions(); + return null; + } + + await renderWithProvider(, { autoConnect: true }); + await observed.promise; + sdkMocks.MockDaemonSessionClient.load.mockImplementationOnce( + async () => target.promise, + ); + let refresh!: Promise; + act(() => { + refresh = requireActions(actions).loadSession('session-a'); + }); + const outcome = refresh.catch((error: unknown) => error); + await act(async () => flushPromises()); + expect(sdkMocks.MockDaemonSessionClient.load).toHaveBeenCalledTimes(1); + + await act(async () => { + terminal.resolve(); + await flushPromises(); + }); + expect(sdkMocks.MockDaemonSessionClient.load).toHaveBeenCalledTimes(2); + await act(async () => { + target.reject(new Error('refresh failed')); + await flushPromises(); + }); + await expect(outcome).resolves.toMatchObject({ message: 'refresh failed' }); + }); + + it('rejects a prepared refresh after the source clientId changes in place', async () => { + const detachFetch = vi.fn( + async (_input: RequestInfo | URL, _init?: RequestInit) => + new Response(null, { status: 204 }), + ); + vi.stubGlobal('fetch', detachFetch); + sdkMocks.capabilities.mockResolvedValue({ + workspaceCwd: '/mock-workspace', + features: ['client_identity'], + }); + const source = createMockSession({ + sessionId: 'session-a', + clientId: 'client-a', + lastEventId: 2, + replaySnapshot: createTextReplaySnapshot('A transcript'), + events: async function* sourceEvents(opts = {}) { + yield { v: 1, type: 'replay_complete', data: {} }; + await new Promise((resolve) => + opts.signal?.addEventListener('abort', () => resolve(), { + once: true, + }), + ); + }, + }); + sdkMocks.sessions.push(source); + const target = createDeferred(); + let actions: DaemonSessionActions | undefined; + let connection: DaemonConnectionState | undefined; + + function Harness() { + actions = useDaemonActions(); + connection = useDaemonConnection(); + return null; + } + + await renderWithProvider(, { autoConnect: true }); + sdkMocks.MockDaemonSessionClient.load.mockImplementationOnce( + async () => target.promise, + ); + let refresh!: Promise; + act(() => { + refresh = requireActions(actions).loadSession('session-a'); + }); + const outcome = refresh.catch((error: unknown) => error); + await act(async () => flushPromises()); + + source.clientId = 'client-healed'; + await act(async () => { + target.resolve( + createMockSession({ + sessionId: 'session-a', + clientId: 'client-a', + lastEventId: 2, + replaySnapshot: createTextReplaySnapshot('stale replacement'), + }), + ); + await flushPromises(); + }); + + await expect(outcome).resolves.toMatchObject({ + message: 'Current attachment changed before refresh commit', + }); + expect(connection).toMatchObject({ + status: 'connected', + sessionId: 'session-a', + clientId: 'client-a', + }); + expect(detachFetch).toHaveBeenCalledOnce(); + }); + + it.each([ + { + name: 'a gap', + events: [ + { + id: 4, + v: 1, + type: 'turn_complete', + data: { promptId: 'remote-prompt', stopReason: 'end_turn' }, + } satisfies DaemonEvent, + ], + expectedMessage: 'Source event sequence contains a gap', + }, + { + name: 'an id-less frame', + events: [ + { + v: 1, + type: 'available_commands_update', + data: { availableCommands: [] }, + } satisfies DaemonEvent, + ], + expectedMessage: 'Source emitted id-less available_commands_update', + }, + { + name: 'an overflow', + maxQueued: 1, + events: [ + { + id: 3, + v: 1, + type: 'available_commands_update', + data: { availableCommands: [] }, + } satisfies DaemonEvent, + { + id: 4, + v: 1, + type: 'turn_complete', + data: { promptId: 'remote-prompt', stopReason: 'end_turn' }, + } satisfies DaemonEvent, + ], + expectedMessage: 'Source event capture exceeded its bound', + }, + ])( + 'rejects a same-session load when the captured source tail has $name', + async ({ events, expectedMessage, maxQueued }) => { + const detachFetch = vi.fn( + async (_input: RequestInfo | URL, _init?: RequestInit) => + new Response(null, { status: 204 }), + ); + vi.stubGlobal('fetch', detachFetch); + sdkMocks.capabilities.mockResolvedValue({ + workspaceCwd: '/mock-workspace', + features: ['client_identity'], + }); + const emitTail = createDeferred(); + const source = createMockSession({ + sessionId: 'session-a', + clientId: 'client-a', + lastEventId: 2, + replaySnapshot: createTextReplaySnapshot('A transcript'), + events: async function* sourceEvents() { + yield { v: 1, type: 'replay_complete', data: {} }; + await emitTail.promise; + for (const event of events) yield event; + }, + }); + sdkMocks.sessions.push(source); + const target = createDeferred(); + let actions: DaemonSessionActions | undefined; + + function Harness() { + actions = useDaemonActions(); + return null; + } + + await renderWithProvider(, { + autoConnect: true, + ...(maxQueued !== undefined ? { maxQueued } : {}), + }); + sdkMocks.MockDaemonSessionClient.load.mockImplementationOnce( + async () => target.promise, + ); + let refresh!: Promise; + act(() => { + refresh = requireActions(actions).loadSession('session-a'); + }); + const outcome = refresh.catch((error: unknown) => error); + await act(async () => { + emitTail.resolve(); + await flushPromises(); + target.resolve( + createMockSession({ + sessionId: 'session-a', + clientId: 'client-a', + lastEventId: 2, + replaySnapshot: createTextReplaySnapshot('replacement'), + }), + ); + await flushPromises(); + }); + + await expect(outcome).resolves.toMatchObject({ + message: expectedMessage, + }); + expect(detachFetch).toHaveBeenCalledOnce(); + }, + ); + + it('keeps capturing source events when an in-flight refresh is retried', async () => { + vi.stubGlobal( + 'fetch', + vi.fn(async () => new Response(null, { status: 204 })), + ); + sdkMocks.capabilities.mockResolvedValue({ + workspaceCwd: '/mock-workspace', + features: ['client_identity'], + }); + const emitTail = createDeferred(); + const source = createMockSession({ + sessionId: 'session-a', + clientId: 'client-a', + lastEventId: 2, + replaySnapshot: createTextReplaySnapshot('A transcript'), + events: async function* sourceEvents() { + yield { v: 1, type: 'replay_complete', data: {} }; + await emitTail.promise; + yield { + id: 3, + v: 1, + type: 'session_update', + data: { + update: { + sessionUpdate: 'agent_message_chunk', + content: { type: 'text', text: 'late captured tail' }, + }, + }, + }; + yield { + id: 4, + v: 1, + type: 'turn_complete', + data: { promptId: 'remote-prompt', stopReason: 'end_turn' }, + }; + }, + }); + sdkMocks.sessions.push(source); + const target = createDeferred(); + let actions: DaemonSessionActions | undefined; + let blocks: readonly DaemonTranscriptBlock[] = []; + + function Harness() { + actions = useDaemonActions(); + blocks = useDaemonTranscriptBlocks(); + return null; + } + + await renderWithProvider(, { autoConnect: true }); + sdkMocks.MockDaemonSessionClient.load.mockClear(); + sdkMocks.MockDaemonSessionClient.load.mockImplementationOnce( + async () => target.promise, + ); + vi.useFakeTimers(); + let first!: Promise; + act(() => { + first = requireActions(actions) + .loadSession('session-a') + .catch((error: unknown) => error); + }); + await act(async () => { + await vi.advanceTimersByTimeAsync(75_000); + }); + await expect(first).resolves.toMatchObject({ + message: 'Session transition timed out', + }); + + let timedOutRetry!: Promise; + act(() => { + timedOutRetry = requireActions(actions) + .loadSession('session-a') + .catch((error: unknown) => error); + }); + await act(async () => { + await vi.advanceTimersByTimeAsync(75_000); + }); + await expect(timedOutRetry).resolves.toMatchObject({ + message: 'Session transition timed out', + }); + + let finalRetry!: Promise; + act(() => { + finalRetry = requireActions(actions).loadSession('session-a'); + }); + await act(async () => { + emitTail.resolve(); + await flushPromises(); + target.resolve( + createMockSession({ + sessionId: 'session-a', + clientId: 'client-a', + lastEventId: 2, + replaySnapshot: createTextReplaySnapshot('refreshed transcript'), + }), + ); + await finalRetry; + await flushPromises(); + }); + + expect(sdkMocks.MockDaemonSessionClient.load).toHaveBeenCalledOnce(); + expect( + blocks.map((block) => ('text' in block ? block.text : undefined)), + ).toContain('late captured tail'); + }); + + it('commits a controlled clientId rebind without replacing the transcript', async () => { + const detachFetch = vi.fn( + async (_input: RequestInfo | URL, _init?: RequestInit) => + new Response(null, { status: 204 }), + ); + vi.stubGlobal('fetch', detachFetch); + sdkMocks.capabilities.mockResolvedValue({ + workspaceCwd: '/mock-workspace', + features: ['client_identity'], + }); + let sourceSignal: AbortSignal | undefined; + const source = createMockSession({ + sessionId: 'session-a', + clientId: 'client-a', + lastEventId: 2, + replaySnapshot: createTextReplaySnapshot('A transcript'), + events: async function* sourceEvents(opts = {}) { + sourceSignal = opts.signal; + yield { v: 1, type: 'replay_complete', data: {} }; + await new Promise((resolve) => + opts.signal?.addEventListener('abort', () => resolve(), { + once: true, + }), + ); + }, + }); + sdkMocks.sessions.push(source); + const candidate = createMockSession({ + sessionId: 'session-a', + clientId: 'client-b', + lastEventId: 2, + }); + sdkMocks.MockDaemonSessionClient.resume.mockResolvedValueOnce(candidate); + let blocks: readonly DaemonTranscriptBlock[] = []; + let connection: DaemonConnectionState | undefined; + + function Harness() { + blocks = useDaemonTranscriptBlocks(); + connection = useDaemonConnection(); + return null; + } + + await renderWithProvider(, { + autoConnect: true, + sessionId: 'session-a', + clientId: 'client-a', + }); + await act(async () => { + root?.render( + + + , + ); + await flushPromises(); + }); + + expect(blocks).toMatchObject([{ text: 'A transcript' }]); + expect(connection).toMatchObject({ + status: 'connected', + sessionId: 'session-a', + clientId: 'client-b', + }); + expect(sourceSignal?.aborted).toBe(true); + const [url, init] = detachFetch.mock.calls[0] ?? []; + expect(String(url)).toContain('/session/session-a/detach'); + expect(new Headers(init?.headers).get('X-Qwen-Client-Id')).toBe('client-a'); + }); + + it('keeps the old client when a controlled clientId rebind fails', async () => { + const detachFetch = vi.fn( + async (_input: RequestInfo | URL, _init?: RequestInit) => + new Response(null, { status: 204 }), + ); + vi.stubGlobal('fetch', detachFetch); + sdkMocks.capabilities.mockResolvedValue({ + workspaceCwd: '/mock-workspace', + features: ['client_identity'], + }); + let sourceSignal: AbortSignal | undefined; + const source = createMockSession({ + sessionId: 'session-a', + clientId: 'client-a', + lastEventId: 2, + replaySnapshot: createTextReplaySnapshot('A transcript'), + events: async function* sourceEvents(opts = {}) { + sourceSignal = opts.signal; + yield { v: 1, type: 'replay_complete', data: {} }; + await new Promise((resolve) => + opts.signal?.addEventListener('abort', () => resolve(), { + once: true, + }), + ); + }, + }); + sdkMocks.sessions.push(source); + const target = createDeferred(); + let blocks: readonly DaemonTranscriptBlock[] = []; + let connection: DaemonConnectionState | undefined; + + function Harness() { + blocks = useDaemonTranscriptBlocks(); + connection = useDaemonConnection(); + return null; + } + + await renderWithProvider(, { + autoConnect: true, + sessionId: 'session-a', + clientId: 'client-a', + }); + sdkMocks.MockDaemonSessionClient.resume.mockImplementationOnce( + async () => target.promise, + ); + await act(async () => { + root?.render( + + + , + ); + await flushPromises(); + }); + expect(sourceSignal?.aborted).toBe(false); + expect(detachFetch).not.toHaveBeenCalled(); + expect(sdkMocks.MockDaemonSessionClient.resume.mock.calls.at(-1)?.[3]).toBe( + 'client-b', + ); + + await act(async () => { + target.reject(new Error('rebind failed')); + await flushPromises(); + }); + expect(blocks).toMatchObject([{ text: 'A transcript' }]); + expect(connection).toMatchObject({ + status: 'connected', + sessionId: 'session-a', + clientId: 'client-a', + }); + expect(detachFetch).not.toHaveBeenCalled(); + }); + + it('uses a full load for a legacy controlled clientId rebind', async () => { + sdkMocks.capabilities.mockResolvedValue({ + workspaceCwd: '/mock-workspace', + features: [], + }); + sdkMocks.sessions.push( + createMockSession({ + sessionId: 'session-a', + clientId: 'client-a', + replaySnapshot: createTextReplaySnapshot('A transcript'), + }), + ); + let actions: DaemonSessionActions | undefined; + + function Harness() { + actions = useDaemonActions(); + useDaemonConnection(); + return null; + } + + await renderWithProvider(, { + autoConnect: true, + sessionId: 'session-a', + clientId: 'client-a', + }); + sdkMocks.MockDaemonSessionClient.resume.mockResolvedValueOnce( + createMockSession({ + sessionId: 'session-a', + clientId: 'client-a', + lastEventId: 2, + }), + ); + let resume!: Promise; + act(() => { + resume = requireActions(actions).resumeSession('session-a'); + }); + await act(async () => flushPromises()); + await expect(resume).resolves.toBeUndefined(); + sdkMocks.MockDaemonSessionClient.load.mockClear(); + sdkMocks.MockDaemonSessionClient.resume.mockClear(); + sdkMocks.MockDaemonSessionClient.load.mockResolvedValueOnce( + createMockSession({ + sessionId: 'session-a', + clientId: 'client-b', + replaySnapshot: createTextReplaySnapshot('reloaded transcript'), + }), + ); + + await act(async () => { + root?.render( + + + , + ); + await flushPromises(); + }); + + expect(sdkMocks.MockDaemonSessionClient.resume).not.toHaveBeenCalled(); + expect(sdkMocks.MockDaemonSessionClient.load).toHaveBeenCalledOnce(); + expect(sdkMocks.MockDaemonSessionClient.load.mock.calls[0]?.[3]).toBe( + 'client-b', + ); + }); + async function renderWithProvider( children: ReactNode, props: Partial = {}, @@ -14400,6 +15691,10 @@ function createMockSession(opts: Partial = {}): MockSession { historyHasMore: opts.historyHasMore ?? false, historyAnchorRecordId: opts.historyAnchorRecordId, replayDegraded: opts.replayDegraded ?? false, + replaySnapshotComplete: opts.replaySnapshotComplete ?? true, + replayPartial: opts.replayPartial ?? false, + replayError: opts.replayError, + eventEpoch: opts.eventEpoch ?? 'epoch-1', lastEventId: opts.lastEventId, setLastEventId: opts.setLastEventId ?? diff --git a/packages/webui/src/daemon/session/DaemonSessionProvider.tsx b/packages/webui/src/daemon/session/DaemonSessionProvider.tsx index 110987e373a..0f5ca1c9f17 100644 --- a/packages/webui/src/daemon/session/DaemonSessionProvider.tsx +++ b/packages/webui/src/daemon/session/DaemonSessionProvider.tsx @@ -167,8 +167,33 @@ type TranscriptHistoryState = Omit & { }; interface SessionRunnerControl { session?: DaemonSessionClient; + capture?: SameSessionCapture; flush(): void; stop(): void; + snapshot(): SessionRunnerSnapshot; +} +interface SessionRunnerSnapshot { + session?: DaemonSessionClient; + clientId?: string; + eventEpoch?: string; + processedEventId?: number; + lastPromptTerminalEventId?: number; + ready: boolean; + activeTurn: boolean; +} +interface SameSessionCapture { + source: DaemonSessionClient; + sourceClientId: string; + eventEpoch: string; + startEventId: number; + lastCapturedEventId: number; + bytes: number; + events: DaemonEvent[]; + invalidReason?: string; +} +interface PreparedRunnerHandoff { + session: DaemonSessionClient; + capabilities: DaemonCapabilities; } interface StagedCrossSession { session: DaemonSessionClient; @@ -190,6 +215,8 @@ interface CrossSessionTarget { targetClientId?: string; mode: 'load' | 'resume'; origin: 'action' | 'controlled'; + sameLogical?: boolean; + signal?: AbortSignal; } interface CrossSessionIntent extends CrossSessionTarget { key: string; @@ -203,6 +230,12 @@ interface CrossSessionIntent extends CrossSessionTarget { deadlineAt?: number; timeout?: ReturnType; retryAttempt?: number; + sourceClientId: string; + capture?: SameSessionCapture; + candidate?: DaemonSessionClient; + candidateCapabilities?: DaemonCapabilities; + deadlineStarted?: true; + removeAbortListener?: () => void; promise: Promise; resolve(): void; reject(error: unknown): void; @@ -212,11 +245,14 @@ const CLIENT_IDENTITY_FEATURE = 'client_identity'; const WORKSPACE_ACP_PREHEAT_FEATURE = 'workspace_acp_preheat'; const WORKSPACE_ACP_STATUS_FEATURE = 'workspace_acp_status'; const STAGING_BATCH_SIZE = 512; +const SAME_SESSION_CAPTURE_MAX_BYTES = 8 * 1024 * 1024; +const UTF8_ENCODER = new TextEncoder(); function crossSessionKey( sessionId: string, workspaceCwd: string | undefined, mode: CrossSessionTarget['mode'], historyPageSize: number | undefined, + clientId: string | undefined, ): string { const replayShape = mode === 'resume' @@ -224,7 +260,7 @@ function crossSessionKey( : historyPageSize === undefined ? 'load:all' : `load:recent:${historyPageSize}`; - return `${sessionId}\0${normalizeWorkspaceIdentity(workspaceCwd)}\0${replayShape}`; + return `${sessionId}\0${normalizeWorkspaceIdentity(workspaceCwd)}\0${replayShape}\0${clientId ?? ''}`; } function transitionState( target: CrossSessionTarget, @@ -246,6 +282,8 @@ function settleCrossSessionIntent( error?: unknown, ): void { if (intent.timeout !== undefined) clearTimeout(intent.timeout); + intent.removeAbortListener?.(); + intent.removeAbortListener = undefined; if (error === undefined) intent.resolve(); else intent.reject(error); } @@ -273,6 +311,7 @@ function stageCrossSession(input: { maxBlocks: number; subagentTranscriptMode: 'full' | 'summary'; eventOptions: { suppressOwnUserEcho: boolean; includeRawEvent: boolean }; + additionalEvents?: readonly DaemonEvent[]; }): StagedCrossSession { const { session, capabilities, maxBlocks, subagentTranscriptMode } = input; const notices: SessionNoticeInput[] = []; @@ -417,6 +456,7 @@ function stageCrossSession(input: { } consume(event); } + for (const event of input.additionalEvents ?? []) consume(event); flush(); const replayTokenUsage = getReplayTokenUsage(session.replaySnapshot.liveJournal) ?? @@ -814,11 +854,13 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { const resolvedBaseUrl = baseUrl ?? workspace?.baseUrl; const resolvedToken = token ?? workspace?.token; const resolvedWorkspaceCwd = workspaceCwd ?? workspace?.workspaceCwd; + const sessionCapabilitiesRef = useRef( + workspace?.capabilities, + ); const environmentRef = useRef({ baseUrl: resolvedBaseUrl, token: resolvedToken, client: workspace?.client, - clientId, maxBlocks, subagentTranscriptMode, generation: 0, @@ -827,15 +869,14 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { environmentRef.current.baseUrl !== resolvedBaseUrl || environmentRef.current.token !== resolvedToken || environmentRef.current.client !== workspace?.client || - environmentRef.current.clientId !== clientId || environmentRef.current.maxBlocks !== maxBlocks || environmentRef.current.subagentTranscriptMode !== subagentTranscriptMode ) { + sessionCapabilitiesRef.current = workspace?.capabilities; environmentRef.current = { baseUrl: resolvedBaseUrl, token: resolvedToken, client: workspace?.client, - clientId, maxBlocks, subagentTranscriptMode, generation: environmentRef.current.generation + 1, @@ -867,7 +908,9 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { ); const sessionRef = useRef(undefined); const runnerControlRef = useRef(undefined); - const preparedRunnerRef = useRef(undefined); + const preparedRunnerRef = useRef( + undefined, + ); const desiredTransitionRef = useRef( undefined, ); @@ -933,10 +976,7 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { const loadWarningsRef = useRef(loadWarnings); const historyPageSizeRef = useRef(historyPageSize); const subagentTranscriptModeRef = useRef(subagentTranscriptMode); - const clientIdRef = useRef(undefined); - if (!clientIdRef.current || clientId) { - clientIdRef.current = getStableClientId(clientId); - } + const clientIdRef = useRef(getStableClientId(clientId)); eventOptionsRef.current = { suppressOwnUserEcho, includeRawEvent }; reconnectConfigRef.current = { reconnectDelayMs, maxReconnectDelayMs }; loadWarningsRef.current = loadWarnings; @@ -964,6 +1004,19 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { }); const connectionRef = useRef(connection); connectionRef.current = connection; + const initialClientIdDependencyRef = useRef(clientId); + const knownCapabilities = + workspace?.capabilities ?? + sessionCapabilitiesRef.current ?? + connection.capabilities; + const legacyClientIdDependency = + knownCapabilities && + !knownCapabilities.features.includes(CLIENT_IDENTITY_FEATURE) + ? clientId + : initialClientIdDependencyRef.current; + if (legacyClientIdDependency) { + clientIdRef.current = getStableClientId(legacyClientIdDependency); + } const setConnectionSynchronous = useCallback( (update: SetStateAction) => { const next = @@ -1133,12 +1186,81 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { cancelTranscriptFlush(); pendingTranscriptEvents = []; }; + let runnerEventEpoch = runnerSession?.eventEpoch; + let processedEventId = runnerSession?.lastEventId; + let lastPromptTerminalEventId: number | undefined; + let runnerReady = false; + let runnerActiveTurn = false; const runnerControl: SessionRunnerControl = { session: runnerSession, flush: flushTranscriptSync, stop: () => abort.abort(), + snapshot: () => ({ + session: runnerControl.session, + clientId: runnerControl.session?.clientId, + eventEpoch: runnerEventEpoch, + processedEventId, + lastPromptTerminalEventId, + ready: runnerReady, + activeTurn: runnerActiveTurn, + }), }; runnerControlRef.current = runnerControl; + const captureSourceEvent = (event: DaemonEvent) => { + const capture = runnerControl.capture; + if (!capture || capture.invalidReason) return; + if ( + runnerControl.session !== capture.source || + capture.source.clientId !== capture.sourceClientId || + capture.source.eventEpoch !== capture.eventEpoch + ) { + capture.invalidReason = 'Source attachment changed during refresh'; + return; + } + if (event.id === undefined) { + if (event.type !== 'replay_complete') { + capture.invalidReason = `Source emitted id-less ${event.type}`; + } + return; + } + if (event.id <= capture.lastCapturedEventId) return; + if (event.id !== capture.lastCapturedEventId + 1) { + capture.invalidReason = 'Source event sequence contains a gap'; + return; + } + let serializedBytes: number; + try { + serializedBytes = UTF8_ENCODER.encode(JSON.stringify(event)).byteLength; + } catch (error) { + capture.invalidReason = + error instanceof Error ? error.message : String(error); + return; + } + if ( + capture.events.length >= maxQueued || + capture.bytes + serializedBytes > SAME_SESSION_CAPTURE_MAX_BYTES + ) { + capture.invalidReason = 'Source event capture exceeded its bound'; + return; + } + capture.events.push(event); + capture.bytes += serializedBytes; + capture.lastCapturedEventId = event.id; + }; + const markSourceEventProcessed = (event: DaemonEvent) => { + const learnedEpoch = runnerControl.session?.eventEpoch; + if (learnedEpoch !== runnerEventEpoch) { + if (runnerControl.capture && !runnerControl.capture.invalidReason) { + runnerControl.capture.invalidReason = + 'Source event epoch changed during refresh'; + } + runnerEventEpoch = learnedEpoch; + } + if (event.id !== undefined) { + processedEventId = Math.max(processedEventId ?? 0, event.id); + } + queueMicrotask(pumpTransitionRef.current); + }; const tryLiveJournalRepair = () => { if (disposed || abort.signal.aborted) return; const repair = liveJournalRepairRef.current; @@ -1301,6 +1423,7 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { : await client.capabilities()); if (disposed || abort.signal.aborted) return; capabilities = caps; + sessionCapabilitiesRef.current = caps; const historyPaginationSupported = Array.isArray(caps.features) && caps.features.includes(SESSION_TRANSCRIPT_PAGINATION_FEATURE); @@ -1450,14 +1573,21 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { } return; } - const restoreMethod = - restoreSessionId && restoreMode === 'resume' - ? DaemonSessionClient.resume - : DaemonSessionClient.load; const targetSessionId = restoreSessionId ?? reconnectSessionId; - const requestClientId = clientId + const requestClientId = legacyClientIdDependency ? clientIdRef.current : getStableClientId(undefined, targetSessionId); + const legacyClientRebind = + targetSessionId !== undefined && + targetSessionId === connectionRef.current.sessionId && + connectionRef.current.clientId !== undefined && + requestClientId !== connectionRef.current.clientId; + const restoreMethod = + restoreSessionId && + restoreMode === 'resume' && + !legacyClientRebind + ? DaemonSessionClient.resume + : DaemonSessionClient.load; loadingRequestedSession = Boolean(restoreSessionId); if (targetSessionId && !preservingTranscriptDuringLoad) { setConnection((current) => ({ @@ -1522,7 +1652,7 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { requestClientId, ); loadingRequestedSession = false; - if (!clientId && nextSession.clientId) { + if (!legacyClientIdDependency && nextSession.clientId) { clientIdRef.current = nextSession.clientId; persistStableClientId( nextSession.clientId, @@ -1694,6 +1824,8 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { const activeSession = session; runnerSession = activeSession; runnerControl.session = activeSession; + runnerEventEpoch = activeSession.eventEpoch; + processedEventId = activeSession.lastEventId; // Prompt activity is session state returned by /load. Surface it // immediately so a refreshed page shows the running state without // waiting for auxiliary data such as providers, commands, or context. @@ -1719,9 +1851,11 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { }; const hasSessionActivePrompt = () => restoredActivePrompt || - activePromptsRef.current.has(activeSession.sessionId); + activePromptsRef.current.has(activeSession.sessionId) || + activePromptsRef.current.has(`${activeSession.sessionId}:shell`); hasCurrentSessionActivePrompt = hasSessionActivePrompt; hasCurrentSessionActivePromptRef.current = hasSessionActivePrompt; + runnerActiveTurn = hasSessionActivePrompt(); setPromptStatus(hasSessionActivePrompt() ? 'streaming' : 'idle'); const pendingLoad = pendingSessionLoadRef.current; @@ -2330,6 +2464,8 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { abort.signal.removeEventListener('abort', abortEventStream); const sseConnectReason = nextSseConnectReason; nextSseConnectReason = undefined; + runnerReady = activeSession.lastEventId === undefined; + if (runnerReady) queueMicrotask(pumpTransitionRef.current); for await (const event of activeSession.events({ signal: eventStreamController.signal, maxQueued, @@ -2338,6 +2474,7 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { if (sessionRef.current !== activeSession) { break; } + captureSourceEvent(event); if (!sawEvent) { sawEvent = true; reconnectAttempt = 0; @@ -2353,314 +2490,350 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { ); } try { - const followupSuggestion = - parseSidechannelFollowupSuggestion(event); - if (followupSuggestion) { - publishSidechannelFollowupSuggestion(followupSuggestion); - continue; - } - const midTurnInjected = parseSidechannelMidTurnInjected(event); - if (midTurnInjected) { - // Keep the sidechannel for queue dedupe, but still normalize the - // event below so chat UIs can render the inserted-message status. - publishSidechannelMidTurnInjected(midTurnInjected); - if (sessionRef.current !== activeSession) break; - } - if (isPendingPromptEvent(event)) { - publishPendingPromptEvent(event); - if (sessionRef.current !== activeSession) break; - if (event.type === 'pending_prompt_started') { - clearPassiveAssistantDoneTimer(passiveAssistantDoneTimerRef); - setPromptStatus('waiting'); + try { + const followupSuggestion = + parseSidechannelFollowupSuggestion(event); + if (followupSuggestion) { + publishSidechannelFollowupSuggestion(followupSuggestion); + continue; } - } - const normalizedUiEvents = normalizeAndFilterEvent( - event, - activeSession.clientId, - eventOptionsRef.current, - setConnection, - ); - const uiEvents = filterDaemonUiEventsForTranscript( - event, - normalizedUiEvents, - addNotice, - dismissNotice, - ); - const transcriptUiEvents = - subagentTranscriptModeRef.current === 'summary' - ? projectMainTranscriptEvents(uiEvents) - : uiEvents; - if (event.type === 'state_resync_required') { - const reason = - typeof event.data === 'object' && event.data !== null - ? (event.data as Record).reason - : undefined; - if (reason === 'epoch_reset') { - requestEpochResetReload(); - break; + const midTurnInjected = parseSidechannelMidTurnInjected(event); + if (midTurnInjected) { + // Keep the sidechannel for queue dedupe, but still normalize the + // event below so chat UIs can render the inserted-message status. + publishSidechannelMidTurnInjected(midTurnInjected); + if (sessionRef.current !== activeSession) break; } - } - bumpWorkspaceEventSignals(uiEvents, setWorkspaceEventSignals); - if (uiEvents.length > 0) { - const hasGenerationSignal = hasActiveGenerationSignal(uiEvents); - setPromptStatus((current) => - current === 'waiting' || - (current === 'idle' && hasGenerationSignal) - ? 'streaming' - : current, + if (isPendingPromptEvent(event)) { + publishPendingPromptEvent(event); + if (sessionRef.current !== activeSession) break; + if (event.type === 'pending_prompt_started') { + runnerActiveTurn = true; + clearPassiveAssistantDoneTimer( + passiveAssistantDoneTimerRef, + ); + setPromptStatus('waiting'); + } + } + const normalizedUiEvents = normalizeAndFilterEvent( + event, + activeSession.clientId, + eventOptionsRef.current, + (update) => { + setConnectionSynchronous((current) => { + if (sessionRef.current !== activeSession) return current; + return typeof update === 'function' + ? update(current) + : update; + }); + }, ); - } - // Flush buffered transcript events before settling a turn so the - // turn's content is applied ahead of the assistant.done that - // settle (and the restored-prompt / observer branches below) - // dispatch. Guarded to turn terminals so steady streaming keeps - // batching. - if ( - event.type === 'turn_complete' || - event.type === 'turn_error' - ) { - flushTranscriptSync(); - } - const activePromptSettled = settleActivePromptFromTurnEvent( - activePromptsRef.current, - settledPromptsRef.current, - activeSession.sessionId, - event, - store, - setPromptStatus, - passiveAssistantDoneTimerRef, - ); - let restoredPromptSettled = false; - if ( - !activePromptSettled && - restoredActivePrompt && - (event.type === 'turn_complete' || event.type === 'turn_error') - ) { - // A refreshed page restores an already-running prompt without a - // local ActivePrompt entry or prompt promise to settle. The daemon - // terminal event is still authoritative, so end the restored - // running state here instead of relying on the observer branch. - settleRestoredActivePrompt(); - restoredPromptSettled = true; - clearPassiveAssistantDoneTimer(passiveAssistantDoneTimerRef); - const stopReason = - event.type === 'turn_complete' - ? ((event.data as DaemonTurnCompleteData | undefined) - ?.stopReason ?? 'end_turn') - : 'error'; - dispatchTranscriptNow( - assistantDoneFromTurnEvent(event, stopReason), + const uiEvents = filterDaemonUiEventsForTranscript( + event, + normalizedUiEvents, + addNotice, + dismissNotice, ); - if (!hasSessionActivePrompt()) { - setPromptStatus('idle'); + const transcriptUiEvents = + subagentTranscriptModeRef.current === 'summary' + ? projectMainTranscriptEvents(uiEvents) + : uiEvents; + if (event.type === 'state_resync_required') { + const reason = + typeof event.data === 'object' && event.data !== null + ? (event.data as Record).reason + : undefined; + if (reason === 'epoch_reset') { + requestEpochResetReload(); + break; + } } - } - // The debug guard below reads the committed store's active - // assistant block, but batching leaves earlier chunks from this - // same burst in the pending buffer until the macrotask flush. An - // observer burst that interleaves a debug event between assistant - // chunks would otherwise miss the still-pending assistant block - // and let the debug event split it. Commit the buffer first so the - // guard sees the effective state. Scoped to observer-mode debug - // events (rare) so steady streaming keeps batching. - if ( - !hasSessionActivePrompt() && - uiEvents.some((e) => e.type === 'debug') - ) { - flushTranscriptSync(); - } - const shouldGuardAssistant = - !hasSessionActivePrompt() && - store.getSnapshot().activeAssistantBlockId != null; - const eventsToDispatch = shouldGuardAssistant - ? transcriptUiEvents.filter((e) => e.type !== 'debug') - : transcriptUiEvents; - enqueueTranscriptEvents(eventsToDispatch); - for (const uiEvent of uiEvents) { + bumpWorkspaceEventSignals(uiEvents, setWorkspaceEventSignals); + if (uiEvents.length > 0) { + const hasGenerationSignal = + hasActiveGenerationSignal(uiEvents); + if (hasGenerationSignal) runnerActiveTurn = true; + setPromptStatus((current) => + current === 'waiting' || + (current === 'idle' && hasGenerationSignal) + ? 'streaming' + : current, + ); + } + // Flush buffered transcript events before settling a turn so the + // turn's content is applied ahead of the assistant.done that + // settle (and the restored-prompt / observer branches below) + // dispatch. Guarded to turn terminals so steady streaming keeps + // batching. if ( - uiEvent.type === 'prompt.cancelled' && - (restoredActivePrompt || - uiEvent.originatorClientId !== activeSession.clientId) + event.type === 'turn_complete' || + event.type === 'turn_error' ) { - dispatchTranscriptNow( - assistantDoneFromTurnEvent(event, 'cancelled'), - ); - const cancellingRestoredPrompt = restoredActivePrompt; + flushTranscriptSync(); + } + const activePromptSettled = settleActivePromptFromTurnEvent( + activePromptsRef.current, + settledPromptsRef.current, + activeSession.sessionId, + event, + store, + setPromptStatus, + passiveAssistantDoneTimerRef, + ); + let restoredPromptSettled = false; + if ( + !activePromptSettled && + restoredActivePrompt && + (event.type === 'turn_complete' || + event.type === 'turn_error') + ) { + // A refreshed page restores an already-running prompt without a + // local ActivePrompt entry or prompt promise to settle. The daemon + // terminal event is still authoritative, so end the restored + // running state here instead of relying on the observer branch. settleRestoredActivePrompt(); restoredPromptSettled = true; clearPassiveAssistantDoneTimer(passiveAssistantDoneTimerRef); - if (!cancellingRestoredPrompt) { - activePromptsRef.current.delete(activeSession.sessionId); - } + const stopReason = + event.type === 'turn_complete' + ? ((event.data as DaemonTurnCompleteData | undefined) + ?.stopReason ?? 'end_turn') + : 'error'; + dispatchTranscriptNow( + assistantDoneFromTurnEvent(event, stopReason), + ); if (!hasSessionActivePrompt()) { setPromptStatus('idle'); } - } else if (uiEvent.type === 'session.replay_complete') { - // Flush first so the awaitingResync read below reflects every - // event up to replay_complete (e.g. a buffered - // state_resync_required from this same burst). + } + // The debug guard below reads the committed store's active + // assistant block, but batching leaves earlier chunks from this + // same burst in the pending buffer until the macrotask flush. An + // observer burst that interleaves a debug event between assistant + // chunks would otherwise miss the still-pending assistant block + // and let the debug event split it. Commit the buffer first so the + // guard sees the effective state. Scoped to observer-mode debug + // events (rare) so steady streaming keeps batching. + if ( + !hasSessionActivePrompt() && + uiEvents.some((e) => e.type === 'debug') + ) { flushTranscriptSync(); - setConnection((c) => ({ ...c, catchingUp: undefined })); - if (store.getSnapshot().awaitingResync) { - store.clearAwaitingResync(); - } - if (!hasSessionActivePrompt()) { + } + const shouldGuardAssistant = + !hasSessionActivePrompt() && + store.getSnapshot().activeAssistantBlockId != null; + const eventsToDispatch = shouldGuardAssistant + ? transcriptUiEvents.filter((e) => e.type !== 'debug') + : transcriptUiEvents; + enqueueTranscriptEvents(eventsToDispatch); + for (const uiEvent of uiEvents) { + if ( + uiEvent.type === 'prompt.cancelled' && + (restoredActivePrompt || + uiEvent.originatorClientId !== activeSession.clientId) + ) { + dispatchTranscriptNow( + assistantDoneFromTurnEvent(event, 'cancelled'), + ); + const cancellingRestoredPrompt = restoredActivePrompt; + settleRestoredActivePrompt(); + restoredPromptSettled = true; clearPassiveAssistantDoneTimer( passiveAssistantDoneTimerRef, ); - dispatchTranscriptNow({ - type: 'assistant.done', - reason: 'replay_complete', - }); - setPromptStatus('idle'); + if (!cancellingRestoredPrompt) { + activePromptsRef.current.delete(activeSession.sessionId); + } + if (!hasSessionActivePrompt()) { + setPromptStatus('idle'); + } + } else if (uiEvent.type === 'session.replay_complete') { + // Flush first so the awaitingResync read below reflects every + // event up to replay_complete (e.g. a buffered + // state_resync_required from this same burst). + flushTranscriptSync(); + setConnection((c) => ({ ...c, catchingUp: undefined })); + if (store.getSnapshot().awaitingResync) { + store.clearAwaitingResync(); + } + runnerReady = true; + queueMicrotask(pumpTransitionRef.current); + if (!hasSessionActivePrompt()) { + clearPassiveAssistantDoneTimer( + passiveAssistantDoneTimerRef, + ); + dispatchTranscriptNow({ + type: 'assistant.done', + reason: 'replay_complete', + }); + setPromptStatus('idle'); + } } } - } - // A restored active prompt is not in activePromptsRef because this - // browser did not submit it. Treat it as active here too; otherwise - // the passive observer timer can briefly mark a still-running turn - // idle between sparse tool/thinking updates. - const isObserver = - !activePromptSettled && - !restoredPromptSettled && - !hasSessionActivePrompt(); - if (isObserver) { - const hasUserMsg = uiEvents.some( - (e) => e.type === 'user.text.delta', - ); - if (hasUserMsg) { - setPromptStatus('waiting'); - } else if (hasActiveGenerationSignal(uiEvents)) { - setPromptStatus((current) => - current === 'idle' ? 'streaming' : current, + // A restored active prompt is not in activePromptsRef because this + // browser did not submit it. Treat it as active here too; otherwise + // the passive observer timer can briefly mark a still-running turn + // idle between sparse tool/thinking updates. + const isObserver = + !activePromptSettled && + !restoredPromptSettled && + !hasSessionActivePrompt(); + if (isObserver) { + const hasUserMsg = uiEvents.some( + (e) => e.type === 'user.text.delta', ); + if (hasUserMsg) { + runnerActiveTurn = true; + setPromptStatus('waiting'); + } else if (hasActiveGenerationSignal(uiEvents)) { + setPromptStatus((current) => + current === 'idle' ? 'streaming' : current, + ); + } } - } - if (isObserver && event.type === 'turn_complete') { - clearPassiveAssistantDoneTimer(passiveAssistantDoneTimerRef); - const stopReason = - (event.data as DaemonTurnCompleteData | undefined) - ?.stopReason ?? 'end_turn'; - dispatchTranscriptNow( - assistantDoneFromTurnEvent(event, stopReason), - ); - setPromptStatus('idle'); - } else if (isObserver && event.type === 'turn_error') { - clearPassiveAssistantDoneTimer(passiveAssistantDoneTimerRef); - dispatchTranscriptNow( - assistantDoneFromTurnEvent(event, 'error'), - ); - setPromptStatus('idle'); - } else if (isObserver && hasActiveGenerationSignal(uiEvents)) { - schedulePassiveAssistantDone( - store, - passiveAssistantDoneTimerRef, - 'passive_observer', - 3000, - () => setPromptStatus('idle'), - ); - } - const pendingRepair = liveJournalRepairRef.current; - if ( - pendingRepair?.sessionId === activeSession.sessionId && - (event.type === 'turn_complete' || - event.type === 'turn_error') && - eventPromptId(event) === pendingRepair.target.promptId - ) { - pendingRepair.terminalSeen = true; - queueMicrotask(tryLiveJournalRepair); - } else if (pendingRepair?.terminalSeen) { - queueMicrotask(tryLiveJournalRepair); - } - // ── state_resync_required handling ────────────────────── - // Resyncs are transcript recovery signals, not prompt terminal - // signals. For epoch_reset and ring_evicted we reload the session - // snapshot; the fresh /load response is the source of truth for - // hasActivePrompt and transcript replay. - if (event.type === 'state_resync_required') { - const reason = - typeof event.data === 'object' && event.data !== null - ? (event.data as Record).reason - : undefined; - if (reason !== 'epoch_reset') { - cancelTransitionRef.current( - 'Session transition cancelled by state resync', + if (isObserver && event.type === 'turn_complete') { + clearPassiveAssistantDoneTimer(passiveAssistantDoneTimerRef); + const stopReason = + (event.data as DaemonTurnCompleteData | undefined) + ?.stopReason ?? 'end_turn'; + dispatchTranscriptNow( + assistantDoneFromTurnEvent(event, stopReason), ); - // Resync asks us to rebuild transcript state, but it is not a - // prompt terminal signal. Keep loading alive for local/restored - // prompts until turn_complete, turn_error, or prompt_cancelled. - if (!hasSessionActivePrompt()) { - setPromptStatus('idle'); - clearPassiveAssistantDoneTimer( - passiveAssistantDoneTimerRef, + setPromptStatus('idle'); + } else if (isObserver && event.type === 'turn_error') { + clearPassiveAssistantDoneTimer(passiveAssistantDoneTimerRef); + dispatchTranscriptNow( + assistantDoneFromTurnEvent(event, 'error'), + ); + setPromptStatus('idle'); + } else if (isObserver && hasActiveGenerationSignal(uiEvents)) { + schedulePassiveAssistantDone( + store, + passiveAssistantDoneTimerRef, + 'passive_observer', + 3000, + () => setPromptStatus('idle'), + ); + } + if ( + event.id !== undefined && + (event.type === 'turn_complete' || + event.type === 'turn_error' || + uiEvents.some( + (uiEvent) => uiEvent.type === 'prompt.cancelled', + )) + ) { + lastPromptTerminalEventId = event.id; + runnerActiveTurn = hasSessionActivePrompt(); + if (!runnerActiveTurn) { + queueMicrotask(pumpTransitionRef.current); + } + } + const pendingRepair = liveJournalRepairRef.current; + if ( + pendingRepair?.sessionId === activeSession.sessionId && + (event.type === 'turn_complete' || + event.type === 'turn_error') && + eventPromptId(event) === pendingRepair.target.promptId + ) { + pendingRepair.terminalSeen = true; + queueMicrotask(tryLiveJournalRepair); + } else if (pendingRepair?.terminalSeen) { + queueMicrotask(tryLiveJournalRepair); + } + // ── state_resync_required handling ────────────────────── + // Resyncs are transcript recovery signals, not prompt terminal + // signals. For epoch_reset and ring_evicted we reload the session + // snapshot; the fresh /load response is the source of truth for + // hasActivePrompt and transcript replay. + if (event.type === 'state_resync_required') { + const reason = + typeof event.data === 'object' && event.data !== null + ? (event.data as Record).reason + : undefined; + if (reason !== 'epoch_reset') { + cancelTransitionRef.current( + 'Session transition cancelled by state resync', + ); + // Resync asks us to rebuild transcript state, but it is not a + // prompt terminal signal. Keep loading alive for local/restored + // prompts until turn_complete, turn_error, or prompt_cancelled. + if (!hasSessionActivePrompt()) { + setPromptStatus('idle'); + clearPassiveAssistantDoneTimer( + passiveAssistantDoneTimerRef, + ); + } + clearPendingTranscriptEvents(); + store.reset(); + // Ring eviction means the SSE replay window has a real gap. + // Resetting and continuing on the same stream can only replay + // the surviving tail; reload the session snapshot instead so + // compactedReplay/liveJournal rebuild the bounded replay + // window. + console.warn( + '[DaemonSessionProvider] ring eviction detected, reloading session (sessionId=%s)', + activeSession.sessionId, ); + resyncRequested = true; + nextSseConnectReason = 'state_resync'; + session = undefined; + sessionRef.current = undefined; + hasCurrentSessionActivePromptRef.current = () => false; + setConnection((current) => ({ + ...current, + status: 'connecting', + error: undefined, + errorStatus: resolveConnectionErrorStatus( + undefined, + current.errorStatus, + ), + })); + break; } - clearPendingTranscriptEvents(); - store.reset(); - // Ring eviction means the SSE replay window has a real gap. - // Resetting and continuing on the same stream can only replay - // the surviving tail; reload the session snapshot instead so - // compactedReplay/liveJournal rebuild the bounded replay - // window. - console.warn( - '[DaemonSessionProvider] ring eviction detected, reloading session (sessionId=%s)', - activeSession.sessionId, - ); - resyncRequested = true; - nextSseConnectReason = 'state_resync'; + } + // session_closed with reason 'client_close' means the + // user explicitly deleted the session. Stop the + // reconnect loop — without this, the next iteration + // would call createOrAttach and auto-create a new + // session, undoing the user's delete action. + // Other reasons (idle_timeout, last_client_detached) + // fall through to the normal reconnect path. + if ( + event.type === 'session_closed' && + (event.data as Record | undefined) + ?.reason === 'client_close' + ) { + userDeletedSession = true; + const closedSessionId = activeSession.sessionId; + const active = activePromptsRef.current.get(closedSessionId); + active?.controller.abort(); + activePromptsRef.current.delete(closedSessionId); session = undefined; sessionRef.current = undefined; - hasCurrentSessionActivePromptRef.current = () => false; - setConnection((current) => ({ - ...current, - status: 'connecting', - error: undefined, - errorStatus: resolveConnectionErrorStatus( - undefined, - current.errorStatus, - ), - })); break; } + } catch (error) { + if (sessionRef.current !== activeSession) break; + const message = + error instanceof Error ? error.message : String(error); + addNotice({ + severity: 'warning', + category: 'protocol', + operation: 'normalize_event', + code: 'daemon.event_malformed', + message: 'Skipped malformed daemon event', + debugMessage: message, + recoverable: true, + }); + console.warn( + '[DaemonSessionProvider] skipped malformed daemon event:', + error, + ); } - // session_closed with reason 'client_close' means the - // user explicitly deleted the session. Stop the - // reconnect loop — without this, the next iteration - // would call createOrAttach and auto-create a new - // session, undoing the user's delete action. - // Other reasons (idle_timeout, last_client_detached) - // fall through to the normal reconnect path. - if ( - event.type === 'session_closed' && - (event.data as Record | undefined)?.reason === - 'client_close' - ) { - userDeletedSession = true; - const closedSessionId = activeSession.sessionId; - const active = activePromptsRef.current.get(closedSessionId); - active?.controller.abort(); - activePromptsRef.current.delete(closedSessionId); - session = undefined; - sessionRef.current = undefined; - break; - } - } catch (error) { - if (sessionRef.current !== activeSession) break; - const message = - error instanceof Error ? error.message : String(error); - addNotice({ - severity: 'warning', - category: 'protocol', - operation: 'normalize_event', - code: 'daemon.event_malformed', - message: 'Skipped malformed daemon event', - debugMessage: message, - recoverable: true, - }); - console.warn( - '[DaemonSessionProvider] skipped malformed daemon event:', - error, - ); + } finally { + markSourceEventProcessed(event); } } if ( @@ -2676,6 +2849,7 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { // so post-loop handling (and consumers reading the snapshot) see a // complete transcript without waiting for the scheduled flush. flushTranscriptSync(); + runnerReady = false; const restartRequested = eventStream.restartRequested; clearEventStream(); if (restartRequested) { @@ -2744,6 +2918,7 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { })); } } catch (error) { + runnerReady = false; const restartRequested = eventStream?.restartRequested === true; clearEventStream(); if (session && sessionRef.current !== session) { @@ -3056,11 +3231,12 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { restoreSessionNonce, attachSessionNonce, newSessionNonce, - clientId, + legacyClientIdDependency, shouldDeferInitialSessionCreation, clearNotices, addNotice, dismissNotice, + setConnectionSynchronous, ]); useEffect(() => { @@ -3232,7 +3408,9 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { category: 'connection', operation: target.mode === 'resume' ? 'resume_session' : 'load_session', code: 'daemon.session_transition.failed', - message: `Could not open session ${target.sessionId}. The current session is still active.`, + message: target.sameLogical + ? `Could not refresh session ${target.sessionId}. The current attachment is still active.` + : `Could not open session ${target.sessionId}. The current session is still active.`, debugMessage: message, recoverable: true, }); @@ -3266,6 +3444,36 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { [], ); + const armTransitionDeadline = useCallback( + (intent: CrossSessionIntent, capabilities: DaemonCapabilities) => { + if (intent.deadlineStarted) return; + intent.deadlineStarted = true; + const timeoutMs = + resolveSessionRestoreTimeouts(capabilities).watchdogTimeoutMs; + if (timeoutMs === undefined) return; + intent.deadlineAt = Date.now() + timeoutMs; + intent.timeout = setTimeout(() => { + if (intent.candidate) { + retireAttachment(intent.candidate, intent); + intent.candidate = undefined; + } + const control = runnerControlRef.current; + if ( + rawTransitionRef.current !== intent && + control && + control.capture === intent.capture + ) { + control.capture = undefined; + } + exposeCrossSessionFailure( + intent, + new Error('Session transition timed out'), + ); + }, timeoutMs); + }, + [exposeCrossSessionFailure, retireAttachment], + ); + const commitCrossSession = useCallback( (intent: CrossSessionIntent, staged: StagedCrossSession): boolean => { if ( @@ -3385,8 +3593,179 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { ], ); + const commitSameSession = useCallback( + ( + intent: CrossSessionIntent, + candidate: DaemonSessionClient, + capabilities: DaemonCapabilities, + capture: SameSessionCapture | undefined, + ): boolean => { + const control = runnerControlRef.current; + if (!control) return false; + const snapshot = control?.snapshot(); + const watermark = candidate.lastEventId; + if ( + !mountedRef.current || + desiredTransitionRef.current !== intent || + intent.lifecycle !== lifecycleRef.current || + intent.environmentGeneration !== environmentRef.current.generation || + sessionRef.current !== intent.source || + intent.source.clientId !== intent.sourceClientId || + snapshot?.session !== intent.source || + snapshot.clientId !== intent.sourceClientId || + snapshot.eventEpoch !== candidate.eventEpoch || + snapshot.eventEpoch === undefined || + snapshot.processedEventId === undefined || + !Number.isSafeInteger(snapshot.processedEventId) || + snapshot.processedEventId < 0 || + !snapshot.ready || + snapshot.activeTurn || + hasCurrentSessionActivePromptRef.current() || + watermark === undefined || + !Number.isSafeInteger(watermark) || + watermark < 0 || + snapshot.processedEventId < watermark || + (intent.deadlineAt !== undefined && Date.now() >= intent.deadlineAt) + ) { + return false; + } + if ( + candidate.hasActivePrompt && + (snapshot.lastPromptTerminalEventId === undefined || + snapshot.lastPromptTerminalEventId <= watermark) + ) { + return false; + } + + let staged: StagedCrossSession | undefined; + if (intent.mode === 'load') { + if ( + !capture || + capture.invalidReason || + capture.source !== intent.source || + capture.sourceClientId !== intent.sourceClientId || + capture.eventEpoch !== candidate.eventEpoch || + watermark < capture.startEventId + ) { + return false; + } + const tail = capture.events.filter( + (event) => + event.id !== undefined && + event.id > watermark && + event.id <= snapshot.processedEventId!, + ); + candidate.setLastEventId(snapshot.processedEventId); + try { + staged = stageCrossSession({ + session: candidate, + capabilities, + maxBlocks, + subagentTranscriptMode: subagentTranscriptModeRef.current, + eventOptions: eventOptionsRef.current, + additionalEvents: tail, + }); + } catch { + return false; + } + if ( + staged.repair || + staged.notices.some( + (notice) => notice.code === 'daemon.replay_event_malformed', + ) + ) { + return false; + } + } + + const finalSnapshot = control?.snapshot(); + if ( + desiredTransitionRef.current !== intent || + sessionRef.current !== intent.source || + intent.source.clientId !== intent.sourceClientId || + finalSnapshot?.session !== intent.source || + finalSnapshot.clientId !== intent.sourceClientId || + finalSnapshot.eventEpoch !== candidate.eventEpoch || + finalSnapshot.processedEventId !== snapshot.processedEventId || + !finalSnapshot.ready || + finalSnapshot.activeTurn || + (intent.deadlineAt !== undefined && Date.now() >= intent.deadlineAt) + ) { + return false; + } + + if (intent.timeout !== undefined) clearTimeout(intent.timeout); + if (control.capture === capture) control.capture = undefined; + control.flush(); + control.stop(); + candidate.setLastEventId(finalSnapshot.processedEventId); + if (staged) { + store.reset(staged.transcript); + transcriptHistoryRef.current = staged.history; + setTranscriptHistoryState({ + hasMore: staged.history.hasMore, + loading: false, + capacityReached: staged.history.capacityReached, + paginationError: false, + }); + } else { + const history = { + ...transcriptHistoryRef.current, + loading: false, + }; + transcriptHistoryRef.current = history; + setTranscriptHistoryState({ + hasMore: history.hasMore, + loading: false, + capacityReached: history.capacityReached, + paginationError: history.paginationError, + }); + } + sessionRef.current = candidate; + lastSessionIdRef.current = candidate.sessionId; + activeWorkspaceCwdRef.current = candidate.workspaceCwd; + clientIdRef.current = candidate.clientId; + persistStableClientId(candidate.clientId!, candidate.sessionId); + setConnectionSynchronous((current) => { + const next = { + ...current, + status: 'connected' as const, + sessionId: candidate.sessionId, + clientId: candidate.clientId, + workspaceCwd: candidate.workspaceCwd, + capabilities, + loadingTranscript: undefined, + catchingUp: undefined, + error: undefined, + errorStatus: undefined, + missingSession: false, + }; + delete next.sessionTransition; + return next; + }); + desiredTransitionRef.current = undefined; + if (candidate.hasActivePrompt) { + settledRestoredActivePromptSessionsRef.current.add(candidate); + } + hasCurrentSessionActivePromptRef.current = () => false; + clearPassiveAssistantDoneTimer(passiveAssistantDoneTimerRef); + setPromptStatus('idle'); + liveJournalRepairRef.current?.controller?.abort(); + liveJournalRepairRef.current = undefined; + preparedRunnerRef.current = { session: candidate, capabilities }; + manualSessionClearRef.current = false; + setRestoreMode(intent.mode); + setRestoreSessionId(candidate.sessionId); + setRestoreWorkspaceCwd(candidate.workspaceCwd); + setRestoreSessionNonce((nonce) => nonce + 1); + settleCrossSessionIntent(intent); + retireAttachment(intent.source, intent); + return true; + }, + [maxBlocks, retireAttachment, setConnectionSynchronous, store], + ); + const pumpCrossSessionTransition = useCallback(() => { - if (rawTransitionRef.current) return; const intent = desiredTransitionRef.current; if (!intent) return; if ( @@ -3410,10 +3789,123 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { return; } const capabilities = - workspaceCapabilitiesRef.current ?? connectionRef.current.capabilities; + workspaceCapabilitiesRef.current ?? + sessionCapabilitiesRef.current ?? + connectionRef.current.capabilities; if (!capabilities?.features.includes(CLIENT_IDENTITY_FEATURE)) return; - const requestClientId = getStableClientId(clientId, intent.sessionId); - intent.targetClientId = requestClientId; + + if (intent.sameLogical && intent.candidate) { + const candidate = intent.candidate; + const snapshot = runnerControlRef.current?.snapshot(); + if ( + intent.capture?.invalidReason || + sessionRef.current !== intent.source || + intent.source.clientId !== intent.sourceClientId || + snapshot?.session !== intent.source || + snapshot.clientId !== intent.sourceClientId || + snapshot.eventEpoch !== candidate.eventEpoch + ) { + retireAttachment(candidate, intent); + exposeCrossSessionFailure( + intent, + new Error( + intent.capture?.invalidReason ?? + 'Current attachment changed before refresh commit', + ), + ); + return; + } + if (intent.deadlineAt !== undefined && Date.now() >= intent.deadlineAt) { + retireAttachment(candidate, intent); + exposeCrossSessionFailure( + intent, + new Error('Session transition timed out'), + ); + return; + } + if ( + !snapshot.ready || + snapshot.activeTurn || + hasCurrentSessionActivePromptRef.current() || + snapshot.processedEventId === undefined || + candidate.lastEventId === undefined || + snapshot.processedEventId < candidate.lastEventId || + (candidate.hasActivePrompt && + (snapshot.lastPromptTerminalEventId === undefined || + snapshot.lastPromptTerminalEventId <= candidate.lastEventId)) + ) { + return; + } + if ( + !commitSameSession( + intent, + candidate, + intent.candidateCapabilities ?? capabilities, + intent.capture, + ) + ) { + retireAttachment(candidate, intent); + exposeCrossSessionFailure( + intent, + new Error('Session refresh failed integrity validation'), + ); + } + return; + } + + if (rawTransitionRef.current) return; + if (intent.sameLogical) { + const snapshot = runnerControlRef.current?.snapshot(); + if ( + snapshot?.session !== intent.source || + snapshot.clientId !== intent.sourceClientId || + snapshot.eventEpoch === undefined || + snapshot.processedEventId === undefined || + !Number.isSafeInteger(snapshot.processedEventId) || + snapshot.processedEventId < 0 + ) { + exposeCrossSessionFailure( + intent, + new Error( + 'Current attachment cursor is unavailable; session was preserved', + ), + ); + return; + } + if ( + !snapshot.ready || + snapshot.activeTurn || + hasCurrentSessionActivePromptRef.current() + ) { + setConnectionSynchronous((current) => ({ + ...current, + sessionTransition: transitionState(intent, 'queued'), + })); + return; + } + const capture: SameSessionCapture = { + source: intent.source, + sourceClientId: intent.sourceClientId, + eventEpoch: snapshot.eventEpoch, + startEventId: snapshot.processedEventId, + lastCapturedEventId: snapshot.processedEventId, + bytes: 0, + events: [], + }; + intent.capture = capture; + if (intent.mode === 'load') { + runnerControlRef.current!.capture = capture; + } + armTransitionDeadline(intent, capabilities); + } + const requestClientId = intent.targetClientId; + if (!requestClientId) { + exposeCrossSessionFailure( + intent, + new Error('Session restore client identity is unavailable'), + ); + return; + } rawTransitionRef.current = intent; setConnectionSynchronous((current) => ({ ...current, @@ -3455,6 +3947,7 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { const latest = desiredTransitionRef.current; if ( !candidate.clientId || + (intent.sameLogical && candidate.clientId !== requestClientId) || candidate.sessionId !== intent.sessionId || normalizeWorkspaceIdentity(candidate.workspaceCwd) !== normalizeWorkspaceIdentity(intent.workspaceCwd) @@ -3468,6 +3961,30 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { } return; } + if ( + intent.sameLogical && + (candidate.eventEpoch === undefined || + candidate.eventEpoch !== intent.capture?.eventEpoch || + candidate.lastEventId === undefined || + !Number.isSafeInteger(candidate.lastEventId) || + candidate.lastEventId < 0 || + candidate.replayPartial || + candidate.replayError !== undefined || + candidate.replayDegraded || + (intent.mode === 'load' && + (!candidate.replaySnapshotComplete || + !intent.capture || + candidate.lastEventId < intent.capture.startEventId))) + ) { + retireAttachment(candidate, intent); + if (latest?.key === intent.key) { + exposeCrossSessionFailure( + latest, + new Error('Session refresh returned an incomplete snapshot'), + ); + } + return; + } if ( intent.resultSuperseded === true || latest?.key !== intent.key || @@ -3477,6 +3994,12 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { retireAttachment(candidate, intent); return; } + if (latest.sameLogical) { + latest.capture = intent.capture; + latest.candidate = candidate; + latest.candidateCapabilities = capabilities; + return; + } let staged: StagedCrossSession; try { staged = stageCrossSession({ @@ -3528,12 +4051,22 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { if (rawTransitionRef.current === intent) { rawTransitionRef.current = undefined; } + const capture = intent.capture; + const latest = desiredTransitionRef.current; + if ( + capture && + latest?.capture !== capture && + runnerControlRef.current?.capture === capture + ) { + runnerControlRef.current.capture = undefined; + } if (!retryScheduled) pumpTransitionRef.current(); }); }, [ + armTransitionDeadline, autoReconnect, - clientId, commitCrossSession, + commitSameSession, exposeCrossSessionFailure, maxBlocks, resolvedBaseUrl, @@ -3549,6 +4082,14 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { const intent = desiredTransitionRef.current; desiredTransitionRef.current = undefined; if (intent) { + if (intent.candidate) retireAttachment(intent.candidate, intent); + if ( + rawTransitionRef.current !== intent && + intent.capture && + runnerControlRef.current?.capture === intent.capture + ) { + runnerControlRef.current.capture = undefined; + } settleCrossSessionIntent( intent, new DOMException(reason, 'AbortError'), @@ -3561,7 +4102,7 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { return next; }); }, - [setConnectionSynchronous], + [retireAttachment, setConnectionSynchronous], ); cancelTransitionRef.current = cancelCrossSessionTransition; @@ -3571,7 +4112,9 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { startLegacy: () => Promise, ): Promise => { const capabilities = - workspaceCapabilitiesRef.current ?? connectionRef.current.capabilities; + workspaceCapabilitiesRef.current ?? + sessionCapabilitiesRef.current ?? + connectionRef.current.capabilities; const rejectPreflight = (error: Error) => { publishCrossSessionFailure(request, error); return Promise.reject(error); @@ -3583,6 +4126,14 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { ), ); } + if (sourceBoundOperationCountRef.current > 0) { + return rejectPreflight( + new DOMException( + 'Another session operation is already in progress', + 'InvalidStateError', + ), + ); + } if (!capabilities.features.includes(CLIENT_IDENTITY_FEATURE)) { return startLegacy(); } @@ -3601,18 +4152,19 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { ), ); } - if (sourceBoundOperationCountRef.current > 0) { + if (pendingSessionLoadRef.current) { return rejectPreflight( new DOMException( - 'Another session operation is already in progress', + 'Another session restore is already in progress', 'InvalidStateError', ), ); } - if (pendingSessionLoadRef.current) { - return rejectPreflight( + const pending = desiredTransitionRef.current; + if (request.sameLogical && pending && !pending.sameLogical) { + return Promise.reject( new DOMException( - 'Another session restore is already in progress', + 'A session switch is still preparing', 'InvalidStateError', ), ); @@ -3623,17 +4175,61 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { capabilities.features.includes(SESSION_TRANSCRIPT_PAGINATION_FEATURE) ? historyPageSizeRef.current : undefined; - const key = crossSessionKey( + const requestShape = crossSessionKey( request.sessionId, request.workspaceCwd, request.mode, effectiveHistoryPageSize, + undefined, ); const raw = rawTransitionRef.current; + const rawShape = raw + ? crossSessionKey( + raw.sessionId, + raw.workspaceCwd, + raw.mode, + raw.effectiveHistoryPageSize, + undefined, + ) + : undefined; + const targetClientId = + clientId ?? + (request.sameLogical + ? source.clientId + : rawShape === requestShape + ? raw?.targetClientId + : undefined) ?? + getStableClientId(undefined, request.sessionId); + const key = crossSessionKey( + request.sessionId, + request.workspaceCwd, + request.mode, + effectiveHistoryPageSize, + targetClientId, + ); if (raw && raw.key !== key) raw.resultSuperseded = true; const current = desiredTransitionRef.current; - if (current?.key === key) return current.promise; + if ( + current?.key === key && + request.signal === undefined && + current.signal === undefined + ) { + return current.promise; + } if (current) { + if (current.candidate) { + retireAttachment(current.candidate, current); + current.candidate = undefined; + } + const control = runnerControlRef.current; + if ( + rawTransitionRef.current !== current && + current.capture && + control && + control.capture === current.capture + ) { + control.capture = undefined; + } settleCrossSessionIntent( current, new DOMException( @@ -3648,11 +4244,6 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { resolve = res; reject = rej; }); - const timeouts = resolveSessionRestoreTimeouts(capabilities); - const deadlineAt = - timeouts.watchdogTimeoutMs === undefined - ? undefined - : Date.now() + timeouts.watchdogTimeoutMs; const intent: CrossSessionIntent = { key, ...(effectiveHistoryPageSize !== undefined @@ -3664,35 +4255,73 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { token: resolvedToken, lifecycle: lifecycleRef.current, environmentGeneration: environmentRef.current.generation, - ...(deadlineAt !== undefined ? { deadlineAt } : {}), + sourceClientId: source.clientId, + targetClientId, promise, resolve, reject, }; - if (timeouts.watchdogTimeoutMs !== undefined) { - intent.timeout = setTimeout(() => { - exposeCrossSessionFailure( + desiredTransitionRef.current = intent; + const adoptingRaw = rawTransitionRef.current?.key === key; + if (request.sameLogical && adoptingRaw) { + armTransitionDeadline(intent, capabilities); + } + if (request.signal) { + const abort = () => { + if (desiredTransitionRef.current !== intent) return; + desiredTransitionRef.current = undefined; + if (intent.candidate) retireAttachment(intent.candidate, intent); + const control = runnerControlRef.current; + if ( + rawTransitionRef.current !== intent && + control && + control.capture === intent.capture + ) { + control.capture = undefined; + } + settleCrossSessionIntent( intent, - new Error('Session transition timed out'), + request.signal?.reason ?? + new DOMException('Session transition cancelled', 'AbortError'), ); - }, timeouts.watchdogTimeoutMs); + setConnectionSynchronous((connectionState) => { + if (!connectionState.sessionTransition) return connectionState; + const next = { ...connectionState }; + delete next.sessionTransition; + return next; + }); + pumpTransitionRef.current(); + }; + request.signal.addEventListener('abort', abort, { once: true }); + intent.removeAbortListener = () => + request.signal?.removeEventListener('abort', abort); + if (request.signal.aborted) { + abort(); + return promise; + } } - desiredTransitionRef.current = intent; + if (!request.sameLogical) armTransitionDeadline(intent, capabilities); setConnectionSynchronous((connectionState) => ({ ...connectionState, sessionTransition: transitionState( request, - rawTransitionRef.current ? 'queued' : 'preparing', + adoptingRaw + ? 'preparing' + : request.sameLogical || rawTransitionRef.current + ? 'queued' + : 'preparing', ), })); pumpTransitionRef.current(); return promise; }, [ - exposeCrossSessionFailure, + armTransitionDeadline, + clientId, publishCrossSessionFailure, resolvedBaseUrl, resolvedToken, + retireAttachment, setConnectionSynchronous, ], ); @@ -3773,7 +4402,10 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { getConnection: () => connectionRef.current, addNotice, setConnection, - setPromptStatus, + setPromptStatus: (update) => { + setPromptStatus(update); + queueMicrotask(pumpTransitionRef.current); + }, setRestoreSessionId, setRestoreWorkspaceCwd, setRestoreMode, @@ -3784,6 +4416,8 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { cancelCrossSessionTransition, isCrossSessionTransitionPending: () => desiredTransitionRef.current !== undefined, + isDifferentLogicalTransitionPending: () => + desiredTransitionRef.current?.sameLogical === false, isSourceBoundOperationInFlight: () => sourceBoundOperationCountRef.current > 0, setSourceBoundOperationInFlight: (inFlight) => @@ -4020,6 +4654,7 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { string | undefined | typeof UNHANDLED_SESSION >(UNHANDLED_SESSION); const lastHandledWorkspaceRef = useRef(undefined); + const lastHandledClientIdRef = useRef(undefined); useEffect(() => { const targetWorkspaceCwd = @@ -4028,12 +4663,14 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { if ( lastHandledSessionIdRef.current === sessionId && normalizeWorkspaceIdentity(lastHandledWorkspaceRef.current) === - normalizeWorkspaceIdentity(targetWorkspaceCwd) + normalizeWorkspaceIdentity(targetWorkspaceCwd) && + lastHandledClientIdRef.current === clientId ) { return; } lastHandledSessionIdRef.current = sessionId; lastHandledWorkspaceRef.current = targetWorkspaceCwd; + lastHandledClientIdRef.current = clientId; if (sessionId === undefined && previousSessionId === undefined) return; @@ -4042,7 +4679,8 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { pending && (pending.sessionId !== sessionId || normalizeWorkspaceIdentity(pending.workspaceCwd) !== - normalizeWorkspaceIdentity(targetWorkspaceCwd)) + normalizeWorkspaceIdentity(targetWorkspaceCwd) || + (clientId !== undefined && pending.targetClientId !== clientId)) ) { cancelTransitionRef.current( 'Session transition cancelled by controlled target change', @@ -4050,11 +4688,25 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { } const currentSessionId = connectionRef.current.sessionId; - if ( + const currentSession = sessionRef.current; + const sameLogicalTarget = sessionId === currentSessionId && normalizeWorkspaceIdentity(targetWorkspaceCwd) === - normalizeWorkspaceIdentity(connectionRef.current.workspaceCwd) - ) { + normalizeWorkspaceIdentity(connectionRef.current.workspaceCwd); + const clientIdChanged = + sameLogicalTarget && + currentSession !== undefined && + clientId !== undefined && + currentSession.clientId !== clientId; + const controlledCapabilities = + workspaceCapabilitiesRef.current ?? + sessionCapabilitiesRef.current ?? + connectionRef.current.capabilities; + const needsTransactionalClientRebind = + clientIdChanged && + (controlledCapabilities === undefined || + controlledCapabilities.features.includes(CLIENT_IDENTITY_FEATURE)); + if (sameLogicalTarget && !needsTransactionalClientRebind) { if (connectionRef.current.sessionTransition?.phase === 'failed') { setConnectionSynchronous((current) => { if (current.sessionTransition?.phase !== 'failed') return current; @@ -4068,7 +4720,9 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { if (sessionId) controlledTransitionOriginRef.current = true; const request = sessionId - ? actions.loadSession(sessionId, { + ? (needsTransactionalClientRebind + ? actions.resumeSession + : actions.loadSession)(sessionId, { ...(targetWorkspaceCwd !== undefined ? { workspaceCwd: targetWorkspaceCwd } : {}), @@ -4085,7 +4739,13 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { error, ); }); - }, [actions, resolvedWorkspaceCwd, sessionId, setConnectionSynchronous]); + }, [ + actions, + clientId, + resolvedWorkspaceCwd, + sessionId, + setConnectionSynchronous, + ]); const ownerGuardValue = useMemo( () => ({ diff --git a/packages/webui/src/daemon/session/actions.test.ts b/packages/webui/src/daemon/session/actions.test.ts index 8afbad95945..1d01be3fb4c 100644 --- a/packages/webui/src/daemon/session/actions.test.ts +++ b/packages/webui/src/daemon/session/actions.test.ts @@ -262,6 +262,181 @@ describe('createDaemonSessionActions', () => { expect(createDetachedSession).not.toHaveBeenCalled(); }); + it('blocks a restore while active-session creation is in flight', async () => { + const existingSession = createMockSession('session-a'); + const nextSession = createMockSession('session-b'); + const created = createDeferred(); + existingSession.client.createOrAttachSession.mockReturnValueOnce( + created.promise, + ); + let sourceBoundOperationCount = 0; + const setSourceBoundOperationInFlight = vi.fn((inFlight: boolean) => { + sourceBoundOperationCount += inFlight ? 1 : -1; + }); + const beginCrossSessionTransition = vi.fn(async () => { + if (sourceBoundOperationCount > 0) { + throw new DOMException( + 'Another session operation is already in progress', + 'InvalidStateError', + ); + } + }); + const { actions } = createActionsHarness({ + beginCrossSessionTransition, + connection: { status: 'connected', sessionId: 'session-a' }, + isSourceBoundOperationInFlight: () => sourceBoundOperationCount > 0, + session: existingSession, + setSourceBoundOperationInFlight, + }); + + const create = actions.createSession(); + await expect(actions.loadSession('session-c')).rejects.toMatchObject({ + name: 'InvalidStateError', + }); + expect(beginCrossSessionTransition).toHaveBeenCalledOnce(); + + created.resolve(nextSession); + await expect(create).resolves.toBe(nextSession); + expect(setSourceBoundOperationInFlight.mock.calls).toEqual([ + [true], + [false], + ]); + }); + + it('keeps restore blocked after create times out until the raw request settles', async () => { + vi.useFakeTimers(); + try { + const existingSession = createMockSession('session-a'); + const nextSession = createMockSession('session-b'); + const created = createDeferred(); + existingSession.client.createOrAttachSession.mockReturnValueOnce( + created.promise, + ); + let sourceBoundOperationCount = 0; + const beginCrossSessionTransition = vi.fn(async () => { + if (sourceBoundOperationCount > 0) { + throw new DOMException( + 'Another session operation is already in progress', + 'InvalidStateError', + ); + } + }); + const setSourceBoundOperationInFlight = vi.fn((inFlight: boolean) => { + sourceBoundOperationCount += inFlight ? 1 : -1; + }); + const { actions } = createActionsHarness({ + beginCrossSessionTransition, + connection: { status: 'connected', sessionId: 'session-a' }, + isSourceBoundOperationInFlight: () => sourceBoundOperationCount > 0, + session: existingSession, + setSourceBoundOperationInFlight, + }); + + const createOutcome = actions + .createSession() + .catch((error: unknown) => error); + await vi.advanceTimersByTimeAsync(30_000); + await expect(createOutcome).resolves.toMatchObject({ + message: expect.stringContaining('Create session timed out'), + }); + await expect(actions.loadSession('session-c')).rejects.toMatchObject({ + name: 'InvalidStateError', + }); + expect(beginCrossSessionTransition).toHaveBeenCalledOnce(); + + created.resolve(nextSession); + await Promise.resolve(); + expect(existingSession.client.detachSession).toHaveBeenCalledWith( + nextSession.sessionId, + nextSession.clientId, + ); + await actions.loadSession('session-c'); + expect(beginCrossSessionTransition).toHaveBeenCalledTimes(2); + } finally { + vi.useRealTimers(); + } + }); + + it('retires a detached create that succeeds after its public timeout', async () => { + vi.useFakeTimers(); + try { + const nextSession = createMockSession('session-b'); + const created = createDeferred(); + const setSourceBoundOperationInFlight = vi.fn(); + const { actions, getConnection, sessionRef } = createActionsHarness({ + connection: { status: 'connected' }, + createDetachedSession: vi.fn(() => created.promise), + setSourceBoundOperationInFlight, + }); + + const createOutcome = actions + .createSession() + .catch((error: unknown) => error); + await vi.advanceTimersByTimeAsync(30_000); + await expect(createOutcome).resolves.toMatchObject({ + message: expect.stringContaining('Create session timed out'), + }); + expect(setSourceBoundOperationInFlight.mock.calls).toEqual([[true]]); + + created.resolve(nextSession as unknown as DaemonSessionClient); + await Promise.resolve(); + expect(nextSession.detach).toHaveBeenCalledOnce(); + expect(setSourceBoundOperationInFlight.mock.calls).toEqual([ + [true], + [false], + ]); + expect(sessionRef.current).toBeUndefined(); + expect(getConnection()).not.toHaveProperty('sessionId'); + } finally { + vi.useRealTimers(); + } + }); + + it('consumes a controlled origin when a source-bound operation blocks restore', async () => { + let controlled = true; + let sourceBound = true; + const beginCrossSessionTransition = vi.fn( + async (_request: { origin: 'action' | 'controlled' }) => { + if (sourceBound) { + throw new DOMException( + 'Another session operation is already in progress', + 'InvalidStateError', + ); + } + }, + ); + const source = createMockSession('session-a', 'client-a'); + const { actions } = createActionsHarness({ + beginCrossSessionTransition, + connection: { + status: 'connected', + sessionId: 'session-a', + workspaceCwd: '/workspace', + }, + getTransitionOrigin: () => { + const origin = controlled ? 'controlled' : 'action'; + controlled = false; + return origin; + }, + isSourceBoundOperationInFlight: () => sourceBound, + session: source, + }); + + await expect(actions.loadSession('session-b')).rejects.toMatchObject({ + name: 'InvalidStateError', + }); + sourceBound = false; + await actions.loadSession('session-b'); + + expect(beginCrossSessionTransition).toHaveBeenCalledWith( + expect.objectContaining({ origin: 'action' }), + expect.any(Function), + ); + expect( + beginCrossSessionTransition.mock.calls.map(([request]) => request.origin), + ).toEqual(['controlled', 'action']); + }); + it('creates a detached session when no active session exists', async () => { const nextSession = createMockSession('session-b'); const createDetachedSession = vi.fn(async () => nextSession); @@ -613,7 +788,7 @@ describe('createDaemonSessionActions', () => { const beginCrossSessionTransition = vi.fn(async () => undefined); const { actions, pendingSessionLoadRef } = createActionsHarness({ beginCrossSessionTransition, - isCrossSessionTransitionPending: () => true, + isDifferentLogicalTransitionPending: () => true, connection: { status: 'connected', sessionId: 'session-a', @@ -625,11 +800,33 @@ describe('createDaemonSessionActions', () => { await expect(actions.loadSession('session-a')).rejects.toMatchObject({ name: 'InvalidStateError', }); + await expect( + actions.reloadSession(new AbortController().signal, { + replaySource: 'configured', + }), + ).rejects.toMatchObject({ name: 'InvalidStateError' }); expect(beginCrossSessionTransition).not.toHaveBeenCalled(); expect(pendingSessionLoadRef.current).toBeUndefined(); expect(source.detach).not.toHaveBeenCalled(); }); + it('keeps an empty-owner load on the bootstrap path', () => { + const beginCrossSessionTransition = vi.fn(async () => undefined); + const { actions, pendingSessionLoadRef } = createActionsHarness({ + beginCrossSessionTransition, + session: undefined, + }); + + void actions.loadSession('session-a').catch(() => undefined); + + expect(beginCrossSessionTransition).not.toHaveBeenCalled(); + expect(pendingSessionLoadRef.current).toMatchObject({ + sessionId: 'session-a', + mode: 'load', + }); + clearTimeout(pendingSessionLoadRef.current?.timeout); + }); + it('consumes the controlled origin when a switch uses the legacy path', () => { const getTransitionOrigin = vi.fn(() => 'controlled' as const); const { actions, pendingSessionLoadRef } = createActionsHarness({ @@ -1332,6 +1529,7 @@ function createActionsHarness( setSourceBoundOperationInFlight?: ReturnType; isSourceBoundOperationInFlight?: () => boolean; isCrossSessionTransitionPending?: () => boolean; + isDifferentLogicalTransitionPending?: () => boolean; setPromptStatus?: ReturnType; hasSessionActivePrompt?: () => boolean; } = {}, @@ -1386,6 +1584,8 @@ function createActionsHarness( clearLiveJournalRepair: opts.clearLiveJournalRepair, beginCrossSessionTransition: opts.beginCrossSessionTransition, isCrossSessionTransitionPending: opts.isCrossSessionTransitionPending, + isDifferentLogicalTransitionPending: + opts.isDifferentLogicalTransitionPending, isSourceBoundOperationInFlight: opts.isSourceBoundOperationInFlight, getTransitionOrigin: opts.getTransitionOrigin, setSourceBoundOperationInFlight: opts.setSourceBoundOperationInFlight, diff --git a/packages/webui/src/daemon/session/actions.ts b/packages/webui/src/daemon/session/actions.ts index 718f03a1b51..b50ddbcadbc 100644 --- a/packages/webui/src/daemon/session/actions.ts +++ b/packages/webui/src/daemon/session/actions.ts @@ -125,11 +125,14 @@ export interface CreateDaemonSessionActionsArgs { mode: 'load' | 'resume'; workspaceCwd?: string; origin: 'action' | 'controlled'; + sameLogical?: boolean; + signal?: AbortSignal; }, startLegacy: () => Promise, ) => Promise; cancelCrossSessionTransition?: (reason: string) => void; isCrossSessionTransitionPending?: () => boolean; + isDifferentLogicalTransitionPending?: () => boolean; isSourceBoundOperationInFlight?: () => boolean; setSourceBoundOperationInFlight?: (inFlight: boolean) => void; getTransitionOrigin?: () => 'action' | 'controlled'; @@ -199,6 +202,7 @@ export function createDaemonSessionActions({ beginCrossSessionTransition, cancelCrossSessionTransition = () => undefined, isCrossSessionTransitionPending = () => false, + isDifferentLogicalTransitionPending = () => false, isSourceBoundOperationInFlight = () => false, setSourceBoundOperationInFlight = () => undefined, getTransitionOrigin = () => 'action', @@ -457,6 +461,7 @@ export function createDaemonSessionActions({ ); } const origin = getTransitionOrigin(); + const sourceBoundOperationInFlight = isSourceBoundOperationInFlight(); const startLegacy = () => startLegacySessionSwitch( sessionId, @@ -466,17 +471,26 @@ export function createDaemonSessionActions({ replaySource, ); const current = sessionRef.current; + if ( + sourceBoundOperationInFlight && + (current === undefined || + replaySource === 'memory' || + !beginCrossSessionTransition) + ) { + return Promise.reject( + new DOMException( + 'Another session operation is already in progress', + 'InvalidStateError', + ), + ); + } const targetWorkspace = workspaceCwd ?? getConnection().workspaceCwd; const crossLogicalTarget = current !== undefined && (current.sessionId !== sessionId || normalizeWorkspaceIdentity(current.workspaceCwd) !== normalizeWorkspaceIdentity(targetWorkspace)); - if ( - !crossLogicalTarget && - replaySource === undefined && - isCrossSessionTransitionPending() - ) { + if (!crossLogicalTarget && isDifferentLogicalTransitionPending()) { return Promise.reject( new DOMException( 'A session switch is still preparing', @@ -485,8 +499,8 @@ export function createDaemonSessionActions({ ); } if ( - !crossLogicalTarget || - replaySource !== undefined || + current === undefined || + replaySource === 'memory' || !beginCrossSessionTransition ) { return startLegacy(); @@ -499,6 +513,8 @@ export function createDaemonSessionActions({ ? { workspaceCwd: targetWorkspace } : {}), origin, + sameLogical: !crossLogicalTarget, + ...(signal ? { signal } : {}), }, startLegacy, ); @@ -854,7 +870,7 @@ export function createDaemonSessionActions({ }, async reloadSession(signal, options) { - requireStableSession(); + if (options?.replaySource === 'memory') requireStableSession(); const session = requireSessionForAction( addNotice, sessionRef.current, @@ -882,6 +898,35 @@ export function createDaemonSessionActions({ branch?: { name: string }; }) { requireStableSession(); + let rawCreateStarted = false; + let rawCreateSettled = false; + let retireLateResult = false; + const trackCreate = ( + request: Promise, + retire: (created: T) => Promise, + ) => { + rawCreateStarted = true; + setSourceBoundOperationInFlight(true); + void request.then( + (created) => { + rawCreateSettled = true; + setSourceBoundOperationInFlight(false); + if (retireLateResult) { + void retire(created).catch((error: unknown) => { + console.warn( + '[DaemonSessionActions] detach after timed-out create failed:', + error, + ); + }); + } + }, + () => { + rawCreateSettled = true; + setSourceBoundOperationInFlight(false); + }, + ); + return request; + }; try { manualSessionClearRef.current = false; // Fold the initial approval mode into the create request so the daemon @@ -909,13 +954,20 @@ export function createDaemonSessionActions({ : undefined; if (activeSession) { const nextSession = await withActionTimeout( - activeSession.client.createOrAttachSession({ - ...getCreateSessionRequest(), - ...(options?.workspaceCwd !== undefined - ? { workspaceCwd: options.workspaceCwd } - : {}), - ...requestOverrides, - }), + trackCreate( + activeSession.client.createOrAttachSession({ + ...getCreateSessionRequest(), + ...(options?.workspaceCwd !== undefined + ? { workspaceCwd: options.workspaceCwd } + : {}), + ...requestOverrides, + }), + (created) => + activeSession.client.detachSession( + created.sessionId, + created.clientId, + ), + ), 'Create session timed out', ); persistStableClientId(nextSession.clientId, nextSession.sessionId); @@ -923,7 +975,10 @@ export function createDaemonSessionActions({ } const nextSession = await withActionTimeout( - createDetachedSession(options?.workspaceCwd, requestOverrides), + trackCreate( + createDetachedSession(options?.workspaceCwd, requestOverrides), + (created) => created.detach(), + ), 'Create session timed out', ); if (manualSessionClearRef.current) { @@ -955,6 +1010,7 @@ export function createDaemonSessionActions({ })); return nextSession; } catch (error) { + if (rawCreateStarted && !rawCreateSettled) retireLateResult = true; throw dispatchActionError( addNotice, `Create session failed${