fix(codex): defer session persistence until user activity

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<Session>
+    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<boolean> {
+    if (signal.aborted || interruptSignal?.aborted) {
+        return false
+    }
+
+    return await new Promise<boolean>((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<boolean> | 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<boolean> {
+        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<boolean> {
+        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<boolean> {
-        if (this.socket.connected) {
-            return true
-        }
-
-        this.socket.connect()
-
-        return await new Promise<boolean>((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<boolean> {
         if (timeoutMs <= 0) {
             return false
@@ -748,6 +1016,60 @@ export class ApiSessionClient extends EventEmitter {
         })
     }
 
+    private async waitForPromise(promise: Promise<unknown>, timeoutMs: number): Promise<boolean> {
+        if (timeoutMs <= 0) {
+            return false
+        }
+
+        return await new Promise<boolean>((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<boolean> {
+        if (this.socket.connected) {
+            return true
+        }
+        if (timeoutMs <= 0) {
+            return false
+        }
+
+        return await new Promise<boolean>((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<boolean> {
+        if (this.state !== 'active') {
+            return false
+        }
         return await this.drainLock(this.metadataLock, timeoutMs)
     }
 
     async flush(options?: { timeoutMs?: number }): Promise<void> {
         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()