fix(hub): preserve invocation activity timestamps (#1249)

hub/src/socket/handlers/cli/sessionHandlers.ts

@@ -311,25 +311,37 @@ export function registerSessionHandlers(socket: CliSocketWithData, deps: Session
             return
         }
         const invokedAt = Date.now()
+        let sessionUpdatedAt: number
         try {
-            store.messages.markMessagesInvoked(data.sid, localIds, invokedAt)
-            onSessionActivity?.(data.sid, invokedAt)
-            // Only drop the queued-thinking grace when the CLI explicitly opts in
-            // (synchronous handlers like slash commands that will never send
-            // their own `thinking=true` keepalive). Normal queue drains still
-            // need the grace so the spinner doesn't flicker between the queue
-            // shift and `backend.prompt` start.
-            if (data.clearQueuedThinkingGrace === true) {
-                onMessagesConsumed?.(data.sid)
-            }
-            // Emit only after the DB write succeeds. Otherwise a transient SQLite
-            // failure would broadcast an `invokedAt` that was never persisted —
-            // live clients would hide the queued rows while a refresh / secondary
-            // client would see them as queued again, diverging the state.
-            onWebappEvent?.({ type: 'messages-consumed', sessionId: data.sid, localIds, invokedAt })
+            sessionUpdatedAt = store.recordMessagesConsumed(
+                data.sid,
+                localIds,
+                invokedAt,
+                sessionAccess.value.namespace
+            )
         } catch (err) {
-            console.error('markMessagesInvoked failed', err)
+            console.error('recordMessagesConsumed failed', err)
+            return
         }
+
+        try {
+            onSessionActivity?.(data.sid, sessionUpdatedAt)
+        } catch (err) {
+            console.error('onSessionActivity failed', err)
+        }
+
+        // Only drop the queued-thinking grace when the CLI explicitly opts in
+        // (synchronous handlers like slash commands that will never send
+        // their own `thinking=true` keepalive). Normal queue drains still
+        // need the grace so the spinner doesn't flicker between the queue
+        // shift and `backend.prompt` start.
+        if (data.clearQueuedThinkingGrace === true) {
+            onMessagesConsumed?.(data.sid)
+        }
+        // Emit only after the DB transaction succeeds. This is an ACK-level
+        // batch contract, so preserve its original timestamp even when IDs are
+        // heterogeneous, replayed, or unknown.
+        onWebappEvent?.({ type: 'messages-consumed', sessionId: data.sid, localIds, invokedAt })
     })
 
     socket.on('session-end', (data: SessionEndPayload) => {

hub/src/store/index.ts

@@ -107,6 +107,36 @@ export class Store {
         this.scratchlist = new ScratchlistStore(this.db)
     }
 
+    /**
+     * Atomically records a CLI prompt-consumption acknowledgement and returns
+     * the persisted session activity timestamp. A duplicate or sibling-stamped
+     * acknowledgement leaves the session untouched while returning its existing
+     * timestamp for replay-safe in-memory cache synchronization.
+     */
+    recordMessagesConsumed(
+        sessionId: string,
+        localIds: string[],
+        invokedAt: number,
+        namespace: string
+    ): number {
+        return this.db.transaction(() => {
+            const changes = this.messages.markMessagesInvoked(sessionId, localIds, invokedAt)
+            if (changes > 0) {
+                this.sessions.touchSessionUpdatedAt(sessionId, invokedAt, namespace)
+            }
+
+            const session = this.sessions.getSessionByNamespace(sessionId, namespace)
+            if (!session) {
+                throw new Error('session not found after messages-consumed transition')
+            }
+            if (changes > 0 && session.updatedAt < invokedAt) {
+                throw new Error('session activity was not persisted after messages-consumed transition')
+            }
+
+            return session.updatedAt
+        })()
+    }
+
     close(): void {
         if (this.closed) return
         this.db.close()

hub/src/store/messageStore.ts

@@ -136,8 +136,8 @@ export class MessageStore {
         return deleteQueuedMessageById(this.db, sessionId, messageId)
     }
 
-    markMessagesInvoked(sessionId: string, localIds: string[], invokedAt: number): void {
-        markMessagesInvoked(this.db, sessionId, localIds, invokedAt)
+    markMessagesInvoked(sessionId: string, localIds: string[], invokedAt: number): number {
+        return markMessagesInvoked(this.db, sessionId, localIds, invokedAt)
     }
 
     mergeSessionMessages(fromSessionId: string, toSessionId: string): { moved: number; oldMaxSeq: number; newMaxSeq: number } {

hub/src/store/messages.test.ts

@@ -175,6 +175,21 @@ describe('cancelQueuedMessage', () => {
     })
 })
 
+describe('recordMessagesConsumed', () => {
+    it('rolls back the invocation transition when the session namespace cannot be verified', () => {
+        const store = makeStore()
+        const session = makeSession(store, 'consumed-rollback-wrong-namespace')
+        store.messages.addMessage(session.id, { role: 'user', content: { type: 'text', text: 'hello' } }, 'local-rollback')
+        const originalUpdatedAt = store.sessions.getSession(session.id)?.updatedAt
+
+        expect(() => store.recordMessagesConsumed(session.id, ['local-rollback'], 2_000, 'other-namespace'))
+            .toThrow('session not found after messages-consumed transition')
+        expect(store.messages.getLocalMessageStates(session.id, ['local-rollback']))
+            .toEqual([{ localId: 'local-rollback', invokedAt: null }])
+        expect(store.sessions.getSession(session.id)?.updatedAt).toBe(originalUpdatedAt)
+    })
+})
+
 describe('position pagination and structural epochs', () => {
     it('returns rows strictly after a cursor and respects an inclusive snapshot head', () => {
         const store = makeStore()

hub/src/store/messages.ts

@@ -629,16 +629,16 @@ export function markMessagesInvoked(
     sessionId: string,
     localIds: string[],
     invokedAt: number
-): void {
-    if (localIds.length === 0) return
+): number {
+    if (localIds.length === 0) return 0
     const placeholders = localIds.map(() => '?').join(', ')
-    db.prepare(
+    return db.prepare(
         `UPDATE messages
          SET invoked_at = ?
          WHERE session_id = ?
            AND local_id IN (${placeholders})
            AND invoked_at IS NULL`
-    ).run(invokedAt, sessionId, ...localIds)
+    ).run(invokedAt, sessionId, ...localIds).changes
 }
 
 export function mergeSessionMessages(

hub/src/sync/sessionActivity.ts

@@ -26,29 +26,11 @@ function hasHumanTextContent(content: unknown): boolean {
         && record.text.trim().length > 0
 }
 
-function isReadyEventContent(content: unknown): boolean {
-    const record = asRecord(content)
-    if (record?.type !== 'event') {
-        return false
-    }
-
-    const data = asRecord(record.data)
-    return data?.type === 'ready'
-}
-
 export function shouldRecordSessionActivity(content: unknown): boolean {
     const message = unwrapRoleWrappedRecordEnvelope(content)
     if (!message) {
         return false
     }
 
-    if (message.role === 'user') {
-        return hasHumanTextContent(message.content)
-    }
-
-    if (message.role !== 'agent') {
-        return false
-    }
-
-    return isReadyEventContent(message.content)
+    return message.role === 'user' && hasHumanTextContent(message.content)
 }

hub/src/sync/sessionModel.test.ts

@@ -612,7 +612,7 @@ describe('session model', () => {
         }
     })
 
-    it('reports session activity when CLI receives a turn-ready event over socket', () => {
+    it('records CLI user text activity but stores and broadcasts ready without recording activity', () => {
         const store = new Store(':memory:')
         const events: SyncEvent[] = []
         const cache = new SessionCache(store, createPublisher(events))
@@ -624,12 +624,13 @@ describe('session model', () => {
         )
         const handlers = new Map<string, (payload: unknown) => void>()
         const activity: Array<{ sessionId: string; updatedAt: number }> = []
+        const roomEvents: unknown[] = []
 
         registerSessionHandlers({
             on: (event: string, handler: (payload: unknown) => void) => {
                 handlers.set(event, handler)
             },
-            to: () => ({ emit() {} })
+            to: () => ({ emit: (_event: string, update: unknown) => roomEvents.push(update) })
         } as never, {
             store,
             resolveSessionAccess: (sessionId) => {
@@ -642,6 +643,10 @@ describe('session model', () => {
             }
         })
 
+        handlers.get('message')?.({
+            sid: session.id,
+            message: JSON.stringify({ role: 'user', content: { type: 'text', text: 'hello' } })
+        })
         handlers.get('message')?.({
             sid: session.id,
             message: JSON.stringify({
@@ -653,9 +658,305 @@ describe('session model', () => {
             })
         })
 
+        const messages = store.messages.getMessages(session.id)
+        expect(messages).toHaveLength(2)
+        expect(roomEvents).toHaveLength(2)
         expect(activity).toHaveLength(1)
         expect(activity[0].sessionId).toBe(session.id)
-        expect(activity[0].updatedAt).toBe(store.messages.getMessages(session.id, 1)[0]?.createdAt)
+        expect(activity[0].updatedAt).toBe(messages[0]?.createdAt)
+    })
+
+    it('records activity only for the first messages-consumed transition while retaining duplicate acknowledgements', () => {
+        const originalDateNow = Date.now
+        let now = 1_000
+        Date.now = () => now
+        try {
+            const store = new Store(':memory:')
+            const events: SyncEvent[] = []
+            const cache = new SessionCache(store, createPublisher(events))
+            const session = cache.getOrCreateSession(
+                'session-cli-consumed-activity',
+                { path: '/tmp/project', host: 'localhost', flavor: 'codex' },
+                null,
+                'default'
+            )
+            const queued = store.messages.addMessage(
+                session.id,
+                { role: 'user', content: { type: 'text', text: 'hello' } },
+                'local-activity'
+            )
+            const handlers = new Map<string, (payload: unknown) => void>()
+            const activity: Array<{ sessionId: string; updatedAt: number }> = []
+            const webEvents: SyncEvent[] = []
+
+            registerSessionHandlers({
+                on: (event: string, handler: (payload: unknown) => void) => {
+                    handlers.set(event, handler)
+                },
+                to: () => ({ emit() {} })
+            } as never, {
+                store,
+                resolveSessionAccess: (sessionId) => {
+                    const stored = store.sessions.getSessionByNamespace(sessionId, 'default')
+                    return stored ? { ok: true, value: stored } : { ok: false, reason: 'not-found' }
+                },
+                emitAccessError: () => {},
+                onSessionActivity: (sessionId, updatedAt) => {
+                    activity.push({ sessionId, updatedAt })
+                    cache.recordSessionActivity(sessionId, updatedAt)
+                },
+                onWebappEvent: (event) => webEvents.push(event)
+            })
+
+            now = 2_000
+            handlers.get('messages-consumed')?.({ sid: session.id, localIds: ['local-activity'] })
+            now = 3_000
+            handlers.get('messages-consumed')?.({ sid: session.id, localIds: ['local-activity'] })
+            now = 4_000
+            handlers.get('message')?.({
+                sid: session.id,
+                message: JSON.stringify({
+                    role: 'agent',
+                    content: { type: 'event', data: { type: 'ready' } }
+                })
+            })
+
+            const invoked = store.messages.getMessages(session.id).find((message) => message.id === queued.id)
+            expect(invoked?.invokedAt).toBe(2_000)
+            expect(activity).toEqual([
+                { sessionId: session.id, updatedAt: 2_000 },
+                { sessionId: session.id, updatedAt: 2_000 }
+            ])
+            expect(store.sessions.getSession(session.id)?.updatedAt).toBe(2_000)
+            expect(events.filter((event) => event.type === 'session-updated')).toHaveLength(1)
+            expect(webEvents.filter((event) => event.type === 'messages-consumed')).toHaveLength(2)
+        } finally {
+            Date.now = originalDateNow
+        }
+    })
+
+    it('replays the persisted invocation timestamp after the first activity callback fails', () => {
+        const originalDateNow = Date.now
+        const originalConsoleError = console.error
+        let now = 1_000
+        Date.now = () => now
+        console.error = () => {}
+        try {
+            const store = new Store(':memory:')
+            const session = store.sessions.getOrCreateSession(
+                'session-cli-consumed-replay',
+                { path: '/tmp/project', host: 'localhost', flavor: 'codex' },
+                null,
+                'default'
+            )
+            store.messages.addMessage(
+                session.id,
+                { role: 'user', content: { type: 'text', text: 'hello' } },
+                'local-replay'
+            )
+            const handlers = new Map<string, (payload: unknown) => void>()
+            const activity: Array<{ sessionId: string; updatedAt: number }> = []
+            const webEvents: SyncEvent[] = []
+            let failActivity = true
+
+            registerSessionHandlers({
+                on: (event: string, handler: (payload: unknown) => void) => {
+                    handlers.set(event, handler)
+                },
+                to: () => ({ emit() {} })
+            } as never, {
+                store,
+                resolveSessionAccess: (sessionId) => {
+                    const stored = store.sessions.getSessionByNamespace(sessionId, 'default')
+                    return stored ? { ok: true, value: stored } : { ok: false, reason: 'not-found' }
+                },
+                emitAccessError: () => {},
+                onSessionActivity: (sessionId, updatedAt) => {
+                    if (failActivity) throw new Error('activity callback failed')
+                    activity.push({ sessionId, updatedAt })
+                },
+                onWebappEvent: (event) => webEvents.push(event)
+            })
+
+            now = 2_000
+            handlers.get('messages-consumed')?.({ sid: session.id, localIds: ['local-replay'] })
+            failActivity = false
+            now = 3_000
+            handlers.get('messages-consumed')?.({ sid: session.id, localIds: ['local-replay'] })
+
+            expect(store.messages.getLocalMessageStates(session.id, ['local-replay']))
+                .toEqual([{ localId: 'local-replay', invokedAt: 2_000 }])
+            expect(store.sessions.getSession(session.id)?.updatedAt).toBe(2_000)
+            expect(activity).toEqual([{ sessionId: session.id, updatedAt: 2_000 }])
+            expect(webEvents.filter((event) => event.type === 'messages-consumed').map((event) => event.invokedAt))
+                .toEqual([2_000, 3_000])
+        } finally {
+            Date.now = originalDateNow
+            console.error = originalConsoleError
+        }
+    })
+
+    it('uses the newest persisted invocation timestamp for a partial messages-consumed batch', () => {
+        const originalDateNow = Date.now
+        let now = 1_000
+        Date.now = () => now
+        try {
+            const store = new Store(':memory:')
+            const session = store.sessions.getOrCreateSession(
+                'session-cli-consumed-partial',
+                { path: '/tmp/project', host: 'localhost', flavor: 'codex' },
+                null,
+                'default'
+            )
+            store.messages.addMessage(session.id, { role: 'user', content: { type: 'text', text: 'old' } }, 'local-old')
+            store.messages.addMessage(session.id, { role: 'user', content: { type: 'text', text: 'fresh' } }, 'local-fresh')
+            store.messages.markMessagesInvoked(session.id, ['local-old'], 1_500)
+            const handlers = new Map<string, (payload: unknown) => void>()
+            const activity: Array<{ sessionId: string; updatedAt: number }> = []
+            const webEvents: SyncEvent[] = []
+
+            registerSessionHandlers({
+                on: (event: string, handler: (payload: unknown) => void) => {
+                    handlers.set(event, handler)
+                },
+                to: () => ({ emit() {} })
+            } as never, {
+                store,
+                resolveSessionAccess: (sessionId) => {
+                    const stored = store.sessions.getSessionByNamespace(sessionId, 'default')
+                    return stored ? { ok: true, value: stored } : { ok: false, reason: 'not-found' }
+                },
+                emitAccessError: () => {},
+                onSessionActivity: (sessionId, updatedAt) => activity.push({ sessionId, updatedAt }),
+                onWebappEvent: (event) => webEvents.push(event)
+            })
+
+            now = 2_000
+            handlers.get('messages-consumed')?.({ sid: session.id, localIds: ['local-old', 'local-fresh'] })
+
+            expect(store.messages.getLocalMessageStates(session.id, ['local-old', 'local-fresh']))
+                .toEqual([
+                    { localId: 'local-old', invokedAt: 1_500 },
+                    { localId: 'local-fresh', invokedAt: 2_000 }
+                ])
+            expect(activity).toEqual([{ sessionId: session.id, updatedAt: 2_000 }])
+            expect(webEvents.filter((event) => event.type === 'messages-consumed').map((event) => event.invokedAt))
+                .toEqual([2_000])
+        } finally {
+            Date.now = originalDateNow
+        }
+    })
+
+    it('keeps the batch ACK timestamp for heterogeneous sibling-preinvoked and unknown IDs', () => {
+        const originalDateNow = Date.now
+        let now = 1_000
+        Date.now = () => now
+        try {
+            const store = new Store(':memory:')
+            const events: SyncEvent[] = []
+            const cache = new SessionCache(store, createPublisher(events))
+            const session = cache.getOrCreateSession(
+                'session-cli-consumed-sibling-preinvoked',
+                { path: '/tmp/project', host: 'localhost', flavor: 'codex' },
+                null,
+                'default'
+            )
+            store.messages.addMessage(
+                session.id,
+                { role: 'user', content: { type: 'text', text: 'already sent by sibling' } },
+                'local-sibling-preinvoked'
+            )
+            store.messages.addMessage(
+                session.id,
+                { role: 'user', content: { type: 'text', text: 'second sibling send' } },
+                'local-sibling-preinvoked-newer'
+            )
+            store.messages.markMessagesInvoked(session.id, ['local-sibling-preinvoked'], 1_500)
+            store.messages.markMessagesInvoked(session.id, ['local-sibling-preinvoked-newer'], 1_800)
+            const handlers = new Map<string, (payload: unknown) => void>()
+            const activity: Array<{ sessionId: string; updatedAt: number }> = []
+            const webEvents: SyncEvent[] = []
+
+            registerSessionHandlers({
+                on: (event: string, handler: (payload: unknown) => void) => {
+                    handlers.set(event, handler)
+                },
+                to: () => ({ emit() {} })
+            } as never, {
+                store,
+                resolveSessionAccess: (sessionId) => {
+                    const stored = store.sessions.getSessionByNamespace(sessionId, 'default')
+                    return stored ? { ok: true, value: stored } : { ok: false, reason: 'not-found' }
+                },
+                emitAccessError: () => {},
+                onSessionActivity: (sessionId, updatedAt) => {
+                    activity.push({ sessionId, updatedAt })
+                    cache.recordSessionActivity(sessionId, updatedAt)
+                },
+                onWebappEvent: (event) => webEvents.push(event)
+            })
+
+            now = 2_000
+            handlers.get('messages-consumed')?.({
+                sid: session.id,
+                localIds: ['local-sibling-preinvoked', 'local-sibling-preinvoked-newer', 'local-unknown']
+            })
+
+            expect(store.messages.getLocalMessageStates(session.id, [
+                'local-sibling-preinvoked',
+                'local-sibling-preinvoked-newer'
+            ])).toEqual([
+                { localId: 'local-sibling-preinvoked', invokedAt: 1_500 },
+                { localId: 'local-sibling-preinvoked-newer', invokedAt: 1_800 }
+            ])
+            expect(activity).toEqual([{ sessionId: session.id, updatedAt: 1_000 }])
+            expect(store.sessions.getSession(session.id)?.updatedAt).toBe(1_000)
+            expect(events.filter((event) => event.type === 'session-updated')).toHaveLength(0)
+            expect(webEvents.filter((event) => event.type === 'messages-consumed').map((event) => event.invokedAt))
+                .toEqual([2_000])
+        } finally {
+            Date.now = originalDateNow
+        }
+    })
+
+    it('keeps the ACK timestamp for messages-consumed SSE when local IDs are unknown', () => {
+        const originalDateNow = Date.now
+        let now = 1_000
+        Date.now = () => now
+        try {
+            const store = new Store(':memory:')
+            const session = store.sessions.getOrCreateSession(
+                'session-cli-consumed-unknown-id',
+                { path: '/tmp/project', host: 'localhost', flavor: 'codex' },
+                null,
+                'default'
+            )
+            const handlers = new Map<string, (payload: unknown) => void>()
+            const webEvents: SyncEvent[] = []
+
+            registerSessionHandlers({
+                on: (event: string, handler: (payload: unknown) => void) => {
+                    handlers.set(event, handler)
+                },
+                to: () => ({ emit() {} })
+            } as never, {
+                store,
+                resolveSessionAccess: (sessionId) => {
+                    const stored = store.sessions.getSessionByNamespace(sessionId, 'default')
+                    return stored ? { ok: true, value: stored } : { ok: false, reason: 'not-found' }
+                },
+                emitAccessError: () => {},
+                onWebappEvent: (event) => webEvents.push(event)
+            })
+
+            now = 2_000
+            handlers.get('messages-consumed')?.({ sid: session.id, localIds: ['local-unknown'] })
+
+            expect(webEvents.filter((event) => event.type === 'messages-consumed').map((event) => event.invokedAt))
+                .toEqual([2_000])
+        } finally {
+            Date.now = originalDateNow
+        }
     })
 
     it('does not report session activity for CLI tool messages', () => {
@@ -3335,9 +3636,11 @@ describe('session model', () => {
                 'default'
             )
 
+            const updatedAtBeforeArchive = store.sessions.getSession(session.id)?.updatedAt
             cache.markSessionArchivedFromHub(session.id, 'Archived from hub (CLI unreachable)')
 
             const meta = cache.getSession(session.id)?.metadata as Record<string, unknown> | null | undefined
+            expect(store.sessions.getSession(session.id)?.updatedAt).toBe(updatedAtBeforeArchive)
             expect(meta?.lifecycleState).toBe('archived')
             expect(meta?.archivedBy).toBe('hub')
             expect(meta?.archiveReason).toBe('Archived from hub (CLI unreachable)')