diff --git a/src/main/runtime/orca-runtime.test.ts b/src/main/runtime/orca-runtime.test.ts index 62e4e3551..053340b89 100644 --- a/src/main/runtime/orca-runtime.test.ts +++ b/src/main/runtime/orca-runtime.test.ts @@ -1478,6 +1478,49 @@ describe('OrcaRuntimeService', () => { } }) + it('does not replay an already-delivered message on a later idle transition', async () => { + vi.useFakeTimers() + try { + const runtime = new OrcaRuntimeService(store) + const db = new OrchestrationDb(':memory:') + const write = vi.fn().mockReturnValue(true) + runtime.setOrchestrationDb(db) + runtime.setPtyController({ + write, + kill: vi.fn(), + getForegroundProcess: async () => null + }) + syncSinglePty(runtime) + + const [terminal] = (await runtime.listTerminals()).terminals + runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100) + runtime.onPtyData('pty-1', '\x1b]0;Codex done\x07', 101) + db.insertMessage({ from: 'term_sender', to: terminal.handle, subject: 'hello' }) + + runtime.deliverPendingMessagesForHandle(terminal.handle) + await vi.advanceTimersByTimeAsync(500) + + const firstInjections = write.mock.calls.filter( + (c) => typeof c[1] === 'string' && c[1].includes('Subject: hello') + ).length + expect(firstInjections).toBe(1) + + // Second idle transition: the row is still unread (no check caller has + // consumed it), but it has been delivered. Push-on-idle must skip it to + // avoid the replay bug. + runtime.deliverPendingMessagesForHandle(terminal.handle) + await vi.advanceTimersByTimeAsync(500) + + const totalInjections = write.mock.calls.filter( + (c) => typeof c[1] === 'string' && c[1].includes('Subject: hello') + ).length + expect(totalInjections).toBe(1) + db.close() + } finally { + vi.useRealTimers() + } + }) + it('adopts preallocated ORCA_TERMINAL_HANDLE as a valid runtime handle', async () => { const runtime = new OrcaRuntimeService(store) const handle = runtime.preAllocateHandleForPty('pty-1') diff --git a/src/main/runtime/orca-runtime.ts b/src/main/runtime/orca-runtime.ts index 44ba0c649..3c57895d4 100644 --- a/src/main/runtime/orca-runtime.ts +++ b/src/main/runtime/orca-runtime.ts @@ -8059,7 +8059,7 @@ export class OrcaRuntimeService { return } - const unread = this._orchestrationDb.getUnreadMessages(handle) + const unread = this._orchestrationDb.getUndeliveredUnreadMessages(handle) if (unread.length === 0) { return } diff --git a/src/main/runtime/orchestration/db.test.ts b/src/main/runtime/orchestration/db.test.ts index 2f6b7f089..4607afb4b 100644 --- a/src/main/runtime/orchestration/db.test.ts +++ b/src/main/runtime/orchestration/db.test.ts @@ -61,6 +61,46 @@ describe('OrchestrationDb', () => { expect(filtered[0].type).toBe('worker_done') }) + it('excludes already-delivered rows from getUndeliveredUnreadMessages', () => { + const d = createDb() + const m1 = d.insertMessage({ from: 'a', to: 'b', subject: 'one' }) + const m2 = d.insertMessage({ from: 'a', to: 'b', subject: 'two' }) + + d.markAsDelivered([m1.id]) + + // Push delivery query: only undelivered, unread. + const pending = d.getUndeliveredUnreadMessages('b') + expect(pending).toHaveLength(1) + expect(pending[0].id).toBe(m2.id) + + // Explicit `check` still sees both (they are still unread). + const unread = d.getUnreadMessages('b') + expect(unread).toHaveLength(2) + }) + + it('creates the undelivered inbox index used by push delivery', () => { + const d = createDb() + const sqlite = (d as unknown as { db: Database.Database }).db + + const indexes = sqlite + .prepare( + `SELECT name FROM sqlite_master WHERE type = 'index' AND tbl_name = 'messages' AND name = 'idx_messages_undelivered_inbox'` + ) + .all() + + expect(indexes).toHaveLength(1) + }) + + it('filters getUndeliveredUnreadMessages by type', () => { + const d = createDb() + d.insertMessage({ from: 'a', to: 'b', subject: 's', type: 'status' }) + const wd = d.insertMessage({ from: 'a', to: 'b', subject: 'd', type: 'worker_done' }) + + const filtered = d.getUndeliveredUnreadMessages('b', ['worker_done']) + expect(filtered).toHaveLength(1) + expect(filtered[0].id).toBe(wd.id) + }) + it('marks messages as read', () => { const d = createDb() const m1 = d.insertMessage({ from: 'a', to: 'b', subject: 'one' }) @@ -740,6 +780,7 @@ describe('OrchestrationDb', () => { const names = new Set(indexes.map((r) => r.name)) expect(names.has('idx_messages_id')).toBe(true) expect(names.has('idx_inbox')).toBe(true) + expect(names.has('idx_messages_undelivered_inbox')).toBe(true) expect(names.has('idx_thread')).toBe(true) // v1 data preserved diff --git a/src/main/runtime/orchestration/db.ts b/src/main/runtime/orchestration/db.ts index 8bfe15991..570a1aa04 100644 --- a/src/main/runtime/orchestration/db.ts +++ b/src/main/runtime/orchestration/db.ts @@ -142,6 +142,7 @@ export class OrchestrationDb { completed_at TEXT ); `) + this.createUndeliveredInboxIndexIfPossible() } // Why: `CREATE TABLE IF NOT EXISTS` is a no-op against an existing on-disk @@ -170,14 +171,12 @@ export class OrchestrationDb { if (!this.messagesTypeCheckAllowsHeartbeat()) { // Why — index list is not optional. createTables() already attached - // idx_messages_id / idx_inbox / idx_thread to the old messages table; - // DROP TABLE removes those indexes with it. CREATE INDEX IF NOT - // EXISTS in createTables() only runs on the next process startup, - // so skipping explicit recreation here would leave every - // getUnreadMessages / getMessageById call full-scanning for the - // rest of this process's lifetime — a silent O(N) perf regression. - // The three CREATE INDEX statements below mirror createTables() - // verbatim so the two definitions cannot drift. + // idx_messages_id / idx_inbox / idx_messages_undelivered_inbox / + // idx_thread to the old messages table; DROP TABLE removes those + // indexes with it. CREATE INDEX IF NOT EXISTS in createTables() only + // runs on the next process startup, so skipping explicit recreation + // here would leave message lookups full-scanning for the rest of this + // process's lifetime — a silent O(N) perf regression. this.db.exec(` CREATE TABLE messages_new ( id TEXT NOT NULL, @@ -212,6 +211,8 @@ export class OrchestrationDb { CREATE UNIQUE INDEX idx_messages_id ON messages(id); CREATE INDEX idx_inbox ON messages(to_handle, read); + CREATE INDEX idx_messages_undelivered_inbox + ON messages(to_handle, read, delivered_at, sequence); CREATE INDEX idx_thread ON messages(thread_id); `) } @@ -233,6 +234,7 @@ export class OrchestrationDb { this.db.exec(`ALTER TABLE tasks ADD COLUMN created_by_terminal_handle TEXT`) } } + this.createUndeliveredInboxIndexIfPossible() this.db.pragma(`user_version = ${SCHEMA_VERSION}`) this.db.exec('COMMIT') @@ -247,6 +249,16 @@ export class OrchestrationDb { return rows.some((r) => r.name === column) } + private createUndeliveredInboxIndexIfPossible(): void { + if (!this.hasColumn('messages', 'delivered_at')) { + return + } + this.db.exec(` + CREATE INDEX IF NOT EXISTS idx_messages_undelivered_inbox + ON messages(to_handle, read, delivered_at, sequence) + `) + } + // Why: sqlite_master stores the original CREATE TABLE SQL including the // CHECK clause. Inspecting that text is the cheapest reliable way to tell // whether the pre-rebuild schema already knows about 'heartbeat' without @@ -303,6 +315,28 @@ export class OrchestrationDb { .all(toHandle) as MessageRow[] } + // Why: push-on-idle delivery must not replay messages that were already + // injected into the PTY. `read` flips only when a check-caller consumes a + // message, so delivered-but-unread rows would otherwise be re-injected on + // every later idle transition (the replay bug). Filter on + // `delivered_at IS NULL` so each row is auto-pushed at most once; explicit + // `check` still sees them via getUnreadMessages. + getUndeliveredUnreadMessages(toHandle: string, types?: MessageType[]): MessageRow[] { + if (types && types.length > 0) { + const placeholders = types.map(() => '?').join(',') + return this.db + .prepare( + `SELECT * FROM messages WHERE to_handle = ? AND read = 0 AND delivered_at IS NULL AND type IN (${placeholders}) ORDER BY sequence` + ) + .all(toHandle, ...types) as MessageRow[] + } + return this.db + .prepare( + 'SELECT * FROM messages WHERE to_handle = ? AND read = 0 AND delivered_at IS NULL ORDER BY sequence' + ) + .all(toHandle) as MessageRow[] + } + getAllMessages(toHandle: string, limit = 20): MessageRow[] { return this.db .prepare('SELECT * FROM messages WHERE to_handle = ? ORDER BY sequence DESC LIMIT ?')