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 }