diff --git a/cli/src/api/apiSession.ts b/cli/src/api/apiSession.ts index 370239b5..404e10a4 100644 --- a/cli/src/api/apiSession.ts +++ b/cli/src/api/apiSession.ts @@ -586,6 +586,14 @@ 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() + }) + } + emitMessagesConsumed(localIds: string[], options?: { clearQueuedThinkingGrace?: boolean }): void { if (localIds.length === 0) return // `clearQueuedThinkingGrace` is an opt-in signal for the hub to drop diff --git a/cli/src/cursor/cursorAcpRemoteLauncher.test.ts b/cli/src/cursor/cursorAcpRemoteLauncher.test.ts index 3aeb7473..164d2065 100644 --- a/cli/src/cursor/cursorAcpRemoteLauncher.test.ts +++ b/cli/src/cursor/cursorAcpRemoteLauncher.test.ts @@ -138,15 +138,7 @@ import { ApiSessionClient } from '@/api/apiSession'; function makeSession(sessionId: string | null): CursorSession { const queue = new MessageQueue2(() => 'mode'); - const client = { - rpcHandlerManager: { - registerHandler: vi.fn() - }, - updateMetadata: vi.fn(), - sendSessionEvent: vi.fn(), - sendAgentMessage: vi.fn(), - keepAlive: vi.fn() - } as unknown as ApiSessionClient; + const client = makeClient(); const session = new CursorSession({ api: {} as never, @@ -168,6 +160,19 @@ function makeSession(sessionId: string | null): CursorSession { return session; } +function makeClient() { + return { + rpcHandlerManager: { + registerHandler: vi.fn() + }, + updateMetadata: vi.fn(), + sendSessionEvent: vi.fn(), + sendAgentMessage: vi.fn(), + keepAlive: vi.fn(), + emitSessionReady: vi.fn() + } as unknown as ApiSessionClient; +} + describe('cursorAcpRemoteLauncher', () => { beforeEach(() => { harness.initializeError = null; @@ -265,6 +270,26 @@ describe('cursorAcpRemoteLauncher', () => { expect(harness.newSessionCalled).toBe(true); expect(harness.loadSessionCalled).toBe(false); expect(session.onSessionFoundWithProtocol).toHaveBeenCalledWith('new-acp-session', 'acp'); + expect(session.client.emitSessionReady).toHaveBeenCalledTimes(1); + }); + + it('emits session-ready after session/load succeeds', async () => { + const session = makeSession('resume-thread-ready'); + await cursorAcpRemoteLauncher(session); + + expect(harness.loadSessionCalled).toBe(true); + expect(session.client.emitSessionReady).toHaveBeenCalledTimes(1); + }); + + it('does not emit session-ready when session/load fails', async () => { + harness.loadSessionError = new Error('session not found'); + const session = makeSession('old-stream-json-id'); + + await expect(cursorAcpRemoteLauncher(session)).rejects.toThrow( + /Legacy stream-json sessions cannot be loaded via ACP/ + ); + + expect(session.client.emitSessionReady).not.toHaveBeenCalled(); }); it('applies debug mode immediately when setPermissionMode is called', async () => { @@ -274,7 +299,8 @@ describe('cursorAcpRemoteLauncher', () => { updateMetadata: vi.fn(), sendSessionEvent: vi.fn(), sendAgentMessage: vi.fn(), - keepAlive: vi.fn() + keepAlive: vi.fn(), + emitSessionReady: vi.fn() } as unknown as ApiSessionClient; const session = new CursorSession({ @@ -318,7 +344,8 @@ describe('cursorAcpRemoteLauncher', () => { updateMetadata: vi.fn(), sendSessionEvent: vi.fn(), sendAgentMessage: vi.fn(), - keepAlive + keepAlive, + emitSessionReady: vi.fn() } as unknown as ApiSessionClient; const session = new CursorSession({ @@ -362,7 +389,8 @@ describe('cursorAcpRemoteLauncher', () => { updateMetadata: vi.fn(), sendSessionEvent: vi.fn(), sendAgentMessage: vi.fn(), - keepAlive + keepAlive, + emitSessionReady: vi.fn() } as unknown as ApiSessionClient; const session = new CursorSession({ @@ -409,7 +437,8 @@ describe('cursorAcpRemoteLauncher', () => { updateMetadata: vi.fn(), sendSessionEvent: vi.fn(), sendAgentMessage: vi.fn(), - keepAlive: vi.fn() + keepAlive: vi.fn(), + emitSessionReady: vi.fn() } as unknown as ApiSessionClient; const session = new CursorSession({ @@ -453,7 +482,8 @@ describe('cursorAcpRemoteLauncher', () => { updateMetadata: vi.fn(), sendSessionEvent: vi.fn(), sendAgentMessage: vi.fn(), - keepAlive: vi.fn() + keepAlive: vi.fn(), + emitSessionReady: vi.fn() } as unknown as ApiSessionClient; const session = new CursorSession({ @@ -509,7 +539,8 @@ describe('cursorAcpRemoteLauncher', () => { updateMetadata: vi.fn(), sendSessionEvent: vi.fn(), sendAgentMessage: vi.fn(), - keepAlive + keepAlive, + emitSessionReady: vi.fn() } as unknown as ApiSessionClient; const session = new CursorSession({ @@ -550,7 +581,8 @@ describe('cursorAcpRemoteLauncher', () => { updateMetadata: vi.fn(), sendSessionEvent: vi.fn(), sendAgentMessage: vi.fn(), - keepAlive: vi.fn() + keepAlive: vi.fn(), + emitSessionReady: vi.fn() } as unknown as ApiSessionClient; const session = new CursorSession({ @@ -595,7 +627,8 @@ describe('cursorAcpRemoteLauncher', () => { updateMetadata: vi.fn(), sendSessionEvent: vi.fn(), sendAgentMessage: vi.fn(), - keepAlive: vi.fn() + keepAlive: vi.fn(), + emitSessionReady: vi.fn() } as unknown as ApiSessionClient; const session = new CursorSession({ @@ -636,6 +669,7 @@ describe('cursorAcpRemoteLauncher', () => { sendSessionEvent: vi.fn(), sendAgentMessage: vi.fn(), keepAlive: vi.fn(), + emitSessionReady: vi.fn(), emitMessagesConsumed: vi.fn() } as unknown as ApiSessionClient; diff --git a/cli/src/cursor/cursorAcpRemoteLauncher.ts b/cli/src/cursor/cursorAcpRemoteLauncher.ts index 58b611be..a3fe2545 100644 --- a/cli/src/cursor/cursorAcpRemoteLauncher.ts +++ b/cli/src/cursor/cursorAcpRemoteLauncher.ts @@ -105,7 +105,7 @@ class CursorAcpRemoteLauncher extends RemoteLauncherBase { mcpServers: mcpServerList }); } catch (error) { - logger.warn('[cursor-acp] session/load failed', error); + logger.warn('[cursor-acp] session/load failed', formatAcpLoadError(error)); throw new Error( 'Failed to resume Cursor ACP session. Legacy stream-json sessions cannot be loaded via ACP.' ); @@ -125,6 +125,8 @@ class CursorAcpRemoteLauncher extends RemoteLauncherBase { session.onSessionFoundWithProtocol(acpSessionId, 'acp'); } + session.client.emitSessionReady(); + syncCursorModelsFromAcp(backend, acpSessionId); const initialMetadata = backend.getSessionModelsMetadata(acpSessionId); @@ -436,6 +438,34 @@ class CursorAcpRemoteLauncher extends RemoteLauncherBase { } } +function formatAcpLoadError(error: unknown): Record { + if (error instanceof Error) { + const record: Record = { + name: error.name, + message: error.message + }; + const code = (error as Error & { code?: unknown }).code; + if (code !== undefined) { + record.code = code; + } + const data = (error as Error & { data?: unknown }).data; + if (data !== undefined) { + record.data = data; + } + const cause = error.cause; + if (cause !== undefined) { + record.cause = cause instanceof Error + ? { name: cause.name, message: cause.message } + : cause; + } + return record; + } + if (typeof error === 'object' && error !== null) { + return { ...(error as Record) }; + } + return { message: String(error) }; +} + function isSpawnDefaultModel(modelId: string): boolean { const normalized = modelId.trim().toLowerCase(); return normalized === 'auto' || normalized === 'default' || normalized === 'default[]'; diff --git a/hub/README.md b/hub/README.md index faed7297..b224b5df 100644 --- a/hub/README.md +++ b/hub/README.md @@ -152,6 +152,7 @@ Namespace: `/cli` - `update-metadata` - Update session metadata. - `update-state` - Update agent state. - `session-alive` - Keep session active. +- `session-ready` - Cursor ACP `session/load` (or `newSession`) succeeded; hub defers merge/dedup until this arrives on reopen. - `session-end` - Mark session ended. - `machine-alive` - Keep machine online. - `rpc-register` - Register RPC handler. diff --git a/hub/src/socket/handlers/cli/index.ts b/hub/src/socket/handlers/cli/index.ts index 223af963..f39b510c 100644 --- a/hub/src/socket/handlers/cli/index.ts +++ b/hub/src/socket/handlers/cli/index.ts @@ -27,6 +27,11 @@ type SessionEndPayload = { time: number } +type SessionReadyPayload = { + sid: string + time: number +} + type MachineAlivePayload = { machineId: string time: number @@ -38,6 +43,7 @@ export type CliHandlersDeps = { rpcRegistry: RpcRegistry terminalRegistry: TerminalRegistry onSessionAlive?: (payload: SessionAlivePayload) => void + onSessionReady?: (payload: SessionReadyPayload) => void onSessionEnd?: (payload: SessionEndPayload) => void onMachineAlive?: (payload: MachineAlivePayload) => void onWebappEvent?: (event: SyncEvent) => void @@ -48,7 +54,7 @@ export type CliHandlersDeps = { } export function registerCliHandlers(socket: CliSocketWithData, deps: CliHandlersDeps): void { - const { io, store, rpcRegistry, terminalRegistry, onSessionAlive, onSessionEnd, onMachineAlive, onWebappEvent, onBackgroundTaskDelta, onSessionActivity, onSweepImmediateQueued, onMessagesConsumed } = deps + const { io, store, rpcRegistry, terminalRegistry, onSessionAlive, onSessionReady, onSessionEnd, onMachineAlive, onWebappEvent, onBackgroundTaskDelta, onSessionActivity, onSweepImmediateQueued, onMessagesConsumed } = deps const terminalNamespace = io.of('/terminal') const namespace = typeof socket.data.namespace === 'string' ? socket.data.namespace : null @@ -106,6 +112,7 @@ export function registerCliHandlers(socket: CliSocketWithData, deps: CliHandlers resolveSessionAccess, emitAccessError, onSessionAlive, + onSessionReady, onSessionEnd, onWebappEvent, onBackgroundTaskDelta, diff --git a/hub/src/socket/handlers/cli/sessionHandlers.ts b/hub/src/socket/handlers/cli/sessionHandlers.ts index 123af75a..7ffec6db 100644 --- a/hub/src/socket/handlers/cli/sessionHandlers.ts +++ b/hub/src/socket/handlers/cli/sessionHandlers.ts @@ -32,6 +32,11 @@ type SessionEndPayload = { reason?: SessionEndReason } +type SessionReadyPayload = { + sid: string + time: number +} + type ResolveSessionAccess = (sessionId: string) => AccessResult type EmitAccessError = (scope: 'session' | 'machine', id: string, reason: AccessErrorReason) => void @@ -62,6 +67,7 @@ export type SessionHandlersDeps = { resolveSessionAccess: ResolveSessionAccess emitAccessError: EmitAccessError onSessionAlive?: (payload: SessionAlivePayload) => void + onSessionReady?: (payload: SessionReadyPayload) => void onSessionEnd?: (payload: SessionEndPayload) => void onWebappEvent?: (event: SyncEvent) => void onBackgroundTaskDelta?: (sessionId: string, delta: { started: number; completed: number }) => void @@ -74,7 +80,7 @@ export type SessionHandlersDeps = { } export function registerSessionHandlers(socket: CliSocketWithData, deps: SessionHandlersDeps): void { - const { store, resolveSessionAccess, emitAccessError, onSessionAlive, onSessionEnd, onWebappEvent, onBackgroundTaskDelta, onSessionActivity, onSweepImmediateQueued, onMessagesConsumed } = deps + const { store, resolveSessionAccess, emitAccessError, onSessionAlive, onSessionReady, onSessionEnd, onWebappEvent, onBackgroundTaskDelta, onSessionActivity, onSweepImmediateQueued, onMessagesConsumed } = deps socket.on('message', (data: unknown) => { const parsed = messageSchema.safeParse(data) @@ -279,6 +285,18 @@ export function registerSessionHandlers(socket: CliSocketWithData, deps: Session onSessionAlive?.(data) }) + socket.on('session-ready', (data: SessionReadyPayload) => { + if (!data || typeof data.sid !== 'string' || typeof data.time !== 'number') { + return + } + const sessionAccess = resolveSessionAccess(data.sid) + if (!sessionAccess.ok) { + emitAccessError('session', data.sid, sessionAccess.reason) + return + } + onSessionReady?.(data) + }) + socket.on('messages-consumed', (data: { sid: string; localIds: string[]; clearQueuedThinkingGrace?: boolean }) => { if (!data || typeof data.sid !== 'string' || !Array.isArray(data.localIds)) { return diff --git a/hub/src/socket/server.ts b/hub/src/socket/server.ts index af7533e5..8f804e37 100644 --- a/hub/src/socket/server.ts +++ b/hub/src/socket/server.ts @@ -37,6 +37,7 @@ export type SocketServerDeps = { getSession?: (sessionId: string) => { active: boolean; namespace: string } | null onWebappEvent?: (event: SyncEvent) => void onSessionAlive?: (payload: { sid: string; time: number; thinking?: boolean; mode?: 'local' | 'remote' }) => void + onSessionReady?: (payload: { sid: string; time: number }) => void onSessionEnd?: (payload: { sid: string; time: number }) => void onMachineAlive?: (payload: { machineId: string; time: number }) => void onBackgroundTaskDelta?: (sessionId: string, delta: { started: number; completed: number }) => void @@ -116,6 +117,7 @@ export function createSocketServer(deps: SocketServerDeps): { rpcRegistry, terminalRegistry, onSessionAlive: deps.onSessionAlive, + onSessionReady: deps.onSessionReady, onSessionEnd: deps.onSessionEnd, onMachineAlive: deps.onMachineAlive, onWebappEvent: deps.onWebappEvent, diff --git a/hub/src/startHub.ts b/hub/src/startHub.ts index 2a07ad70..58a47344 100644 --- a/hub/src/startHub.ts +++ b/hub/src/startHub.ts @@ -185,6 +185,7 @@ export async function startHub(options: StartHubOptions = {}): Promise syncEngine?.handleRealtimeEvent(event), onSessionAlive: (payload) => syncEngine?.handleSessionAlive(payload), + onSessionReady: (payload) => syncEngine?.handleSessionReady(payload), onSessionEnd: (payload) => syncEngine?.handleSessionEnd(payload), onMachineAlive: (payload) => syncEngine?.handleMachineAlive(payload), onBackgroundTaskDelta: (sessionId, delta) => syncEngine?.handleBackgroundTaskDelta(sessionId, delta), diff --git a/hub/src/sync/sessionModel.test.ts b/hub/src/sync/sessionModel.test.ts index 8489ade9..7ea9950e 100644 --- a/hub/src/sync/sessionModel.test.ts +++ b/hub/src/sync/sessionModel.test.ts @@ -1128,6 +1128,276 @@ describe('session model', () => { } }) + it('defers mergeSessions for cursor reopen until session-ready (load failure leaves old row)', async () => { + const store = new Store(':memory:') + const engine = new SyncEngine( + store, + {} as never, + new RpcRegistry(), + { broadcast() {} } as never + ) + + try { + const oldSession = engine.getOrCreateSession( + 'cursor-reopen-old', + { + path: '/tmp/project', + host: 'localhost', + machineId: 'machine-1', + flavor: 'cursor', + cursorSessionId: 'cursor-csid-load-fail', + cursorSessionProtocol: 'acp' + }, + null, + 'default' + ) + engine.getOrCreateMachine( + 'machine-1', + { host: 'localhost', platform: 'linux', happyCliVersion: '0.1.0' }, + null, + 'default' + ) + engine.handleMachineAlive({ machineId: 'machine-1', time: Date.now() }) + engine.handleSessionEnd({ sid: oldSession.id, time: Date.now() }) + + const spawnedSession = engine.getOrCreateSession( + 'cursor-reopen-spawned', + { + path: '/tmp/project', + host: 'localhost', + machineId: 'machine-1', + flavor: 'cursor', + cursorSessionId: 'cursor-csid-load-fail', + cursorSessionProtocol: 'acp' + }, + null, + 'default' + ) + const spawnedSessionId = spawnedSession.id + + let mergeCalls = 0 + const sessionCache = (engine as any).sessionCache + const mergeSessions = sessionCache.mergeSessions.bind(sessionCache) + sessionCache.mergeSessions = async (oldSessionId: string, newSessionId: string, namespace: string) => { + mergeCalls += 1 + return mergeSessions(oldSessionId, newSessionId, namespace) + } + + ;(engine as any).rpcGateway.spawnSession = async () => { + engine.handleSessionAlive({ sid: spawnedSessionId, time: Date.now() }) + return { type: 'success', sessionId: spawnedSessionId } + } + ;(engine as any).waitForSessionActive = async () => true + ;(engine as any).waitForSessionReady = async () => 'ended' + + const result = await engine.resumeSession(oldSession.id, 'default') + + expect(result).toEqual({ + type: 'error', + message: 'Session ended before Cursor ACP load completed', + code: 'resume_failed' + }) + expect(mergeCalls).toBe(0) + expect(store.sessions.getSession(oldSession.id)).not.toBeNull() + } finally { + engine.stop() + } + }) + + it('does not dedup-merge when ACP spawn ends without session-ready', async () => { + const store = new Store(':memory:') + const engine = new SyncEngine( + store, + {} as never, + new RpcRegistry(), + { broadcast() {} } as never + ) + + try { + const oldSession = engine.getOrCreateSession( + 'cursor-acp-dedup-old', + { + path: '/tmp/project', + host: 'localhost', + machineId: 'machine-1', + flavor: 'cursor', + cursorSessionId: 'cursor-csid-dedup-fail', + cursorSessionProtocol: 'acp' + }, + null, + 'default' + ) + const spawnedSession = engine.getOrCreateSession( + 'cursor-acp-dedup-spawned', + { + path: '/tmp/project', + host: 'localhost', + machineId: 'machine-1', + flavor: 'cursor', + cursorSessionId: 'cursor-csid-dedup-fail', + cursorSessionProtocol: 'acp' + }, + null, + 'default' + ) + + let mergeCalls = 0 + const sessionCache = (engine as any).sessionCache + const mergeSessions = sessionCache.mergeSessions.bind(sessionCache) + sessionCache.mergeSessions = async (oldSessionId: string, newSessionId: string, namespace: string) => { + mergeCalls += 1 + return mergeSessions(oldSessionId, newSessionId, namespace) + } + + engine.handleSessionAlive({ sid: spawnedSession.id, time: Date.now() }) + engine.handleSessionEnd({ sid: spawnedSession.id, time: Date.now(), reason: 'error' }) + + expect(mergeCalls).toBe(0) + expect(store.sessions.getSession(oldSession.id)).not.toBeNull() + } finally { + engine.stop() + } + }) + + it('mergeSessions runs for cursor reopen after session-ready', async () => { + const store = new Store(':memory:') + const engine = new SyncEngine( + store, + {} as never, + new RpcRegistry(), + { broadcast() {} } as never + ) + + try { + const oldSession = engine.getOrCreateSession( + 'cursor-reopen-old-ready', + { + path: '/tmp/project', + host: 'localhost', + machineId: 'machine-1', + flavor: 'cursor', + cursorSessionId: 'cursor-csid-load-ok', + cursorSessionProtocol: 'acp' + }, + null, + 'default' + ) + engine.getOrCreateMachine( + 'machine-1', + { host: 'localhost', platform: 'linux', happyCliVersion: '0.1.0' }, + null, + 'default' + ) + engine.handleMachineAlive({ machineId: 'machine-1', time: Date.now() }) + engine.handleSessionEnd({ sid: oldSession.id, time: Date.now() }) + + const spawnedSession = engine.getOrCreateSession( + 'cursor-reopen-spawned-ready', + { + path: '/tmp/project', + host: 'localhost', + machineId: 'machine-1', + flavor: 'cursor', + cursorSessionId: 'cursor-csid-load-ok', + cursorSessionProtocol: 'acp' + }, + null, + 'default' + ) + const spawnedSessionId = spawnedSession.id + + let mergeCalls = 0 + const sessionCache = (engine as any).sessionCache + const mergeSessions = sessionCache.mergeSessions.bind(sessionCache) + sessionCache.mergeSessions = async (oldSessionId: string, newSessionId: string, namespace: string) => { + mergeCalls += 1 + return mergeSessions(oldSessionId, newSessionId, namespace) + } + + ;(engine as any).rpcGateway.spawnSession = async () => { + engine.handleSessionAlive({ sid: spawnedSessionId, time: Date.now() }) + engine.handleSessionReady({ sid: spawnedSessionId, time: Date.now() }) + return { type: 'success', sessionId: spawnedSessionId } + } + ;(engine as any).waitForSessionActive = async () => true + + const result = await engine.resumeSession(oldSession.id, 'default') + + expect(result).toEqual({ type: 'success', sessionId: spawnedSessionId }) + expect(mergeCalls).toBe(1) + expect(store.sessions.getSession(oldSession.id)).toBeNull() + } finally { + engine.stop() + } + }) + + it('does not wait for session-ready on cursor stream-json reopen', async () => { + const store = new Store(':memory:') + const engine = new SyncEngine( + store, + {} as never, + new RpcRegistry(), + { broadcast() {} } as never + ) + + try { + const oldSession = engine.getOrCreateSession( + 'cursor-legacy-reopen-old', + { + path: '/tmp/project', + host: 'localhost', + machineId: 'machine-1', + flavor: 'cursor', + cursorSessionId: 'legacy-csid', + cursorSessionProtocol: 'stream-json' + }, + null, + 'default' + ) + engine.getOrCreateMachine( + 'machine-1', + { host: 'localhost', platform: 'linux', happyCliVersion: '0.1.0' }, + null, + 'default' + ) + engine.handleMachineAlive({ machineId: 'machine-1', time: Date.now() }) + engine.handleSessionEnd({ sid: oldSession.id, time: Date.now() }) + + const spawnedSession = engine.getOrCreateSession( + 'cursor-legacy-reopen-spawned', + { + path: '/tmp/project', + host: 'localhost', + machineId: 'machine-1', + flavor: 'cursor', + cursorSessionId: 'legacy-csid', + cursorSessionProtocol: 'stream-json' + }, + null, + 'default' + ) + const spawnedSessionId = spawnedSession.id + + let waitForSessionReadyCalls = 0 + ;(engine as any).waitForSessionReady = async () => { + waitForSessionReadyCalls += 1 + return 'timeout' + } + ;(engine as any).rpcGateway.spawnSession = async () => { + engine.handleSessionAlive({ sid: spawnedSessionId, time: Date.now() }) + return { type: 'success', sessionId: spawnedSessionId } + } + ;(engine as any).waitForSessionActive = async () => true + + const result = await engine.resumeSession(oldSession.id, 'default') + + expect(result).toEqual({ type: 'success', sessionId: spawnedSessionId }) + expect(waitForSessionReadyCalls).toBe(0) + } finally { + engine.stop() + } + }) + it('resolves a local resume target for a Codex session', () => { const store = new Store(':memory:') const engine = new SyncEngine( diff --git a/hub/src/sync/syncEngine.ts b/hub/src/sync/syncEngine.ts index 1d733cee..29c442f7 100644 --- a/hub/src/sync/syncEngine.ts +++ b/hub/src/sync/syncEngine.ts @@ -133,6 +133,8 @@ export class SyncEngine { private readonly messageService: MessageService private readonly rpcGateway: RpcGateway private inactivityTimer: NodeJS.Timeout | null = null + /** Sessions that emitted `session-ready` (Cursor ACP load/newSession complete). */ + private readonly sessionReadyIds = new Set() constructor( private readonly store: Store, @@ -269,6 +271,9 @@ export class SyncEngine { this.sessionCache.refreshSession(event.sessionId) const after = this.sessionCache.getSession(event.sessionId) if (after?.metadata && !this.hasSameAgentSessionIds(before?.metadata ?? null, after.metadata)) { + if (!this.canRunCursorDedup(after)) { + return + } void this.sessionCache.deduplicateByAgentSessionId(event.sessionId).catch(() => { // best-effort: dedup failure is harmless, web-side safety net hides remaining duplicates }) @@ -306,11 +311,21 @@ export class SyncEngine { this.triggerDedupIfNeeded(payload.sid) } + handleSessionReady(payload: { sid: string; time: number }): void { + this.sessionReadyIds.add(payload.sid) + this.triggerDedupIfNeeded(payload.sid) + } + clearQueuedThinkingGrace(sessionId: string): void { this.sessionCache.clearQueuedThinkingGrace(sessionId) } handleSessionEnd(payload: { sid: string; time: number; reason?: 'completed' | 'terminated' | 'error' }): void { + const before = this.sessionCache.getSession(payload.sid) + const isCursorAcp = before?.metadata?.flavor === 'cursor' + && before.metadata.cursorSessionProtocol === 'acp' + const shouldRetryDedup = !isCursorAcp || this.sessionReadyIds.has(payload.sid) + this.sessionCache.handleSessionEnd(payload) this.eventPublisher.emit({ type: 'session-ended', @@ -318,8 +333,12 @@ export class SyncEngine { reason: payload.reason }) // Retry dedup now that this session is inactive — a prior dedup may have - // skipped it because it was still active at the time. - this.triggerDedupIfNeeded(payload.sid) + // skipped it because it was still active at the time. Cursor ACP rows that + // never reached session-ready must not dedup-merge the original on failure. + if (shouldRetryDedup) { + this.triggerDedupIfNeeded(payload.sid) + } + this.sessionReadyIds.delete(payload.sid) } handleBackgroundTaskDelta(sessionId: string, delta: { started: number; completed: number }): void { @@ -1163,6 +1182,19 @@ export class SyncEngine { // permissionMode is passed to spawnSession above; do not call set-session-config here. // session-alive can arrive before the CLI registers that RPC handler, which caused resume_failed. + const needsReadyBeforeMerge = spawnResult.sessionId !== access.sessionId + && flavor === 'cursor' + && metadata.cursorSessionProtocol === 'acp' + if (needsReadyBeforeMerge) { + const readyResult = await this.waitForSessionReady(spawnResult.sessionId) + if (readyResult !== 'ready') { + const message = readyResult === 'ended' + ? 'Session ended before Cursor ACP load completed' + : 'Session failed to become ready' + return { type: 'error', message, code: 'resume_failed' } + } + } + if (spawnResult.sessionId !== access.sessionId) { // The old session may have already been merged by the automatic dedup path // (triggered when the spawned CLI sets its agent session ID in metadata). @@ -1410,9 +1442,22 @@ export class SyncEngine { && (prev?.kimiSessionId ?? null) === (next.kimiSessionId ?? null) } + private canRunCursorDedup(session: Session): boolean { + if (session.metadata?.flavor !== 'cursor') { + return true + } + if (session.metadata?.cursorSessionProtocol !== 'acp') { + return true + } + return this.sessionReadyIds.has(session.id) + } + private triggerDedupIfNeeded(sessionId: string): void { const session = this.sessionCache.getSession(sessionId) if (session?.metadata) { + if (!this.canRunCursorDedup(session)) { + return + } void this.sessionCache.deduplicateByAgentSessionId(sessionId).catch(() => { // best-effort: web-side safety net hides remaining duplicates }) @@ -1431,6 +1476,21 @@ export class SyncEngine { return false } + async waitForSessionReady(sessionId: string, timeoutMs: number = 60_000): Promise<'ready' | 'ended' | 'timeout'> { + const start = Date.now() + while (Date.now() - start < timeoutMs) { + if (this.sessionReadyIds.has(sessionId)) { + return 'ready' + } + const session = this.getSession(sessionId) + if (!session?.active) { + return 'ended' + } + await new Promise((resolve) => setTimeout(resolve, 250)) + } + return 'timeout' + } + async waitForSessionInactive(sessionId: string, timeoutMs: number = 15_000): Promise { const start = Date.now() while (Date.now() - start < timeoutMs) { diff --git a/shared/src/socket.ts b/shared/src/socket.ts index 9d6e2e08..e050b0df 100644 --- a/shared/src/socket.ts +++ b/shared/src/socket.ts @@ -213,6 +213,8 @@ export interface ClientToServerEvents { serviceTier?: string | null collaborationMode?: CodexCollaborationMode }) => void + /** CLI agent finished session/load (or equivalent) and can accept prompts. */ + 'session-ready': (data: { sid: string; time: number }) => void 'session-end': (data: { sid: string; time: number; reason?: SessionEndReason }) => void 'messages-consumed': (data: { sid: string; localIds: string[] }) => void 'update-metadata': (data: { sid: string; expectedVersion: number; metadata: unknown }, cb: (answer: UpdateMetadataAck) => void) => void