diff --git a/hub/src/socket/handlers/cli/sessionHandlers.ts b/hub/src/socket/handlers/cli/sessionHandlers.ts index b3a89292..43ea7207 100644 --- a/hub/src/socket/handlers/cli/sessionHandlers.ts +++ b/hub/src/socket/handlers/cli/sessionHandlers.ts @@ -150,7 +150,8 @@ export function registerSessionHandlers(socket: CliSocketWithData, deps: Session seq: msg.seq, localId: msg.localId, content: msg.content, - createdAt: msg.createdAt + createdAt: msg.createdAt, + invokedAt: msg.invokedAt } }) }) @@ -274,7 +275,18 @@ export function registerSessionHandlers(socket: CliSocketWithData, deps: Session emitAccessError('session', data.sid, sessionAccess.reason) return } - onWebappEvent?.({ type: 'messages-consumed', sessionId: data.sid, localIds }) + const invokedAt = Date.now() + try { + store.messages.markMessagesInvoked(data.sid, localIds, invokedAt) + onSessionActivity?.(data.sid, invokedAt) + // 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 }) + } catch (err) { + console.error('markMessagesInvoked failed', err) + } }) socket.on('session-end', (data: SessionEndPayload) => { @@ -286,6 +298,30 @@ export function registerSessionHandlers(socket: CliSocketWithData, deps: Session emitAccessError('session', data.sid, sessionAccess.reason) return } + + // Force-invoke any user messages that are still queued at session end. + // Without this, the floating bar pins the queued rows after the CLI is + // gone — there is no longer an ack path (no CLI to emit + // messages-consumed) so they would stay queued forever. + try { + const queued = store.messages.getUninvokedLocalMessages(data.sid) + const localIds = queued + .map((m) => m.localId) + .filter((id): id is string => typeof id === 'string') + if (localIds.length > 0) { + const invokedAt = Date.now() + store.messages.markMessagesInvoked(data.sid, localIds, invokedAt) + onWebappEvent?.({ + type: 'messages-consumed', + sessionId: data.sid, + localIds, + invokedAt + }) + } + } catch (err) { + console.error('session-end markMessagesInvoked failed', err) + } + onSessionEnd?.(data) }) } diff --git a/hub/src/store/index.ts b/hub/src/store/index.ts index f9ac76ba..f8c08cde 100644 --- a/hub/src/store/index.ts +++ b/hub/src/store/index.ts @@ -22,7 +22,7 @@ export { PushStore } from './pushStore' export { SessionStore } from './sessionStore' export { UserStore } from './userStore' -const SCHEMA_VERSION: number = 7 +const SCHEMA_VERSION: number = 8 const REQUIRED_TABLES = [ 'sessions', 'machines', @@ -85,9 +85,35 @@ export class Store { private initSchema(): void { const currentVersion = this.getUserVersion() + // V1/V2/V3 entries cover legacy DBs that pre-date our migration ladder. + // Each step is idempotent (column-existence guards inside) so we can + // safely run the full V1→V8 chain in the legacy branch where the DB + // shape is unknown. + const buildStepMigrations = (legacy: boolean): Record void> => ({ + 1: () => this.migrateFromV1ToV2(legacy), + 2: () => this.migrateFromV2ToV3(), + 3: () => this.migrateFromV3ToV4(), + 4: () => this.migrateFromV4ToV5(), + 5: () => this.migrateFromV5ToV6(), + 6: () => this.migrateFromV6ToV7(), + 7: () => this.migrateFromV7ToV8(), + }) + if (currentVersion === 0) { if (this.hasAnyUserTables()) { this.migrateLegacySchemaIfNeeded() + // Run the full step ladder BEFORE createSchema so legacy tables + // pick up every later-version column (e.g. invoked_at) via ALTER + // TABLE. Without this, createSchema below would try to build + // idx_messages_session_position over a column that does not + // exist yet, and CREATE TABLE IF NOT EXISTS would not add the + // missing column to the existing table. + const legacySteps = buildStepMigrations(true) + for (let v = 1; v < SCHEMA_VERSION; v++) { + legacySteps[v]?.() + } + // Backfill any *missing* tables (sessions, machines, ...) that + // a partially-built legacy DB may not have yet. this.createSchema() this.setUserVersion(SCHEMA_VERSION) return @@ -98,60 +124,13 @@ export class Store { return } - if (currentVersion === 1 && SCHEMA_VERSION === 2) { - this.migrateFromV1ToV2() - this.setUserVersion(SCHEMA_VERSION) - return - } - - if (currentVersion === 2 && SCHEMA_VERSION === 3) { - this.migrateFromV2ToV3() - this.setUserVersion(SCHEMA_VERSION) - return - } - - if (currentVersion === 3 && SCHEMA_VERSION === 4) { - this.migrateFromV3ToV4() - this.setUserVersion(SCHEMA_VERSION) - return - } - - if (currentVersion === 4 && SCHEMA_VERSION === 5) { - this.migrateFromV4ToV5() - this.setUserVersion(SCHEMA_VERSION) - return - } - - if (currentVersion === 5 && SCHEMA_VERSION === 6) { - this.migrateFromV5ToV6() - this.setUserVersion(SCHEMA_VERSION) - return - } - - if (currentVersion === 6 && SCHEMA_VERSION === 7) { - this.migrateFromV6ToV7() - this.setUserVersion(SCHEMA_VERSION) - return - } - - if (currentVersion === 4 && SCHEMA_VERSION === 6) { - this.migrateFromV4ToV5() - this.migrateFromV5ToV6() - this.setUserVersion(SCHEMA_VERSION) - return - } - - if (currentVersion === 4 && SCHEMA_VERSION === 7) { - this.migrateFromV4ToV5() - this.migrateFromV5ToV6() - this.migrateFromV6ToV7() - this.setUserVersion(SCHEMA_VERSION) - return - } - - if (currentVersion === 5 && SCHEMA_VERSION === 7) { - this.migrateFromV5ToV6() - this.migrateFromV6ToV7() + const stepMigrations = buildStepMigrations(false) + if (currentVersion < SCHEMA_VERSION && stepMigrations[currentVersion]) { + for (let v = currentVersion; v < SCHEMA_VERSION; v++) { + const step = stepMigrations[v] + if (!step) throw this.buildSchemaMismatchError(currentVersion) + step() + } this.setUserVersion(SCHEMA_VERSION) return } @@ -212,10 +191,13 @@ export class Store { created_at INTEGER NOT NULL, seq INTEGER NOT NULL, local_id TEXT, + invoked_at INTEGER, FOREIGN KEY (session_id) REFERENCES sessions(id) ON DELETE CASCADE ); CREATE INDEX IF NOT EXISTS idx_messages_session ON messages(session_id, seq); CREATE UNIQUE INDEX IF NOT EXISTS idx_messages_local_id ON messages(session_id, local_id) WHERE local_id IS NOT NULL; + CREATE INDEX IF NOT EXISTS idx_messages_session_position + ON messages(session_id, COALESCE(invoked_at, created_at) DESC, seq DESC); CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, @@ -259,9 +241,14 @@ export class Store { } } - private migrateFromV1ToV2(): void { + private migrateFromV1ToV2(legacy: boolean = false): void { const columns = this.getMachineColumnNames() if (columns.size === 0) { + // In the legacy branch the table may not exist yet — createSchema + // will build the up-to-date one. When invoked from the regular + // upgrade path (user_version >= 1), missing the machines table is + // still an error. + if (legacy) return throw new Error('SQLite schema missing machines table for v1 to v2 migration.') } @@ -273,6 +260,7 @@ export class Store { } if (!hasDaemon) { + if (legacy) return throw new Error('SQLite schema missing daemon_state columns for v1 to v2 migration.') } @@ -333,6 +321,11 @@ export class Store { private migrateFromV3ToV4(): void { const columns = this.getSessionColumnNames() + // When the legacy branch invokes the full step ladder, an upstream-only + // DB may not have the sessions table yet — createSchema runs after the + // ladder. Skip ALTERs in that case; createSchema will build the table + // with the up-to-date columns. + if (columns.size === 0) return if (!columns.has('team_state')) { this.db.exec('ALTER TABLE sessions ADD COLUMN team_state TEXT') } @@ -343,6 +336,7 @@ export class Store { private migrateFromV4ToV5(): void { const columns = this.getSessionColumnNames() + if (columns.size === 0) return if (!columns.has('model')) { this.db.exec('ALTER TABLE sessions ADD COLUMN model TEXT') } @@ -350,6 +344,7 @@ export class Store { private migrateFromV5ToV6(): void { const columns = this.getSessionColumnNames() + if (columns.size === 0) return if (!columns.has('effort')) { this.db.exec('ALTER TABLE sessions ADD COLUMN effort TEXT') } @@ -357,11 +352,31 @@ export class Store { private migrateFromV6ToV7(): void { const columns = this.getSessionColumnNames() + if (columns.size === 0) return if (!columns.has('model_reasoning_effort')) { this.db.exec('ALTER TABLE sessions ADD COLUMN model_reasoning_effort TEXT') } } + private migrateFromV7ToV8(): void { + const columns = this.getMessageColumnNames() + if (columns.size === 0) { + // No messages table yet — createSchema will build the up-to-date one. + return + } + if (!columns.has('invoked_at')) { + this.db.exec('ALTER TABLE messages ADD COLUMN invoked_at INTEGER') + } + // Idempotent (WHERE invoked_at IS NULL); safe to re-run if a previous attempt + // crashed between ALTER and UPDATE before user_version was bumped. + this.db.exec('UPDATE messages SET invoked_at = created_at WHERE invoked_at IS NULL') + // Position index for byPosition pagination — idempotent via IF NOT EXISTS. + this.db.exec(` + CREATE INDEX IF NOT EXISTS idx_messages_session_position + ON messages(session_id, COALESCE(invoked_at, created_at) DESC, seq DESC) + `) + } + 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)) @@ -372,6 +387,11 @@ export class Store { return new Set(rows.map((row) => row.name)) } + private getMessageColumnNames(): Set { + const rows = this.db.prepare('PRAGMA table_info(messages)').all() as Array<{ name: string }> + return new Set(rows.map((row) => row.name)) + } + private getUserVersion(): number { const row = this.db.prepare('PRAGMA user_version').get() as { user_version: number } | undefined return row?.user_version ?? 0 diff --git a/hub/src/store/messageStore.ts b/hub/src/store/messageStore.ts index c25f5732..3dec8002 100644 --- a/hub/src/store/messageStore.ts +++ b/hub/src/store/messageStore.ts @@ -1,7 +1,7 @@ import type { Database } from 'bun:sqlite' import type { StoredMessage } from './types' -import { addMessage, getMessages, getMessagesAfter, mergeSessionMessages } from './messages' +import { addMessage, getMessages, getMessagesAfter, getMessagesByPosition, getUninvokedLocalMessages, markMessagesInvoked, mergeSessionMessages } from './messages' export class MessageStore { private readonly db: Database @@ -22,6 +22,18 @@ export class MessageStore { return getMessagesAfter(this.db, sessionId, afterSeq, limit) } + getMessagesByPosition(sessionId: string, limit: number, before?: { at: number; seq: number }): StoredMessage[] { + return getMessagesByPosition(this.db, sessionId, limit, before) + } + + getUninvokedLocalMessages(sessionId: string): StoredMessage[] { + return getUninvokedLocalMessages(this.db, sessionId) + } + + markMessagesInvoked(sessionId: string, localIds: string[], invokedAt: number): void { + markMessagesInvoked(this.db, sessionId, localIds, invokedAt) + } + mergeSessionMessages(fromSessionId: string, toSessionId: string): { moved: number; oldMaxSeq: number; newMaxSeq: number } { return mergeSessionMessages(this.db, fromSessionId, toSessionId) } diff --git a/hub/src/store/messages.ts b/hub/src/store/messages.ts index bb850c0c..c315ea7e 100644 --- a/hub/src/store/messages.ts +++ b/hub/src/store/messages.ts @@ -11,6 +11,7 @@ type DbMessageRow = { created_at: number seq: number local_id: string | null + invoked_at: number | null } function toStoredMessage(row: DbMessageRow): StoredMessage { @@ -20,7 +21,8 @@ function toStoredMessage(row: DbMessageRow): StoredMessage { content: safeJsonParse(row.content), createdAt: row.created_at, seq: row.seq, - localId: row.local_id + localId: row.local_id, + invokedAt: row.invoked_at ?? null } } @@ -49,11 +51,16 @@ export function addMessage( const id = randomUUID() const json = JSON.stringify(content) + // Messages without a localId have no ack path (markMessagesInvoked matches by localId). + // Treat them as already-invoked at insert time so they land in the thread normally instead + // of being stuck in the queued floating bar forever. + const invokedAt = localId ? null : now + db.prepare(` INSERT INTO messages ( - id, session_id, content, created_at, seq, local_id + id, session_id, content, created_at, seq, local_id, invoked_at ) VALUES ( - @id, @session_id, @content, @created_at, @seq, @local_id + @id, @session_id, @content, @created_at, @seq, @local_id, @invoked_at ) `).run({ id, @@ -61,7 +68,8 @@ export function addMessage( content: json, created_at: now, seq: msgSeq, - local_id: localId ?? null + local_id: localId ?? null, + invoked_at: invokedAt }) const row = db.prepare('SELECT * FROM messages WHERE id = ?').get(id) as DbMessageRow | undefined @@ -106,6 +114,49 @@ export function getMessagesAfter( return rows.map(toStoredMessage) } +/** Paginate messages by COALESCE(invoked_at, created_at) DESC, seq DESC. + * Used for V8 byPosition mode. Results are returned in ascending display order. */ +export function getMessagesByPosition( + db: Database, + sessionId: string, + limit: number, + before?: { at: number; seq: number } +): StoredMessage[] { + const safeLimit = Number.isFinite(limit) ? Math.max(1, Math.min(200, limit)) : 200 + const beforeClause = before + ? 'AND (COALESCE(invoked_at, created_at) < @beforeAt OR (COALESCE(invoked_at, created_at) = @beforeAt AND seq < @beforeSeq))' + : '' + const rows = db.prepare(` + SELECT *, COALESCE(invoked_at, created_at) AS position_at + FROM messages + WHERE session_id = @sessionId + ${beforeClause} + ORDER BY position_at DESC, seq DESC + LIMIT @limit + `).all({ + sessionId, + beforeAt: before?.at ?? null, + beforeSeq: before?.seq ?? null, + limit: safeLimit + }) as DbMessageRow[] + // Reverse so results are in ascending display order (oldest first) + return rows.reverse().map(toStoredMessage) +} + +/** Returns user messages that have a localId but no invoked_at. + * Used to surface queued messages on refresh / secondary clients even when they + * fall outside the latest position-ordered page (their position key is the send + * time, but the floating bar still needs to render them). */ +export function getUninvokedLocalMessages( + db: Database, + sessionId: string +): StoredMessage[] { + const rows = db.prepare( + 'SELECT * FROM messages WHERE session_id = ? AND invoked_at IS NULL AND local_id IS NOT NULL ORDER BY seq ASC' + ).all(sessionId) as DbMessageRow[] + return rows.map(toStoredMessage) +} + export function getMaxSeq(db: Database, sessionId: string): number { const row = db.prepare( 'SELECT COALESCE(MAX(seq), 0) AS maxSeq FROM messages WHERE session_id = ?' @@ -113,6 +164,28 @@ export function getMaxSeq(db: Database, sessionId: string): number { return row?.maxSeq ?? 0 } +/** Mark messages as invoked at the given server timestamp. + * Only updates rows whose local_id is in localIds. + * First-write-wins: rows with a non-NULL invoked_at are not updated. A duplicate + * ack (e.g. a CLI re-emit) would otherwise re-stamp the timestamp and shuffle + * the message's position in the byPosition-ordered thread. */ +export function markMessagesInvoked( + db: Database, + sessionId: string, + localIds: string[], + invokedAt: number +): void { + if (localIds.length === 0) return + const placeholders = localIds.map(() => '?').join(', ') + db.prepare( + `UPDATE messages + SET invoked_at = ? + WHERE session_id = ? + AND local_id IN (${placeholders}) + AND invoked_at IS NULL` + ).run(invokedAt, sessionId, ...localIds) +} + export function mergeSessionMessages( db: Database, fromSessionId: string, @@ -145,8 +218,15 @@ export function mergeSessionMessages( if (collisions.length > 0) { const localIds = collisions.map((row) => row.local_id) const placeholders = localIds.map(() => '?').join(', ') + // Force-invoke the older copy: clearing local_id severs its ack path + // (markMessagesInvoked matches by local_id), so leaving invoked_at + // NULL would strand the row in the queued floating bar forever. + // Use COALESCE so an already-invoked row keeps its server timestamp. db.prepare( - `UPDATE messages SET local_id = NULL WHERE session_id = ? AND local_id IN (${placeholders})` + `UPDATE messages + SET local_id = NULL, + invoked_at = COALESCE(invoked_at, created_at) + WHERE session_id = ? AND local_id IN (${placeholders})` ).run(fromSessionId, ...localIds) } diff --git a/hub/src/store/migration-v8.test.ts b/hub/src/store/migration-v8.test.ts new file mode 100644 index 00000000..a19cea0d --- /dev/null +++ b/hub/src/store/migration-v8.test.ts @@ -0,0 +1,811 @@ +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' + +/** + * Tests for V7→V8 schema migration: adding invoked_at column to messages table. + * All migration tests open a real Store to exercise the actual migration code path. + */ +describe('Store V7→V8 migration: invoked_at column', () => { + it('fresh DB has invoked_at column in messages', () => { + const store = new Store(':memory:') + const cols = getMessageColumns(store) + expect(cols).toContain('invoked_at') + }) + + it('V7 DB migrates to V8 via Store: invoked_at added, existing rows backfilled to created_at', () => { + const dir = mkdtempSync(join(tmpdir(), 'hapi-migration-v8-test-')) + const dbPath = join(dir, 'test.db') + try { + // Build a V7 DB on disk, insert rows, then open via Store to trigger migration + const db = new Database(dbPath, { create: true, readwrite: true, strict: true }) + db.exec('PRAGMA journal_mode = WAL') + db.exec('PRAGMA foreign_keys = ON') + createV7Schema(db) + db.exec('PRAGMA user_version = 7') + db.exec(`INSERT INTO sessions (id, namespace, created_at, updated_at, seq) + VALUES ('s1', 'default', 1000, 1000, 0)`) + db.exec(`INSERT INTO messages (id, session_id, content, created_at, seq) + VALUES ('m1', 's1', '"hello"', 1000, 1)`) + db.exec(`INSERT INTO messages (id, session_id, content, created_at, seq) + VALUES ('m2', 's1', '"world"', 2000, 2)`) + db.close() + + // Open via Store — should auto-migrate V7→V8 + const store = new Store(dbPath) + const cols = getMessageColumns(store) + expect(cols).toContain('invoked_at') + + // Backfill: existing rows must have invoked_at == created_at (not NULL) + const msgs = store.messages.getMessages('s1') + expect(msgs).toHaveLength(2) + const m1 = msgs.find(m => m.id === 'm1')! + const m2 = msgs.find(m => m.id === 'm2')! + expect(m1.invokedAt).toBe(1000) + expect(m2.invokedAt).toBe(2000) + } finally { + rmSync(dir, { recursive: true, force: true }) + } + }) + + it('V6 DB migrates to V8 (multi-hop)', () => { + const dir = mkdtempSync(join(tmpdir(), 'hapi-migration-v6-test-')) + const dbPath = join(dir, 'test.db') + try { + const db = new Database(dbPath, { create: true, readwrite: true, strict: true }) + db.exec('PRAGMA journal_mode = WAL') + db.exec('PRAGMA foreign_keys = ON') + createV6Schema(db) + db.exec('PRAGMA user_version = 6') + db.close() + + const store = new Store(dbPath) + const cols = getMessageColumns(store) + expect(cols).toContain('invoked_at') + // sessions table should have model_reasoning_effort (added in V6→V7) + const sessionCols = getSessionColumns(store) + expect(sessionCols).toContain('model_reasoning_effort') + } finally { + rmSync(dir, { recursive: true, force: true }) + } + }) + + it('V5 DB migrates to V8 (multi-hop)', () => { + const dir = mkdtempSync(join(tmpdir(), 'hapi-migration-v5-test-')) + const dbPath = join(dir, 'test.db') + try { + const db = new Database(dbPath, { create: true, readwrite: true, strict: true }) + db.exec('PRAGMA journal_mode = WAL') + db.exec('PRAGMA foreign_keys = ON') + createV5Schema(db) + db.exec('PRAGMA user_version = 5') + db.close() + + const store = new Store(dbPath) + const cols = getMessageColumns(store) + expect(cols).toContain('invoked_at') + } finally { + rmSync(dir, { recursive: true, force: true }) + } + }) + + it('V4 DB migrates to V8 (multi-hop)', () => { + const dir = mkdtempSync(join(tmpdir(), 'hapi-migration-v4-test-')) + const dbPath = join(dir, 'test.db') + try { + const db = new Database(dbPath, { create: true, readwrite: true, strict: true }) + db.exec('PRAGMA journal_mode = WAL') + db.exec('PRAGMA foreign_keys = ON') + createV4Schema(db) + db.exec('PRAGMA user_version = 4') + db.close() + + const store = new Store(dbPath) + const cols = getMessageColumns(store) + expect(cols).toContain('invoked_at') + } finally { + rmSync(dir, { recursive: true, force: true }) + } + }) + + it('V8 DB reopen is idempotent: schema unchanged', () => { + const dir = mkdtempSync(join(tmpdir(), 'hapi-migration-v8-idempotent-')) + const dbPath = join(dir, 'test.db') + try { + const store1 = new Store(dbPath) + const cols1 = getMessageColumns(store1) + expect(cols1).toContain('invoked_at') + + // Re-open same DB — version is already 8, must not throw or alter schema + const store2 = new Store(dbPath) + const cols2 = getMessageColumns(store2) + expect(cols2).toEqual(cols1) + } finally { + rmSync(dir, { recursive: true, force: true }) + } + }) + + it('migrateFromV7ToV8 PRAGMA guard: invoked_at column appears exactly once', () => { + const dir = mkdtempSync(join(tmpdir(), 'hapi-migration-v8-guard-')) + const dbPath = join(dir, 'test.db') + try { + const db = new Database(dbPath, { create: true, readwrite: true, strict: true }) + db.exec('PRAGMA journal_mode = WAL') + db.exec('PRAGMA foreign_keys = ON') + createV7Schema(db) + db.exec('PRAGMA user_version = 7') + db.close() + + const store = new Store(dbPath) + const cols = getMessageColumns(store) + const count = cols.filter(c => c === 'invoked_at').length + expect(count).toBe(1) + } finally { + rmSync(dir, { recursive: true, force: true }) + } + }) + + it('markMessagesInvoked sets invoked_at on matching messages', () => { + const store = new Store(':memory:') + const session = store.sessions.getOrCreateSession('test', { path: '/tmp' }, null, 'default') + const msg1 = store.messages.addMessage(session.id, 'hello', 'local-1') + const msg2 = store.messages.addMessage(session.id, 'world', 'local-2') + + // Initially both have invokedAt = null (new messages added to fresh V8 DB) + expect(store.messages.getMessages(session.id).map(m => m.invokedAt)).toEqual([null, null]) + + const ts = Date.now() + store.messages.markMessagesInvoked(session.id, ['local-1'], ts) + + const msgs = store.messages.getMessages(session.id) + const m1 = msgs.find(m => m.id === msg1.id)! + const m2 = msgs.find(m => m.id === msg2.id)! + expect(m1.invokedAt).toBe(ts) + expect(m2.invokedAt).toBeNull() + }) + + it('markMessagesInvoked is first-write-wins (subsequent calls are no-ops)', () => { + const store = new Store(':memory:') + const session = store.sessions.getOrCreateSession('test', { path: '/tmp' }, null, 'default') + store.messages.addMessage(session.id, 'hi', 'local-x') + + const ts1 = 1000 + const ts2 = 2000 + store.messages.markMessagesInvoked(session.id, ['local-x'], ts1) + // A duplicate ack (CLI re-emit) must not overwrite the original timestamp: + // re-stamping invoked_at would shuffle the message in the byPosition-ordered + // thread for every subscribed client. + store.messages.markMessagesInvoked(session.id, ['local-x'], ts2) + + const msgs = store.messages.getMessages(session.id) + expect(msgs[0].invokedAt).toBe(ts1) + }) + + it('addMessage with localId leaves invoked_at NULL (ack path is messages-consumed)', () => { + const store = new Store(':memory:') + const session = store.sessions.getOrCreateSession('test', { path: '/tmp' }, null, 'default') + const msg = store.messages.addMessage(session.id, 'content', 'local-1') + expect(msg.invokedAt).toBeNull() + }) + + it('addMessage without localId sets invoked_at = created_at (no ack path)', () => { + const store = new Store(':memory:') + const session = store.sessions.getOrCreateSession('test', { path: '/tmp' }, null, 'default') + const msg = store.messages.addMessage(session.id, 'content') + expect(msg.invokedAt).toBe(msg.createdAt) + }) + + it('getUninvokedLocalMessages returns rows with localId and null invoked_at', () => { + const store = new Store(':memory:') + const session = store.sessions.getOrCreateSession('test', { path: '/tmp' }, null, 'default') + const queued = store.messages.addMessage(session.id, 'q', 'local-q') + store.messages.addMessage(session.id, 'no-localid') // invoked_at = createdAt, excluded + store.messages.addMessage(session.id, 'invoked', 'local-i') + store.messages.markMessagesInvoked(session.id, ['local-i'], Date.now()) + + const uninvoked = store.messages.getUninvokedLocalMessages(session.id) + expect(uninvoked.map(m => m.id)).toEqual([queued.id]) + }) + + it('getUninvokedLocalMessages returns empty for session with no queued messages', () => { + const store = new Store(':memory:') + const session = store.sessions.getOrCreateSession('test', { path: '/tmp' }, null, 'default') + store.messages.addMessage(session.id, 'plain') // invoked_at set + const sent = store.messages.addMessage(session.id, 'sent', 'local-s') + store.messages.markMessagesInvoked(session.id, ['local-s'], Date.now()) + expect(sent.invokedAt).toBeNull() // value at insert; row has been updated since + expect(store.messages.getUninvokedLocalMessages(session.id)).toEqual([]) + }) + + // Mirrors the session-end handler's auto-invoke contract at the store + // layer: when a CLI exits, we sweep every queued message for the session + // (getUninvokedLocalMessages) and stamp them with a single timestamp + // (markMessagesInvoked). After the sweep, no queued ghosts may remain — + // otherwise the floating bar would survive across reloads even though the + // CLI is no longer running. + it('session-end pattern: getUninvokedLocalMessages + markMessagesInvoked clears all queued', () => { + const store = new Store(':memory:') + const session = store.sessions.getOrCreateSession('test', { path: '/tmp' }, null, 'default') + store.messages.addMessage(session.id, 'q1', 'local-1') + store.messages.addMessage(session.id, 'q2', 'local-2') + store.messages.addMessage(session.id, 'q3', 'local-3') + + const queuedBefore = store.messages.getUninvokedLocalMessages(session.id) + expect(queuedBefore).toHaveLength(3) + const localIds = queuedBefore + .map(m => m.localId) + .filter((id): id is string => id !== null) + expect(localIds).toHaveLength(3) + + const ts = Date.now() + store.messages.markMessagesInvoked(session.id, localIds, ts) + + const queuedAfter = store.messages.getUninvokedLocalMessages(session.id) + expect(queuedAfter).toHaveLength(0) + // And every row now carries the same invokedAt — a partial sweep would + // leave the floating bar half-cleared on the web client. + const allMessages = store.messages.getMessages(session.id) + for (const msg of allMessages) { + expect(msg.invokedAt).toBe(ts) + } + }) +}) + +describe('Store V8 byPosition pagination', () => { + it('getMessagesByPosition returns messages in ascending order', () => { + const store = new Store(':memory:') + const session = store.sessions.getOrCreateSession('test', { path: '/tmp' }, null, 'default') + const msg1 = store.messages.addMessage(session.id, 'a', 'loc-1') + const msg2 = store.messages.addMessage(session.id, 'b', 'loc-2') + const msg3 = store.messages.addMessage(session.id, 'c', 'loc-3') + // All start with null invokedAt; set different invokedAt values + store.messages.markMessagesInvoked(session.id, ['loc-1'], 1000) + store.messages.markMessagesInvoked(session.id, ['loc-2'], 2000) + store.messages.markMessagesInvoked(session.id, ['loc-3'], 3000) + + const result = store.messages.getMessagesByPosition(session.id, 50) + expect(result.map(m => m.id)).toEqual([msg1.id, msg2.id, msg3.id]) + }) + + it('getMessagesByPosition sorts by invokedAt DESC, seq DESC (latest first, reversed to ascending)', () => { + const store = new Store(':memory:') + const session = store.sessions.getOrCreateSession('test', { path: '/tmp' }, null, 'default') + // Insert 3 messages: msg1 has low seq but high invokedAt (queued message that was consumed late) + const msg1 = store.messages.addMessage(session.id, 'queued', 'loc-q') + const msg2 = store.messages.addMessage(session.id, 'normal-1') // no localId → invokedAt = createdAt + const msg3 = store.messages.addMessage(session.id, 'normal-2') // no localId → invokedAt = createdAt + + // Simulate: msg1 (seq=1) is invoked much later than msg2 and msg3 + store.messages.markMessagesInvoked(session.id, ['loc-q'], msg3.createdAt + 10_000) + + const result = store.messages.getMessagesByPosition(session.id, 50) + // Expected order by position_at ASC: msg2, msg3, msg1 (msg1 has highest invokedAt) + expect(result[result.length - 1].id).toBe(msg1.id) + expect(result[0].id).toBe(msg2.id) + }) + + it('getMessagesByPosition composite cursor: second page has no gap or duplicate', () => { + const store = new Store(':memory:') + const session = store.sessions.getOrCreateSession('test', { path: '/tmp' }, null, 'default') + // Add 5 messages with distinct invokedAt timestamps + const messages = [] + for (let i = 0; i < 5; i++) { + const msg = store.messages.addMessage(session.id, `msg-${i}`) + messages.push(msg) + } + + // First page: limit=3 (gets last 3 by position DESC, reversed to ASC) + const page1 = store.messages.getMessagesByPosition(session.id, 3) + expect(page1).toHaveLength(3) + + // Derive cursor from oldest in page1 (first element after reverse) + const oldest = page1[0] + const cursorAt = oldest.invokedAt ?? oldest.createdAt + const cursorSeq = oldest.seq + + // Second page + const page2 = store.messages.getMessagesByPosition(session.id, 3, { at: cursorAt, seq: cursorSeq }) + expect(page2).toHaveLength(2) + + // No overlap between pages + const page1Ids = new Set(page1.map(m => m.id)) + const page2Ids = new Set(page2.map(m => m.id)) + for (const id of page2Ids) { + expect(page1Ids.has(id)).toBe(false) + } + + // Together they cover all 5 messages + expect(page1Ids.size + page2Ids.size).toBe(5) + }) + + it('long session: low-seq late-invokedAt message appears in first page', () => { + const store = new Store(':memory:') + const session = store.sessions.getOrCreateSession('test', { path: '/tmp' }, null, 'default') + // Insert many normal messages first (low invokedAt) + for (let i = 0; i < 10; i++) { + store.messages.addMessage(session.id, `normal-${i}`) + } + // Insert a queued message (low seq, but invoked much later) + const queued = store.messages.addMessage(session.id, 'queued', 'loc-q') + store.messages.markMessagesInvoked(session.id, ['loc-q'], Date.now() + 1_000_000) + + // The queued message should appear in first page (highest position_at) + const page1 = store.messages.getMessagesByPosition(session.id, 5) + const ids = page1.map(m => m.id) + expect(ids).toContain(queued.id) + // It should be the last (most recent) in ascending result + expect(ids[ids.length - 1]).toBe(queued.id) + }) + + it('V7 mode getMessages is unchanged after V8 migration', () => { + const dir = mkdtempSync(join(tmpdir(), 'hapi-v7-compat-')) + const dbPath = join(dir, 'test.db') + try { + const db = new Database(dbPath, { create: true, readwrite: true, strict: true }) + db.exec('PRAGMA journal_mode = WAL') + db.exec('PRAGMA foreign_keys = ON') + createV7Schema(db) + db.exec('PRAGMA user_version = 7') + db.exec(`INSERT INTO sessions (id, namespace, created_at, updated_at, seq) + VALUES ('s1', 'default', 1000, 1000, 0)`) + db.exec(`INSERT INTO messages (id, session_id, content, created_at, seq) + VALUES ('m1', 's1', '"hello"', 1000, 1), ('m2', 's1', '"world"', 2000, 2)`) + db.close() + + const store = new Store(dbPath) + // V7 getMessages (seq-based) must still work + const msgs = store.messages.getMessages('s1') + expect(msgs).toHaveLength(2) + expect(msgs[0].seq).toBe(1) + expect(msgs[1].seq).toBe(2) + } finally { + rmSync(dir, { recursive: true, force: true }) + } + }) + + it('idx_messages_session_position index exists on fresh DB', () => { + const store = new Store(':memory:') + const db: Database = (store as any).db + const rows = db.prepare( + "SELECT name FROM sqlite_master WHERE type='index' AND name='idx_messages_session_position'" + ).all() as Array<{ name: string }> + expect(rows).toHaveLength(1) + }) + + it('idx_messages_session_position index exists after V7→V8 migration', () => { + const dir = mkdtempSync(join(tmpdir(), 'hapi-index-v7-v8-')) + const dbPath = join(dir, 'test.db') + try { + const db = new Database(dbPath, { create: true, readwrite: true, strict: true }) + db.exec('PRAGMA journal_mode = WAL') + db.exec('PRAGMA foreign_keys = ON') + createV7Schema(db) + db.exec('PRAGMA user_version = 7') + db.close() + + const store = new Store(dbPath) + const db2: Database = (store as any).db + const rows = db2.prepare( + "SELECT name FROM sqlite_master WHERE type='index' AND name='idx_messages_session_position'" + ).all() as Array<{ name: string }> + expect(rows).toHaveLength(1) + } finally { + rmSync(dir, { recursive: true, force: true }) + } + }) + + // The web client renders queued messages from the union of the latest + // page (getMessagesByPosition) and the uninvoked-local set + // (getUninvokedLocalMessages). This test pins that contract at the store + // layer: a low-position queued row must NOT appear in the latest page once + // it's been pushed out, but it must still be discoverable via the + // uninvoked set so the floating bar can render it. + it('latest page + uninvoked union: queued rows pushed out of the page are still surfaced', () => { + const store = new Store(':memory:') + const session = store.sessions.getOrCreateSession('test', { path: '/tmp' }, null, 'default') + + // Queued message lands first → low createdAt; invoked_at stays NULL, + // so its position_at = createdAt (the lowest in the session). + const queued = store.messages.addMessage(session.id, 'queued', 'local-q') + + // Add enough later (auto-invoked) messages to push the queued row out + // of a 3-row latest page. + for (let i = 0; i < 5; i++) { + store.messages.addMessage(session.id, `later-${i}`) + } + + const pageRows = store.messages.getMessagesByPosition(session.id, 3) + expect(pageRows).toHaveLength(3) + expect(pageRows.find(m => m.id === queued.id)).toBeUndefined() + + // ...but the uninvoked union still surfaces it. + const queuedRows = store.messages.getUninvokedLocalMessages(session.id) + expect(queuedRows.map(m => m.id)).toContain(queued.id) + }) + + // Pins the latest-page ordering contract used as the cursor anchor on the + // web side: page rows are returned in ascending position order, so + // pageRows[0] is the oldest row in the page and is the correct anchor for + // the next older fetch. + it('getMessagesByPosition ascending order: pageRows[0] is the oldest in the page', () => { + const store = new Store(':memory:') + const session = store.sessions.getOrCreateSession('test', { path: '/tmp' }, null, 'default') + const m1 = store.messages.addMessage(session.id, 'm1') + const m2 = store.messages.addMessage(session.id, 'm2') + const m3 = store.messages.addMessage(session.id, 'm3') + + const page = store.messages.getMessagesByPosition(session.id, 10) + expect(page).toHaveLength(3) + // Ascending by position_at, with seq as the tiebreaker — m1 is oldest, + // m3 is newest. If this ever flips, the web client's + // `oldestPositionAt = pageRows[0].position` would anchor to the wrong + // end of the page and the next loadMore would either gap or duplicate. + expect(page[0].id).toBe(m1.id) + expect(page[2].id).toBe(m3.id) + }) + + it('legacy DB (user_version=0 with V7-shape tables): step ladder backfills invoked_at and index', () => { + const dir = mkdtempSync(join(tmpdir(), 'hapi-legacy-v0-')) + const dbPath = join(dir, 'test.db') + try { + // Build a V7-shape schema but leave user_version = 0 (legacy DB + // predating the version stamping). The legacy branch in initSchema + // must run the step ladder so the messages table picks up + // invoked_at + idx_messages_session_position. + const db = new Database(dbPath, { create: true, readwrite: true, strict: true }) + db.exec('PRAGMA journal_mode = WAL') + db.exec('PRAGMA foreign_keys = ON') + createV7Schema(db) + // Intentionally do NOT set user_version — leaves it at 0. + db.exec(`INSERT INTO sessions (id, namespace, created_at, updated_at, seq) + VALUES ('s1', 'default', 1000, 1000, 0)`) + db.exec(`INSERT INTO messages (id, session_id, content, created_at, seq) + VALUES ('m1', 's1', '"hi"', 1500, 1)`) + db.close() + + const store = new Store(dbPath) + const cols = getMessageColumns(store) + expect(cols).toContain('invoked_at') + + // Backfill should have happened via V7→V8 step running in the legacy branch. + const msgs = store.messages.getMessages('s1') + expect(msgs).toHaveLength(1) + expect(msgs[0].invokedAt).toBe(1500) + + // The position index must exist for byPosition pagination to work. + const db2: Database = (store as any).db + const rows = db2.prepare( + "SELECT name FROM sqlite_master WHERE type='index' AND name='idx_messages_session_position'" + ).all() as Array<{ name: string }> + expect(rows).toHaveLength(1) + } finally { + rmSync(dir, { recursive: true, force: true }) + } + }) +}) + +function getMessageColumns(store: Store): string[] { + // Access internal db via reflection — safe for test only + const db: Database = (store as any).db + const rows = db.prepare('PRAGMA table_info(messages)').all() as Array<{ name: string }> + return rows.map(r => r.name) +} + +function getSessionColumns(store: Store): string[] { + const db: Database = (store as any).db + const rows = db.prepare('PRAGMA table_info(sessions)').all() as Array<{ name: string }> + return rows.map(r => r.name) +} + +/** V7 schema: messages table without invoked_at */ +function createV7Schema(db: Database): void { + db.exec(` + CREATE TABLE IF NOT EXISTS 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, + 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 INDEX IF NOT EXISTS idx_sessions_tag ON sessions(tag); + CREATE INDEX IF NOT EXISTS idx_sessions_tag_namespace ON sessions(tag, namespace); + + CREATE TABLE IF NOT EXISTS 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 INDEX IF NOT EXISTS idx_machines_namespace ON machines(namespace); + + CREATE TABLE IF NOT EXISTS 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, + FOREIGN KEY (session_id) REFERENCES sessions(id) ON DELETE CASCADE + ); + CREATE INDEX IF NOT EXISTS idx_messages_session ON messages(session_id, seq); + CREATE UNIQUE INDEX IF NOT EXISTS idx_messages_local_id ON messages(session_id, local_id) WHERE local_id IS NOT NULL; + + CREATE TABLE IF NOT EXISTS 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 INDEX IF NOT EXISTS idx_users_platform ON users(platform); + CREATE INDEX IF NOT EXISTS idx_users_platform_namespace ON users(platform, namespace); + + CREATE TABLE IF NOT EXISTS 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 INDEX IF NOT EXISTS idx_push_subscriptions_namespace ON push_subscriptions(namespace); + `) +} + +/** V6 schema: sessions without model_reasoning_effort; messages without invoked_at */ +function createV6Schema(db: Database): void { + db.exec(` + CREATE TABLE IF NOT EXISTS 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, + effort 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 INDEX IF NOT EXISTS idx_sessions_tag ON sessions(tag); + CREATE INDEX IF NOT EXISTS idx_sessions_tag_namespace ON sessions(tag, namespace); + + CREATE TABLE IF NOT EXISTS 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 INDEX IF NOT EXISTS idx_machines_namespace ON machines(namespace); + + CREATE TABLE IF NOT EXISTS 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, + FOREIGN KEY (session_id) REFERENCES sessions(id) ON DELETE CASCADE + ); + CREATE INDEX IF NOT EXISTS idx_messages_session ON messages(session_id, seq); + CREATE UNIQUE INDEX IF NOT EXISTS idx_messages_local_id ON messages(session_id, local_id) WHERE local_id IS NOT NULL; + + CREATE TABLE IF NOT EXISTS 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 INDEX IF NOT EXISTS idx_users_platform ON users(platform); + CREATE INDEX IF NOT EXISTS idx_users_platform_namespace ON users(platform, namespace); + + CREATE TABLE IF NOT EXISTS 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 INDEX IF NOT EXISTS idx_push_subscriptions_namespace ON push_subscriptions(namespace); + `) +} + +/** V5 schema: sessions without effort, model_reasoning_effort; messages without invoked_at */ +function createV5Schema(db: Database): void { + db.exec(` + CREATE TABLE IF NOT EXISTS 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, + 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 INDEX IF NOT EXISTS idx_sessions_tag ON sessions(tag); + CREATE INDEX IF NOT EXISTS idx_sessions_tag_namespace ON sessions(tag, namespace); + + CREATE TABLE IF NOT EXISTS 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 INDEX IF NOT EXISTS idx_machines_namespace ON machines(namespace); + + CREATE TABLE IF NOT EXISTS 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, + FOREIGN KEY (session_id) REFERENCES sessions(id) ON DELETE CASCADE + ); + CREATE INDEX IF NOT EXISTS idx_messages_session ON messages(session_id, seq); + CREATE UNIQUE INDEX IF NOT EXISTS idx_messages_local_id ON messages(session_id, local_id) WHERE local_id IS NOT NULL; + + CREATE TABLE IF NOT EXISTS 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 INDEX IF NOT EXISTS idx_users_platform ON users(platform); + CREATE INDEX IF NOT EXISTS idx_users_platform_namespace ON users(platform, namespace); + + CREATE TABLE IF NOT EXISTS 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 INDEX IF NOT EXISTS idx_push_subscriptions_namespace ON push_subscriptions(namespace); + `) +} + +/** V4 schema: sessions without model, effort, model_reasoning_effort; messages without invoked_at */ +function createV4Schema(db: Database): void { + db.exec(` + CREATE TABLE IF NOT EXISTS 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, + 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 INDEX IF NOT EXISTS idx_sessions_tag ON sessions(tag); + CREATE INDEX IF NOT EXISTS idx_sessions_tag_namespace ON sessions(tag, namespace); + + CREATE TABLE IF NOT EXISTS 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 INDEX IF NOT EXISTS idx_machines_namespace ON machines(namespace); + + CREATE TABLE IF NOT EXISTS 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, + FOREIGN KEY (session_id) REFERENCES sessions(id) ON DELETE CASCADE + ); + CREATE INDEX IF NOT EXISTS idx_messages_session ON messages(session_id, seq); + CREATE UNIQUE INDEX IF NOT EXISTS idx_messages_local_id ON messages(session_id, local_id) WHERE local_id IS NOT NULL; + + CREATE TABLE IF NOT EXISTS 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 INDEX IF NOT EXISTS idx_users_platform ON users(platform); + CREATE INDEX IF NOT EXISTS idx_users_platform_namespace ON users(platform, namespace); + + CREATE TABLE IF NOT EXISTS 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 INDEX IF NOT EXISTS idx_push_subscriptions_namespace ON push_subscriptions(namespace); + `) +} diff --git a/hub/src/store/types.ts b/hub/src/store/types.ts index 9297a5e6..a81762f7 100644 --- a/hub/src/store/types.ts +++ b/hub/src/store/types.ts @@ -42,6 +42,7 @@ export type StoredMessage = { createdAt: number seq: number localId: string | null + invokedAt: number | null } export type StoredUser = { diff --git a/hub/src/sync/messageService.ts b/hub/src/sync/messageService.ts index e9be24b8..0aba66a1 100644 --- a/hub/src/sync/messageService.ts +++ b/hub/src/sync/messageService.ts @@ -27,7 +27,8 @@ export class MessageService { seq: message.seq, localId: message.localId, content: message.content, - createdAt: message.createdAt + createdAt: message.createdAt, + invokedAt: message.invokedAt })) let oldestSeq: number | null = null @@ -53,6 +54,75 @@ export class MessageService { } } + getMessagesPageByPosition( + sessionId: string, + options: { limit: number; before?: { at: number; seq: number } | null } + ): { + messages: DecryptedMessage[] + page: { + limit: number + nextBeforeSeq: number | null + nextBeforeAt: number | null + hasMore: boolean + } + } { + const before = options.before ?? undefined + const pageRows = this.store.messages.getMessagesByPosition(sessionId, options.limit, before) + + // 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. + const queuedRows = before === undefined + ? this.store.messages.getUninvokedLocalMessages(sessionId) + : [] + + const byId = new Map() + for (const row of pageRows) byId.set(row.id, row) + for (const row of queuedRows) byId.set(row.id, row) + + const stored = [...byId.values()].sort((a, b) => { + const at = (a.invokedAt ?? a.createdAt) - (b.invokedAt ?? b.createdAt) + return at !== 0 ? at : a.seq - b.seq + }) + + const messages: DecryptedMessage[] = stored.map((message) => ({ + id: message.id, + seq: message.seq, + localId: message.localId, + content: message.content, + createdAt: message.createdAt, + invokedAt: message.invokedAt + })) + + // The cursor is the oldest row in the actual position-ordered page (pageRows[0]). + // Out-of-band queued rows are not part of the cursor — they are pinned to + // every latest-page response. + const oldest = pageRows[0] ?? null + const oldestSeq: number | null = oldest?.seq ?? null + const oldestPositionAt: number | null = oldest + ? oldest.invokedAt ?? oldest.createdAt + : null + + const hasMore = oldestSeq !== null && oldestPositionAt !== null + && this.store.messages.getMessagesByPosition( + sessionId, + 1, + { at: oldestPositionAt, seq: oldestSeq } + ).length > 0 + + return { + messages, + page: { + limit: options.limit, + nextBeforeSeq: oldestSeq, + nextBeforeAt: oldestPositionAt, + hasMore + } + } + } + getMessagesAfter(sessionId: string, options: { afterSeq: number; limit: number }): DecryptedMessage[] { const stored = this.store.messages.getMessagesAfter(sessionId, options.afterSeq, options.limit) return stored.map((message) => ({ @@ -60,7 +130,8 @@ export class MessageService { seq: message.seq, localId: message.localId, content: message.content, - createdAt: message.createdAt + createdAt: message.createdAt, + invokedAt: message.invokedAt })) } @@ -116,7 +187,8 @@ export class MessageService { seq: msg.seq, localId: msg.localId, content: msg.content, - createdAt: msg.createdAt + createdAt: msg.createdAt, + invokedAt: msg.invokedAt } }) } diff --git a/hub/src/sync/syncEngine.ts b/hub/src/sync/syncEngine.ts index b2246cd3..4c561f04 100644 --- a/hub/src/sync/syncEngine.ts +++ b/hub/src/sync/syncEngine.ts @@ -166,6 +166,21 @@ export class SyncEngine { return this.messageService.getMessagesPage(sessionId, options) } + getMessagesPageByPosition( + sessionId: string, + options: { limit: number; before?: { at: number; seq: number } | null } + ): { + messages: DecryptedMessage[] + page: { + limit: number + nextBeforeSeq: number | null + nextBeforeAt: number | null + hasMore: boolean + } + } { + return this.messageService.getMessagesPageByPosition(sessionId, options) + } + getMessagesAfter(sessionId: string, options: { afterSeq: number; limit: number }): DecryptedMessage[] { return this.messageService.getMessagesAfter(sessionId, options) } diff --git a/hub/src/web/routes/messages.ts b/hub/src/web/routes/messages.ts index 492298f2..1ce8e026 100644 --- a/hub/src/web/routes/messages.ts +++ b/hub/src/web/routes/messages.ts @@ -7,7 +7,9 @@ import { requireSessionFromParam, requireSyncEngine } from './guards' const querySchema = z.object({ limit: z.coerce.number().int().min(1).max(200).optional(), - beforeSeq: z.coerce.number().int().min(1).optional() + beforeSeq: z.coerce.number().int().min(1).optional(), + byPosition: z.string().optional(), + beforeAt: z.coerce.number().int().min(0).optional(), }) const sendMessageBodySchema = z.object({ @@ -33,6 +35,18 @@ export function createMessagesRoutes(getSyncEngine: () => SyncEngine | null): Ho const parsed = querySchema.safeParse(c.req.query()) const limit = parsed.success ? (parsed.data.limit ?? 50) : 50 + + // V8 byPosition mode: use composite (position_at, seq) cursor + if (parsed.success && parsed.data.byPosition === '1') { + const beforeAt = parsed.data.beforeAt + const beforeSeq = parsed.data.beforeSeq + const before = (beforeAt !== undefined && beforeSeq !== undefined) + ? { at: beforeAt, seq: beforeSeq } + : null + return c.json(engine.getMessagesPageByPosition(sessionId, { limit, before })) + } + + // V7-compatible path: seq-based cursor const beforeSeq = parsed.success ? (parsed.data.beforeSeq ?? null) : null return c.json(engine.getMessagesPage(sessionId, { limit, beforeSeq })) }) diff --git a/shared/src/schemas.ts b/shared/src/schemas.ts index cb9f448d..9a4ca011 100644 --- a/shared/src/schemas.ts +++ b/shared/src/schemas.ts @@ -154,7 +154,8 @@ export const DecryptedMessageSchema = z.object({ seq: z.number().nullable(), localId: z.string().nullable(), content: z.unknown(), - createdAt: z.number() + createdAt: z.number(), + invokedAt: z.number().nullable().optional() }) export type DecryptedMessage = z.infer @@ -236,7 +237,8 @@ export const SyncEventSchema = z.discriminatedUnion('type', [ }), SessionChangedSchema.extend({ type: z.literal('messages-consumed'), - localIds: z.array(z.string()) + localIds: z.array(z.string()), + invokedAt: z.number().optional() }), SessionEventBaseSchema.extend({ type: z.literal('heartbeat'), diff --git a/web/src/api/client.ts b/web/src/api/client.ts index 15034358..b5903f2a 100644 --- a/web/src/api/client.ts +++ b/web/src/api/client.ts @@ -191,8 +191,22 @@ export class ApiClient { return await this.request(`/api/sessions/${encodeURIComponent(sessionId)}`) } - async getMessages(sessionId: string, options: { beforeSeq?: number | null; limit?: number }): Promise { + async getMessages( + sessionId: string, + options: { + beforeSeq?: number | null + beforeAt?: number | null + byPosition?: boolean + limit?: number + } + ): Promise { const params = new URLSearchParams() + if (options.byPosition || options.beforeAt !== undefined && options.beforeAt !== null) { + params.set('byPosition', '1') + } + if (options.beforeAt !== undefined && options.beforeAt !== null) { + params.set('beforeAt', `${options.beforeAt}`) + } if (options.beforeSeq !== undefined && options.beforeSeq !== null) { params.set('beforeSeq', `${options.beforeSeq}`) } diff --git a/web/src/components/AssistantChat/QueuedMessagesBar.tsx b/web/src/components/AssistantChat/QueuedMessagesBar.tsx new file mode 100644 index 00000000..d2fc9846 --- /dev/null +++ b/web/src/components/AssistantChat/QueuedMessagesBar.tsx @@ -0,0 +1,109 @@ +import { useCallback, useSyncExternalStore } from 'react' +import { getMessageWindowState, subscribeMessageWindow } from '@/lib/message-window-store' +import { isQueuedForInvocation } from '@/lib/messages' +import { EMPTY_STATE } from '@/hooks/queries/useMessages' +import { normalizeDecryptedMessage } from '@/chat/normalize' +import type { DecryptedMessage } from '@/types/api' + +function ClockIcon() { + return ( + + ) +} + +/** + * Returns user messages that haven't been invoked yet (invokedAt == null and not sent/failed). + * Covers both optimistic (status='queued') and server-loaded (status=undefined, invokedAt=null) cases. + */ +function useQueuedMessages(sessionId: string): DecryptedMessage[] { + const state = useSyncExternalStore( + useCallback((listener) => subscribeMessageWindow(sessionId, listener), [sessionId]), + useCallback(() => getMessageWindowState(sessionId), [sessionId]), + () => EMPTY_STATE + ) + + // `invokedAt` is the source of truth for invocation; see isQueuedForInvocation + // (lib/messages) for the shared predicate used by the thread filter and the + // window store trim helpers. + const allMessages = [...state.messages, ...state.pending] + return allMessages.filter(isQueuedForInvocation) +} + +function getTextFromMessage(msg: DecryptedMessage): string { + const normalized = normalizeDecryptedMessage(msg) + if (!normalized || normalized.role !== 'user') { + return '' + } + const text = (normalized.content.text ?? '').trim() + if (text) { + return text + } + // Attachment-only sends: the composer / POST /messages allow empty text + // when attachments are present. Fall back to the filenames so the chip + // is not blank. + const attachments = normalized.content.attachments ?? [] + if (attachments.length === 0) { + return '' + } + return attachments.map((a) => a.filename ?? 'attachment').join(', ') +} + +/** + * Floating bar above the composer showing queued (pending invocation) messages. + * Disappears automatically when all queued messages are invoked or consumed. + * + * TODO PR 2: add cancel/edit buttons per item. + */ +export function QueuedMessagesBar({ sessionId }: { sessionId: string }) { + const queued = useQueuedMessages(sessionId) + + if (queued.length === 0) { + return null + } + + return ( +
+
+
+ + Queued +
+
    + {queued.map((msg) => { + const text = getTextFromMessage(msg) + return ( +
  • + {text} + {/* TODO PR 2: cancel/edit buttons */} +
    • + ) + })} +
+
+
+ ) +} diff --git a/web/src/components/AssistantChat/messages/UserMessage.tsx b/web/src/components/AssistantChat/messages/UserMessage.tsx index 75141470..bffa7023 100644 --- a/web/src/components/AssistantChat/messages/UserMessage.tsx +++ b/web/src/components/AssistantChat/messages/UserMessage.tsx @@ -47,7 +47,7 @@ export function HappyUserMessage() { const canRetry = status === 'failed' && typeof localId === 'string' && Boolean(ctx.onRetryMessage) const onRetry = canRetry ? () => ctx.onRetryMessage!(localId) : undefined - const userBubbleClass = `w-fit min-w-0 max-w-[92%] ml-auto rounded-xl bg-[var(--app-secondary-bg)] px-3 py-2 text-[var(--app-fg)] shadow-sm${status === 'queued' ? ' opacity-60' : ''}` + const userBubbleClass = `w-fit min-w-0 max-w-[92%] ml-auto rounded-xl bg-[var(--app-secondary-bg)] px-3 py-2 text-[var(--app-fg)] shadow-sm` if (isCliOutput) { return ( diff --git a/web/src/components/SessionChat.tsx b/web/src/components/SessionChat.tsx index a27cbbf9..12c840cb 100644 --- a/web/src/components/SessionChat.tsx +++ b/web/src/components/SessionChat.tsx @@ -16,8 +16,10 @@ import { normalizeDecryptedMessage } from '@/chat/normalize' import { reduceChatBlocks } from '@/chat/reducer' import { reconcileChatBlocks } from '@/chat/reconcile' import { buildConversationOutline } from '@/chat/outline' +import { isQueuedForInvocation } from '@/lib/messages' import { HappyComposer } from '@/components/AssistantChat/HappyComposer' import { HappyThread } from '@/components/AssistantChat/HappyThread' +import { QueuedMessagesBar } from '@/components/AssistantChat/QueuedMessagesBar' import { useHappyRuntime } from '@/lib/assistant-runtime' import { createAttachmentAdapter } from '@/lib/attachmentAdapter' import { useTranslation } from '@/lib/use-translation' @@ -209,6 +211,15 @@ export function SessionChat(props: { setOutlineOpen(false) }, [props.session.id]) + // Exclude user messages that haven't been invoked yet — those appear in the + // QueuedMessagesBar above the composer, not in the thread timeline. The + // `isQueuedForInvocation` predicate is shared with the window store and the + // floating bar so the three views never disagree about queued state. + const visibleMessages = useMemo( + () => props.messages.filter((m) => !isQueuedForInvocation(m)), + [props.messages] + ) + const normalizedMessages: NormalizedMessage[] = useMemo(() => { // Clear caches immediately when session changes (before useEffect runs) if (prevSessionIdRef.current !== null && prevSessionIdRef.current !== props.session.id) { @@ -220,7 +231,7 @@ export function SessionChat(props: { const cache = normalizedCacheRef.current const normalized: NormalizedMessage[] = [] const seen = new Set() - for (const message of props.messages) { + for (const message of visibleMessages) { seen.add(message.id) const cached = cache.get(message.id) if (cached && cached.source === message) { @@ -237,7 +248,7 @@ export function SessionChat(props: { } } return normalized - }, [props.messages]) + }, [visibleMessages]) const reduced = useMemo( () => reduceChatBlocks(normalizedMessages, props.session.agentState), @@ -408,7 +419,7 @@ export function SessionChat(props: { isLoadingMoreMessages={props.isLoadingMoreMessages} onLoadMore={props.onLoadMore} pendingCount={props.pendingCount} - rawMessagesCount={props.messages.length} + rawMessagesCount={visibleMessages.length} normalizedMessagesCount={normalizedMessages.length} messagesVersion={props.messagesVersion} forceScrollToken={forceScrollToken} @@ -426,6 +437,10 @@ export function SessionChat(props: { ) : null} +
+ +
+ message.id)) + const regular = messages.filter((message) => !queuedIds.has(message.id)) + const budget = Math.max(0, limit - queued.length) + const trimmedRegular = mode === 'prepend' + ? regular.slice(0, budget) + : regular.slice(Math.max(0, regular.length - budget)) + const droppedRegular = mode === 'prepend' + ? regular.slice(budget) + : regular.slice(0, Math.max(0, regular.length - budget)) + return { kept: mergeMessages(trimmedRegular, queued), dropped: droppedRegular } +} + function trimVisible(messages: DecryptedMessage[], mode: 'append' | 'prepend'): DecryptedMessage[] { - if (messages.length <= VISIBLE_WINDOW_SIZE) { - return messages - } - if (mode === 'prepend') { - return messages.slice(0, VISIBLE_WINDOW_SIZE) - } - return messages.slice(messages.length - VISIBLE_WINDOW_SIZE) + return trimPreservingQueued(messages, VISIBLE_WINDOW_SIZE, mode).kept } function trimPending( @@ -272,11 +315,12 @@ function trimPending( if (messages.length <= PENDING_WINDOW_SIZE) { return { pending: messages, dropped: 0, droppedVisible: 0 } } - const cutoff = messages.length - PENDING_WINDOW_SIZE - const droppedMessages = messages.slice(0, cutoff) - const pending = messages.slice(cutoff) - const droppedVisible = countVisiblePendingMessages(sessionId, droppedMessages) - return { pending, dropped: droppedMessages.length, droppedVisible } + // Symmetric with trimVisible: agents that overflow the pending window + // (200) must not evict queued user messages — the floating bar holds the + // only client-visible reference to them until the CLI ack arrives. + const { kept, dropped } = trimPreservingQueued(messages, PENDING_WINDOW_SIZE, 'append') + const droppedVisible = countVisiblePendingMessages(sessionId, dropped) + return { pending: kept, dropped: dropped.length, droppedVisible } } function filterPendingAgainstVisible(pending: DecryptedMessage[], visible: DecryptedMessage[]): DecryptedMessage[] { @@ -360,6 +404,8 @@ export function seedMessageWindowFromSession(fromSessionId: string, toSessionId: pendingOverflowCount: source.pendingOverflowCount, pendingOverflowVisibleCount: source.pendingOverflowVisibleCount, hasMore: source.hasMore, + oldestPositionAt: source.oldestPositionAt, + oldestPositionSeq: source.oldestPositionSeq, warning: source.warning, atBottom: source.atBottom, isLoading: false, @@ -376,7 +422,17 @@ export async function fetchLatestMessages(api: ApiClient, sessionId: string): Pr updateState(sessionId, (prev) => buildState(prev, { isLoading: true, warning: null })) try { - const response = await api.getMessages(sessionId, { limit: PAGE_SIZE, beforeSeq: null }) + // Always request byPosition mode (V8). If the hub is V7 it ignores byPosition and + // returns the standard seq-based response (no nextBeforeAt field) — we fall back + // to seq-cursor mode seamlessly. + const response = await api.getMessages(sessionId, { byPosition: true, limit: PAGE_SIZE }) + // Derive composite cursor pair from server response. Both values come from + // the same row on the server; we keep them paired so the next older fetch + // doesn't mix `beforeAt` from the server with a recomputed minimum `seq`. + const nextBeforeAt = response.page.nextBeforeAt ?? null + const nextBeforeSeq = response.page.nextBeforeSeq ?? null + const isV8Cursor = nextBeforeAt !== null && nextBeforeSeq !== null + updateState(sessionId, (prev) => { if (prev.atBottom) { const merged = mergeMessages(prev.messages, [...prev.pending, ...response.messages]) @@ -388,6 +444,8 @@ export async function fetchLatestMessages(api: ApiClient, sessionId: string): Pr pendingVisibleCount: 0, pendingOverflowVisibleCount: 0, hasMore: response.page.hasMore, + oldestPositionAt: isV8Cursor ? nextBeforeAt : null, + oldestPositionSeq: isV8Cursor ? nextBeforeSeq : null, isLoading: false, warning: null, }) @@ -398,6 +456,12 @@ export async function fetchLatestMessages(api: ApiClient, sessionId: string): Pr pendingVisibleCount: pendingResult.pendingVisibleCount, pendingOverflowCount: pendingResult.pendingOverflowCount, pendingOverflowVisibleCount: pendingResult.pendingOverflowVisibleCount, + // Persist the V8 cursor pair on the non-at-bottom path too. Without this + // a refresh while scrolled up dropped the composite cursor and the next + // loadMore fell back to V7 seq mode against a V8 hub — the same + // asymmetric class of bug the at-bottom branch already guards against. + oldestPositionAt: isV8Cursor ? nextBeforeAt : null, + oldestPositionSeq: isV8Cursor ? nextBeforeSeq : null, isLoading: false, warning: pendingResult.warning, }) @@ -419,13 +483,31 @@ export async function fetchOlderMessages(api: ApiClient, sessionId: string): Pro updateState(sessionId, (prev) => buildState(prev, { isLoadingMore: true })) try { - const response = await api.getMessages(sessionId, { limit: PAGE_SIZE, beforeSeq: initial.oldestSeq }) + // V8 mode: use the server-provided cursor pair as-is. Mixing `beforeAt` from + // the server with a recomputed minimum `seq` from the local window can refer + // to different rows after a low-seq message is invoked late. + const useV8Cursor = initial.oldestPositionAt !== null && initial.oldestPositionSeq !== null + const response = useV8Cursor + ? await api.getMessages(sessionId, { + byPosition: true, + beforeAt: initial.oldestPositionAt!, + beforeSeq: initial.oldestPositionSeq!, + limit: PAGE_SIZE + }) + : await api.getMessages(sessionId, { beforeSeq: initial.oldestSeq, limit: PAGE_SIZE }) + + const nextBeforeAt = response.page.nextBeforeAt ?? null + const nextBeforeSeq = response.page.nextBeforeSeq ?? null + const isV8Cursor = nextBeforeAt !== null && nextBeforeSeq !== null + updateState(sessionId, (prev) => { const merged = mergeMessages(response.messages, prev.messages) const trimmed = trimVisible(merged, 'prepend') return buildState(prev, { messages: trimmed, hasMore: response.page.hasMore, + oldestPositionAt: isV8Cursor ? nextBeforeAt : null, + oldestPositionSeq: isV8Cursor ? nextBeforeSeq : null, isLoadingMore: false, }) }) @@ -538,24 +620,78 @@ export function updateMessageStatus(sessionId: string, localId: string, status: }) } -/** Transition the queued messages whose localIds match to 'sent'. Driven by the - * CLI ack (messages-consumed). Unmatched messages remain queued. */ -export function markMessagesConsumed(sessionId: string, localIds: string[]): void { +/** Transition the queued messages whose localIds match to 'sent' and record invokedAt. + * Driven by the CLI ack (messages-consumed). Unmatched messages remain queued. + * Also handles server-loaded messages (status=undefined) that have a matching localId. + * V7 hub compat: if `invokedAt` is undefined the SyncEvent had no server timestamp, + * so we fall back to client time — without it the row would stay queued forever + * under the strict-null filter. The fallback only affects display ordering on + * this client; the persisted server value is the authoritative one when present. */ +export function markMessagesConsumed(sessionId: string, localIds: string[], invokedAt: number | undefined): void { if (localIds.length === 0) return const idSet = new Set(localIds) + const effectiveInvokedAt = invokedAt ?? Date.now() updateState(sessionId, (prev) => { let changed = false const updateList = (list: DecryptedMessage[]) => { return list.map((message) => { - if (message.status !== 'queued' || !message.localId || !idSet.has(message.localId)) { + if (!message.localId || !idSet.has(message.localId)) { + return message + } + if (message.status === 'failed') { + return message + } + // Apply the ack even if the message is already 'sent' (optimistic) — otherwise + // a message that flipped to 'sent' before the consume event arrives would + // never receive `invokedAt` and keep sorting by send time. + // First-write-wins on `invokedAt`: mirror the hub's UPDATE guard so a + // duplicate `messages-consumed` (e.g. CLI re-emit) doesn't restamp a + // message and shuffle its byPosition slot on live clients while the + // DB still holds the original timestamp. + const needsStatus = message.status !== 'sent' + // Strict null to stay consistent with isQueuedForInvocation and the rest + // of this file. The idSet filter already shields V7-stamped rows from + // this path, but the strict-null contract should not vary by call site. + const needsInvokedAt = message.invokedAt === null + if (!needsStatus && !needsInvokedAt) { return message } changed = true - return { ...message, status: 'sent' as MessageStatus } + const update: Partial = {} + if (needsStatus) { + update.status = 'sent' as MessageStatus + } + if (needsInvokedAt) { + update.invokedAt = effectiveInvokedAt + } + return { ...message, ...update } }) } - const messages = updateList(prev.messages) - const pending = updateList(prev.pending) + // Migrate just-acked pending entries into the visible thread. Without + // this step, an at-bottom=false user that is stuck in pending never + // sees their own message at the invocation slot — it stays in the + // pending bucket until they scroll, even though the floating bar + // already cleared. Identifying the migrated rows by (localId, + // invokedAt = effectiveInvokedAt) ensures we only move rows whose + // ack just arrived, not unrelated pending entries. + const updatedPending = updateList(prev.pending) + const consumedFromPending: DecryptedMessage[] = [] + const remainingPending = updatedPending.filter((message) => { + if ( + message.localId && + idSet.has(message.localId) && + message.invokedAt === effectiveInvokedAt + ) { + consumedFromPending.push(message) + return false + } + return true + }) + // After update, re-merge to re-sort by the position key (`invokedAt ?? createdAt`): + // a queued message that just received `invokedAt` should move to its invocation + // position, not stay at its original send-time slot until the next fetch. + const messages = mergeMessages(updateList(prev.messages), consumedFromPending) + const pending = mergeMessages([], remainingPending) if (!changed) { return prev } diff --git a/web/src/lib/messages.ts b/web/src/lib/messages.ts index aaee5522..e263451b 100644 --- a/web/src/lib/messages.ts +++ b/web/src/lib/messages.ts @@ -1,5 +1,4 @@ -import type { InfiniteData } from '@tanstack/react-query' -import type { DecryptedMessage, MessagesResponse } from '@/types/api' +import type { DecryptedMessage } from '@/types/api' import { randomId } from '@/lib/randomId' export function makeClientSideId(prefix: string): string { @@ -14,21 +13,33 @@ export function isUserMessage(msg: DecryptedMessage): boolean { return false } +/** A user message that is still waiting for the CLI ack (messages-consumed). + * Strict null on `invokedAt` so a pre-V8 hub response that omits the field + * (`undefined`) is treated as already-invoked; only optimistic / V8-loaded + * rows that explicitly carry `invokedAt: null` are queued. `failed` rows are + * not queued either — they're surfaced as send errors, not pending work. */ +export function isQueuedForInvocation(msg: DecryptedMessage): boolean { + return isUserMessage(msg) && msg.invokedAt === null && msg.status !== 'failed' +} + function isOptimisticMessage(msg: DecryptedMessage): boolean { return Boolean(msg.localId && msg.id === msg.localId) } function compareMessages(a: DecryptedMessage, b: DecryptedMessage): number { + const aTime = a.invokedAt ?? a.createdAt + const bTime = b.invokedAt ?? b.createdAt + + if (aTime !== bTime) { + return aTime - bTime + } + const aSeq = typeof a.seq === 'number' ? a.seq : null const bSeq = typeof b.seq === 'number' ? b.seq : null if (aSeq !== null && bSeq !== null && aSeq !== bSeq) { return aSeq - bSeq } - - if (a.createdAt !== b.createdAt) { - return a.createdAt - b.createdAt - } return a.id.localeCompare(b.id) } @@ -58,12 +69,18 @@ export function mergeMessages(existing: DecryptedMessage[], incoming: DecryptedM } // If we received stored messages with a localId, drop any optimistic bubbles with the same localId. - // Preserve client-side status (e.g. 'queued') on the replacing server message. + // Preserve client-side status (e.g. 'queued') and invokedAt on the replacing server message. if (incomingStoredLocalIds.size > 0) { const optimisticStatusByLocalId = new Map() + const optimisticInvokedAtByLocalId = new Map() for (const msg of merged) { - if (msg.localId && isOptimisticMessage(msg) && incomingStoredLocalIds.has(msg.localId) && msg.status) { - optimisticStatusByLocalId.set(msg.localId, msg.status) + if (msg.localId && isOptimisticMessage(msg) && incomingStoredLocalIds.has(msg.localId)) { + if (msg.status) { + optimisticStatusByLocalId.set(msg.localId, msg.status) + } + if (msg.invokedAt !== undefined) { + optimisticInvokedAtByLocalId.set(msg.localId, msg.invokedAt) + } } } merged = merged.filter((msg) => { @@ -72,10 +89,21 @@ export function mergeMessages(existing: DecryptedMessage[], incoming: DecryptedM } return !isOptimisticMessage(msg) }) - if (optimisticStatusByLocalId.size > 0) { + if (optimisticStatusByLocalId.size > 0 || optimisticInvokedAtByLocalId.size > 0) { merged = merged.map((msg) => { - if (msg.localId && optimisticStatusByLocalId.has(msg.localId) && !msg.status) { - return { ...msg, status: optimisticStatusByLocalId.get(msg.localId) } + if (!msg.localId) return msg + const update: Partial = {} + if (optimisticStatusByLocalId.has(msg.localId) && !msg.status) { + update.status = optimisticStatusByLocalId.get(msg.localId) + } + if (optimisticInvokedAtByLocalId.has(msg.localId) && msg.invokedAt == null) { + const optimisticInvokedAt = optimisticInvokedAtByLocalId.get(msg.localId) + if (optimisticInvokedAt != null) { + update.invokedAt = optimisticInvokedAt + } + } + if (Object.keys(update).length > 0) { + return { ...msg, ...update } } return msg }) @@ -90,9 +118,14 @@ export function mergeMessages(existing: DecryptedMessage[], incoming: DecryptedM for (const optimistic of optimisticMessages) { if (optimistic.status === 'sent') { + // Compare by the position key (invokedAt ?? createdAt). A late ack can + // attach `invokedAt` long after `createdAt`, so the optimistic copy and + // the server echo end up at the same byPosition slot — using + // `createdAt` alone misses that match and renders both as duplicates. + const optimisticTime = optimistic.invokedAt ?? optimistic.createdAt const hasServerUserMessage = nonOptimisticMessages.some((m) => isUserMessage(m) && - Math.abs(m.createdAt - optimistic.createdAt) < 10_000 + Math.abs((m.invokedAt ?? m.createdAt) - optimisticTime) < 10_000 ) if (hasServerUserMessage) { continue @@ -104,39 +137,3 @@ export function mergeMessages(existing: DecryptedMessage[], incoming: DecryptedM result.sort(compareMessages) return result } - -export function upsertMessagesInCache( - data: InfiniteData | undefined, - incoming: DecryptedMessage[], -): InfiniteData { - const mergedIncoming = mergeMessages([], incoming) - - if (!data || data.pages.length === 0) { - return { - pages: [ - { - messages: mergedIncoming, - page: { - limit: 50, - beforeSeq: null, - nextBeforeSeq: null, - hasMore: false, - }, - }, - ], - pageParams: [null], - } - } - - const pages = data.pages.slice() - const first = pages[0] - pages[0] = { - ...first, - messages: mergeMessages(first.messages, mergedIncoming), - } - - return { - ...data, - pages, - } -} diff --git a/web/src/types/api.ts b/web/src/types/api.ts index 0dd55158..71f34792 100644 --- a/web/src/types/api.ts +++ b/web/src/types/api.ts @@ -40,6 +40,7 @@ export type MessageStatus = 'queued' | 'sending' | 'sent' | 'failed' export type DecryptedMessage = ProtocolDecryptedMessage & { status?: MessageStatus originalText?: string + invokedAt?: number | null } export type RunnerState = { @@ -87,8 +88,9 @@ export type MessagesResponse = { messages: DecryptedMessage[] page: { limit: number - beforeSeq: number | null + beforeSeq?: number | null nextBeforeSeq: number | null + nextBeforeAt?: number | null hasMore: boolean } }