perf(hub): stop heartbeat replay scanning session history (#1745)

hub/src/store/index.ts

@@ -42,7 +42,7 @@ export {
     WorkGraphValidationError
 } from './workGraph'
 
-const SCHEMA_VERSION: number = 25
+const SCHEMA_VERSION: number = 26
 const REQUIRED_TABLES = [
     'sessions',
     'machines',
@@ -347,6 +347,7 @@ export class Store {
             22: () => this.migrateFromV22ToV23(),
             23: () => this.migrateFromV23ToV24(),
             24: () => this.migrateFromV24ToV25(),
+            25: () => this.migrateFromV25ToV26(),
         })
 
         if (currentVersion === 0) {
@@ -456,6 +457,12 @@ export class Store {
             CREATE INDEX IF NOT EXISTS idx_messages_scheduled_pending
                 ON messages(scheduled_at)
                 WHERE scheduled_at IS NOT NULL AND invoked_at IS NULL;
+            CREATE INDEX IF NOT EXISTS idx_messages_immediate_queued
+                ON messages(session_id, seq)
+                WHERE invoked_at IS NULL
+                  AND local_id IS NOT NULL
+                  AND scheduled_at IS NULL
+                  AND delivery_state = 'queued';
 
             CREATE TABLE IF NOT EXISTS message_epochs (
                 session_id TEXT PRIMARY KEY,
@@ -977,6 +984,18 @@ export class Store {
         }
     }
 
+    /** v25→v26: make empty immediate-queue heartbeat replay an indexed lookup. */
+    private migrateFromV25ToV26(): void {
+        this.db.exec(`
+            CREATE INDEX IF NOT EXISTS idx_messages_immediate_queued
+                ON messages(session_id, seq)
+                WHERE invoked_at IS NULL
+                  AND local_id IS NOT NULL
+                  AND scheduled_at IS NULL
+                  AND delivery_state = 'queued';
+        `)
+    }
+
     /**
      * A2A Layer 1 / P1 (#1374) + P3 substrate: hub work-graph ledger tables.
      * Namespace + principal_json required on every events row.

hub/src/store/migration-v13.test.ts

@@ -10,7 +10,7 @@ describe('Store V12/V13→V14 schema reconciliation', () => {
         const store = new Store(':memory:')
         expect(tableExists(store, 'message_epochs')).toBe(true)
         expect(tableExists(store, 'session_scratchlist')).toBe(true)
-        expect(getUserVersion(store)).toBe(25)
+        expect(getUserVersion(store)).toBe(26)
         store.close()
     })
 
@@ -34,7 +34,7 @@ describe('Store V12/V13→V14 schema reconciliation', () => {
             store = new Store(dbPath)
             expect(tableExists(store, 'message_epochs')).toBe(true)
             expect(tableExists(store, 'session_scratchlist')).toBe(true)
-            expect(getUserVersion(store)).toBe(25)
+            expect(getUserVersion(store)).toBe(26)
             expect(store.messages.getMessageEpoch('session-1')).toBe(0)
             expect(store.messages.getMessages('session-1')).toHaveLength(1)
         } finally {
@@ -72,7 +72,7 @@ describe('Store V12/V13→V14 schema reconciliation', () => {
             store = new Store(dbPath)
             expect(tableExists(store, 'message_epochs')).toBe(true)
             expect(tableExists(store, 'session_scratchlist')).toBe(true)
-            expect(getUserVersion(store)).toBe(25)
+            expect(getUserVersion(store)).toBe(26)
             expect(store.messages.getMessages('session-1')).toHaveLength(1)
         } finally {
             store?.close()

hub/src/store/migration-v15.test.ts

@@ -19,7 +19,7 @@ describe('Store V14→V15 migration: scratchlist attachments column', () => {
         expect(cols).toContain('attachments')
         expect(getColumns(store, 'usage_events')).toContain('last_input_tokens')
         expect(getColumns(store, 'usage_scan_state')).toContain('last_seq')
-        expect(getUserVersion(store)).toBe(25)
+        expect(getUserVersion(store)).toBe(26)
         store.close()
     })
 
@@ -40,7 +40,7 @@ describe('Store V14→V15 migration: scratchlist attachments column', () => {
             expect(cols).toContain('attachments')
             expect(getColumns(store, 'usage_events')).toContain('last_input_tokens')
             expect(getColumns(store, 'usage_scan_state')).toContain('last_seq')
-            expect(getUserVersion(store)).toBe(25)
+            expect(getUserVersion(store)).toBe(26)
         } finally {
             store?.close()
             rmSync(dir, { recursive: true, force: true })
@@ -60,7 +60,7 @@ describe('Store V14→V15 migration: scratchlist attachments column', () => {
             store2 = new Store(dbPath)
             const cols2 = getColumns(store2, 'session_scratchlist')
             expect(cols2).toEqual(cols1)
-            expect(getUserVersion(store2)).toBe(25)
+            expect(getUserVersion(store2)).toBe(26)
         } finally {
             store2?.close()
             store1?.close()

hub/src/store/migration-v18.test.ts

@@ -38,7 +38,7 @@ describe('Store V18->V19 migration: usage scan state', () => {
             const usageRows = internalDb.prepare('SELECT COUNT(*) AS count FROM usage_events').get() as { count: number }
 
             expect(table?.name).toBe('usage_scan_state')
-            expect(version.user_version).toBe(25)
+            expect(version.user_version).toBe(26)
             expect(usageRows.count).toBe(0)
         } finally {
             store?.close()

hub/src/store/migration-v19.test.ts

@@ -37,7 +37,7 @@ describe('Store V19->current migration: usage re-index', () => {
             const scanRows = internalDb.prepare('SELECT COUNT(*) AS count FROM usage_scan_state').get() as { count: number }
             const usageRows = internalDb.prepare('SELECT COUNT(*) AS count FROM usage_events').get() as { count: number }
 
-            expect(version.user_version).toBe(25)
+            expect(version.user_version).toBe(26)
             expect(scanRows.count).toBe(0)
             expect(usageRows.count).toBe(0)
         } finally {

hub/src/store/migration-v20.test.ts

@@ -43,7 +43,7 @@ describe('Store V20->V21 migration: usage semantics re-index', () => {
             store = new Store(dbPath)
             const internalDb = (store as unknown as { db: Database }).db
             const count = (table: string): number => (internalDb.prepare(`SELECT COUNT(*) AS count FROM ${table}`).get() as { count: number }).count
-            expect((internalDb.prepare('PRAGMA user_version').get() as { user_version: number }).user_version).toBe(25)
+            expect((internalDb.prepare('PRAGMA user_version').get() as { user_version: number }).user_version).toBe(26)
             expect(count('usage_events')).toBe(0)
             expect(count('usage_scan_state')).toBe(0)
             expect(store.messages.getMessages(session.id)).toHaveLength(1)

hub/src/store/migration-v23.test.ts

@@ -13,7 +13,7 @@ afterEach(() => {
     }
 })
 
-describe('schema migration v22 to v25', () => {
+describe('schema migration v22 to v26', () => {
     it('adds events and event_links tables to a V22 database', () => {
         const dir = mkdtempSync(join(tmpdir(), 'hapi-migration-v23-'))
         tempDirs.push(dir)
@@ -42,7 +42,7 @@ describe('schema migration v22 to v25', () => {
         expect(links?.name).toBe('event_links')
         const columns = internalDb.prepare('PRAGMA table_info(messages)').all() as Array<{ name: string }>
         expect(columns.map((column) => column.name)).toContain('delivery_state')
-        expect(version.user_version).toBe(25)
+        expect(version.user_version).toBe(26)
         migrated.close()
     })
 })

hub/src/store/migration-v24.test.ts

@@ -13,7 +13,7 @@ afterEach(() => {
     }
 })
 
-describe('schema migration v23 to v25', () => {
+describe('schema migration v23 to v26', () => {
     it('adds fcm_devices.push_key to a V23 database and keeps existing rows', () => {
         const dir = mkdtempSync(join(tmpdir(), 'hapi-migration-v24-'))
         tempDirs.push(dir)
@@ -37,7 +37,7 @@ describe('schema migration v23 to v25', () => {
         expect(columns.some((col) => col.name === 'push_key')).toBe(true)
         const messageColumns = internalDb.prepare('PRAGMA table_info(messages)').all() as Array<{ name: string }>
         expect(messageColumns.some((col) => col.name === 'delivery_state')).toBe(true)
-        expect(version.user_version).toBe(25)
+        expect(version.user_version).toBe(26)
 
         // Existing Android rows survive with a NULL push key.
         const devices = migrated.fcm.getDevicesByNamespace('default')
@@ -74,7 +74,7 @@ describe('schema migration v23 to v25', () => {
         const columns = internalDb.prepare('PRAGMA table_info(messages)').all() as Array<{ name: string }>
         const version = internalDb.prepare('PRAGMA user_version').get() as { user_version: number }
         expect(columns.some((col) => col.name === 'delivery_state')).toBe(true)
-        expect(version.user_version).toBe(25)
+        expect(version.user_version).toBe(26)
         migrated.close()
     })
 })

hub/src/store/migration-v25.test.ts

@@ -0,0 +1,48 @@
+import { afterEach, describe, expect, it } from 'bun:test'
+import { Database } from 'bun:sqlite'
+import { mkdtempSync, rmSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { Store } from './index'
+
+const tempDirs: string[] = []
+
+afterEach(() => {
+    for (const dir of tempDirs.splice(0)) {
+        rmSync(dir, { recursive: true, force: true })
+    }
+})
+
+describe('schema migration v25 to v26', () => {
+    it('adds an index used by immediate queued-message replay', () => {
+        const dir = mkdtempSync(join(tmpdir(), 'hapi-migration-v25-'))
+        tempDirs.push(dir)
+        const dbPath = join(dir, 'hapi.db')
+
+        new Store(dbPath).close()
+        const legacy = new Database(dbPath)
+        legacy.exec(`
+            DROP INDEX idx_messages_immediate_queued;
+            PRAGMA user_version = 25;
+        `)
+        legacy.close()
+
+        const migrated = new Store(dbPath)
+        const internalDb = (migrated as unknown as { db: Database }).db
+        const version = internalDb.prepare('PRAGMA user_version').get() as { user_version: number }
+        const plan = internalDb.prepare(`
+            EXPLAIN QUERY PLAN
+            SELECT * FROM messages
+            WHERE session_id = ?
+              AND invoked_at IS NULL
+              AND local_id IS NOT NULL
+              AND scheduled_at IS NULL
+              AND delivery_state = 'queued'
+            ORDER BY seq ASC
+        `).all('session-id') as Array<{ detail: string }>
+
+        expect(version.user_version).toBe(26)
+        expect(plan.some((row) => row.detail.includes('idx_messages_immediate_queued'))).toBe(true)
+        migrated.close()
+    })
+})

hub/src/sync/messageService.test.ts

@@ -1192,6 +1192,21 @@ describe('MessageService.sendMessage deliveryMode', () => {
         return { io, cliEmitted }
     }
 
+    it('skips the OpenCode delivery-gate scan when heartbeat replay has no queued messages', () => {
+        const store = makeStore()
+        const session = makeSession(store, 'empty-heartbeat-replay')
+        const service = new MessageService(store, makeTrackingIo().io, makePublisher() as any)
+        const originalGateCheck = store.isOpenCodeClearDeliveryGated.bind(store)
+        let gateChecks = 0
+        store.isOpenCodeClearDeliveryGated = (sessionId: string) => {
+            gateChecks += 1
+            return originalGateCheck(sessionId)
+        }
+
+        expect(service.replayImmediateQueuedMessages(session.id)).toBe(0)
+        expect(gateChecks).toBe(0)
+    })
+
     it('persists Pi steer provenance but downgrades every deferred CLI delivery to queue', async () => {
         const store = makeStore()
         const session = store.sessions.getOrCreateSession(

hub/src/sync/messageService.ts

@@ -982,8 +982,8 @@ export class MessageService {
 
     /** Replay durable immediate prompts whenever their CLI session attaches. */
     replayImmediateQueuedMessages(sessionId: string): number {
-        if (this.store.isOpenCodeClearDeliveryGated(sessionId)) return 0
         const queued = this.store.messages.getImmediateQueuedLocalMessages(sessionId)
+        if (queued.length === 0 || this.store.isOpenCodeClearDeliveryGated(sessionId)) return 0
         for (const msg of queued) {
             const update = {
                 id: msg.id,