From 5a377e38b9c43d38d88ffc5ff67b7426c8ea3108 Mon Sep 17 00:00:00 2001 From: weishu Date: Sun, 12 Jul 2026 10:59:15 +0800 Subject: [PATCH] fix(codex): defer session persistence until user activity --- cli/src/agent/runnerLifecycle.test.ts | 9 + cli/src/agent/runnerLifecycle.ts | 2 +- cli/src/agent/sessionFactory.test.ts | 68 ++- cli/src/agent/sessionFactory.ts | 82 +++ cli/src/api/api.ts | 27 +- cli/src/api/apiSession.test.ts | 325 +++++++++++- cli/src/api/apiSession.ts | 470 +++++++++++++++--- cli/src/claude/utils/sessionHookForwarder.ts | 26 +- cli/src/claude/utils/startHookServer.ts | 8 +- cli/src/codex/codexLocalLauncher.test.ts | 82 ++- cli/src/codex/codexLocalLauncher.ts | 39 +- cli/src/codex/runCodex.test.ts | 29 ++ cli/src/codex/runCodex.ts | 7 +- cli/src/codex/session.ts | 4 + cli/src/codex/utils/codexCliOverrides.test.ts | 10 + cli/src/codex/utils/codexCliOverrides.ts | 18 + .../codex/utils/codexEventConverter.test.ts | 16 +- cli/src/codex/utils/codexEventConverter.ts | 18 +- .../codex/utils/codexSessionScanner.test.ts | 26 + cli/src/codex/utils/codexSessionScanner.ts | 6 +- .../utils/codexTranscriptLocator.test.ts | 250 ++++++++++ cli/src/codex/utils/codexTranscriptLocator.ts | 380 ++++++++++++++ cli/src/codex/utils/codexVersion.test.ts | 4 +- cli/src/codex/utils/codexVersion.ts | 2 + cli/src/commands/codex.test.ts | 14 + cli/src/commands/codex.ts | 6 +- cli/src/utils/autoStartServer.ts | 31 +- hub/src/store/sessionStore.ts | 5 +- hub/src/store/sessions.test.ts | 61 +++ hub/src/store/sessions.ts | 25 +- hub/src/sync/sessionCache.ts | 14 +- hub/src/sync/syncEngine.ts | 14 +- hub/src/web/routes/cli.test.ts | 104 +++- hub/src/web/routes/cli.ts | 43 +- shared/src/apiTypes.ts | 24 +- 35 files changed, 2118 insertions(+), 131 deletions(-) create mode 100644 cli/src/codex/utils/codexTranscriptLocator.test.ts create mode 100644 cli/src/codex/utils/codexTranscriptLocator.ts diff --git a/cli/src/agent/runnerLifecycle.test.ts b/cli/src/agent/runnerLifecycle.test.ts index 172dca0a..f5b5b2d6 100644 --- a/cli/src/agent/runnerLifecycle.test.ts +++ b/cli/src/agent/runnerLifecycle.test.ts @@ -100,6 +100,15 @@ describe('createRunnerLifecycle', () => { await lc.cleanup(); expect(session.sendSessionDeath).toHaveBeenCalledWith('completed'); }); + + it('limits the final connected flush budget to one second', async () => { + const session = createMockApiSession(); + const lc = createRunnerLifecycle({ session, logTag: 'test' }); + + await lc.cleanup(); + + expect(session.flush).toHaveBeenCalledWith({ timeoutMs: 1_000 }); + }); }); }); diff --git a/cli/src/agent/runnerLifecycle.ts b/cli/src/agent/runnerLifecycle.ts index 16a04951..d6ae1419 100644 --- a/cli/src/agent/runnerLifecycle.ts +++ b/cli/src/agent/runnerLifecycle.ts @@ -62,7 +62,7 @@ export function createRunnerLifecycle(options: RunnerLifecycleOptions): RunnerLi })) options.session.sendSessionDeath(sessionEndReason) - await options.session.flush() + await options.session.flush({ timeoutMs: 1_000 }) await options.session.close() } diff --git a/cli/src/agent/sessionFactory.test.ts b/cli/src/agent/sessionFactory.test.ts index 70a7c31a..246cc2a1 100644 --- a/cli/src/agent/sessionFactory.test.ts +++ b/cli/src/agent/sessionFactory.test.ts @@ -3,12 +3,14 @@ import type { Session } from '@/api/types' const { getSessionMock, + getOrCreateSessionMock, getOrCreateMachineMock, sessionSyncClientMock, notifyRunnerSessionStartedMock, readSettingsMock } = vi.hoisted(() => ({ getSessionMock: vi.fn(), + getOrCreateSessionMock: vi.fn(), getOrCreateMachineMock: vi.fn(), sessionSyncClientMock: vi.fn(), notifyRunnerSessionStartedMock: vi.fn(async () => ({})), @@ -19,6 +21,7 @@ vi.mock('@/api/api', () => ({ ApiClient: { create: async () => ({ getSession: getSessionMock, + getOrCreateSession: getOrCreateSessionMock, getOrCreateMachine: getOrCreateMachineMock, sessionSyncClient: sessionSyncClientMock }) @@ -47,7 +50,7 @@ vi.mock('@/ui/logger', () => ({ } })) -import { bootstrapExistingSession, buildSessionMetadata } from './sessionFactory' +import { bootstrapExistingSession, bootstrapLazySession, buildSessionMetadata } from './sessionFactory' function createSession(): Session { return { @@ -83,6 +86,7 @@ function createSession(): Session { describe('bootstrapExistingSession', () => { beforeEach(() => { getSessionMock.mockReset() + getOrCreateSessionMock.mockReset() getOrCreateMachineMock.mockReset() sessionSyncClientMock.mockReset() notifyRunnerSessionStartedMock.mockClear() @@ -194,3 +198,65 @@ describe('bootstrapExistingSession', () => { expect(metadata.capabilities?.terminal).toBe(true) }) }) + +describe('bootstrapLazySession', () => { + beforeEach(() => { + getOrCreateSessionMock.mockReset() + getOrCreateMachineMock.mockReset() + sessionSyncClientMock.mockReset() + notifyRunnerSessionStartedMock.mockClear() + readSettingsMock.mockReset() + }) + + it('does not persist a machine or session until materialization', async () => { + const pendingClient = { isPending: () => true } + sessionSyncClientMock.mockReturnValue(pendingClient) + readSettingsMock.mockResolvedValue({ machineId: 'machine-1' }) + + const result = await bootstrapLazySession({ + flavor: 'codex', + startedBy: 'terminal', + workingDirectory: '/tmp/project', + agentState: { controlledByUser: false } + }) + + expect(result.session).toBe(pendingClient) + expect(getOrCreateMachineMock).not.toHaveBeenCalled() + expect(getOrCreateSessionMock).not.toHaveBeenCalled() + expect(notifyRunnerSessionStartedMock).not.toHaveBeenCalled() + + const [provisional, options] = sessionSyncClientMock.mock.calls[0] + expect(provisional.id).toMatch(/^[0-9a-f-]{36}$/) + expect(provisional.metadata).toEqual(expect.objectContaining({ + machineId: 'machine-1', + path: '/tmp/project', + flavor: 'codex' + })) + + const materialized = createSession() + materialized.id = provisional.id + getOrCreateSessionMock.mockResolvedValue(materialized) + const snapshot = { + metadata: { + ...provisional.metadata, + codexSessionId: 'codex-thread-1' + }, + agentState: { controlledByUser: true } + } + await options.materialize(snapshot, new AbortController().signal) + + expect(getOrCreateSessionMock).toHaveBeenCalledWith(expect.objectContaining({ + id: provisional.id, + metadata: snapshot.metadata, + state: snapshot.agentState, + timeoutMs: 10_000, + machine: expect.objectContaining({ id: 'machine-1' }) + })) + + options.onMaterialized(materialized, snapshot) + expect(notifyRunnerSessionStartedMock).toHaveBeenCalledWith( + provisional.id, + expect.objectContaining({ codexSessionId: 'codex-thread-1' }) + ) + }) +}) diff --git a/cli/src/agent/sessionFactory.ts b/cli/src/agent/sessionFactory.ts index fcc26296..124ae124 100644 --- a/cli/src/agent/sessionFactory.ts +++ b/cli/src/agent/sessionFactory.ts @@ -184,6 +184,88 @@ export async function bootstrapSession(options: SessionBootstrapOptions): Promis } } +export async function bootstrapLazySession(options: SessionBootstrapOptions): Promise { + const workingDirectory = options.workingDirectory ?? getInvokedCwd() + const startedBy = options.startedBy ?? 'terminal' + if (startedBy !== 'terminal') { + throw new Error('Lazy session bootstrap is only supported for terminal sessions') + } + + const api = await ApiClient.create() + const machineId = await getMachineIdOrExit() + const machineMetadata = buildMachineMetadata() + const metadata = buildSessionMetadata({ + flavor: options.flavor, + startedBy, + workingDirectory, + machineId, + metadataOverrides: options.metadataOverrides + }) + const agentState = options.agentState === undefined ? {} : options.agentState + const now = Date.now() + const requestedId = randomUUID() + const sessionTag = options.tag ?? randomUUID() + const sessionInfo: Session = { + id: requestedId, + namespace: 'pending', + seq: 0, + createdAt: now, + updatedAt: now, + active: false, + activeAt: now, + metadata, + metadataVersion: 0, + agentState, + agentStateVersion: 0, + thinking: false, + thinkingAt: now, + todos: [], + model: options.model ?? null, + modelReasoningEffort: options.modelReasoningEffort ?? null, + effort: options.effort ?? null, + serviceTier: null, + permissionMode: undefined, + collaborationMode: undefined + } + + const session = api.sessionSyncClient(sessionInfo, { + materialize: async (snapshot, signal) => { + const materialized = await api.getOrCreateSession({ + id: requestedId, + tag: sessionTag, + metadata: snapshot.metadata ?? metadata, + state: snapshot.agentState, + model: options.model, + modelReasoningEffort: options.modelReasoningEffort, + effort: options.effort, + machine: { + id: machineId, + metadata: machineMetadata + }, + timeoutMs: 10_000, + signal + }) + if (materialized.id !== requestedId) { + throw new Error(`Hub returned unexpected session id ${materialized.id}`) + } + return materialized + }, + onMaterialized: (materialized, snapshot) => { + void reportSessionStarted(materialized.id, snapshot.metadata ?? metadata) + } + }) + + return { + api, + session, + sessionInfo, + metadata, + machineId, + startedBy, + workingDirectory + } +} + export async function bootstrapExistingSession(options: { sessionId: string flavor: string diff --git a/cli/src/api/api.ts b/cli/src/api/api.ts index df52e454..275c7196 100644 --- a/cli/src/api/api.ts +++ b/cli/src/api/api.ts @@ -17,7 +17,7 @@ import { configuration } from '@/configuration' import { getAuthToken } from '@/api/auth' import { apiValidationError } from '@/utils/errorUtils' import { ApiMachineClient } from './apiMachine' -import { ApiSessionClient } from './apiSession' +import { ApiSessionClient, type ApiSessionClientOptions } from './apiSession' import { buildHubRequestHeaders } from './hubExtraHeaders' export class ApiClient { @@ -35,29 +35,46 @@ export class ApiClient { } async getOrCreateSession(opts: { + id?: string tag: string metadata: Metadata state: AgentState | null model?: string modelReasoningEffort?: string effort?: string + machine?: { + id: string + metadata: MachineMetadata + runnerState?: RunnerState + } + timeoutMs?: number + signal?: AbortSignal }): Promise { const response = await axios.post( `${configuration.apiUrl}/cli/sessions`, { + id: opts.id, tag: opts.tag, metadata: opts.metadata, agentState: opts.state, model: opts.model, modelReasoningEffort: opts.modelReasoningEffort, - effort: opts.effort + effort: opts.effort, + machine: opts.machine + ? { + id: opts.machine.id, + metadata: opts.machine.metadata, + runnerState: opts.machine.runnerState ?? null + } + : undefined }, { headers: buildHubRequestHeaders({ Authorization: `Bearer ${this.token}`, 'Content-Type': 'application/json' }), - timeout: 60_000 + timeout: opts.timeoutMs ?? 60_000, + signal: opts.signal } ) @@ -255,8 +272,8 @@ export class ApiClient { } } - sessionSyncClient(session: Session): ApiSessionClient { - return new ApiSessionClient(this.token, session) + sessionSyncClient(session: Session, options?: ApiSessionClientOptions): ApiSessionClient { + return new ApiSessionClient(this.token, session, options) } machineSyncClient(machine: Machine, options?: { workspaceRoots?: string[] }): ApiMachineClient { diff --git a/cli/src/api/apiSession.test.ts b/cli/src/api/apiSession.test.ts index a44ebc86..5b4b5a44 100644 --- a/cli/src/api/apiSession.test.ts +++ b/cli/src/api/apiSession.test.ts @@ -1,5 +1,326 @@ -import { describe, expect, it } from 'vitest' -import { isExternalUserMessage, IncomingMessageFilter } from './apiSession' +import { describe, expect, it, vi } from 'vitest' +import type { Session } from './types' + +const socketHarness = vi.hoisted(() => ({ + sockets: [] as Array<{ + connected: boolean + connectCalls: number + connectImmediately: boolean + emitted: Array<{ event: string; args: unknown[] }> + listeners: Map void>> + triggerConnect: () => void + triggerConnectError: () => void + }> +})) + +vi.mock('socket.io-client', () => ({ + io: () => { + const state = { + connected: false, + connectCalls: 0, + connectImmediately: true, + emitted: [] as Array<{ event: string; args: unknown[] }>, + listeners: new Map void>>(), + triggerConnect: () => {}, + triggerConnectError: () => {} + } + const triggerConnect = () => { + state.connected = true + for (const listener of state.listeners.get('connect') ?? []) listener() + } + state.triggerConnect = triggerConnect + state.triggerConnectError = () => { + for (const listener of state.listeners.get('connect_error') ?? []) { + listener(new Error('connect failed')) + } + } + const socket = { + get connected() { + return state.connected + }, + on: (event: string, listener: (...args: any[]) => void) => { + const listeners = state.listeners.get(event) ?? [] + listeners.push(listener) + state.listeners.set(event, listeners) + return socket + }, + off: (event: string, listener: (...args: any[]) => void) => { + const listeners = state.listeners.get(event) ?? [] + state.listeners.set(event, listeners.filter((candidate) => candidate !== listener)) + return socket + }, + emit: (event: string, ...args: unknown[]) => { + state.emitted.push({ event, args }) + return socket + }, + emitWithAck: async () => ({}), + timeout: () => ({ emitWithAck: async () => ({}) }), + connect: () => { + state.connectCalls += 1 + if (state.connectImmediately) { + triggerConnect() + } + return socket + }, + disconnect: () => { + state.connected = false + return socket + } + } + Object.assign(socket, { volatile: socket }) + socketHarness.sockets.push(state) + return socket + } +})) + +import { ApiSessionClient, isExternalUserMessage, IncomingMessageFilter } from './apiSession' + +function createSession(overrides: Partial = {}): Session { + return { + id: '11111111-1111-4111-8111-111111111111', + namespace: 'pending', + seq: 0, + createdAt: 1, + updatedAt: 1, + active: false, + activeAt: 1, + metadata: null, + metadataVersion: 0, + agentState: { controlledByUser: false }, + agentStateVersion: 0, + thinking: false, + thinkingAt: 1, + todos: [], + model: null, + modelReasoningEffort: null, + effort: null, + serviceTier: null, + permissionMode: undefined, + collaborationMode: undefined, + ...overrides + } +} + +function deferred() { + let resolve!: (value: T) => void + let reject!: (error: unknown) => void + const promise = new Promise((promiseResolve, promiseReject) => { + resolve = promiseResolve + reject = promiseReject + }) + return { promise, resolve, reject } +} + +describe('ApiSessionClient lazy materialization', () => { + it('does not connect or materialize without a real user message', async () => { + socketHarness.sockets.length = 0 + const materialize = vi.fn(async () => createSession()) + const client = new ApiSessionClient('token', createSession(), { materialize }) + + client.updateMetadata(() => ({ path: '/tmp/project', host: 'localhost', codexSessionId: 'codex-thread' })) + client.sendSessionEvent({ type: 'ready' }) + client.keepAlive(false, 'local') + await client.flush({ timeoutMs: 100 }) + + expect(client.getState()).toBe('pending') + expect(materialize).not.toHaveBeenCalled() + expect(socketHarness.sockets[0]?.connectCalls).toBe(0) + expect(socketHarness.sockets[0]?.emitted).toEqual([]) + client.close() + }) + + it('materializes on the first user message and replays queued events', async () => { + socketHarness.sockets.length = 0 + const materialize = vi.fn(async (snapshot) => createSession({ + namespace: 'default', + metadata: snapshot.metadata, + metadataVersion: 1, + agentState: snapshot.agentState, + agentStateVersion: 1 + })) + const client = new ApiSessionClient('token', createSession(), { materialize }) + client.updateMetadata(() => ({ path: '/tmp/project', host: 'localhost', codexSessionId: 'codex-thread' })) + client.sendSessionEvent({ type: 'ready' }) + + client.sendUserMessage('hello') + expect(await client.materialize()).toBe(true) + + expect(materialize).toHaveBeenCalledWith({ + metadata: { path: '/tmp/project', host: 'localhost', codexSessionId: 'codex-thread' }, + agentState: { controlledByUser: false } + }, expect.any(AbortSignal)) + expect(client.getState()).toBe('active') + expect(socketHarness.sockets[0]?.connectCalls).toBe(1) + expect(socketHarness.sockets[0]?.emitted.map((entry) => entry.event)).toEqual([ + 'message', + 'message', + 'session-alive' + ]) + client.close() + }) + + it('materializes on non-text user activity and preserves following agent events', async () => { + socketHarness.sockets.length = 0 + const pendingMaterialization = deferred() + const materialize = vi.fn(async () => await pendingMaterialization.promise) + const client = new ApiSessionClient('token', createSession(), { materialize }) + + client.notifyUserActivity() + client.sendAgentMessage({ type: 'message', message: 'image response' }) + expect(client.getState()).toBe('materializing') + + pendingMaterialization.resolve(createSession({ namespace: 'default' })) + expect(await client.materialize()).toBe(true) + + expect(materialize).toHaveBeenCalledTimes(1) + const messages = socketHarness.sockets[0]?.emitted.filter((entry) => entry.event === 'message') + expect(messages).toHaveLength(1) + client.close() + }) + + it('preserves all replayed transcript messages while materialization is in flight', async () => { + socketHarness.sockets.length = 0 + const pendingMaterialization = deferred() + const client = new ApiSessionClient('token', createSession(), { + materialize: async () => await pendingMaterialization.promise + }) + const expectedMessages: string[] = [] + + for (let index = 0; index < 150; index += 1) { + const userMessage = `user-${index}` + const agentMessage = `agent-${index}` + expectedMessages.push(userMessage, agentMessage) + client.sendUserMessage(userMessage) + client.sendAgentMessage({ type: 'message', message: agentMessage }) + } + + pendingMaterialization.resolve(createSession({ namespace: 'default' })) + expect(await client.materialize()).toBe(true) + + const emittedMessages = socketHarness.sockets[0]?.emitted + .filter((entry) => entry.event === 'message') + .map((entry) => { + const payload = entry.args[0] as { + message: { + role: 'user' | 'agent' + content: { text?: string; data?: { message?: string } } + } + } + return payload.message.role === 'user' + ? payload.message.content.text + : payload.message.content.data?.message + }) + + expect(emittedMessages).toEqual(expectedMessages) + client.close() + }) + + it('drains in-flight materialization and initial socket delivery before closing', async () => { + socketHarness.sockets.length = 0 + const pendingMaterialization = deferred() + const client = new ApiSessionClient('token', createSession(), { + materialize: async () => await pendingMaterialization.promise + }) + const socket = socketHarness.sockets[0] + if (!socket) throw new Error('expected socket') + socket.connectImmediately = false + + client.sendUserMessage('persist me') + client.sendAgentMessage({ type: 'message', message: 'persist response' }) + client.sendSessionDeath('completed') + + let flushed = false + const flushTask = client.flush({ timeoutMs: 1_000 }).then(() => { + flushed = true + }) + await Promise.resolve() + expect(flushed).toBe(false) + + pendingMaterialization.resolve(createSession({ namespace: 'default' })) + await vi.waitFor(() => expect(socket.connectCalls).toBe(1)) + expect(flushed).toBe(false) + + socket.triggerConnectError() + await Promise.resolve() + expect(flushed).toBe(false) + + socket.triggerConnect() + await flushTask + + expect(socket.emitted.map((entry) => entry.event)).toEqual([ + 'message', + 'message', + 'session-end', + 'session-alive' + ]) + client.close() + }) + + it('skips materialization backoff and performs one final attempt during shutdown drain', async () => { + socketHarness.sockets.length = 0 + const materialize = vi.fn(async () => { + if (materialize.mock.calls.length === 1) { + throw Object.assign(new Error('hub unavailable'), { isAxiosError: true }) + } + return createSession({ namespace: 'default' }) + }) + const client = new ApiSessionClient('token', createSession(), { materialize }) + + client.sendUserMessage('hello') + await vi.waitFor(() => expect(materialize).toHaveBeenCalledTimes(1)) + await Promise.resolve() + + await client.flush({ timeoutMs: 500 }) + + expect(materialize).toHaveBeenCalledTimes(2) + expect(client.getState()).toBe('active') + expect(socketHarness.sockets[0]?.emitted.some((entry) => entry.event === 'message')).toBe(true) + client.close() + }) + + it('aborts in-flight materialization when closed', async () => { + socketHarness.sockets.length = 0 + const observedSignals: AbortSignal[] = [] + const materialize = vi.fn(async (_snapshot, signal: AbortSignal) => { + observedSignals.push(signal) + return await new Promise((_resolve, reject) => { + signal.addEventListener('abort', () => reject(new Error('aborted')), { once: true }) + }) + }) + const client = new ApiSessionClient('token', createSession(), { materialize }) + + const task = client.materialize() + await Promise.resolve() + client.close() + + expect(await task).toBe(false) + expect(observedSignals[0]?.aborted).toBe(true) + expect(client.getState()).toBe('closed') + }) + + it('reconnects a disconnected active session during final flush', async () => { + socketHarness.sockets.length = 0 + const client = new ApiSessionClient('token', createSession({ namespace: 'default' })) + const socket = socketHarness.sockets[0] + if (!socket) throw new Error('expected socket') + socket.connected = false + socket.connectImmediately = false + client.sendSessionDeath('completed') + + let flushed = false + const flushTask = client.flush({ timeoutMs: 500 }).then(() => { + flushed = true + }) + await vi.waitFor(() => expect(socket.connectCalls).toBe(2)) + expect(flushed).toBe(false) + + socket.triggerConnect() + await flushTask + + expect(socket.emitted.some((entry) => entry.event === 'session-end')).toBe(true) + client.close() + }) +}) describe('isExternalUserMessage', () => { const baseUserMsg = { diff --git a/cli/src/api/apiSession.ts b/cli/src/api/apiSession.ts index cd98d69a..3e98814a 100644 --- a/cli/src/api/apiSession.ts +++ b/cli/src/api/apiSession.ts @@ -152,6 +152,72 @@ export class IncomingMessageFilter { } } +export type ApiSessionClientState = 'pending' | 'materializing' | 'active' | 'closed' + +export type PendingSessionSnapshot = { + metadata: Metadata | null + agentState: AgentState | null +} + +export type ApiSessionClientOptions = { + materialize?: (snapshot: PendingSessionSnapshot, signal: AbortSignal) => Promise + onMaterialized?: (session: Session, snapshot: PendingSessionSnapshot) => void +} + +type PendingOutboundEvent = { + emit: () => void + retention: 'lossless' | 'droppable' +} + +const MAX_PENDING_DROPPABLE_EVENTS = 256 +const MATERIALIZATION_RETRY_MIN_MS = 1_000 +const MATERIALIZATION_RETRY_MAX_MS = 30_000 + +function isTransientMaterializationError(error: unknown): boolean { + if (!axios.isAxiosError(error)) { + return false + } + if (!error.response) { + return true + } + const status = error.response.status + return status === 408 || status === 425 || status === 429 || status >= 500 +} + +async function waitForAbortableDelay( + ms: number, + signal: AbortSignal, + interruptSignal?: AbortSignal +): Promise { + if (signal.aborted || interruptSignal?.aborted) { + return false + } + + return await new Promise((resolve) => { + let settled = false + const finish = (completed: boolean) => { + if (settled) return + settled = true + clearTimeout(timeout) + signal.removeEventListener('abort', onAbort) + interruptSignal?.removeEventListener('abort', onAbort) + resolve(completed) + } + const timeout = setTimeout(() => { + finish(true) + }, ms) + const onAbort = () => { + finish(false) + } + signal.addEventListener('abort', onAbort, { once: true }) + interruptSignal?.addEventListener('abort', onAbort, { once: true }) + }) +} + +function hasSameJsonValue(left: unknown, right: unknown): boolean { + return JSON.stringify(left) === JSON.stringify(right) +} + export class ApiSessionClient extends EventEmitter { private readonly token: string readonly sessionId: string @@ -171,8 +237,20 @@ export class ApiSessionClient extends EventEmitter { private readonly terminalManager: TerminalManager private agentStateLock = new AsyncLock() private metadataLock = new AsyncLock() + private state: ApiSessionClientState + private readonly materializer?: ApiSessionClientOptions['materialize'] + private readonly onMaterialized?: ApiSessionClientOptions['onMaterialized'] + private materializationTask: Promise | null = null + private materializationAbortController: AbortController | null = null + private materializationRetryAbortController: AbortController | null = null + private materializationDrainRequested = false + private awaitingMaterializedConnection = false + private metadataChangedDuringAttempt = false + private agentStateChangedDuringAttempt = false + private readonly pendingOutboundEvents: PendingOutboundEvent[] = [] + private didWarnPendingQueueFull = false - constructor(token: string, session: Session) { + constructor(token: string, session: Session, options: ApiSessionClientOptions = {}) { super() this.token = token this.sessionId = session.id @@ -180,6 +258,9 @@ export class ApiSessionClient extends EventEmitter { this.metadataVersion = session.metadataVersion this.agentState = session.agentState this.agentStateVersion = session.agentStateVersion + this.materializer = options.materialize + this.onMaterialized = options.onMaterialized + this.state = this.materializer ? 'pending' : 'active' this.rpcHandlerManager = new RpcHandlerManager({ scopePrefix: this.sessionId, @@ -217,6 +298,7 @@ export class ApiSessionClient extends EventEmitter { this.socket.on('connect', () => { logger.debug('Socket connected successfully') + this.awaitingMaterializedConnection = false this.rpcHandlerManager.onSocketConnect(this.socket) if (this.hasConnectedOnce) { this.needsBackfill = true @@ -332,7 +414,186 @@ export class ApiSessionClient extends EventEmitter { } }) - this.socket.connect() + if (this.state === 'active') { + this.socket.connect() + } + } + + getState(): ApiSessionClientState { + return this.state + } + + isPending(): boolean { + return this.state === 'pending' || this.state === 'materializing' + } + + private isClosed(): boolean { + return this.state === 'closed' + } + + async materialize(): Promise { + if (this.state === 'active') { + return true + } + if (this.state === 'closed' || !this.materializer || this.materializationDrainRequested) { + return false + } + if (this.materializationTask) { + return await this.materializationTask + } + + this.state = 'materializing' + const abortController = new AbortController() + this.materializationAbortController = abortController + this.materializationTask = this.runMaterializationLoop(abortController.signal) + .finally(() => { + this.materializationTask = null + if (this.materializationAbortController === abortController) { + this.materializationAbortController = null + } + }) + return await this.materializationTask + } + + private async runMaterializationLoop(signal: AbortSignal): Promise { + let retryDelayMs = MATERIALIZATION_RETRY_MIN_MS + let finalDrainAttemptStarted = false + + while (!signal.aborted && !this.isClosed()) { + this.metadataChangedDuringAttempt = false + this.agentStateChangedDuringAttempt = false + const snapshot: PendingSessionSnapshot = { + metadata: this.metadata, + agentState: this.agentState + } + + try { + const materialized = await this.materializer!(snapshot, signal) + if (signal.aborted || this.isClosed()) { + return false + } + + const latestMetadata = this.metadata + const latestAgentState = this.agentState + const shouldSyncMetadata = this.metadataChangedDuringAttempt + || !hasSameJsonValue(materialized.metadata, latestMetadata) + const shouldSyncAgentState = this.agentStateChangedDuringAttempt + || !hasSameJsonValue(materialized.agentState, latestAgentState) + + this.metadata = materialized.metadata + this.metadataVersion = materialized.metadataVersion + this.agentState = materialized.agentState + this.agentStateVersion = materialized.agentStateVersion + this.state = 'active' + + if (shouldSyncMetadata && latestMetadata) { + this.updateMetadata(() => latestMetadata) + } + if (shouldSyncAgentState && latestAgentState) { + this.updateAgentState(() => latestAgentState) + } + + const pendingEvents = this.pendingOutboundEvents.splice(0) + this.awaitingMaterializedConnection = pendingEvents.length > 0 + || shouldSyncMetadata + || shouldSyncAgentState + for (const pendingEvent of pendingEvents) { + pendingEvent.emit() + } + this.socket.connect() + try { + this.onMaterialized?.(materialized, { + metadata: latestMetadata, + agentState: latestAgentState + }) + } catch (error) { + logger.debug(`[API] Post-materialization callback failed for ${this.sessionId}`, error) + } + logger.debug(`[API] Materialized pending session ${this.sessionId}`) + return true + } catch (error) { + if (signal.aborted || this.isClosed()) { + return false + } + if (!isTransientMaterializationError(error)) { + this.state = 'pending' + logger.warn(`[API] Failed to materialize pending session ${this.sessionId}`, error) + return false + } + if (this.materializationDrainRequested) { + if (finalDrainAttemptStarted) { + this.state = 'pending' + return false + } + finalDrainAttemptStarted = true + logger.debug(`[API] Retrying materialization once during final drain for ${this.sessionId}`) + continue + } + + logger.debug( + `[API] Hub unavailable while materializing ${this.sessionId}; retrying in ${retryDelayMs}ms`, + error + ) + const retryAbortController = new AbortController() + this.materializationRetryAbortController = retryAbortController + const completedDelay = await waitForAbortableDelay( + retryDelayMs, + signal, + retryAbortController.signal + ) + if (this.materializationRetryAbortController === retryAbortController) { + this.materializationRetryAbortController = null + } + if (!completedDelay) { + if (signal.aborted || this.isClosed()) { + return false + } + if (this.materializationDrainRequested && !finalDrainAttemptStarted) { + finalDrainAttemptStarted = true + logger.debug(`[API] Skipping materialization backoff during final drain for ${this.sessionId}`) + continue + } + this.state = 'pending' + return false + } + retryDelayMs = Math.min(retryDelayMs * 2, MATERIALIZATION_RETRY_MAX_MS) + } + } + + return false + } + + private emitOrQueue( + emit: () => void, + retention: PendingOutboundEvent['retention'] = 'lossless' + ): void { + if (this.state === 'active') { + emit() + return + } + if (this.state === 'closed') { + return + } + + if (retention === 'droppable') { + const droppableCount = this.pendingOutboundEvents.reduce( + (count, event) => count + (event.retention === 'droppable' ? 1 : 0), + 0 + ) + if (droppableCount >= MAX_PENDING_DROPPABLE_EVENTS) { + const oldestDroppableIndex = this.pendingOutboundEvents.findIndex( + (event) => event.retention === 'droppable' + ) + if (oldestDroppableIndex >= 0) { + this.pendingOutboundEvents.splice(oldestDroppableIndex, 1) + } + if (!this.didWarnPendingQueueFull) { + this.didWarnPendingQueueFull = true + logger.warn(`[API] Pending control event queue full for ${this.sessionId}; dropping oldest control event`) + } + } + } + this.pendingOutboundEvents.push({ emit, retention }) } onUserMessage(callback: (data: UserMessage, localId?: string) => void): void { @@ -482,9 +743,11 @@ export class ApiSessionClient extends EventEmitter { } } - this.socket.emit('message', { - sid: this.sessionId, - message: content + this.emitOrQueue(() => { + this.socket.emit('message', { + sid: this.sessionId, + message: content + }) }) if (body.type === 'summary' && 'summary' in body && 'leafUuid' in body) { @@ -515,10 +778,17 @@ export class ApiSessionClient extends EventEmitter { } } - this.socket.emit('message', { - sid: this.sessionId, - message: content + this.emitOrQueue(() => { + this.socket.emit('message', { + sid: this.sessionId, + message: content + }) }) + this.notifyUserActivity() + } + + notifyUserActivity(): void { + void this.materialize() } sendAgentMessage(body: unknown): void { @@ -532,9 +802,11 @@ export class ApiSessionClient extends EventEmitter { sentFrom: 'cli' } } - this.socket.emit('message', { - sid: this.sessionId, - message: content + this.emitOrQueue(() => { + this.socket.emit('message', { + sid: this.sessionId, + message: content + }) }) } @@ -559,10 +831,12 @@ export class ApiSessionClient extends EventEmitter { } } - this.socket.emit('message', { - sid: this.sessionId, - message: content - }) + this.emitOrQueue(() => { + this.socket.emit('message', { + sid: this.sessionId, + message: content + }) + }, event.type === 'message' ? 'lossless' : 'droppable') } keepAlive( @@ -577,6 +851,9 @@ export class ApiSessionClient extends EventEmitter { collaborationMode?: SessionCollaborationMode } ): void { + if (this.state !== 'active') { + return + } this.socket.volatile.emit('session-alive', { sid: this.sessionId, time: Date.now(), @@ -588,10 +865,12 @@ export class ApiSessionClient extends EventEmitter { /** Hub waits for this before mergeSessions on Cursor ACP reopen (tiann/hapi#939). */ emitSessionReady(): void { - this.socket.emit('session-ready', { - sid: this.sessionId, - time: Date.now() - }) + this.emitOrQueue(() => { + this.socket.emit('session-ready', { + sid: this.sessionId, + time: Date.now() + }) + }, 'droppable') } emitMessagesConsumed(localIds: string[], options?: { clearQueuedThinkingGrace?: boolean }): void { @@ -609,15 +888,28 @@ export class ApiSessionClient extends EventEmitter { if (options?.clearQueuedThinkingGrace) { payload.clearQueuedThinkingGrace = true } - this.socket.emit('messages-consumed', payload) + this.emitOrQueue(() => this.socket.emit('messages-consumed', payload)) } sendSessionDeath(reason?: SessionEndReason): void { - void cleanupUploadDir(this.sessionId) - this.socket.emit('session-end', { sid: this.sessionId, time: Date.now(), reason }) + if (this.state === 'active') { + void cleanupUploadDir(this.sessionId) + } + this.emitOrQueue(() => { + this.socket.emit('session-end', { sid: this.sessionId, time: Date.now(), reason }) + }) } updateMetadata(handler: (metadata: Metadata) => Metadata): void { + if (this.state !== 'active') { + if (this.state === 'closed') return + const current = this.metadata ?? ({} as Metadata) + this.metadata = handler(current) + if (this.state === 'materializing') { + this.metadataChangedDuringAttempt = true + } + return + } this.metadataLock.inLock(async () => { await backoff(async () => { const current = this.metadata ?? ({} as Metadata) @@ -654,6 +946,15 @@ export class ApiSessionClient extends EventEmitter { } updateAgentState(handler: (state: AgentState) => AgentState): void { + if (this.state !== 'active') { + if (this.state === 'closed') return + const current = this.agentState ?? ({} as AgentState) + this.agentState = handler(current) + if (this.state === 'materializing') { + this.agentStateChangedDuringAttempt = true + } + return + } this.agentStateLock.inLock(async () => { await backoff(async () => { const current = this.agentState ?? ({} as AgentState) @@ -689,39 +990,6 @@ export class ApiSessionClient extends EventEmitter { }) } - private async waitForConnected(timeoutMs: number): Promise { - if (this.socket.connected) { - return true - } - - this.socket.connect() - - return await new Promise((resolve) => { - let settled = false - - const cleanup = () => { - this.socket.off('connect', onConnect) - clearTimeout(timeout) - } - - const onConnect = () => { - if (settled) return - settled = true - cleanup() - resolve(true) - } - - const timeout = setTimeout(() => { - if (settled) return - settled = true - cleanup() - resolve(false) - }, Math.max(0, timeoutMs)) - - this.socket.on('connect', onConnect) - }) - } - private async drainLock(lock: AsyncLock, timeoutMs: number): Promise { if (timeoutMs <= 0) { return false @@ -748,6 +1016,60 @@ export class ApiSessionClient extends EventEmitter { }) } + private async waitForPromise(promise: Promise, timeoutMs: number): Promise { + if (timeoutMs <= 0) { + return false + } + + return await new Promise((resolve) => { + let settled = false + const timeout = setTimeout(() => finish(false), timeoutMs) + const finish = (value: boolean) => { + if (settled) return + settled = true + clearTimeout(timeout) + resolve(value) + } + + promise.then(() => finish(true)).catch(() => finish(true)) + }) + } + + private async waitForConnected(timeoutMs: number): Promise { + if (this.socket.connected) { + return true + } + if (timeoutMs <= 0) { + return false + } + + return await new Promise((resolve) => { + let settled = false + const cleanup = () => { + clearTimeout(timeout) + this.socket.off('connect', onConnect) + } + const finish = (value: boolean) => { + if (settled) return + settled = true + cleanup() + resolve(value) + } + const onConnect = () => finish(true) + const timeout = setTimeout(() => finish(false), timeoutMs) + + this.socket.on('connect', onConnect) + + if (!this.awaitingMaterializedConnection) { + this.socket.connect() + } + + if (this.socket.connected) { + finish(true) + } + }) + } + /** * tiann/hapi#913: wait until any pending `update-metadata` writes have * been acked by the hub (or the timeout elapses). `updateMetadata` is @@ -760,14 +1082,34 @@ export class ApiSessionClient extends EventEmitter { * Returns true when the lock drained, false when the timeout fired. */ async flushMetadata(timeoutMs: number = 5_000): Promise { + if (this.state !== 'active') { + return false + } return await this.drainLock(this.metadataLock, timeoutMs) } async flush(options?: { timeoutMs?: number }): Promise { const deadlineMs = Date.now() + (options?.timeoutMs ?? 5_000) - const remainingMs = () => Math.max(0, deadlineMs - Date.now()) + const materializationTask = this.materializationTask + if (materializationTask) { + this.materializationDrainRequested = true + this.materializationRetryAbortController?.abort() + await this.waitForPromise(materializationTask, remainingMs()) + } + + if (this.state !== 'active') { + return + } + + if (!this.socket.connected) { + const connected = await this.waitForConnected(remainingMs()) + if (!connected) { + return + } + } + await this.drainLock(this.metadataLock, remainingMs()) await this.drainLock(this.agentStateLock, remainingMs()) @@ -775,11 +1117,6 @@ export class ApiSessionClient extends EventEmitter { return } - const connected = await this.waitForConnected(remainingMs()) - if (!connected) { - return - } - const pingTimeoutMs = remainingMs() if (pingTimeoutMs === 0) { return @@ -787,12 +1124,23 @@ export class ApiSessionClient extends EventEmitter { try { await this.socket.timeout(pingTimeoutMs).emitWithAck('ping') + this.awaitingMaterializedConnection = false } catch { // best effort } } close(): void { + if (this.state === 'closed') { + return + } + this.state = 'closed' + this.materializationAbortController?.abort() + this.materializationAbortController = null + this.materializationRetryAbortController?.abort() + this.materializationRetryAbortController = null + this.awaitingMaterializedConnection = false + this.pendingOutboundEvents.length = 0 this.rpcHandlerManager.onSocketDisconnect() this.terminalManager.closeAll() this.socket.disconnect() diff --git a/cli/src/claude/utils/sessionHookForwarder.ts b/cli/src/claude/utils/sessionHookForwarder.ts index 8cc206d3..b34183ac 100644 --- a/cli/src/claude/utils/sessionHookForwarder.ts +++ b/cli/src/claude/utils/sessionHookForwarder.ts @@ -1,5 +1,7 @@ import { request } from 'node:http'; +export const SESSION_HOOK_FORWARD_TIMEOUT_MS = 1_000; + function logError(message: string, error?: unknown): void { const detail = error instanceof Error ? error.message : (error ? String(error) : ''); const suffix = detail ? `: ${detail}` : ''; @@ -93,6 +95,13 @@ export async function runSessionHookForwarder(args: string[]): Promise { let hadError = false; await new Promise((resolve) => { + let settled = false; + let timedOut = false; + const finish = () => { + if (settled) return; + settled = true; + resolve(); + }; const req = request({ host: '127.0.0.1', port, @@ -111,16 +120,25 @@ export async function runSessionHookForwarder(args: string[]): Promise { res.on('error', (error) => { hadError = true; logError('Error reading hook server response', error); - resolve(); + finish(); }); - res.on('end', () => resolve()); + res.on('end', finish); res.resume(); }); req.on('error', (error) => { hadError = true; - logError('Failed to send hook request', error); - resolve(); + if (!timedOut) { + logError('Failed to send hook request', error); + } + finish(); + }); + req.setTimeout(SESSION_HOOK_FORWARD_TIMEOUT_MS, () => { + timedOut = true; + hadError = true; + logError(`Hook request timed out after ${SESSION_HOOK_FORWARD_TIMEOUT_MS}ms`); + req.destroy(); + finish(); }); req.end(body); }); diff --git a/cli/src/claude/utils/startHookServer.ts b/cli/src/claude/utils/startHookServer.ts index 02073edf..096ac6f1 100644 --- a/cli/src/claude/utils/startHookServer.ts +++ b/cli/src/claude/utils/startHookServer.ts @@ -107,7 +107,6 @@ export async function startHookServer(options: HookServerOptions): Promise { + try { + onSessionHook(sessionId, data); + } catch (error) { + logger.debug('[hookServer] Error dispatching session hook:', error); + } + }); } catch (error) { clearTimeout(timeout); if (timedOut) { diff --git a/cli/src/codex/codexLocalLauncher.test.ts b/cli/src/codex/codexLocalLauncher.test.ts index 74e7237a..2594ca5a 100644 --- a/cli/src/codex/codexLocalLauncher.test.ts +++ b/cli/src/codex/codexLocalLauncher.test.ts @@ -70,11 +70,13 @@ function createSessionStub( codexArgs?: string[], path = '/tmp/worktree', initialTranscriptPath: string | null = null, - replayTranscriptHistoryOnStart = false + replayTranscriptHistoryOnStart = false, + pendingClient = false ) { const sessionEvents: Array<{ type: string; message?: string }> = []; const userMessages: string[] = []; const agentMessages: unknown[] = []; + let userActivityCount = 0; let localLaunchFailure: { message: string; exitReason: 'switch' | 'exit' } | null = null; let sessionId: string | null = null; let transcriptPath: string | null = initialTranscriptPath; @@ -94,6 +96,7 @@ function createSessionStub( codexArgs, replayTranscriptHistoryOnStart, client: { + isPending: () => pendingClient, rpcHandlerManager: { registerHandler: () => {} } @@ -130,6 +133,9 @@ function createSessionStub( sendUserMessage: (message: string) => { userMessages.push(message); }, + notifyUserActivity: () => { + userActivityCount += 1; + }, sendAgentMessage: (message: unknown) => { agentMessages.push(message); }, @@ -138,6 +144,7 @@ function createSessionStub( sessionEvents, userMessages, agentMessages, + getUserActivityCount: () => userActivityCount, getLocalLaunchFailure: () => localLaunchFailure }; } @@ -354,6 +361,68 @@ describe('codexLocalLauncher', () => { }); }); + it('falls back to fresh transcript activity when SessionStart does not arrive', async () => { + const originalCodexHome = process.env.CODEX_HOME; + process.env.CODEX_HOME = tempDir; + const now = new Date(); + const sessionDirectory = join( + tempDir, + 'sessions', + String(now.getUTCFullYear()), + String(now.getUTCMonth() + 1).padStart(2, '0'), + String(now.getUTCDate()).padStart(2, '0') + ); + await mkdir(sessionDirectory, { recursive: true }); + const transcriptPath = join(sessionDirectory, 'rollout-fallback-thread.jsonl'); + const { session, userMessages } = createSessionStub( + 'default', + ['--cd', '/tmp/effective-codex-cwd'], + '/tmp/worktree', + null, + true, + true + ); + let releaseRunBarrier: (() => void) | undefined; + harness.runBarrier = new Promise((resolve) => { + releaseRunBarrier = resolve; + }); + + try { + const launcherPromise = codexLocalLauncher(session as never); + await vi.waitFor(() => expect(harness.launches).toHaveLength(1)); + expect(session.sessionId).toBeNull(); + + await writeFile(transcriptPath, [ + JSON.stringify({ + type: 'session_meta', + payload: { id: 'fallback-thread', cwd: '/tmp/effective-codex-cwd' } + }), + JSON.stringify({ + timestamp: new Date().toISOString(), + type: 'event_msg', + payload: { type: 'user_message', message: 'fallback prompt' } + }) + ].join('\n') + '\n'); + + await vi.waitFor( + () => expect(session.sessionId).toBe('fallback-thread'), + { timeout: 3_000, interval: 50 } + ); + if (releaseRunBarrier) releaseRunBarrier(); + await launcherPromise; + + expect(session.transcriptPath).toBe(transcriptPath); + expect(userMessages).toContain('fallback prompt'); + } finally { + if (releaseRunBarrier) releaseRunBarrier(); + if (originalCodexHome === undefined) { + delete process.env.CODEX_HOME; + } else { + process.env.CODEX_HOME = originalCodexHome; + } + } + }); + it('replays existing transcript messages when importing a Codex thread into a new Hapi session', async () => { const transcriptPath = join(tempDir, 'codex-import-transcript.jsonl'); const { session, agentMessages } = createSessionStub('default', undefined, '/tmp/worktree', null, true); @@ -392,7 +461,7 @@ describe('codexLocalLauncher', () => { it('replays existing response_item chat messages when importing a Codex thread into a new Hapi session', async () => { const transcriptPath = join(tempDir, 'codex-import-response-item-transcript.jsonl'); - const { session, userMessages, agentMessages } = createSessionStub('default', undefined, '/tmp/worktree', null, true); + const { session, userMessages, agentMessages, getUserActivityCount } = createSessionStub('default', undefined, '/tmp/worktree', null, true); let releaseRunBarrier: (() => void) | undefined; harness.runBarrier = new Promise((resolve) => { releaseRunBarrier = resolve; @@ -417,6 +486,14 @@ describe('codexLocalLauncher', () => { role: 'assistant', content: [{ type: 'output_text', text: 'old response_item assistant message' }] } + }), + JSON.stringify({ + type: 'response_item', + payload: { + type: 'message', + role: 'user', + content: [{ type: 'input_image', image_url: 'data:image/png;base64,abc' }] + } }) ].join('\n') + '\n' ); @@ -435,6 +512,7 @@ describe('codexLocalLauncher', () => { await launcherPromise; expect(userMessages).toContain('old response_item user message'); + expect(getUserActivityCount()).toBe(1); expect(agentMessages).toContainEqual({ type: 'message', message: 'old response_item assistant message', diff --git a/cli/src/codex/codexLocalLauncher.ts b/cli/src/codex/codexLocalLauncher.ts index a02e23ed..7df3e955 100644 --- a/cli/src/codex/codexLocalLauncher.ts +++ b/cli/src/codex/codexLocalLauncher.ts @@ -1,4 +1,5 @@ import { logger } from '@/ui/logger'; +import { resolve } from 'node:path'; import { startHookServer } from '@/claude/utils/startHookServer'; import { codexLocal } from './codexLocal'; import type { ReasoningEffort } from './appServerTypes'; @@ -6,9 +7,10 @@ import { CodexSession } from './session'; import { createCodexSessionScanner, type CodexSessionScanner } from './utils/codexSessionScanner'; import { convertCodexEvent } from './utils/codexEventConverter'; import { buildHapiMcpBridge } from './utils/buildHapiMcpBridge'; -import { stripCodexCliOverrides } from './utils/codexCliOverrides'; +import { parseCodexCliOverrides, stripCodexCliOverrides } from './utils/codexCliOverrides'; import { buildCodexPermissionModeCliArgs } from './utils/permissionModeConfig'; import { BaseLocalLauncher } from '@/modules/common/launcher/BaseLocalLauncher'; +import { createCodexTranscriptLocator, type CodexTranscriptLocator } from './utils/codexTranscriptLocator'; export async function codexLocalLauncher(session: CodexSession): Promise<'switch' | 'exit'> { const resumeSessionId = session.sessionId; @@ -18,6 +20,7 @@ export async function codexLocalLauncher(session: CodexSession): Promise<'switch let hookReady = false; let shuttingDown = false; let pendingScannerSetup: Promise | null = null; + let transcriptLocator: CodexTranscriptLocator | null = null; const permissionMode = session.getPermissionMode(); const managedPermissionMode = permissionMode === 'read-only' || permissionMode === 'safe-yolo' || permissionMode === 'yolo' ? permissionMode @@ -28,6 +31,8 @@ export async function codexLocalLauncher(session: CodexSession): Promise<'switch ...stripCodexCliOverrides(session.codexArgs) ] : session.codexArgs; + const cwdOverride = parseCodexCliOverrides(session.codexArgs).cwd; + const effectiveCodexCwd = cwdOverride ? resolve(session.path, cwdOverride) : session.path; // Start hapi hub for MCP bridge (same as remote mode) const { server: happyServer, mcpServers } = await buildHapiMcpBridge(session.client); @@ -103,6 +108,8 @@ export async function codexLocalLauncher(session: CodexSession): Promise<'switch } if (converted?.userMessage) { session.sendUserMessage(converted.userMessage); + } else if (converted?.userActivity) { + session.notifyUserActivity(); } if (converted?.message) { session.sendAgentMessage(converted.message); @@ -142,6 +149,12 @@ export async function codexLocalLauncher(session: CodexSession): Promise<'switch const hookSource = typeof data.source === 'string' ? data.source : null; const shouldAllowSessionSwitch = hookSource === 'clear'; + if (transcriptPath) { + const activeLocator = transcriptLocator; + transcriptLocator = null; + void activeLocator?.cleanup(); + } + if (!transcriptPath) { handleSessionFound(sessionId, shouldAllowSessionSwitch); return; @@ -160,6 +173,27 @@ export async function codexLocalLauncher(session: CodexSession): Promise<'switch }); logger.debug(`[codex-local]: Started Codex SessionStart hook server on port ${hookServer.port}`); + if (session.client.isPending()) { + const createdLocator = createCodexTranscriptLocator({ + cwd: effectiveCodexCwd, + startupTimestampMs: Date.now(), + resumeSessionId, + onLocated: ({ sessionId, transcriptPath }) => { + if (shuttingDown || hookReady || primaryTranscriptPath) { + return; + } + transcriptLocator = null; + bindPrimarySession(sessionId, transcriptPath); + }, + onAmbiguous: (paths) => { + transcriptLocator = null; + logger.warn(`[codex-local]: Transcript fallback was ambiguous (${paths.length} active candidates)`); + } + }); + transcriptLocator = createdLocator; + await createdLocator.ready; + } + const launcher = new BaseLocalLauncher({ label: 'codex-local', failureLabel: 'Local Codex process failed', @@ -204,6 +238,9 @@ export async function codexLocalLauncher(session: CodexSession): Promise<'switch shuttingDown = true; session.removeTranscriptPathCallback(handleTranscriptPathCallback); hookServer.stop(); + const activeLocator = transcriptLocator; + transcriptLocator = null; + void activeLocator?.cleanup(); if (pendingScannerSetup) { await pendingScannerSetup; } diff --git a/cli/src/codex/runCodex.test.ts b/cli/src/codex/runCodex.test.ts index 7c138bdd..8b2ab0a2 100644 --- a/cli/src/codex/runCodex.test.ts +++ b/cli/src/codex/runCodex.test.ts @@ -32,6 +32,14 @@ vi.mock('@/agent/sessionFactory', () => ({ sessionInfo: harness.sessionInfo } }), + bootstrapLazySession: vi.fn(async (options: Record) => { + harness.bootstrapArgs.push({ ...options, lazy: true }) + return { + api: {}, + session: harness.session, + sessionInfo: harness.sessionInfo + } + }), bootstrapExistingSession: vi.fn(async (options: Record) => { harness.bootstrapArgs.push(options) return { @@ -202,6 +210,27 @@ describe('runCodex', () => { expect(mockCodexSession.setServiceTier).not.toHaveBeenCalled() }) + it('uses lazy bootstrap for a fresh terminal launch', async () => { + await runCodexImpl({ workingDirectory: '/tmp/project' }) + + expect(harness.bootstrapArgs[0]).toEqual(expect.objectContaining({ + workingDirectory: '/tmp/project', + lazy: true + })) + expect(harness.loopArgs[0]).toEqual(expect.objectContaining({ + replayTranscriptHistoryOnStart: true + })) + }) + + it('keeps eager bootstrap for runner launches', async () => { + await runCodexImpl({ + startedBy: 'runner', + workingDirectory: '/tmp/project' + }) + + expect(harness.bootstrapArgs[0]).not.toHaveProperty('lazy') + }) + it('replays transcript history when attaching a new Hapi session to an existing Codex thread', async () => { await runCodexImpl({ workingDirectory: '/tmp/project', diff --git a/cli/src/codex/runCodex.ts b/cli/src/codex/runCodex.ts index de907c22..fe3a0f70 100644 --- a/cli/src/codex/runCodex.ts +++ b/cli/src/codex/runCodex.ts @@ -7,7 +7,7 @@ import { registerKillSessionHandler } from '@/claude/registerKillSessionHandler' import type { AgentState } from '@/api/types'; import type { CodexSession } from './session'; import { parseCodexCliOverrides } from './utils/codexCliOverrides'; -import { bootstrapExistingSession, bootstrapSession } from '@/agent/sessionFactory'; +import { bootstrapExistingSession, bootstrapLazySession, bootstrapSession } from '@/agent/sessionFactory'; import { registerLocalHandoffHandler } from '@/agent/localHandoff'; import { createModeChangeHandler, createRunnerLifecycle, setControlledByUser } from '@/agent/runnerLifecycle'; import { isPermissionModeAllowedForFlavor } from '@hapi/protocol'; @@ -44,6 +44,7 @@ export async function runCodex(opts: { let state: AgentState = { controlledByUser: false }; + const useLazyBootstrap = !opts.existingSessionId && startedBy === 'terminal'; const bootstrap = opts.existingSessionId ? await bootstrapExistingSession({ sessionId: opts.existingSessionId, @@ -51,7 +52,7 @@ export async function runCodex(opts: { startedBy, workingDirectory }) - : await bootstrapSession({ + : await (useLazyBootstrap ? bootstrapLazySession : bootstrapSession)({ flavor: 'codex', startedBy, workingDirectory, @@ -77,7 +78,7 @@ export async function runCodex(opts: { const sessionWrapperRef: { current: CodexSession | null } = { current: null }; // 中文注释：当用户直接把现成的 Codex thread 导入到一个全新的 Hapi 会话时， // 需要在首次附着 transcript 时回放已有历史；恢复已有 Hapi 会话时则保持原来的增量模式，避免重复灌入旧消息。 - const replayTranscriptHistoryOnStart = Boolean(opts.resumeSessionId && !opts.existingSessionId); + const replayTranscriptHistoryOnStart = useLazyBootstrap || Boolean(opts.resumeSessionId && !opts.existingSessionId); let currentPermissionMode: PermissionMode = opts.permissionMode ?? 'default'; let currentModel = opts.model; diff --git a/cli/src/codex/session.ts b/cli/src/codex/session.ts index c8892419..1fdf985f 100644 --- a/cli/src/codex/session.ts +++ b/cli/src/codex/session.ts @@ -144,6 +144,10 @@ export class CodexSession extends AgentSessionBase { this.client.sendUserMessage(text); }; + notifyUserActivity = (): void => { + this.client.notifyUserActivity(); + }; + sendSessionEvent = (event: Parameters[0]): void => { this.client.sendSessionEvent(event); }; diff --git a/cli/src/codex/utils/codexCliOverrides.test.ts b/cli/src/codex/utils/codexCliOverrides.test.ts index c236bf86..06d14960 100644 --- a/cli/src/codex/utils/codexCliOverrides.test.ts +++ b/cli/src/codex/utils/codexCliOverrides.test.ts @@ -41,6 +41,16 @@ describe('parseCodexCliOverrides', () => { expect(parseCodexCliOverrides(['-a', 'untrusted', '-a', 'on-failure'])).toEqual({ approvalPolicy: 'on-failure' }); + + expect(parseCodexCliOverrides(['-C', 'first', '--cd=second'])).toEqual({ + cwd: 'second' + }); + }); + + it('parses cwd overrides before the argument terminator', () => { + expect(parseCodexCliOverrides(['--cd', '../project'])).toEqual({ cwd: '../project' }); + expect(parseCodexCliOverrides(['-C=/tmp/project'])).toEqual({ cwd: '/tmp/project' }); + expect(parseCodexCliOverrides(['--', '--cd', '/tmp/ignored'])).toEqual({}); }); it('ignores invalid values and stops at terminator', () => { diff --git a/cli/src/codex/utils/codexCliOverrides.ts b/cli/src/codex/utils/codexCliOverrides.ts index f2d6eeed..6e06014c 100644 --- a/cli/src/codex/utils/codexCliOverrides.ts +++ b/cli/src/codex/utils/codexCliOverrides.ts @@ -1,6 +1,7 @@ export type CodexCliOverrides = { sandbox?: 'read-only' | 'workspace-write' | 'danger-full-access'; approvalPolicy?: 'untrusted' | 'on-failure' | 'on-request' | 'never'; + cwd?: string; }; const SANDBOX_VALUES = new Set([ @@ -46,6 +47,23 @@ export function parseCodexCliOverrides(args?: string[]): CodexCliOverrides { continue; } + if (arg === '-C' || arg === '--cd') { + const value = args[i + 1]; + if (value && value !== '--') { + overrides.cwd = value; + i += 1; + } + continue; + } + + if (arg.startsWith('--cd=') || arg.startsWith('-C=')) { + const value = arg.slice(arg.indexOf('=') + 1); + if (value) { + overrides.cwd = value; + } + continue; + } + if (arg === '-s' || arg === '--sandbox') { const value = args[i + 1]; if (SANDBOX_VALUES.has(value as CodexCliOverrides['sandbox'])) { diff --git a/cli/src/codex/utils/codexEventConverter.test.ts b/cli/src/codex/utils/codexEventConverter.test.ts index 13afe113..adbc3bd3 100644 --- a/cli/src/codex/utils/codexEventConverter.test.ts +++ b/cli/src/codex/utils/codexEventConverter.test.ts @@ -43,10 +43,24 @@ describe('convertCodexEvent', () => { }); expect(result).toEqual({ - userMessage: 'hello from response_item user' + userMessage: 'hello from response_item user', + userActivity: true }); }); + it('marks image-only response_item messages as user activity', () => { + const result = convertCodexEvent({ + type: 'response_item', + payload: { + type: 'message', + role: 'user', + content: [{ type: 'input_image', image_url: 'data:image/png;base64,abc' }] + } + }); + + expect(result).toEqual({ userActivity: true }); + }); + it('converts response_item assistant messages', () => { const result = convertCodexEvent({ type: 'response_item', diff --git a/cli/src/codex/utils/codexEventConverter.ts b/cli/src/codex/utils/codexEventConverter.ts index efd7394c..39304805 100644 --- a/cli/src/codex/utils/codexEventConverter.ts +++ b/cli/src/codex/utils/codexEventConverter.ts @@ -42,6 +42,7 @@ export type CodexConversionResult = { sessionId?: string; message?: CodexMessage; userMessage?: string; + userActivity?: true; }; function asRecord(value: unknown): Record | null { @@ -146,11 +147,9 @@ export function convertCodexEvent(rawEvent: unknown): CodexConversionResult | nu const message = asString(payloadRecord.message) ?? asString(payloadRecord.text) ?? asString(payloadRecord.content); - if (!message) { - return null; - } return { - userMessage: message + userActivity: true, + ...(message ? { userMessage: message } : {}) }; } @@ -221,13 +220,16 @@ export function convertCodexEvent(rawEvent: unknown): CodexConversionResult | nu if (itemType === 'message') { const role = asString(payloadRecord.role); const text = extractCodexText(payloadRecord.content); - if (!text) { - return null; - } if (role === 'user') { - return { userMessage: text }; + return { + userActivity: true, + ...(text ? { userMessage: text } : {}) + }; } if (role === 'assistant') { + if (!text) { + return null; + } return { message: { type: 'message', diff --git a/cli/src/codex/utils/codexSessionScanner.test.ts b/cli/src/codex/utils/codexSessionScanner.test.ts index d6a7db80..7feddcd1 100644 --- a/cli/src/codex/utils/codexSessionScanner.test.ts +++ b/cli/src/codex/utils/codexSessionScanner.test.ts @@ -163,4 +163,30 @@ describe('codexSessionScanner', () => { expect(events).toHaveLength(1); expect(events[0]?.payload).toEqual({ type: 'agent_message', message: 'after-truncate' }); }); + + it('retries an unterminated final record after it is completed', async () => { + await writeFile( + transcriptPath, + JSON.stringify({ type: 'session_meta', payload: { id: 'session-partial' } }) + '\n' + ); + scanner = await createCodexSessionScanner({ + transcriptPath, + onEvent: (event) => events.push(event) + }); + const event = JSON.stringify({ + type: 'event_msg', + payload: { type: 'agent_message', message: 'completed later' } + }); + const splitAt = Math.floor(event.length / 2); + + await appendFile(transcriptPath, event.slice(0, splitAt)); + await wait(300); + expect(events).toEqual([]); + + await appendFile(transcriptPath, event.slice(splitAt)); + await wait(700); + + expect(events).toHaveLength(1); + expect(events[0]?.payload).toEqual({ type: 'agent_message', message: 'completed later' }); + }); }); diff --git a/cli/src/codex/utils/codexSessionScanner.ts b/cli/src/codex/utils/codexSessionScanner.ts index 06bd6670..2a2be616 100644 --- a/cli/src/codex/utils/codexSessionScanner.ts +++ b/cli/src/codex/utils/codexSessionScanner.ts @@ -122,6 +122,7 @@ class CodexSessionScannerImpl extends BaseSessionScanner { const lines = content.split('\n'); const hasTrailingEmpty = lines.length > 0 && lines[lines.length - 1] === ''; const totalLines = hasTrailingEmpty ? lines.length - 1 : lines.length; + let nextCursor = totalLines; const currentSize = Buffer.byteLength(content); const previousSize = this.fileSizeByPath.get(filePath); let effectiveStartLine = startLine; @@ -145,6 +146,9 @@ class CodexSessionScannerImpl extends BaseSessionScanner { parsed = JSON.parse(line); } catch (error) { logger.debug(`[codex-session-scanner] Failed to parse transcript line ${filePath}:${lineIndex + 1}: ${error}`); + if (!hasTrailingEmpty && lineIndex === totalLines - 1) { + nextCursor = lineIndex; + } continue; } @@ -169,7 +173,7 @@ class CodexSessionScannerImpl extends BaseSessionScanner { return { events, - nextCursor: totalLines + nextCursor }; } diff --git a/cli/src/codex/utils/codexTranscriptLocator.test.ts b/cli/src/codex/utils/codexTranscriptLocator.test.ts new file mode 100644 index 00000000..b633f51b --- /dev/null +++ b/cli/src/codex/utils/codexTranscriptLocator.test.ts @@ -0,0 +1,250 @@ +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { appendFile, mkdir, rm, writeFile } from 'node:fs/promises'; +import { join } from 'node:path'; +import { tmpdir } from 'node:os'; +import { createCodexTranscriptLocator, type CodexTranscriptLocator } from './codexTranscriptLocator'; + +const wait = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +describe('codexTranscriptLocator', () => { + let codexHome: string; + let sessionDirectory: string; + let locator: CodexTranscriptLocator | null = null; + const originalCodexHome = process.env.CODEX_HOME; + + beforeEach(async () => { + codexHome = join(tmpdir(), `codex-transcript-locator-${Date.now()}-${Math.random()}`); + const now = new Date(); + sessionDirectory = join( + codexHome, + 'sessions', + String(now.getUTCFullYear()), + String(now.getUTCMonth() + 1).padStart(2, '0'), + String(now.getUTCDate()).padStart(2, '0') + ); + await mkdir(sessionDirectory, { recursive: true }); + process.env.CODEX_HOME = codexHome; + }); + + afterEach(async () => { + await locator?.cleanup(); + locator = null; + if (originalCodexHome === undefined) { + delete process.env.CODEX_HOME; + } else { + process.env.CODEX_HOME = originalCodexHome; + } + await rm(codexHome, { recursive: true, force: true }); + }); + + it('does not attach for session metadata alone', async () => { + const located: string[] = []; + locator = createCodexTranscriptLocator({ + cwd: '/tmp/project', + startupTimestampMs: Date.now(), + intervalMs: 25, + onLocated: (result) => located.push(result.transcriptPath) + }); + await locator.ready; + const transcriptPath = await createTranscript('thread-meta-only', '/tmp/project'); + + await wait(150); + expect(located).toEqual([]); + expect(transcriptPath).toContain('thread-meta-only'); + }); + + it('attaches after fresh real user activity in the matching cwd', async () => { + const located: string[] = []; + locator = createCodexTranscriptLocator({ + cwd: '/tmp/project', + startupTimestampMs: Date.now(), + intervalMs: 25, + settlementMs: 25, + onLocated: (result) => located.push(result.transcriptPath) + }); + await locator.ready; + const transcriptPath = await createTranscript('thread-user', '/tmp/project'); + + await appendFile(transcriptPath, `${JSON.stringify({ + timestamp: new Date().toISOString(), + type: 'event_msg', + payload: { type: 'user_message', message: 'hello' } + })}\n`); + await wait(150); + + expect(located).toEqual([transcriptPath]); + }); + + it('attaches after image-only user activity', async () => { + const located: string[] = []; + locator = createCodexTranscriptLocator({ + cwd: '/tmp/project', + startupTimestampMs: Date.now(), + intervalMs: 25, + settlementMs: 0, + onLocated: (result) => located.push(result.transcriptPath) + }); + await locator.ready; + const transcriptPath = await createTranscript('thread-image', '/tmp/project'); + + await appendFile(transcriptPath, `${JSON.stringify({ + timestamp: new Date().toISOString(), + type: 'response_item', + payload: { + type: 'message', + role: 'user', + content: [{ type: 'input_image', image_url: 'data:image/png;base64,abc' }] + } + })}\n`); + await wait(100); + + expect(located).toEqual([transcriptPath]); + }); + + it('refuses fallback when fresh activity is ambiguous', async () => { + const located: string[] = []; + const ambiguous: string[][] = []; + locator = createCodexTranscriptLocator({ + cwd: '/tmp/project', + startupTimestampMs: Date.now(), + intervalMs: 25, + settlementMs: 50, + onLocated: (result) => located.push(result.transcriptPath), + onAmbiguous: (paths) => ambiguous.push(paths) + }); + await locator.ready; + const first = await createTranscript('thread-a', '/tmp/project'); + const second = await createTranscript('thread-b', '/tmp/project'); + + const userEvent = `${JSON.stringify({ + timestamp: new Date().toISOString(), + type: 'event_msg', + payload: { type: 'user_message', message: 'hello' } + })}\n`; + await Promise.all([appendFile(first, userEvent), appendFile(second, userEvent)]); + await wait(150); + + expect(located).toEqual([]); + expect(ambiguous).toHaveLength(1); + expect(new Set(ambiguous[0])).toEqual(new Set([first, second])); + }); + + it('rejects candidates whose activity arrives in adjacent polling cycles', async () => { + const located: string[] = []; + const ambiguous: string[][] = []; + locator = createCodexTranscriptLocator({ + cwd: '/tmp/project', + startupTimestampMs: Date.now(), + intervalMs: 25, + settlementMs: 150, + onLocated: (result) => located.push(result.transcriptPath), + onAmbiguous: (paths) => ambiguous.push(paths) + }); + await locator.ready; + const first = await createTranscript('thread-staggered-a', '/tmp/project'); + const second = await createTranscript('thread-staggered-b', '/tmp/project'); + const userEvent = (message: string) => JSON.stringify({ + timestamp: new Date().toISOString(), + type: 'event_msg', + payload: { type: 'user_message', message } + }); + + await appendFile(first, `${userEvent('first')}\n`); + await wait(60); + await appendFile(second, `${userEvent('second')}\n`); + await wait(150); + + expect(located).toEqual([]); + expect(ambiguous).toHaveLength(1); + expect(new Set(ambiguous[0])).toEqual(new Set([first, second])); + }); + + it('retries an unterminated final JSON record after it is completed', async () => { + const located: string[] = []; + locator = createCodexTranscriptLocator({ + cwd: '/tmp/project', + startupTimestampMs: Date.now(), + intervalMs: 25, + settlementMs: 0, + onLocated: (result) => located.push(result.transcriptPath) + }); + await locator.ready; + const transcriptPath = await createTranscript('thread-partial', '/tmp/project'); + const userEvent = JSON.stringify({ + timestamp: new Date().toISOString(), + type: 'event_msg', + payload: { type: 'user_message', message: 'completed later' } + }); + const splitAt = Math.floor(userEvent.length / 2); + + await appendFile(transcriptPath, userEvent.slice(0, splitAt)); + await wait(75); + expect(located).toEqual([]); + + await appendFile(transcriptPath, userEvent.slice(splitAt)); + await wait(100); + expect(located).toEqual([transcriptPath]); + }); + + it('ignores pre-existing fresh transcripts even when they receive new activity', async () => { + const transcriptPath = await createTranscript('thread-existing', '/tmp/project'); + const located: string[] = []; + locator = createCodexTranscriptLocator({ + cwd: '/tmp/project', + startupTimestampMs: Date.now(), + intervalMs: 25, + settlementMs: 0, + onLocated: (result) => located.push(result.transcriptPath) + }); + await locator.ready; + + await appendFile(transcriptPath, `${JSON.stringify({ + timestamp: new Date().toISOString(), + type: 'event_msg', + payload: { type: 'user_message', message: 'other terminal' } + })}\n`); + await wait(100); + + expect(located).toEqual([]); + }); + + it('polls only the exact resume transcript once it is found', async () => { + const unrelated = await createTranscript('thread-unrelated', '/tmp/project'); + const target = await createTranscript('thread-resume', '/tmp/original-project'); + await appendFile(unrelated, `${JSON.stringify({ + timestamp: new Date().toISOString(), + type: 'event_msg', + payload: { type: 'user_message', message: 'unrelated activity' } + })}\n`); + const located: string[] = []; + const ambiguous: string[][] = []; + locator = createCodexTranscriptLocator({ + cwd: '/tmp/current-project', + startupTimestampMs: Date.now(), + resumeSessionId: 'thread-resume', + intervalMs: 25, + onLocated: (result) => located.push(result.transcriptPath), + onAmbiguous: (paths) => ambiguous.push(paths) + }); + await locator.ready; + + await appendFile(target, `${JSON.stringify({ + timestamp: new Date().toISOString(), + type: 'event_msg', + payload: { type: 'user_message', message: 'resume activity' } + })}\n`); + await wait(100); + + expect(located).toEqual([target]); + expect(ambiguous).toEqual([]); + }); + + async function createTranscript(sessionId: string, cwd: string): Promise { + const transcriptPath = join(sessionDirectory, `rollout-${sessionId}.jsonl`); + await writeFile(transcriptPath, `${JSON.stringify({ + type: 'session_meta', + payload: { id: sessionId, cwd } + })}\n`); + return transcriptPath; + } +}); diff --git a/cli/src/codex/utils/codexTranscriptLocator.ts b/cli/src/codex/utils/codexTranscriptLocator.ts new file mode 100644 index 00000000..6d5bcd50 --- /dev/null +++ b/cli/src/codex/utils/codexTranscriptLocator.ts @@ -0,0 +1,380 @@ +import { homedir } from 'node:os'; +import { join, resolve } from 'node:path'; +import { open, readdir, stat } from 'node:fs/promises'; +import { logger } from '@/ui/logger'; +import { convertCodexEvent, type CodexSessionEvent } from './codexEventConverter'; + +export type LocatedCodexTranscript = { + sessionId: string; + transcriptPath: string; +}; + +export type CodexTranscriptLocator = { + ready: Promise; + cleanup: () => Promise; +}; + +type TranscriptState = { + offset: number; + size: number; + mtimeMs: number; + ino: number; + sessionId: string | null; + cwd: string | null; +}; + +type CodexTranscriptLocatorOptions = { + cwd: string; + startupTimestampMs: number; + resumeSessionId?: string | null; + intervalMs?: number; + settlementMs?: number; + onLocated: (located: LocatedCodexTranscript) => void; + onAmbiguous?: (paths: string[]) => void; +}; + +const DEFAULT_LOCATOR_INTERVAL_MS = 500; + +export function createCodexTranscriptLocator(options: CodexTranscriptLocatorOptions): CodexTranscriptLocator { + const locator = new CodexTranscriptLocatorImpl(options); + const ready = locator.start().catch((error) => { + logger.debug('[codex-transcript-locator] Failed to initialize transcript fallback', error); + }); + return { + ready, + cleanup: async () => { + await locator.cleanup(); + await ready; + } + }; +} + +class CodexTranscriptLocatorImpl { + private readonly sessionsRoot: string; + private readonly targetCwd: string; + private readonly startupTimestampMs: number; + private readonly resumeSessionId: string | null; + private readonly intervalMs: number; + private readonly settlementMs: number; + private readonly onLocated: CodexTranscriptLocatorOptions['onLocated']; + private readonly onAmbiguous?: CodexTranscriptLocatorOptions['onAmbiguous']; + private readonly states = new Map(); + private readonly initialFreshPaths = new Set(); + private readonly pendingCandidates = new Map(); + private resumeTranscriptPaths: string[] | null = null; + private firstCandidateTimestampMs: number | null = null; + private interval: ReturnType | null = null; + private scanPromise: Promise | null = null; + private stopped = false; + + constructor(options: CodexTranscriptLocatorOptions) { + const codexHome = process.env.CODEX_HOME || join(homedir(), '.codex'); + this.sessionsRoot = join(codexHome, 'sessions'); + this.targetCwd = normalizePath(options.cwd); + this.startupTimestampMs = options.startupTimestampMs; + this.resumeSessionId = options.resumeSessionId ?? null; + this.intervalMs = options.intervalMs ?? DEFAULT_LOCATOR_INTERVAL_MS; + this.settlementMs = options.settlementMs ?? this.intervalMs; + this.onLocated = options.onLocated; + this.onAmbiguous = options.onAmbiguous; + } + + async start(): Promise { + if (!this.resumeSessionId) { + const existingPaths = await this.listNearbyTranscriptFiles(); + for (const transcriptPath of existingPaths) { + this.initialFreshPaths.add(transcriptPath); + } + } + if (this.stopped) return; + + void this.scan(); + this.interval = setInterval(() => void this.scan(), this.intervalMs); + this.interval.unref?.(); + } + + async cleanup(): Promise { + this.stopped = true; + if (this.interval) { + clearInterval(this.interval); + this.interval = null; + } + await this.scanPromise?.catch(() => {}); + } + + private async scan(): Promise { + if (this.stopped || this.scanPromise) { + return this.scanPromise ?? Promise.resolve(); + } + + this.scanPromise = this.runScan(); + try { + await this.scanPromise; + } finally { + this.scanPromise = null; + } + } + + private async runScan(): Promise { + const files = await this.listCandidateFiles(); + for (const transcriptPath of files) { + if (this.stopped) return; + const candidate = await this.scanFile(transcriptPath); + if (candidate) { + this.pendingCandidates.set(candidate.transcriptPath, candidate); + } + } + + if (this.stopped || this.pendingCandidates.size === 0) { + return; + } + + if (this.pendingCandidates.size > 1) { + const paths = [...this.pendingCandidates.keys()]; + logger.warn('[codex-transcript-locator] Ambiguous Codex transcript activity; refusing fallback attachment', paths); + this.stopPolling(); + this.onAmbiguous?.(paths); + return; + } + + const [located] = this.pendingCandidates.values(); + if (!located) return; + + if (!this.resumeSessionId) { + if (this.firstCandidateTimestampMs === null) { + this.firstCandidateTimestampMs = Date.now(); + } + if (Date.now() - this.firstCandidateTimestampMs < this.settlementMs) { + return; + } + } + + logger.debug(`[codex-transcript-locator] Located ${located.sessionId} at ${located.transcriptPath}`); + this.stopPolling(); + this.onLocated(located); + } + + private async scanFile(transcriptPath: string): Promise { + let fileStats: Awaited>; + try { + fileStats = await stat(transcriptPath); + } catch { + return null; + } + if (!fileStats.isFile()) return null; + + const previous = this.states.get(transcriptPath); + let state: TranscriptState = previous ?? { + offset: 0, + size: 0, + mtimeMs: 0, + ino: fileStats.ino, + sessionId: null, + cwd: null + }; + + const replaced = previous && previous.ino !== fileStats.ino; + const truncated = previous && fileStats.size < previous.offset; + const rewrittenAtSameSize = previous + && fileStats.size === previous.size + && fileStats.mtimeMs !== previous.mtimeMs + && previous.offset === previous.size; + if (replaced || truncated || rewrittenAtSameSize) { + state = { + offset: 0, + size: 0, + mtimeMs: 0, + ino: fileStats.ino, + sessionId: null, + cwd: null + }; + } else if (previous + && fileStats.size === previous.size + && fileStats.mtimeMs === previous.mtimeMs) { + return null; + } + + if (fileStats.size <= state.offset) { + state.size = fileStats.size; + state.mtimeMs = fileStats.mtimeMs; + state.ino = fileStats.ino; + this.states.set(transcriptPath, state); + return null; + } + + let content: Buffer; + try { + content = await readBytes(transcriptPath, state.offset, fileStats.size - state.offset); + } catch { + return null; + } + + const startOffset = state.offset; + const text = content.toString('utf8'); + const lines = text.split('\n'); + let consumedBytes = 0; + let sawFreshUserActivity = false; + + for (let index = 0; index < lines.length; index += 1) { + const line = lines[index] ?? ''; + const terminated = index < lines.length - 1; + const recordBytes = Buffer.byteLength(line) + (terminated ? 1 : 0); + + if (!line.trim()) { + if (terminated) consumedBytes += recordBytes; + continue; + } + + let event: CodexSessionEvent; + try { + event = JSON.parse(line) as CodexSessionEvent; + } catch { + if (terminated) consumedBytes += recordBytes; + continue; + } + + if (event.type === 'session_meta') { + const metadata = asRecord(event.payload); + state.sessionId = asString(metadata?.id) ?? state.sessionId; + const eventCwd = asString(metadata?.cwd); + state.cwd = eventCwd ? normalizePath(eventCwd) : state.cwd; + } + + if (convertCodexEvent(event)?.userActivity) { + const eventTimestamp = parseTimestamp(event.timestamp); + if (eventTimestamp !== null && eventTimestamp >= this.startupTimestampMs) { + sawFreshUserActivity = true; + } + } + + consumedBytes += recordBytes; + } + + state.offset = startOffset + consumedBytes; + state.size = startOffset + content.length; + state.mtimeMs = fileStats.mtimeMs; + state.ino = fileStats.ino; + this.states.set(transcriptPath, state); + + if (!sawFreshUserActivity || !state.sessionId) { + return null; + } + if (this.resumeSessionId) { + if (state.sessionId !== this.resumeSessionId) { + return null; + } + } else if (state.cwd !== this.targetCwd) { + return null; + } + + return { sessionId: state.sessionId, transcriptPath }; + } + + private async listCandidateFiles(): Promise { + if (this.resumeSessionId) { + if (this.resumeTranscriptPaths) { + return this.resumeTranscriptPaths; + } + const suffix = `-${this.resumeSessionId}.jsonl`; + const matches = await listJsonlFiles(this.sessionsRoot, (name) => name.endsWith(suffix)); + if (matches.length > 0) { + this.resumeTranscriptPaths = matches; + } + return matches; + } + + const files = await this.listNearbyTranscriptFiles(); + return files.filter((transcriptPath) => !this.initialFreshPaths.has(transcriptPath)); + } + + private async listNearbyTranscriptFiles(): Promise { + const roots = getNearbyDateRoots(this.sessionsRoot, this.startupTimestampMs); + const groups = await Promise.all(roots.map((root) => listJsonlFiles(root))); + return groups.flat(); + } + + private stopPolling(): void { + this.stopped = true; + if (this.interval) { + clearInterval(this.interval); + this.interval = null; + } + } +} + +async function readBytes(filePath: string, offset: number, length: number): Promise { + const handle = await open(filePath, 'r'); + try { + const buffer = Buffer.alloc(length); + let totalBytesRead = 0; + while (totalBytesRead < length) { + const { bytesRead } = await handle.read( + buffer, + totalBytesRead, + length - totalBytesRead, + offset + totalBytesRead + ); + if (bytesRead === 0) break; + totalBytesRead += bytesRead; + } + return buffer.subarray(0, totalBytesRead); + } finally { + await handle.close(); + } +} + +async function listJsonlFiles( + directory: string, + matchesName: (name: string) => boolean = () => true +): Promise { + try { + const entries = await readdir(directory, { withFileTypes: true }); + const groups = await Promise.all(entries.map(async (entry) => { + const fullPath = join(directory, entry.name); + if (entry.isDirectory()) { + return await listJsonlFiles(fullPath, matchesName); + } + return entry.isFile() && entry.name.endsWith('.jsonl') && matchesName(entry.name) + ? [fullPath] + : []; + })); + return groups.flat(); + } catch { + return []; + } +} + +function getNearbyDateRoots(sessionsRoot: string, timestampMs: number): string[] { + const roots: string[] = []; + for (const offsetDays of [-1, 0, 1]) { + const date = new Date(timestampMs + offsetDays * 24 * 60 * 60 * 1000); + roots.push(join( + sessionsRoot, + String(date.getUTCFullYear()), + String(date.getUTCMonth() + 1).padStart(2, '0'), + String(date.getUTCDate()).padStart(2, '0') + )); + } + return roots; +} + +function asRecord(value: unknown): Record | null { + return value && 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 parseTimestamp(value: unknown): number | null { + if (typeof value !== 'string') return null; + const parsed = Date.parse(value); + return Number.isNaN(parsed) ? null : parsed; +} + +function normalizePath(value: string): string { + const normalized = resolve(value); + return process.platform === 'win32' ? normalized.toLowerCase() : normalized; +} diff --git a/cli/src/codex/utils/codexVersion.test.ts b/cli/src/codex/utils/codexVersion.test.ts index e2b57571..47efcbbd 100644 --- a/cli/src/codex/utils/codexVersion.test.ts +++ b/cli/src/codex/utils/codexVersion.test.ts @@ -18,6 +18,7 @@ vi.mock('cross-spawn', () => ({ import { assertCodexLocalSupported, + CODEX_VERSION_TIMEOUT_MS, isCodexVersionAtLeast, MIN_CODEX_HOOKS_VERSION, parseCodexVersion @@ -74,7 +75,8 @@ describe('codexVersion', () => { 'node', [codexScriptPath, '--version'], expect.objectContaining({ - encoding: 'utf8' + encoding: 'utf8', + timeout: CODEX_VERSION_TIMEOUT_MS }) ) }) diff --git a/cli/src/codex/utils/codexVersion.ts b/cli/src/codex/utils/codexVersion.ts index aa976b78..deeb5b0b 100644 --- a/cli/src/codex/utils/codexVersion.ts +++ b/cli/src/codex/utils/codexVersion.ts @@ -3,6 +3,7 @@ import { withBunRuntimeEnv } from '@/utils/bunRuntime' import { resolveCodexCommand } from './codexExecutable' export const MIN_CODEX_HOOKS_VERSION = '0.124.0' +export const CODEX_VERSION_TIMEOUT_MS = 3_000 const SEMVER_PATTERN = /\b(\d+)\.(\d+)\.(\d+)\b/ @@ -59,6 +60,7 @@ export function assertCodexLocalSupported(): void { const result = spawn.sync(codexCommand.command, [...codexCommand.args, '--version'], { encoding: 'utf8', env: withBunRuntimeEnv(), + timeout: CODEX_VERSION_TIMEOUT_MS, windowsHide: process.platform === 'win32' }) diff --git a/cli/src/commands/codex.test.ts b/cli/src/commands/codex.test.ts index 7e04fc33..1e5fba1c 100644 --- a/cli/src/commands/codex.test.ts +++ b/cli/src/commands/codex.test.ts @@ -62,6 +62,20 @@ describe('codexCommand', () => { expect(runCodexMock).toHaveBeenCalledWith({}) }) + it('does not block local Codex startup on Hub auto-start readiness', async () => { + maybeAutoStartServerMock.mockImplementationOnce(async () => { + await new Promise(() => {}) + }) + + await codexCommand.run(createCommandContext([])) + + expect(runCodexMock).toHaveBeenCalledOnce() + expect(maybeAutoStartServerMock).toHaveBeenCalledWith({ + waitForReady: false, + quiet: true + }) + }) + it('checks Codex version before resuming a local session', async () => { await codexCommand.run(createCommandContext(['resume', 'session-123'])) diff --git a/cli/src/commands/codex.ts b/cli/src/commands/codex.ts index 196adc94..579bf5cb 100644 --- a/cli/src/commands/codex.ts +++ b/cli/src/commands/codex.ts @@ -106,7 +106,11 @@ export const codexCommand: CommandDefinition = { } await initializeToken() - await maybeAutoStartServer() + if (options.startedBy === 'runner') { + await maybeAutoStartServer() + } else { + void maybeAutoStartServer({ waitForReady: false, quiet: true }) + } await authAndSetupMachineIfNeeded() await runCodex(options) } catch (error) { diff --git a/cli/src/utils/autoStartServer.ts b/cli/src/utils/autoStartServer.ts index c565d962..3cf20814 100644 --- a/cli/src/utils/autoStartServer.ts +++ b/cli/src/utils/autoStartServer.ts @@ -142,7 +142,10 @@ function startServerAsChild(): void { /** * Main entry point: auto-start hub if conditions are met */ -export async function maybeAutoStartServer(): Promise { +export async function maybeAutoStartServer(options?: { + waitForReady?: boolean + quiet?: boolean +}): Promise { try { const shouldStart = await shouldAutoStartServer() if (!shouldStart) { @@ -150,24 +153,36 @@ export async function maybeAutoStartServer(): Promise { } logger.debug('[AUTO-START] Starting hub automatically...') - console.log(chalk.gray('Starting HAPI hub in background...')) + if (!options?.quiet) { + console.log(chalk.gray('Starting HAPI hub in background...')) + } startServerAsChild() + if (options?.waitForReady === false) { + return + } + const isReady = await waitForServerReady(configuration.apiUrl) if (!isReady) { - console.log(chalk.yellow('Warning: Hub did not start within expected time')) - console.log(chalk.gray(' Try running `hapi hub` manually to see errors')) + if (!options?.quiet) { + console.log(chalk.yellow('Warning: Hub did not start within expected time')) + console.log(chalk.gray(' Try running `hapi hub` manually to see errors')) + } return } - console.log(chalk.green('HAPI hub started')) + if (!options?.quiet) { + console.log(chalk.green('HAPI hub started')) + } } catch (error) { logger.debug('[AUTO-START] Error during hub auto-start', error) - console.log(chalk.yellow('Warning: Failed to auto-start hub')) - if (error instanceof Error) { - console.log(chalk.gray(` Error: ${error.message}`)) + if (!options?.quiet) { + console.log(chalk.yellow('Warning: Failed to auto-start hub')) + if (error instanceof Error) { + console.log(chalk.gray(` Error: ${error.message}`)) + } } } } diff --git a/hub/src/store/sessionStore.ts b/hub/src/store/sessionStore.ts index 0e18a859..6b123658 100644 --- a/hub/src/store/sessionStore.ts +++ b/hub/src/store/sessionStore.ts @@ -33,9 +33,10 @@ export class SessionStore { namespace: string, model?: string, effort?: string, - modelReasoningEffort?: string + modelReasoningEffort?: string, + requestedId?: string ): StoredSession { - return getOrCreateSession(this.db, tag, metadata, agentState, namespace, model, effort, modelReasoningEffort) + return getOrCreateSession(this.db, tag, metadata, agentState, namespace, model, effort, modelReasoningEffort, requestedId) } updateSessionMetadata( diff --git a/hub/src/store/sessions.test.ts b/hub/src/store/sessions.test.ts index f00a0377..90a89f87 100644 --- a/hub/src/store/sessions.test.ts +++ b/hub/src/store/sessions.test.ts @@ -1,5 +1,7 @@ import { describe, expect, it } from 'bun:test' import { Store } from './index' +import { randomUUID } from 'node:crypto' +import { SessionIdentityConflictError } from './sessions' function makeStore(): Store { return new Store(':memory:') @@ -10,6 +12,65 @@ function getMetadata(store: Store, id: string): Record | null { return (row?.metadata ?? null) as Record | null } +describe('getOrCreateSession: requested identity', () => { + it('creates and idempotently reloads a client-requested id', () => { + const store = makeStore() + const requestedId = randomUUID() + + const created = store.sessions.getOrCreateSession( + 'lazy-session-tag', + { path: '/tmp/project' }, + { controlledByUser: true }, + 'default', + undefined, + undefined, + undefined, + requestedId + ) + const reloaded = store.sessions.getOrCreateSession( + 'lazy-session-tag', + { path: '/tmp/ignored' }, + null, + 'default', + undefined, + undefined, + undefined, + requestedId + ) + + expect(created.id).toBe(requestedId) + expect(reloaded.id).toBe(requestedId) + expect(store.sessions.getSessionsByNamespace('default')).toHaveLength(1) + store.close() + }) + + it('rejects a tag already bound to another requested id', () => { + const store = makeStore() + const firstId = randomUUID() + store.sessions.getOrCreateSession( + 'conflicting-tag', {}, null, 'default', undefined, undefined, undefined, firstId + ) + + expect(() => store.sessions.getOrCreateSession( + 'conflicting-tag', {}, null, 'default', undefined, undefined, undefined, randomUUID() + )).toThrow(SessionIdentityConflictError) + store.close() + }) + + it('rejects a requested id already bound to another tag', () => { + const store = makeStore() + const requestedId = randomUUID() + store.sessions.getOrCreateSession( + 'first-tag', {}, null, 'default', undefined, undefined, undefined, requestedId + ) + + expect(() => store.sessions.getOrCreateSession( + 'second-tag', {}, null, 'default', undefined, undefined, undefined, requestedId + )).toThrow(SessionIdentityConflictError) + store.close() + }) +}) + describe('updateSessionMetadata: protocol resume token preservation', () => { it('preserves cursorSessionId when archive payload omits it (Cursor crash-archive)', () => { const store = makeStore() diff --git a/hub/src/store/sessions.ts b/hub/src/store/sessions.ts index 537216c0..a5f80dcd 100644 --- a/hub/src/store/sessions.ts +++ b/hub/src/store/sessions.ts @@ -187,18 +187,32 @@ export function getOrCreateSession( namespace: string, model?: string, effort?: string, - modelReasoningEffort?: string + modelReasoningEffort?: string, + requestedId?: string ): StoredSession { const existing = db.prepare( 'SELECT * FROM sessions WHERE tag = ? AND namespace = ? ORDER BY created_at DESC LIMIT 1' ).get(tag, namespace) as DbSessionRow | undefined if (existing) { + if (requestedId && existing.id !== requestedId) { + throw new SessionIdentityConflictError('Session tag is already bound to a different id') + } return toStoredSession(existing) } const now = Date.now() - const id = randomUUID() + const id = requestedId ?? randomUUID() + + if (requestedId) { + const existingById = getSession(db, requestedId) + if (existingById) { + if (existingById.namespace === namespace && existingById.tag === tag) { + return existingById + } + throw new SessionIdentityConflictError('Session id is already bound to a different session') + } + } const metadataJson = JSON.stringify(metadata) const agentStateJson = agentState === null || agentState === undefined ? null : JSON.stringify(agentState) @@ -243,6 +257,13 @@ export function getOrCreateSession( return row } +export class SessionIdentityConflictError extends Error { + constructor(message: string) { + super(message) + this.name = 'SessionIdentityConflictError' + } +} + export function updateSessionMetadata( db: Database, id: string, diff --git a/hub/src/sync/sessionCache.ts b/hub/src/sync/sessionCache.ts index 304b01cc..bdfafecd 100644 --- a/hub/src/sync/sessionCache.ts +++ b/hub/src/sync/sessionCache.ts @@ -75,9 +75,19 @@ export class SessionCache { namespace: string, model?: string, effort?: string, - modelReasoningEffort?: string + modelReasoningEffort?: string, + requestedId?: string ): Session { - const stored = this.store.sessions.getOrCreateSession(tag, metadata, agentState, namespace, model, effort, modelReasoningEffort) + const stored = this.store.sessions.getOrCreateSession( + tag, + metadata, + agentState, + namespace, + model, + effort, + modelReasoningEffort, + requestedId + ) return this.refreshSession(stored.id) ?? (() => { throw new Error('Failed to load session') })() } diff --git a/hub/src/sync/syncEngine.ts b/hub/src/sync/syncEngine.ts index 7986a81d..414da247 100644 --- a/hub/src/sync/syncEngine.ts +++ b/hub/src/sync/syncEngine.ts @@ -387,9 +387,19 @@ export class SyncEngine { namespace: string, model?: string, effort?: string, - modelReasoningEffort?: string + modelReasoningEffort?: string, + requestedId?: string ): Session { - return this.sessionCache.getOrCreateSession(tag, metadata, agentState, namespace, model, effort, modelReasoningEffort) + return this.sessionCache.getOrCreateSession( + tag, + metadata, + agentState, + namespace, + model, + effort, + modelReasoningEffort, + requestedId + ) } getOrCreateMachine(id: string, metadata: unknown, runnerState: unknown, namespace: string): Machine { diff --git a/hub/src/web/routes/cli.test.ts b/hub/src/web/routes/cli.test.ts index 448a9249..637f95bb 100644 --- a/hub/src/web/routes/cli.test.ts +++ b/hub/src/web/routes/cli.test.ts @@ -1,8 +1,9 @@ -import { beforeAll, describe, expect, it } from 'bun:test' +import { beforeAll, describe, expect, it, mock } from 'bun:test' import { Hono } from 'hono' import type { SyncEngine } from '../../sync/syncEngine' import { createConfiguration } from '../../configuration' import { createCliRoutes } from './cli' +import { SessionIdentityConflictError } from '../../store/sessions' function createApp(engine: Partial) { const app = new Hono() @@ -114,3 +115,104 @@ describe('cli resume routes', () => { }) }) }) + +describe('cli lazy session creation', () => { + const sessionId = '11111111-1111-4111-8111-111111111111' + + it('creates the machine and requested session identity in one request', async () => { + const getOrCreateMachine = mock(() => ({ id: 'machine-1' })) + const getOrCreateSession = mock(() => ({ id: sessionId })) + const app = createApp({ + getMachine: () => null, + getOrCreateMachine, + getOrCreateSession + } as never) + + const response = await app.request('/cli/sessions', { + method: 'POST', + headers: { + ...authHeaders(), + 'content-type': 'application/json' + }, + body: JSON.stringify({ + id: sessionId, + tag: 'lazy-tag', + metadata: { path: '/tmp/project' }, + agentState: { controlledByUser: true }, + machine: { + id: 'machine-1', + metadata: { host: 'localhost' } + } + }) + }) + + expect(response.status).toBe(200) + expect(getOrCreateMachine).toHaveBeenCalledWith( + 'machine-1', + { host: 'localhost' }, + null, + 'default' + ) + expect(getOrCreateSession).toHaveBeenCalledWith( + 'lazy-tag', + { path: '/tmp/project' }, + { controlledByUser: true }, + 'default', + undefined, + undefined, + undefined, + sessionId + ) + }) + + it('rejects an embedded machine owned by another namespace', async () => { + const getOrCreateMachine = mock(() => ({ id: 'machine-1' })) + const getOrCreateSession = mock(() => ({ id: sessionId })) + const app = createApp({ + getMachine: () => ({ id: 'machine-1', namespace: 'other' }), + getOrCreateMachine, + getOrCreateSession + } as never) + + const response = await app.request('/cli/sessions', { + method: 'POST', + headers: { + ...authHeaders(), + 'content-type': 'application/json' + }, + body: JSON.stringify({ + id: sessionId, + tag: 'lazy-tag', + metadata: {}, + machine: { id: 'machine-1', metadata: {} } + }) + }) + + expect(response.status).toBe(403) + expect(getOrCreateMachine).not.toHaveBeenCalled() + expect(getOrCreateSession).not.toHaveBeenCalled() + }) + + it('returns 409 for a requested identity conflict', async () => { + const app = createApp({ + getOrCreateSession: () => { + throw new SessionIdentityConflictError('Session tag is already bound to a different id') + } + }) + + const response = await app.request('/cli/sessions', { + method: 'POST', + headers: { + ...authHeaders(), + 'content-type': 'application/json' + }, + body: JSON.stringify({ + id: sessionId, + tag: 'lazy-tag', + metadata: {} + }) + }) + + expect(response.status).toBe(409) + }) +}) diff --git a/hub/src/web/routes/cli.ts b/hub/src/web/routes/cli.ts index 30a24124..fc0abe43 100644 --- a/hub/src/web/routes/cli.ts +++ b/hub/src/web/routes/cli.ts @@ -10,6 +10,7 @@ import { getConfiguration } from '../../configuration' import { constantTimeEquals } from '../../utils/crypto' import { parseAccessToken } from '../../utils/accessToken' import type { Machine, Session, SyncEngine } from '../../sync/syncEngine' +import { SessionIdentityConflictError } from '../../store/sessions' const bearerSchema = z.string().regex(/^Bearer\s+(.+)$/i) @@ -94,16 +95,38 @@ export function createCliRoutes(getSyncEngine: () => SyncEngine | null): Hono { diff --git a/shared/src/apiTypes.ts b/shared/src/apiTypes.ts index e6381da7..0167c12b 100644 --- a/shared/src/apiTypes.ts +++ b/shared/src/apiTypes.ts @@ -15,17 +15,6 @@ import type { } from './schemas' import type { SessionSummary } from './sessionSummary' -export const CreateOrLoadSessionRequestSchema = z.object({ - tag: z.string().min(1), - metadata: z.unknown(), - agentState: z.unknown().nullable().optional(), - model: z.string().optional(), - modelReasoningEffort: z.string().optional(), - effort: z.string().optional() -}) - -export type CreateOrLoadSessionRequest = z.infer - export const CreateOrLoadMachineRequestSchema = z.object({ id: z.string().min(1), metadata: z.unknown(), @@ -34,6 +23,19 @@ export const CreateOrLoadMachineRequestSchema = z.object({ export type CreateOrLoadMachineRequest = z.infer +export const CreateOrLoadSessionRequestSchema = z.object({ + id: z.string().uuid().optional(), + tag: z.string().min(1), + metadata: z.unknown(), + agentState: z.unknown().nullable().optional(), + model: z.string().optional(), + modelReasoningEffort: z.string().optional(), + effort: z.string().optional(), + machine: CreateOrLoadMachineRequestSchema.optional() +}) + +export type CreateOrLoadSessionRequest = z.infer + export const CliMessagesResponseSchema = z.object({ messages: z.array(z.object({ id: z.string(),