From 7f82df87c8768f1e3acab13c7c135d45275bcad3 Mon Sep 17 00:00:00 2001 From: Qianli Liu <73509271+Liu-KM@users.noreply.github.com> Date: Sat, 25 Apr 2026 20:17:22 +0800 Subject: [PATCH] fix(hub): refresh session activity on completed turns (#524) --- hub/src/index.ts | 3 +- hub/src/socket/handlers/cli/index.ts | 6 +- .../socket/handlers/cli/sessionHandlers.ts | 7 +- hub/src/socket/server.ts | 4 +- hub/src/store/sessionStore.ts | 5 + hub/src/store/sessions.ts | 26 +++ hub/src/sync/messageService.ts | 4 +- hub/src/sync/sessionActivity.ts | 54 ++++++ hub/src/sync/sessionCache.ts | 34 ++++ hub/src/sync/sessionModel.test.ts | 164 ++++++++++++++++++ hub/src/sync/syncEngine.ts | 11 +- 11 files changed, 311 insertions(+), 7 deletions(-) create mode 100644 hub/src/sync/sessionActivity.ts diff --git a/hub/src/index.ts b/hub/src/index.ts index 3f6cbc29..bdc78958 100644 --- a/hub/src/index.ts +++ b/hub/src/index.ts @@ -180,7 +180,8 @@ async function main() { onSessionAlive: (payload) => syncEngine?.handleSessionAlive(payload), onSessionEnd: (payload) => syncEngine?.handleSessionEnd(payload), onMachineAlive: (payload) => syncEngine?.handleMachineAlive(payload), - onBackgroundTaskDelta: (sessionId, delta) => syncEngine?.handleBackgroundTaskDelta(sessionId, delta) + onBackgroundTaskDelta: (sessionId, delta) => syncEngine?.handleBackgroundTaskDelta(sessionId, delta), + onSessionActivity: (sessionId, updatedAt) => syncEngine?.recordSessionActivity(sessionId, updatedAt) }) syncEngine = new SyncEngine(store, socketServer.io, socketServer.rpcRegistry, sseManager) diff --git a/hub/src/socket/handlers/cli/index.ts b/hub/src/socket/handlers/cli/index.ts index 2d0ca220..a531437f 100644 --- a/hub/src/socket/handlers/cli/index.ts +++ b/hub/src/socket/handlers/cli/index.ts @@ -42,10 +42,11 @@ export type CliHandlersDeps = { onMachineAlive?: (payload: MachineAlivePayload) => void onWebappEvent?: (event: SyncEvent) => void onBackgroundTaskDelta?: (sessionId: string, delta: { started: number; completed: number }) => void + onSessionActivity?: (sessionId: string, updatedAt: number) => void } export function registerCliHandlers(socket: CliSocketWithData, deps: CliHandlersDeps): void { - const { io, store, rpcRegistry, terminalRegistry, onSessionAlive, onSessionEnd, onMachineAlive, onWebappEvent, onBackgroundTaskDelta } = deps + const { io, store, rpcRegistry, terminalRegistry, onSessionAlive, onSessionEnd, onMachineAlive, onWebappEvent, onBackgroundTaskDelta, onSessionActivity } = deps const terminalNamespace = io.of('/terminal') const namespace = typeof socket.data.namespace === 'string' ? socket.data.namespace : null @@ -105,7 +106,8 @@ export function registerCliHandlers(socket: CliSocketWithData, deps: CliHandlers onSessionAlive, onSessionEnd, onWebappEvent, - onBackgroundTaskDelta + onBackgroundTaskDelta, + onSessionActivity }) registerMachineHandlers(socket, { store, diff --git a/hub/src/socket/handlers/cli/sessionHandlers.ts b/hub/src/socket/handlers/cli/sessionHandlers.ts index ab14052d..940a7d6b 100644 --- a/hub/src/socket/handlers/cli/sessionHandlers.ts +++ b/hub/src/socket/handlers/cli/sessionHandlers.ts @@ -7,6 +7,7 @@ import type { SyncEvent } from '../../../sync/syncEngine' import { extractTodoWriteTodosFromMessageContent } from '../../../sync/todos' import { extractTeamStateFromMessageContent, applyTeamStateDelta } from '../../../sync/teams' import { extractBackgroundTaskDelta } from '../../../sync/backgroundTasks' +import { shouldRecordSessionActivity } from '../../../sync/sessionActivity' import type { CliSocketWithData } from '../../socketTypes' import type { AccessErrorReason, AccessResult } from './types' @@ -60,10 +61,11 @@ export type SessionHandlersDeps = { onSessionEnd?: (payload: SessionEndPayload) => void onWebappEvent?: (event: SyncEvent) => void onBackgroundTaskDelta?: (sessionId: string, delta: { started: number; completed: number }) => void + onSessionActivity?: (sessionId: string, updatedAt: number) => void } export function registerSessionHandlers(socket: CliSocketWithData, deps: SessionHandlersDeps): void { - const { store, resolveSessionAccess, emitAccessError, onSessionAlive, onSessionEnd, onWebappEvent, onBackgroundTaskDelta } = deps + const { store, resolveSessionAccess, emitAccessError, onSessionAlive, onSessionEnd, onWebappEvent, onBackgroundTaskDelta, onSessionActivity } = deps socket.on('message', (data: unknown) => { const parsed = messageSchema.safeParse(data) @@ -92,6 +94,9 @@ export function registerSessionHandlers(socket: CliSocketWithData, deps: Session const session = sessionAccess.value const msg = store.messages.addMessage(sid, content, localId) + if (shouldRecordSessionActivity(content)) { + onSessionActivity?.(sid, msg.createdAt) + } const todos = extractTodoWriteTodosFromMessageContent(content) if (todos) { diff --git a/hub/src/socket/server.ts b/hub/src/socket/server.ts index 19086a95..b03b31ec 100644 --- a/hub/src/socket/server.ts +++ b/hub/src/socket/server.ts @@ -40,6 +40,7 @@ export type SocketServerDeps = { onSessionEnd?: (payload: { sid: string; time: number }) => void onMachineAlive?: (payload: { machineId: string; time: number }) => void onBackgroundTaskDelta?: (sessionId: string, delta: { started: number; completed: number }) => void + onSessionActivity?: (sessionId: string, updatedAt: number) => void } export function createSocketServer(deps: SocketServerDeps): { @@ -115,7 +116,8 @@ export function createSocketServer(deps: SocketServerDeps): { onSessionEnd: deps.onSessionEnd, onMachineAlive: deps.onMachineAlive, onWebappEvent: deps.onWebappEvent, - onBackgroundTaskDelta: deps.onBackgroundTaskDelta + onBackgroundTaskDelta: deps.onBackgroundTaskDelta, + onSessionActivity: deps.onSessionActivity })) terminalNs.use(async (socket, next) => { diff --git a/hub/src/store/sessionStore.ts b/hub/src/store/sessionStore.ts index c6af8ad4..4be487f3 100644 --- a/hub/src/store/sessionStore.ts +++ b/hub/src/store/sessionStore.ts @@ -13,6 +13,7 @@ import { setSessionModelReasoningEffort, setSessionTeamState, setSessionTodos, + touchSessionUpdatedAt, updateSessionAgentState, updateSessionMetadata } from './sessions' @@ -80,6 +81,10 @@ export class SessionStore { return setSessionEffort(this.db, id, effort, namespace, options) } + touchSessionUpdatedAt(id: string, updatedAt: number, namespace: string): boolean { + return touchSessionUpdatedAt(this.db, id, updatedAt, namespace) + } + getSession(id: string): StoredSession | null { return getSession(this.db, id) } diff --git a/hub/src/store/sessions.ts b/hub/src/store/sessions.ts index 95a33c5a..82264b83 100644 --- a/hub/src/store/sessions.ts +++ b/hub/src/store/sessions.ts @@ -342,6 +342,32 @@ export function setSessionEffort( } } +export function touchSessionUpdatedAt( + db: Database, + id: string, + updatedAt: number, + namespace: string +): boolean { + try { + const result = db.prepare(` + UPDATE sessions + SET updated_at = @updated_at, + seq = seq + 1 + WHERE id = @id + AND namespace = @namespace + AND updated_at < @updated_at + `).run({ + id, + namespace, + updated_at: updatedAt + }) + + return result.changes === 1 + } catch { + return false + } +} + export function getSession(db: Database, id: string): StoredSession | null { const row = db.prepare('SELECT * FROM sessions WHERE id = ?').get(id) as DbSessionRow | undefined return row ? toStoredSession(row) : null diff --git a/hub/src/sync/messageService.ts b/hub/src/sync/messageService.ts index c5bfd316..e9be24b8 100644 --- a/hub/src/sync/messageService.ts +++ b/hub/src/sync/messageService.ts @@ -7,7 +7,8 @@ export class MessageService { constructor( private readonly store: Store, private readonly io: Server, - private readonly publisher: EventPublisher + private readonly publisher: EventPublisher, + private readonly onSessionActivity?: (sessionId: string, updatedAt: number) => void ) { } @@ -87,6 +88,7 @@ export class MessageService { } const msg = this.store.messages.addMessage(sessionId, content, payload.localId ?? undefined) + this.onSessionActivity?.(sessionId, msg.createdAt) const update = { id: msg.id, diff --git a/hub/src/sync/sessionActivity.ts b/hub/src/sync/sessionActivity.ts new file mode 100644 index 00000000..08c2f45e --- /dev/null +++ b/hub/src/sync/sessionActivity.ts @@ -0,0 +1,54 @@ +import { unwrapRoleWrappedRecordEnvelope } from '@hapi/protocol' + +function asRecord(value: unknown): Record | null { + return value !== null && typeof value === 'object' && !Array.isArray(value) + ? value as Record + : null +} + +function hasHumanTextContent(content: unknown): boolean { + if (typeof content === 'string') { + return content.trim().length > 0 + } + + if (Array.isArray(content)) { + return content.some((block) => { + const record = asRecord(block) + return record?.type === 'text' + && typeof record.text === 'string' + && record.text.trim().length > 0 + }) + } + + const record = asRecord(content) + return record?.type === 'text' + && typeof record.text === 'string' + && 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) +} diff --git a/hub/src/sync/sessionCache.ts b/hub/src/sync/sessionCache.ts index be42edf6..ca4ea56f 100644 --- a/hub/src/sync/sessionCache.ts +++ b/hub/src/sync/sessionCache.ts @@ -264,6 +264,40 @@ export class SessionCache { }) } + recordSessionActivity(sessionId: string, updatedAt: number): void { + if (!Number.isFinite(updatedAt)) { + return + } + + const stored = this.store.sessions.getSession(sessionId) + if (!stored) { + return + } + + const nextUpdatedAt = Math.max(stored.updatedAt, updatedAt) + const touched = this.store.sessions.touchSessionUpdatedAt(sessionId, nextUpdatedAt, stored.namespace) + const session = this.sessions.get(sessionId) + + if (!session) { + if (touched) { + this.refreshSession(sessionId) + } + return + } + + if (nextUpdatedAt <= session.updatedAt && !touched) { + return + } + + session.updatedAt = Math.max(session.updatedAt, nextUpdatedAt) + this.publisher.emit({ + type: 'session-updated', + sessionId, + namespace: session.namespace, + data: { updatedAt: session.updatedAt } + }) + } + handleSessionEnd(payload: { sid: string; time: number }): void { const t = clampAliveTime(payload.time) ?? Date.now() diff --git a/hub/src/sync/sessionModel.test.ts b/hub/src/sync/sessionModel.test.ts index 37a9f3c9..f743fe65 100644 --- a/hub/src/sync/sessionModel.test.ts +++ b/hub/src/sync/sessionModel.test.ts @@ -3,6 +3,7 @@ import { toSessionSummary } from '@hapi/protocol' import type { SyncEvent } from '@hapi/protocol/types' import { Store } from '../store' import { RpcRegistry } from '../socket/rpcRegistry' +import { registerSessionHandlers } from '../socket/handlers/cli/sessionHandlers' import type { EventPublisher } from './eventPublisher' import { SessionCache } from './sessionCache' import { SyncEngine } from './syncEngine' @@ -265,6 +266,169 @@ describe('session model', () => { expect(cache.getSession(session.id)?.collaborationMode).toBe('default') }) + it('touches session updatedAt when new message activity is recorded', () => { + const store = new Store(':memory:') + const events: SyncEvent[] = [] + const cache = new SessionCache(store, createPublisher(events)) + + const session = cache.getOrCreateSession( + 'session-message-activity', + { path: '/tmp/project', host: 'localhost', flavor: 'codex' }, + null, + 'default' + ) + const activityAt = session.updatedAt + 60_000 + + cache.recordSessionActivity(session.id, activityAt) + + expect(store.sessions.getSession(session.id)?.updatedAt).toBe(activityAt) + expect(cache.getSession(session.id)?.updatedAt).toBe(activityAt) + expect(events).toContainEqual({ + type: 'session-updated', + sessionId: session.id, + namespace: 'default', + data: { updatedAt: activityAt } + }) + }) + + it('touches session updatedAt when web sends a message through sync engine', async () => { + const store = new Store(':memory:') + const engine = new SyncEngine( + store, + { of: () => ({ to: () => ({ emit() {} }) }) } as never, + new RpcRegistry(), + { broadcast() {} } as never + ) + + try { + const session = engine.getOrCreateSession( + 'session-web-message-activity', + { path: '/tmp/project', host: 'localhost', flavor: 'codex' }, + null, + 'default' + ) + const before = store.sessions.getSession(session.id)?.updatedAt ?? 0 + + await new Promise((resolve) => setTimeout(resolve, 2)) + await engine.sendMessage(session.id, { text: 'hello' }) + + const after = store.sessions.getSession(session.id)?.updatedAt ?? 0 + expect(after).toBeGreaterThan(before) + expect(engine.getSession(session.id)?.updatedAt).toBe(after) + } finally { + engine.stop() + } + }) + + it('reports session activity when CLI receives a turn-ready event over socket', () => { + const store = new Store(':memory:') + const events: SyncEvent[] = [] + const cache = new SessionCache(store, createPublisher(events)) + const session = cache.getOrCreateSession( + 'session-cli-message-activity', + { path: '/tmp/project', host: 'localhost', flavor: 'codex' }, + null, + 'default' + ) + const handlers = new Map void>() + const activity: Array<{ sessionId: string; updatedAt: number }> = [] + + 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 }) + } + }) + + handlers.get('message')?.({ + sid: session.id, + message: JSON.stringify({ + role: 'agent', + content: { + type: 'event', + data: { type: 'ready' } + } + }) + }) + + expect(activity).toHaveLength(1) + expect(activity[0].sessionId).toBe(session.id) + expect(activity[0].updatedAt).toBe(store.messages.getMessages(session.id, 1)[0]?.createdAt) + }) + + it('does not report session activity for CLI tool messages', () => { + const store = new Store(':memory:') + const events: SyncEvent[] = [] + const cache = new SessionCache(store, createPublisher(events)) + const session = cache.getOrCreateSession( + 'session-cli-tool-activity', + { path: '/tmp/project', host: 'localhost', flavor: 'codex' }, + null, + 'default' + ) + const handlers = new Map void>() + const activity: Array<{ sessionId: string; updatedAt: number }> = [] + + 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 }) + } + }) + + handlers.get('message')?.({ + sid: session.id, + message: JSON.stringify({ + role: 'agent', + content: { + type: 'codex', + data: { + type: 'tool-call', + name: 'CodexBash', + callId: 'call-1', + input: { cmd: 'date' } + } + } + }) + }) + handlers.get('message')?.({ + sid: session.id, + message: JSON.stringify({ + role: 'agent', + content: { + type: 'codex', + data: { + type: 'tool-call-result', + callId: 'call-1', + output: { stdout: 'Sat Apr 25' } + } + } + }) + }) + + expect(activity).toHaveLength(0) + }) + it('passes the stored model when respawning a resumed session', async () => { const store = new Store(':memory:') const engine = new SyncEngine( diff --git a/hub/src/sync/syncEngine.ts b/hub/src/sync/syncEngine.ts index b1c7998e..c75b2222 100644 --- a/hub/src/sync/syncEngine.ts +++ b/hub/src/sync/syncEngine.ts @@ -63,7 +63,12 @@ export class SyncEngine { this.eventPublisher = new EventPublisher(sseManager, (event) => this.resolveNamespace(event)) this.sessionCache = new SessionCache(store, this.eventPublisher) this.machineCache = new MachineCache(store, this.eventPublisher) - this.messageService = new MessageService(store, io, this.eventPublisher) + this.messageService = new MessageService( + store, + io, + this.eventPublisher, + (sessionId, updatedAt) => this.recordSessionActivity(sessionId, updatedAt) + ) this.rpcGateway = new RpcGateway(io, rpcRegistry) this.reloadAll() this.inactivityTimer = setInterval(() => this.expireInactive(), 5_000) @@ -220,6 +225,10 @@ export class SyncEngine { this.sessionCache.applyBackgroundTaskDelta(sessionId, delta) } + recordSessionActivity(sessionId: string, updatedAt: number): void { + this.sessionCache.recordSessionActivity(sessionId, updatedAt) + } + handleMachineAlive(payload: { machineId: string; time: number }): void { this.machineCache.handleMachineAlive(payload) }