diff --git a/src/main/runtime/orchestration-mailbox-detached-routing.test.ts b/src/main/runtime/orchestration-mailbox-detached-routing.test.ts index 453af2368e1..654f01acef1 100644 --- a/src/main/runtime/orchestration-mailbox-detached-routing.test.ts +++ b/src/main/runtime/orchestration-mailbox-detached-routing.test.ts @@ -450,4 +450,106 @@ describe('orchestration detached mailbox routing', () => { expect(db.getMessageById(message.id)?.to_handle).toBe(TERMINAL_HANDLE) db.close() }) + + it('does not route active Dispatch mail through a paged Run check', () => { + const db = createDatabase('orca-mailbox-paged-coordinator-dispatch-overlap-') + const run = db.createRun({ + objective: 'Paged coordinator and Dispatch overlap', + coordinatorHandle: TERMINAL_HANDLE, + coordinatorPaneKey: PANE_KEY + }) + const task = db.createTask({ spec: 'Same-handle worker', runId: run.id }) + const dispatch = db.createDispatchContext(task.id, TERMINAL_HANDLE, PANE_KEY) + const message = db.insertMessage({ + from: 'term_sender', + to: TERMINAL_HANDLE, + subject: 'Dispatch-owned paged mail', + type: 'dispatch', + runId: run.id, + deliveryContract: 'current_delivery' + }) + + expect(db.routeUnreadDirectMessagesToRunMailbox(run.id, TERMINAL_HANDLE)).toMatchObject({ + routedCount: 0, + hasMore: false + }) + expect(db.getMessageById(message.id)?.to_handle).toBe(TERMINAL_HANDLE) + expect( + db.routeUnreadDirectMessagesToDispatchMailbox(dispatch.id, run.id, TERMINAL_HANDLE) + ).toMatchObject({ routedCount: 1, hasMore: false }) + expect(db.getMessageById(message.id)?.to_handle).toBe(`dispatch:${dispatch.id}`) + db.close() + }) + + it('uses primary-key lookups for bounded routing updates', () => { + const db = createDatabase('orca-mailbox-bounded-routing-indexes-') + const sqlite = sqliteFor(db) + const run = db.createRun({ + objective: 'Bounded routing indexes', + coordinatorHandle: TERMINAL_HANDLE, + coordinatorPaneKey: PANE_KEY + }) + const directMessage = db.insertMessage({ + from: 'term_sender', + to: TERMINAL_HANDLE, + subject: 'Direct Run mail', + type: 'status', + runId: run.id, + deliveryContract: 'current_delivery' + }) + sqlite + .prepare('UPDATE messages SET to_handle = ? WHERE id = ?') + .run(TERMINAL_HANDLE, directMessage.id) + const task = db.createTask({ spec: 'Dispatch migration', runId: run.id }) + const dispatch = db.createDispatchContext(task.id, 'term_dispatch', PANE_KEY) + db.insertMessage({ + from: 'term_sender', + to: `dispatch:${dispatch.id}`, + subject: 'Dispatch mailbox mail', + type: 'status', + runId: run.id, + deliveryContract: 'current_delivery' + }) + const foreignHandle = 'term_foreign_bounded_routing' + const foreignRun = db.createRun({ + objective: 'Foreign routing index', + coordinatorHandle: foreignHandle, + coordinatorPaneKey: + '55555555-5555-4555-8555-555555555555:66666666-6666-4666-8666-666666666666' + }) + const foreignMessage = db.insertMessage({ + from: 'term_sender', + to: foreignHandle, + subject: 'Foreign direct mail', + type: 'status', + runId: foreignRun.id, + deliveryContract: 'current_delivery' + }) + sqlite + .prepare('UPDATE messages SET to_handle = ? WHERE id = ?') + .run(foreignHandle, foreignMessage.id) + const prepare = vi.spyOn(sqlite, 'prepare') + + db.routeUnreadDirectMessagesToRunMailbox(run.id, TERMINAL_HANDLE) + db.routeUnreadDispatchMailboxToRunMailbox(dispatch.id, run.id) + db.routeForeignDirectMessagesToOwnedMailboxes(foreignHandle) + + const updateSql = [ + ...new Set( + prepare.mock.calls + .map(([sql]) => sql) + .filter((sql) => sql.includes('UPDATE messages') && sql.includes('id IN')) + ) + ] + expect(updateSql).toHaveLength(3) + for (const sql of updateSql) { + expect(sql).toContain('INDEXED BY idx_messages_id') + const parameters = Array.from({ length: sql.split('?').length - 1 }, () => 'probe') + const plan = sqlite.prepare(`EXPLAIN QUERY PLAN ${sql}`).all(...parameters) as { + detail: string + }[] + expect(plan.map((row) => row.detail).join(' ')).toContain('idx_messages_id') + } + db.close() + }) }) diff --git a/src/main/runtime/orchestration/db.ts b/src/main/runtime/orchestration/db.ts index 52b94562ae3..c619513d6cc 100644 --- a/src/main/runtime/orchestration/db.ts +++ b/src/main/runtime/orchestration/db.ts @@ -3765,11 +3765,20 @@ export class OrchestrationDb { mailboxHandle: string, runId: string, directHandle: string, - throughSequence?: number + throughSequence?: number, + preserveActiveDispatchOwnership = false ): MailboxRoutingPage { this.db.exec('BEGIN IMMEDIATE') try { const throughClause = throughSequence === undefined ? '' : ' AND sequence <= ?' + const dispatchOwnershipClause = preserveActiveDispatchOwnership + ? ` AND NOT EXISTS ( + SELECT 1 FROM dispatch_contexts + WHERE dispatch_contexts.run_id = messages.run_id + AND dispatch_contexts.assignee_handle = messages.to_handle + AND dispatch_contexts.status IN ('pending', 'dispatched') + )` + : '' const params: (string | number)[] = [runId, directHandle] if (throughSequence !== undefined) { params.push(throughSequence) @@ -3779,7 +3788,7 @@ export class OrchestrationDb { .prepare( `SELECT id, type FROM messages WHERE run_id = ? AND to_handle = ? AND read = 0 - AND delivery_contract = 'current_delivery'${throughClause} + AND delivery_contract = 'current_delivery'${throughClause}${dispatchOwnershipClause} ORDER BY sequence LIMIT ?` ) .all(...params) as { @@ -3794,7 +3803,7 @@ export class OrchestrationDb { const placeholders = page.map(() => '?').join(',') const result = this.db .prepare( - `UPDATE messages SET to_handle = ? + `UPDATE messages INDEXED BY idx_messages_id SET to_handle = ? WHERE run_id = ? AND to_handle = ? AND read = 0 AND delivery_contract = 'current_delivery' AND id IN (${placeholders})` ) @@ -3832,7 +3841,7 @@ export class OrchestrationDb { directHandle: string, throughSequence?: number ): MailboxRoutingPage { - return this.routeDirectMessagePage(`run:${runId}`, runId, directHandle, throughSequence) + return this.routeDirectMessagePage(`run:${runId}`, runId, directHandle, throughSequence, true) } routeUnreadDirectMessagesToDispatchMailbox( @@ -3882,7 +3891,7 @@ export class OrchestrationDb { const placeholders = page.map(() => '?').join(',') const result = this.db .prepare( - `UPDATE messages SET to_handle = ? + `UPDATE messages INDEXED BY idx_messages_id SET to_handle = ? WHERE run_id = ? AND to_handle = ? AND read = 0 AND delivery_contract = 'current_delivery' AND id IN (${placeholders})` @@ -4047,7 +4056,7 @@ export class OrchestrationDb { const placeholders = ids.map(() => '?').join(',') this.db .prepare( - `UPDATE messages SET to_handle = ? + `UPDATE messages INDEXED BY idx_messages_id SET to_handle = ? WHERE to_handle = ? AND read = 0 AND delivered_at IS NULL AND delivery_contract = 'current_delivery' AND id IN (${placeholders})` ) @@ -4081,23 +4090,6 @@ export class OrchestrationDb { ).map((row) => row.type) } - areUndeliveredUnreadMessages(toHandle: string, ids: string[]): boolean { - let matched = 0 - for (let offset = 0; offset < ids.length; offset += MESSAGE_ID_UPDATE_BATCH_SIZE) { - const batch = ids.slice(offset, offset + MESSAGE_ID_UPDATE_BATCH_SIZE) - const placeholders = batch.map(() => '?').join(',') - const row = this.db - .prepare( - `SELECT COUNT(*) AS count FROM messages INDEXED BY idx_messages_id - WHERE to_handle = ? AND read = 0 AND delivered_at IS NULL - AND delivery_contract = 'current_delivery' AND id IN (${placeholders})` - ) - .get(toHandle, ...batch) as { count: number } - matched += row.count - } - return matched === ids.length - } - areUnreadMessages(toHandle: string, ids: string[]): boolean { let matched = 0 for (let offset = 0; offset < ids.length; offset += MESSAGE_ID_UPDATE_BATCH_SIZE) {