mirror of
https://github.com/wu736139669/hapi.git
synced 2026-08-07 06:52:28 +00:00
feat: message-level conversation fork and rewind (#1263)
* feat: add message-level conversation fork and rewind Expose native Codex/Grok/Claude history controls through hub REST+RPC and web ConfirmDialog actions, without file rewind or composed forks. Also reconcile the duplicate hub V14→V15 migration so typecheck can pass. Co-authored-by: Cursor <cursoragent@cursor.com> * fix: hydrate fork transcript and consume Claude --fork-session Forked HAPI children now copy the source transcript prefix so navigation is not a blank thread, and Claude drops --fork-session after the first launch so relaunches do not branch again. Co-authored-by: Cursor <cursoragent@cursor.com> * fix: harden fork/rewind concurrency and durable history points Skip pending scheduled rows when hydrating fork transcripts, serialize fork/rewind per session, and persist conversation history points/indexes across existing-session bootstrap and Grok relaunches. Co-authored-by: Cursor <cursoragent@cursor.com> * fix: close remaining fork/rewind races and UI anchoring Block sends and scheduled maturation while history actions run, order fork prefixes by invocation time, inherit history locators into children, and only offer Fork current on the live tail boundary. Co-authored-by: Cursor <cursoragent@cursor.com> * fix: address remaining fork/rewind bot findings Materialize Claude --fork-session before the first child prompt, validate HAPI history boundaries before native RPC, expose forkCurrent on a latest user boundary, fully demote unsupported conversationHistory capabilities, and fix the truncate test setup order. Co-authored-by: Cursor <cursoragent@cursor.com> * fix: align fork-current ids and Claude fork bootstrap Compare the latest fork boundary in assistant-ui threadMessageId space, spawn Claude forks with the persisted session mode, and preserve forkedFrom across existing-session bootstrap. Co-authored-by: Cursor <cursoragent@cursor.com> * fix: close fork/rewind consistency holes at the contract layer Hold the source history lock until Claude child binds a distinct native id, persist Codex localId→turnId locators, and mark/block diverged sessions when native rewind outruns HAPI truncate. Co-authored-by: Cursor <cursoragent@cursor.com> * fix: use Codex stable lastTurnId for historical fork Map HAPI's exclusive boundary to the previous turn's inclusive lastTurnId so native fork context matches the hydrated transcript. Co-authored-by: Cursor <cursoragent@cursor.com> * fix: require exact Grok native resume for fork children Reject newSession fallback when forkedFrom is set, and keep the hub history lock until the child binds the forked grokSessionId. Co-authored-by: Cursor <cursoragent@cursor.com> * fix: kill active fork children before failed-fork cleanup Bind/readiness failures can leave the child process running; deleteSession rejects active rows, so terminate first then remove the HAPI session. Co-authored-by: Cursor <cursoragent@cursor.com> * fix: close remaining fork lock, hydrate, and todos gaps Reject mode switches during history actions, batch-copy fork transcripts in one SQLite transaction, and rebuild todos after fork hydrate / rewind truncate. Co-authored-by: Cursor <cursoragent@cursor.com> * fix: allow Codex historical fork before the first turn Use experimental beforeTurnId when there is no previous turn for the stable inclusive lastTurnId boundary. Co-authored-by: Cursor <cursoragent@cursor.com> * fix: mark Grok history busy immediately after dequeue Hub idle checks clear once messages-consumed fires; hold the busy flag across permission sync and rewind-points lookup before prompt starts. Co-authored-by: Cursor <cursoragent@cursor.com> * fix: encode copied conversation history content * fix(web): hide local conversation history actions * style(codex): remove trailing whitespace * fix(fork): preserve children when cleanup is unconfirmed * fix(history): confirm cleanup and guard rewind divergence * fix(history): probe capabilities before advertising --------- Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
SSU-WEI HUANG
@@ -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<T = unknown>(
|
||||
method: string,
|
||||
params: Record<string, unknown>,
|
||||
options?: { timeoutMs?: number }
|
||||
): Promise<T> {
|
||||
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
|
||||
|
||||
@@ -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]
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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) {
|
||||
|
||||
+121
-61
@@ -25,6 +25,8 @@ export async function claudeRemote(opts: {
|
||||
hookSettingsPath: string,
|
||||
signal?: AbortSignal,
|
||||
canCallTool: (toolName: string, input: unknown, mode: EnhancedMode, options: { signal: AbortSignal }) => Promise<PermissionResult>,
|
||||
/** 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<typeof parseSpecialCommand> = { 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<SDKUserMessage>();
|
||||
|
||||
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<SDKUserMessage>();
|
||||
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);
|
||||
}
|
||||
|
||||
|
||||
@@ -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);
|
||||
},
|
||||
|
||||
@@ -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<void> {
|
||||
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);
|
||||
|
||||
@@ -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(','))
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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'])
|
||||
})
|
||||
})
|
||||
@@ -122,8 +122,10 @@ export class Session extends AgentSessionBase<EnhancedMode> {
|
||||
};
|
||||
|
||||
/**
|
||||
* 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<EnhancedMode> {
|
||||
// --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]);
|
||||
}
|
||||
|
||||
@@ -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<ThreadResumeParams, 'history' | 'path'> {
|
||||
/** 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 {
|
||||
|
||||
@@ -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<boolean> {
|
||||
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<ThreadReadResponse> {
|
||||
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<TurnStartResponse> {
|
||||
const response = await this.sendRequest('turn/start', params, {
|
||||
signal: options?.signal,
|
||||
|
||||
@@ -1018,6 +1018,7 @@ function createSessionStub(
|
||||
rpcHandlers.set(method, handler);
|
||||
}
|
||||
},
|
||||
updateMetadata(_handler: (metadata: Record<string, unknown>) => Record<string, unknown>) {},
|
||||
updateAgentState(handler: (state: FakeAgentState) => FakeAgentState) {
|
||||
agentState = handler(agentState);
|
||||
},
|
||||
|
||||
@@ -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<ReturnType<typeof registerGeneratedImage> | null> {
|
||||
@@ -58,7 +59,13 @@ async function registerGeneratedImageFromPath(args: { id: string; path: string;
|
||||
}
|
||||
|
||||
type HappyServer = Awaited<ReturnType<typeof buildHapiMcpBridge>>['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<string, string>();
|
||||
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;
|
||||
|
||||
|
||||
@@ -0,0 +1,143 @@
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import { CodexConversationHistory } from './conversationHistory'
|
||||
|
||||
function createClient(overrides?: {
|
||||
fork?: (params: Record<string, unknown>) => Promise<{ thread: { id: string } }>
|
||||
rollback?: (params: { threadId: string; numTurns: number }) => Promise<unknown>
|
||||
read?: () => Promise<{ thread: { id: string; turns: Array<Record<string, unknown>> } }>
|
||||
}) {
|
||||
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<string, unknown>) => {
|
||||
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<string, unknown>) => {
|
||||
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<string, unknown>) => {
|
||||
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<string, unknown>) => {
|
||||
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')
|
||||
})
|
||||
})
|
||||
@@ -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<string, unknown> | null {
|
||||
return value !== null && typeof value === 'object' && !Array.isArray(value)
|
||||
? value as Record<string, unknown>
|
||||
: 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<string, string>()
|
||||
private busy = false
|
||||
private publishCapabilities: (() => Promise<void>) | null = null
|
||||
|
||||
constructor(private readonly getClient: () => CodexAppServerClient | null) {}
|
||||
|
||||
setPublishCapabilities(fn: () => Promise<void>): void {
|
||||
this.publishCapabilities = fn
|
||||
}
|
||||
|
||||
setBusy(busy: boolean): void {
|
||||
this.busy = busy
|
||||
}
|
||||
|
||||
setThreadId(threadId: string | null): void {
|
||||
this.threadId = threadId
|
||||
}
|
||||
|
||||
restoreTurns(turns: Record<string, string> | null | undefined): void {
|
||||
if (!turns) return
|
||||
for (const [localId, turnId] of Object.entries(turns)) {
|
||||
if (localId && turnId) this.turnByLocalId.set(localId, turnId)
|
||||
}
|
||||
}
|
||||
|
||||
getTurns(): Record<string, string> {
|
||||
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<void> {
|
||||
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<ForkConversationRpcResult> {
|
||||
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<RewindConversationRpcResult> {
|
||||
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<string> {
|
||||
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<TurnInfo[]> {
|
||||
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]
|
||||
}))
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -50,6 +50,12 @@ export function parseRemoteAgentCommandOptions<TPermissionMode extends Permissio
|
||||
throw new Error('Missing --resume value')
|
||||
}
|
||||
options.resumeSessionId = sessionId
|
||||
} else if (arg === '--existing-session-id') {
|
||||
const sessionId = args[++i]
|
||||
if (!sessionId) {
|
||||
throw new Error('Missing --existing-session-id value')
|
||||
}
|
||||
options.existingSessionId = sessionId
|
||||
} else if (arg === '-s' || arg === '--session') {
|
||||
// OpenCode-native resume flags (hapi opencode -s / --session <id>)
|
||||
const sessionId = args[++i]
|
||||
|
||||
@@ -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('-')) {
|
||||
|
||||
@@ -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<string, unknown>) => {
|
||||
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<string, unknown>) => {
|
||||
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<string, unknown>) => {
|
||||
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<string, unknown>) => {
|
||||
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()
|
||||
})
|
||||
})
|
||||
@@ -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<string, unknown> | null {
|
||||
return value !== null && typeof value === 'object' && !Array.isArray(value)
|
||||
? value as Record<string, unknown>
|
||||
: 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<string, number>()
|
||||
private busy = false
|
||||
private publishCapabilities: (() => Promise<void>) | null = null
|
||||
|
||||
constructor(private readonly getBackend: () => AcpSdkBackend | null) {}
|
||||
|
||||
setPublishCapabilities(fn: () => Promise<void>): 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<string, true> {
|
||||
const points: Record<string, true> = {}
|
||||
for (const localId of this.promptIndexByLocalId.keys()) {
|
||||
points[localId] = true
|
||||
}
|
||||
return points
|
||||
}
|
||||
|
||||
getHistoryIndexes(): Record<string, number> {
|
||||
const indexes: Record<string, number> = {}
|
||||
for (const [localId, promptIndex] of this.promptIndexByLocalId.entries()) {
|
||||
indexes[localId] = promptIndex
|
||||
}
|
||||
return indexes
|
||||
}
|
||||
|
||||
restorePromptIndexes(indexes: Record<string, number> | 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<void> {
|
||||
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<ForkConversationRpcResult> {
|
||||
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<string, unknown> = {
|
||||
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<Record<string, unknown>>(
|
||||
'_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<RewindConversationRpcResult> {
|
||||
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<Record<string, unknown>>(
|
||||
'_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
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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'])
|
||||
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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 =
|
||||
|
||||
@@ -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',
|
||||
|
||||
+15
-4
@@ -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);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user