diff --git a/hub/src/store/index.ts b/hub/src/store/index.ts index 5c12c5eb..7e1e6cf0 100644 --- a/hub/src/store/index.ts +++ b/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. diff --git a/hub/src/store/migration-v13.test.ts b/hub/src/store/migration-v13.test.ts index ec0ff42c..7c355f6e 100644 --- a/hub/src/store/migration-v13.test.ts +++ b/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() diff --git a/hub/src/store/migration-v15.test.ts b/hub/src/store/migration-v15.test.ts index 57113fe4..f3bdf273 100644 --- a/hub/src/store/migration-v15.test.ts +++ b/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() diff --git a/hub/src/store/migration-v18.test.ts b/hub/src/store/migration-v18.test.ts index 3e6b4912..2c31127b 100644 --- a/hub/src/store/migration-v18.test.ts +++ b/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() diff --git a/hub/src/store/migration-v19.test.ts b/hub/src/store/migration-v19.test.ts index ba3d802d..888b7455 100644 --- a/hub/src/store/migration-v19.test.ts +++ b/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 { diff --git a/hub/src/store/migration-v20.test.ts b/hub/src/store/migration-v20.test.ts index 2af4f259..fba1a455 100644 --- a/hub/src/store/migration-v20.test.ts +++ b/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) diff --git a/hub/src/store/migration-v23.test.ts b/hub/src/store/migration-v23.test.ts index cd2733b3..191fa1cd 100644 --- a/hub/src/store/migration-v23.test.ts +++ b/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() }) }) diff --git a/hub/src/store/migration-v24.test.ts b/hub/src/store/migration-v24.test.ts index 7dd95179..4e727880 100644 --- a/hub/src/store/migration-v24.test.ts +++ b/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() }) }) diff --git a/hub/src/store/migration-v25.test.ts b/hub/src/store/migration-v25.test.ts new file mode 100644 index 00000000..c72bac7e --- /dev/null +++ b/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() + }) +}) diff --git a/hub/src/sync/messageService.test.ts b/hub/src/sync/messageService.test.ts index 6841da44..2ef3febe 100644 --- a/hub/src/sync/messageService.test.ts +++ b/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( diff --git a/hub/src/sync/messageService.ts b/hub/src/sync/messageService.ts index c1e310e8..01be0990 100644 --- a/hub/src/sync/messageService.ts +++ b/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,