From 8e652e1ef7c8f67af8045e6c7af4ea7c04bf3547 Mon Sep 17 00:00:00 2001 From: ytahdn <1294726970@qq.com> Date: Mon, 22 Jun 2026 19:18:01 +0800 Subject: [PATCH] feat(web-shell): support daemon session branching (#5613) * feat(web-shell): support daemon session branching * fix(webui): keep branch events on new session * fix(web-shell): harden fork session handling * fix(cli): neutralize fork history marker --------- Co-authored-by: ytahdn --- packages/acp-bridge/src/bridge.test.ts | 96 +++++++++++++ packages/acp-bridge/src/bridge.ts | 84 ++++++++++- packages/acp-bridge/src/bridgeTypes.ts | 17 +++ packages/acp-bridge/src/status.ts | 1 + .../cli/src/acp-integration/acpAgent.test.ts | 81 +++++++++++ packages/cli/src/acp-integration/acpAgent.ts | 113 +++++++++++++++ packages/cli/src/serve/server.test.ts | 135 ++++++++++++++++-- packages/cli/src/serve/server.ts | 29 ++++ .../core/src/services/sessionService.test.ts | 78 ++++++++++ packages/core/src/services/sessionService.ts | 24 ++-- packages/sdk-typescript/scripts/build.js | 3 +- .../sdk-typescript/src/daemon/DaemonClient.ts | 23 +++ .../src/daemon/DaemonSessionClient.ts | 9 ++ packages/sdk-typescript/src/daemon/index.ts | 2 + packages/sdk-typescript/src/daemon/types.ts | 10 ++ .../src/daemon/ui/normalizer.ts | 48 +++++++ .../sdk-typescript/src/daemon/ui/terminal.ts | 12 ++ .../src/daemon/ui/transcript.ts | 73 ++++++++++ .../sdk-typescript/src/daemon/ui/types.ts | 18 +++ .../sdk-typescript/test/unit/daemonUi.test.ts | 111 ++++++++++++++ packages/web-shell/client/App.tsx | 50 +++++++ .../adapters/transcriptToMessages.test.ts | 35 +++++ .../client/adapters/transcriptToMessages.ts | 80 ++++++++++- .../client/constants/localCommands.ts | 10 ++ .../client/hooks/useBackgroundTasks.ts | 7 + .../web-shell/client/hooks/useMessages.ts | 5 +- packages/web-shell/client/i18n.tsx | 20 +++ .../session/DaemonSessionProvider.test.tsx | 66 +++++++++ packages/webui/src/daemon/session/actions.ts | 66 +++++++++ packages/webui/src/daemon/session/types.ts | 7 + 30 files changed, 1286 insertions(+), 27 deletions(-) diff --git a/packages/acp-bridge/src/bridge.test.ts b/packages/acp-bridge/src/bridge.test.ts index 35eac91e98..3b187a7217 100644 --- a/packages/acp-bridge/src/bridge.test.ts +++ b/packages/acp-bridge/src/bridge.test.ts @@ -34,6 +34,7 @@ import { RestoreInProgressError, SessionShellClientRequiredError, SessionShellDisabledError, + SessionBusyError, SessionNotFoundError, WorkspaceMismatchError, } from './bridgeErrors.js'; @@ -2946,6 +2947,70 @@ describe('createAcpSessionBridge', () => { await bridge.shutdown(); }); + it('publishes session_branched only on the new session stream', async () => { + const factory: ChannelFactory = async () => + makeChannel({ + extMethodImpl: async (method) => { + if (method !== 'qwen/control/session/branch') return {}; + return { newSessionId: 'branch-1', title: 'Branch 1' }; + }, + resumeSessionImpl: () => ({}), + }).channel; + const bridge = makeBridge({ channelFactory: factory }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + const sourceAbort = new AbortController(); + const sourceIter = bridge + .subscribeEvents(session.sessionId, { signal: sourceAbort.signal }) + [Symbol.asyncIterator](); + + const branch = await bridge.branchSession(session.sessionId, { + name: 'Branch 1', + }); + + const sourceEvent = await Promise.race([ + sourceIter.next(), + new Promise<'timeout'>((resolve) => setTimeout(resolve, 25, 'timeout')), + ]); + expect(sourceEvent).toBe('timeout'); + sourceAbort.abort(); + + const sourceReplayAbort = new AbortController(); + const sourceReplayIter = bridge + .subscribeEvents(session.sessionId, { + lastEventId: 0, + signal: sourceReplayAbort.signal, + }) + [Symbol.asyncIterator](); + const sourceReplayEvent = await Promise.race([ + sourceReplayIter.next(), + new Promise<'timeout'>((resolve) => setTimeout(resolve, 25, 'timeout')), + ]); + expect(sourceReplayEvent).toMatchObject({ + value: { type: 'replay_complete' }, + }); + const sourceReplayNext = await Promise.race([ + sourceReplayIter.next(), + new Promise<'timeout'>((resolve) => setTimeout(resolve, 25, 'timeout')), + ]); + expect(sourceReplayNext).toBe('timeout'); + sourceReplayAbort.abort(); + + const branchedIter = bridge + .subscribeEvents(branch.sessionId, { lastEventId: 0 }) + [Symbol.asyncIterator](); + const replayed = await branchedIter.next(); + expect(replayed.value).toMatchObject({ + type: 'session_branched', + data: { + sourceSessionId: session.sessionId, + newSessionId: branch.sessionId, + displayName: 'Branch 1', + }, + }); + + await bridge.shutdown(); + }); + it('a failed prompt does not poison the queue for subsequent prompts', async () => { let promptCount = 0; const handles: ChannelHandle[] = []; @@ -2985,6 +3050,37 @@ describe('createAcpSessionBridge', () => { await bridge.shutdown(); }); + it('rejects launchSessionForkAgent while a prompt is active', async () => { + let releasePrompt: (() => void) | undefined; + const handle = makeChannel({ + promptImpl: async () => + new Promise((resolve) => { + releasePrompt = () => resolve({ stopReason: 'end_turn' }); + }), + }); + const bridge = makeBridge({ channelFactory: async () => handle.channel }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + + const active = bridge.sendPrompt(session.sessionId, { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'active' }], + }); + await vi.waitFor(() => expect(releasePrompt).toBeDefined()); + + await expect( + bridge.launchSessionForkAgent(session.sessionId, 'review this'), + ).rejects.toBeInstanceOf(SessionBusyError); + expect( + handle.agent.extMethodCalls.some( + (call) => call.method === 'qwen/control/session/fork_agent', + ), + ).toBe(false); + + releasePrompt!(); + await expect(active).resolves.toEqual({ stopReason: 'end_turn' }); + await bridge.shutdown(); + }); + it('throws SessionNotFoundError for unknown session ids', async () => { const bridge = makeBridge({ channelFactory: async () => { diff --git a/packages/acp-bridge/src/bridge.ts b/packages/acp-bridge/src/bridge.ts index 402f216683..d90e8eedef 100644 --- a/packages/acp-bridge/src/bridge.ts +++ b/packages/acp-bridge/src/bridge.ts @@ -3394,14 +3394,14 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { let restored; try { - restored = await restoreSession('resume', { + restored = await restoreSession('load', { sessionId: result.newSessionId, workspaceCwd: boundWorkspace, clientId: context?.clientId, }); } catch (restoreErr) { writeStderrLine( - `qwen serve: branchSession resume failed for ${result.newSessionId}, attempting cleanup...`, + `qwen serve: branchSession load failed for ${result.newSessionId}, attempting cleanup...`, ); try { await ci.connection.extMethod( @@ -3429,8 +3429,9 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { data: eventData, ...(originatorClientId ? { originatorClientId } : {}), }; - entry.events.publish(branchEnvelope); - broadcastWorkspaceEvent(branchEnvelope, sessionId); + // The branch announcement belongs to the new session only. Publishing + // it on the source session would persist in that session's replay ring. + newEntry?.events.publish(branchEnvelope); return { ...restored, @@ -4350,6 +4351,81 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { }; }, + async launchSessionForkAgent(sessionId, directive, context) { + const entry = byId.get(sessionId); + if (!entry) throw new SessionNotFoundError(sessionId); + const info = channelInfoForEntry(entry); + if (!info || info.isDying) throw new SessionNotFoundError(sessionId); + resolveTrustedClientId(entry, context?.clientId); + + const trimmed = directive.trim(); + if (!trimmed) { + throw new Error('Fork directive is required'); + } + if (entry.pendingPromptCount > 0 || entry.promptActive) { + throw new SessionBusyError( + sessionId, + 'Cannot fork while a response or tool call is in progress', + ); + } + return entry.promptQueue.then(async () => { + if (entry.pendingPromptCount > 0 || entry.promptActive) { + throw new SessionBusyError( + sessionId, + 'Cannot fork while a response or tool call is in progress', + ); + } + + opts.onDiagnosticLine?.( + `qwen serve: launchSessionForkAgent requested for session=${sessionId}`, + 'info', + ); + + let response: { + description?: string; + launched?: boolean; + }; + try { + response = (await Promise.race([ + withTimeout( + entry.connection.extMethod( + SERVE_CONTROL_EXT_METHODS.sessionForkAgent, + { + sessionId, + directive: trimmed, + }, + ), + initTimeoutMs, + SERVE_CONTROL_EXT_METHODS.sessionForkAgent, + ), + getTransportClosedReject(entry), + ])) as { + description?: string; + launched?: boolean; + }; + } catch (error) { + opts.onDiagnosticLine?.( + `qwen serve: launchSessionForkAgent failed for session=${sessionId}: ${ + error instanceof Error ? error.message : String(error) + }`, + 'warn', + ); + throw error; + } + + const result = { + sessionId: entry.sessionId, + description: response.description ?? trimmed.slice(0, 60), + launched: response.launched === true, + }; + opts.onDiagnosticLine?.( + `qwen serve: launchSessionForkAgent completed for session=${sessionId} launched=${result.launched}`, + 'info', + ); + return result; + }); + }, + async executeShellCommand( sessionId, command, diff --git a/packages/acp-bridge/src/bridgeTypes.ts b/packages/acp-bridge/src/bridgeTypes.ts index 16a51f7a31..42429746a6 100644 --- a/packages/acp-bridge/src/bridgeTypes.ts +++ b/packages/acp-bridge/src/bridgeTypes.ts @@ -113,6 +113,12 @@ export interface BridgeBranchedSession extends BridgeRestoredSession { forkedFrom: { sessionId: string; displayName: string }; } +export interface BridgeForkAgentResult { + sessionId: string; + description: string; + launched: boolean; +} + /** Sparse summary used by `GET /workspace/:id/sessions`. */ export interface BridgeSessionSummary { sessionId: string; @@ -572,6 +578,17 @@ export interface AcpSessionBridge { context?: BridgeClientRequestContext, ): Promise<{ sessionId: string; answer: string | null }>; + /** + * Launch a background fork agent that inherits the live session's current + * conversation context. This is CLI `/fork`, not ACP `session/fork` + * (which maps to `/branch`). + */ + launchSessionForkAgent( + sessionId: string, + directive: string, + context?: BridgeClientRequestContext, + ): Promise; + /** * Queue a mid-turn user message for the running turn. The ACP child drains * it between tool batches via the `craft/drainMidTurnQueue` ext-method so diff --git a/packages/acp-bridge/src/status.ts b/packages/acp-bridge/src/status.ts index cb1ec00955..7f1ead748b 100644 --- a/packages/acp-bridge/src/status.ts +++ b/packages/acp-bridge/src/status.ts @@ -125,6 +125,7 @@ export const SERVE_CONTROL_EXT_METHODS = { sessionClose: 'qwen/control/session/close', sessionApprovalMode: 'qwen/control/session/approval_mode', sessionBranch: 'qwen/control/session/branch', + sessionForkAgent: 'qwen/control/session/fork_agent', sessionRecap: 'qwen/control/session/recap', sessionBtw: 'qwen/control/session/btw', sessionShellHistory: 'qwen/control/session/shell_history', diff --git a/packages/cli/src/acp-integration/acpAgent.test.ts b/packages/cli/src/acp-integration/acpAgent.test.ts index f10d45056f..501dea50b8 100644 --- a/packages/cli/src/acp-integration/acpAgent.test.ts +++ b/packages/cli/src/acp-integration/acpAgent.test.ts @@ -125,6 +125,10 @@ vi.mock('@qwen-code/qwen-code-core', () => ({ USE_GEMINI: 'gemini', USE_VERTEX_AI: 'vertex-ai', }, + ToolNames: { + AGENT: 'agent', + }, + FORK_SUBAGENT_TYPE: 'fork', ALL_PROVIDERS: [ { id: 'deepseek', @@ -2103,6 +2107,83 @@ describe('QwenAgent MCP SSE/HTTP support', () => { await agentPromise; }); + it('launches fork agents with neutral history text', async () => { + const sessionId = '11111111-1111-1111-1111-111111111111'; + const innerConfig = await setupSessionMocks(sessionId); + const addHistory = vi.fn(); + const execute = vi.fn().mockResolvedValue({ llmContent: 'ok' }); + const build = vi.fn().mockReturnValue({ execute }); + const directive = `review this\nbranch ${'x'.repeat(220)}`; + const collapsed = `review this branch ${'x'.repeat(220)}`; + + Object.assign(innerConfig, { + getGeminiClient: vi.fn().mockReturnValue({ + isInitialized: vi.fn().mockReturnValue(true), + initialize: vi.fn().mockResolvedValue(undefined), + waitForMcpReady: vi.fn().mockResolvedValue(undefined), + getHistoryShallow: vi + .fn() + .mockReturnValue([{ role: 'user', parts: [{ text: 'before' }] }]), + addHistory, + }), + getToolRegistry: vi.fn().mockReturnValue({ + getTool: vi.fn((name: string) => + name === 'agent' ? { build } : undefined, + ), + }), + }); + + const agentPromise = runAcpAgent( + mockConfig, + makeSessionSettings(), + mockArgv, + ); + await vi.waitFor(() => expect(capturedAgentFactory).toBeDefined()); + + const agent = capturedAgentFactory!({ + get closed() { + return mockConnectionState.promise; + }, + }) as AgentLike; + + await agent.newSession({ cwd: '/tmp', mcpServers: [] }); + await expect( + agent.extMethod(SERVE_CONTROL_EXT_METHODS.sessionForkAgent, { + sessionId, + directive, + }), + ).resolves.toEqual({ + sessionId, + description: `${collapsed.slice(0, 57)}…`, + launched: true, + }); + + expect(build).toHaveBeenCalledWith({ + description: `${collapsed.slice(0, 57)}…`, + prompt: directive.trim(), + subagent_type: 'fork', + run_in_background: true, + }); + expect(execute).toHaveBeenCalledTimes(1); + expect(addHistory).toHaveBeenCalledWith({ + role: 'user', + parts: [ + { + text: `User launched a background fork via /fork. Directive (truncated): ${collapsed.slice( + 0, + 197, + )}…`, + }, + ], + }); + expect(addHistory.mock.calls[0]?.[0]?.parts[0]?.text).not.toContain( + '[system]', + ); + + mockConnectionState.resolve(); + await agentPromise; + }); + it('allows cancelling paused agent tasks', async () => { const sessionId = '11111111-1111-1111-1111-111111111111'; const innerConfig = await setupSessionMocks(sessionId); diff --git a/packages/cli/src/acp-integration/acpAgent.ts b/packages/cli/src/acp-integration/acpAgent.ts index b569446534..df6292db96 100644 --- a/packages/cli/src/acp-integration/acpAgent.ts +++ b/packages/cli/src/acp-integration/acpAgent.ts @@ -58,9 +58,12 @@ import { redactUrlCredentials, computeUniqueBranchTitle, unregisterGoalHook, + ToolNames, + FORK_SUBAGENT_TYPE, } from '@qwen-code/qwen-code-core'; import { randomUUID } from 'node:crypto'; import type { + AgentParams, ApprovalMode, Config, ConversationRecord, @@ -215,6 +218,32 @@ const debugLogger = createDebugLogger('ACP_AGENT'); // aborts before the bridge's backstop timer fires. const BTW_CHILD_TIMEOUT_MS = 55_000; +function collapseForkDirective(directive: string, maxLength: number): string { + const oneLine = directive.replace(/\s+/g, ' ').trim(); + return oneLine.length > maxLength + ? `${oneLine.slice(0, maxLength - 3)}…` + : oneLine; +} + +function deriveForkDescription(directive: string): string { + return collapseForkDirective(directive, 60); +} + +function truncateForkDirectiveForHistory(directive: string): string { + return collapseForkDirective(directive, 200); +} + +function hasFailedDisplayStatus( + display: unknown, +): display is { status: 'failed' } { + return ( + display !== null && + typeof display === 'object' && + 'status' in display && + (display as { status?: unknown }).status === 'failed' + ); +} + function sanitizeProviderBaseUrl(baseUrl: string): string { const scheme = baseUrl.match(/^[A-Za-z][A-Za-z\d+.-]*:\/\//); if (!scheme) { @@ -5733,6 +5762,90 @@ class QwenAgent implements Agent { } return { sessionId, answer: result.text || null }; } + case SERVE_CONTROL_EXT_METHODS.sessionForkAgent: { + const sessionId = params['sessionId']; + if (typeof sessionId !== 'string' || sessionId.length === 0) { + throw RequestError.invalidParams( + undefined, + 'Invalid or missing sessionId', + ); + } + const directive = + typeof params['directive'] === 'string' ? params['directive'] : ''; + const trimmed = directive.trim(); + if (!trimmed) { + throw RequestError.invalidParams( + undefined, + 'Invalid or missing directive', + ); + } + + const session = this.sessionOrThrow(sessionId); + const config = session.getConfig(); + if (!config.getModel()) { + throw RequestError.invalidParams(undefined, 'No model configured.'); + } + + let hasHistory = false; + try { + hasHistory = + (config.getGeminiClient().getHistoryShallow() ?? []).length > 0; + } catch (error) { + debugLogger.debug('Failed to read history before /fork:', error); + } + if (!hasHistory) { + throw RequestError.invalidParams( + undefined, + 'Cannot fork before the first conversation turn.', + ); + } + + const agentTool = config.getToolRegistry().getTool(ToolNames.AGENT); + if (!agentTool) { + throw RequestError.invalidParams( + undefined, + 'The agent tool is unavailable; cannot fork.', + ); + } + + const description = deriveForkDescription(trimmed); + const agentParams: AgentParams = { + description, + prompt: trimmed, + subagent_type: FORK_SUBAGENT_TYPE, + run_in_background: true, + }; + const result = await agentTool + .build(agentParams) + .execute(new AbortController().signal); + if (hasFailedDisplayStatus(result?.returnDisplay)) { + const reason = + typeof result.llmContent === 'string' && result.llmContent.trim() + ? result.llmContent.trim() + : 'the background agent could not be started.'; + throw RequestError.invalidParams( + undefined, + `Failed to launch fork: ${reason}`, + ); + } + + try { + config.getGeminiClient().addHistory({ + role: 'user', + parts: [ + { + text: `User launched a background fork via /fork. Directive (truncated): ${truncateForkDirectiveForHistory( + trimmed, + )}`, + }, + ], + }); + } catch (error) { + debugLogger.debug('Failed to record fork event in history:', error); + } + + return { sessionId, description, launched: true }; + } case SERVE_CONTROL_EXT_METHODS.sessionShellHistory: { const sessionId = params['sessionId']; if (typeof sessionId !== 'string' || sessionId.length === 0) { diff --git a/packages/cli/src/serve/server.test.ts b/packages/cli/src/serve/server.test.ts index d2d915e10d..d73d377416 100644 --- a/packages/cli/src/serve/server.test.ts +++ b/packages/cli/src/serve/server.test.ts @@ -62,6 +62,7 @@ import { RestoreInProgressError, SessionShellClientRequiredError, SessionShellDisabledError, + SessionBusyError, SessionLimitExceededError, SessionNotFoundError, WorkspaceMismatchError, @@ -382,6 +383,11 @@ interface FakeBridgeOpts { sessionId: string, context?: BridgeClientRequestContext, ) => Promise<{ sessionId: string; recap: string | null }>; + launchSessionForkAgentImpl?: ( + sessionId: string, + directive: string, + context?: BridgeClientRequestContext, + ) => Promise<{ sessionId: string; description: string; launched: boolean }>; setToolEnabledImpl?: ( toolName: string, enabled: boolean, @@ -552,6 +558,11 @@ interface FakeBridge extends AcpSessionBridge { sessionId: string; context?: BridgeClientRequestContext; }>; + forkCalls: Array<{ + sessionId: string; + directive: string; + context?: BridgeClientRequestContext; + }>; setToolEnabledCalls: Array<{ toolName: string; enabled: boolean; @@ -841,6 +852,14 @@ function fakeBridge(opts: FakeBridgeOpts = {}): FakeBridge { sessionId, recap: 'Default fake recap.', })); + const forkCalls: FakeBridge['forkCalls'] = []; + const launchSessionForkAgentImpl = + opts.launchSessionForkAgentImpl ?? + (async (sessionId: string, directive: string) => ({ + sessionId, + description: directive.slice(0, 60), + launched: true, + })); const setToolEnabledCalls: FakeBridge['setToolEnabledCalls'] = []; const setToolEnabledImpl = opts.setToolEnabledImpl ?? @@ -963,6 +982,7 @@ function fakeBridge(opts: FakeBridgeOpts = {}): FakeBridge { setApprovalModeCalls, shellCalls, generateSessionRecapCalls, + forkCalls, setToolEnabledCalls, initWorkspaceCalls, restartMcpServerCalls, @@ -1188,6 +1208,14 @@ function fakeBridge(opts: FakeBridgeOpts = {}): FakeBridge { async generateSessionBtw(sessionId, _question, _signal, _context) { return { sessionId, answer: 'mock btw answer' }; }, + async launchSessionForkAgent(sessionId, directive, context) { + forkCalls.push({ + sessionId, + directive, + ...(context ? { context } : {}), + }); + return launchSessionForkAgentImpl(sessionId, directive, context); + }, enqueueMidTurnMessage(sessionId, message, context) { enqueueMidTurnCalls.push({ sessionId, @@ -6093,6 +6121,94 @@ describe('createServeApp', () => { }); }); + describe('POST /session/:id/fork', () => { + it('202 with directive and client identity on success', async () => { + const bridge = fakeBridge(); + const app = createServeApp(baseOpts, undefined, { bridge }); + + const res = await request(app) + .post('/session/session-A/fork') + .set('Host', `127.0.0.1:${baseOpts.port}`) + .set('X-Qwen-Client-Id', 'client-1') + .send({ directive: 'review the current code' }); + + expect(res.status).toBe(202); + expect(res.body).toEqual({ + sessionId: 'session-A', + description: 'review the current code', + launched: true, + }); + expect(bridge.forkCalls).toEqual([ + { + sessionId: 'session-A', + directive: 'review the current code', + context: { clientId: 'client-1' }, + }, + ]); + }); + + it('400 when directive is missing or empty', async () => { + const bridge = fakeBridge(); + const app = createServeApp(baseOpts, undefined, { bridge }); + + const missing = await request(app) + .post('/session/session-A/fork') + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({}); + expect(missing.status).toBe(400); + expect(missing.body.code).toBe('missing_directive'); + + const empty = await request(app) + .post('/session/session-A/fork') + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ directive: ' ' }); + expect(empty.status).toBe(400); + expect(empty.body.code).toBe('missing_directive'); + expect(bridge.forkCalls).toEqual([]); + }); + + it('404 when bridge throws SessionNotFoundError', async () => { + const bridge = fakeBridge({ + launchSessionForkAgentImpl: async (sessionId) => { + throw new SessionNotFoundError(sessionId); + }, + }); + const app = createServeApp(baseOpts, undefined, { bridge }); + + const res = await request(app) + .post('/session/missing/fork') + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ directive: 'review the current code' }); + + expect(res.status).toBe(404); + expect(res.body.sessionId).toBe('missing'); + }); + + it('409 when bridge reports the session is busy', async () => { + const bridge = fakeBridge({ + launchSessionForkAgentImpl: async (sessionId) => { + throw new SessionBusyError( + sessionId, + 'Cannot fork while a response or tool call is in progress', + ); + }, + }); + const app = createServeApp(baseOpts, undefined, { bridge }); + + const res = await request(app) + .post('/session/session-A/fork') + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ directive: 'review the current code' }); + + expect(res.status).toBe(409); + expect(res.body).toMatchObject({ + code: 'session_busy', + sessionId: 'session-A', + }); + expect(res.headers['retry-after']).toBe('5'); + }); + }); + describe('POST /session/:id/language', () => { it('200 with language result on success', async () => { const bridge = fakeBridge(); @@ -10936,15 +11052,16 @@ describe('auth device-flow routes', () => { // through `sendBridgeError`'s generic 500 path. Build a fake // provider whose start always throws. const { UpstreamDeviceFlowError } = await import('./auth/device-flow.js'); - const failingProvider: import('./auth/device-flow.js').DeviceFlowProvider = { - providerId: 'qwen-oauth', - async start() { - throw new UpstreamDeviceFlowError('mocked upstream outage'); - }, - async poll() { - return { kind: 'pending' as const }; - }, - }; + const failingProvider: import('./auth/device-flow.js').DeviceFlowProvider = + { + providerId: 'qwen-oauth', + async start() { + throw new UpstreamDeviceFlowError('mocked upstream outage'); + }, + async poll() { + return { kind: 'pending' as const }; + }, + }; const bridge = fakeBridge(); const app = createServeApp({ ...baseOpts, token: 'tkn' }, undefined, { bridge, diff --git a/packages/cli/src/serve/server.ts b/packages/cli/src/serve/server.ts index 09b8cb80cc..e0cf66e8cb 100644 --- a/packages/cli/src/serve/server.ts +++ b/packages/cli/src/serve/server.ts @@ -2999,6 +2999,35 @@ export function createServeApp( } }); + app.post('/session/:id/fork', mutate(), async (req, res) => { + const sessionId = requireSessionId(req, res); + if (sessionId === null) return; + const body = safeBody(req); + const directive = body['directive']; + if (typeof directive !== 'string' || directive.trim().length === 0) { + res.status(400).json({ + error: '`directive` is required and must be a non-empty string', + code: 'missing_directive', + }); + return; + } + const clientId = parseClientIdHeader(req, res); + if (clientId === null) return; + try { + const result = await bridge.launchSessionForkAgent( + sessionId, + directive, + clientId !== undefined ? { clientId } : undefined, + ); + res.status(202).json(result); + } catch (err) { + sendBridgeError(res, err, { + route: 'POST /session/:id/fork', + sessionId, + }); + } + }); + app.get('/session/:id/context', async (req, res) => { const sessionId = requireSessionId(req, res); if (sessionId === null) return; diff --git a/packages/core/src/services/sessionService.test.ts b/packages/core/src/services/sessionService.test.ts index d670b9fbae..9562c9c5da 100644 --- a/packages/core/src/services/sessionService.test.ts +++ b/packages/core/src/services/sessionService.test.ts @@ -1941,6 +1941,84 @@ describe('SessionService', () => { expect(srcLines.every((r) => !r.forkedFrom)).toBe(true); }); + it('forks only the active branch after rewind', async () => { + const oldId = '12121212-1212-1212-1212-121212121212'; + const newId = '34343434-3434-3434-3434-343434343434'; + const chatsDir = realPath.join( + service['storage'].getProjectDir(), + 'chats', + ); + fs.mkdirSync(chatsDir, { recursive: true }); + fs.writeFileSync( + realPath.join(chatsDir, `${oldId}.jsonl`), + [ + { + uuid: 'u1', + parentUuid: null, + sessionId: oldId, + type: 'user', + timestamp: '2026-04-22T00:00:00.000Z', + cwd, + version: 'test', + message: { role: 'user', parts: [{ text: 'first' }] }, + }, + { + uuid: 'u2', + parentUuid: 'u1', + sessionId: oldId, + type: 'assistant', + timestamp: '2026-04-22T00:00:01.000Z', + cwd, + version: 'test', + message: { role: 'model', parts: [{ text: 'first reply' }] }, + }, + { + uuid: 'u3', + parentUuid: 'u2', + sessionId: oldId, + type: 'user', + timestamp: '2026-04-22T00:00:02.000Z', + cwd, + version: 'test', + message: { role: 'user', parts: [{ text: 'second' }] }, + }, + { + uuid: 'u4', + parentUuid: 'u3', + sessionId: oldId, + type: 'assistant', + timestamp: '2026-04-22T00:00:03.000Z', + cwd, + version: 'test', + message: { role: 'model', parts: [{ text: 'second reply' }] }, + }, + { + uuid: 'rewind-1', + parentUuid: 'u2', + sessionId: oldId, + type: 'system', + subtype: 'rewind', + timestamp: '2026-04-22T00:00:04.000Z', + cwd, + version: 'test', + systemPayload: { targetTurnIndex: 1, truncatedCount: 2 }, + }, + ] + .map((line) => JSON.stringify(line)) + .join('\n') + '\n', + ); + + const result = await service.forkSession(oldId, newId); + const loaded = await service.loadSession(newId); + + expect(result.copiedCount).toBe(3); + expect( + loaded?.conversation.messages.flatMap( + (message) => message.message?.parts?.map((part) => part.text) ?? [], + ), + ).toEqual(['first', 'first reply']); + }); + it('throws when the source session does not exist', async () => { const oldId = '33333333-3333-3333-3333-333333333333'; const newId = '44444444-4444-4444-4444-444444444444'; diff --git a/packages/core/src/services/sessionService.ts b/packages/core/src/services/sessionService.ts index 93a8a966de..ca73b57240 100644 --- a/packages/core/src/services/sessionService.ts +++ b/packages/core/src/services/sessionService.ts @@ -1003,11 +1003,11 @@ export class SessionService { /** * Forks a session to a new sessionId. * - * Reads the source JSONL into memory, rewrites every record's `sessionId` - * to `newSessionId`, stamps `cwd` to this service's project root, rebuilds - * the `parentUuid` chain in write order so the fork is a linear - * continuation, stamps `forkedFrom: { sessionId, messageUuid }` on every - * copied record for audit, and writes the result to `.jsonl`. + * Reads the source JSONL into memory, reconstructs the active record chain, + * rewrites every record's `sessionId` to `newSessionId`, stamps `cwd` to this + * service's project root, rebuilds the `parentUuid` chain so the fork is a + * linear continuation, stamps `forkedFrom: { sessionId, messageUuid }` on + * every copied record for audit, and writes the result to `.jsonl`. * * Mirrors Claude Code's `/branch` storage model: full in-memory copy + per- * message forkedFrom (see claude-code/src/commands/branch/branch.ts). @@ -1049,10 +1049,18 @@ export class SessionService { ); } - // Rebuild the parentUuid chain in write order so the fork is a clean - // linear descendant. `forkedFrom` captures the origin of each message. + // Copy only the active branch. Rewind leaves old records in the JSONL as + // abandoned parentUuid branches; copying raw records would resurrect them. + const sourceRecords = this.reconstructHistory(records); + if (sourceRecords.length === 0) { + throw new Error(`Source session not found or empty: ${sourceSessionId}`); + } + + // Rebuild the parentUuid chain in active-history order so the fork is a + // clean linear descendant. `forkedFrom` captures the origin of each + // message. let prevUuid: string | null = null; - const forked: ChatRecord[] = records.map((record) => { + const forked: ChatRecord[] = sourceRecords.map((record) => { const next: ChatRecord = { ...record, sessionId: newSessionId, diff --git a/packages/sdk-typescript/scripts/build.js b/packages/sdk-typescript/scripts/build.js index a0227c6e0a..b31ca59e32 100755 --- a/packages/sdk-typescript/scripts/build.js +++ b/packages/sdk-typescript/scripts/build.js @@ -27,7 +27,8 @@ const rootDir = join(__dirname, '..'); // `mid_turn_message_injected` event type/guard/registration, ~150 bytes). // Bumped from 119KB to 122KB for the workspace extension management surface // (install/update/enable/disable/uninstall/refresh/check update endpoints). -const MAX_DAEMON_BROWSER_BUNDLE_BYTES = 122 * 1024; +// Bumped from 122KB to 124KB for daemon fork-session APIs/events. +const MAX_DAEMON_BROWSER_BUNDLE_BYTES = 124 * 1024; rmSync(join(rootDir, 'dist'), { recursive: true, force: true }); mkdirSync(join(rootDir, 'dist'), { recursive: true }); diff --git a/packages/sdk-typescript/src/daemon/DaemonClient.ts b/packages/sdk-typescript/src/daemon/DaemonClient.ts index 7cf712ff6b..ef70778268 100644 --- a/packages/sdk-typescript/src/daemon/DaemonClient.ts +++ b/packages/sdk-typescript/src/daemon/DaemonClient.ts @@ -29,6 +29,7 @@ import type { DaemonSessionContextUsageStatus, BranchSessionRequest, DaemonBranchedSession, + DaemonForkSessionResult, DaemonRestoredSession, DaemonSession, DaemonSessionSummary, @@ -79,6 +80,7 @@ import type { DaemonToolToggleResult, DaemonRewindSnapshotInfo, DaemonRewindResult, + ForkSessionRequest, DaemonSessionHooksStatus, DaemonWorkspaceExtensionsStatus, ExtensionMutationResponse, @@ -1265,6 +1267,27 @@ export class DaemonClient { ); } + async forkSession( + sessionId: string, + req: ForkSessionRequest, + clientId?: string, + ): Promise { + return await this.fetchWithTimeout( + `${this.baseUrl}/session/${encodeURIComponent(sessionId)}/fork`, + { + method: 'POST', + headers: this.headers({ 'Content-Type': 'application/json' }, clientId), + body: JSON.stringify({ directive: req.directive }), + }, + async (res) => { + if (!res.ok) { + throw await this.failOnError(res, 'POST /session/:id/fork'); + } + return (await res.json()) as DaemonForkSessionResult; + }, + ); + } + async sessionContext( sessionId: string, clientId?: string, diff --git a/packages/sdk-typescript/src/daemon/DaemonSessionClient.ts b/packages/sdk-typescript/src/daemon/DaemonSessionClient.ts index 5948113018..21064cc220 100644 --- a/packages/sdk-typescript/src/daemon/DaemonSessionClient.ts +++ b/packages/sdk-typescript/src/daemon/DaemonSessionClient.ts @@ -16,6 +16,7 @@ import { type SubscribeOptions, } from './DaemonClient.js'; import type { + DaemonForkSessionResult, DaemonEvent, DaemonRewindResult, DaemonRewindSnapshotInfo, @@ -341,6 +342,14 @@ export class DaemonSessionClient { }); } + async fork(directive: string): Promise { + return await this.client.forkSession( + this.sessionId, + { directive }, + this.clientId, + ); + } + /** * One-sentence "where did I leave off" recap of this session. See * `DaemonClient.recapSession` for the full contract: best-effort diff --git a/packages/sdk-typescript/src/daemon/index.ts b/packages/sdk-typescript/src/daemon/index.ts index c71d5b7a81..301c7f8ded 100644 --- a/packages/sdk-typescript/src/daemon/index.ts +++ b/packages/sdk-typescript/src/daemon/index.ts @@ -332,6 +332,8 @@ export type { DaemonProtocolVersions, BranchSessionRequest, DaemonBranchedSession, + DaemonForkSessionResult, + ForkSessionRequest, DaemonRestoredSession, DaemonSession, DaemonAuthProviderId, diff --git a/packages/sdk-typescript/src/daemon/types.ts b/packages/sdk-typescript/src/daemon/types.ts index fb1a42df78..a0bbaf43cd 100644 --- a/packages/sdk-typescript/src/daemon/types.ts +++ b/packages/sdk-typescript/src/daemon/types.ts @@ -190,6 +190,16 @@ export interface DaemonBranchedSession extends DaemonRestoredSession { forkedFrom: { sessionId: string; displayName: string }; } +export interface ForkSessionRequest { + directive: string; +} + +export interface DaemonForkSessionResult { + sessionId: string; + description: string; + launched: boolean; +} + /** Sparse session record returned by `GET /workspace/:id/sessions`. */ export interface DaemonSessionSummary { sessionId: string; diff --git a/packages/sdk-typescript/src/daemon/ui/normalizer.ts b/packages/sdk-typescript/src/daemon/ui/normalizer.ts index 3b708ffe3d..0b7523574a 100644 --- a/packages/sdk-typescript/src/daemon/ui/normalizer.ts +++ b/packages/sdk-typescript/src/daemon/ui/normalizer.ts @@ -176,6 +176,12 @@ export function normalizeDaemonEvent( case 'state_resync_required': return normalizeStateResyncRequired(event, base); + case 'session_rewound': + return normalizeSessionRewound(event, base); + + case 'session_branched': + return normalizeSessionBranched(event, base); + case 'prompt_cancelled': { // Forward the optional `reason` (e.g. `'forward_failed'` from the // bridge's C3 compensating broadcast) so consumers can distinguish a @@ -334,6 +340,48 @@ function normalizeStateResyncRequired( ]; } +function normalizeSessionRewound( + event: DaemonEvent, + base: NormalizedEventBase, +): DaemonUiEvent[] { + const promptId = getString(event.data, 'promptId'); + const targetTurnIndex = numberField(event.data, 'targetTurnIndex'); + if (!promptId || targetTurnIndex === undefined) { + return fallbackDebug(event, base, 'malformed session_rewound payload'); + } + const sessionId = getString(event.data, 'sessionId'); + return [ + { + ...base, + type: 'session.rewound', + promptId, + targetTurnIndex, + ...(sessionId ? { sessionId } : {}), + }, + ]; +} + +function normalizeSessionBranched( + event: DaemonEvent, + base: NormalizedEventBase, +): DaemonUiEvent[] { + const sourceSessionId = getString(event.data, 'sourceSessionId'); + const newSessionId = getString(event.data, 'newSessionId'); + const displayName = getString(event.data, 'displayName'); + if (!sourceSessionId || !newSessionId || !displayName) { + return fallbackDebug(event, base, 'malformed session_branched payload'); + } + return [ + { + ...base, + type: 'session.branched', + sourceSessionId, + newSessionId, + displayName, + }, + ]; +} + function normalizeFollowupSuggestion( event: DaemonEvent, base: NormalizedEventBase, diff --git a/packages/sdk-typescript/src/daemon/ui/terminal.ts b/packages/sdk-typescript/src/daemon/ui/terminal.ts index 50153140de..bb092a3489 100644 --- a/packages/sdk-typescript/src/daemon/ui/terminal.ts +++ b/packages/sdk-typescript/src/daemon/ui/terminal.ts @@ -80,6 +80,18 @@ export function daemonUiEventToTerminalText(event: DaemonUiEvent): string { `caught up (${event.replayedCount} replayed)`, '2', ); + case 'session.rewound': + return terminalLine( + 'rewound', + `${event.promptId} → turn ${event.targetTurnIndex}`, + '2', + ); + case 'session.branched': + return terminalLine( + 'branched', + `${event.sourceSessionId} → ${event.newSessionId} (${event.displayName})`, + '2', + ); case 'prompt.cancelled': return terminalLine('cancelled', 'prompt cancelled', '33'); case 'followup.suggestion': diff --git a/packages/sdk-typescript/src/daemon/ui/transcript.ts b/packages/sdk-typescript/src/daemon/ui/transcript.ts index 61109c4678..ee2bfc538c 100644 --- a/packages/sdk-typescript/src/daemon/ui/transcript.ts +++ b/packages/sdk-typescript/src/daemon/ui/transcript.ts @@ -327,6 +327,17 @@ function applyDaemonTranscriptEvent( // stream (or `selectors`) to drop a catch-up indicator. No // transcript mutation. break; + case 'session.rewound': + rewindTranscriptToUserTurn(next, event.targetTurnIndex); + break; + case 'session.branched': + appendStatusBlock( + next, + 'status', + `Branched conversation "${event.displayName}". You are now in the branch.`, + event, + ); + break; case 'workspace.memory.changed': case 'workspace.agent.changed': case 'workspace.tool.toggled': @@ -969,6 +980,16 @@ function appendStatusBlock( event.data !== undefined ? { data: event.data } : {}), + ...(event?.type === 'session.branched' + ? { + source: 'session_branched', + data: { + sourceSessionId: event.sourceSessionId, + newSessionId: event.newSessionId, + displayName: event.displayName, + }, + } + : {}), }; appendBlock(state, block); if (opts.clearActiveText !== false) clearActiveText(state); @@ -1173,6 +1194,58 @@ function takeBlocksOwnership(state: DaemonTranscriptState): void { ownedBlocks.set(state, state.blocks); } +// Applies a daemon rewind event to this in-memory transcript only. The target +// user turn and everything after it are removed so the rendered session view +// matches the already-rewound backend state. +function rewindTranscriptToUserTurn( + state: DaemonTranscriptState, + targetTurnIndex: number, +): void { + let userTurn = 0; + let lastUserIndex = -1; + for (let index = 0; index < state.blocks.length; index += 1) { + if (state.blocks[index]?.kind !== 'user') continue; + lastUserIndex = index; + if (userTurn === targetTurnIndex) { + truncateTranscriptBeforeBlock(state, index); + return; + } + userTurn += 1; + } + if (lastUserIndex >= 0 && targetTurnIndex >= userTurn) { + truncateTranscriptBeforeBlock(state, lastUserIndex); + } +} + +function truncateTranscriptBeforeBlock( + state: DaemonTranscriptState, + blockIndex: number, +): void { + takeBlocksOwnership(state); + state.blocks = state.blocks.slice(0, blockIndex); + ownedBlocks.set(state, state.blocks); + state.blockIndexById = rebuildDaemonTranscriptBlockIndex(state.blocks); + state.toolBlockByCallId = {}; + state.permissionBlockByRequestId = {}; + for (const block of state.blocks) { + if (block.kind === 'tool') + state.toolBlockByCallId[block.toolCallId] = block.id; + if (block.kind === 'permission') { + state.permissionBlockByRequestId[block.requestId] = block.id; + } + } + state.trimmedToolNotificationByCallId = {}; + state.activeUserBlockId = undefined; + state.activeAssistantBlockId = undefined; + state.activeThoughtBlockId = undefined; + state.activeAssistantBlockByParent = {}; + state.activeThoughtBlockByParent = {}; + state.currentToolCallId = undefined; + state.pendingUserShellCommand = undefined; + state.lastFollowupSuggestion = undefined; + state.toolProgress = {}; +} + function appendBlock( state: DaemonTranscriptState, block: DaemonTranscriptBlock, diff --git a/packages/sdk-typescript/src/daemon/ui/types.ts b/packages/sdk-typescript/src/daemon/ui/types.ts index 7ea228eabf..90178e55e0 100644 --- a/packages/sdk-typescript/src/daemon/ui/types.ts +++ b/packages/sdk-typescript/src/daemon/ui/types.ts @@ -38,6 +38,8 @@ export type DaemonUiEventType = | 'session.available_commands' | 'session.state_resync_required' | 'session.replay_complete' + | 'session.rewound' + | 'session.branched' // Prompt lifecycle (cross-client) | 'prompt.cancelled' // Daemon assist push (server-side ghost-text suggestion) @@ -359,6 +361,20 @@ export interface DaemonUiReplayCompleteEvent extends DaemonUiEventBase { lastReplayedEventId?: number; } +export interface DaemonUiSessionRewoundEvent extends DaemonUiEventBase { + type: 'session.rewound'; + sessionId?: string; + promptId: string; + targetTurnIndex: number; +} + +export interface DaemonUiSessionBranchedEvent extends DaemonUiEventBase { + type: 'session.branched'; + sourceSessionId: string; + newSessionId: string; + displayName: string; +} + /* ────────────────────────────────────────────────────────────────────────── * Workspace events (Wave 3-4) * ──────────────────────────────────────────────────────────────────────── */ @@ -517,6 +533,8 @@ export type DaemonUiEvent = | DaemonUiSessionAvailableCommandsEvent | DaemonUiStateResyncRequiredEvent | DaemonUiReplayCompleteEvent + | DaemonUiSessionRewoundEvent + | DaemonUiSessionBranchedEvent // Prompt lifecycle (cross-client) | DaemonUiPromptCancelledEvent // Daemon assist push (server-side ghost-text suggestion) diff --git a/packages/sdk-typescript/test/unit/daemonUi.test.ts b/packages/sdk-typescript/test/unit/daemonUi.test.ts index 20df7ffc27..bd8d8c1550 100644 --- a/packages/sdk-typescript/test/unit/daemonUi.test.ts +++ b/packages/sdk-typescript/test/unit/daemonUi.test.ts @@ -1164,6 +1164,117 @@ describe('daemon UI normalizer and transcript reducer', () => { expect((malformed[0] as { text: string }).text).not.toContain('secret'); }); + it('normalizes session branch events as structured sidechannel events', () => { + const events = normalizeDaemonEvent({ + id: 59, + v: 1, + type: 'session_branched', + data: { + sourceSessionId: '9976ed52-1bd3-48cd-b8dc-0f045009ad7d', + newSessionId: '7497af5d-b62f-42f4-82d7-6f2a81daf439', + displayName: 'support-branch-new3 (Branch 2)', + }, + }); + + expect(events).toEqual([ + expect.objectContaining({ + type: 'session.branched', + sourceSessionId: '9976ed52-1bd3-48cd-b8dc-0f045009ad7d', + newSessionId: '7497af5d-b62f-42f4-82d7-6f2a81daf439', + displayName: 'support-branch-new3 (Branch 2)', + }), + ]); + + const state = reduceDaemonTranscriptEvents( + createDaemonTranscriptState({ now: 1 }), + events, + { now: 2 }, + ); + expect(state.blocks).toEqual([ + expect.objectContaining({ + kind: 'status', + source: 'session_branched', + data: { + sourceSessionId: '9976ed52-1bd3-48cd-b8dc-0f045009ad7d', + newSessionId: '7497af5d-b62f-42f4-82d7-6f2a81daf439', + displayName: 'support-branch-new3 (Branch 2)', + }, + }), + ]); + }); + + it('rewinds to the last user turn when targetTurnIndex is out of range', () => { + let state = createDaemonTranscriptState({ now: 1 }); + state = appendLocalUserTranscriptMessage(state, 'first', { now: 2 }); + state = reduceDaemonTranscriptEvents( + state, + [{ type: 'assistant.text.delta', text: 'answer one' }], + { now: 3 }, + ); + state = appendLocalUserTranscriptMessage(state, 'second', { now: 4 }); + state = reduceDaemonTranscriptEvents( + state, + [{ type: 'assistant.text.delta', text: 'answer two' }], + { now: 5 }, + ); + + state = reduceDaemonTranscriptEvents( + state, + [ + { + type: 'session.rewound', + promptId: 'prompt-2', + targetTurnIndex: 99, + }, + ], + { now: 6 }, + ); + + expect(state.blocks).toMatchObject([ + { kind: 'user', text: 'first' }, + { kind: 'assistant', text: 'answer one' }, + ]); + }); + + it('rewinds to an exact user turn index', () => { + let state = createDaemonTranscriptState({ now: 1 }); + state = appendLocalUserTranscriptMessage(state, 'first', { now: 2 }); + state = reduceDaemonTranscriptEvents( + state, + [{ type: 'assistant.text.delta', text: 'answer one' }], + { now: 3 }, + ); + state = appendLocalUserTranscriptMessage(state, 'second', { now: 4 }); + state = reduceDaemonTranscriptEvents( + state, + [{ type: 'assistant.text.delta', text: 'answer two' }], + { now: 5 }, + ); + state = appendLocalUserTranscriptMessage(state, 'third', { now: 6 }); + state = reduceDaemonTranscriptEvents( + state, + [{ type: 'assistant.text.delta', text: 'answer three' }], + { now: 7 }, + ); + + state = reduceDaemonTranscriptEvents( + state, + [ + { + type: 'session.rewound', + promptId: 'prompt-2', + targetTurnIndex: 1, + }, + ], + { now: 8 }, + ); + + expect(state.blocks).toMatchObject([ + { kind: 'user', text: 'first' }, + { kind: 'assistant', text: 'answer one' }, + ]); + }); + it('normalizes plan session updates as visible tool blocks', () => { const state = reduceDaemonTranscriptEvents( createDaemonTranscriptState({ now: 1 }), diff --git a/packages/web-shell/client/App.tsx b/packages/web-shell/client/App.tsx index 3445922232..28466f4772 100644 --- a/packages/web-shell/client/App.tsx +++ b/packages/web-shell/client/App.tsx @@ -904,9 +904,12 @@ export function App({ () => getBackgroundTaskActivityKey(messages), [messages], ); + const [backgroundTasksRefreshTrigger, setBackgroundTasksRefreshTrigger] = + useState(0); const backgroundTasks = useBackgroundTasks( backgroundTaskActivityKey, connection.status === 'connected', + backgroundTasksRefreshTrigger, ); const footerTasks = useMemo( () => (renderFooter ? backgroundTasks.map(mapToWebShellTaskInfo) : []), @@ -2083,6 +2086,53 @@ export function App({ setShowReleaseDialog(true); return true; } + if (cmd === 'branch') { + if (promptBlocked) return enqueuePrompt(text, images); + const branchName = text.slice(match[0].length).trim(); + sessionActions + .branchSession(branchName || undefined) + .then((result) => { + store.dispatch([ + { + type: 'status', + text: t('branch.success', { + name: result.displayName, + }), + }, + ]); + }) + .catch((error: unknown) => { + reportError(error, t('branch.failed')); + }); + return true; + } + if (cmd === 'fork') { + if (promptBlocked) return enqueuePrompt(text, images); + const directive = text.slice(match[0].length).trim(); + if (!directive) { + pushToast('error', t('fork.empty')); + return true; + } + sessionActions + .forkSession(directive) + .then((result) => { + if (!result.launched) { + pushToast('warning', t('fork.notStarted')); + return; + } + setBackgroundTasksRefreshTrigger((value) => value + 1); + pushToast( + 'success', + t('fork.started', { name: result.description }), + ); + }) + .catch((error: unknown) => { + const reason = + error instanceof Error ? error.message : String(error); + reportError(error, t('fork.failed', { reason })); + }); + return true; + } if (cmd === 'auth') { store.appendLocalUserMessage(text); setAuthInlineOpen(true); diff --git a/packages/web-shell/client/adapters/transcriptToMessages.test.ts b/packages/web-shell/client/adapters/transcriptToMessages.test.ts index ba5cd3cf91..e9cfd7dcbf 100644 --- a/packages/web-shell/client/adapters/transcriptToMessages.test.ts +++ b/packages/web-shell/client/adapters/transcriptToMessages.test.ts @@ -169,6 +169,41 @@ describe('transcriptBlocksToDaemonMessages', () => { ]); }); + it('localizes structured session branch status blocks', () => { + const messages = transcriptBlocksToDaemonMessages( + [ + statusBlock('branch-1', 'Branched conversation "old"', 1, { + source: 'session_branched', + data: { + sourceSessionId: 'source', + newSessionId: 'new', + displayName: 'support-branch-new3 (Branch 2)', + }, + }), + ], + { + labels: { + branchSuccess: (name) => + `已复制会话，新会话名称为： "${name}"，当前已切换到新的会话。`, + }, + }, + ); + + expect(messages).toEqual([ + expect.objectContaining({ + role: 'system', + content: + '已复制会话，新会话名称为： "support-branch-new3 (Branch 2)"，当前已切换到新的会话。', + source: 'session_branched', + data: { + sourceSessionId: 'source', + newSessionId: 'new', + displayName: 'support-branch-new3 (Branch 2)', + }, + }), + ]); + }); + it('ignores daemon plan entries without content', () => { const plan = { sessionUpdate: 'plan', diff --git a/packages/web-shell/client/adapters/transcriptToMessages.ts b/packages/web-shell/client/adapters/transcriptToMessages.ts index 86e8862a17..d04994e451 100644 --- a/packages/web-shell/client/adapters/transcriptToMessages.ts +++ b/packages/web-shell/client/adapters/transcriptToMessages.ts @@ -38,14 +38,64 @@ type ExtendedDaemonStatusTranscriptBlock = DaemonStatusTranscriptBlock & { data?: unknown; }; +type ExtendedDaemonTextTranscriptBlock = DaemonTextTranscriptBlock & { + meta?: Record; +}; + interface TranscriptMessageLabels { promptCancelled?: string; + branchSuccess?: (name: string) => string; } interface TranscriptMessageOptions { labels?: TranscriptMessageLabels; } +function isIgnoredWebShellStatus(text: string): boolean { + return ( + text.startsWith('language_changed (unrecognized daemon event):') || + text.startsWith('Model switched: ') + ); +} + +function getSessionBranchDisplayName(data: unknown): string | null { + if (!data || typeof data !== 'object') return null; + const branchData = data as { + displayName?: unknown; + newSessionId?: unknown; + }; + if (typeof branchData.displayName === 'string' && branchData.displayName) { + return branchData.displayName; + } + return typeof branchData.newSessionId === 'string' + ? branchData.newSessionId.slice(0, 8) + : null; +} + +function isBackgroundNotificationAssistantBlock( + block: DaemonTextTranscriptBlock, +): boolean { + const extended = block as ExtendedDaemonTextTranscriptBlock; + const meta = extended.meta; + return ( + meta?.['source'] === 'background_notification' && + meta['qwenDiscreteMessage'] === true && + meta['backgroundTask'] !== undefined + ); +} + +function normalizeAssistantTextBlock( + block: DaemonTextTranscriptBlock, +): DaemonTextTranscriptBlock | null { + if (isBackgroundNotificationAssistantBlock(block)) return null; + if (!block.text && !block.usage) return null; + return block; +} + +function isTextBlockEmpty(block: DaemonTextTranscriptBlock): boolean { + return block.text.length === 0; +} + function parseDaemonTodoItemsFromEntries( entries: readonly unknown[], ): DaemonMessageTodoItem[] | undefined { @@ -145,7 +195,10 @@ export function transcriptBlocksToDaemonMessages( } case 'assistant': { - const textBlock = block as DaemonTextTranscriptBlock; + const textBlock = normalizeAssistantTextBlock( + block as DaemonTextTranscriptBlock, + ); + if (!textBlock) break; const parentSubAgent = textBlock.parentToolCallId ? toolsByCallId.get(textBlock.parentToolCallId) @@ -210,7 +263,12 @@ export function transcriptBlocksToDaemonMessages( currentAssistantIdx !== null ? messages[currentAssistantIdx] : undefined; - if (target && target.role === 'assistant' && !needsNewContentMessage) { + if ( + target && + target.role === 'assistant' && + !needsNewContentMessage && + !isTextBlockEmpty(textBlock) + ) { const usage = mergeAssistantUsage(target.usage, textBlock.usage); messages[currentAssistantIdx!] = { ...target, @@ -219,7 +277,7 @@ export function transcriptBlocksToDaemonMessages( ...(usage ? { usage } : {}), }; needsNewContentMessage = false; - } else { + } else if (!isTextBlockEmpty(textBlock)) { messages.push({ id: block.id, role: 'assistant', @@ -230,6 +288,12 @@ export function transcriptBlocksToDaemonMessages( }); currentAssistantIdx = messages.length - 1; needsNewContentMessage = false; + } else if (textBlock.usage && target && target.role === 'assistant') { + const usage = mergeAssistantUsage(target.usage, textBlock.usage); + messages[currentAssistantIdx!] = { + ...target, + ...(usage ? { usage } : {}), + }; } break; } @@ -444,7 +508,15 @@ export function transcriptBlocksToDaemonMessages( case 'status': case 'debug': { const statusBlock = block as ExtendedDaemonStatusTranscriptBlock; - const text = statusBlock.text; + const branchDisplayName = + statusBlock.source === 'session_branched' + ? getSessionBranchDisplayName(statusBlock.data) + : null; + const text = + branchDisplayName && options.labels?.branchSuccess + ? options.labels.branchSuccess(branchDisplayName) + : statusBlock.text; + if (isIgnoredWebShellStatus(text)) break; const todos = parsePlanTodos(text); if (todos) { messages.push({ diff --git a/packages/web-shell/client/constants/localCommands.ts b/packages/web-shell/client/constants/localCommands.ts index 81e03e1ff5..cbbe909e06 100644 --- a/packages/web-shell/client/constants/localCommands.ts +++ b/packages/web-shell/client/constants/localCommands.ts @@ -83,6 +83,16 @@ export function getLocalCommands(t: Translate): CommandInfo[] { }, { name: 'tasks', description: t('local.tasks') }, { name: 'recap', description: t('local.recap') }, + { + name: 'branch', + description: t('local.branch'), + argumentHint: '[]', + }, + { + name: 'fork', + description: t('local.fork'), + argumentHint: '', + }, { name: 'clear', description: t('local.clear') }, { name: 'new', description: t('local.new') }, { name: 'reset', description: t('local.reset') }, diff --git a/packages/web-shell/client/hooks/useBackgroundTasks.ts b/packages/web-shell/client/hooks/useBackgroundTasks.ts index be791b7901..9acca18e3a 100644 --- a/packages/web-shell/client/hooks/useBackgroundTasks.ts +++ b/packages/web-shell/client/hooks/useBackgroundTasks.ts @@ -15,6 +15,7 @@ function hasActiveTask(tasks: readonly DaemonSessionTaskStatus[]): boolean { export function useBackgroundTasks( taskActivityKey: string, connected: boolean, + refreshTrigger = 0, ): DaemonSessionTaskStatus[] { const actions = useActions(); const [tasks, setTasks] = useState([]); @@ -38,6 +39,12 @@ export function useBackgroundTasks( setPollingActive(true); }, [connected, taskActivityKey]); + useEffect(() => { + if (!connected || refreshTrigger === 0) return; + emptyPollsRef.current = 0; + setPollingActive(true); + }, [connected, refreshTrigger]); + useEffect(() => { if (tasksPanelActive) return; if (!connected || !pollingActive) return; diff --git a/packages/web-shell/client/hooks/useMessages.ts b/packages/web-shell/client/hooks/useMessages.ts index abdfbab578..bd0fbce72a 100644 --- a/packages/web-shell/client/hooks/useMessages.ts +++ b/packages/web-shell/client/hooks/useMessages.ts @@ -10,7 +10,10 @@ export function useMessages( return useMemo( () => transcriptBlocksToDaemonMessages(blocks, { - labels: { promptCancelled: t('request.cancelled') }, + labels: { + promptCancelled: t('request.cancelled'), + branchSuccess: (name) => t('branch.success', { name }), + }, }), [blocks, t], ); diff --git a/packages/web-shell/client/i18n.tsx b/packages/web-shell/client/i18n.tsx index b06d805ced..dc807666f0 100644 --- a/packages/web-shell/client/i18n.tsx +++ b/packages/web-shell/client/i18n.tsx @@ -348,6 +348,14 @@ const EN: Messages = { 'help.shortcut.compact': 'Toggle compact mode', 'retry.hint': 'Press Ctrl+Y to retry or click to retry', 'retry.none': 'No failed request to retry.', + 'branch.failed': 'Failed to branch session.', + 'branch.success': (v) => + `Copied session. New session name: "${v?.name ?? ''}". Switched to the new session.`, + 'fork.empty': 'Please provide a directive. Usage: /fork ', + 'fork.failed': (v) => `Failed to launch fork: ${v?.reason ?? ''}`, + 'fork.notStarted': 'Background agent was not launched.', + 'fork.started': (v) => + `Started background agent: "${v?.name ?? ''}". Track it in background tasks.`, 'command.hidden': 'This command is not available.', 'help.shortcut.approvals': 'Cycle approval modes', 'help.shortcut.cancel': 'Close dialogs or cancel operation', @@ -434,6 +442,8 @@ const EN: Messages = { 'local.copy': 'Copy last output or snippet', 'local.delete': 'Delete a session permanently', 'local.release': 'Release a live session', + 'local.branch': 'Copy the current conversation into a new session', + 'local.fork': 'Start a background agent from this conversation', 'local.help': 'Show help and commands', 'local.language': 'Change UI language', 'local.mcp': 'Manage MCP servers', @@ -1321,6 +1331,14 @@ const ZH: Messages = { 'help.shortcut.compact': '切换紧凑模式', 'retry.hint': '按 Ctrl+Y 重试或点击重试', 'retry.none': '没有可重试的失败请求。', + 'branch.failed': '分支会话失败。', + 'branch.success': (v) => + `已复制会话，新会话名称为： "${v?.name ?? ''}"，当前已切换到新的会话。`, + 'fork.empty': '请提供任务指令。用法：/fork <指令>', + 'fork.failed': (v) => `启动后台智能体失败：${v?.reason ?? ''}`, + 'fork.notStarted': '后台智能体未启动。', + 'fork.started': (v) => + `已启动后台智能体："${v?.name ?? ''}"，可在后台任务中查看。`, 'command.hidden': '该命令不可用。', 'help.shortcut.approvals': '切换审批模式', 'help.shortcut.cancel': '关闭弹窗或取消操作', @@ -1403,6 +1421,8 @@ const ZH: Messages = { 'local.copy': '复制最后输出或代码片段', 'local.delete': '永久删除会话', 'local.release': '释放 live session', + 'local.branch': '将当前对话复制到新会话', + 'local.fork': '基于当前对话启动后台智能体', 'local.help': '查看帮助和可用命令', 'local.language': '切换 UI 语言', 'local.mcp': '管理 MCP servers', diff --git a/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx b/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx index b654f4e02b..fd623f1c59 100644 --- a/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx +++ b/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx @@ -101,6 +101,15 @@ interface MockClient { getWorkspaceAgent: () => Promise; createWorkspaceAgent: () => Promise; deleteWorkspaceAgent: () => Promise; + branchSession: ( + sessionId: string, + req: { name?: string }, + clientId?: string, + ) => Promise<{ + sessionId: string; + displayName: string; + clientId?: string; + }>; } const sdkMocks = vi.hoisted(() => { @@ -123,6 +132,7 @@ const sdkMocks = vi.hoisted(() => { const getWorkspaceAgent = vi.fn(); const createWorkspaceAgent = vi.fn(); const deleteWorkspaceAgent = vi.fn(); + const branchSession = vi.fn(); class MockDaemonClient { constructor(_opts: unknown) {} @@ -145,6 +155,7 @@ const sdkMocks = vi.hoisted(() => { getWorkspaceAgent = getWorkspaceAgent; createWorkspaceAgent = createWorkspaceAgent; deleteWorkspaceAgent = deleteWorkspaceAgent; + branchSession = branchSession; dispose = vi.fn(); } @@ -176,6 +187,7 @@ const sdkMocks = vi.hoisted(() => { MockDaemonClient, MockDaemonSessionClient, workspaceMcpTools, + branchSession, reset() { sessions.length = 0; capabilities.mockReset(); @@ -251,6 +263,12 @@ const sdkMocks = vi.hoisted(() => { createWorkspaceAgent.mockResolvedValue({ ok: true }); deleteWorkspaceAgent.mockReset(); deleteWorkspaceAgent.mockResolvedValue(undefined); + branchSession.mockReset(); + branchSession.mockResolvedValue({ + sessionId: 'branch-session', + displayName: 'Branch Session', + clientId: 'branch-client', + }); MockDaemonSessionClient.createOrAttach.mockReset(); MockDaemonSessionClient.createOrAttach.mockImplementation( async (client: unknown, _req: unknown): Promise => @@ -2626,6 +2644,54 @@ describe('DaemonSessionProvider', () => { expect(loadCalls[1]?.[3]).toBe('client-a'); }); + it('reuses the branched session client when switching after branch', async () => { + window.sessionStorage.clear(); + const sourceSession = createMockSession({ + sessionId: 'session-a', + clientId: 'client-a', + }); + const branchedSession = createMockSession({ + sessionId: 'session-b', + clientId: 'client-b', + }); + sdkMocks.branchSession.mockResolvedValue({ + sessionId: 'session-b', + displayName: 'Branch 1', + clientId: 'client-b', + }); + sdkMocks.sessions.push(sourceSession, branchedSession); + let actions: DaemonSessionActions | undefined; + + function Harness() { + actions = useDaemonActions(); + return null; + } + + await renderWithProvider(, { autoConnect: true }); + await act(async () => { + await flushPromises(); + }); + + const branch = requireActions(actions).branchSession('Branch 1'); + await act(async () => { + await wait(5); + await flushPromises(); + }); + await expect(branch).resolves.toEqual({ + sessionId: 'session-b', + displayName: 'Branch 1', + }); + + expect(sdkMocks.branchSession).toHaveBeenCalledWith( + 'session-a', + { name: 'Branch 1' }, + 'client-a', + ); + const loadCalls = sdkMocks.MockDaemonSessionClient.load.mock.calls; + expect(loadCalls[0]?.[1]).toBe('session-b'); + expect(loadCalls[0]?.[3]).toBe('client-b'); + }); + it('exposes daemon capabilities on the connection state', async () => { sdkMocks.capabilities.mockResolvedValue({ v: 1, diff --git a/packages/webui/src/daemon/session/actions.ts b/packages/webui/src/daemon/session/actions.ts index cb0ae7f635..e63d9356f7 100644 --- a/packages/webui/src/daemon/session/actions.ts +++ b/packages/webui/src/daemon/session/actions.ts @@ -9,6 +9,7 @@ import type { DaemonSessionContextStatus, DaemonSessionClient, DaemonSessionBtwResult, + DaemonForkSessionResult, DaemonMidTurnMessageResult, DaemonSessionRecapResult, DaemonSessionTaskStatus, @@ -27,6 +28,7 @@ import { withActionTimeout, type TimerRef, } from '../timing.js'; +import { persistStableClientId } from './clientLifecycle.js'; import type { ActivePrompt, AddDaemonSessionNotice, @@ -778,6 +780,70 @@ export function createDaemonSessionActions({ ); } }, + + async branchSession(name?: string) { + const session = requireSessionForAction( + addNotice, + sessionRef.current, + 'Branch session failed', + 'branch_session', + ); + try { + const result = await withActionTimeout( + session.client.branchSession( + session.sessionId, + { name }, + session.clientId, + ), + 'Branch session timed out', + ); + persistStableClientId(result.clientId, result.sessionId); + void startSessionSwitch(result.sessionId, 'load').catch( + (switchError: unknown) => { + if (isAbortError(switchError)) return; + dispatchActionError( + addNotice, + 'Branch session failed', + switchError, + 'branch_session', + ); + }, + ); + return { + sessionId: result.sessionId, + displayName: result.displayName, + }; + } catch (error) { + throw dispatchActionError( + addNotice, + 'Branch session failed', + error, + 'branch_session', + ); + } + }, + + async forkSession(directive: string): Promise { + const session = requireSessionForAction( + addNotice, + sessionRef.current, + 'Fork session failed', + 'fork_session', + ); + try { + return await withActionTimeout( + session.fork(directive), + 'Fork session timed out', + ); + } catch (error) { + throw dispatchActionError( + addNotice, + 'Fork session failed', + error, + 'fork_session', + ); + } + }, }; } diff --git a/packages/webui/src/daemon/session/types.ts b/packages/webui/src/daemon/session/types.ts index 0b30db6fc4..866684f13a 100644 --- a/packages/webui/src/daemon/session/types.ts +++ b/packages/webui/src/daemon/session/types.ts @@ -11,6 +11,7 @@ import type { DaemonApprovalMode, DaemonApprovalModeResult, DaemonAvailableCommand, + DaemonForkSessionResult, DaemonSessionBtwResult, DaemonMidTurnMessageResult, DaemonSessionContextStatus, @@ -161,6 +162,8 @@ export type DaemonNoticeOperation = | 'refresh_commands' | 'recap_session' | 'btw_session' + | 'branch_session' + | 'fork_session' | 'stream' | 'normalize_event'; @@ -307,6 +310,10 @@ export interface DaemonSessionActions { ): Promise<{ cancelled: boolean }>; clearGoal(): Promise<{ cleared: boolean; condition?: string }>; getStats(): Promise; + branchSession( + name?: string, + ): Promise<{ sessionId: string; displayName: string }>; + forkSession(directive: string): Promise; } export interface DaemonSessionContextValue {