From 3c3bffdfbddfc6fdc827473197c97668fdc4f89c Mon Sep 17 00:00:00 2001 From: SSU-WEI HUANG Date: Mon, 3 Aug 2026 09:28:01 +0800 Subject: [PATCH] feat: message-level conversation fork and rewind (#1263) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat: add message-level conversation fork and rewind Expose native Codex/Grok/Claude history controls through hub REST+RPC and web ConfirmDialog actions, without file rewind or composed forks. Also reconcile the duplicate hub V14→V15 migration so typecheck can pass. Co-authored-by: Cursor * fix: hydrate fork transcript and consume Claude --fork-session Forked HAPI children now copy the source transcript prefix so navigation is not a blank thread, and Claude drops --fork-session after the first launch so relaunches do not branch again. Co-authored-by: Cursor * fix: harden fork/rewind concurrency and durable history points Skip pending scheduled rows when hydrating fork transcripts, serialize fork/rewind per session, and persist conversation history points/indexes across existing-session bootstrap and Grok relaunches. Co-authored-by: Cursor * fix: close remaining fork/rewind races and UI anchoring Block sends and scheduled maturation while history actions run, order fork prefixes by invocation time, inherit history locators into children, and only offer Fork current on the live tail boundary. Co-authored-by: Cursor * fix: address remaining fork/rewind bot findings Materialize Claude --fork-session before the first child prompt, validate HAPI history boundaries before native RPC, expose forkCurrent on a latest user boundary, fully demote unsupported conversationHistory capabilities, and fix the truncate test setup order. Co-authored-by: Cursor * fix: align fork-current ids and Claude fork bootstrap Compare the latest fork boundary in assistant-ui threadMessageId space, spawn Claude forks with the persisted session mode, and preserve forkedFrom across existing-session bootstrap. Co-authored-by: Cursor * fix: close fork/rewind consistency holes at the contract layer Hold the source history lock until Claude child binds a distinct native id, persist Codex localId→turnId locators, and mark/block diverged sessions when native rewind outruns HAPI truncate. Co-authored-by: Cursor * fix: use Codex stable lastTurnId for historical fork Map HAPI's exclusive boundary to the previous turn's inclusive lastTurnId so native fork context matches the hydrated transcript. Co-authored-by: Cursor * fix: require exact Grok native resume for fork children Reject newSession fallback when forkedFrom is set, and keep the hub history lock until the child binds the forked grokSessionId. Co-authored-by: Cursor * fix: kill active fork children before failed-fork cleanup Bind/readiness failures can leave the child process running; deleteSession rejects active rows, so terminate first then remove the HAPI session. Co-authored-by: Cursor * fix: close remaining fork lock, hydrate, and todos gaps Reject mode switches during history actions, batch-copy fork transcripts in one SQLite transaction, and rebuild todos after fork hydrate / rewind truncate. Co-authored-by: Cursor * fix: allow Codex historical fork before the first turn Use experimental beforeTurnId when there is no previous turn for the stable inclusive lastTurnId boundary. Co-authored-by: Cursor * fix: mark Grok history busy immediately after dequeue Hub idle checks clear once messages-consumed fires; hold the busy flag across permission sync and rewind-points lookup before prompt starts. Co-authored-by: Cursor * fix: encode copied conversation history content * fix(web): hide local conversation history actions * style(codex): remove trailing whitespace * fix(fork): preserve children when cleanup is unconfirmed * fix(history): confirm cleanup and guard rewind divergence * fix(history): probe capabilities before advertising --------- Co-authored-by: Cursor --- cli/src/agent/backends/acp/AcpSdkBackend.ts | 17 + cli/src/agent/sessionFactory.test.ts | 12 +- cli/src/agent/sessionFactory.ts | 44 +- cli/src/api/apiMachine.ts | 5 +- cli/src/claude/claudeRemote.ts | 182 ++++--- cli/src/claude/claudeRemoteLauncher.ts | 5 + cli/src/claude/runClaude.ts | 32 ++ cli/src/claude/sdk/query.ts | 2 + cli/src/claude/sdk/types.ts | 5 + .../session.consumeOneTimeFlags.test.ts | 43 ++ cli/src/claude/session.ts | 8 +- cli/src/codex/appServerTypes.ts | 23 + cli/src/codex/codexAppServerClient.ts | 21 + cli/src/codex/codexRemoteLauncher.test.ts | 1 + cli/src/codex/codexRemoteLauncher.ts | 81 ++- cli/src/codex/conversationHistory.test.ts | 143 +++++ cli/src/codex/conversationHistory.ts | 265 ++++++++++ cli/src/codex/utils/appServerConfig.ts | 5 + cli/src/commands/agentCommandOptions.ts | 6 + cli/src/commands/claude.ts | 6 + cli/src/grok/conversationHistory.test.ts | 80 +++ cli/src/grok/conversationHistory.ts | 224 ++++++++ cli/src/grok/grokRemoteLauncher.test.ts | 40 +- cli/src/grok/grokRemoteLauncher.ts | 100 +++- cli/src/modules/common/rpcTypes.ts | 2 + cli/src/runner/buildCliArgs.test.ts | 15 + cli/src/runner/run.ts | 19 +- hub/src/store/messageStore.ts | 22 + hub/src/store/messages.copyBatch.test.ts | 33 ++ hub/src/store/messages.truncate.test.ts | 28 + hub/src/store/messages.ts | 122 +++++ hub/src/store/sessionStore.ts | 10 + hub/src/store/sessions.ts | 32 ++ hub/src/sync/forkTranscript.test.ts | 47 ++ hub/src/sync/forkTranscript.ts | 36 ++ hub/src/sync/messageService.ts | 5 +- hub/src/sync/rpcGateway.ts | 30 +- hub/src/sync/sessionCache.ts | 31 ++ hub/src/sync/syncEngine.ts | 490 +++++++++++++++++- hub/src/web/routes/sessions.test.ts | 72 ++- hub/src/web/routes/sessions.ts | 69 +++ shared/package.json | 1 + shared/src/apiTypes.ts | 39 ++ shared/src/conversationHistory.test.ts | 25 + shared/src/conversationHistory.ts | 64 +++ shared/src/index.ts | 1 + shared/src/rpcMethods.ts | 4 +- shared/src/schemas.ts | 25 +- web/src/api/client.ts | 20 + .../components/AssistantChat/HappyThread.tsx | 8 + web/src/components/AssistantChat/context.tsx | 4 + .../messages/AssistantMessage.tsx | 20 +- .../messages/MessageActions.test.tsx | 59 ++- .../AssistantChat/messages/MessageActions.tsx | 147 +++++- .../AssistantChat/messages/UserMessage.tsx | 43 +- web/src/components/SessionChat.tsx | 53 +- web/src/lib/locales/en.ts | 8 + web/src/lib/locales/zh-CN.ts | 8 + web/src/types/api.ts | 9 + 59 files changed, 2829 insertions(+), 122 deletions(-) create mode 100644 cli/src/claude/session.consumeOneTimeFlags.test.ts create mode 100644 cli/src/codex/conversationHistory.test.ts create mode 100644 cli/src/codex/conversationHistory.ts create mode 100644 cli/src/grok/conversationHistory.test.ts create mode 100644 cli/src/grok/conversationHistory.ts create mode 100644 hub/src/store/messages.copyBatch.test.ts create mode 100644 hub/src/store/messages.truncate.test.ts create mode 100644 hub/src/sync/forkTranscript.test.ts create mode 100644 hub/src/sync/forkTranscript.ts create mode 100644 shared/src/conversationHistory.test.ts create mode 100644 shared/src/conversationHistory.ts diff --git a/cli/src/agent/backends/acp/AcpSdkBackend.ts b/cli/src/agent/backends/acp/AcpSdkBackend.ts index 06d32db5..4d21f39b 100644 --- a/cli/src/agent/backends/acp/AcpSdkBackend.ts +++ b/cli/src/agent/backends/acp/AcpSdkBackend.ts @@ -388,6 +388,23 @@ export class AcpSdkBackend implements AgentBackend { this.captureSessionMetadata(sessionId, response); } + /** + * Low-level extension RPC for agent-specific methods (e.g. Grok `_x.ai/*`). + * Keep method names and schemas in the agent adapter — not here. + */ + async sendExtensionRequest( + method: string, + params: Record, + options?: { timeoutMs?: number } + ): Promise { + if (!this.transport) { + throw new Error('ACP transport not initialized'); + } + return await this.transport.sendRequest(method, params, { + timeoutMs: options?.timeoutMs + }) as T; + } + /** * Returns the per-session models metadata captured from session/new (or * session/load, or session/set_model). Returns undefined if the agent did diff --git a/cli/src/agent/sessionFactory.test.ts b/cli/src/agent/sessionFactory.test.ts index 0662852d..9c53580d 100644 --- a/cli/src/agent/sessionFactory.test.ts +++ b/cli/src/agent/sessionFactory.test.ts @@ -158,7 +158,11 @@ describe('bootstrapExistingSession', () => { updatedAt: 100 }, tools: ['read_file'], - slashCommands: ['/compact'] + slashCommands: ['/compact'], + capabilities: { + terminal: true, + conversationHistory: { forkCurrent: true } + } } const sessionClient = { updateMetadata: vi.fn() @@ -193,7 +197,11 @@ describe('bootstrapExistingSession', () => { updatedAt: 100 }, tools: ['read_file'], - slashCommands: ['/compact'] + slashCommands: ['/compact'], + capabilities: { + terminal: true, + conversationHistory: { forkCurrent: true } + } })) expect(sessionClient.updateMetadata).toHaveBeenCalledOnce() const updateHandler = sessionClient.updateMetadata.mock.calls[0][0] diff --git a/cli/src/agent/sessionFactory.ts b/cli/src/agent/sessionFactory.ts index 8f256771..93fe9717 100644 --- a/cli/src/agent/sessionFactory.ts +++ b/cli/src/agent/sessionFactory.ts @@ -116,6 +116,27 @@ function pickExistingSessionMetadata(metadata: Metadata | null | undefined): Par if (metadata.piAvailableModels !== undefined) preserved.piAvailableModels = metadata.piAvailableModels // Preserve provider-qualified Pi model selection (disambiguates duplicate modelIds). if (metadata.piSelectedModel !== undefined) preserved.piSelectedModel = metadata.piSelectedModel + if (metadata.conversationHistoryPoints !== undefined) { + preserved.conversationHistoryPoints = metadata.conversationHistoryPoints + } + if (metadata.conversationHistoryIndexes !== undefined) { + preserved.conversationHistoryIndexes = metadata.conversationHistoryIndexes + } + if (metadata.conversationHistoryTurns !== undefined) { + preserved.conversationHistoryTurns = metadata.conversationHistoryTurns + } + if (metadata.conversationHistoryDiverged !== undefined) { + preserved.conversationHistoryDiverged = metadata.conversationHistoryDiverged + } + if (metadata.forkedFrom !== undefined) { + preserved.forkedFrom = metadata.forkedFrom + } + if (metadata.capabilities?.conversationHistory !== undefined) { + preserved.capabilities = { + ...preserved.capabilities, + conversationHistory: metadata.capabilities.conversationHistory + } + } return preserved } @@ -302,17 +323,20 @@ export async function bootstrapExistingSession(options: { workingDirectory: options.workingDirectory, machineId }) - const metadata = { - ...baseMetadata, - ...pickExistingSessionMetadata(sessionInfo.metadata), - ...options.metadataOverrides + const buildUpdatedMetadata = (current: Metadata | null | undefined): Metadata => { + const preserved = pickExistingSessionMetadata(current) + return { + ...baseMetadata, + ...preserved, + ...options.metadataOverrides, + capabilities: { + ...baseMetadata.capabilities, + ...preserved.capabilities, + ...options.metadataOverrides?.capabilities + } + } } - - const buildUpdatedMetadata = (current: Metadata): Metadata => ({ - ...baseMetadata, - ...pickExistingSessionMetadata(current), - ...options.metadataOverrides - }) + const metadata = buildUpdatedMetadata(sessionInfo.metadata) const session = api.sessionSyncClient(sessionInfo) session.updateMetadata(buildUpdatedMetadata) diff --git a/cli/src/api/apiMachine.ts b/cli/src/api/apiMachine.ts index 0759538f..97c9beda 100644 --- a/cli/src/api/apiMachine.ts +++ b/cli/src/api/apiMachine.ts @@ -357,7 +357,7 @@ export class ApiMachineClient { setRPCHandlers({ spawnSession, stopSession, requestShutdown }: MachineRpcHandlers): void { this.rpcHandlerManager.registerHandler(RPC_METHODS.SpawnHappySession, async (params: any) => { - const { directory, sessionId, existingSessionId, resumeSessionId, machineId, approvedNewDirectoryCreation, agent, model, effort, modelReasoningEffort, yolo, permissionMode, serviceTier, collaborationMode, token, sessionType, worktreeName } = params || {} + const { directory, sessionId, existingSessionId, resumeSessionId, machineId, approvedNewDirectoryCreation, agent, model, effort, modelReasoningEffort, yolo, permissionMode, serviceTier, collaborationMode, token, sessionType, worktreeName, forkSession } = params || {} if (!directory) { throw new Error('Directory is required') @@ -385,7 +385,8 @@ export class ApiMachineClient { collaborationMode, token, sessionType, - worktreeName + worktreeName, + forkSession: forkSession === true }) switch (result.type) { diff --git a/cli/src/claude/claudeRemote.ts b/cli/src/claude/claudeRemote.ts index 07043811..8041462d 100644 --- a/cli/src/claude/claudeRemote.ts +++ b/cli/src/claude/claudeRemote.ts @@ -25,6 +25,8 @@ export async function claudeRemote(opts: { hookSettingsPath: string, signal?: AbortSignal, canCallTool: (toolName: string, input: unknown, mode: EnhancedMode, options: { signal: AbortSignal }) => Promise, + /** Session modes used to spawn Claude before the first fork child prompt. */ + bootstrapMode?: EnhancedMode, // Dynamic parameters nextMessage: () => Promise<{ message: string, mode: EnhancedMode } | null>, @@ -32,7 +34,7 @@ export async function claudeRemote(opts: { isAborted: (toolCallId: string) => boolean, // Callbacks - onSessionFound: (id: string) => void, + onSessionFound: (id: string, extras?: { forkedFrom?: string }) => void, onThinkingChange?: (thinking: boolean) => void, onMessage: (message: SDKMessage) => void, onFirstResult?: (initialMessage: string) => void, @@ -81,67 +83,100 @@ export async function claudeRemote(opts: { } process.env.DISABLE_AUTOUPDATER = '1'; - // Get initial message - let initial; - try { - initial = await opts.nextMessage(); - } catch (e) { - if (e instanceof AbortError) { - logger.debug(`[claudeRemote] Aborted during initial message`); - return; - } - throw e; + // Message-level Fork current passes `--fork-session` via claudeArgs from the runner. + const forkSession = Boolean(opts.claudeArgs?.includes('--fork-session')); + if (forkSession) { + logger.debug(`[claudeRemote] --fork-session requested via claudeArgs`); } - if (!initial) { // No initial message - exit - logger.debug(`${debugPrefix} initial nextMessage returned null; exiting`); - return; - } - logger.debug(`${debugPrefix} initial message acquired`); + const forkedFrom = forkSession ? startFrom : null; - // Handle special commands - const specialCommand = parseSpecialCommand(initial.message); - - // Handle /clear command - if (specialCommand.type === 'clear') { - if (opts.onCompletionEvent) { - opts.onCompletionEvent('Context was reset'); - } - if (opts.onSessionReset) { - opts.onSessionReset(); - } - return; - } - - // Handle /compact command - let isCompactCommand = false; + // Mode starts from the persisted session for fork bootstrap; updated when + // the first child prompt arrives. plan/auto must be present at process start. + const bootstrapMode: EnhancedMode = opts.bootstrapMode ?? { permissionMode: 'default' }; + let mode: EnhancedMode = bootstrapMode; + let initial: { message: string; mode: EnhancedMode } | null = null; + let specialCommand: ReturnType = { type: null }; // Claude reports the /compact outcome on a `system`/`status` message that // arrives before the `result` message. Hold it here so the completion event // can report what actually happened. Stays null unless a failure is // reported, so an unseen or successful status keeps the success path. + let isCompactCommand = false; let compactFailure: string | null = null; - if (specialCommand.type === 'compact') { - logger.debug('[claudeRemote] /compact command detected - will process as normal but with compaction behavior'); - isCompactCommand = true; - if (opts.onCompletionEvent) { - opts.onCompletionEvent('Compaction started'); - } - } + let awaitingForkInit = forkSession; - // Prepare SDK options - let mode = initial.mode; + const messages = new PushableAsyncIterable(); + + const applyInitialTurn = async (): Promise<{ message: string; mode: EnhancedMode } | null> => { + let next: { message: string; mode: EnhancedMode } | null; + try { + next = await opts.nextMessage(); + } catch (e) { + if (e instanceof AbortError) { + logger.debug(`[claudeRemote] Aborted during initial message`); + messages.end(); + return null; + } + throw e; + } + if (!next) { + logger.debug(`${debugPrefix} initial nextMessage returned null; exiting`); + messages.end(); + return null; + } + logger.debug(`${debugPrefix} initial message acquired`); + + specialCommand = parseSpecialCommand(next.message); + if (specialCommand.type === 'clear') { + if (opts.onCompletionEvent) { + opts.onCompletionEvent('Context was reset'); + } + if (opts.onSessionReset) { + opts.onSessionReset(); + } + messages.end(); + return null; + } + if (specialCommand.type === 'compact') { + logger.debug('[claudeRemote] /compact command detected - will process as normal but with compaction behavior'); + isCompactCommand = true; + if (opts.onCompletionEvent) { + opts.onCompletionEvent('Compaction started'); + } + } + + mode = next.mode; + messages.push({ + type: 'user', + message: { + role: 'user', + content: next.message, + }, + }); + return next; + }; + + // Prepare SDK options. For --fork-session, start query() before waiting for the + // first child prompt so the native fork materializes at the clicked source state. const sdkOptions: Options = { additionalArgs: filterCatalogAffectingClaudeArgs(opts.claudeArgs), cwd: opts.path, resume: startFrom ?? undefined, + forkSession, mcpServers: opts.mcpServers, - permissionMode: initial.mode.permissionMode, - model: initial.mode.model, - effort: initial.mode.effort, - fallbackModel: initial.mode.fallbackModel, - customSystemPrompt: initial.mode.customSystemPrompt ? initial.mode.customSystemPrompt + '\n\n' + systemPrompt : undefined, - appendSystemPrompt: initial.mode.appendSystemPrompt ? initial.mode.appendSystemPrompt + '\n\n' + systemPrompt : systemPrompt, - allowedTools: initial.mode.allowedTools ? initial.mode.allowedTools.concat(opts.allowedTools) : opts.allowedTools, - disallowedTools: initial.mode.disallowedTools, + permissionMode: bootstrapMode.permissionMode, + model: bootstrapMode.model, + effort: bootstrapMode.effort, + fallbackModel: bootstrapMode.fallbackModel, + customSystemPrompt: bootstrapMode.customSystemPrompt + ? bootstrapMode.customSystemPrompt + '\n\n' + systemPrompt + : undefined, + appendSystemPrompt: bootstrapMode.appendSystemPrompt + ? bootstrapMode.appendSystemPrompt + '\n\n' + systemPrompt + : systemPrompt, + allowedTools: bootstrapMode.allowedTools + ? bootstrapMode.allowedTools.concat(opts.allowedTools) + : opts.allowedTools, + disallowedTools: bootstrapMode.disallowedTools, canCallTool: (toolName: string, input: unknown, options: { signal: AbortSignal }) => opts.canCallTool(toolName, input, mode, options), abort: opts.signal, pathToClaudeCodeExecutable: getDefaultClaudeCodePath(), @@ -149,6 +184,28 @@ export async function claudeRemote(opts: { additionalDirectories: [getHapiBlobsDir()], } + if (!awaitingForkInit) { + const first = await applyInitialTurn(); + if (!first) { + return; + } + initial = first; + sdkOptions.permissionMode = first.mode.permissionMode; + sdkOptions.model = first.mode.model; + sdkOptions.effort = first.mode.effort; + sdkOptions.fallbackModel = first.mode.fallbackModel; + sdkOptions.customSystemPrompt = first.mode.customSystemPrompt + ? first.mode.customSystemPrompt + '\n\n' + systemPrompt + : undefined; + sdkOptions.appendSystemPrompt = first.mode.appendSystemPrompt + ? first.mode.appendSystemPrompt + '\n\n' + systemPrompt + : systemPrompt; + sdkOptions.allowedTools = first.mode.allowedTools + ? first.mode.allowedTools.concat(opts.allowedTools) + : opts.allowedTools; + sdkOptions.disallowedTools = first.mode.disallowedTools; + } + // Track thinking state let thinking = false; const updateThinking = (newThinking: boolean) => { @@ -161,16 +218,6 @@ export async function claudeRemote(opts: { } }; - // Push initial message - let messages = new PushableAsyncIterable(); - messages.push({ - type: 'user', - message: { - role: 'user', - content: initial.message, - }, - }); - // Start the loop const response = query({ prompt: messages, @@ -258,7 +305,20 @@ export async function claudeRemote(opts: { const projectDir = getProjectPath(opts.path); const found = await awaitFileExist(join(projectDir, `${systemInit.session_id}.jsonl`)); logger.debug(`[claudeRemote] Session file found: ${systemInit.session_id} ${found}`); - opts.onSessionFound(systemInit.session_id); + const extras = forkedFrom && forkedFrom !== systemInit.session_id + ? { forkedFrom } + : undefined; + opts.onSessionFound(systemInit.session_id, extras); + } + + // Fork: only accept the first child prompt after the native branch exists. + if (awaitingForkInit) { + awaitingForkInit = false; + const first = await applyInitialTurn(); + if (!first) { + return; + } + initial = first; } } @@ -285,7 +345,7 @@ export async function claudeRemote(opts: { `(nextInFlight=${nextMessageFetchInFlight}, inputEnded=${inputEnded})` ); - if (resultSeq === 1 && specialCommand.type === null) { + if (resultSeq === 1 && specialCommand.type === null && initial) { opts.onFirstResult?.(initial.message); } diff --git a/cli/src/claude/claudeRemoteLauncher.ts b/cli/src/claude/claudeRemoteLauncher.ts index 4da0dc04..2fa8e541 100644 --- a/cli/src/claude/claudeRemoteLauncher.ts +++ b/cli/src/claude/claudeRemoteLauncher.ts @@ -387,6 +387,11 @@ class ClaudeRemoteLauncher extends RemoteLauncherBase { mcpServers: session.mcpServers, hookSettingsPath: session.hookSettingsPath, canCallTool: permissionHandler.handleToolCall, + bootstrapMode: { + permissionMode: session.getPermissionMode() ?? 'default', + model: session.getModel() ?? undefined, + effort: session.getEffort() ?? undefined, + }, isAborted: (toolCallId: string) => { return permissionHandler.isAborted(toolCallId); }, diff --git a/cli/src/claude/runClaude.ts b/cli/src/claude/runClaude.ts index 1b14c497..8d51680c 100644 --- a/cli/src/claude/runClaude.ts +++ b/cli/src/claude/runClaude.ts @@ -23,6 +23,10 @@ import { normalizeClaudeSessionModel } from './model'; import { normalizeClaudeSessionEffort } from './effort'; import { normalizeHookPermissionMode } from './utils/hookPermissionMode'; import { getInvokedCwd } from '@/utils/invokedCwd'; +import { + CLAUDE_CONVERSATION_HISTORY, + toConversationHistoryCapabilities +} from '@hapi/protocol/conversationHistory'; import { listSkills, type SkillSummary } from '@/modules/common/skills'; export interface StartOptions { @@ -229,6 +233,34 @@ export async function runClaude(options: StartOptions = {}): Promise { registerKillSessionHandler(session.rpcHandlerManager, lifecycle); registerLocalHandoffHandler(session.rpcHandlerManager, lifecycle); + const conversationHistory = toConversationHistoryCapabilities(CLAUDE_CONVERSATION_HISTORY) + session.updateMetadata((metadata) => ({ + ...metadata, + path: metadata?.path ?? workingDirectory, + host: metadata?.host ?? 'unknown', + capabilities: { + ...metadata?.capabilities, + ...(conversationHistory ? { conversationHistory } : {}) + } + })) + + session.rpcHandlerManager.registerHandler(RPC_METHODS.ForkConversation, async (payload: unknown) => { + if (payload && typeof payload === 'object' && 'messageLocalId' in payload && (payload as { messageLocalId?: unknown }).messageLocalId) { + throw new Error('Historical fork is not supported for Claude') + } + const nativeSessionId = currentSessionRef.current?.sessionId + ?? session.getMetadata()?.claudeSessionId + ?? sessionInfo.metadata?.claudeSessionId + ?? null + if (!nativeSessionId) { + throw new Error('Claude session id is not ready') + } + return { nativeSessionId, forkSession: true as const } + }) + session.rpcHandlerManager.registerHandler(RPC_METHODS.RewindConversation, async () => { + throw new Error('Rewind is not supported for Claude') + }) + // Set initial agent state const startingMode = options.startingMode ?? (startedBy === 'runner' ? 'remote' : 'local'); setControlledByUser(session, startingMode); diff --git a/cli/src/claude/sdk/query.ts b/cli/src/claude/sdk/query.ts index a8fc1d08..f8d43348 100644 --- a/cli/src/claude/sdk/query.ts +++ b/cli/src/claude/sdk/query.ts @@ -310,6 +310,7 @@ export function query(config: { permissionMode = 'default', continue: continueConversation, resume, + forkSession, model, effort, fallbackModel, @@ -342,6 +343,7 @@ export function query(config: { } if (continueConversation) args.push('--continue') if (resume) args.push('--resume', resume) + if (forkSession) args.push('--fork-session') args.push(...additionalArgs) if (settingsPath) args.push('--settings', settingsPath) if (allowedTools.length > 0) args.push('--allowedTools', allowedTools.join(',')) diff --git a/cli/src/claude/sdk/types.ts b/cli/src/claude/sdk/types.ts index f05616c9..a05bbf66 100644 --- a/cli/src/claude/sdk/types.ts +++ b/cli/src/claude/sdk/types.ts @@ -195,6 +195,11 @@ export interface QueryOptions { permissionMode?: ClaudePermissionMode continue?: boolean resume?: string + /** + * When resuming, branch with `--fork-session` instead of taking over the + * existing Claude session id. + */ + forkSession?: boolean model?: string effort?: string fallbackModel?: string diff --git a/cli/src/claude/session.consumeOneTimeFlags.test.ts b/cli/src/claude/session.consumeOneTimeFlags.test.ts new file mode 100644 index 00000000..bda3f1e1 --- /dev/null +++ b/cli/src/claude/session.consumeOneTimeFlags.test.ts @@ -0,0 +1,43 @@ +import { describe, expect, it } from 'vitest' +import { Session } from './session' + +function makeSession(claudeArgs: string[] | undefined): Session { + return new Session({ + api: {} as never, + client: { + updateMetadata() {}, + keepAlive() {}, + emitMessagesConsumed() {} + } as never, + path: '/tmp', + logPath: '/tmp/test.log', + sessionId: null, + claudeArgs, + mcpServers: {}, + messageQueue: { onBatchConsumed: null } as never, + onModeChange: () => {}, + startedBy: 'runner', + startingMode: 'remote', + hookSettingsPath: '/tmp/hooks.json' + }) +} + +describe('Session.consumeOneTimeFlags', () => { + it('consumes --resume and --fork-session together', () => { + const session = makeSession(['--resume', 'claude-source-id', '--fork-session', '--permission-mode', 'default']) + session.consumeOneTimeFlags() + expect(session.claudeArgs).toEqual(['--permission-mode', 'default']) + }) + + it('consumes a lone --fork-session flag', () => { + const session = makeSession(['--fork-session']) + session.consumeOneTimeFlags() + expect(session.claudeArgs).toBeUndefined() + }) + + it('leaves unrelated args alone', () => { + const session = makeSession(['--permission-mode', 'acceptEdits']) + session.consumeOneTimeFlags() + expect(session.claudeArgs).toEqual(['--permission-mode', 'acceptEdits']) + }) +}) diff --git a/cli/src/claude/session.ts b/cli/src/claude/session.ts index 1dfc2ffa..06fd37ce 100644 --- a/cli/src/claude/session.ts +++ b/cli/src/claude/session.ts @@ -122,8 +122,10 @@ export class Session extends AgentSessionBase { }; /** - * Consume one-time Claude flags from claudeArgs after Claude spawn - * Currently handles: --resume (with or without session ID) + * Consume one-time Claude flags from claudeArgs after Claude spawn. + * Handles: --resume (with or without session ID) and --fork-session. + * `--fork-session` must be one-shot; keeping it across relaunches would + * branch again off the already-forked native id. */ consumeOneTimeFlags = (): void => { if (!this.claudeArgs) return; @@ -147,6 +149,8 @@ export class Session extends AgentSessionBase { // --resume at the end of args logger.debug('[Session] Consumed --resume flag (no session ID)'); } + } else if (this.claudeArgs[i] === '--fork-session') { + logger.debug('[Session] Consumed --fork-session flag'); } else { filteredArgs.push(this.claudeArgs[i]); } diff --git a/cli/src/codex/appServerTypes.ts b/cli/src/codex/appServerTypes.ts index 6791ca1f..6615b662 100644 --- a/cli/src/codex/appServerTypes.ts +++ b/cli/src/codex/appServerTypes.ts @@ -156,7 +156,28 @@ export interface ThreadResumeResponse { [key: string]: unknown; } +export interface ThreadReadParams { + threadId: string; + includeTurns?: boolean; +} + +export interface ThreadReadResponse { + thread: { + id: string; + turns?: Array<{ + id?: string; + status?: string; + items?: ResponseItem[]; + }>; + }; + [key: string]: unknown; +} + export interface ThreadForkParams extends Omit { + /** Inclusive terminal turn for the fork (stable). */ + lastTurnId?: string | null; + /** Exclusive: copy history strictly before this turn (experimental). */ + beforeTurnId?: string | null; } export interface ThreadForkResponse { @@ -239,6 +260,8 @@ export interface TurnStartParams { personality?: string; outputSchema?: unknown; collaborationMode?: CollaborationMode; + /** Optional client identity echoed back as userMessage.clientId. */ + clientUserMessageId?: string; } export interface TurnStartResponse { diff --git a/cli/src/codex/codexAppServerClient.ts b/cli/src/codex/codexAppServerClient.ts index 54411101..5c0d6a77 100644 --- a/cli/src/codex/codexAppServerClient.ts +++ b/cli/src/codex/codexAppServerClient.ts @@ -17,6 +17,8 @@ import type { ThreadResumeResponse, ThreadForkParams, ThreadForkResponse, + ThreadReadParams, + ThreadReadResponse, TurnStartParams, TurnStartResponse, TurnInterruptParams, @@ -294,6 +296,25 @@ export class CodexAppServerClient extends JsonLineParser { return response as ThreadForkResponse; } + async supportsMethod(method: 'thread/fork' | 'thread/rollback'): Promise { + try { + await this.sendRequest(method, { threadId: '__hapi_capability_probe__' }, { timeoutMs: 30_000 }); + return true; + } catch (error) { + return !/method not found|unknown method|unsupported/i.test( + error instanceof Error ? error.message : String(error) + ); + } + } + + async readThread(params: ThreadReadParams, options?: { signal?: AbortSignal }): Promise { + const response = await this.sendRequest('thread/read', params, { + signal: options?.signal, + timeoutMs: CodexAppServerClient.DEFAULT_TIMEOUT_MS + }); + return response as ThreadReadResponse; + } + async startTurn(params: TurnStartParams, options?: { signal?: AbortSignal }): Promise { const response = await this.sendRequest('turn/start', params, { signal: options?.signal, diff --git a/cli/src/codex/codexRemoteLauncher.test.ts b/cli/src/codex/codexRemoteLauncher.test.ts index 2dc6d1a0..36bffd99 100644 --- a/cli/src/codex/codexRemoteLauncher.test.ts +++ b/cli/src/codex/codexRemoteLauncher.test.ts @@ -1018,6 +1018,7 @@ function createSessionStub( rpcHandlers.set(method, handler); } }, + updateMetadata(_handler: (metadata: Record) => Record) {}, updateAgentState(handler: (state: FakeAgentState) => FakeAgentState) { agentState = handler(agentState); }, diff --git a/cli/src/codex/codexRemoteLauncher.ts b/cli/src/codex/codexRemoteLauncher.ts index e877b202..b6d52767 100644 --- a/cli/src/codex/codexRemoteLauncher.ts +++ b/cli/src/codex/codexRemoteLauncher.ts @@ -27,6 +27,7 @@ import { type RemoteLauncherDisplayContext, type RemoteLauncherExitReason } from '@/modules/common/remote/RemoteLauncherBase'; +import { CodexConversationHistory } from './conversationHistory'; async function registerGeneratedImageFromPath(args: { id: string; path: string; fileName?: string | null }): Promise | null> { @@ -58,7 +59,13 @@ async function registerGeneratedImageFromPath(args: { id: string; path: string; } type HappyServer = Awaited>['server']; -type QueuedMessage = { message: string; mode: EnhancedMode; isolate: boolean; hash: string }; +type QueuedMessage = { + message: string + mode: EnhancedMode + isolate: boolean + hash: string + items?: Array<{ message: string; localId?: string }> +} type ChildAgentRuntime = { reasoningProcessor: ReasoningProcessor; diffProcessor: DiffProcessor; @@ -227,6 +234,7 @@ class CodexRemoteLauncher extends RemoteLauncherBase { private currentThreadId: string | null = null; private currentTurnId: string | null = null; private readonly activeChildTurns = new Map(); + readonly conversationHistory = new CodexConversationHistory(() => this.appServerClient); constructor(session: CodexSession) { super(process.env.DEBUG ? session.logPath : undefined); @@ -2360,6 +2368,8 @@ class CodexRemoteLauncher extends RemoteLauncherBase { if (threadId) { if (!this.currentThreadId || this.currentThreadId === threadId) { this.currentThreadId = threadId; + this.conversationHistory.setThreadId(threadId); + void this.conversationHistory.probeCapabilities().catch(() => {}); session.onSessionFound(threadId); } else { logger.debug( @@ -2740,6 +2750,7 @@ class CodexRemoteLauncher extends RemoteLauncherBase { } if (isTerminalEvent) { turnInFlight = false; + this.conversationHistory.setBusy(false); allowAnonymousTerminalEvent = false; if (session.thinking) { logger.debug('thinking completed'); @@ -3198,6 +3209,44 @@ class CodexRemoteLauncher extends RemoteLauncherBase { } }); + const publishConversationHistoryCapabilities = async () => { + const conversationHistory = this.conversationHistory.getCapabilitiesForMetadata()?.conversationHistory + try { + session.client.updateMetadata((metadata) => { + const capabilities = { ...metadata?.capabilities } + delete capabilities.conversationHistory + if (conversationHistory) { + capabilities.conversationHistory = conversationHistory + } + return { + ...metadata, + path: metadata?.path ?? session.path, + host: metadata?.host ?? 'unknown', + capabilities + } + }) + } catch { + // best-effort; tests and transient hub disconnects must not crash the loop + } + } + this.conversationHistory.setPublishCapabilities(publishConversationHistoryCapabilities) + this.conversationHistory.restoreTurns( + typeof session.client.getMetadata === 'function' + ? session.client.getMetadata()?.conversationHistoryTurns + : undefined + ) + session.client.rpcHandlerManager.registerHandler(RPC_METHODS.ForkConversation, async (payload: unknown) => { + const messageLocalId = payload && typeof payload === 'object' && typeof (payload as { messageLocalId?: unknown }).messageLocalId === 'string' + ? (payload as { messageLocalId: string }).messageLocalId + : undefined + return await this.conversationHistory.fork(messageLocalId) + }) + session.client.rpcHandlerManager.registerHandler(RPC_METHODS.RewindConversation, async (payload: unknown) => { + if (!payload || typeof payload !== 'object' || typeof (payload as { messageLocalId?: unknown }).messageLocalId !== 'string') { + throw new Error('messageLocalId is required') + } + return await this.conversationHistory.rewind((payload as { messageLocalId: string }).messageLocalId) + }) try { await refreshNativeSkills(false); } catch (error) { @@ -3325,6 +3374,8 @@ class CodexRemoteLauncher extends RemoteLauncherBase { const threadId = asString(resumeThread?.id) ?? resumeCandidate; applyResolvedModel(resumeRecord?.model); this.currentThreadId = threadId; + this.conversationHistory.setThreadId(threadId); + void this.conversationHistory.probeCapabilities().catch(() => {}); session.onSessionFound(threadId); hasThread = true; logger.debug(`[Codex] Resumed app-server thread ${threadId} for /compact`); @@ -3387,6 +3438,8 @@ class CodexRemoteLauncher extends RemoteLauncherBase { const threadId = asString(resumeThread?.id) ?? resumeCandidate; applyResolvedModel(resumeRecord?.model); this.currentThreadId = threadId; + this.conversationHistory.setThreadId(threadId); + void this.conversationHistory.probeCapabilities().catch(() => {}); session.onSessionFound(threadId); hasThread = true; return threadId; @@ -3418,6 +3471,8 @@ class CodexRemoteLauncher extends RemoteLauncherBase { throw new Error('app-server thread/start did not return thread.id'); } this.currentThreadId = threadId; + this.conversationHistory.setThreadId(threadId); + void this.conversationHistory.probeCapabilities().catch(() => {}); session.onSessionFound(threadId); hasThread = true; return threadId; @@ -3709,6 +3764,8 @@ class CodexRemoteLauncher extends RemoteLauncherBase { } this.currentThreadId = threadId; + this.conversationHistory.setThreadId(threadId); + void this.conversationHistory.probeCapabilities().catch(() => {}); session.onSessionFound(threadId); hasThread = true; } else { @@ -3721,6 +3778,7 @@ class CodexRemoteLauncher extends RemoteLauncherBase { } turnInFlight = true; + this.conversationHistory.setBusy(true); allowAnonymousTerminalEvent = false; const mode = { ...message.mode, @@ -3728,12 +3786,16 @@ class CodexRemoteLauncher extends RemoteLauncherBase { }; const shouldSendCollaborationMode = supportsTurnCollaborationMode && Boolean(mode.collaborationMode); + const clientUserMessageId = message.items + ?.map((item) => item.localId) + .find((id): id is string => typeof id === 'string' && id.length > 0); const buildParams = (suppressCollaborationMode: boolean) => buildTurnStartParams({ threadId: this.currentThreadId!, message: message.message, cwd: session.path, mode, cliOverrides: session.codexCliOverrides, + clientUserMessageId, skills: nativeSkills, overrides: suppressCollaborationMode ? { suppressCollaborationMode: true } @@ -3775,6 +3837,22 @@ class CodexRemoteLauncher extends RemoteLauncherBase { if (turnInFlight) { if (turnId) { this.currentTurnId = turnId; + if (clientUserMessageId) { + this.conversationHistory.rememberLocalIdTurn(clientUserMessageId, turnId); + session.client.updateMetadata((metadata) => ({ + ...metadata, + path: metadata?.path ?? session.path, + host: metadata?.host ?? 'unknown', + conversationHistoryPoints: { + ...metadata?.conversationHistoryPoints, + [clientUserMessageId]: true as const + }, + conversationHistoryTurns: { + ...metadata?.conversationHistoryTurns, + [clientUserMessageId]: turnId + } + })) + } } else if (!this.currentTurnId) { allowAnonymousTerminalEvent = true; } @@ -3783,6 +3861,7 @@ class CodexRemoteLauncher extends RemoteLauncherBase { logger.warn('Error in codex session:', error); const isAbortError = error instanceof Error && error.name === 'AbortError'; turnInFlight = false; + this.conversationHistory.setBusy(false); allowAnonymousTerminalEvent = false; this.currentTurnId = null; diff --git a/cli/src/codex/conversationHistory.test.ts b/cli/src/codex/conversationHistory.test.ts new file mode 100644 index 00000000..8bd6ee4f --- /dev/null +++ b/cli/src/codex/conversationHistory.test.ts @@ -0,0 +1,143 @@ +import { describe, expect, it, vi } from 'vitest' +import { CodexConversationHistory } from './conversationHistory' + +function createClient(overrides?: { + fork?: (params: Record) => Promise<{ thread: { id: string } }> + rollback?: (params: { threadId: string; numTurns: number }) => Promise + read?: () => Promise<{ thread: { id: string; turns: Array> } }> +}) { + return { + supportsMethod: async () => true, + forkThread: overrides?.fork ?? (async () => ({ thread: { id: 'forked-1' } })), + rollbackThread: overrides?.rollback ?? (async () => ({ thread: { id: 'thread-1' } })), + readThread: overrides?.read ?? (async () => ({ + thread: { + id: 'thread-1', + turns: [ + { id: 'turn-a', items: [{ type: 'userMessage', clientId: 'local-a' }] }, + { id: 'turn-b', items: [{ type: 'userMessage', clientId: 'local-b' }] }, + { id: 'turn-c', items: [{ type: 'userMessage', clientId: 'local-c' }] } + ] + } + })) + } +} + +describe('CodexConversationHistory', () => { + it('only publishes methods confirmed by the app server', async () => { + const supportsMethod = vi.fn(async (method: string) => method === 'thread/fork') + const history = new CodexConversationHistory(() => ({ + ...createClient(), + supportsMethod + }) as never) + history.setThreadId('thread-1') + await history.probeCapabilities() + expect(history.getCapabilitiesForMetadata()?.conversationHistory).toEqual({ + forkCurrent: true, + forkAtMessage: true + }) + }) + + it('forks current without a turn boundary', async () => { + const fork = vi.fn(async (params: Record) => { + expect(params.beforeTurnId).toBeUndefined() + return { thread: { id: 'forked-current' } } + }) + const history = new CodexConversationHistory(() => createClient({ fork }) as never) + history.setThreadId('thread-1') + const result = await history.fork() + expect(result).toEqual({ nativeSessionId: 'forked-current' }) + expect(fork).toHaveBeenCalledTimes(1) + }) + + it('historical fork passes lastTurnId of the previous turn', async () => { + const fork = vi.fn(async (params: Record) => { + expect(params.lastTurnId).toBe('turn-a') + expect(params.beforeTurnId).toBeUndefined() + return { thread: { id: 'forked-hist' } } + }) + const history = new CodexConversationHistory(() => createClient({ fork }) as never) + history.setThreadId('thread-1') + const result = await history.fork('local-b') + expect(result.nativeSessionId).toBe('forked-hist') + }) + + it('historical fork of the first turn uses beforeTurnId', async () => { + const fork = vi.fn(async (params: Record) => { + expect(params.beforeTurnId).toBe('turn-a') + expect(params.lastTurnId).toBeUndefined() + return { thread: { id: 'forked-first' } } + }) + const history = new CodexConversationHistory(() => createClient({ fork }) as never) + history.setThreadId('thread-1') + const result = await history.fork('local-a') + expect(result.nativeSessionId).toBe('forked-first') + }) + + it('computes rewind numTurns from selected turn', async () => { + const rollback = vi.fn(async (params: { threadId: string; numTurns: number }) => { + expect(params).toEqual({ threadId: 'thread-1', numTurns: 2 }) + return { thread: { id: 'thread-1' } } + }) + const history = new CodexConversationHistory(() => createClient({ rollback }) as never) + history.setThreadId('thread-1') + const result = await history.rewind('local-b') + expect(result).toEqual({ + success: true, + truncateFromLocalId: 'local-b', + messages: [] + }) + expect(rollback).toHaveBeenCalledTimes(1) + }) + + it('marks rewind unsupported on method-not-found without affecting fork', async () => { + const rollback = vi.fn(async () => { + throw new Error('thread/rollback is unsupported') + }) + const fork = vi.fn(async () => ({ thread: { id: 'forked-ok' } })) + const history = new CodexConversationHistory(() => createClient({ rollback, fork }) as never) + history.setThreadId('thread-1') + await expect(history.rewind('local-a')).rejects.toThrow(/unsupported/) + const caps = history.getCapabilitiesForMetadata()?.conversationHistory + expect(caps?.rewindToMessage).toBeUndefined() + const forked = await history.fork() + expect(forked.nativeSessionId).toBe('forked-ok') + }) + + it('does not call native fork when selected turn is missing', async () => { + const fork = vi.fn(async () => ({ thread: { id: 'x' } })) + const history = new CodexConversationHistory(() => createClient({ + fork, + read: async () => ({ thread: { id: 'thread-1', turns: [] } }) + }) as never) + history.setThreadId('thread-1') + await expect(history.fork('missing-local')).rejects.toThrow(/No native history point/) + expect(fork).not.toHaveBeenCalled() + }) + + it('restores durable localId→turnId locators across relaunches', async () => { + const fork = vi.fn(async (params: Record) => { + expect(params.lastTurnId).toBe('turn-a') + expect(params.beforeTurnId).toBeUndefined() + return { thread: { id: 'forked-restored' } } + }) + const history = new CodexConversationHistory(() => createClient({ + fork, + // Simulate a relaunch where thread/read no longer exposes clientIds. + read: async () => ({ + thread: { + id: 'thread-1', + turns: [ + { id: 'turn-a', items: [] }, + { id: 'turn-b', items: [] } + ] + } + }) + }) as never) + history.setThreadId('thread-1') + history.restoreTurns({ 'local-b': 'turn-b' }) + const result = await history.fork('local-b') + expect(result.nativeSessionId).toBe('forked-restored') + expect(history.getTurns()['local-b']).toBe('turn-b') + }) +}) diff --git a/cli/src/codex/conversationHistory.ts b/cli/src/codex/conversationHistory.ts new file mode 100644 index 00000000..1cc9580a --- /dev/null +++ b/cli/src/codex/conversationHistory.ts @@ -0,0 +1,265 @@ +import type { CodexAppServerClient } from './codexAppServerClient' +import type { Metadata } from '@/api/types' +import { + CODEX_CONVERSATION_HISTORY_INITIAL, + markSupported, + markUnsupported, + toConversationHistoryCapabilities, + type ConversationHistoryCapabilityStates +} from '@hapi/protocol/conversationHistory' +import type { + ForkConversationRpcResult, + RewindConversationRpcResult +} from '@hapi/protocol/apiTypes' +import { logger } from '@/ui/logger' + +function asRecord(value: unknown): Record | null { + return value !== null && typeof value === 'object' && !Array.isArray(value) + ? value as Record + : null +} + +function asString(value: unknown): string | null { + return typeof value === 'string' && value.length > 0 ? value : null +} + +function isMethodNotFound(error: unknown): boolean { + const message = error instanceof Error ? error.message : String(error) + return /method not found|unknown method|unsupported/i.test(message) +} + +type TurnInfo = { + id: string + status?: string + clientIds: string[] +} + +export class CodexConversationHistory { + private states: ConversationHistoryCapabilityStates = { ...CODEX_CONVERSATION_HISTORY_INITIAL } + private threadId: string | null = null + private readonly turnByLocalId = new Map() + private busy = false + private publishCapabilities: (() => Promise) | null = null + + constructor(private readonly getClient: () => CodexAppServerClient | null) {} + + setPublishCapabilities(fn: () => Promise): void { + this.publishCapabilities = fn + } + + setBusy(busy: boolean): void { + this.busy = busy + } + + setThreadId(threadId: string | null): void { + this.threadId = threadId + } + + restoreTurns(turns: Record | null | undefined): void { + if (!turns) return + for (const [localId, turnId] of Object.entries(turns)) { + if (localId && turnId) this.turnByLocalId.set(localId, turnId) + } + } + + getTurns(): Record { + return Object.fromEntries(this.turnByLocalId.entries()) + } + + rememberLocalIdTurn(localId: string | undefined, turnId: string | null | undefined): void { + if (!localId || !turnId) return + this.turnByLocalId.set(localId, turnId) + } + + getCapabilityStates(): ConversationHistoryCapabilityStates { + return this.states + } + + getCapabilitiesForMetadata(): Metadata['capabilities'] { + const conversationHistory = toConversationHistoryCapabilities(this.states) + return conversationHistory ? { conversationHistory } : undefined + } + + /** Probe fork/rollback once thread is live. Never optimistic. */ + async probeCapabilities(): Promise { + const client = this.getClient() + const threadId = this.threadId + if (!client || !threadId) return + + if (this.states.forkCurrent === 'unknown' || this.states.forkAtMessage === 'unknown') { + if (await client.supportsMethod('thread/fork')) { + this.states = markSupported(this.states, 'forkCurrent') + this.states = markSupported(this.states, 'forkAtMessage') + } else { + this.states = markUnsupported(this.states, 'forkCurrent') + this.states = markUnsupported(this.states, 'forkAtMessage') + } + } + + if (this.states.rewindToMessage === 'unknown') { + this.states = await client.supportsMethod('thread/rollback') + ? markSupported(this.states, 'rewindToMessage') + : markUnsupported(this.states, 'rewindToMessage') + } + + await this.publishCapabilities?.() + } + + async fork(messageLocalId?: string): Promise { + if (this.busy) throw new Error('Session is busy') + const client = this.getClient() + const threadId = this.threadId + if (!client || !threadId) throw new Error('Codex thread is not ready') + + if (messageLocalId) { + if (this.states.forkAtMessage === 'unsupported') { + throw new Error('Historical fork is not supported') + } + // HAPI historical fork excludes the selected boundary turn. Prefer the + // stable inclusive `lastTurnId` of the previous turn over experimental + // `beforeTurnId`, so native context matches the hydrated transcript. + const turns = await this.listTurns() + const selectedTurnId = await this.resolveTurnId(messageLocalId, turns) + const selectedIndex = turns.findIndex((turn) => turn.id === selectedTurnId) + if (selectedIndex < 0) { + throw new Error('Selected turn not found') + } + // Prefer stable inclusive lastTurnId of the previous turn. The first + // turn has no predecessor, so fall back to experimental beforeTurnId + // (exclusive) for that single boundary. + const boundary = selectedIndex === 0 + ? { beforeTurnId: selectedTurnId } + : { lastTurnId: turns[selectedIndex - 1]!.id } + try { + const response = await client.forkThread({ + threadId, + ...boundary + }) + const nativeSessionId = asString(asRecord(response.thread)?.id) + if (!nativeSessionId) throw new Error('thread/fork did not return thread.id') + this.states = markSupported(this.states, 'forkAtMessage') + this.states = markSupported(this.states, 'forkCurrent') + await this.publishCapabilities?.() + return { nativeSessionId } + } catch (error) { + if (isMethodNotFound(error)) { + this.states = markUnsupported(this.states, 'forkAtMessage') + await this.publishCapabilities?.() + } + throw error + } + } + + if (this.states.forkCurrent === 'unsupported') { + throw new Error('Fork current is not supported') + } + try { + const response = await client.forkThread({ threadId }) + const nativeSessionId = asString(asRecord(response.thread)?.id) + if (!nativeSessionId) throw new Error('thread/fork did not return thread.id') + this.states = markSupported(this.states, 'forkCurrent') + await this.publishCapabilities?.() + return { nativeSessionId } + } catch (error) { + if (isMethodNotFound(error)) { + this.states = markUnsupported(this.states, 'forkCurrent') + await this.publishCapabilities?.() + } + throw error + } + } + + async rewind(messageLocalId: string): Promise { + if (this.busy) throw new Error('Session is busy') + const client = this.getClient() + const threadId = this.threadId + if (!client || !threadId) throw new Error('Codex thread is not ready') + if (this.states.rewindToMessage === 'unsupported') { + throw new Error('Rewind is not supported') + } + + const turns = await this.listTurns() + const turnId = await this.resolveTurnId(messageLocalId, turns) + const index = turns.findIndex((turn) => turn.id === turnId) + if (index < 0) throw new Error('Selected turn not found') + if (turns[index]?.status === 'inProgress' || turns[index]?.status === 'in_progress') { + throw new Error('Cannot rewind an in-progress turn') + } + const numTurns = turns.length - index + if (numTurns <= 0) throw new Error('Invalid rewind count') + + try { + await client.rollbackThread({ threadId, numTurns }) + this.states = markSupported(this.states, 'rewindToMessage') + await this.publishCapabilities?.() + } catch (error) { + if (isMethodNotFound(error)) { + this.states = markUnsupported(this.states, 'rewindToMessage') + await this.publishCapabilities?.() + throw new Error('thread/rollback is unsupported') + } + throw error + } + + // Re-read remaining turns for hydrate; return empty messages so hub truncates + // and child clients reset via epoch. Native history is source of truth on resume. + return { + success: true, + truncateFromLocalId: messageLocalId, + messages: [] + } + } + + private async resolveTurnId(localId: string, turns?: TurnInfo[]): Promise { + const cached = this.turnByLocalId.get(localId) + if (cached) return cached + + const list = turns ?? await this.listTurns() + for (const turn of list) { + if (turn.clientIds.includes(localId)) { + this.turnByLocalId.set(localId, turn.id) + return turn.id + } + } + throw new Error(`No native history point for message ${localId}`) + } + + private async listTurns(): Promise { + const client = this.getClient() + const threadId = this.threadId + if (!client || !threadId) return [] + + try { + const response = await client.readThread({ threadId, includeTurns: true }) + const thread = asRecord(response.thread) + const turns = Array.isArray(thread?.turns) ? thread.turns : [] + return turns.flatMap((entry) => { + const record = asRecord(entry) + const id = asString(record?.id) + if (!id) return [] + const clientIds: string[] = [] + const items = Array.isArray(record?.items) ? record.items : [] + for (const item of items) { + const itemRecord = asRecord(item) + const type = asString(itemRecord?.type) ?? asString(itemRecord?.itemType) + if (type === 'userMessage' || type === 'user_message') { + const clientId = asString(itemRecord?.clientId) ?? asString(itemRecord?.client_id) + if (clientId) clientIds.push(clientId) + } + } + return [{ + id, + status: asString(record?.status) ?? undefined, + clientIds + }] + }) + } catch (error) { + logger.debug(`[Codex] thread/read failed: ${error instanceof Error ? error.message : String(error)}`) + // Fall back to in-memory mapping only + return Array.from(this.turnByLocalId.entries()).map(([localId, id]) => ({ + id, + clientIds: [localId] + })) + } + } +} diff --git a/cli/src/codex/utils/appServerConfig.ts b/cli/src/codex/utils/appServerConfig.ts index 69781063..3d639354 100644 --- a/cli/src/codex/utils/appServerConfig.ts +++ b/cli/src/codex/utils/appServerConfig.ts @@ -246,6 +246,7 @@ export function buildTurnStartParams(args: { cliOverrides?: CodexCliOverrides; baseInstructions?: string; developerInstructions?: string; + clientUserMessageId?: string; skills?: readonly SkillMetadata[]; overrides?: { approvalPolicy?: TurnStartParams['approvalPolicy']; @@ -260,6 +261,10 @@ export function buildTurnStartParams(args: { input: buildUserInputFromMessage(args.message, args.skills) }; + if (args.clientUserMessageId) { + params.clientUserMessageId = args.clientUserMessageId; + } + const allowCliOverrides = args.mode?.permissionMode === 'default'; const cliOverrides = allowCliOverrides ? args.cliOverrides : undefined; const approvalPolicy = args.overrides?.approvalPolicy diff --git a/cli/src/commands/agentCommandOptions.ts b/cli/src/commands/agentCommandOptions.ts index 549aaaae..561749d8 100644 --- a/cli/src/commands/agentCommandOptions.ts +++ b/cli/src/commands/agentCommandOptions.ts @@ -50,6 +50,12 @@ export function parseRemoteAgentCommandOptions) const sessionId = args[++i] diff --git a/cli/src/commands/claude.ts b/cli/src/commands/claude.ts index 4b959383..01401de2 100644 --- a/cli/src/commands/claude.ts +++ b/cli/src/commands/claude.ts @@ -72,6 +72,12 @@ export const claudeCommand: CommandDefinition = { unknownArgs.push('--effort', effort) } else if (arg === '--started-by') { options.startedBy = args[++i] as 'runner' | 'terminal' + } else if (arg === '--existing-session-id') { + const sessionId = args[++i] + if (!sessionId) { + throw new Error('Missing --existing-session-id value') + } + options.existingSessionId = sessionId } else { unknownArgs.push(arg) if (i + 1 < args.length && !args[i + 1].startsWith('-')) { diff --git a/cli/src/grok/conversationHistory.test.ts b/cli/src/grok/conversationHistory.test.ts new file mode 100644 index 00000000..c04bc7c7 --- /dev/null +++ b/cli/src/grok/conversationHistory.test.ts @@ -0,0 +1,80 @@ +import { describe, expect, it, vi } from 'vitest' +import { GrokConversationHistory } from './conversationHistory' + +describe('GrokConversationHistory', () => { + it('probes fork independently from rewind support', async () => { + const send = vi.fn(async (method: string) => { + if (method === '_x.ai/session/fork') throw new Error('Method not found: -32601') + return { points: [] } + }) + const history = new GrokConversationHistory(() => ({ sendExtensionRequest: send }) as never) + history.setSession('sess-1', '/tmp/proj') + await history.probeCapabilities() + expect(history.getCapabilitiesForMetadata()?.conversationHistory).toEqual({ + rewindToMessage: true + }) + }) + + it('current fork omits targetPromptIndex', async () => { + const send = vi.fn(async (method: string, params: Record) => { + expect(method).toBe('_x.ai/session/fork') + expect(params.targetPromptIndex).toBeUndefined() + return { newSessionId: 'grok-fork-1' } + }) + const history = new GrokConversationHistory(() => ({ sendExtensionRequest: send }) as never) + history.setSession('sess-1', '/tmp/proj') + const result = await history.fork() + expect(result).toEqual({ nativeSessionId: 'grok-fork-1' }) + }) + + it('historical fork passes targetPromptIndex from persisted mapping', async () => { + const send = vi.fn(async (_method: string, params: Record) => { + expect(params.targetPromptIndex).toBe(2) + return { newSessionId: 'grok-fork-2' } + }) + const history = new GrokConversationHistory(() => ({ sendExtensionRequest: send }) as never) + history.setSession('sess-1', '/tmp/proj') + history.rememberPromptIndex('local-x', 2) + await history.fork('local-x') + expect(send).toHaveBeenCalled() + }) + + it('restores prompt indexes from durable metadata', async () => { + const send = vi.fn(async (_method: string, params: Record) => { + expect(params.targetPromptIndex).toBe(4) + return { newSessionId: 'grok-fork-restored' } + }) + const history = new GrokConversationHistory(() => ({ sendExtensionRequest: send }) as never) + history.setSession('sess-1', '/tmp/proj') + history.restorePromptIndexes({ 'local-restored': 4 }) + await history.fork('local-restored') + expect(history.getHistoryIndexes()).toEqual({ 'local-restored': 4 }) + expect(history.getHistoryPoints()).toEqual({ 'local-restored': true }) + }) + + it('rewind always uses conversation_only and never all/files_only', async () => { + const send = vi.fn(async (method: string, params: Record) => { + expect(method).toBe('_x.ai/rewind/execute') + expect(params.mode).toBe('conversation_only') + expect(params.force).toBe(false) + return { success: true } + }) + const history = new GrokConversationHistory(() => ({ sendExtensionRequest: send }) as never) + history.setSession('sess-1', '/tmp/proj') + history.rememberPromptIndex('local-y', 1) + const result = await history.rewind('local-y') + expect(result.success).toBe(true) + expect(result.truncateFromLocalId).toBe('local-y') + }) + + it('marks capability unsupported on method-not-found', async () => { + const send = vi.fn(async () => { + throw new Error('Method not found: -32601') + }) + const history = new GrokConversationHistory(() => ({ sendExtensionRequest: send }) as never) + history.setSession('sess-1', '/tmp/proj') + history.rememberPromptIndex('local-z', 0) + await expect(history.rewind('local-z')).rejects.toThrow(/Method not found/) + expect(history.getCapabilitiesForMetadata()?.conversationHistory?.rewindToMessage).toBeUndefined() + }) +}) diff --git a/cli/src/grok/conversationHistory.ts b/cli/src/grok/conversationHistory.ts new file mode 100644 index 00000000..1f4de100 --- /dev/null +++ b/cli/src/grok/conversationHistory.ts @@ -0,0 +1,224 @@ +import type { AcpSdkBackend } from '@/agent/backends/acp/AcpSdkBackend' +import type { Metadata } from '@/api/types' +import { + GROK_CONVERSATION_HISTORY_INITIAL, + markSupported, + markUnsupported, + toConversationHistoryCapabilities, + type ConversationHistoryCapabilityStates +} from '@hapi/protocol/conversationHistory' +import type { + ForkConversationRpcResult, + RewindConversationRpcResult +} from '@hapi/protocol/apiTypes' + +function isMethodNotFound(error: unknown): boolean { + const message = error instanceof Error ? error.message : String(error) + return /method not found|-32601/i.test(message) +} + +function asString(value: unknown): string | null { + return typeof value === 'string' && value.length > 0 ? value : null +} + +function asRecord(value: unknown): Record | null { + return value !== null && typeof value === 'object' && !Array.isArray(value) + ? value as Record + : null +} + +export class GrokConversationHistory { + private states: ConversationHistoryCapabilityStates = { ...GROK_CONVERSATION_HISTORY_INITIAL } + private sessionId: string | null = null + private cwd: string | null = null + private readonly promptIndexByLocalId = new Map() + private busy = false + private publishCapabilities: (() => Promise) | null = null + + constructor(private readonly getBackend: () => AcpSdkBackend | null) {} + + setPublishCapabilities(fn: () => Promise): void { + this.publishCapabilities = fn + } + + setBusy(busy: boolean): void { + this.busy = busy + } + + setSession(sessionId: string | null, cwd: string | null): void { + this.sessionId = sessionId + this.cwd = cwd + } + + rememberPromptIndex(localId: string | undefined, promptIndex: number | null | undefined): void { + if (!localId || promptIndex == null || !Number.isInteger(promptIndex) || promptIndex < 0) return + this.promptIndexByLocalId.set(localId, promptIndex) + } + + getCapabilitiesForMetadata(): Metadata['capabilities'] { + const conversationHistory = toConversationHistoryCapabilities(this.states) + return conversationHistory ? { conversationHistory } : undefined + } + + getHistoryPoints(): Record { + const points: Record = {} + for (const localId of this.promptIndexByLocalId.keys()) { + points[localId] = true + } + return points + } + + getHistoryIndexes(): Record { + const indexes: Record = {} + for (const [localId, promptIndex] of this.promptIndexByLocalId.entries()) { + indexes[localId] = promptIndex + } + return indexes + } + + restorePromptIndexes(indexes: Record | null | undefined): void { + if (!indexes) return + for (const [localId, promptIndex] of Object.entries(indexes)) { + if (typeof localId !== 'string' || localId.length === 0) continue + if (!Number.isInteger(promptIndex) || promptIndex < 0) continue + this.promptIndexByLocalId.set(localId, promptIndex) + } + } + + async probeCapabilities(): Promise { + const backend = this.getBackend() + const sessionId = this.sessionId + if (!backend || !sessionId) return + + if (this.states.rewindToMessage === 'unknown' || this.states.forkAtMessage === 'unknown') { + try { + await backend.sendExtensionRequest('_x.ai/rewind/points', { sessionId }) + this.states = markSupported(this.states, 'rewindToMessage') + } catch (error) { + if (isMethodNotFound(error)) { + this.states = markUnsupported(this.states, 'rewindToMessage') + // Fork may still work independently — probe separately below + } + } + } + + if (this.states.forkCurrent === 'unknown' || this.states.forkAtMessage === 'unknown') { + try { + await backend.sendExtensionRequest('_x.ai/session/fork', { + sourceSessionId: '__hapi_capability_probe__', + sourceCwd: this.cwd ?? '', + newCwd: this.cwd ?? '' + }) + this.states = markSupported(this.states, 'forkCurrent') + this.states = markSupported(this.states, 'forkAtMessage') + } catch (error) { + if (isMethodNotFound(error)) { + this.states = markUnsupported(this.states, 'forkCurrent') + this.states = markUnsupported(this.states, 'forkAtMessage') + } else { + this.states = markSupported(this.states, 'forkCurrent') + this.states = markSupported(this.states, 'forkAtMessage') + } + } + } + + await this.publishCapabilities?.() + } + + async fork(messageLocalId?: string): Promise { + if (this.busy) throw new Error('Session is busy') + const backend = this.getBackend() + const sessionId = this.sessionId + const cwd = this.cwd + if (!backend || !sessionId || !cwd) throw new Error('Grok session is not ready') + + const params: Record = { + sourceSessionId: sessionId, + sourceCwd: cwd, + newCwd: cwd + } + if (messageLocalId) { + if (this.states.forkAtMessage === 'unsupported') { + throw new Error('Historical fork is not supported') + } + const targetPromptIndex = this.promptIndexByLocalId.get(messageLocalId) + if (targetPromptIndex == null) { + throw new Error(`No native history point for message ${messageLocalId}`) + } + params.targetPromptIndex = targetPromptIndex + } else if (this.states.forkCurrent === 'unsupported') { + throw new Error('Fork current is not supported') + } + + try { + const response = await backend.sendExtensionRequest>( + '_x.ai/session/fork', + params + ) + const nativeSessionId = asString(response.newSessionId) + ?? asString(asRecord(response)?.sessionId) + ?? asString(response.sessionId) + if (!nativeSessionId) throw new Error('x.ai/session/fork did not return newSessionId') + this.states = markSupported(this.states, messageLocalId ? 'forkAtMessage' : 'forkCurrent') + if (!messageLocalId) this.states = markSupported(this.states, 'forkCurrent') + else { + this.states = markSupported(this.states, 'forkAtMessage') + this.states = markSupported(this.states, 'forkCurrent') + } + await this.publishCapabilities?.() + return { nativeSessionId } + } catch (error) { + if (isMethodNotFound(error)) { + if (messageLocalId) { + this.states = markUnsupported(this.states, 'forkAtMessage') + } else { + this.states = markUnsupported(this.states, 'forkCurrent') + } + await this.publishCapabilities?.() + } + throw error + } + } + + async rewind(messageLocalId: string): Promise { + if (this.busy) throw new Error('Session is busy') + const backend = this.getBackend() + const sessionId = this.sessionId + if (!backend || !sessionId) throw new Error('Grok session is not ready') + if (this.states.rewindToMessage === 'unsupported') { + throw new Error('Rewind is not supported') + } + const targetPromptIndex = this.promptIndexByLocalId.get(messageLocalId) + if (targetPromptIndex == null) { + throw new Error(`No native history point for message ${messageLocalId}`) + } + + try { + const response = await backend.sendExtensionRequest>( + '_x.ai/rewind/execute', + { + sessionId, + targetPromptIndex, + mode: 'conversation_only', + force: false + } + ) + if (response.success === false) { + throw new Error(asString(response.error) ?? 'Rewind point is no longer available') + } + this.states = markSupported(this.states, 'rewindToMessage') + await this.publishCapabilities?.() + return { + success: true, + truncateFromLocalId: messageLocalId, + messages: [] + } + } catch (error) { + if (isMethodNotFound(error)) { + this.states = markUnsupported(this.states, 'rewindToMessage') + await this.publishCapabilities?.() + } + throw error + } + } +} diff --git a/cli/src/grok/grokRemoteLauncher.test.ts b/cli/src/grok/grokRemoteLauncher.test.ts index 9fa8e979..851964a2 100644 --- a/cli/src/grok/grokRemoteLauncher.test.ts +++ b/cli/src/grok/grokRemoteLauncher.test.ts @@ -11,13 +11,25 @@ const harness = vi.hoisted(() => ({ sessionInfoUpdateHandler: null as null | ((update: { title?: string | null }) => void), nativeTitle: null as string | null, nativeTitleSent: false, + loadSessionCalls: [] as string[], + loadSessionError: null as Error | null, + newSessionCalls: 0, })) vi.mock('./utils/grokBackend', () => ({ createGrokBackend: vi.fn(() => ({ initialize: vi.fn(async () => {}), - newSession: vi.fn(async () => 'grok-session-1'), - loadSession: vi.fn(async () => 'grok-session-1'), + newSession: vi.fn(async () => { + harness.newSessionCalls += 1 + return 'grok-new-session' + }), + loadSession: vi.fn(async (params: { sessionId: string }) => { + harness.loadSessionCalls.push(params.sessionId) + if (harness.loadSessionError) { + throw harness.loadSessionError + } + return params.sessionId + }), setModel: vi.fn(async (sessionId: string, modelId: string, opts?: { flavor?: string }) => { harness.setModels.push({ sessionId, modelId, flavor: opts?.flavor }) }), @@ -90,6 +102,8 @@ function createSession() { rpcHandlerManager: { registerHandler(method: string, handler: () => unknown) { rpcHandlers.set(method, handler) } }, + updateMetadata: vi.fn(), + getMetadata: vi.fn(() => null), sendAgentMessage: vi.fn(), sendSessionEvent: vi.fn(), sendClaudeSessionMessage: vi.fn() @@ -134,6 +148,9 @@ describe('grokRemoteLauncher runtime config', () => { harness.nativeTitle = null harness.nativeTitleSent = false harness.autoCommandAvailable = true + harness.loadSessionCalls = [] + harness.loadSessionError = null + harness.newSessionCalls = 0 }) it('switches model and effort between turns and exposes session catalogs', async () => { @@ -148,10 +165,10 @@ describe('grokRemoteLauncher runtime config', () => { expect(discovered).toEqual([{ model: 'grok-a', effort: 'low' }]) expect(harness.setModels).toEqual([ - { sessionId: 'grok-session-1', modelId: 'grok-b', flavor: 'grok' } + { sessionId: 'grok-new-session', modelId: 'grok-b', flavor: 'grok' } ]) expect(harness.setModes).toEqual([ - { sessionId: 'grok-session-1', modeId: 'medium' } + { sessionId: 'grok-new-session', modeId: 'medium' } ]) expect(harness.prompts).toHaveLength(3) expect(session.sendSessionEvent).not.toHaveBeenCalledWith(expect.objectContaining({ @@ -167,6 +184,21 @@ describe('grokRemoteLauncher runtime config', () => { expect(await rpcHandlers.get('listGrokReasoningEffortOptions')?.()).toMatchObject({ success: true, currentValue: 'low' }) }) + it('does not fall back to newSession when a fork child cannot load its native id', async () => { + const { session } = createSession() + session.sessionId = 'grok-forked-native' + vi.mocked(session.client.getMetadata).mockReturnValue({ forkedFrom: 'parent-session' } as never) + harness.loadSessionError = new Error('session/load rejected') + + await expect(grokRemoteLauncher(session as never, { + model: 'grok-a', + effort: 'low' + })).rejects.toThrow(/session\/load rejected/) + + expect(harness.loadSessionCalls).toEqual(['grok-forked-native']) + expect(harness.newSessionCalls).toBe(0) + }) + it('uses Grok slash commands to enter and leave Auto permission mode without model turns', async () => { const { session } = createPermissionSession(['auto', 'default']) diff --git a/cli/src/grok/grokRemoteLauncher.ts b/cli/src/grok/grokRemoteLauncher.ts index 6ac0de4d..54acb67c 100644 --- a/cli/src/grok/grokRemoteLauncher.ts +++ b/cli/src/grok/grokRemoteLauncher.ts @@ -20,6 +20,8 @@ import { import { GrokPermissionHandler } from './utils/permissionHandler' import { RPC_METHODS } from '@hapi/protocol/rpcMethods' import { GROK_TITLE_INSTRUCTION } from './utils/systemPrompt' +import { GrokConversationHistory } from './conversationHistory' +import { isObject } from '@hapi/protocol' const PLAN_MODE_INSTRUCTION = 'Work in plan-only mode. Analyze and propose a plan, but do not execute commands or modify files.' @@ -46,6 +48,7 @@ class GrokRemoteLauncher extends RemoteLauncherBase { private defaultBackendEffort: string | null = null private currentBackendPermissionMode: 'default' | 'auto' | null = null private instructionsSent = false + private readonly conversationHistory = new GrokConversationHistory(() => this.backend) constructor( private readonly session: GrokSession, @@ -102,6 +105,9 @@ class GrokRemoteLauncher extends RemoteLauncherBase { await backend.initialize() const acpMcpServers = toAcpMcpServers(mcpServers) + // Fork children must load the exact native id hub forked. Falling back to + // newSession() would leave hydrated HAPI history without matching model context. + const strictForkResume = session.client.getMetadata()?.forkedFrom != null let acpSessionId: string try { if (session.sessionId) { @@ -112,6 +118,9 @@ class GrokRemoteLauncher extends RemoteLauncherBase { mcpServers: acpMcpServers }) } catch (error) { + if (strictForkResume) { + throw error + } logger.warn('[grok-remote] resume failed, starting new session', error) session.sendSessionEvent({ type: 'message', @@ -135,6 +144,53 @@ class GrokRemoteLauncher extends RemoteLauncherBase { } session.registerExistingNativeSession(acpSessionId) + this.conversationHistory.setSession(acpSessionId, session.path) + this.conversationHistory.setPublishCapabilities(async () => { + const conversationHistory = this.conversationHistory.getCapabilitiesForMetadata()?.conversationHistory + try { + session.client.updateMetadata((metadata) => { + const capabilities = { ...metadata?.capabilities } + delete capabilities.conversationHistory + if (conversationHistory) { + capabilities.conversationHistory = conversationHistory + } + return { + ...metadata, + path: metadata?.path ?? session.path, + host: metadata?.host ?? 'unknown', + capabilities, + conversationHistoryPoints: { + ...metadata?.conversationHistoryPoints, + ...this.conversationHistory.getHistoryPoints() + }, + conversationHistoryIndexes: { + ...metadata?.conversationHistoryIndexes, + ...this.conversationHistory.getHistoryIndexes() + } + } + }) + } catch { + // best-effort; tests and transient hub disconnects must not crash the loop + } + }) + this.conversationHistory.restorePromptIndexes( + typeof session.client.getMetadata === 'function' + ? session.client.getMetadata()?.conversationHistoryIndexes + : undefined + ) + session.client.rpcHandlerManager.registerHandler(RPC_METHODS.ForkConversation, async (payload: unknown) => { + const messageLocalId = isObject(payload) && typeof payload.messageLocalId === 'string' + ? payload.messageLocalId + : undefined + return await this.conversationHistory.fork(messageLocalId) + }) + session.client.rpcHandlerManager.registerHandler(RPC_METHODS.RewindConversation, async (payload: unknown) => { + if (!isObject(payload) || typeof payload.messageLocalId !== 'string') { + throw new Error('messageLocalId is required') + } + return await this.conversationHistory.rewind(payload.messageLocalId) + }) + void this.conversationHistory.probeCapabilities().catch(() => {}) const modelMetadata = backend.getSessionModelsMetadata(acpSessionId) const effortMetadata = backend.getThoughtLevelConfigOption(acpSessionId) this.currentBackendModel = modelMetadata?.currentModelId ?? this.opts.model ?? null @@ -186,6 +242,12 @@ class GrokRemoteLauncher extends RemoteLauncherBase { break } + // collectBatch already emitted messages-consumed; hub idle checks + // see an empty queue. Hold the history busy flag across the whole + // turn setup (model/permission sync, rewind-points, prompt). + this.conversationHistory.setBusy(true) + session.onThinkingChange(true) + try { const requestedModel = batch.mode.model === null ? this.defaultBackendModel : batch.mode.model @@ -248,18 +310,54 @@ class GrokRemoteLauncher extends RemoteLauncherBase { this.instructionsSent = true } const promptContent: PromptContent[] = [{ type: 'text', text }] + const localId = batch.items + ?.map((item) => item.localId) + .find((id): id is string => typeof id === 'string' && id.length > 0) + + // Official prompt index: count rewind points before the prompt; the new + // point lands at that index after a successful turn. + let nextPromptIndex: number | null = null + try { + const points = await backend.sendExtensionRequest<{ points?: unknown[] } | unknown[]>( + '_x.ai/rewind/points', + { sessionId: acpSessionId } + ) + const list = Array.isArray(points) + ? points + : (isObject(points) && Array.isArray(points.points) ? points.points : null) + if (list) nextPromptIndex = list.length + } catch { + nextPromptIndex = null + } - session.onThinkingChange(true) try { await backend.prompt(acpSessionId, promptContent, (message: AgentMessage) => { this.handleAgentMessage(message) }) + if (localId && nextPromptIndex != null) { + this.conversationHistory.rememberPromptIndex(localId, nextPromptIndex) + session.client.updateMetadata((metadata) => ({ + ...metadata, + path: metadata?.path ?? session.path, + host: metadata?.host ?? 'unknown', + conversationHistoryPoints: { + ...metadata?.conversationHistoryPoints, + [localId]: true as const + }, + conversationHistoryIndexes: { + ...metadata?.conversationHistoryIndexes, + [localId]: nextPromptIndex + } + })) + } } catch (error) { const message = formatGrokError(error) logger.warn('[grok-remote] prompt failed', error) session.sendSessionEvent({ type: 'message', message: `Grok prompt failed: ${message}` }) this.messageBuffer.addMessage(`Grok prompt failed: ${message}`, 'status') + } } finally { + this.conversationHistory.setBusy(false) session.onThinkingChange(false) await this.permissionHandler?.cancelAll('Prompt finished') if (session.queue.size() === 0 && !this.shouldExit) { diff --git a/cli/src/modules/common/rpcTypes.ts b/cli/src/modules/common/rpcTypes.ts index fdc8a12c..e1a6e219 100644 --- a/cli/src/modules/common/rpcTypes.ts +++ b/cli/src/modules/common/rpcTypes.ts @@ -18,6 +18,8 @@ export interface SpawnSessionOptions { token?: string sessionType?: 'simple' | 'worktree' worktreeName?: string + /** Claude: spawn with --fork-session after --resume. */ + forkSession?: boolean } export type SpawnSessionResult = diff --git a/cli/src/runner/buildCliArgs.test.ts b/cli/src/runner/buildCliArgs.test.ts index 3cffb258..8496da01 100644 --- a/cli/src/runner/buildCliArgs.test.ts +++ b/cli/src/runner/buildCliArgs.test.ts @@ -242,6 +242,21 @@ describe('buildCliArgs', () => { expect(args).toContain('some-claude-session-id') }) + it('passes --fork-session and --existing-session-id for Claude message-level fork', () => { + const args = buildCliArgs('claude', { + directory: '/tmp', + resumeSessionId: 'claude-source-id', + existingSessionId: 'hapi-child-id', + forkSession: true, + }) + expect(args).toContain('--resume') + expect(args).toContain('claude-source-id') + expect(args).toContain('--fork-session') + expect(args.indexOf('--fork-session')).toBeGreaterThan(args.indexOf('--resume')) + expect(args).toContain('--existing-session-id') + expect(args).toContain('hapi-child-id') + }) + it('passes --effort for pi agent', () => { const args = buildCliArgs('pi', { directory: '/tmp', diff --git a/cli/src/runner/run.ts b/cli/src/runner/run.ts index 2d19bdd1..f7c6ffc9 100644 --- a/cli/src/runner/run.ts +++ b/cli/src/runner/run.ts @@ -1336,16 +1336,27 @@ export function buildCliArgs( args.push('--resume', options.resumeSessionId); } } + // Message-level Fork current for Claude: must follow --resume. + if (options.forkSession && agentCommand === 'claude') { + args.push('--fork-session'); + } args.push('--hapi-starting-mode', 'remote', '--started-by', 'runner'); - // Codex, Cursor ACP, and Pi native resume reuse the original HAPI row via - // --existing-session-id. Pi is reported successful only after the hub sees - // its validated native get_state/session-ready signal. - if (agent === 'codex' || agent === 'cursor' || agent === 'pi') { + // Codex, Cursor ACP, Pi native resume, and Claude message-level forks + // reuse the original HAPI row via --existing-session-id. + if (agent === 'codex' || agent === 'cursor' || agent === 'pi' + || (agentCommand === 'claude' && options.forkSession)) { const existingSessionId = options.existingSessionId ?? options.sessionId; if (existingSessionId) { args.push('--existing-session-id', existingSessionId); } } + // Grok fork children also bind the pending HAPI session id. + if (agent === 'grok') { + const existingSessionId = options.existingSessionId ?? options.sessionId; + if (existingSessionId && !args.includes('--existing-session-id')) { + args.push('--existing-session-id', existingSessionId); + } + } if (options.model) { args.push('--model', options.model); } diff --git a/hub/src/store/messageStore.ts b/hub/src/store/messageStore.ts index cc1c5f38..643d6d1c 100644 --- a/hub/src/store/messageStore.ts +++ b/hub/src/store/messageStore.ts @@ -25,7 +25,9 @@ import { markMessagesInvoked, mergeSessionMessages, copyMessageToSession as copyStoredMessageToSession, + copyMessagesToSession as copyStoredMessagesToSession, getAllMessages, + truncateMessagesFromLocalId, type CancelQueuedMessageResult, type LookupQueuedMessageResult, type LocalMessageState, @@ -51,6 +53,13 @@ export class MessageStore { return copyStoredMessageToSession(this.db, sessionId, message) } + copyMessagesToSession( + sessionId: string, + messages: Array> + ): number { + return copyStoredMessagesToSession(this.db, sessionId, messages) + } + getAllMessages(sessionId: string): StoredMessage[] { return getAllMessages(this.db, sessionId) } @@ -143,4 +152,17 @@ export class MessageStore { mergeSessionMessages(fromSessionId: string, toSessionId: string): { moved: number; oldMaxSeq: number; newMaxSeq: number } { return mergeSessionMessages(this.db, fromSessionId, toSessionId) } + + truncateMessagesFromLocalId( + sessionId: string, + localId: string, + replacement: Array<{ + content: unknown + localId?: string | null + createdAt?: number + invokedAt?: number | null + }> = [] + ): { deleted: number; inserted: number; epoch: number } { + return truncateMessagesFromLocalId(this.db, sessionId, localId, replacement) + } } diff --git a/hub/src/store/messages.copyBatch.test.ts b/hub/src/store/messages.copyBatch.test.ts new file mode 100644 index 00000000..d2dddbd2 --- /dev/null +++ b/hub/src/store/messages.copyBatch.test.ts @@ -0,0 +1,33 @@ +import { describe, expect, it } from 'bun:test' +import { Store } from './index' + +describe('copyMessagesToSession', () => { + it('copies a prefix in one transaction and bumps epoch once', () => { + const store = new Store(':memory:') + const source = store.sessions.getOrCreateSession('src', { path: '/tmp' }, null, 'default') + const child = store.sessions.getOrCreateSession('child', { path: '/tmp' }, null, 'default') + + store.messages.addMessage(source.id, { role: 'user', content: { type: 'text', text: 'one' } }, 'local-1') + store.messages.markMessagesInvoked(source.id, ['local-1'], Date.now()) + store.messages.addMessage(source.id, { role: 'agent', content: { type: 'text', text: 'a1' } }) + store.messages.addMessage(source.id, { role: 'user', content: { type: 'text', text: 'two' } }, 'local-2') + store.messages.markMessagesInvoked(source.id, ['local-2'], Date.now()) + + const beforeEpoch = store.messages.getMessageEpoch(child.id) + const prefix = store.messages.getAllMessages(source.id).slice(0, 2) + const copied = store.messages.copyMessagesToSession( + child.id, + prefix.map((message) => ({ + content: message.content, + createdAt: message.createdAt, + localId: message.localId, + invokedAt: message.invokedAt, + scheduledAt: message.scheduledAt + })) + ) + + expect(copied).toBe(2) + expect(store.messages.getAllMessages(child.id)).toHaveLength(2) + expect(store.messages.getMessageEpoch(child.id)).toBe(beforeEpoch + 1) + }) +}) diff --git a/hub/src/store/messages.truncate.test.ts b/hub/src/store/messages.truncate.test.ts new file mode 100644 index 00000000..128d2cc1 --- /dev/null +++ b/hub/src/store/messages.truncate.test.ts @@ -0,0 +1,28 @@ +import { describe, expect, it } from 'bun:test' +import { Store } from './index' + +describe('truncateMessagesFromLocalId', () => { + it('deletes the target and later messages and bumps epoch', () => { + const store = new Store(':memory:') + const session = store.sessions.getOrCreateSession('t', { path: '/tmp' }, null, 'default') + + // Mirror real delivery order: invoke each user turn before its agent reply + // is written. Bulk-stamping invokedAt after the fact collapses timestamps and + // can leave later agent rows "before" the rewind boundary. + store.messages.addMessage(session.id, { role: 'user', content: { type: 'text', text: 'one' } }, 'local-1') + store.messages.markMessagesInvoked(session.id, ['local-1'], Date.now()) + store.messages.addMessage(session.id, { role: 'agent', content: { type: 'text', text: 'a1' } }) + store.messages.addMessage(session.id, { role: 'user', content: { type: 'text', text: 'two' } }, 'local-2') + store.messages.markMessagesInvoked(session.id, ['local-2'], Date.now()) + store.messages.addMessage(session.id, { role: 'agent', content: { type: 'text', text: 'a2' } }) + + const beforeEpoch = store.messages.getMessageEpoch(session.id) + const result = store.messages.truncateMessagesFromLocalId(session.id, 'local-2', []) + expect(result.deleted).toBeGreaterThanOrEqual(2) + expect(result.epoch).toBeGreaterThan(beforeEpoch) + + const remaining = store.messages.getAllMessages(session.id) + expect(remaining.some((message) => message.localId === 'local-2')).toBe(false) + expect(remaining.some((message) => message.localId === 'local-1')).toBe(true) + }) +}) diff --git a/hub/src/store/messages.ts b/hub/src/store/messages.ts index 3195883c..14981a5c 100644 --- a/hub/src/store/messages.ts +++ b/hub/src/store/messages.ts @@ -160,6 +160,61 @@ export function copyMessageToSession( return toStoredMessage(row) } +/** + * Batch-hydrate a fork child: one transaction, contiguous seq allocation, + * single epoch bump. Avoids O(n) max-seq lookups and epoch writes. + */ +export function copyMessagesToSession( + db: Database, + sessionId: string, + messages: CopyStoredMessageInput[] +): number { + if (messages.length === 0) return 0 + + return db.transaction(() => { + let nextSeq = getMaxSeq(db, sessionId) + 1 + const insert = db.prepare(` + INSERT INTO messages ( + id, session_id, content, created_at, seq, local_id, invoked_at, scheduled_at + ) VALUES ( + @id, @session_id, @content, @created_at, @seq, @local_id, @invoked_at, @scheduled_at + ) + `) + const collisionCheck = db.prepare( + 'SELECT 1 FROM messages WHERE session_id = ? AND local_id = ? LIMIT 1' + ) + + for (const message of messages) { + const createdAt = Number.isFinite(message.createdAt) ? message.createdAt : Date.now() + let localId = message.localId + if (localId) { + const collision = collisionCheck.get(sessionId, localId) as { 1: number } | undefined + if (collision) { + localId = `${localId}:merged:${randomUUID().slice(0, 8)}` + } + } + if (message.scheduledAt != null && !localId && message.invokedAt === null) { + localId = `merged-scheduled:${randomUUID()}` + } + const invokedAt = localId ? message.invokedAt : (message.invokedAt ?? createdAt) + insert.run({ + id: randomUUID(), + session_id: sessionId, + content: encodeMessageContent(message.content), + created_at: createdAt, + seq: nextSeq, + local_id: localId ?? null, + invoked_at: invokedAt ?? null, + scheduled_at: message.scheduledAt ?? null + }) + nextSeq += 1 + } + + bumpMessageEpoch(db, sessionId) + return messages.length + })() +} + export function getMessages( db: Database, sessionId: string, @@ -704,3 +759,70 @@ export function mergeSessionMessages( throw error } } + +/** + * Truncate transcript at/after the message with `localId`, optionally replacing + * the removed suffix with `replacement` messages. Bumps message epoch so web + * clients reset their window. + */ +export function truncateMessagesFromLocalId( + db: Database, + sessionId: string, + localId: string, + replacement: Array<{ + content: unknown + localId?: string | null + createdAt?: number + invokedAt?: number | null + }> = [] +): { deleted: number; inserted: number; epoch: number } { + return db.transaction(() => { + const target = db.prepare(` + SELECT id, seq, COALESCE(invoked_at, created_at) AS position_at + FROM messages + WHERE session_id = ? AND local_id = ? + LIMIT 1 + `).get(sessionId, localId) as { id: string; seq: number; position_at: number } | undefined + + if (!target) { + throw new Error(`Message not found for localId: ${localId}`) + } + + const deleted = db.prepare(` + DELETE FROM messages + WHERE session_id = ? + AND ( + COALESCE(invoked_at, created_at) > ? + OR (COALESCE(invoked_at, created_at) = ? AND seq >= ?) + ) + `).run(sessionId, target.position_at, target.position_at, target.seq) + + let inserted = 0 + for (const message of replacement) { + const now = Date.now() + const msgSeqRow = db.prepare( + 'SELECT COALESCE(MAX(seq), 0) + 1 AS nextSeq FROM messages WHERE session_id = ?' + ).get(sessionId) as { nextSeq: number } + const id = randomUUID() + const createdAt = message.createdAt ?? now + const invokedAt = message.invokedAt === undefined ? createdAt : message.invokedAt + const rowLocalId = message.localId ?? null + db.prepare(` + INSERT INTO messages (id, session_id, content, created_at, seq, local_id, invoked_at, scheduled_at) + VALUES (?, ?, ?, ?, ?, ?, ?, NULL) + `).run( + id, + sessionId, + encodeMessageContent(message.content), + createdAt, + msgSeqRow.nextSeq, + rowLocalId, + invokedAt + ) + inserted += 1 + } + + const epoch = bumpMessageEpoch(db, sessionId) + return { deleted: deleted.changes, inserted, epoch } + })() +} diff --git a/hub/src/store/sessionStore.ts b/hub/src/store/sessionStore.ts index 1b8c2ca0..d8857605 100644 --- a/hub/src/store/sessionStore.ts +++ b/hub/src/store/sessionStore.ts @@ -15,6 +15,7 @@ import { setSessionActive, setSessionTeamState, setSessionTodos, + replaceSessionTodos, touchSessionUpdatedAt, updateSessionAgentState, updateSessionMetadata @@ -63,6 +64,15 @@ export class SessionStore { return setSessionTodos(this.db, id, todos, todosUpdatedAt, namespace) } + replaceSessionTodos( + id: string, + todos: unknown, + todosUpdatedAt: number | null, + namespace: string + ): boolean { + return replaceSessionTodos(this.db, id, todos, todosUpdatedAt, namespace) + } + setSessionTeamState(id: string, teamState: unknown, updatedAt: number, namespace: string): boolean { return setSessionTeamState(this.db, id, teamState, updatedAt, namespace) } diff --git a/hub/src/store/sessions.ts b/hub/src/store/sessions.ts index 9fae4f1f..c6bd11ad 100644 --- a/hub/src/store/sessions.ts +++ b/hub/src/store/sessions.ts @@ -376,6 +376,38 @@ export function setSessionTodos( } } +/** Force-replace todos after rewind/fork (ignores monotonic timestamp guard). */ +export function replaceSessionTodos( + db: Database, + id: string, + todos: unknown, + todosUpdatedAt: number | null, + namespace: string +): boolean { + try { + const json = todos === null || todos === undefined ? null : JSON.stringify(todos) + const now = Date.now() + const result = db.prepare(` + UPDATE sessions + SET todos = @todos, + todos_updated_at = @todos_updated_at, + updated_at = CASE WHEN updated_at > @updated_at THEN updated_at ELSE @updated_at END, + seq = seq + 1 + WHERE id = @id + AND namespace = @namespace + `).run({ + id, + todos: json, + todos_updated_at: todosUpdatedAt, + updated_at: now, + namespace + }) + return result.changes === 1 + } catch { + return false + } +} + export function setSessionTeamState( db: Database, id: string, diff --git a/hub/src/sync/forkTranscript.test.ts b/hub/src/sync/forkTranscript.test.ts new file mode 100644 index 00000000..2bffd9be --- /dev/null +++ b/hub/src/sync/forkTranscript.test.ts @@ -0,0 +1,47 @@ +import { describe, expect, it } from 'bun:test' +import { selectForkTranscriptPrefix } from './forkTranscript' + +describe('selectForkTranscriptPrefix', () => { + const messages = [ + { localId: 'a', text: '1', invokedAt: 1, createdAt: 1, seq: 1 }, + { localId: null, text: '2', invokedAt: 2, createdAt: 2, seq: 2 }, + { localId: 'b', text: '3', invokedAt: 3, createdAt: 3, seq: 3 }, + { localId: null, text: '4', invokedAt: 4, createdAt: 4, seq: 4 }, + { localId: 'pending', text: 'scheduled', invokedAt: null, createdAt: 5, seq: 5 } + ] + + it('copies the full invoked transcript for current fork', () => { + expect(selectForkTranscriptPrefix(messages).map((message) => message.text)).toEqual([ + '1', '2', '3', '4' + ]) + }) + + it('excludes the boundary message and later turns for historical fork', () => { + expect(selectForkTranscriptPrefix(messages, 'b').map((message) => message.text)).toEqual([ + '1', '2' + ]) + }) + + it('never copies pending scheduled rows', () => { + expect(selectForkTranscriptPrefix(messages).some((message) => message.localId === 'pending')).toBe(false) + expect(selectForkTranscriptPrefix(messages, 'pending').map((message) => message.text)).toEqual([ + '1', '2', '3', '4' + ]) + }) + + it('orders by invocation time before slicing, not insertion seq', () => { + const queuedThenAnswered = [ + { localId: 'user-b', text: 'B', invokedAt: 30, createdAt: 10, seq: 1 }, + { localId: null, text: 'A-reply', invokedAt: 20, createdAt: 20, seq: 2 } + ] + expect(selectForkTranscriptPrefix(queuedThenAnswered, 'user-b').map((message) => message.text)).toEqual([ + 'A-reply' + ]) + }) + + it('throws when the boundary localId is missing', () => { + expect(() => selectForkTranscriptPrefix(messages, 'missing')).toThrow( + 'Fork boundary message not found' + ) + }) +}) diff --git a/hub/src/sync/forkTranscript.ts b/hub/src/sync/forkTranscript.ts new file mode 100644 index 00000000..2ac6d75e --- /dev/null +++ b/hub/src/sync/forkTranscript.ts @@ -0,0 +1,36 @@ +/** + * Select the HAPI transcript prefix to hydrate into a forked child session. + * Historical fork excludes the boundary message and everything after it. + * Current fork copies the full source transcript. + * Pending scheduled/queued rows (`invokedAt == null`) are never copied — they + * are not part of the native history being forked and would otherwise fire on + * both the source and the child. + * + * Messages are ordered by invocation/display time (then seq) before slicing so + * a late-seq agent reply that appeared before a queued user turn is retained. + */ +export function selectForkTranscriptPrefix( + messages: T[], + messageLocalId?: string +): T[] { + const ordered = messages.slice().sort((a, b) => { + const byTime = (a.invokedAt ?? a.createdAt) - (b.invokedAt ?? b.createdAt) + return byTime !== 0 ? byTime : a.seq - b.seq + }) + let scoped: T[] + if (!messageLocalId) { + scoped = ordered + } else { + const cutoff = ordered.findIndex((message) => message.localId === messageLocalId) + if (cutoff < 0) { + throw new Error('Fork boundary message not found') + } + scoped = ordered.slice(0, cutoff) + } + return scoped.filter((message) => message.invokedAt != null) +} diff --git a/hub/src/sync/messageService.ts b/hub/src/sync/messageService.ts index 557de4ff..911021c2 100644 --- a/hub/src/sync/messageService.ts +++ b/hub/src/sync/messageService.ts @@ -699,10 +699,13 @@ export class MessageService { * preserved). Web client surfaces this as 'sent' in the thread. * See messageService.test.ts "cancel × mature race" for the documented * expected behaviour. */ - releaseMatureScheduledMessages(now: number): void { + releaseMatureScheduledMessages(now: number, skipSessionIds?: ReadonlySet): void { const mature = this.store.messages.getMatureScheduledMessages(now) const maturedSessionIds = new Set() for (const msg of mature) { + if (skipSessionIds?.has(msg.sessionId)) { + continue + } const localId = msg.localId if (typeof localId === 'string' && !this.scheduledMatureNotifiedLocalIds.has(localId)) { this.scheduledMatureNotifiedLocalIds.add(localId) diff --git a/hub/src/sync/rpcGateway.ts b/hub/src/sync/rpcGateway.ts index 7cd031b0..69647e0b 100644 --- a/hub/src/sync/rpcGateway.ts +++ b/hub/src/sync/rpcGateway.ts @@ -165,7 +165,8 @@ export class RpcGateway { permissionMode?: PermissionMode, serviceTier?: string, existingSessionId?: string, - collaborationMode?: CodexCollaborationMode + collaborationMode?: CodexCollaborationMode, + forkSession?: boolean ): Promise<{ type: 'success'; sessionId: string } | { type: 'error'; message: string }> { try { const result = await this.machineRpc( @@ -186,7 +187,8 @@ export class RpcGateway { serviceTier, existingSessionId, sessionId: existingSessionId, - collaborationMode + collaborationMode, + forkSession: forkSession === true } ) if (result && typeof result === 'object') { @@ -366,6 +368,30 @@ export class RpcGateway { return await this.sessionRpc(sessionId, method, params ?? {}, timeoutMs ?? DEFAULT_RPC_TIMEOUT_MS) as T } + async forkConversation( + sessionId: string, + params: { messageLocalId?: string } + ): Promise { + return await this.sessionRpc( + sessionId, + RPC_METHODS.ForkConversation, + params, + 120_000 + ) as import('@hapi/protocol/apiTypes').ForkConversationRpcResult + } + + async rewindConversation( + sessionId: string, + params: { messageLocalId: string } + ): Promise { + return await this.sessionRpc( + sessionId, + RPC_METHODS.RewindConversation, + params, + 120_000 + ) as import('@hapi/protocol/apiTypes').RewindConversationRpcResult + } + async listOpencodeReasoningEffortOptionsForSession(sessionId: string): Promise { return await this.sessionRpc(sessionId, RPC_METHODS.ListOpencodeReasoningEffortOptions, {}) as RpcListOpencodeReasoningEffortOptionsResponse } diff --git a/hub/src/sync/sessionCache.ts b/hub/src/sync/sessionCache.ts index 5f372c1c..23e3c2b5 100644 --- a/hub/src/sync/sessionCache.ts +++ b/hub/src/sync/sessionCache.ts @@ -91,6 +91,37 @@ export class SessionCache { return this.refreshSession(stored.id) ?? (() => { throw new Error('Failed to load session') })() } + /** + * After fork hydrate / rewind truncate, re-scan the transcript for the + * latest TodoWrite (or clear todos). Bypasses the one-shot backfill flag + * and the monotonic todosUpdatedAt guard. + */ + rebuildTodosFromTranscript(sessionId: string): void { + const stored = this.store.sessions.getSession(sessionId) + if (!stored) return + + this.todoBackfillAttemptedSessionIds.delete(sessionId) + const messages = this.store.messages.getAllMessages(sessionId) + let found: { todos: unknown; at: number } | null = null + for (let i = messages.length - 1; i >= 0; i -= 1) { + const message = messages[i] + if (!message) continue + const todos = extractTodoWriteTodosFromMessageContent(message.content) + if (todos) { + found = { todos, at: message.createdAt } + break + } + } + this.store.sessions.replaceSessionTodos( + sessionId, + found?.todos ?? null, + found?.at ?? null, + stored.namespace + ) + this.todoBackfillAttemptedSessionIds.add(sessionId) + this.refreshSession(sessionId) + } + refreshSession(sessionId: string): Session | null { let stored = this.store.sessions.getSession(sessionId) if (!stored) { diff --git a/hub/src/sync/syncEngine.ts b/hub/src/sync/syncEngine.ts index adf4cd17..5ec9e1ab 100644 --- a/hub/src/sync/syncEngine.ts +++ b/hub/src/sync/syncEngine.ts @@ -12,6 +12,7 @@ import type { CursorChatStoreStatus, CursorMigrateOutcome, CursorMigrateToAcpReq import type { AgentFlavor, CodexCollaborationMode, DecryptedMessage, PermissionMode, Session, SyncEvent } from '@hapi/protocol/types' import { unwrapRoleWrappedRecordEnvelope } from '@hapi/protocol/messages' import type { Server } from 'socket.io' +import { randomUUID } from 'node:crypto' import type { Store, CancelQueuedMessageResult } from '../store' import type { HapiSessionExportResult } from '@hapi/protocol/sessionExport' import type { RpcRegistry } from '../socket/rpcRegistry' @@ -21,6 +22,7 @@ import { CursorLegacyMigrator, type CursorLegacyMigratorOptions } from '../curso import { EventPublisher, type SyncEventListener } from './eventPublisher' import { MachineCache, type Machine } from './machineCache' import { MessageService } from './messageService' +import { selectForkTranscriptPrefix } from './forkTranscript' import { RpcGateway, RpcTargetMissingError, @@ -159,6 +161,8 @@ export class SyncEngine { private readonly piUnexpectedTempOriginalIds = new Map() /** Serialize scratchlist uploads per session so disk-byte caps cannot race. */ private readonly scratchlistUploadTails = new Map>() + /** Serialize fork/rewind per session so concurrent native rollbacks cannot stack. */ + private readonly historyActionsInFlight = new Set() constructor( private readonly store: Store, @@ -759,7 +763,7 @@ async uploadScratchlistAttachment( this.machineCache.expireInactive() // Piggybacked on the inactivity tick; not a logical part of expireInactive // but shares its 5s cadence (avoids a second timer). - this.messageService.releaseMatureScheduledMessages(Date.now()) + this.messageService.releaseMatureScheduledMessages(Date.now(), this.historyActionsInFlight) } private reloadAll(): void { @@ -810,6 +814,9 @@ async uploadScratchlistAttachment( scheduledAt?: number | null } ): Promise { + if (this.historyActionsInFlight.has(sessionId)) { + throw new Error('Conversation history action already in progress') + } await this.messageService.sendMessage(sessionId, payload) this.sessionCache.markMessageQueued(sessionId) this.sessionCache.recordSessionActivity(sessionId, Date.now()) @@ -849,6 +856,484 @@ async uploadScratchlistAttachment( await this.rpcGateway.abortSession(sessionId) } + private assertConversationHistoryIdle(session: Session): void { + if (!session.active) { + throw new Error('Session must be active') + } + if (session.agentState?.controlledByUser === true) { + throw new Error('Conversation history actions require a remote session') + } + if (session.thinking) { + throw new Error('Session is busy') + } + if (session.metadata?.conversationHistoryDiverged === true) { + throw new Error('Conversation history is diverged; refuse further fork/rewind') + } + const queued = this.store.messages.getUninvokedLocalMessages(session.id) + if (queued.length > 0) { + throw new Error('Session has queued messages') + } + } + + /** Stale localIds must fail before native fork/rewind mutates agent history. */ + private assertInvokedHistoryBoundary(sessionId: string, messageLocalId: string): void { + const boundary = this.store.messages.getAllMessages(sessionId).find( + (message) => message.localId === messageLocalId && message.invokedAt != null + ) + if (!boundary) { + throw new Error('History boundary message not found or not yet invoked') + } + } + + /** + * Claude `--fork-session` materializes only after the child process starts. + * Poll until the child metadata has a native id distinct from the source. + */ + private async waitForClaudeForkBound( + childId: string, + sourceNativeSessionId: string, + timeoutMs: number = 60_000 + ): Promise { + const startedAt = Date.now() + while (Date.now() - startedAt < timeoutMs) { + this.sessionCache.refreshSession(childId) + const child = this.sessionCache.getSession(childId) + const boundId = child?.metadata?.claudeSessionId + if ( + typeof boundId === 'string' + && boundId.length > 0 + && boundId !== sourceNativeSessionId + ) { + return true + } + // Give the runner a few seconds to come up before treating inactivity as failure. + if (child && !child.active && Date.now() - startedAt > 5_000) { + return false + } + await new Promise((resolve) => setTimeout(resolve, 250)) + } + return false + } + + /** + * Grok RPC already created `expectedNativeSessionId`. Wait until the child + * binds that exact id — a different id means load failed and fell back. + */ + private async waitForGrokForkBound( + childId: string, + expectedNativeSessionId: string, + timeoutMs: number = 60_000 + ): Promise { + const startedAt = Date.now() + while (Date.now() - startedAt < timeoutMs) { + this.sessionCache.refreshSession(childId) + const child = this.sessionCache.getSession(childId) + const boundId = child?.metadata?.grokSessionId + if (typeof boundId === 'string' && boundId.length > 0) { + return boundId === expectedNativeSessionId + } + if (child && !child.active && Date.now() - startedAt > 5_000) { + return false + } + await new Promise((resolve) => setTimeout(resolve, 250)) + } + return false + } + + /** Drop history locators that no longer have a corresponding message row. */ + private scrubHistoryLocators(sessionId: string, namespace: string): void { + const remainingLocalIds = new Set( + this.store.messages.getAllMessages(sessionId) + .flatMap((message) => (message.localId ? [message.localId] : [])) + ) + for (let attempt = 0; attempt < 3; attempt += 1) { + const session = this.sessionCache.getSessionByNamespace(sessionId, namespace) + ?? this.sessionCache.refreshSession(sessionId) + if (!session?.metadata) return + + const nextMetadata: Record = { ...session.metadata } + let changed = false + + const points = session.metadata.conversationHistoryPoints + if (points) { + const nextPoints = Object.fromEntries( + Object.entries(points).filter(([localId]) => remainingLocalIds.has(localId)) + ) + if (Object.keys(nextPoints).length !== Object.keys(points).length) { + changed = true + if (Object.keys(nextPoints).length > 0) { + nextMetadata.conversationHistoryPoints = nextPoints + } else { + delete nextMetadata.conversationHistoryPoints + } + } + } + + const indexes = session.metadata.conversationHistoryIndexes + if (indexes) { + const nextIndexes = Object.fromEntries( + Object.entries(indexes).filter(([localId]) => remainingLocalIds.has(localId)) + ) + if (Object.keys(nextIndexes).length !== Object.keys(indexes).length) { + changed = true + if (Object.keys(nextIndexes).length > 0) { + nextMetadata.conversationHistoryIndexes = nextIndexes + } else { + delete nextMetadata.conversationHistoryIndexes + } + } + } + + const turns = session.metadata.conversationHistoryTurns + if (turns) { + const nextTurns = Object.fromEntries( + Object.entries(turns).filter(([localId]) => remainingLocalIds.has(localId)) + ) + if (Object.keys(nextTurns).length !== Object.keys(turns).length) { + changed = true + if (Object.keys(nextTurns).length > 0) { + nextMetadata.conversationHistoryTurns = nextTurns + } else { + delete nextMetadata.conversationHistoryTurns + } + } + } + + if (!changed) return + + const result = this.store.sessions.updateSessionMetadata( + sessionId, + nextMetadata, + session.metadataVersion, + namespace, + { touchUpdatedAt: false } + ) + if (result.result === 'success') { + this.sessionCache.refreshSession(sessionId) + return + } + if (result.result !== 'version-mismatch') return + this.sessionCache.refreshSession(sessionId) + } + } + + private markConversationHistoryDiverged(sessionId: string, namespace: string): void { + for (let attempt = 0; attempt < 3; attempt += 1) { + const session = this.sessionCache.getSessionByNamespace(sessionId, namespace) + ?? this.sessionCache.refreshSession(sessionId) + if (!session?.metadata) return + if (session.metadata.conversationHistoryDiverged === true) return + + const result = this.store.sessions.updateSessionMetadata( + sessionId, + { ...session.metadata, conversationHistoryDiverged: true }, + session.metadataVersion, + namespace, + { touchUpdatedAt: false } + ) + if (result.result === 'success') { + this.sessionCache.refreshSession(sessionId) + return + } + if (result.result !== 'version-mismatch') return + this.sessionCache.refreshSession(sessionId) + } + } + + async forkConversation( + sessionId: string, + namespace: string, + messageLocalId?: string + ): Promise<{ type: 'success'; sessionId: string } | { type: 'error'; message: string }> { + if (this.historyActionsInFlight.has(sessionId)) { + return { type: 'error', message: 'Conversation history action already in progress' } + } + this.historyActionsInFlight.add(sessionId) + try { + return await this.forkConversationUnlocked(sessionId, namespace, messageLocalId) + } finally { + this.historyActionsInFlight.delete(sessionId) + } + } + + private async forkConversationUnlocked( + sessionId: string, + namespace: string, + messageLocalId?: string + ): Promise<{ type: 'success'; sessionId: string } | { type: 'error'; message: string }> { + const access = this.resolveSessionAccess(sessionId, namespace) + if (!access.ok) { + return { type: 'error', message: access.reason === 'not-found' ? 'Session not found' : 'Access denied' } + } + const source = access.session + try { + this.assertConversationHistoryIdle(source) + } catch (error) { + return { type: 'error', message: error instanceof Error ? error.message : String(error) } + } + + const history = source.metadata?.capabilities?.conversationHistory + if (messageLocalId) { + if (history?.forkAtMessage !== true) { + return { type: 'error', message: 'Historical fork is not supported for this session' } + } + try { + this.assertInvokedHistoryBoundary(sessionId, messageLocalId) + } catch (error) { + return { type: 'error', message: error instanceof Error ? error.message : String(error) } + } + } else if (history?.forkCurrent !== true) { + return { type: 'error', message: 'Fork current is not supported for this session' } + } + + const machineId = source.metadata?.machineId + const directory = source.metadata?.path + if (!machineId || !directory) { + return { type: 'error', message: 'Session is missing machine or path metadata' } + } + + let rpcResult: Awaited> + try { + rpcResult = await this.rpcGateway.forkConversation( + sessionId, + messageLocalId ? { messageLocalId } : {} + ) + } catch (error) { + return { type: 'error', message: error instanceof Error ? error.message : String(error) } + } + + if (!rpcResult?.nativeSessionId) { + return { type: 'error', message: 'Native fork did not return a session id' } + } + + const flavor = this.resolveFlavor(source) + const childId = randomUUID() + let prefix + try { + prefix = selectForkTranscriptPrefix(this.store.messages.getAllMessages(sessionId), messageLocalId) + } catch (error) { + return { type: 'error', message: error instanceof Error ? error.message : String(error) } + } + const copiedLocalIds = new Set( + prefix.flatMap((message) => (message.localId ? [message.localId] : [])) + ) + const childMetadata: Record = { + path: directory, + host: source.metadata?.host ?? 'unknown', + machineId, + flavor, + forkedFrom: sessionId, + startedBy: 'runner', + capabilities: source.metadata?.capabilities, + conversationHistoryPoints: Object.fromEntries( + Object.entries(source.metadata?.conversationHistoryPoints ?? {}) + .filter(([localId]) => copiedLocalIds.has(localId)) + ), + conversationHistoryIndexes: Object.fromEntries( + Object.entries(source.metadata?.conversationHistoryIndexes ?? {}) + .filter(([localId]) => copiedLocalIds.has(localId)) + ), + conversationHistoryTurns: Object.fromEntries( + Object.entries(source.metadata?.conversationHistoryTurns ?? {}) + .filter(([localId]) => copiedLocalIds.has(localId)) + ) + } + if (flavor === 'codex') { + childMetadata.codexSessionId = rpcResult.nativeSessionId + } else if (flavor === 'grok') { + childMetadata.grokSessionId = rpcResult.nativeSessionId + } else if (flavor === 'claude') { + // Child will bind the forked Claude id after --fork-session starts. + childMetadata.claudeSessionId = rpcResult.forkSession ? undefined : rpcResult.nativeSessionId + } + + let childCreated = false + let spawnAttempted = false + try { + this.sessionCache.getOrCreateSession( + `fork:${childId}`, + childMetadata, + null, + namespace, + source.model ?? undefined, + source.effort ?? undefined, + source.modelReasoningEffort ?? undefined, + childId + ) + childCreated = true + + // Native fork keeps agent context, but the new HAPI row starts empty. + // Hydrate the transcript prefix so web navigation is not a blank thread. + this.store.messages.copyMessagesToSession( + childId, + prefix.map((message) => ({ + content: message.content, + createdAt: message.createdAt, + localId: message.localId, + invokedAt: message.invokedAt, + scheduledAt: message.scheduledAt + })) + ) + this.sessionCache.rebuildTodosFromTranscript(childId) + this.sessionCache.refreshSession(childId) + + spawnAttempted = true + const spawn = await this.rpcGateway.spawnSession( + machineId, + directory, + flavor, + source.model ?? undefined, + source.modelReasoningEffort ?? undefined, + undefined, + 'simple', + undefined, + rpcResult.nativeSessionId, + source.effort ?? undefined, + source.permissionMode, + source.serviceTier ?? undefined, + childId, + source.collaborationMode, + rpcResult.forkSession === true + ) + if (spawn.type !== 'success') { + throw new Error(spawn.message) + } + + // Claude fork is spawn+flag, not an RPC-time snapshot. Keep the + // source history lock (caller holds historyActionsInFlight) until + // the child binds a distinct native id — otherwise the source can + // advance before --fork-session materializes. + if (flavor === 'claude' && rpcResult.forkSession === true) { + const bound = await this.waitForClaudeForkBound(childId, rpcResult.nativeSessionId) + if (!bound) { + throw new Error('Claude fork did not materialize before timeout') + } + } + + // Grok forks at RPC time, but spawn may still fall back to a blank + // session if load fails. Do not report success until the child is + // bound to the exact forked native id. + if (flavor === 'grok') { + const bound = await this.waitForGrokForkBound(childId, rpcResult.nativeSessionId) + if (!bound) { + throw new Error('Grok fork could not load the forked native session') + } + } + + return { type: 'success', sessionId: childId } + } catch (error) { + if (childCreated) { + try { + await this.cleanupFailedForkChild(childId, machineId, spawnAttempted) + } catch (cleanupError) { + const message = cleanupError instanceof Error ? cleanupError.message : String(cleanupError) + return { type: 'error', message: `Fork failed; child cleanup was not confirmed: ${message}` } + } + } + return { type: 'error', message: error instanceof Error ? error.message : String(error) } + } + } + + /** Kill an active fork child (if any) then delete the HAPI row. */ + private async cleanupFailedForkChild( + childId: string, + machineId: string, + spawnAttempted: boolean + ): Promise { + if (spawnAttempted) { + const status = await this.rpcGateway.stopRunnerSession(machineId, childId) + if (status === 'still_alive') { + throw new Error('Fork child termination was not confirmed') + } + } + const child = this.sessionCache.refreshSession(childId) + if (child?.active) { + this.handleSessionEnd({ sid: childId, time: Date.now(), reason: 'error' }) + } + await this.deleteSession(childId) + } + + async rewindConversation( + sessionId: string, + namespace: string, + messageLocalId: string + ): Promise<{ type: 'success' } | { type: 'error'; message: string; hydrateFailed?: boolean }> { + if (this.historyActionsInFlight.has(sessionId)) { + return { type: 'error', message: 'Conversation history action already in progress' } + } + this.historyActionsInFlight.add(sessionId) + try { + return await this.rewindConversationUnlocked(sessionId, namespace, messageLocalId) + } finally { + this.historyActionsInFlight.delete(sessionId) + } + } + + private async rewindConversationUnlocked( + sessionId: string, + namespace: string, + messageLocalId: string + ): Promise<{ type: 'success' } | { type: 'error'; message: string; hydrateFailed?: boolean }> { + const access = this.resolveSessionAccess(sessionId, namespace) + if (!access.ok) { + return { type: 'error', message: access.reason === 'not-found' ? 'Session not found' : 'Access denied' } + } + const session = access.session + try { + this.assertConversationHistoryIdle(session) + } catch (error) { + return { type: 'error', message: error instanceof Error ? error.message : String(error) } + } + if (session.metadata?.capabilities?.conversationHistory?.rewindToMessage !== true) { + return { type: 'error', message: 'Rewind is not supported for this session' } + } + try { + this.assertInvokedHistoryBoundary(sessionId, messageLocalId) + } catch (error) { + return { type: 'error', message: error instanceof Error ? error.message : String(error) } + } + + let rpcResult: Awaited> + try { + rpcResult = await this.rpcGateway.rewindConversation(sessionId, { messageLocalId }) + } catch (error) { + if (!(error instanceof RpcTargetMissingError)) { + this.markConversationHistoryDiverged(sessionId, namespace) + return { + type: 'error', + hydrateFailed: true, + message: 'Rewind outcome is unknown; session history requires reconciliation' + } + } + return { type: 'error', message: error.message } + } + + if (rpcResult?.success !== true) { + return { type: 'error', message: 'Native rewind failed' } + } + + try { + this.store.messages.truncateMessagesFromLocalId( + sessionId, + rpcResult.truncateFromLocalId ?? messageLocalId, + rpcResult.messages ?? [] + ) + this.scrubHistoryLocators(sessionId, namespace) + this.sessionCache.rebuildTodosFromTranscript(sessionId) + this.eventPublisher.emit({ type: 'messages-invalidated', sessionId, namespace }) + this.sessionCache.refreshSession(sessionId) + return { type: 'success' } + } catch (error) { + // Native history already changed; refuse further history actions until repaired. + this.markConversationHistoryDiverged(sessionId, namespace) + return { + type: 'error', + message: error instanceof Error ? error.message : String(error), + hydrateFailed: true + } + } + } + async archiveSession(sessionId: string): Promise { // tiann/hapi#916: when the CLI is already gone (e.g. after a // hub-restart cascade SIGTERMed the runner but the in-memory @@ -1041,6 +1526,9 @@ async uploadScratchlistAttachment( } async switchSession(sessionId: string, to: 'remote' | 'local'): Promise { + if (this.historyActionsInFlight.has(sessionId)) { + throw new Error('Conversation history action already in progress') + } await this.rpcGateway.switchSession(sessionId, to) } diff --git a/hub/src/web/routes/sessions.test.ts b/hub/src/web/routes/sessions.test.ts index ad6f18c8..609f6f4b 100644 --- a/hub/src/web/routes/sessions.test.ts +++ b/hub/src/web/routes/sessions.test.ts @@ -64,6 +64,8 @@ function createApp(session: Session, opts?: { sessionExists?: boolean archiveSession?: (sessionId: string) => Promise getCursorChatStoreStatus?: SyncEngine['getCursorChatStoreStatus'] + forkConversation?: SyncEngine['forkConversation'] + rewindConversation?: SyncEngine['rewindConversation'] }) { const applySessionConfigCalls: Array<[string, Record]> = [] const applySessionConfig = async (sessionId: string, config: Record) => { @@ -147,7 +149,9 @@ function createApp(session: Session, opts?: { listSlashCommands: opts?.listSlashCommands ?? (async () => ({ success: true, commands: [] - })) + })), + forkConversation: opts?.forkConversation ?? (async () => ({ type: 'success', sessionId: 'child-1' })), + rewindConversation: opts?.rewindConversation ?? (async () => ({ type: 'success' })) } as Partial const app = new Hono() @@ -1273,4 +1277,70 @@ describe('sessions routes', () => { }) }) + it('forks via POST /sessions/:id/fork and returns the child session id', async () => { + const calls: Array<{ sessionId: string; namespace: string; messageLocalId?: string }> = [] + const session = createSession({ + metadata: { + path: '/tmp/project', + host: 'localhost', + flavor: 'codex', + capabilities: { conversationHistory: { forkCurrent: true } } + } + }) + const { app } = createApp(session, { + forkConversation: async (sessionId, namespace, messageLocalId) => { + calls.push({ sessionId, namespace, messageLocalId }) + return { type: 'success', sessionId: 'forked-child' } + } + }) + + const response = await app.request('/api/sessions/session-1/fork', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({}) + }) + + expect(response.status).toBe(200) + expect(await response.json()).toEqual({ sessionId: 'forked-child' }) + expect(calls).toEqual([{ sessionId: 'session-1', namespace: 'default', messageLocalId: undefined }]) + }) + + it('rewinds via POST /sessions/:id/rewind', async () => { + const calls: Array<{ sessionId: string; messageLocalId: string }> = [] + const session = createSession({ + metadata: { + path: '/tmp/project', + host: 'localhost', + flavor: 'codex', + capabilities: { conversationHistory: { rewindToMessage: true } } + } + }) + const { app } = createApp(session, { + rewindConversation: async (sessionId, _namespace, messageLocalId) => { + calls.push({ sessionId, messageLocalId }) + return { type: 'success' } + } + }) + + const response = await app.request('/api/sessions/session-1/rewind', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ messageLocalId: 'local-2' }) + }) + + expect(response.status).toBe(200) + expect(await response.json()).toEqual({ success: true }) + expect(calls).toEqual([{ sessionId: 'session-1', messageLocalId: 'local-2' }]) + }) + + it('rejects rewind without messageLocalId', async () => { + const { app } = createApp(createSession()) + const response = await app.request('/api/sessions/session-1/rewind', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({}) + }) + expect(response.status).toBe(400) + }) + }) diff --git a/hub/src/web/routes/sessions.ts b/hub/src/web/routes/sessions.ts index ac92b1fd..b1bd315a 100644 --- a/hub/src/web/routes/sessions.ts +++ b/hub/src/web/routes/sessions.ts @@ -1,10 +1,12 @@ import { CursorMigrateToAcpRequestSchema, DeleteUploadRequestSchema, + ForkConversationRequestSchema, getPermissionModesForFlavor, isPermissionModeAllowedForFlavor, RenameSessionRequestSchema, ResumeSessionRequestSchema, + RewindConversationRequestSchema, SCRATCHLIST_MAX_ENTRIES, ScratchlistEntryCreateRequestSchema, ScratchlistEntryUpdateRequestSchema, @@ -326,6 +328,73 @@ export function createSessionsRoutes(getSyncEngine: () => SyncEngine | null): Ho return c.json({ ok: true }) }) + app.post('/sessions/:id/fork', async (c) => { + const engine = requireSyncEngine(c, getSyncEngine) + if (engine instanceof Response) { + return engine + } + + const sessionResult = requireSessionFromParam(c, engine, { requireActive: true }) + if (sessionResult instanceof Response) { + return sessionResult + } + + const rawBody = await c.req.text() + let body: unknown = {} + if (rawBody.trim()) { + try { + body = JSON.parse(rawBody) + } catch { + return c.json({ error: 'Invalid JSON body' }, 400) + } + } + const parsed = ForkConversationRequestSchema.safeParse(body) + if (!parsed.success) { + return c.json({ error: 'Invalid body' }, 400) + } + + const result = await engine.forkConversation( + sessionResult.sessionId, + c.get('namespace'), + parsed.data.messageLocalId + ) + if (result.type === 'error') { + return c.json({ error: result.message }, 409) + } + return c.json({ sessionId: result.sessionId }) + }) + + app.post('/sessions/:id/rewind', async (c) => { + const engine = requireSyncEngine(c, getSyncEngine) + if (engine instanceof Response) { + return engine + } + + const sessionResult = requireSessionFromParam(c, engine, { requireActive: true }) + if (sessionResult instanceof Response) { + return sessionResult + } + + const body = await c.req.json().catch(() => null) + const parsed = RewindConversationRequestSchema.safeParse(body) + if (!parsed.success) { + return c.json({ error: 'Invalid body' }, 400) + } + + const result = await engine.rewindConversation( + sessionResult.sessionId, + c.get('namespace'), + parsed.data.messageLocalId + ) + if (result.type === 'error') { + return c.json({ + error: result.message, + hydrateFailed: result.hydrateFailed === true + }, result.hydrateFailed ? 500 : 409) + } + return c.json({ success: true as const }) + }) + app.post('/sessions/:id/archive', async (c) => { // tiann/hapi#916: relax the blanket `requireActive: true` guard so // the endpoint is idempotent for already-archived rows AND can clean diff --git a/shared/package.json b/shared/package.json index 1513cb3c..7a485846 100644 --- a/shared/package.json +++ b/shared/package.json @@ -11,6 +11,7 @@ "./messages": "./src/messages.ts", "./slashCommands": "./src/slashCommands.ts", "./buildInfo": "./src/buildInfo.ts", + "./conversationHistory": "./src/conversationHistory.ts", "./modes": "./src/modes.ts", "./rpcMethods": "./src/rpcMethods.ts", "./schemas": "./src/schemas.ts", diff --git a/shared/src/apiTypes.ts b/shared/src/apiTypes.ts index bf4d3355..6b6b06b0 100644 --- a/shared/src/apiTypes.ts +++ b/shared/src/apiTypes.ts @@ -427,6 +427,45 @@ export const SendMessageRequestSchema = z.object({ export type SendMessageRequest = z.infer +export const ForkConversationRequestSchema = z.object({ + messageLocalId: z.string().min(1).optional() +}) + +export type ForkConversationRequest = z.infer + +export type ForkConversationResponse = { + sessionId: string +} + +export const RewindConversationRequestSchema = z.object({ + messageLocalId: z.string().min(1) +}) + +export type RewindConversationRequest = z.infer + +export type RewindConversationResponse = { + success: true +} + +/** CLI → hub RPC result for native fork (before HAPI child binding). */ +export type ForkConversationRpcResult = { + nativeSessionId: string + /** When true, hub must spawn with --fork-session (Claude). */ + forkSession?: boolean +} + +export type RewindConversationRpcResult = { + success: true + /** Truncate HAPI transcript at/after this localId, then accept rehydrated history. */ + truncateFromLocalId: string + messages?: Array<{ + content: unknown + localId?: string | null + createdAt?: number + invokedAt?: number | null + }> +} + export const QueuedStateRequestSchema = z.object({ localIds: z.array(z.string().min(1)).max(1000) }) diff --git a/shared/src/conversationHistory.test.ts b/shared/src/conversationHistory.test.ts new file mode 100644 index 00000000..81971b39 --- /dev/null +++ b/shared/src/conversationHistory.test.ts @@ -0,0 +1,25 @@ +import { describe, expect, it } from 'bun:test' +import { + CLAUDE_CONVERSATION_HISTORY, + markUnsupported, + toConversationHistoryCapabilities +} from './conversationHistory' + +describe('conversationHistory capabilities', () => { + it('only exposes supported flags', () => { + expect(toConversationHistoryCapabilities(CLAUDE_CONVERSATION_HISTORY)).toEqual({ + forkCurrent: true + }) + }) + + it('keeps unsupported sticky', () => { + const next = markUnsupported( + { forkCurrent: 'supported', forkAtMessage: 'supported', rewindToMessage: 'supported' }, + 'rewindToMessage' + ) + expect(toConversationHistoryCapabilities(next)).toEqual({ + forkCurrent: true, + forkAtMessage: true + }) + }) +}) diff --git a/shared/src/conversationHistory.ts b/shared/src/conversationHistory.ts new file mode 100644 index 00000000..0b907c22 --- /dev/null +++ b/shared/src/conversationHistory.ts @@ -0,0 +1,64 @@ +export type CapabilityState = 'unknown' | 'supported' | 'unsupported' + +export type ConversationHistoryCapabilityStates = { + forkCurrent: CapabilityState + forkAtMessage: CapabilityState + rewindToMessage: CapabilityState +} + +export type ConversationHistoryCapabilities = { + forkCurrent?: boolean + forkAtMessage?: boolean + rewindToMessage?: boolean +} + +/** Only `supported` becomes true in session metadata; never optimistic. */ +export function toConversationHistoryCapabilities( + states: ConversationHistoryCapabilityStates +): ConversationHistoryCapabilities | undefined { + const capabilities: ConversationHistoryCapabilities = {} + if (states.forkCurrent === 'supported') capabilities.forkCurrent = true + if (states.forkAtMessage === 'supported') capabilities.forkAtMessage = true + if (states.rewindToMessage === 'supported') capabilities.rewindToMessage = true + return Object.keys(capabilities).length > 0 ? capabilities : undefined +} + +export function markUnsupported( + states: ConversationHistoryCapabilityStates, + key: keyof ConversationHistoryCapabilityStates +): ConversationHistoryCapabilityStates { + if (states[key] === 'unsupported') return states + return { ...states, [key]: 'unsupported' } +} + +export function markSupported( + states: ConversationHistoryCapabilityStates, + key: keyof ConversationHistoryCapabilityStates +): ConversationHistoryCapabilityStates { + if (states[key] === 'unsupported') return states + return { ...states, [key]: 'supported' } +} + +export const UNSUPPORTED_CONVERSATION_HISTORY: ConversationHistoryCapabilityStates = { + forkCurrent: 'unsupported', + forkAtMessage: 'unsupported', + rewindToMessage: 'unsupported' +} + +export const CLAUDE_CONVERSATION_HISTORY: ConversationHistoryCapabilityStates = { + forkCurrent: 'supported', + forkAtMessage: 'unsupported', + rewindToMessage: 'unsupported' +} + +export const CODEX_CONVERSATION_HISTORY_INITIAL: ConversationHistoryCapabilityStates = { + forkCurrent: 'unknown', + forkAtMessage: 'unknown', + rewindToMessage: 'unknown' +} + +export const GROK_CONVERSATION_HISTORY_INITIAL: ConversationHistoryCapabilityStates = { + forkCurrent: 'unknown', + forkAtMessage: 'unknown', + rewindToMessage: 'unknown' +} diff --git a/shared/src/index.ts b/shared/src/index.ts index 3ea41a83..02853a1e 100644 --- a/shared/src/index.ts +++ b/shared/src/index.ts @@ -3,6 +3,7 @@ export * from './apiTypes' export * from './cursorCliSku' export * from './messages' export * from './buildInfo' +export * from './conversationHistory' export * from './effort' export * from './flavors' export * from './models' diff --git a/shared/src/rpcMethods.ts b/shared/src/rpcMethods.ts index 9c461a72..0ffd2da3 100644 --- a/shared/src/rpcMethods.ts +++ b/shared/src/rpcMethods.ts @@ -37,7 +37,9 @@ export const RPC_METHODS = { ListGrokModelsForCwd: 'listGrokModelsForCwd', ListGrokModels: 'listGrokModels', ListGrokReasoningEffortOptions: 'listGrokReasoningEffortOptions', - ListOpencodeReasoningEffortOptions: 'listOpencodeReasoningEffortOptions' + ListOpencodeReasoningEffortOptions: 'listOpencodeReasoningEffortOptions', + ForkConversation: 'fork-conversation', + RewindConversation: 'rewind-conversation', } as const export type RpcMethod = typeof RPC_METHODS[keyof typeof RPC_METHODS] diff --git a/shared/src/schemas.ts b/shared/src/schemas.ts index a0214b30..1483c4cf 100644 --- a/shared/src/schemas.ts +++ b/shared/src/schemas.ts @@ -11,10 +11,19 @@ const MetadataSummarySchema = z.object({ updatedAt: z.number() }) -const SessionCapabilitiesSchema = z.object({ - terminal: z.boolean().optional() +const ConversationHistoryCapabilitiesSchema = z.object({ + forkCurrent: z.boolean().optional(), + forkAtMessage: z.boolean().optional(), + rewindToMessage: z.boolean().optional() }) +const SessionCapabilitiesSchema = z.object({ + terminal: z.boolean().optional(), + conversationHistory: ConversationHistoryCapabilitiesSchema.optional() +}) + +export type ConversationHistoryCapabilities = z.infer + export const WorktreeMetadataSchema = z.object({ basePath: z.string(), branch: z.string(), @@ -34,6 +43,10 @@ export const MetadataSchema = z.object({ summary: MetadataSummarySchema.optional(), machineId: z.string().optional(), claudeSessionId: z.string().optional(), + // Parent HAPI session id when this session was created by message-level fork + // (`claude --resume --fork-session`). Lets the web list mark the new + // session as a branch of `` instead of an unrelated duplicate. + forkedFrom: z.string().optional(), codexSessionId: z.string().optional(), // 原始 Codex thread id。导入 Codex 历史后，HAPI 会 fork 出自己的续写 thread； // codexSessionId 保存 fork 后的 thread，codexSourceSessionId 保留来源 thread 便于同步/展示。 @@ -81,6 +94,14 @@ export const MetadataSchema = z.object({ preferredPermissionMode: PermissionModeSchema.optional(), flavor: z.string().nullish(), capabilities: SessionCapabilitiesSchema.optional(), + conversationHistoryPoints: z.record(z.string(), z.literal(true)).optional(), + // Native locators for historical fork/rewind (e.g. Grok prompt indexes). + // Kept separately from the boolean UI markers above. + conversationHistoryIndexes: z.record(z.string(), z.number().int().nonnegative()).optional(), + // Codex localId → turnId mapping (durable across runner relaunches). + conversationHistoryTurns: z.record(z.string(), z.string().min(1)).optional(), + // Set when native rewind succeeded but HAPI truncate/hydrate failed. + conversationHistoryDiverged: z.boolean().optional(), worktree: WorktreeMetadataSchema.optional(), // Cached Pi model list — written by CLI, read by web (inactive session fallback). // Minimal shape: each entry must have modelId; other fields (provider, name, etc.) pass through. diff --git a/web/src/api/client.ts b/web/src/api/client.ts index cca70656..e36bdf37 100644 --- a/web/src/api/client.ts +++ b/web/src/api/client.ts @@ -468,6 +468,26 @@ export class ApiClient { }) } + async forkConversation(sessionId: string, messageLocalId?: string): Promise<{ sessionId: string }> { + return await this.request<{ sessionId: string }>( + `/api/sessions/${encodeURIComponent(sessionId)}/fork`, + { + method: 'POST', + body: JSON.stringify(messageLocalId ? { messageLocalId } : {}) + } + ) + } + + async rewindConversation(sessionId: string, messageLocalId: string): Promise<{ success: true }> { + return await this.request<{ success: true }>( + `/api/sessions/${encodeURIComponent(sessionId)}/rewind`, + { + method: 'POST', + body: JSON.stringify({ messageLocalId }) + } + ) + } + async archiveSession(sessionId: string): Promise { await this.request(`/api/sessions/${encodeURIComponent(sessionId)}/archive`, { method: 'POST', diff --git a/web/src/components/AssistantChat/HappyThread.tsx b/web/src/components/AssistantChat/HappyThread.tsx index a2a1835a..c5d4a3dc 100644 --- a/web/src/components/AssistantChat/HappyThread.tsx +++ b/web/src/components/AssistantChat/HappyThread.tsx @@ -421,6 +421,10 @@ export function HappyThread(props: { disabled: boolean onRefresh: () => void onRetryMessage?: (localId: string) => void + historyActionPending?: boolean + onForkConversation?: (messageLocalId?: string) => Promise + onRewindConversation?: (messageLocalId: string) => Promise + isLatestCompletedBoundary?: (messageId: string) => boolean onViewModeChange: (mode: 'tail' | 'history') => void isSyncingTail: boolean messagesWarning: string | null @@ -1439,6 +1443,10 @@ export function HappyThread(props: { disabled: props.disabled, onRefresh: props.onRefresh, onRetryMessage: props.onRetryMessage, + historyActionPending: props.historyActionPending, + onForkConversation: props.onForkConversation, + onRewindConversation: props.onRewindConversation, + isLatestCompletedBoundary: props.isLatestCompletedBoundary, onShareTurn: handleShareTurn, hasMoreMessages: props.hasMoreMessages, isSyncingTail: props.isSyncingTail, diff --git a/web/src/components/AssistantChat/context.tsx b/web/src/components/AssistantChat/context.tsx index 651cd416..1e9f8826 100644 --- a/web/src/components/AssistantChat/context.tsx +++ b/web/src/components/AssistantChat/context.tsx @@ -14,6 +14,10 @@ export type HappyChatContextValue = { disabled: boolean onRefresh: () => void onRetryMessage?: (localId: string) => void + historyActionPending?: boolean + onForkConversation?: (messageLocalId?: string) => Promise + onRewindConversation?: (messageLocalId: string) => Promise + isLatestCompletedBoundary?: (messageId: string) => boolean onShareTurn?: ( messageElement: HTMLElement | string | null, clientY?: number, diff --git a/web/src/components/AssistantChat/messages/AssistantMessage.tsx b/web/src/components/AssistantChat/messages/AssistantMessage.tsx index 37ae1b13..f89d0e9b 100644 --- a/web/src/components/AssistantChat/messages/AssistantMessage.tsx +++ b/web/src/components/AssistantChat/messages/AssistantMessage.tsx @@ -8,6 +8,7 @@ import { getAssistantCopyText } from '@/components/AssistantChat/messages/assist import { getConversationMessageAnchorId } from '@/chat/outline' import { CodexReviewCard } from '@/components/AssistantChat/messages/CodexReviewCard' import { MessageActions } from '@/components/AssistantChat/messages/MessageActions' +import { useHappyChatContext } from '@/components/AssistantChat/context' const TOOL_COMPONENTS = { Fallback: HappyToolMessage @@ -21,6 +22,7 @@ const MESSAGE_PART_COMPONENTS = { } as const export function HappyAssistantMessage() { + const ctx = useHappyChatContext() const messageId = useAuiState((s) => s.message.id) const elementId = getConversationMessageAnchorId(messageId) const isCliOutput = useAuiState((s) => { @@ -53,6 +55,14 @@ export function HappyAssistantMessage() { const metadata = { durationMs, usage, model: messageModel ?? null, turnCount } + const history = ctx.metadata?.capabilities?.conversationHistory + const showForkCurrent = Boolean( + history?.forkCurrent + && ctx.isLatestCompletedBoundary?.(messageId) + && !ctx.disabled + && ctx.onForkConversation + ) + const rootClass = toolOnly ? 'py-1 min-w-0 max-w-full overflow-x-hidden' : 'px-1 min-w-0 max-w-full overflow-x-hidden' @@ -68,7 +78,15 @@ export function HappyAssistantMessage() { : codexReview ? : } - + ctx.onForkConversation!() : undefined} + /> ) } diff --git a/web/src/components/AssistantChat/messages/MessageActions.test.tsx b/web/src/components/AssistantChat/messages/MessageActions.test.tsx index 79ab511b..3765d093 100644 --- a/web/src/components/AssistantChat/messages/MessageActions.test.tsx +++ b/web/src/components/AssistantChat/messages/MessageActions.test.tsx @@ -7,11 +7,28 @@ import { MessageActions } from './MessageActions' const copy = vi.fn() vi.mock('@assistant-ui/react', () => ({ - useAuiState: (selector: (state: { message: { createdAt: Date } }) => unknown) => selector({ - message: { createdAt: new Date(2026, 6, 12, 10, 30) } + useAuiState: (selector: (state: { message: { createdAt: Date }; thread: { isRunning: boolean } }) => unknown) => selector({ + message: { createdAt: new Date(2026, 6, 12, 10, 30) }, + thread: { isRunning: false } }) })) +vi.mock('@/components/ui/ConfirmDialog', () => ({ + ConfirmDialog: (props: { + isOpen: boolean + title: string + confirmLabel: string + onConfirm: () => Promise + onClose: () => void + }) => props.isOpen ? ( +
+
{props.title}
+ + +
+ ) : null +})) + vi.mock('@radix-ui/react-popover', () => ({ Root: ({ children }: PropsWithChildren) => <>{children}, Trigger: ({ children }: PropsWithChildren) => <>{children}, @@ -116,4 +133,42 @@ describe('MessageActions', () => { expect(row).not.toBeNull() expect(row!.className.split(' ')).not.toContain('happy-message-actions-desktop-only-row') }) + + it('hides Fork and Rewind when capabilities are off', () => { + renderActions({ align: 'end', copyText: 'body' }) + expect(screen.queryByRole('button', { name: 'Fork' })).toBeNull() + expect(screen.queryByRole('button', { name: 'Rewind' })).toBeNull() + }) + + it('shows Fork confirm dialog and calls onFork only after confirm', async () => { + const onFork = vi.fn(async () => {}) + renderActions({ align: 'start', copyText: 'body', showFork: true, onFork }) + + fireEvent.click(screen.getByRole('button', { name: 'Fork' })) + expect(onFork).not.toHaveBeenCalled() + expect(screen.getByText('Fork conversation')).toBeTruthy() + + fireEvent.click(screen.getByRole('button', { name: 'Cancel' })) + expect(onFork).not.toHaveBeenCalled() + + fireEvent.click(screen.getByRole('button', { name: 'Fork' })) + fireEvent.click(screen.getAllByRole('button', { name: 'Fork' }).at(-1)!) + expect(onFork).toHaveBeenCalledTimes(1) + }) + + it('shows Rewind destructive confirm and calls onRewind only after confirm', async () => { + const onRewind = vi.fn(async () => {}) + renderActions({ align: 'end', copyText: 'body', showRewind: true, onRewind }) + + fireEvent.click(screen.getByRole('button', { name: 'Rewind' })) + expect(onRewind).not.toHaveBeenCalled() + expect(screen.getByText('Rewind conversation')).toBeTruthy() + + fireEvent.click(screen.getByRole('button', { name: 'Cancel' })) + expect(onRewind).not.toHaveBeenCalled() + + fireEvent.click(screen.getByRole('button', { name: 'Rewind' })) + fireEvent.click(screen.getAllByRole('button', { name: 'Rewind' }).at(-1)!) + expect(onRewind).toHaveBeenCalledTimes(1) + }) }) diff --git a/web/src/components/AssistantChat/messages/MessageActions.tsx b/web/src/components/AssistantChat/messages/MessageActions.tsx index b8e6a36c..37e0eb17 100644 --- a/web/src/components/AssistantChat/messages/MessageActions.tsx +++ b/web/src/components/AssistantChat/messages/MessageActions.tsx @@ -1,4 +1,5 @@ import * as Popover from '@radix-ui/react-popover' +import { useState } from 'react' import { useAuiState } from '@assistant-ui/react' import { CheckIcon, CopyIcon, InfoIcon } from '@/components/icons' import { useCopyToClipboard } from '@/hooks/useCopyToClipboard' @@ -7,20 +8,47 @@ import { MessageMetadata, buildMessageMetadataLabels, type MessageMetadataProps import { MessageTimestamp } from './MessageTimestamp' import { cn } from '@/lib/utils' import { ShareTurnButton } from './ShareTurnButton' +import { ConfirmDialog } from '@/components/ui/ConfirmDialog' + +export type MessageHistoryAction = { + kind: 'forkCurrent' | 'forkAtMessage' | 'rewind' + messageLocalId?: string +} type MessageActionsProps = { align: 'start' | 'end' copyText?: string metadata?: Omit messageElementId?: string + showFork?: boolean + showRewind?: boolean + historyActionPending?: boolean + onFork?: () => Promise + onRewind?: () => Promise } -export function MessageActions({ align, copyText, metadata, messageElementId }: MessageActionsProps) { +export function MessageActions({ + align, + copyText, + metadata, + messageElementId, + showFork = false, + showRewind = false, + historyActionPending = false, + onFork, + onRewind +}: MessageActionsProps) { const { copied, copy } = useCopyToClipboard() const { t } = useTranslation() const threadIsRunning = useAuiState(({ thread }) => thread?.isRunning ?? false) const canCopy = Boolean(copyText) const hasMetadata = metadata ? buildMessageMetadataLabels(metadata).length > 0 : false + const [forkOpen, setForkOpen] = useState(false) + const [rewindOpen, setRewindOpen] = useState(false) + const [forkPending, setForkPending] = useState(false) + const [rewindPending, setRewindPending] = useState(false) + const actionsLocked = historyActionPending || forkPending || rewindPending || threadIsRunning + const shareButton = messageElementId && !threadIsRunning ? ( ) : null - return ( -
- {align === 'end' ? : null} - {align === 'end' && hasMetadata && metadata ? : null} - {align === 'end' ? shareButton : null} - {canCopy ? ( + const historyButtons = ( + <> + {showFork && onFork ? ( ) : null} - {align === 'start' ? shareButton : null} - {align === 'start' && hasMetadata && metadata ? : null} - {align === 'start' ? : null} -
+ {showRewind && onRewind ? ( + + ) : null} + + ) + + return ( + <> +
+ {align === 'end' ? : null} + {align === 'end' && hasMetadata && metadata ? : null} + {align === 'end' ? shareButton : null} + {canCopy ? ( + + ) : null} + {historyButtons} + {align === 'start' ? shareButton : null} + {align === 'start' && hasMetadata && metadata ? : null} + {align === 'start' ? : null} +
+ + { + if (!forkPending) setForkOpen(false) + }} + title={t('message.fork.confirmTitle')} + description={t('message.fork.confirmDescription')} + confirmLabel={t('message.fork')} + confirmingLabel={t('message.fork.confirming')} + isPending={forkPending} + onConfirm={async () => { + if (!onFork) return + setForkPending(true) + try { + await onFork() + setForkOpen(false) + } finally { + setForkPending(false) + } + }} + /> + + { + if (!rewindPending) setRewindOpen(false) + }} + title={t('message.rewind.confirmTitle')} + description={t('message.rewind.confirmDescription')} + confirmLabel={t('message.rewind')} + confirmingLabel={t('message.rewind.confirming')} + isPending={rewindPending} + destructive + onConfirm={async () => { + if (!onRewind) return + setRewindPending(true) + try { + await onRewind() + setRewindOpen(false) + } finally { + setRewindPending(false) + } + }} + /> + ) } diff --git a/web/src/components/AssistantChat/messages/UserMessage.tsx b/web/src/components/AssistantChat/messages/UserMessage.tsx index b0c1ed51..aa3df25a 100644 --- a/web/src/components/AssistantChat/messages/UserMessage.tsx +++ b/web/src/components/AssistantChat/messages/UserMessage.tsx @@ -46,6 +46,32 @@ export function HappyUserMessage() { const onRetry = canRetry ? () => ctx.onRetryMessage!(localId) : undefined const showStatus = shouldShowMessageStatus(status) + const history = ctx.metadata?.capabilities?.conversationHistory + const hasNativePoint = typeof localId === 'string' + && localId.length > 0 + && ctx.metadata?.conversationHistoryPoints?.[localId] === true + const isLatestBoundary = ctx.isLatestCompletedBoundary?.(messageId) === true + const showCurrentFork = Boolean( + history?.forkCurrent + && isLatestBoundary + && !ctx.disabled + && ctx.onForkConversation + ) + const showHistoricalFork = Boolean( + history?.forkAtMessage + && hasNativePoint + && !isLatestBoundary + && !ctx.disabled + && ctx.onForkConversation + ) + const showFork = showCurrentFork || showHistoricalFork + const showRewind = Boolean( + history?.rewindToMessage + && hasNativePoint + && !ctx.disabled + && ctx.onRewindConversation + ) + if (isCliOutput) { return ( - + ctx.onForkConversation!() + : showHistoricalFork && localId + ? () => ctx.onForkConversation!(localId) + : undefined} + onRewind={showRewind && localId + ? () => ctx.onRewindConversation!(localId) + : undefined} + /> ) } diff --git a/web/src/components/SessionChat.tsx b/web/src/components/SessionChat.tsx index 59a2d5d5..a1c4918e 100644 --- a/web/src/components/SessionChat.tsx +++ b/web/src/components/SessionChat.tsx @@ -19,7 +19,7 @@ import { normalizeDecryptedMessage } from '@/chat/normalize' import { reduceChatBlocks } from '@/chat/reducer' import { reconcileChatBlocks } from '@/chat/reconcile' import { buildConversationOutline } from '@/chat/outline' -import { buildVisibleChatBlocks, isToolGroupBlock, type ToolGroupBlock } from '@/chat/toolGroups' +import { buildVisibleChatBlocks, isToolGroupBlock, visibleBlockRole, type ToolGroupBlock } from '@/chat/toolGroups' import { useUnseenBlockCount } from '@/hooks/useUnseenBlockCount' import { isQueuedForInvocation } from '@/lib/messages' import { inactiveSessionCanResume } from '@/lib/sessionResume' @@ -42,7 +42,7 @@ import { classifySessionAttention, getSessionAttentionLabelKey } from '@/lib/ses import { getSessionLastSeenAt } from '@/lib/sessionLastSeen' import { formatRelativeTime } from '@/lib/relativeTime' import { ScratchlistMigrationBanner } from '@/components/AssistantChat/ScratchlistMigrationBanner' -import { useHappyRuntime } from '@/lib/assistant-runtime' +import { assignThreadMessageIds, useHappyRuntime } from '@/lib/assistant-runtime' import type { OlderLoadOutcome } from '@/lib/message-window-store' import { createAttachmentAdapter } from '@/lib/attachmentAdapter' import { createScratchlistAttachmentAdapter } from '@/lib/scratchlistAttachmentAdapter' @@ -468,6 +468,27 @@ function SessionChatInner(props: SessionChatProps) { const { haptic } = usePlatform() const { t } = useTranslation() const navigate = useNavigate() + const [historyActionPending, setHistoryActionPending] = useState(false) + + const onForkConversation = useCallback(async (messageLocalId?: string) => { + setHistoryActionPending(true) + try { + const result = await props.api.forkConversation(props.session.id, messageLocalId) + await navigate({ to: '/sessions/$sessionId', params: { sessionId: result.sessionId } }) + } finally { + setHistoryActionPending(false) + } + }, [navigate, props.api, props.session.id]) + + const onRewindConversation = useCallback(async (messageLocalId: string) => { + setHistoryActionPending(true) + try { + await props.api.rewindConversation(props.session.id, messageLocalId) + props.onRefresh() + } finally { + setHistoryActionPending(false) + } + }, [props.api, props.onRefresh, props.session.id]) const sessionInactive = !props.session.active const inactiveCanResume = inactiveSessionCanResume( props.session, @@ -1049,6 +1070,30 @@ function SessionChatInner(props: SessionChatProps) { [reconciled.blocks, props.hasMoreMessages] ) + // Fork-current must compare against assistant-ui message ids (`kind:id`), + // not raw hub message ids — MessageActions receive the rendered card id, + // and adjacent assistant blocks join under the first block's id. + const latestCompletedBoundaryId = useMemo(() => { + if (props.viewMode !== 'tail') return null + let candidate: string | null = null + let previousRole: ReturnType | null = null + for (const { block, threadMessageId } of assignThreadMessageIds(visibleBlocks)) { + const role = visibleBlockRole(block) + if ( + (role === 'user' && block.invokedAt != null) + || (role === 'assistant' && previousRole !== 'assistant') + ) { + candidate = threadMessageId + } + previousRole = role + } + return candidate + }, [props.viewMode, visibleBlocks]) + + const isLatestCompletedBoundary = useCallback((messageId: string) => { + return latestCompletedBoundaryId === messageId + }, [latestCompletedBoundaryId]) + useEffect(() => { visibleGroupsRef.current = visibleBlocks.filter(isToolGroupBlock) }, [visibleBlocks]) @@ -1366,6 +1411,10 @@ function SessionChatInner(props: SessionChatProps) { disabled={sessionInactive} onRefresh={props.onRefresh} onRetryMessage={props.onRetryMessage} + historyActionPending={historyActionPending} + onForkConversation={controlledByUser ? undefined : onForkConversation} + onRewindConversation={controlledByUser ? undefined : onRewindConversation} + isLatestCompletedBoundary={isLatestCompletedBoundary} onViewModeChange={props.onViewModeChange} isSyncingTail={props.isSyncingTail} messagesWarning={props.messagesWarning} diff --git a/web/src/lib/locales/en.ts b/web/src/lib/locales/en.ts index 27bfa097..9473f24a 100644 --- a/web/src/lib/locales/en.ts +++ b/web/src/lib/locales/en.ts @@ -11,6 +11,14 @@ export default { 'message.copy': 'Copy', 'message.copied': 'Copied', 'message.info': 'Message details', + 'message.fork': 'Fork', + 'message.rewind': 'Rewind', + 'message.fork.confirmTitle': 'Fork conversation', + 'message.fork.confirmDescription': 'Create a new session from this point?\nThe current session will not be changed.', + 'message.fork.confirming': 'Forking…', + 'message.rewind.confirmTitle': 'Rewind conversation', + 'message.rewind.confirmDescription': 'Rewind this session to this point?\nLater conversation history will be permanently removed. Files will not be changed.', + 'message.rewind.confirming': 'Rewinding…', 'message.shareTurn': 'Share turn as image', 'shareTurn.title': 'Share turn as image', 'shareTurn.badge': 'Shared turn', diff --git a/web/src/lib/locales/zh-CN.ts b/web/src/lib/locales/zh-CN.ts index bbf6dc6a..6abe0977 100644 --- a/web/src/lib/locales/zh-CN.ts +++ b/web/src/lib/locales/zh-CN.ts @@ -11,6 +11,14 @@ export default { 'message.copy': '复制', 'message.copied': '已复制', 'message.info': '消息详情', + 'message.fork': 'Fork', + 'message.rewind': 'Rewind', + 'message.fork.confirmTitle': '分叉对话', + 'message.fork.confirmDescription': '从此处创建新会话？\n当前会话不会被修改。', + 'message.fork.confirming': '分叉中…', + 'message.rewind.confirmTitle': '回退对话', + 'message.rewind.confirmDescription': '将此会话回退到此处？\n之后的对话历史将永久移除。文件不会被修改。', + 'message.rewind.confirming': '回退中…', 'message.shareTurn': '将本轮对话分享为图片', 'shareTurn.title': '将本轮对话分享为图片', 'shareTurn.badge': '分享会话', diff --git a/web/src/types/api.ts b/web/src/types/api.ts index 18d3b308..0b5d9bf7 100644 --- a/web/src/types/api.ts +++ b/web/src/types/api.ts @@ -84,7 +84,16 @@ export type SessionMetadataSummary = { flavor?: string | null capabilities?: { terminal?: boolean + conversationHistory?: { + forkCurrent?: boolean + forkAtMessage?: boolean + rewindToMessage?: boolean + } } + conversationHistoryPoints?: Record + conversationHistoryIndexes?: Record + conversationHistoryTurns?: Record + conversationHistoryDiverged?: boolean worktree?: WorktreeMetadata }