From faf70c64ddebfb022bc30e0a7de0a0db2e918c93 Mon Sep 17 00:00:00 2001 From: weishu Date: Tue, 28 Jul 2026 10:11:44 +0800 Subject: [PATCH] refactor(sync): replace message reloads with incremental tail sync --- hub/src/store/index.ts | 20 +- hub/src/store/messageStore.ts | 30 +- hub/src/store/messages.test.ts | 71 + hub/src/store/messages.ts | 106 +- hub/src/store/migration-v12.test.ts | 7 +- hub/src/store/migration-v13.test.ts | 137 ++ hub/src/sync/messageService.test.ts | 97 + hub/src/sync/messageService.ts | 135 +- hub/src/sync/syncEngine.ts | 16 +- hub/src/web/routes/messages.test.ts | 131 +- hub/src/web/routes/messages.ts | 14 +- shared/src/apiTypes.test.ts | 40 +- shared/src/apiTypes.ts | 39 +- web/src/App.tsx | 6 +- web/src/api/client.ts | 20 + .../AssistantChat/HappyThread.test.tsx | 86 + .../components/AssistantChat/HappyThread.tsx | 316 ++- .../AssistantChat/QueuedMessagesBar.tsx | 3 +- web/src/components/SessionChat.test.ts | 10 +- web/src/components/SessionChat.tsx | 34 +- .../hooks/mutations/useSendMessage.test.tsx | 8 +- web/src/hooks/mutations/useSendMessage.ts | 3 - web/src/hooks/queries/useMessages.ts | 83 +- web/src/lib/message-window-store.test.ts | 1845 +++++++++-------- web/src/lib/message-window-store.ts | 1706 +++++++-------- .../lib/queued-state-reconciliation.test.ts | 15 +- web/src/lib/queued-state-reconciliation.ts | 4 +- web/src/router.tsx | 22 +- 28 files changed, 2927 insertions(+), 2077 deletions(-) create mode 100644 hub/src/store/migration-v13.test.ts diff --git a/hub/src/store/index.ts b/hub/src/store/index.ts index b5c594d7..b189382e 100644 --- a/hub/src/store/index.ts +++ b/hub/src/store/index.ts @@ -29,11 +29,12 @@ export { ScratchlistStore } from './scratchlistStore' export { SessionStore } from './sessionStore' export { UserStore } from './userStore' -const SCHEMA_VERSION: number = 12 +const SCHEMA_VERSION: number = 13 const REQUIRED_TABLES = [ 'sessions', 'machines', 'messages', + 'message_epochs', 'users', 'push_subscriptions', 'fcm_devices', @@ -138,6 +139,7 @@ export class Store { 9: () => this.migrateFromV9ToV10(), 10: () => this.migrateFromV10ToV11(), 11: () => this.migrateFromV11ToV12(), + 12: () => this.migrateFromV12ToV13(), }) if (currentVersion === 0) { @@ -245,6 +247,12 @@ export class Store { ON messages(scheduled_at) WHERE scheduled_at IS NOT NULL AND invoked_at IS NULL; + CREATE TABLE IF NOT EXISTS message_epochs ( + session_id TEXT PRIMARY KEY, + epoch INTEGER NOT NULL DEFAULT 0, + FOREIGN KEY (session_id) REFERENCES sessions(id) ON DELETE CASCADE + ); + CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, platform TEXT NOT NULL, @@ -519,6 +527,16 @@ export class Store { `) } + private migrateFromV12ToV13(): void { + this.db.exec(` + CREATE TABLE IF NOT EXISTS message_epochs ( + session_id TEXT PRIMARY KEY, + epoch INTEGER NOT NULL DEFAULT 0, + FOREIGN KEY (session_id) REFERENCES sessions(id) ON DELETE CASCADE + ) + `) + } + private getSessionColumnNames(): Set { const rows = this.db.prepare('PRAGMA table_info(sessions)').all() as Array<{ name: string }> return new Set(rows.map((row) => row.name)) diff --git a/hub/src/store/messageStore.ts b/hub/src/store/messageStore.ts index 38cd443e..8256b3c1 100644 --- a/hub/src/store/messageStore.ts +++ b/hub/src/store/messageStore.ts @@ -10,6 +10,10 @@ import { getFirstMessages, getDeliverableMessagesAfter, getMessagesByPosition, + getMessagesAfterPosition, + getNewestMessagePosition, + getMessageEpoch, + bumpMessageEpoch, getLocalMessageStates, getUninvokedLocalMessages, getMatureScheduledMessages, @@ -25,6 +29,7 @@ import { type CancelQueuedMessageResult, type LookupQueuedMessageResult, type LocalMessageState, + type MessagePosition, } from './messages' export class MessageStore { @@ -66,6 +71,27 @@ export class MessageStore { return getMessagesByPosition(this.db, sessionId, limit, before) } + getMessagesAfterPosition( + sessionId: string, + limit: number, + after: MessagePosition, + until?: MessagePosition + ): StoredMessage[] { + return getMessagesAfterPosition(this.db, sessionId, limit, after, until) + } + + getNewestMessagePosition(sessionId: string): MessagePosition | null { + return getNewestMessagePosition(this.db, sessionId) + } + + getMessageEpoch(sessionId: string): number { + return getMessageEpoch(this.db, sessionId) + } + + bumpMessageEpoch(sessionId: string): number { + return bumpMessageEpoch(this.db, sessionId) + } + getLocalMessageStates(sessionId: string, localIds: string[]): LocalMessageState[] { return getLocalMessageStates(this.db, sessionId, localIds) } @@ -106,8 +132,8 @@ export class MessageStore { return lookupQueuedMessage(this.db, sessionId, messageId) } - deleteQueuedMessageById(sessionId: string, messageId: string): void { - deleteQueuedMessageById(this.db, sessionId, messageId) + deleteQueuedMessageById(sessionId: string, messageId: string): boolean { + return deleteQueuedMessageById(this.db, sessionId, messageId) } markMessagesInvoked(sessionId: string, localIds: string[], invokedAt: number): void { diff --git a/hub/src/store/messages.test.ts b/hub/src/store/messages.test.ts index 6c164b34..81edf2a2 100644 --- a/hub/src/store/messages.test.ts +++ b/hub/src/store/messages.test.ts @@ -24,6 +24,7 @@ describe('cancelQueuedMessage', () => { // Row should be gone from uninvoked list const remaining = store.messages.getUninvokedLocalMessages(session.id) expect(remaining).toHaveLength(0) + expect(store.messages.getMessageEpoch(session.id)).toBe(1) }) it('already-invoked: returns status=invoked with full message row, row stays in DB', () => { @@ -67,6 +68,7 @@ describe('cancelQueuedMessage', () => { if (second.status === 'cancelled') { expect(second.localId).toBeNull() } + expect(store.messages.getMessageEpoch(session.id)).toBe(1) }) it('non-existent messageId: returns status=cancelled with localId=null', () => { @@ -173,6 +175,75 @@ describe('cancelQueuedMessage', () => { }) }) +describe('position pagination and structural epochs', () => { + it('returns rows strictly after a cursor and respects an inclusive snapshot head', () => { + const store = makeStore() + const session = makeSession(store, 'position-after') + const first = store.messages.addMessage(session.id, { text: 'first' }) + const second = store.messages.addMessage(session.id, { text: 'second' }) + const third = store.messages.addMessage(session.id, { text: 'third' }) + store.messages.addMessage(session.id, { text: 'fourth' }) + + const rows = store.messages.getMessagesAfterPosition( + session.id, + 10, + { at: first.invokedAt ?? first.createdAt, seq: first.seq }, + { at: third.invokedAt ?? third.createdAt, seq: third.seq } + ) + + expect(rows.map((message) => message.id)).toEqual([second.id, third.id]) + }) + + it('reports the newest composite position', () => { + const store = makeStore() + const session = makeSession(store, 'position-head') + const first = store.messages.addMessage(session.id, { text: 'first' }) + const second = store.messages.addMessage(session.id, { text: 'second' }) + + expect(store.messages.getNewestMessagePosition(session.id)).toEqual({ + at: second.invokedAt ?? second.createdAt, + seq: second.seq + }) + expect(first.seq).toBeLessThan(second.seq) + }) + + it('bumps both epochs when session history is merged', () => { + const store = makeStore() + const source = makeSession(store, 'epoch-merge-source') + const target = makeSession(store, 'epoch-merge-target') + store.messages.addMessage(source.id, { text: 'source' }) + store.messages.addMessage(target.id, { text: 'target' }) + + const result = store.messages.mergeSessionMessages(source.id, target.id) + + expect(result.moved).toBe(1) + expect(store.messages.getMessageEpoch(source.id)).toBe(1) + expect(store.messages.getMessageEpoch(target.id)).toBe(1) + }) + + it('bumps the target epoch when a copied message lands behind the cached head', () => { + const store = makeStore() + const target = makeSession(store, 'epoch-copy-target') + const head = store.messages.addMessage(target.id, { text: 'head' }) + const headPosition = { + at: head.invokedAt ?? head.createdAt, + seq: head.seq + } + + const copied = store.messages.copyMessageToSession(target.id, { + content: { text: 'historical' }, + createdAt: headPosition.at - 1_000, + localId: null, + invokedAt: headPosition.at - 1_000, + scheduledAt: null + }) + + expect(copied.seq).toBeGreaterThan(head.seq) + expect(store.messages.getMessagesAfterPosition(target.id, 10, headPosition)).toEqual([]) + expect(store.messages.getMessageEpoch(target.id)).toBe(1) + }) +}) + describe('addMessage: scheduledAt invariants', () => { it('rejects scheduledAt without a localId — would silently invoke immediately', () => { const store = makeStore() diff --git a/hub/src/store/messages.ts b/hub/src/store/messages.ts index 4994b3c4..4fcd219f 100644 --- a/hub/src/store/messages.ts +++ b/hub/src/store/messages.ts @@ -15,6 +15,11 @@ type DbMessageRow = { scheduled_at: number | null } +export type MessagePosition = { + at: number + seq: number +} + function toStoredMessage(row: DbMessageRow): StoredMessage { return { id: row.id, @@ -144,6 +149,11 @@ export function copyMessageToSession( if (!row) { throw new Error('Failed to copy message into target session') } + + // Copies preserve the source display timestamp, so a new high-seq row can + // still land behind a Web client's cached composite tail cursor. Mark the + // target history as structurally changed so incremental readers reset. + bumpMessageEpoch(db, sessionId) return toStoredMessage(row) } @@ -222,7 +232,7 @@ export function getMessagesByPosition( db: Database, sessionId: string, limit: number, - before?: { at: number; seq: number } + before?: MessagePosition ): StoredMessage[] { const safeLimit = Number.isFinite(limit) ? Math.max(1, Math.min(200, limit)) : 200 const beforeClause = before @@ -245,6 +255,72 @@ export function getMessagesByPosition( return rows.reverse().map(toStoredMessage) } +/** Return messages strictly after a display-position cursor in ascending order. + * `until`, when supplied, is an inclusive fixed snapshot head so a catch-up + * loop does not chase messages appended while it is running. */ +export function getMessagesAfterPosition( + db: Database, + sessionId: string, + limit: number, + after: MessagePosition, + until?: MessagePosition +): StoredMessage[] { + const safeLimit = Number.isFinite(limit) ? Math.max(1, Math.min(200, limit)) : 200 + const untilClause = until + ? `AND ( + COALESCE(invoked_at, created_at) < @untilAt + OR (COALESCE(invoked_at, created_at) = @untilAt AND seq <= @untilSeq) + )` + : '' + const rows = db.prepare(` + SELECT *, COALESCE(invoked_at, created_at) AS position_at + FROM messages + WHERE session_id = @sessionId + AND ( + COALESCE(invoked_at, created_at) > @afterAt + OR (COALESCE(invoked_at, created_at) = @afterAt AND seq > @afterSeq) + ) + ${untilClause} + ORDER BY position_at ASC, seq ASC + LIMIT @limit + `).all({ + sessionId, + afterAt: after.at, + afterSeq: after.seq, + untilAt: until?.at ?? null, + untilSeq: until?.seq ?? null, + limit: safeLimit + }) as DbMessageRow[] + return rows.map(toStoredMessage) +} + +export function getNewestMessagePosition(db: Database, sessionId: string): MessagePosition | null { + const row = db.prepare(` + SELECT COALESCE(invoked_at, created_at) AS position_at, seq + FROM messages + WHERE session_id = ? + ORDER BY position_at DESC, seq DESC + LIMIT 1 + `).get(sessionId) as { position_at: number; seq: number } | undefined + return row ? { at: row.position_at, seq: row.seq } : null +} + +export function getMessageEpoch(db: Database, sessionId: string): number { + const row = db.prepare( + 'SELECT epoch FROM message_epochs WHERE session_id = ?' + ).get(sessionId) as { epoch: number } | undefined + return row?.epoch ?? 0 +} + +export function bumpMessageEpoch(db: Database, sessionId: string): number { + db.prepare(` + INSERT INTO message_epochs (session_id, epoch) + VALUES (?, 1) + ON CONFLICT(session_id) DO UPDATE SET epoch = epoch + 1 + `).run(sessionId) + return getMessageEpoch(db, sessionId) +} + /** Returns user messages that have a localId but no invoked_at. * Includes future scheduled messages — used to surface all queued messages * (including scheduled) for the Web floating bar on refresh / secondary clients. */ @@ -471,11 +547,15 @@ export function cancelQueuedMessage( return { status: 'invoked' as const, message: toStoredMessage(row) } } - db.prepare(` + const deleted = db.prepare(` DELETE FROM messages WHERE session_id = ? AND (id = ? OR local_id = ?) AND invoked_at IS NULL `).run(sessionId, messageId, messageId) + if (deleted.changes > 0) { + bumpMessageEpoch(db, sessionId) + } + return { status: 'cancelled' as const, localId: row.local_id } })() } @@ -525,11 +605,18 @@ export function deleteQueuedMessageById( db: Database, sessionId: string, messageId: string -): void { - db.prepare(` - DELETE FROM messages - WHERE session_id = ? AND (id = ? OR local_id = ?) AND invoked_at IS NULL - `).run(sessionId, messageId, messageId) +): boolean { + return db.transaction(() => { + const deleted = db.prepare(` + DELETE FROM messages + WHERE session_id = ? AND (id = ? OR local_id = ?) AND invoked_at IS NULL + `).run(sessionId, messageId, messageId) + if (deleted.changes > 0) { + bumpMessageEpoch(db, sessionId) + return true + } + return false + })() } /** Mark messages as invoked at the given server timestamp. @@ -602,6 +689,11 @@ export function mergeSessionMessages( 'UPDATE messages SET session_id = ? WHERE session_id = ?' ).run(toSessionId, fromSessionId) + if (result.changes > 0) { + bumpMessageEpoch(db, fromSessionId) + bumpMessageEpoch(db, toSessionId) + } + db.exec('COMMIT') return { moved: result.changes, oldMaxSeq, newMaxSeq } } catch (error) { diff --git a/hub/src/store/migration-v12.test.ts b/hub/src/store/migration-v12.test.ts index 13986cb3..82e07a7a 100644 --- a/hub/src/store/migration-v12.test.ts +++ b/hub/src/store/migration-v12.test.ts @@ -32,7 +32,7 @@ describe('Store V11→V12 migration: session_scratchlist table', () => { expect(rows).toHaveLength(1) }) - it('V11 DB migrates to V12 via Store: session_scratchlist created', () => { + it('V11 DB migrates through V13 via Store: session_scratchlist created', () => { const dir = mkdtempSync(join(tmpdir(), 'hapi-migration-v12-test-')) const dbPath = join(dir, 'test.db') let store: Store | undefined @@ -61,7 +61,7 @@ describe('Store V11→V12 migration: session_scratchlist table', () => { } }) - it('V9 DB migrates to V12 (multi-hop service_tier + fcm_devices + scratchlist)', () => { + it('V9 DB migrates through V13 (multi-hop service_tier + fcm_devices + scratchlist)', () => { const dir = mkdtempSync(join(tmpdir(), 'hapi-migration-v9-to-v12-')) const dbPath = join(dir, 'test.db') let store: Store | undefined @@ -84,7 +84,7 @@ describe('Store V11→V12 migration: session_scratchlist table', () => { } }) - it('V12 DB reopen is idempotent: schema unchanged', () => { + it('current DB reopen is idempotent: schema unchanged', () => { const dir = mkdtempSync(join(tmpdir(), 'hapi-migration-v12-idempotent-')) const dbPath = join(dir, 'test.db') let store1: Store | undefined @@ -334,4 +334,3 @@ function createV11Schema(db: Database): void { CREATE INDEX IF NOT EXISTS idx_fcm_devices_token ON fcm_devices(token); `) } - diff --git a/hub/src/store/migration-v13.test.ts b/hub/src/store/migration-v13.test.ts new file mode 100644 index 00000000..4b11fe35 --- /dev/null +++ b/hub/src/store/migration-v13.test.ts @@ -0,0 +1,137 @@ +import { describe, expect, it } from 'bun:test' +import { Database } from 'bun:sqlite' +import { mkdtempSync, rmSync } from 'node:fs' +import { join } from 'node:path' +import { tmpdir } from 'node:os' +import { Store } from './index' + +describe('Store V12→V13 migration: message_epochs', () => { + it('fresh DB has message_epochs table', () => { + const store = new Store(':memory:') + expect(tableExists(store, 'message_epochs')).toBe(true) + store.close() + }) + + it('V12 DB migrates to V13 and preserves existing messages', () => { + const dir = mkdtempSync(join(tmpdir(), 'hapi-migration-v13-test-')) + const dbPath = join(dir, 'test.db') + let store: Store | undefined + try { + const db = new Database(dbPath, { create: true, readwrite: true, strict: true }) + db.exec('PRAGMA journal_mode = WAL') + db.exec('PRAGMA foreign_keys = ON') + createV12Schema(db) + db.exec(` + INSERT INTO sessions (id, created_at, updated_at) VALUES ('session-1', 1, 1); + INSERT INTO messages (id, session_id, content, created_at, seq, invoked_at) + VALUES ('message-1', 'session-1', '{}', 1, 1, 1); + PRAGMA user_version = 12; + `) + db.close() + + store = new Store(dbPath) + expect(tableExists(store, 'message_epochs')).toBe(true) + expect(store.messages.getMessageEpoch('session-1')).toBe(0) + expect(store.messages.getMessages('session-1')).toHaveLength(1) + } finally { + store?.close() + rmSync(dir, { recursive: true, force: true }) + } + }) +}) + +function tableExists(store: Store, name: string): boolean { + const db: Database = (store as unknown as { db: Database }).db + const row = db.prepare( + "SELECT name FROM sqlite_master WHERE type='table' AND name=?" + ).get(name) as { name: string } | null + return row !== null +} + +function createV12Schema(db: Database): void { + db.exec(` + CREATE TABLE sessions ( + id TEXT PRIMARY KEY, + tag TEXT, + namespace TEXT NOT NULL DEFAULT 'default', + machine_id TEXT, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL, + metadata TEXT, + metadata_version INTEGER DEFAULT 1, + agent_state TEXT, + agent_state_version INTEGER DEFAULT 1, + model TEXT, + model_reasoning_effort TEXT, + effort TEXT, + service_tier TEXT, + todos TEXT, + todos_updated_at INTEGER, + team_state TEXT, + team_state_updated_at INTEGER, + active INTEGER DEFAULT 0, + active_at INTEGER, + seq INTEGER DEFAULT 0 + ); + CREATE TABLE machines ( + id TEXT PRIMARY KEY, + namespace TEXT NOT NULL DEFAULT 'default', + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL, + metadata TEXT, + metadata_version INTEGER DEFAULT 1, + runner_state TEXT, + runner_state_version INTEGER DEFAULT 1, + active INTEGER DEFAULT 0, + active_at INTEGER, + seq INTEGER DEFAULT 0 + ); + CREATE TABLE messages ( + id TEXT PRIMARY KEY, + session_id TEXT NOT NULL, + content TEXT NOT NULL, + created_at INTEGER NOT NULL, + seq INTEGER NOT NULL, + local_id TEXT, + invoked_at INTEGER, + scheduled_at INTEGER, + FOREIGN KEY (session_id) REFERENCES sessions(id) ON DELETE CASCADE + ); + CREATE TABLE users ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + platform TEXT NOT NULL, + platform_user_id TEXT NOT NULL, + namespace TEXT NOT NULL DEFAULT 'default', + created_at INTEGER NOT NULL, + UNIQUE(platform, platform_user_id) + ); + CREATE TABLE push_subscriptions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + namespace TEXT NOT NULL, + endpoint TEXT NOT NULL, + p256dh TEXT NOT NULL, + auth TEXT NOT NULL, + created_at INTEGER NOT NULL, + UNIQUE(namespace, endpoint) + ); + CREATE TABLE fcm_devices ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + namespace TEXT NOT NULL, + token TEXT NOT NULL, + platform TEXT NOT NULL, + device_id TEXT NOT NULL, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL, + UNIQUE(namespace, device_id, platform) + ); + CREATE TABLE session_scratchlist ( + session_id TEXT NOT NULL, + entry_id TEXT NOT NULL, + text TEXT NOT NULL, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL, + PRIMARY KEY (session_id, entry_id), + FOREIGN KEY (session_id) REFERENCES sessions(id) ON DELETE CASCADE + ); + `) +} diff --git a/hub/src/sync/messageService.test.ts b/hub/src/sync/messageService.test.ts index 794ca0aa..d4338147 100644 --- a/hub/src/sync/messageService.test.ts +++ b/hub/src/sync/messageService.test.ts @@ -243,6 +243,11 @@ describe('MessageService message pagination', () => { expect(page.messages.map((message) => message.id)).toEqual([second.id, third.id]) expect(page.page.nextBeforeAt).toBe(2_000) expect(page.page.nextBeforeSeq).toBe(second.seq) + expect(page.page.direction).toBe('latest') + expect(page.page.epoch).toBe(0) + expect(page.page.reset).toBe(false) + expect(page.page.snapshotHeadAt).toBe(3_000) + expect(page.page.snapshotHeadSeq).toBe(third.seq) expect(page.page.hasMore).toBe(true) expect(first.id).toBeDefined() }) @@ -266,6 +271,7 @@ describe('MessageService message pagination', () => { expect(older.messages.map((message) => message.id)).toEqual([first.id]) expect(older.page.nextBeforeAt).toBe(1_000) expect(older.page.nextBeforeSeq).toBe(first.seq) + expect(older.page.direction).toBe('before') expect(older.page.hasMore).toBe(false) expect(second.id).toBeDefined() expect(third.id).toBeDefined() @@ -305,6 +311,97 @@ describe('MessageService message pagination', () => { expect(page.page.nextBeforeSeq).toBe(invoked.seq) expect(page.page.hasMore).toBe(true) }) + + it('pages forward to a fixed snapshot head', () => { + const store = makeStore() + const session = makeSession(store, 'page-after') + const first = store.messages.addMessage(session.id, 'first', 'local-first') + const second = store.messages.addMessage(session.id, 'second', 'local-second') + const third = store.messages.addMessage(session.id, 'third', 'local-third') + const fourth = store.messages.addMessage(session.id, 'fourth', 'local-fourth') + store.messages.markMessagesInvoked(session.id, ['local-first'], 1_000) + store.messages.markMessagesInvoked(session.id, ['local-second'], 2_000) + store.messages.markMessagesInvoked(session.id, ['local-third'], 3_000) + store.messages.markMessagesInvoked(session.id, ['local-fourth'], 4_000) + + const service = makeService(store) + const firstDelta = service.getMessagesPage(session.id, { + limit: 1, + after: { at: 1_000, seq: first.seq }, + epoch: 0 + }) + + expect(firstDelta.messages.map((message) => message.id)).toEqual([second.id]) + expect(firstDelta.page).toMatchObject({ + direction: 'after', + nextAfterAt: 2_000, + nextAfterSeq: second.seq, + snapshotHeadAt: 4_000, + snapshotHeadSeq: fourth.seq, + hasMore: true, + reset: false + }) + + const fifth = store.messages.addMessage(session.id, 'fifth', 'local-fifth') + store.messages.markMessagesInvoked(session.id, ['local-fifth'], 5_000) + + const secondDelta = service.getMessagesPage(session.id, { + limit: 10, + after: { at: firstDelta.page.nextAfterAt!, seq: firstDelta.page.nextAfterSeq! }, + until: { at: firstDelta.page.snapshotHeadAt!, seq: firstDelta.page.snapshotHeadSeq! }, + epoch: firstDelta.page.epoch + }) + + expect(secondDelta.messages.map((message) => message.id)).toEqual([third.id, fourth.id]) + expect(secondDelta.page.hasMore).toBe(false) + expect(secondDelta.messages.some((message) => message.id === fifth.id)).toBe(false) + }) + + it('completes a fixed snapshot when no raw row remains before its head', () => { + const store = makeStore() + const session = makeSession(store, 'page-after-empty-snapshot-gap') + const first = store.messages.addMessage(session.id, 'first', 'local-first') + const latest = store.messages.addMessage(session.id, 'latest', 'local-latest') + store.messages.markMessagesInvoked(session.id, ['local-first'], 1_000) + store.messages.markMessagesInvoked(session.id, ['local-latest'], 4_000) + + const response = makeService(store).getMessagesPage(session.id, { + limit: 10, + after: { at: 2_000, seq: first.seq }, + until: { at: 3_000, seq: latest.seq }, + epoch: 0 + }) + + expect(response.messages).toEqual([]) + expect(response.page).toMatchObject({ + direction: 'after', + nextAfterAt: 3_000, + nextAfterSeq: latest.seq, + snapshotHeadAt: 3_000, + snapshotHeadSeq: latest.seq, + hasMore: false + }) + }) + + it('returns a reset latest page when the structural epoch changed', () => { + const store = makeStore() + const session = makeSession(store, 'page-after-reset') + const first = store.messages.addMessage(session.id, 'first', 'local-first') + store.messages.markMessagesInvoked(session.id, ['local-first'], 1_000) + const queued = store.messages.addMessage(session.id, 'queued', 'local-queued') + store.messages.cancelQueuedMessage(session.id, queued.id) + + const response = makeService(store).getMessagesPage(session.id, { + limit: 10, + after: { at: 1_000, seq: first.seq }, + epoch: 0 + }) + + expect(response.page.direction).toBe('latest') + expect(response.page.reset).toBe(true) + expect(response.page.epoch).toBe(1) + expect(response.messages.map((message) => message.id)).toEqual([first.id]) + }) }) describe('MessageService.getQueuedState', () => { diff --git a/hub/src/sync/messageService.ts b/hub/src/sync/messageService.ts index fed7ab7d..95b3fdc6 100644 --- a/hub/src/sync/messageService.ts +++ b/hub/src/sync/messageService.ts @@ -10,13 +10,25 @@ import { unwrapRoleWrappedRecordEnvelope } from '@hapi/protocol/messages' import { isObject } from '@hapi/protocol' -import type { QueuedStateResponse } from '@hapi/protocol/apiTypes' +import type { MessagesResponse, QueuedStateResponse } from '@hapi/protocol/apiTypes' import type { Server } from 'socket.io' import { randomUUID } from 'node:crypto' import type { Store, CancelQueuedMessageResult } from '../store' import { EventPublisher } from './eventPublisher' type StoredMessageForDelivery = ReturnType[number] +type MessagePosition = { at: number; seq: number } + +function messagePosition(message: StoredMessageForDelivery): MessagePosition { + return { + at: message.invokedAt ?? message.createdAt, + seq: message.seq + } +} + +function comparePosition(a: MessagePosition, b: MessagePosition): number { + return a.at !== b.at ? a.at - b.at : a.seq - b.seq +} function isWebVisibleStoredMessage(message: StoredMessageForDelivery): boolean { return !isRedundantGoalStatusEventContent(message.content) @@ -139,25 +151,54 @@ export class MessageService { getMessagesPage( sessionId: string, - options: { limit: number; before?: { at: number; seq: number } | null } - ): { - messages: DecryptedMessage[] - page: { + options: { limit: number - nextBeforeSeq: number | null - nextBeforeAt: number | null - hasMore: boolean + before?: MessagePosition | null + after?: MessagePosition | null + until?: MessagePosition | null + epoch?: number | null } - } { - let before = options.before ?? undefined - let pageRows = this.store.messages.getMessagesByPosition(sessionId, options.limit, before) + ): MessagesResponse { + const epoch = this.store.messages.getMessageEpoch(sessionId) + if (options.after) { + if (options.epoch !== undefined && options.epoch !== null && options.epoch !== epoch) { + return this.getLatestOrBeforeMessagesPage(sessionId, options.limit, null, epoch, true) + } + return this.getAfterMessagesPage( + sessionId, + options.limit, + options.after, + options.until ?? null, + epoch + ) + } + return this.getLatestOrBeforeMessagesPage( + sessionId, + options.limit, + options.before ?? null, + epoch, + false + ) + } + + private getLatestOrBeforeMessagesPage( + sessionId: string, + limit: number, + requestedBefore: MessagePosition | null, + epoch: number, + reset: boolean + ): MessagesResponse { + const direction = requestedBefore ? 'before' as const : 'latest' as const + const snapshotHead = this.store.messages.getNewestMessagePosition(sessionId) + let before = requestedBefore ?? undefined + let pageRows = this.store.messages.getMessagesByPosition(sessionId, limit, requestedBefore ?? undefined) // Latest-page request (no cursor): also include uninvoked local user messages // out-of-band, so refresh / secondary clients can still see queued rows even // when their position key (createdAt) places them outside the latest page. // The cursor stays anchored to pageRows so out-of-band rows don't affect // pagination of older pages. - let queuedRows = before === undefined + let queuedRows = requestedBefore === null ? this.store.messages.getUninvokedLocalMessages(sessionId) : [] @@ -190,7 +231,7 @@ export class MessageService { while (messages.length === 0 && hasMore && oldestSeq !== null && oldestPositionAt !== null) { before = { at: oldestPositionAt, seq: oldestSeq } - pageRows = this.store.messages.getMessagesByPosition(sessionId, options.limit, before) + pageRows = this.store.messages.getMessagesByPosition(sessionId, limit, before) queuedRows = [] byId = new Map() @@ -219,9 +260,75 @@ export class MessageService { return { messages, page: { - limit: options.limit, + direction, + limit, + epoch, + reset, nextBeforeSeq: oldestSeq, nextBeforeAt: oldestPositionAt, + nextAfterSeq: null, + nextAfterAt: null, + snapshotHeadSeq: snapshotHead?.seq ?? null, + snapshotHeadAt: snapshotHead?.at ?? null, + hasMore + } + } + } + + private getAfterMessagesPage( + sessionId: string, + limit: number, + after: MessagePosition, + requestedUntil: MessagePosition | null, + epoch: number + ): MessagesResponse { + const currentHead = this.store.messages.getNewestMessagePosition(sessionId) + const snapshotHead = currentHead && requestedUntil + ? (comparePosition(requestedUntil, currentHead) <= 0 ? requestedUntil : currentHead) + : requestedUntil ?? currentHead + + if (!snapshotHead || comparePosition(snapshotHead, after) <= 0) { + return { + messages: [], + page: { + direction: 'after', + limit, + epoch, + reset: false, + nextBeforeSeq: null, + nextBeforeAt: null, + nextAfterSeq: after.seq, + nextAfterAt: after.at, + snapshotHeadSeq: snapshotHead?.seq ?? null, + snapshotHeadAt: snapshotHead?.at ?? null, + hasMore: false + } + } + } + + const pageRows = this.store.messages.getMessagesAfterPosition( + sessionId, + limit, + after, + snapshotHead + ) + const last = pageRows[pageRows.length - 1] ?? null + const nextAfter = last ? messagePosition(last) : snapshotHead + const hasMore = last !== null && comparePosition(nextAfter, snapshotHead) < 0 + + return { + messages: toVisibleDecryptedMessages(pageRows), + page: { + direction: 'after', + limit, + epoch, + reset: false, + nextBeforeSeq: null, + nextBeforeAt: null, + nextAfterSeq: nextAfter.seq, + nextAfterAt: nextAfter.at, + snapshotHeadSeq: snapshotHead.seq, + snapshotHeadAt: snapshotHead.at, hasMore } } diff --git a/hub/src/sync/syncEngine.ts b/hub/src/sync/syncEngine.ts index ec0ec31c..3654b9f2 100644 --- a/hub/src/sync/syncEngine.ts +++ b/hub/src/sync/syncEngine.ts @@ -8,7 +8,7 @@ */ import { isKnownFlavor, type LocalResumeTarget, type ResumableSession } from '@hapi/protocol' -import type { CursorChatStoreStatus, CursorMigrateOutcome, CursorMigrateToAcpRequest, QueuedStateResponse, SlashCommandsResponse } from '@hapi/protocol/apiTypes' +import type { CursorChatStoreStatus, CursorMigrateOutcome, CursorMigrateToAcpRequest, MessagesResponse, QueuedStateResponse, SlashCommandsResponse } from '@hapi/protocol/apiTypes' import type { AgentFlavor, CodexCollaborationMode, DecryptedMessage, PermissionMode, Session, SyncEvent } from '@hapi/protocol/types' import { unwrapRoleWrappedRecordEnvelope } from '@hapi/protocol/messages' import type { Server } from 'socket.io' @@ -323,16 +323,14 @@ export class SyncEngine { getMessagesPage( sessionId: string, - options: { limit: number; before?: { at: number; seq: number } | null } - ): { - messages: DecryptedMessage[] - page: { + options: { limit: number - nextBeforeSeq: number | null - nextBeforeAt: number | null - hasMore: boolean + before?: { at: number; seq: number } | null + after?: { at: number; seq: number } | null + until?: { at: number; seq: number } | null + epoch?: number | null } - } { + ): MessagesResponse { return this.messageService.getMessagesPage(sessionId, options) } diff --git a/hub/src/web/routes/messages.test.ts b/hub/src/web/routes/messages.test.ts index 346fc693..f43b0197 100644 --- a/hub/src/web/routes/messages.test.ts +++ b/hub/src/web/routes/messages.test.ts @@ -11,6 +11,8 @@ import type { SyncEngine } from '../../sync/syncEngine' import type { WebAppEnv } from '../middleware/auth' import { createMessagesRoutes } from './messages' +type GetMessagesPage = SyncEngine['getMessagesPage'] + // TS note: engine is cast to unknown→SyncEngine so test helpers don't need to // satisfy the full SyncEngine shape (only the subset the route under test uses). @@ -21,6 +23,7 @@ import { createMessagesRoutes } from './messages' function createApp(opts: { active?: boolean sendMessage?: (sessionId: string, payload: unknown) => Promise + getMessagesPage?: GetMessagesPage getQueuedState?: (sessionId: string, localIds: string[]) => { queuedLocalIds: string[] invokedLocalMessages: Array<{ localId: string; invokedAt: number }> @@ -40,6 +43,22 @@ function createApp(opts: { .map((localId) => ({ localId, invokedAt: 1_000 })) } }) + const getMessagesPage = opts.getMessagesPage ?? (() => ({ + messages: [], + page: { + direction: 'latest', + limit: 50, + epoch: 0, + reset: false, + nextBeforeSeq: null, + nextBeforeAt: null, + nextAfterSeq: null, + nextAfterAt: null, + snapshotHeadSeq: null, + snapshotHeadAt: null, + hasMore: false + } + })) const engine = { resolveSessionAccess: () => ({ @@ -50,7 +69,7 @@ function createApp(opts: { sendMessage, getQueuedState, cancelQueuedMessage: async () => ({ status: 'cancelled' }), - getMessagesPage: () => ({ messages: [], page: {} }), + getMessagesPage, } as unknown as SyncEngine const app = new Hono() @@ -63,6 +82,116 @@ function createApp(opts: { return { app, sentMessages, queuedStateCalls } } +describe('GET /api/sessions/:id/messages', () => { + it('uses latest mode by default and returns the full page metadata', async () => { + const calls: Array<{ sessionId: string; options: Parameters[1] }> = [] + const { app } = createApp({ + getMessagesPage: (sessionId, options) => { + calls.push({ sessionId, options }) + return { + messages: [], + page: { + direction: 'latest', + limit: options.limit, + epoch: 4, + reset: false, + nextBeforeSeq: 10, + nextBeforeAt: 1_000, + nextAfterSeq: null, + nextAfterAt: null, + snapshotHeadSeq: 20, + snapshotHeadAt: 2_000, + hasMore: true + } + } + } + }) + + const response = await app.request('/api/sessions/session-1/messages') + + expect(response.status).toBe(200) + expect(calls).toEqual([{ + sessionId: 'session-1', + options: { limit: 50, before: null, after: null, until: null, epoch: null } + }]) + expect(await response.json()).toEqual({ + messages: [], + page: { + direction: 'latest', + limit: 50, + epoch: 4, + reset: false, + nextBeforeSeq: 10, + nextBeforeAt: 1_000, + nextAfterSeq: null, + nextAfterAt: null, + snapshotHeadSeq: 20, + snapshotHeadAt: 2_000, + hasMore: true + } + }) + }) + + it('forwards after, snapshot-head, epoch, and limit query parameters', async () => { + const calls: Array<{ sessionId: string; options: Parameters[1] }> = [] + const { app } = createApp({ + getMessagesPage: (sessionId, options) => { + calls.push({ sessionId, options }) + return { + messages: [], + page: { + direction: 'after', + limit: options.limit, + epoch: options.epoch ?? 0, + reset: false, + nextBeforeSeq: null, + nextBeforeAt: null, + nextAfterSeq: 11, + nextAfterAt: 1_100, + snapshotHeadSeq: 20, + snapshotHeadAt: 2_000, + hasMore: true + } + } + } + }) + + const response = await app.request( + '/api/sessions/session-1/messages?afterAt=1000&afterSeq=10&untilAt=2000&untilSeq=20&epoch=3&limit=25' + ) + + expect(response.status).toBe(200) + expect(calls).toEqual([{ + sessionId: 'session-1', + options: { + limit: 25, + before: null, + after: { at: 1_000, seq: 10 }, + until: { at: 2_000, seq: 20 }, + epoch: 3 + } + }]) + }) + + it('rejects mixed directional cursors before calling the engine', async () => { + let called = false + const { app } = createApp({ + getMessagesPage: () => { + called = true + throw new Error('must not be called') + } + }) + + const response = await app.request( + '/api/sessions/session-1/messages?beforeAt=1000&beforeSeq=10&afterAt=2000&afterSeq=20' + ) + + expect(response.status).toBe(400) + expect(await response.json()).toMatchObject({ error: 'Invalid query' }) + expect(called).toBe(false) + }) +}) + // --------------------------------------------------------------------------- // #2 server-side scheduledAt upper bound // --------------------------------------------------------------------------- diff --git a/hub/src/web/routes/messages.ts b/hub/src/web/routes/messages.ts index f9bbc333..9e3442ac 100644 --- a/hub/src/web/routes/messages.ts +++ b/hub/src/web/routes/messages.ts @@ -28,7 +28,19 @@ export function createMessagesRoutes(getSyncEngine: () => SyncEngine | null): Ho const before = parsed.data.beforeAt !== undefined && parsed.data.beforeSeq !== undefined ? { at: parsed.data.beforeAt, seq: parsed.data.beforeSeq } : null - return c.json(engine.getMessagesPage(sessionId, { limit, before })) + const after = parsed.data.afterAt !== undefined && parsed.data.afterSeq !== undefined + ? { at: parsed.data.afterAt, seq: parsed.data.afterSeq } + : null + const until = parsed.data.untilAt !== undefined && parsed.data.untilSeq !== undefined + ? { at: parsed.data.untilAt, seq: parsed.data.untilSeq } + : null + return c.json(engine.getMessagesPage(sessionId, { + limit, + before, + after, + until, + epoch: parsed.data.epoch ?? null + })) }) app.delete('/sessions/:id/messages/:messageId', async (c) => { diff --git a/shared/src/apiTypes.test.ts b/shared/src/apiTypes.test.ts index 8decaff9..88c1dd68 100644 --- a/shared/src/apiTypes.test.ts +++ b/shared/src/apiTypes.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from 'vitest' -import { ListCodexSessionsRpcResponseSchema } from './apiTypes' +import { ListCodexSessionsRpcResponseSchema, MessagesQuerySchema } from './apiTypes' describe('ListCodexSessionsRpcResponseSchema', () => { it('preserves Codex session messages when parsing runner RPC responses', () => { @@ -29,3 +29,41 @@ describe('ListCodexSessionsRpcResponseSchema', () => { } }) }) + +describe('MessagesQuerySchema', () => { + it('parses a forward cursor with a bounded snapshot and epoch', () => { + expect(MessagesQuerySchema.parse({ + afterAt: '1000', + afterSeq: '10', + untilAt: '2000', + untilSeq: '20', + epoch: '3', + limit: '200' + })).toEqual({ + afterAt: 1000, + afterSeq: 10, + untilAt: 2000, + untilSeq: 20, + epoch: 3, + limit: 200 + }) + }) + + it('rejects mixed before and after directions', () => { + expect(MessagesQuerySchema.safeParse({ + beforeAt: 1000, + beforeSeq: 10, + afterAt: 2000, + afterSeq: 20 + }).success).toBe(false) + }) + + it('rejects an unpaired or unscoped until cursor', () => { + expect(MessagesQuerySchema.safeParse({ untilAt: 2000, untilSeq: 20 }).success).toBe(false) + expect(MessagesQuerySchema.safeParse({ afterAt: 1000, afterSeq: 10, untilAt: 2000 }).success).toBe(false) + }) + + it('rejects epoch without a forward cursor', () => { + expect(MessagesQuerySchema.safeParse({ epoch: 1 }).success).toBe(false) + }) +}) diff --git a/shared/src/apiTypes.ts b/shared/src/apiTypes.ts index 9a122c86..d5a910ea 100644 --- a/shared/src/apiTypes.ts +++ b/shared/src/apiTypes.ts @@ -78,9 +78,16 @@ export type SessionResponse = { session: Session } export type MessagesResponse = { messages: DecryptedMessage[] page: { + direction: 'latest' | 'before' | 'after' limit: number + epoch: number + reset: boolean nextBeforeSeq: number | null nextBeforeAt: number | null + nextAfterSeq: number | null + nextAfterAt: number | null + snapshotHeadSeq: number | null + snapshotHeadAt: number | null hasMore: boolean } } @@ -330,10 +337,36 @@ export const MessagesQuerySchema = z.object({ limit: z.coerce.number().int().min(1).max(200).optional(), beforeSeq: z.coerce.number().int().min(1).optional(), beforeAt: z.coerce.number().int().min(0).optional(), -}).refine((data) => (data.beforeAt === undefined) === (data.beforeSeq === undefined), { - message: 'beforeAt and beforeSeq must be provided together', - path: ['beforeAt'], + afterSeq: z.coerce.number().int().min(1).optional(), + afterAt: z.coerce.number().int().min(0).optional(), + untilSeq: z.coerce.number().int().min(1).optional(), + untilAt: z.coerce.number().int().min(0).optional(), + epoch: z.coerce.number().int().min(0).optional(), }) + .refine((data) => (data.beforeAt === undefined) === (data.beforeSeq === undefined), { + message: 'beforeAt and beforeSeq must be provided together', + path: ['beforeAt'], + }) + .refine((data) => (data.afterAt === undefined) === (data.afterSeq === undefined), { + message: 'afterAt and afterSeq must be provided together', + path: ['afterAt'], + }) + .refine((data) => (data.untilAt === undefined) === (data.untilSeq === undefined), { + message: 'untilAt and untilSeq must be provided together', + path: ['untilAt'], + }) + .refine((data) => data.beforeAt === undefined || data.afterAt === undefined, { + message: 'before and after cursors are mutually exclusive', + path: ['afterAt'], + }) + .refine((data) => data.untilAt === undefined || data.afterAt !== undefined, { + message: 'until cursor requires an after cursor', + path: ['untilAt'], + }) + .refine((data) => data.epoch === undefined || data.afterAt !== undefined, { + message: 'epoch requires an after cursor', + path: ['epoch'], + }) export type MessagesQuery = z.infer diff --git a/web/src/App.tsx b/web/src/App.tsx index fe3285d7..070f6764 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -15,7 +15,7 @@ import { useViewportHeight } from '@/hooks/useViewportHeight' import { useVisibilityReporter } from '@/hooks/useVisibilityReporter' import { queryKeys } from '@/lib/query-keys' import { AppContextProvider } from '@/lib/app-context' -import { clearMessageWindow, fetchLatestMessages } from '@/lib/message-window-store' +import { clearMessageWindow, syncTailMessages } from '@/lib/message-window-store' import { useAppGoBack } from '@/hooks/useAppGoBack' import { useTranslation } from '@/lib/use-translation' import { VoiceProvider } from '@/lib/voice-context' @@ -229,7 +229,7 @@ function AppInner() { queryClient.invalidateQueries({ queryKey: ['session'] }) ] const refreshMessages = (selectedSessionId && api) - ? fetchLatestMessages(api, selectedSessionId) + ? syncTailMessages(api, selectedSessionId) : Promise.resolve() Promise.all([...invalidations, refreshMessages]) .catch((error) => { @@ -259,7 +259,7 @@ function AppInner() { return } clearMessageWindow(event.sessionId) - void fetchLatestMessages(api, event.sessionId) + void syncTailMessages(api, event.sessionId) }, [api, selectedSessionId]) const handleSessionSseConnect = useCallback(() => { diff --git a/web/src/api/client.ts b/web/src/api/client.ts index 085059c8..0ef8a6ed 100644 --- a/web/src/api/client.ts +++ b/web/src/api/client.ts @@ -283,6 +283,11 @@ export class ApiClient { options: { beforeSeq?: number | null beforeAt?: number | null + afterSeq?: number | null + afterAt?: number | null + untilSeq?: number | null + untilAt?: number | null + epoch?: number | null limit?: number } ): Promise { @@ -293,6 +298,21 @@ export class ApiClient { if (options.beforeSeq !== undefined && options.beforeSeq !== null) { params.set('beforeSeq', `${options.beforeSeq}`) } + if (options.afterAt !== undefined && options.afterAt !== null) { + params.set('afterAt', `${options.afterAt}`) + } + if (options.afterSeq !== undefined && options.afterSeq !== null) { + params.set('afterSeq', `${options.afterSeq}`) + } + if (options.untilAt !== undefined && options.untilAt !== null) { + params.set('untilAt', `${options.untilAt}`) + } + if (options.untilSeq !== undefined && options.untilSeq !== null) { + params.set('untilSeq', `${options.untilSeq}`) + } + if (options.epoch !== undefined && options.epoch !== null) { + params.set('epoch', `${options.epoch}`) + } if (options.limit !== undefined && options.limit !== null) { params.set('limit', `${options.limit}`) } diff --git a/web/src/components/AssistantChat/HappyThread.test.tsx b/web/src/components/AssistantChat/HappyThread.test.tsx index 271e9278..a72e4b20 100644 --- a/web/src/components/AssistantChat/HappyThread.test.tsx +++ b/web/src/components/AssistantChat/HappyThread.test.tsx @@ -5,10 +5,13 @@ import { I18nProvider } from '@/lib/i18n-context' import { ConversationOutlinePanel, captureScrollAnchor, + getHistoryCoverageRetryDelay, getScrollIntent, + loadOlderUntilViewportCovered, locateOutlineTargetMessage, prependMissingUserSnapshot, restoreScrollAnchor, + shouldLoadOlderForViewport, shouldCancelInitialScrollSettling, } from '@/components/AssistantChat/HappyThread' import type { ConversationOutlineItem } from '@/chat/outline' @@ -230,6 +233,89 @@ describe('scroll anchor helpers', () => { }) }) +describe('viewport-driven history coverage', () => { + it('recognizes an underfilled viewport and a top sentinel inside the preload margin', () => { + expect(shouldLoadOlderForViewport({ + scrollHeight: 300, + clientHeight: 500, + viewportTop: 100, + sentinelTop: 100, + sentinelBottom: 101 + })).toBe(true) + expect(shouldLoadOlderForViewport({ + scrollHeight: 1_000, + clientHeight: 500, + viewportTop: 100, + sentinelTop: -200, + sentinelBottom: -199 + })).toBe(false) + }) + + it('loads raw pages until rendered DOM grows by roughly one viewport', async () => { + let scrollHeight = 100 + const growthByPage = [0, 250, 300] + const loadOlderPage = vi.fn(async () => { + scrollHeight += growthByPage.shift() ?? 0 + return true + }) + + const loaded = await loadOlderUntilViewportCovered({ + hasMoreMessages: () => true, + needsCoverage: () => true, + getScrollHeight: () => scrollHeight, + getClientHeight: () => 500, + loadOlderPage, + waitForRender: async () => {} + }) + + expect(loaded).toBe(3) + expect(loadOlderPage).toHaveBeenCalledTimes(3) + expect(scrollHeight).toBe(650) + }) + + it('stops when history is exhausted even if raw pages produced no DOM growth', async () => { + let remainingPages = 2 + const loadOlderPage = vi.fn(async () => { + remainingPages -= 1 + return true + }) + + const loaded = await loadOlderUntilViewportCovered({ + hasMoreMessages: () => remainingPages > 0, + needsCoverage: () => true, + getScrollHeight: () => 100, + getClientHeight: () => 500, + loadOlderPage, + waitForRender: async () => {} + }) + + expect(loaded).toBe(2) + expect(loadOlderPage).toHaveBeenCalledTimes(2) + }) + + it('caps one gesture when pages keep producing no rendered output', async () => { + const loadOlderPage = vi.fn(async () => true) + + const loaded = await loadOlderUntilViewportCovered({ + hasMoreMessages: () => true, + needsCoverage: () => true, + getScrollHeight: () => 100, + getClientHeight: () => 500, + loadOlderPage, + waitForRender: async () => {}, + maxPages: 4 + }) + + expect(loaded).toBe(4) + expect(loadOlderPage).toHaveBeenCalledTimes(4) + }) + + it('defers an intersection signal until the initial scroll-settling deadline', () => { + expect(getHistoryCoverageRetryDelay(2_800, 1_000)).toBe(1_816) + expect(getHistoryCoverageRetryDelay(900, 1_000)).toBe(16) + }) +}) + describe('outline target loading', () => { it('loads older messages through the scroll-preserving wrapper until the target appears', async () => { const loadOlderPreservingScroll = vi.fn<() => Promise>() diff --git a/web/src/components/AssistantChat/HappyThread.tsx b/web/src/components/AssistantChat/HappyThread.tsx index 319d558d..7498c8eb 100644 --- a/web/src/components/AssistantChat/HappyThread.tsx +++ b/web/src/components/AssistantChat/HappyThread.tsx @@ -83,6 +83,8 @@ const AUTO_SCROLL_RESUME_THRESHOLD_PX = 120 const MANUAL_SCROLL_EPSILON_PX = 1 const INITIAL_SCROLL_SETTLE_MS = 1800 const INITIAL_SCROLL_SETTLE_DELAYS_MS = [0, 16, 50, 120, 250, 500, 900, 1400, 1800] as const +const HISTORY_PRELOAD_MARGIN_PX = 200 +const HISTORY_COVERAGE_PAGE_CAP = 8 type ScrollIntent = { distanceFromBottom: number @@ -156,6 +158,63 @@ export async function locateOutlineTargetMessage(options: LocateOutlineTargetOpt return target } +export function shouldLoadOlderForViewport(params: { + scrollHeight: number + clientHeight: number + viewportTop: number + sentinelTop: number + sentinelBottom: number + preloadMarginPx?: number +}): boolean { + const preloadMarginPx = params.preloadMarginPx ?? HISTORY_PRELOAD_MARGIN_PX + if (params.scrollHeight <= params.clientHeight + 1) { + return true + } + return params.sentinelBottom >= params.viewportTop - preloadMarginPx + && params.sentinelTop <= params.viewportTop + preloadMarginPx +} + +export async function loadOlderUntilViewportCovered(options: { + hasMoreMessages: () => boolean + needsCoverage: () => boolean + getScrollHeight: () => number + getClientHeight: () => number + loadOlderPage: () => Promise + waitForRender: () => Promise + forceFirstPage?: boolean + maxPages?: number +}): Promise { + if (!options.hasMoreMessages()) { + return 0 + } + if (!options.forceFirstPage && !options.needsCoverage()) { + return 0 + } + + const baselineHeight = options.getScrollHeight() + const targetGrowth = Math.max(1, options.getClientHeight()) + const maxPages = options.maxPages ?? HISTORY_COVERAGE_PAGE_CAP + let loadedPages = 0 + + while (loadedPages < maxPages && options.hasMoreMessages()) { + if (loadedPages > 0 && options.getScrollHeight() - baselineHeight >= targetGrowth) { + break + } + const loaded = await options.loadOlderPage() + if (!loaded) { + break + } + loadedPages += 1 + await options.waitForRender() + } + + return loadedPages +} + +export function getHistoryCoverageRetryDelay(deadline: number, now: number): number { + return Math.max(0, deadline - now) + 16 +} + function NewMessagesIndicator(props: { count: number; onClick: () => void }) { const { t } = useTranslation() if (props.count === 0) { @@ -341,17 +400,17 @@ export function HappyThread(props: { disabled: boolean onRefresh: () => void onRetryMessage?: (localId: string) => void - onFlushPending: () => void - onAtBottomChange: (atBottom: boolean) => void - isLoadingMessages: boolean + onViewModeChange: (mode: 'tail' | 'history') => void + isSyncingTail: boolean messagesWarning: string | null hasMoreMessages: boolean isLoadingMoreMessages: boolean - onLoadMore: () => Promise - pendingCount: number + onLoadMore: () => Promise + unseenCount: number rawMessagesCount: number normalizedMessagesCount: number messagesVersion: number + historyVersion: number forceScrollToken: number outlineOpen: boolean outlineItems: readonly ConversationOutlineItem[] @@ -367,20 +426,24 @@ export function HappyThread(props: { const topSentinelRef = useRef(null) const loadLockRef = useRef(false) const pendingScrollRef = useRef(null) - const prevLoadingMoreRef = useRef(false) - const loadStartedRef = useRef(false) const isLoadingMoreRef = useRef(props.isLoadingMoreMessages) const hasMoreMessagesRef = useRef(props.hasMoreMessages) - const isLoadingMessagesRef = useRef(props.isLoadingMessages) + const isSyncingTailRef = useRef(props.isSyncingTail) const messagesVersionRef = useRef(props.messagesVersion) + const historyVersionRef = useRef(props.historyVersion) const onLoadMoreRef = useRef(props.onLoadMore) const handleLoadMoreRef = useRef<() => void>(() => {}) const pendingLoadPromiseRef = useRef | null>(null) const pendingLoadResolveRef = useRef<((value: boolean) => void) | null>(null) - const pendingLoadBaselineRef = useRef<{ messagesVersion: number; hasMoreMessages: boolean } | null>(null) + const pendingLoadBaselineRef = useRef<{ + messagesVersion: number + historyVersion: number + hasMoreMessages: boolean + } | null>(null) + const coveragePromiseRef = useRef | null>(null) + const coverageRetryTimerRef = useRef(null) const atBottomRef = useRef(true) - const onAtBottomChangeRef = useRef(props.onAtBottomChange) - const onFlushPendingRef = useRef(props.onFlushPending) + const onViewModeChangeRef = useRef(props.onViewModeChange) const forceScrollTokenRef = useRef(props.forceScrollToken) const lastScrollTopRef = useRef(0) const sessionIdRef = useRef(props.sessionId) @@ -391,20 +454,20 @@ export function HappyThread(props: { // Smart scroll state: enabled only while the user is intentionally at the bottom. const autoScrollEnabledRef = useRef(true) useEffect(() => { - onAtBottomChangeRef.current = props.onAtBottomChange - }, [props.onAtBottomChange]) - useEffect(() => { - onFlushPendingRef.current = props.onFlushPending - }, [props.onFlushPending]) + onViewModeChangeRef.current = props.onViewModeChange + }, [props.onViewModeChange]) useEffect(() => { hasMoreMessagesRef.current = props.hasMoreMessages }, [props.hasMoreMessages]) useEffect(() => { - isLoadingMessagesRef.current = props.isLoadingMessages - }, [props.isLoadingMessages]) + isSyncingTailRef.current = props.isSyncingTail + }, [props.isSyncingTail]) useEffect(() => { messagesVersionRef.current = props.messagesVersion }, [props.messagesVersion]) + useEffect(() => { + historyVersionRef.current = props.historyVersion + }, [props.historyVersion]) useEffect(() => { onLoadMoreRef.current = props.onLoadMore }, [props.onLoadMore]) @@ -424,6 +487,22 @@ export function HappyThread(props: { initialScrollTimersRef.current = [] }, []) + const clearCoverageRetryTimer = useCallback(() => { + if (coverageRetryTimerRef.current !== null) { + window.clearTimeout(coverageRetryTimerRef.current) + coverageRetryTimerRef.current = null + } + }, []) + + const waitForRenderedFrame = useCallback((): Promise => { + return new Promise((resolve) => { + const schedule = typeof window.requestAnimationFrame === 'function' + ? window.requestAnimationFrame.bind(window) + : (callback: FrameRequestCallback) => window.setTimeout(() => callback(performance.now()), 0) + schedule(() => schedule(() => resolve())) + }) + }, []) + const settlePendingLoad = useCallback((result: boolean) => { const resolve = pendingLoadResolveRef.current const baseline = pendingLoadBaselineRef.current @@ -439,6 +518,7 @@ export function HappyThread(props: { } resolve( messagesVersionRef.current !== baseline.messagesVersion + || historyVersionRef.current !== baseline.historyVersion || hasMoreMessagesRef.current !== baseline.hasMoreMessages ) }, []) @@ -462,10 +542,7 @@ export function HappyThread(props: { return } atBottomRef.current = atBottom - onAtBottomChangeRef.current(atBottom) - if (atBottom) { - onFlushPendingRef.current() - } + onViewModeChangeRef.current(atBottom ? 'tail' : 'history') } const handleScroll = () => { @@ -525,9 +602,8 @@ export function HappyThread(props: { autoScrollEnabledRef.current = true if (!atBottomRef.current) { atBottomRef.current = true - onAtBottomChangeRef.current(true) + onViewModeChangeRef.current('tail') } - onFlushPendingRef.current() }, []) // Reset state when session changes @@ -535,25 +611,22 @@ export function HappyThread(props: { autoScrollEnabledRef.current = true lastScrollTopRef.current = viewportRef.current?.scrollTop ?? 0 atBottomRef.current = true - onAtBottomChangeRef.current(true) - // Re-entry forces the thread to the bottom, so release anything the - // non-at-bottom cold load parked in pending — otherwise new messages - // stay invisible until the user manually scrolls to bottom. - onFlushPendingRef.current() + onViewModeChangeRef.current('tail') forceScrollTokenRef.current = props.forceScrollToken pendingScrollRef.current = null loadLockRef.current = false - loadStartedRef.current = false + coveragePromiseRef.current = null initialScrollSessionRef.current = null initialScrollDeadlineRef.current = 0 clearInitialScrollTimers() + clearCoverageRetryTimer() settlePendingLoad(false) - }, [props.sessionId, clearInitialScrollTimers, settlePendingLoad]) + }, [props.sessionId, clearInitialScrollTimers, clearCoverageRetryTimer, settlePendingLoad]) useLayoutEffect(() => { if ( initialScrollSessionRef.current === props.sessionId - || props.isLoadingMessages + || props.isSyncingTail || props.rawMessagesCount === 0 || pendingScrollRef.current ) { @@ -563,8 +636,7 @@ export function HappyThread(props: { initialScrollSessionRef.current = props.sessionId autoScrollEnabledRef.current = true atBottomRef.current = true - onAtBottomChangeRef.current(true) - onFlushPendingRef.current() + onViewModeChangeRef.current('tail') scrollToBottomInstant() initialScrollDeadlineRef.current = Date.now() + INITIAL_SCROLL_SETTLE_MS @@ -581,7 +653,7 @@ export function HappyThread(props: { }, delay)) }, [ props.sessionId, - props.isLoadingMessages, + props.isSyncingTail, props.rawMessagesCount, props.messagesVersion, scrollToBottomInstant, @@ -591,9 +663,10 @@ export function HappyThread(props: { useEffect(() => { return () => { clearInitialScrollTimers() + clearCoverageRetryTimer() settlePendingLoad(false) } - }, [clearInitialScrollTimers, settlePendingLoad]) + }, [clearInitialScrollTimers, clearCoverageRetryTimer, settlePendingLoad]) useEffect(() => { if (forceScrollTokenRef.current === props.forceScrollToken) { @@ -609,7 +682,7 @@ export function HappyThread(props: { } if ( isInitialScrollSettling() - || isLoadingMessagesRef.current + || isSyncingTailRef.current || !hasMoreMessagesRef.current || isLoadingMoreRef.current || loadLockRef.current @@ -627,46 +700,95 @@ export function HappyThread(props: { } autoScrollEnabledRef.current = false loadLockRef.current = true - loadStartedRef.current = false pendingLoadBaselineRef.current = { messagesVersion: messagesVersionRef.current, + historyVersion: historyVersionRef.current, hasMoreMessages: hasMoreMessagesRef.current } const loadPromise = new Promise((resolve) => { pendingLoadResolveRef.current = resolve }) pendingLoadPromiseRef.current = loadPromise - try { - void onLoadMoreRef.current().catch((error) => { + void (async () => { + try { + const loaded = await onLoadMoreRef.current() + if (loaded) { + return + } + pendingScrollRef.current = null + loadLockRef.current = false + settlePendingLoad(false) + } catch (error) { pendingScrollRef.current = null loadLockRef.current = false settlePendingLoad(false) console.error('Failed to load older messages:', error) - }).finally(() => { - if (!loadStartedRef.current && !isLoadingMoreRef.current) { - if (pendingScrollRef.current) { - pendingScrollRef.current = null - loadLockRef.current = false - } - settlePendingLoad(true) - } - }) - } catch (error) { - pendingScrollRef.current = null - loadLockRef.current = false - settlePendingLoad(false) - console.error('Failed to load older messages:', error) - } + } + })() return loadPromise }, [isInitialScrollSettling, settlePendingLoad]) + const needsViewportCoverage = useCallback((): boolean => { + const viewport = viewportRef.current + const sentinel = topSentinelRef.current + if (!viewport || !sentinel) { + return false + } + const viewportRect = viewport.getBoundingClientRect() + const sentinelRect = sentinel.getBoundingClientRect() + return shouldLoadOlderForViewport({ + scrollHeight: viewport.scrollHeight, + clientHeight: viewport.clientHeight, + viewportTop: viewportRect.top, + sentinelTop: sentinelRect.top, + sentinelBottom: sentinelRect.bottom + }) + }, []) + + const loadOlderWithCoverage = useCallback((forceFirstPage = false): Promise => { + if (coveragePromiseRef.current) { + return coveragePromiseRef.current + } + const viewport = viewportRef.current + if (!viewport) { + return Promise.resolve(false) + } + const run = loadOlderUntilViewportCovered({ + hasMoreMessages: () => hasMoreMessagesRef.current, + needsCoverage: needsViewportCoverage, + getScrollHeight: () => viewport.scrollHeight, + getClientHeight: () => viewport.clientHeight, + loadOlderPage: loadOlderPreservingScroll, + waitForRender: waitForRenderedFrame, + forceFirstPage + }).then((loadedPages) => loadedPages > 0) + let tracked: Promise + tracked = run.finally(() => { + if (coveragePromiseRef.current === tracked) { + coveragePromiseRef.current = null + } + }) + coveragePromiseRef.current = tracked + return tracked + }, [loadOlderPreservingScroll, needsViewportCoverage, waitForRenderedFrame]) + + const scheduleCoverageAfterSettling = useCallback(() => { + clearCoverageRetryTimer() + const delay = getHistoryCoverageRetryDelay(initialScrollDeadlineRef.current, Date.now()) + coverageRetryTimerRef.current = window.setTimeout(() => { + coverageRetryTimerRef.current = null + void loadOlderWithCoverage(false) + }, delay) + }, [clearCoverageRetryTimer, loadOlderWithCoverage]) + const loadOlderFromUserAction = useCallback((): Promise => { // Initial settling protects the automatic top sentinel from racing the // first scroll-to-bottom pass. It must not swallow an explicit click. initialScrollDeadlineRef.current = 0 clearInitialScrollTimers() - return loadOlderPreservingScroll() - }, [clearInitialScrollTimers, loadOlderPreservingScroll]) + clearCoverageRetryTimer() + return loadOlderWithCoverage(true) + }, [clearInitialScrollTimers, clearCoverageRetryTimer, loadOlderWithCoverage]) const handleOutlineSelect = useCallback(async (item: ConversationOutlineItem) => { const target = await locateOutlineTargetMessage({ @@ -685,14 +807,14 @@ export function HappyThread(props: { useEffect(() => { handleLoadMoreRef.current = () => { - void loadOlderPreservingScroll() + void loadOlderWithCoverage(false) } - }, [loadOlderPreservingScroll]) + }, [loadOlderWithCoverage]) useEffect(() => { const sentinel = topSentinelRef.current const viewport = viewportRef.current - if (!sentinel || !viewport || !props.hasMoreMessages || props.isLoadingMessages) { + if (!sentinel || !viewport || !props.hasMoreMessages || props.isSyncingTail) { return } if (typeof IntersectionObserver === 'undefined') { @@ -704,6 +826,7 @@ export function HappyThread(props: { for (const entry of entries) { if (entry.isIntersecting) { if (isInitialScrollSettling()) { + scheduleCoverageAfterSettling() continue } handleLoadMoreRef.current() @@ -712,13 +835,36 @@ export function HappyThread(props: { }, { root: viewport, - rootMargin: '200px 0px 0px 0px' + rootMargin: `${HISTORY_PRELOAD_MARGIN_PX}px 0px 0px 0px` } ) observer.observe(sentinel) return () => observer.disconnect() - }, [props.hasMoreMessages, props.isLoadingMessages, isInitialScrollSettling]) + }, [ + props.hasMoreMessages, + props.isSyncingTail, + isInitialScrollSettling, + scheduleCoverageAfterSettling + ]) + + useEffect(() => { + if (!props.hasMoreMessages || props.isSyncingTail) { + clearCoverageRetryTimer() + return + } + if (isInitialScrollSettling() && needsViewportCoverage()) { + scheduleCoverageAfterSettling() + } + }, [ + props.hasMoreMessages, + props.isSyncingTail, + props.messagesVersion, + isInitialScrollSettling, + needsViewportCoverage, + scheduleCoverageAfterSettling, + clearCoverageRetryTimer + ]) useEffect(() => { const content = contentRef.current @@ -737,10 +883,22 @@ export function HappyThread(props: { ) { scrollToBottomInstant() } + if ( + hasMoreMessagesRef.current + && !isInitialScrollSettling() + && needsViewportCoverage() + ) { + void loadOlderWithCoverage(false) + } }) observer.observe(content) return () => observer.disconnect() - }, [scrollToBottomInstant]) + }, [ + scrollToBottomInstant, + isInitialScrollSettling, + needsViewportCoverage, + loadOlderWithCoverage + ]) useLayoutEffect(() => { const pending = pendingScrollRef.current @@ -763,24 +921,13 @@ export function HappyThread(props: { if (atBottomRef.current && autoScrollEnabledRef.current) { scrollToBottomInstant() } - }, [props.messagesVersion, scrollToBottomInstant, settlePendingLoad]) + }, [props.messagesVersion, props.historyVersion, scrollToBottomInstant, settlePendingLoad]) useEffect(() => { isLoadingMoreRef.current = props.isLoadingMoreMessages - if (props.isLoadingMoreMessages) { - loadStartedRef.current = true - } - if (prevLoadingMoreRef.current && !props.isLoadingMoreMessages) { - if (pendingScrollRef.current) { - pendingScrollRef.current = null - loadLockRef.current = false - } - settlePendingLoad(true) - } - prevLoadingMoreRef.current = props.isLoadingMoreMessages - }, [props.isLoadingMoreMessages, settlePendingLoad]) + }, [props.isLoadingMoreMessages]) - const showSkeleton = props.isLoadingMessages && props.rawMessagesCount === 0 && props.pendingCount === 0 + const showSkeleton = props.isSyncingTail && props.rawMessagesCount === 0 const handleShareTurn = useCallback(( messageTarget: HTMLElement | string | null, clientY?: number, @@ -863,6 +1010,15 @@ export function HappyThread(props: { loadOlderMessagesPreservingScroll: loadOlderFromUserAction }}> + {props.isSyncingTail && props.rawMessagesCount > 0 ? ( +
+ + {t('misc.loadingMessages')} +
+ ) : null} ) : null} - {props.hasMoreMessages && !props.isLoadingMessages ? ( + {props.hasMoreMessages && !props.isSyncingTail ? (
- + {props.outlineOpen ? ( <>