diff --git a/hub/src/socket/handlers/cli/sessionHandlers.ts b/hub/src/socket/handlers/cli/sessionHandlers.ts index 7ffec6db..16769888 100644 --- a/hub/src/socket/handlers/cli/sessionHandlers.ts +++ b/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) => { diff --git a/hub/src/store/index.ts b/hub/src/store/index.ts index 878e8655..9e8440f3 100644 --- a/hub/src/store/index.ts +++ b/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() diff --git a/hub/src/store/messageStore.ts b/hub/src/store/messageStore.ts index 8256b3c1..cc1c5f38 100644 --- a/hub/src/store/messageStore.ts +++ b/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 } { diff --git a/hub/src/store/messages.test.ts b/hub/src/store/messages.test.ts index 81edf2a2..1540b53e 100644 --- a/hub/src/store/messages.test.ts +++ b/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() diff --git a/hub/src/store/messages.ts b/hub/src/store/messages.ts index 4fcd219f..b2a18b19 100644 --- a/hub/src/store/messages.ts +++ b/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( diff --git a/hub/src/sync/sessionActivity.ts b/hub/src/sync/sessionActivity.ts index 08c2f45e..80d5bce6 100644 --- a/hub/src/sync/sessionActivity.ts +++ b/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) } diff --git a/hub/src/sync/sessionModel.test.ts b/hub/src/sync/sessionModel.test.ts index 36aaaac7..c6f9b71a 100644 --- a/hub/src/sync/sessionModel.test.ts +++ b/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 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 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 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 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 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 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 | 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)')