fix(hub+cli): defer mergeSessions on cursor ACP reopen until session/load succeeds (closes #939) (#948)

* fix(hub+cli): defer mergeSessions on cursor ACP reopen until session/load succeeds

Emit session-ready from the CLI after ACP load/newSession completes; hub
resumeSession and cursor dedup wait for that signal before merging rows so a
failed session/load no longer deletes the archived session the operator can retry.

Refs #917. Closes #939.

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(hub): gate session-ready wait on cursor ACP protocol only

Legacy stream-json Cursor resumes use cursorLegacyRemoteLauncher, which does
not emit session-ready; limiting the defer-merge and dedup gates to ACP avoids
60s resume_failed timeouts on those sessions.

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(hub): block ACP dedup until session-ready, including on session-end

Inactive ACP spawns that never emitted session-ready could still trigger
deduplicateByAgentSessionId on session-end and delete the original row.
Require session-ready for all ACP dedup paths and skip end-of-session dedup
when load never succeeded.

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(hub): restore session-end dedup for non-ACP cursor duplicates

Only skip the session-end dedup retry for Cursor ACP rows that never emitted
session-ready. Codex/Claude/legacy Cursor duplicates still merge when the live
row ends.

Co-authored-by: Cursor <cursoragent@cursor.com>

---------

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
HeavyGee
2026-06-18 10:10:10 +08:00
committed by GitHub
co-authored by Cursor
parent 0bf17b818e
commit 26d3c2eb34
11 changed files with 455 additions and 22 deletions
+8
View File
@@ -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
+51 -17
View File
@@ -138,15 +138,7 @@ import { ApiSessionClient } from '@/api/apiSession';
function makeSession(sessionId: string | null): CursorSession {
const queue = new MessageQueue2<EnhancedMode>(() => '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;
+31 -1
View File
@@ -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<string, unknown> {
if (error instanceof Error) {
const record: Record<string, unknown> = {
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<string, unknown>) };
}
return { message: String(error) };
}
function isSpawnDefaultModel(modelId: string): boolean {
const normalized = modelId.trim().toLowerCase();
return normalized === 'auto' || normalized === 'default' || normalized === 'default[]';
+1
View File
@@ -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.
+8 -1
View File
@@ -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,
+19 -1
View File
@@ -32,6 +32,11 @@ type SessionEndPayload = {
reason?: SessionEndReason
}
type SessionReadyPayload = {
sid: string
time: number
}
type ResolveSessionAccess = (sessionId: string) => AccessResult<StoredSession>
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
+2
View File
@@ -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,
+1
View File
@@ -185,6 +185,7 @@ export async function startHub(options: StartHubOptions = {}): Promise<HubInstan
},
onWebappEvent: (event: SyncEvent) => 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),
+270
View File
@@ -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(
+62 -2
View File
@@ -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<string>()
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<boolean> {
const start = Date.now()
while (Date.now() - start < timeoutMs) {
+2
View File
@@ -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