starred/orca
1
0
Fork 0
mirror of https://github.com/stablyai/orca.git synced 2026-10-22 08:03:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(orchestration): preserve paged mailbox ownership

Browse Source
This commit is contained in:
Jinwoo-H
2026-08-14 20:30:46 -04:00
parent b002e00d4b
commit 57152c2c08
2 changed files with 117 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/runtime/orchestration-mailbox-detached-routing.test.ts
+102
View File
@@ -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()
})
})
src/main/runtime/orchestration/db.ts
+15 -23
View File
@@ -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) {