fix: prevent orchestration message replay (#2003)

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')

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
     }

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

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 ?')