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

fix(orchestration): make pointer staging restart-safe

Browse Source
This commit is contained in:
Jinwoo-H
2026-08-14 16:35:55 -04:00
committed by Jinwoo-H
parent ddac8cbe7c
commit 9ddab8150c
12 changed files with 1277 additions and 373 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/providers/ssh-pty-provider.ts
+3 -2
View File
@@ -23,6 +23,7 @@ import { SshPtySpawnExitRaceTracker } from './ssh-pty-spawn-exit-race'
import { SshAgentSessionCapabilities } from './ssh-agent-session-capabilities'
import type { PtyProcessInspection } from './pty-process-inspection'
import { SSH_SESSION_EXPIRED_ERROR } from './ssh-pty-errors'
import { writeToSshPty } from './ssh-pty-write'
// Why: sequential relay teardown calls share one absolute budget; convert to the mux-relative timeout only at dispatch.
function relayTimeoutOptions(deadlineMs: number | undefined): { timeoutMs: number } | undefined {
@@ -204,8 +205,8 @@ export class SshPtyProvider implements IPtyProvider {
})
}
write(id: string, data: string): void {
this.mux.notify('pty.data', { id: this.toRelayPtyId(id), data })
write(id: string, data: string): boolean {
return writeToSshPty(this.mux, this.toRelayPtyId(id), data)
}
resize(id: string, cols: number, rows: number): void {
src/main/providers/ssh-pty-write.test.ts
+16
View File
@@ -0,0 +1,16 @@
import { describe, expect, it, vi } from 'vitest'
import { SshPtyProvider } from './ssh-pty-provider'
describe('SSH PTY writes', () => {
it('rejects writes synchronously after the transport is disposed', () => {
const mux = {
isDisposed: vi.fn().mockReturnValue(true),
notify: vi.fn(),
onNotification: vi.fn()
}
const provider = new SshPtyProvider('conn-1', mux as never)
expect(provider.write('ssh:conn-1@@pty-1', 'pointer')).toBe(false)
expect(mux.notify).not.toHaveBeenCalled()
})
})
src/main/providers/ssh-pty-write.ts
+13
View File
@@ -0,0 +1,13 @@
import type { SshChannelMultiplexer } from '../ssh/ssh-channel-multiplexer'
export function writeToSshPty(
mux: SshChannelMultiplexer,
relayPtyId: string,
data: string
): boolean {
if (mux.isDisposed()) {
return false
}
mux.notify('pty.data', { id: relayPtyId, data })
return !mux.isDisposed()
}
src/main/runtime/orca-runtime.test.ts
+84 -10
View File
@@ -1154,6 +1154,42 @@ class InMemoryOrchestrationMessages {
return [...this.runs.values()].find((run) => run.coordinator_pane_key === paneKey)
}
hasUndeliveredDirectMessageForRun(runId: string, directHandle: string): boolean {
return this.messages.some(
(message) =>
message.run_id === runId &&
message.to_handle === directHandle &&
message.read === 0 &&
!message.delivered_at
)
}
routeUnreadDirectMessagesToRunMailbox(
runId: string,
directHandle: string
): { routedCount: number; hasMore: boolean; types: MessageType[] } {
const routed = this.messages.filter(
(message) =>
message.run_id === runId && message.to_handle === directHandle && message.read === 0
)
for (const message of routed) {
message.to_handle = `run:${runId}`
}
return {
routedCount: routed.length,
hasMore: false,
types: [...new Set(routed.map((message) => message.type))]
}
}
areUnreadMessages(toHandle: string, ids: string[]): boolean {
return ids.every((id) =>
this.messages.some(
(message) => message.id === id && message.to_handle === toHandle && message.read === 0
)
)
}
markAsDelivered(ids: string[]): void {
const deliveredIds = new Set(ids)
for (const message of this.messages) {
@@ -1163,6 +1199,15 @@ class InMemoryOrchestrationMessages {
}
}
markAsUndelivered(ids: string[]): void {
const releasedIds = new Set(ids)
for (const message of this.messages) {
if (releasedIds.has(message.id) && message.read === 0) {
message.delivered_at = null
}
}
}
close(): void {}
}
@@ -1173,6 +1218,15 @@ function setInMemoryOrchestrationMessages(
runtime.setOrchestrationDb(db as unknown as OrchestrationDb)
}
function bindSinglePtyRun(db: InMemoryOrchestrationMessages, terminalHandle: string): string {
db.setRun({
id: 'run_test',
coordinator_handle: terminalHandle,
coordinator_pane_key: 'tab-1:pane:1'
})
return 'run:run_test'
}
function expectStablePaneKeyEnv(env: Record<string, string>): string {
expect(env.ORCA_TAB_ID).toMatch(UUID_RE)
const leafId = env.ORCA_PANE_KEY?.slice(`${env.ORCA_TAB_ID}:`.length)
@@ -19572,6 +19626,7 @@ describe('OrcaRuntimeService', () => {
syncSinglePty(runtime)
const [terminal] = (await runtime.listTerminals()).terminals
const mailbox = bindSinglePtyRun(db, terminal.handle)
runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100)
runtime.onPtyData('pty-1', '\x1b]0;Codex done\x07', 101)
db.setActiveCoordinatorRun({ coordinator_handle: 'term_other' })
@@ -19586,7 +19641,7 @@ describe('OrcaRuntimeService', () => {
await vi.advanceTimersByTimeAsync(500)
expect(write).toHaveBeenCalledWith('pty-1', '\r')
const unread = db.getUnreadMessages(terminal.handle)
const unread = db.getUnreadMessages(mailbox)
expect(unread).toHaveLength(1)
expect(unread[0].read).toBe(0)
expect(unread[0].delivered_at).toEqual(expect.any(String))
@@ -19611,6 +19666,7 @@ describe('OrcaRuntimeService', () => {
syncSinglePty(runtime)
const [terminal] = (await runtime.listTerminals()).terminals
const mailbox = bindSinglePtyRun(db, terminal.handle)
runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100)
runtime.onPtyData('pty-1', '\x1b]0;Codex done\x07', 101)
db.setActiveCoordinatorRun({ coordinator_handle: terminal.handle })
@@ -19632,7 +19688,7 @@ describe('OrcaRuntimeService', () => {
)
expect(submitWrites).toHaveLength(1)
const unread = db.getUnreadMessages(terminal.handle)
const unread = db.getUnreadMessages(mailbox)
expect(unread).toHaveLength(1)
expect(unread[0].read).toBe(0)
expect(unread[0].delivered_at).toEqual(expect.any(String))
@@ -19657,6 +19713,7 @@ describe('OrcaRuntimeService', () => {
syncSinglePty(runtime)
const [terminal] = (await runtime.listTerminals()).terminals
const mailbox = bindSinglePtyRun(db, terminal.handle)
runtime.onPtyData('pty-1', '\x1b]0;\u280b Cursor Agent\x07', 100)
runtime.onPtyData('pty-1', '\x1b]0;Cursor ready\x07', 101)
db.insertMessage({ from: 'term_sender', to: terminal.handle, subject: 'hello cursor' })
@@ -19673,7 +19730,7 @@ describe('OrcaRuntimeService', () => {
)
expect(submitWrites).toHaveLength(0)
const unread = db.getUnreadMessages(terminal.handle)
const unread = db.getUnreadMessages(mailbox)
expect(unread).toHaveLength(1)
expect(unread[0].read).toBe(0)
expect(unread[0].delivered_at).toEqual(expect.any(String))
@@ -19698,6 +19755,7 @@ describe('OrcaRuntimeService', () => {
syncSinglePty(runtime, 'pty-1', { tabTitle: 'cursor-repro-branch' })
const [terminal] = (await runtime.listTerminals()).terminals
bindSinglePtyRun(db, terminal.handle)
runtime.onPtyData('pty-1', '\x1b]0;. Investigate Cursor Agent\x07', 100)
runtime.onPtyData('pty-1', '\x1b]0;* Investigate Cursor Agent\x07', 101)
db.insertMessage({ from: 'term_sender', to: terminal.handle, subject: 'hello claude' })
@@ -19731,6 +19789,7 @@ describe('OrcaRuntimeService', () => {
syncSinglePty(runtime)
const [terminal] = (await runtime.listTerminals()).terminals
bindSinglePtyRun(db, terminal.handle)
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' })
@@ -34577,6 +34636,7 @@ describe('OrcaRuntimeService', () => {
syncSinglePty(runtime)
const [terminal] = (await runtime.listTerminals()).terminals
bindSinglePtyRun(db, terminal.handle)
runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100)
runtime.onPtyData('pty-1', '\x1b]0;Codex done\x07', 101)
await runtime.waitForTerminal(terminal.handle, { condition: 'tui-idle' })
@@ -35009,6 +35069,7 @@ describe('OrcaRuntimeService', () => {
syncSinglePty(runtime)
const [terminal] = (await runtime.listTerminals()).terminals
bindSinglePtyRun(db, terminal.handle)
runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100)
const message = db.insertMessage({
from: 'sender',
@@ -35042,6 +35103,7 @@ describe('OrcaRuntimeService', () => {
syncSinglePty(runtime)
const [terminal] = (await runtime.listTerminals()).terminals
bindSinglePtyRun(db, terminal.handle)
runtime.seedTerminalRestoreTail('pty-1', { lastTitle: 'Codex done' })
const message = db.insertMessage({
from: 'sender',
@@ -35084,6 +35146,7 @@ describe('OrcaRuntimeService', () => {
syncSinglePty(runtime)
const [terminal] = (await runtime.listTerminals()).terminals
bindSinglePtyRun(db, terminal.handle)
// Why: the persisted title is historical — the agent may have gone busy
// across the relaunch, so a seeded 'idle' must not authorize a PTY write.
runtime.seedTerminalRestoreTail('pty-1', { lastTitle: 'Codex done' })
@@ -35131,6 +35194,7 @@ describe('OrcaRuntimeService', () => {
syncSinglePty(runtime)
const [terminal] = (await runtime.listTerminals()).terminals
const mailbox = bindSinglePtyRun(db, terminal.handle)
runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100)
runtime.onPtyData('pty-1', '\x1b]0;Codex done\x07', 101)
await runtime.waitForTerminal(terminal.handle, { condition: 'tui-idle' })
@@ -35141,8 +35205,8 @@ describe('OrcaRuntimeService', () => {
// shared in-flight promise put a no-waiter notify inside that window, so the
// push must not inject rows the resolved check is about to return.
const consumed = runtime
.waitForMessage(terminal.handle, { timeoutMs: 5_000 })
.then(() => db.getUnreadMessages(terminal.handle).map((row) => (row.read = 1)))
.waitForMessage(mailbox, { timeoutMs: 5_000 })
.then(() => db.getUnreadMessages(mailbox).map((row) => (row.read = 1)))
const first = db.insertMessage({ from: 'sender', to: terminal.handle, subject: 'pulled' })
runtime.notifyMessageArrived(terminal.handle, 'status')
const second = db.insertMessage({
@@ -35178,6 +35242,7 @@ describe('OrcaRuntimeService', () => {
syncSinglePty(runtime)
const [terminal] = (await runtime.listTerminals()).terminals
const mailbox = bindSinglePtyRun(db, terminal.handle)
runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100)
runtime.onPtyData('pty-1', '\x1b]0;Codex done\x07', 101)
await runtime.waitForTerminal(terminal.handle, { condition: 'tui-idle' })
@@ -35187,7 +35252,7 @@ describe('OrcaRuntimeService', () => {
// `status` notify is unclaimed and pushes, but the worker_done row landing
// in the same drain belongs to this waiter's check — injecting it too would
// deliver that completion twice (pane + check return).
const waitPromise = runtime.waitForMessage(terminal.handle, {
const waitPromise = runtime.waitForMessage(mailbox, {
typeFilter: ['worker_done'],
timeoutMs: 5_000
})
@@ -35212,7 +35277,7 @@ describe('OrcaRuntimeService', () => {
.map(([, data]) => data)
.filter((data): data is string => typeof data === 'string')
expect(payloads).toContain(
'\nYou have 1 orchestration message. Run `orca orchestration check`.\n'
'\nYou have 1 orchestration message. Run `orca orchestration check --run run_test`.\n'
)
expect(payloads.some((data) => data.includes('reserved completion'))).toBe(false)
expect(status.delivered_at).toEqual(expect.any(String))
@@ -35238,6 +35303,7 @@ describe('OrcaRuntimeService', () => {
syncSinglePty(runtime)
const [terminal] = (await runtime.listTerminals()).terminals
const mailbox = bindSinglePtyRun(db, terminal.handle)
runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100)
runtime.onPtyData('pty-1', '\x1b]0;Codex done\x07', 101)
await runtime.waitForTerminal(terminal.handle, { condition: 'tui-idle' })
@@ -35253,7 +35319,7 @@ describe('OrcaRuntimeService', () => {
// blocks before the deferred push runs still owns this row, so only the
// push-time read of live waiters can keep it out of the pane.
runtime.notifyMessageArrived(terminal.handle, 'status')
const waitPromise = runtime.waitForMessage(terminal.handle, {
const waitPromise = runtime.waitForMessage(mailbox, {
typeFilter: ['status'],
timeoutMs: 5_000
})
@@ -35340,6 +35406,7 @@ describe('OrcaRuntimeService', () => {
setInMemoryOrchestrationMessages(runtime, db)
const [republished] = (await runtime.listTerminals()).terminals
bindSinglePtyRun(db, republished.handle)
const message = db.insertMessage({
from: 'sender',
to: republished.handle,
@@ -35372,6 +35439,7 @@ describe('OrcaRuntimeService', () => {
syncSinglePty(runtime)
const [terminal] = (await runtime.listTerminals()).terminals
bindSinglePtyRun(db, terminal.handle)
runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100)
runtime.onPtyData('pty-1', '\x1b]0;Codex done\x07', 101)
await runtime.waitForTerminal(terminal.handle, { condition: 'tui-idle' })
@@ -35429,6 +35497,7 @@ describe('OrcaRuntimeService', () => {
runtime.onPtyExit('pty-1', 0)
runtime.onPtySpawned('pty-1', undefined, { awaitsRegistration: false })
setInMemoryOrchestrationMessages(runtime, db)
bindSinglePtyRun(db, terminal.handle)
const message = db.insertMessage({
from: 'sender',
to: terminal.handle,
@@ -35473,6 +35542,7 @@ describe('OrcaRuntimeService', () => {
syncSinglePty(runtime)
const [terminal] = (await runtime.listTerminals()).terminals
const mailbox = bindSinglePtyRun(db, terminal.handle)
runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100)
runtime.onPtyData('pty-1', '\x1b]0;Codex done\x07', 101)
await runtime.waitForTerminal(terminal.handle, { condition: 'tui-idle' })
@@ -35480,7 +35550,7 @@ describe('OrcaRuntimeService', () => {
// Why: a `check --wait --types worker_done` waiter never returns a status
// row — check re-reads under the same filter — so it is not the consumer
// and treating it as one would strand the message (#12536).
const waitPromise = runtime.waitForMessage(terminal.handle, {
const waitPromise = runtime.waitForMessage(mailbox, {
typeFilter: ['worker_done'],
timeoutMs: 5_000
})
@@ -35526,6 +35596,7 @@ describe('OrcaRuntimeService', () => {
syncSinglePty(runtime)
const [terminal] = (await runtime.listTerminals()).terminals
const mailbox = bindSinglePtyRun(db, terminal.handle)
runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100)
runtime.onPtyData('pty-1', '\x1b]0;Codex done\x07', 101)
await runtime.waitForTerminal(terminal.handle, { condition: 'tui-idle' })
@@ -35538,7 +35609,7 @@ describe('OrcaRuntimeService', () => {
// Why: blocked orchestration.check --wait is an explicit pull; push must
// not stamp delivered_at or type into the pane (double delivery, #12584).
const waitPromise = runtime.waitForMessage(terminal.handle, { timeoutMs: 5_000 })
const waitPromise = runtime.waitForMessage(mailbox, { timeoutMs: 5_000 })
runtime.notifyMessageArrived(terminal.handle, 'status')
await expect(waitPromise).resolves.toBe('notified')
@@ -35565,6 +35636,7 @@ describe('OrcaRuntimeService', () => {
syncSinglePty(runtime)
const [terminal] = (await runtime.listTerminals()).terminals
bindSinglePtyRun(db, terminal.handle)
runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100)
runtime.onPtyData('pty-1', '\x1b]0;Codex done\x07', 101)
await runtime.waitForTerminal(terminal.handle, { condition: 'tui-idle' })
@@ -35612,6 +35684,7 @@ describe('OrcaRuntimeService', () => {
syncSinglePty(runtime)
const [terminal] = (await runtime.listTerminals()).terminals
bindSinglePtyRun(db, terminal.handle)
runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100)
runtime.onPtyData('pty-1', '\x1b]0;Codex done\x07', 101)
await runtime.waitForTerminal(terminal.handle, { condition: 'tui-idle' })
@@ -35670,6 +35743,7 @@ describe('OrcaRuntimeService', () => {
syncSinglePty(runtime)
const [terminal] = (await runtime.listTerminals()).terminals
bindSinglePtyRun(db, terminal.handle)
runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100)
runtime.onPtyData('pty-1', '\x1b]0;Codex done\x07', 101)
await runtime.waitForTerminal(terminal.handle, { condition: 'tui-idle' })
src/main/runtime/orca-runtime.ts
+74 -82
View File
@@ -2899,13 +2899,10 @@ export class OrcaRuntimeService {
private handles = new Map<string, TerminalHandleRecord>()
private handleByLeafKey = new Map<string, string>()
private handleByPtyId = new Map<string, string>()
// Why: provisional pointer state serializes PTY injection before the durable delivered_at commit.
private readonly lastPointedMessageSequenceByHandle = new Map<string, number>()
// Why: a waiter can reserve an older row while a newer row advances the sequence watermark.
private readonly pointedMessageIdsByHandle = new Map<string, Set<string>>()
private readonly mailPointerRepointScheduler = new MailPointerRepointScheduler((handle) =>
this.repointPendingMessagesForHandle(handle)
)
// Why: in-process reservations serialize delayed Enter after durable pointer staging.
private readonly pointedMessageWatermarkOwnerByHandle = new Map<
string,
{ ptyId: string; sequence: number; leafKey: string; active: boolean }
@@ -2915,7 +2912,7 @@ export class OrcaRuntimeService {
string,
ReadonlySet<string> | null
>()
private readonly pendingForeignDirectReconciliationsByLeafKey = new Set<string>()
private readonly pendingDirectMailboxReconciliationsByLeafKey = new Set<string>()
private syntheticTerminalHandles = new Set<string>()
private detachedPreAllocatedLeaves = new Map<string, RuntimeLeafRecord>()
private graphSyncCallbacks: (() => void)[] = []
@@ -3953,8 +3950,6 @@ export class OrcaRuntimeService {
setOrchestrationDb(db: OrchestrationDb): void {
this.stopOrchestrationFederationRelay()
this.mailPointerRepointScheduler.clear()
this.lastPointedMessageSequenceByHandle.clear()
this.pointedMessageIdsByHandle.clear()
this._orchestrationDb = db
this.ensureOrchestrationFederationRelay()
this.scheduleRestoredMessageRepoints()
@@ -32428,14 +32423,14 @@ export class OrcaRuntimeService {
}
}
private scheduleForeignDirectReconciliation(leaf: RuntimeLeafRecord): void {
private scheduleDirectMailboxReconciliation(leaf: RuntimeLeafRecord): void {
const leafKey = this.getLeafKey(leaf.tabId, leaf.leafId)
if (this.pendingForeignDirectReconciliationsByLeafKey.has(leafKey)) {
if (this.pendingDirectMailboxReconciliationsByLeafKey.has(leafKey)) {
return
}
this.pendingForeignDirectReconciliationsByLeafKey.add(leafKey)
this.pendingDirectMailboxReconciliationsByLeafKey.add(leafKey)
setImmediate(() => {
this.pendingForeignDirectReconciliationsByLeafKey.delete(leafKey)
this.pendingDirectMailboxReconciliationsByLeafKey.delete(leafKey)
const currentLeaf = this.leaves.get(leafKey)
if (!currentLeaf || currentLeaf.ptyId !== leaf.ptyId) {
return
@@ -32443,6 +32438,7 @@ export class OrcaRuntimeService {
this.notifyForwardedOrchestrationMailboxes(
this.routeForeignDirectMessagesForLeaf(currentLeaf)
)
this.resolveActionableMailboxForLeaf(currentLeaf)
})
}
@@ -32476,7 +32472,7 @@ export class OrcaRuntimeService {
paneKey
)
if (routed?.hasMore) {
this.scheduleForeignDirectReconciliation(leaf)
this.scheduleDirectMailboxReconciliation(leaf)
}
return routed?.mailboxes ?? []
}
@@ -32491,6 +32487,25 @@ export class OrcaRuntimeService {
}
}
private wakeRoutedMessageWaiters(mailboxHandle: string, types: readonly string[]): void {
const waiters = [...(this.messageWaitersByHandle.get(mailboxHandle) ?? [])]
if (waiters.length === 0 || types.length === 0) {
return
}
// Why: the arrival hook must snapshot waiter ownership before routed rows wake those waiters.
queueMicrotask(() => {
const liveWaiters = this.messageWaitersByHandle.get(mailboxHandle)
for (const waiter of waiters) {
if (
liveWaiters?.has(waiter) &&
(!waiter.typeFilter || types.some((type) => waiter.typeFilter?.includes(type)))
) {
this.resolveMessageWaiter(waiter, 'notified')
}
}
})
}
private resolveActionableMailboxForLeaf(
leaf: RuntimeLeafRecord,
requestedMailbox?: string,
@@ -32515,7 +32530,10 @@ export class OrcaRuntimeService {
return null
}
const hasDirectRunMail =
this._orchestrationDb.hasUndeliveredDirectMessageForRun?.(run.id, terminalHandle) ?? false
options.routeDirectMail === false
? false
: (this._orchestrationDb.hasUndeliveredDirectMessageForRun?.(run.id, terminalHandle) ??
false)
if (
options.requireRequestedMail &&
requestedMailbox === terminalHandle &&
@@ -32524,7 +32542,14 @@ export class OrcaRuntimeService {
return null
}
if (hasDirectRunMail && options.routeDirectMail !== false) {
this._orchestrationDb.routeUnreadDirectMessagesToRunMailbox?.(run.id, terminalHandle)
const routed = this._orchestrationDb.routeUnreadDirectMessagesToRunMailbox?.(
run.id,
terminalHandle
)
this.wakeRoutedMessageWaiters(runMailbox, routed?.types ?? [])
if (routed?.hasMore) {
this.scheduleDirectMailboxReconciliation(leaf)
}
}
return runMailbox
}
@@ -32540,10 +32565,12 @@ export class OrcaRuntimeService {
return null
}
const hasDirectDispatchMail =
this._orchestrationDb.hasUndeliveredDirectMessageForRun?.(
dispatch.run_id,
terminalHandle
) ?? false
options.routeDirectMail === false
? false
: (this._orchestrationDb.hasUndeliveredDirectMessageForRun?.(
dispatch.run_id,
terminalHandle
) ?? false)
if (
options.requireRequestedMail &&
requestedMailbox === terminalHandle &&
@@ -32552,11 +32579,15 @@ export class OrcaRuntimeService {
return null
}
if (hasDirectDispatchMail && options.routeDirectMail !== false) {
this._orchestrationDb.routeUnreadDirectMessagesToDispatchMailbox?.(
const routed = this._orchestrationDb.routeUnreadDirectMessagesToDispatchMailbox?.(
dispatch.id,
dispatch.run_id,
terminalHandle
)
this.wakeRoutedMessageWaiters(dispatchMailbox, routed?.types ?? [])
if (routed?.hasMore) {
this.scheduleDirectMailboxReconciliation(leaf)
}
}
return dispatchMailbox
}
@@ -33237,7 +33268,7 @@ export class OrcaRuntimeService {
// retire must not clear a newer same-id flight or flush its parked trigger.
private readonly messageDeliveryFlightsByPtyId = new Map<
string,
{ enterTimer: ReturnType<typeof setTimeout> | null }
{ enterTimer: ReturnType<typeof setTimeout> | null; stagedMessageIds: string[] }
>()
private readonly parkedMessageRedeliveriesByPtyId = new Map<
@@ -33247,7 +33278,7 @@ export class OrcaRuntimeService {
private settlePendingMessageDelivery(
ptyId: string,
flight: { enterTimer: ReturnType<typeof setTimeout> | null }
flight: { enterTimer: ReturnType<typeof setTimeout> | null; stagedMessageIds: string[] }
): void {
if (this.messageDeliveryFlightsByPtyId.get(ptyId) !== flight) {
return
@@ -33338,7 +33369,7 @@ export class OrcaRuntimeService {
reservedTypes: ReadonlySet<string> | undefined
): void {
const prior = this.parkedMessageRedeliveryTypesByMailboxHandle.get(mailboxHandle)
if (prior === null || (prior === undefined && reservedTypes === undefined)) {
if (!this.parkedMessageRedeliveryTypesByMailboxHandle.has(mailboxHandle) && !reservedTypes) {
this.parkedMessageRedeliveryTypesByMailboxHandle.set(mailboxHandle, null)
return
}
@@ -33369,19 +33400,18 @@ export class OrcaRuntimeService {
if (flight?.enterTimer != null) {
clearTimeout(flight.enterTimer)
}
if (flight?.stagedMessageIds.length) {
this._orchestrationDb?.markAsUndelivered(flight.stagedMessageIds)
}
this.messageDeliveryFlightsByPtyId.delete(ptyId)
this.parkedMessageRedeliveriesByPtyId.delete(ptyId)
for (const leaf of this.getLeavesForPty(ptyId)) {
const handle = this.handleByLeafKey.get(this.getLeafKey(leaf.tabId, leaf.leafId))
if (handle) {
this.lastPointedMessageSequenceByHandle.delete(handle)
this.pointedMessageIdsByHandle.delete(handle)
this.mailPointerRepointScheduler.schedule(handle)
}
const run = this._orchestrationDb?.getCurrentRunForPane?.(`${leaf.tabId}:${leaf.leafId}`)
if (run) {
this.lastPointedMessageSequenceByHandle.delete(`run:${run.id}`)
this.pointedMessageIdsByHandle.delete(`run:${run.id}`)
this.mailPointerRepointScheduler.schedule(`run:${run.id}`)
}
}
@@ -33416,7 +33446,7 @@ export class OrcaRuntimeService {
if (messages.some((message) => messageTypeHasLiveWaiter(waiters, message.type))) {
return 'release'
}
return (this._orchestrationDb?.areUndeliveredUnreadMessages?.(
return (this._orchestrationDb?.areUnreadMessages?.(
mailboxHandle,
messages.map((message) => message.id)
) ?? true)
@@ -33442,11 +33472,11 @@ export class OrcaRuntimeService {
return
}
const mailboxHandle = options.mailboxHandle ?? handle
if (!mailboxHandle.startsWith('run:')) {
return
}
if (
mailboxHandle.startsWith('run:') &&
this._orchestrationDb.hasOutstandingRunDelivery?.(mailboxHandle.slice('run:'.length))
) {
if (this._orchestrationDb.hasOutstandingRunDelivery?.(mailboxHandle.slice('run:'.length))) {
return
}
@@ -33489,18 +33519,6 @@ export class OrcaRuntimeService {
excludeTypes: [...excludedTypes],
limit: ORCHESTRATION_DELIVERY_BATCH_LIMIT
})
const pendingIds = new Set(pending.map((message) => message.id))
const pointedIds = this.pointedMessageIdsByHandle.get(mailboxHandle)
if (pointedIds) {
for (const id of pointedIds) {
if (!pendingIds.has(id)) {
pointedIds.delete(id)
}
}
if (pointedIds.size === 0) {
this.pointedMessageIdsByHandle.delete(mailboxHandle)
}
}
const unread = pending
.filter(
(message) =>
@@ -33512,16 +33530,6 @@ export class OrcaRuntimeService {
return
}
const watermark = this.lastPointedMessageSequenceByHandle.get(mailboxHandle) ?? -1
const priorPointedIds = this.pointedMessageIdsByHandle.get(mailboxHandle)
if (
!unread.some(
(message) => message.sequence > watermark || priorPointedIds?.has(message.id) !== true
)
) {
return
}
if (!leaf.writable || !leaf.ptyId) {
return
}
@@ -33599,7 +33607,10 @@ export class OrcaRuntimeService {
}
const deliveryPtyId = leaf.ptyId
const flight: { enterTimer: ReturnType<typeof setTimeout> | null } = { enterTimer: null }
const flight: {
enterTimer: ReturnType<typeof setTimeout> | null
stagedMessageIds: string[]
} = { enterTimer: null, stagedMessageIds: [] }
this.messageDeliveryFlightsByPtyId.set(deliveryPtyId, flight)
let settlesInEnterCallback = false
try {
@@ -33608,16 +33619,8 @@ export class OrcaRuntimeService {
if (!wrote) {
return
}
this.lastPointedMessageSequenceByHandle.set(
mailboxHandle,
Math.max(watermark, newestSequence)
)
const pointedIdsAfterWrite =
this.pointedMessageIdsByHandle.get(mailboxHandle) ?? new Set<string>()
for (const message of unread) {
pointedIdsAfterWrite.add(message.id)
}
this.pointedMessageIdsByHandle.set(mailboxHandle, pointedIdsAfterWrite)
flight.stagedMessageIds = unread.map((message) => message.id)
this._orchestrationDb.markAsDelivered(flight.stagedMessageIds)
this.setPointedMessageWatermark(
mailboxHandle,
newestSequence,
@@ -33627,17 +33630,7 @@ export class OrcaRuntimeService {
const tabTitle = this.tabs.get(leaf.tabId)?.title
if (isCursorAgentOrchestrationTarget(leaf, tabTitle)) {
let delivered = false
try {
this._orchestrationDb.markAsDelivered(unread.map((message) => message.id))
delivered = true
} finally {
if (delivered) {
this.clearPointedMessageWatermark(mailboxHandle, newestSequence, deliveryPtyId)
} else {
this.deactivatePointedMessageWatermark(mailboxHandle, newestSequence, deliveryPtyId)
}
}
this.clearPointedMessageWatermark(mailboxHandle, newestSequence, deliveryPtyId)
this.redrivePendingMessageMailbox(mailboxHandle)
return
}
@@ -33645,7 +33638,7 @@ export class OrcaRuntimeService {
// Why: agent TUIs can swallow a \r in the same PTY write; submit separately after a delay.
flight.enterTimer = setTimeout(() => {
let clearAndRedrive = false
let delivered = false
let submitted = false
let releaseWithoutRedrive = false
let finalizeReservation = true
void this.isLeafPtyProvenAbsent(deliveryPtyId)
@@ -33679,11 +33672,7 @@ export class OrcaRuntimeService {
releaseWithoutRedrive = true
return
}
const submitted = this.ptyController?.write(deliveryPtyId, '\r') ?? false
if (submitted) {
this._orchestrationDb?.markAsDelivered(unread.map((message) => message.id))
delivered = true
}
submitted = this.ptyController?.write(deliveryPtyId, '\r') ?? false
})
.catch(() => {
// Terminal liveness is uncertain; mail remains queued for explicit check.
@@ -33691,8 +33680,11 @@ export class OrcaRuntimeService {
.finally(() => {
let released = false
if (finalizeReservation) {
if (clearAndRedrive) {
this._orchestrationDb?.markAsUndelivered(unread.map((message) => message.id))
}
released =
delivered || clearAndRedrive || releaseWithoutRedrive
submitted || clearAndRedrive || releaseWithoutRedrive
? this.clearPointedMessageWatermark(mailboxHandle, newestSequence, deliveryPtyId)
: this.deactivatePointedMessageWatermark(
mailboxHandle,
src/main/runtime/orchestration-mailbox-notification-consistency.test.ts
+202 -216
View File
@@ -3,9 +3,29 @@ import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, describe, expect, it, vi } from 'vitest'
import { ORCHESTRATION_CONTRACT_VERSION } from '../../shared/protocol-version'
import type Database from '../sqlite/sync-database'
import { OrcaRuntimeService } from './orca-runtime'
import { OrchestrationDb } from './orchestration/db'
import {
checkBoundMailbox,
createBoundRun,
createDatabase,
createRuntime,
driveToLiveIdle,
insertDirectRunMessage,
LAUNCH_TOKEN,
LEAF_ID,
PANE_KEY,
pointerCount,
PTY_ID,
registerSecondPane,
SECOND_LAUNCH_TOKEN,
SECOND_LEAF_ID,
SECOND_PANE_KEY,
SECOND_PTY_ID,
SECOND_TERMINAL_HANDLE,
sqliteFor,
temporaryDirectories,
TERMINAL_HANDLE
} from './orchestration-mailbox-notification-test-harness'
import { RpcDispatcher } from './rpc/dispatcher'
import { ORCHESTRATION_METHODS } from './rpc/methods/orchestration'
@@ -16,212 +36,6 @@ vi.mock('electron', () => ({
webContents: { fromId: vi.fn(() => null) }
}))
const TAB_ID = '11111111-1111-4111-8111-111111111111'
const LEAF_ID = '22222222-2222-4222-8222-222222222222'
const PANE_KEY = `${TAB_ID}:${LEAF_ID}`
const PTY_ID = 'pty-mailbox'
const TERMINAL_HANDLE = 'term_mailbox_consistency'
const SECOND_TAB_ID = '33333333-3333-4333-8333-333333333333'
const SECOND_LEAF_ID = '44444444-4444-4444-8444-444444444444'
const SECOND_PANE_KEY = `${SECOND_TAB_ID}:${SECOND_LEAF_ID}`
const SECOND_PTY_ID = 'pty-mailbox-second'
const SECOND_TERMINAL_HANDLE = 'term_mailbox_consistency_second'
const SECOND_LAUNCH_TOKEN = 'mailbox-consistency-second-launch'
const WORKTREE_ID = 'repo-mailbox::/tmp/mailbox'
const LAUNCH_TOKEN = 'mailbox-consistency-launch'
const temporaryDirectories: string[] = []
function createDatabase(prefix: string): OrchestrationDb {
const directory = mkdtempSync(join(tmpdir(), prefix))
temporaryDirectories.push(directory)
return new OrchestrationDb(join(directory, 'orchestration.db'))
}
function sqliteFor(db: OrchestrationDb): Database.Database {
return (db as unknown as { db: Database.Database }).db
}
function createBoundRun(db: OrchestrationDb, objective: string) {
return db.createRun({
objective,
coordinatorHandle: TERMINAL_HANDLE,
coordinatorPaneKey: PANE_KEY
})
}
function insertDirectRunMessage(db: OrchestrationDb, runId: string, subject: string) {
return db.insertMessage({
from: 'term_worker',
to: TERMINAL_HANDLE,
subject,
type: 'status',
runId,
deliveryContract: 'current_delivery'
})
}
type Harness = {
runtime: OrcaRuntimeService
write: ReturnType<typeof vi.fn>
}
type CheckResult = {
runId: string
dispatchId?: string
deliveryId: string | null
count: number
messages: unknown[]
acknowledged?: string | null
}
function createRuntime(db: OrchestrationDb): Harness {
const runtime = new OrcaRuntimeService(null, undefined, {
attestAgentHookCompatibilityAuthority: ({ paneKey }) =>
paneKey === PANE_KEY || paneKey.startsWith(`${SECOND_TAB_ID}:`)
? { paneKey, source: 'current_hook' }
: null
})
const write = vi.fn(() => true)
runtime.setOrchestrationDb(db)
runtime.setPtyController({
write,
kill: vi.fn(),
getForegroundProcess: async () => null
})
runtime.registerPty(PTY_ID, WORKTREE_ID, null, {
tabId: TAB_ID,
leafId: LEAF_ID,
incarnationId: 'mailbox-incarnation',
agentLaunchAuthority: { launchToken: LAUNCH_TOKEN, launchAgent: 'codex' }
})
runtime.registerPreAllocatedHandleForPty(PTY_ID, TERMINAL_HANDLE)
runtime.attachWindow(1)
runtime.syncWindowGraph(1, {
tabs: [
{
tabId: TAB_ID,
worktreeId: WORKTREE_ID,
title: 'Codex',
activeLeafId: LEAF_ID,
layout: null
}
],
leaves: [
{
tabId: TAB_ID,
worktreeId: WORKTREE_ID,
leafId: LEAF_ID,
paneRuntimeId: 1,
ptyId: PTY_ID
}
]
})
return { runtime, write }
}
function registerSecondPane(
runtime: OrcaRuntimeService,
leafId = SECOND_LEAF_ID,
includeFirstPane = true
): void {
runtime.registerPty(SECOND_PTY_ID, WORKTREE_ID, null, {
tabId: SECOND_TAB_ID,
leafId,
incarnationId: 'mailbox-second-incarnation',
agentLaunchAuthority: { launchToken: SECOND_LAUNCH_TOKEN, launchAgent: 'codex' }
})
runtime.registerPreAllocatedHandleForPty(SECOND_PTY_ID, SECOND_TERMINAL_HANDLE)
runtime.syncWindowGraph(1, {
tabs: [
...(includeFirstPane
? [
{
tabId: TAB_ID,
worktreeId: WORKTREE_ID,
title: 'Codex',
activeLeafId: LEAF_ID,
layout: null
}
]
: []),
{
tabId: SECOND_TAB_ID,
worktreeId: WORKTREE_ID,
title: 'Codex',
activeLeafId: leafId,
layout: null
}
],
leaves: [
...(includeFirstPane
? [
{
tabId: TAB_ID,
worktreeId: WORKTREE_ID,
leafId: LEAF_ID,
paneRuntimeId: 1,
ptyId: PTY_ID
}
]
: []),
{
tabId: SECOND_TAB_ID,
worktreeId: WORKTREE_ID,
leafId,
paneRuntimeId: 2,
ptyId: SECOND_PTY_ID
}
]
})
}
async function driveToLiveIdle(runtime: OrcaRuntimeService): Promise<void> {
await runtime.listTerminals()
runtime.onPtyData(PTY_ID, '\x1b]0;Codex working\x07', 1)
runtime.onPtyData(PTY_ID, '\x1b]0;Codex done\x07', 2)
await Promise.resolve()
}
function pointerCount(write: ReturnType<typeof vi.fn>): number {
return write.mock.calls.filter(([, payload]) =>
String(payload).includes('orca orchestration check')
).length
}
async function checkBoundMailbox(
runtime: OrcaRuntimeService,
options: {
ack?: string
wait?: boolean
terminal?: string
paneKey?: string
launchToken?: string
} = {}
): Promise<CheckResult> {
const terminal = options.terminal ?? TERMINAL_HANDLE
const response = await new RpcDispatcher({ runtime, methods: ORCHESTRATION_METHODS }).dispatch({
id: 'req-mailbox-consistency',
authToken: 'test-auth-token',
method: 'orchestration.check',
orchestrationContractVersion: ORCHESTRATION_CONTRACT_VERSION,
orchestrationCompatibilityEvidence: {
terminalHandle: terminal,
paneKey: options.paneKey ?? PANE_KEY,
launchToken: options.launchToken ?? LAUNCH_TOKEN
},
params: {
terminal,
...(options.ack ? { ack: options.ack } : {}),
...(options.wait ? { wait: true, timeoutMs: 5_000 } : {})
}
})
expect(response.ok).toBe(true)
if (!response.ok) {
throw new Error(response.error.message)
}
return response.result as CheckResult
}
describe('orchestration notification mailbox consistency', () => {
afterEach(() => {
vi.useRealTimers()
@@ -338,6 +152,46 @@ describe('orchestration notification mailbox consistency', () => {
db.close()
})
it('uses composite indexes for direct ownership reconciliation', () => {
const db = createDatabase('orca-mailbox-routing-indexes-')
const sqlite = sqliteFor(db)
const plan = (sql: string, ...params: string[]): string =>
(sqlite.prepare(`EXPLAIN QUERY PLAN ${sql}`).all(...params) as { detail: string }[])
.map((row) => row.detail)
.join(' ')
expect(
plan(
`SELECT * FROM dispatch_contexts
WHERE run_id = ? AND assignee_handle = ? AND status IN ('pending', 'dispatched')
ORDER BY rowid DESC LIMIT 1`,
'run_indexed',
TERMINAL_HANDLE
)
).toContain('idx_dispatch_active_run_assignee_handle')
expect(
plan(
`SELECT * FROM dispatch_contexts
WHERE run_id = ? AND assignee_pane_key IS NOT NULL
AND status IN ('pending', 'dispatched') AND instr(assignee_pane_key, ':') > 1
AND substr(assignee_pane_key, instr(assignee_pane_key, ':') + 1) = ?
ORDER BY rowid DESC LIMIT 1`,
'run_indexed',
LEAF_ID
)
).toContain('idx_dispatch_active_run_pane_leaf')
expect(
plan(
`SELECT 1 FROM messages INDEXED BY idx_messages_undelivered_direct_run
WHERE run_id = ? AND to_handle = ? AND read = 0 AND delivered_at IS NULL
AND delivery_contract = 'current_delivery' LIMIT 1`,
'run_indexed',
TERMINAL_HANDLE
)
).toContain('idx_messages_undelivered_direct_run')
db.close()
})
it('does not submit or replay a pointer after ownership changes during Enter delay', async () => {
vi.useFakeTimers()
const directory = mkdtempSync(join(tmpdir(), 'orca-mailbox-restart-'))
@@ -371,6 +225,33 @@ describe('orchestration notification mailbox consistency', () => {
restartedDb.close()
})
it('does not replay a staged same-Run pointer when the runtime restarts before Enter', async () => {
vi.useFakeTimers()
const directory = mkdtempSync(join(tmpdir(), 'orca-mailbox-staged-restart-'))
temporaryDirectories.push(directory)
const dbPath = join(directory, 'orchestration.db')
const firstDb = new OrchestrationDb(dbPath)
const first = createRuntime(firstDb)
const run = createBoundRun(firstDb, 'Staged restart Run')
const message = insertDirectRunMessage(firstDb, run.id, 'Visible before restart')
await driveToLiveIdle(first.runtime)
expect(pointerCount(first.write)).toBe(1)
expect(first.write.mock.calls.filter(([, payload]) => payload === '\r')).toHaveLength(0)
expect(firstDb.getMessageById(message.id)?.delivered_at).toEqual(expect.any(String))
firstDb.close()
const restartedDb = new OrchestrationDb(dbPath)
const restarted = createRuntime(restartedDb)
await driveToLiveIdle(restarted.runtime)
const checked = await checkBoundMailbox(restarted.runtime)
expect(pointerCount(restarted.write)).toBe(0)
expect(checked).toMatchObject({ runId: run.id, count: 1 })
expect(checked.messages).toEqual([expect.objectContaining({ id: message.id })])
restartedDb.close()
})
it('fences the pointed mailbox instead of checking a rebound empty Run', async () => {
vi.useFakeTimers()
const db = createDatabase('orca-mailbox-post-submit-rebind-')
@@ -514,7 +395,7 @@ describe('orchestration notification mailbox consistency', () => {
await vi.advanceTimersByTimeAsync(500)
expect(harness.write.mock.calls.filter(([, payload]) => payload === '\r')).toHaveLength(0)
expect(db.getMessageById(message.id)?.delivered_at).toBeNull()
expect(db.getMessageById(message.id)?.delivered_at).toEqual(expect.any(String))
harness.runtime.onPtyData(PTY_ID, '\x1b]0;Codex done\x07', 4)
await Promise.resolve()
expect(pointerCount(harness.write)).toBe(1)
@@ -564,7 +445,7 @@ describe('orchestration notification mailbox consistency', () => {
})
)
const notificationQuery = vi.spyOn(db, 'getUndeliveredUnreadMessages')
const submitRevalidation = vi.spyOn(db, 'areUndeliveredUnreadMessages')
const submitRevalidation = vi.spyOn(db, 'areUnreadMessages')
const prepare = vi.spyOn(sqliteFor(db), 'prepare')
await driveToLiveIdle(harness.runtime)
@@ -642,7 +523,7 @@ describe('orchestration notification mailbox consistency', () => {
restartedDb.close()
})
it('keeps mail retryable when the provider rejects Enter', async () => {
it('does not replay a visible pointer when the provider rejects Enter', async () => {
vi.useFakeTimers()
const db = createDatabase('orca-mailbox-provider-refusal-')
const first = createRuntime(db)
@@ -662,11 +543,11 @@ describe('orchestration notification mailbox consistency', () => {
await driveToLiveIdle(first.runtime)
await vi.advanceTimersByTimeAsync(500)
expect(pointerCount(first.write)).toBe(1)
expect(db.getMessageById(message.id)?.delivered_at).toBeNull()
expect(db.getMessageById(message.id)?.delivered_at).toEqual(expect.any(String))
const restarted = createRuntime(db)
await driveToLiveIdle(restarted.runtime)
expect(pointerCount(restarted.write)).toBe(1)
expect(pointerCount(restarted.write)).toBe(0)
db.close()
})
@@ -702,7 +583,7 @@ describe('orchestration notification mailbox consistency', () => {
db.close()
})
it('routes active worker direct mail through the Dispatch check', async () => {
it('routes active worker direct mail without injecting an unpinned Dispatch pointer', async () => {
vi.useFakeTimers()
const db = createDatabase('orca-mailbox-dispatch-')
const harness = createRuntime(db)
@@ -727,13 +608,13 @@ describe('orchestration notification mailbox consistency', () => {
await vi.advanceTimersByTimeAsync(500)
const checked = await checkBoundMailbox(harness.runtime)
expect(pointerCount(harness.write)).toBe(1)
expect(pointerCount(harness.write)).toBe(0)
expect(checked).toMatchObject({ runId: run.id, dispatchId: dispatch.id, count: 1 })
expect(checked.messages).toEqual([expect.objectContaining({ id: message.id })])
expect(db.getMessageById(message.id)).toMatchObject({
to_handle: `dispatch:${dispatch.id}`,
read: 1,
delivered_at: expect.any(String)
delivered_at: null
})
const internals = harness.runtime as unknown as {
pointedMessageWatermarkOwnerByHandle: Map<string, unknown>
@@ -877,4 +758,109 @@ describe('orchestration notification mailbox consistency', () => {
expect(pointerCount(harness.write)).toBe(0)
db.close()
})
it('drains persisted Run pages before installing a filtered waiter', async () => {
const db = createDatabase('orca-mailbox-filtered-run-backlog-')
const harness = createRuntime(db)
const run = createBoundRun(db, 'Filtered Run backlog')
for (let index = 0; index < 50; index += 1) {
insertDirectRunMessage(db, run.id, `Status ${index}`)
}
const question = db.insertMessage({
from: 'term_worker',
to: TERMINAL_HANDLE,
subject: 'Question behind first page',
type: 'question',
runId: run.id
})
const checked = await checkBoundMailbox(harness.runtime, {
wait: true,
types: 'question'
})
expect(checked).toMatchObject({ runId: run.id, count: 50 })
expect(checked.messages).not.toContainEqual(expect.objectContaining({ id: question.id }))
expect(db.getMessageById(question.id)?.to_handle).toBe(`run:${run.id}`)
const next = await checkBoundMailbox(harness.runtime, {
ack: checked.deliveryId!,
types: 'question'
})
expect(next.messages).toEqual(
expect.arrayContaining([expect.objectContaining({ id: question.id })])
)
db.close()
})
it('wakes a filtered waiter when reconciliation moves its type on a later page', async () => {
const db = createDatabase('orca-mailbox-filtered-reconciliation-wake-')
const harness = createRuntime(db)
const run = createBoundRun(db, 'Filtered reconciliation wake')
const waiting = checkBoundMailbox(harness.runtime, { wait: true, types: 'question' })
const internals = harness.runtime as unknown as {
messageWaitersByHandle: Map<string, Set<unknown>>
}
await vi.waitFor(() => {
expect(internals.messageWaitersByHandle.has(`run:${run.id}`)).toBe(true)
})
for (let index = 0; index < 50; index += 1) {
insertDirectRunMessage(db, run.id, `Status before question ${index}`)
}
const question = db.insertMessage({
from: 'term_worker',
to: TERMINAL_HANDLE,
subject: 'Question moved by continuation',
type: 'question',
runId: run.id
})
const arrivingStatus = insertDirectRunMessage(db, run.id, 'Status arrival trigger')
harness.runtime.notifyMessageArrived(TERMINAL_HANDLE, arrivingStatus.type)
const checked = await waiting
expect(checked).toMatchObject({ runId: run.id, count: 50 })
expect(checked.messages).not.toContainEqual(expect.objectContaining({ id: question.id }))
expect(db.getMessageById(question.id)?.to_handle).toBe(`run:${run.id}`)
const next = await checkBoundMailbox(harness.runtime, {
ack: checked.deliveryId!,
types: 'question'
})
expect(next.messages).toEqual(
expect.arrayContaining([expect.objectContaining({ id: question.id })])
)
db.close()
})
it('drains persisted Dispatch pages before installing a filtered waiter', async () => {
const db = createDatabase('orca-mailbox-filtered-dispatch-backlog-')
const harness = createRuntime(db)
const run = db.createRun({
objective: 'Filtered Dispatch backlog',
coordinatorHandle: 'term_coordinator',
coordinatorPaneKey:
'55555555-5555-4555-8555-555555555555:66666666-6666-4666-8666-666666666666'
})
const task = db.createTask({ spec: 'Worker task', runId: run.id })
const dispatch = db.createDispatchContext(task.id, TERMINAL_HANDLE, PANE_KEY)
for (let index = 0; index < 50; index += 1) {
insertDirectRunMessage(db, run.id, `Worker status ${index}`)
}
const question = db.insertMessage({
from: 'term_coordinator',
to: TERMINAL_HANDLE,
subject: 'Worker question behind first page',
type: 'question',
runId: run.id
})
const checked = await checkBoundMailbox(harness.runtime, {
wait: true,
types: 'question'
})
expect(checked).toMatchObject({ runId: run.id, dispatchId: dispatch.id, count: 1 })
expect(checked.messages).toEqual([expect.objectContaining({ id: question.id })])
expect(db.getMessageById(question.id)?.to_handle).toBe(`dispatch:${dispatch.id}`)
db.close()
})
})
src/main/runtime/orchestration-mailbox-notification-test-harness.ts
+231
View File
@@ -0,0 +1,231 @@
import { mkdtempSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { expect, vi } from 'vitest'
import { ORCHESTRATION_CONTRACT_VERSION } from '../../shared/protocol-version'
import type Database from '../sqlite/sync-database'
import { OrcaRuntimeService } from './orca-runtime'
import { OrchestrationDb } from './orchestration/db'
import { RpcDispatcher } from './rpc/dispatcher'
import { ORCHESTRATION_METHODS } from './rpc/methods/orchestration'
export const TAB_ID = '11111111-1111-4111-8111-111111111111'
export const LEAF_ID = '22222222-2222-4222-8222-222222222222'
export const PANE_KEY = `${TAB_ID}:${LEAF_ID}`
export const PTY_ID = 'pty-mailbox'
export const TERMINAL_HANDLE = 'term_mailbox_consistency'
export const SECOND_TAB_ID = '33333333-3333-4333-8333-333333333333'
export const SECOND_LEAF_ID = '44444444-4444-4444-8444-444444444444'
export const SECOND_PANE_KEY = `${SECOND_TAB_ID}:${SECOND_LEAF_ID}`
export const SECOND_PTY_ID = 'pty-mailbox-second'
export const SECOND_TERMINAL_HANDLE = 'term_mailbox_consistency_second'
export const SECOND_LAUNCH_TOKEN = 'mailbox-consistency-second-launch'
export const WORKTREE_ID = 'repo-mailbox::/tmp/mailbox'
export const LAUNCH_TOKEN = 'mailbox-consistency-launch'
export const temporaryDirectories: string[] = []
export function createDatabase(prefix: string): OrchestrationDb {
const directory = mkdtempSync(join(tmpdir(), prefix))
temporaryDirectories.push(directory)
return new OrchestrationDb(join(directory, 'orchestration.db'))
}
export function sqliteFor(db: OrchestrationDb): Database.Database {
return (db as unknown as { db: Database.Database }).db
}
export function createBoundRun(db: OrchestrationDb, objective: string) {
return db.createRun({
objective,
coordinatorHandle: TERMINAL_HANDLE,
coordinatorPaneKey: PANE_KEY
})
}
export function insertDirectRunMessage(db: OrchestrationDb, runId: string, subject: string) {
return db.insertMessage({
from: 'term_worker',
to: TERMINAL_HANDLE,
subject,
type: 'status',
runId,
deliveryContract: 'current_delivery'
})
}
export type MailboxNotificationHarness = {
runtime: OrcaRuntimeService
write: ReturnType<typeof vi.fn>
}
export type MailboxCheckResult = {
runId: string
dispatchId?: string
deliveryId: string | null
count: number
messages: unknown[]
acknowledged?: string | null
}
export type MailboxCheckOptions = {
ack?: string
wait?: boolean
terminal?: string
paneKey?: string
launchToken?: string
types?: string
signal?: AbortSignal
}
export function createRuntime(db: OrchestrationDb): MailboxNotificationHarness {
const runtime = new OrcaRuntimeService(null, undefined, {
attestAgentHookCompatibilityAuthority: ({ paneKey }) =>
paneKey === PANE_KEY || paneKey.startsWith(`${SECOND_TAB_ID}:`)
? { paneKey, source: 'current_hook' }
: null
})
const write = vi.fn(() => true)
runtime.setOrchestrationDb(db)
runtime.setPtyController({
write,
kill: vi.fn(),
getForegroundProcess: async () => null
})
runtime.registerPty(PTY_ID, WORKTREE_ID, null, {
tabId: TAB_ID,
leafId: LEAF_ID,
incarnationId: 'mailbox-incarnation',
agentLaunchAuthority: { launchToken: LAUNCH_TOKEN, launchAgent: 'codex' }
})
runtime.registerPreAllocatedHandleForPty(PTY_ID, TERMINAL_HANDLE)
runtime.attachWindow(1)
runtime.syncWindowGraph(1, {
tabs: [
{
tabId: TAB_ID,
worktreeId: WORKTREE_ID,
title: 'Codex',
activeLeafId: LEAF_ID,
layout: null
}
],
leaves: [
{
tabId: TAB_ID,
worktreeId: WORKTREE_ID,
leafId: LEAF_ID,
paneRuntimeId: 1,
ptyId: PTY_ID
}
]
})
return { runtime, write }
}
export function registerSecondPane(
runtime: OrcaRuntimeService,
leafId = SECOND_LEAF_ID,
includeFirstPane = true
): void {
runtime.registerPty(SECOND_PTY_ID, WORKTREE_ID, null, {
tabId: SECOND_TAB_ID,
leafId,
incarnationId: 'mailbox-second-incarnation',
agentLaunchAuthority: { launchToken: SECOND_LAUNCH_TOKEN, launchAgent: 'codex' }
})
runtime.registerPreAllocatedHandleForPty(SECOND_PTY_ID, SECOND_TERMINAL_HANDLE)
runtime.syncWindowGraph(1, {
tabs: [
...(includeFirstPane
? [
{
tabId: TAB_ID,
worktreeId: WORKTREE_ID,
title: 'Codex',
activeLeafId: LEAF_ID,
layout: null
}
]
: []),
{
tabId: SECOND_TAB_ID,
worktreeId: WORKTREE_ID,
title: 'Codex',
activeLeafId: leafId,
layout: null
}
],
leaves: [
...(includeFirstPane
? [
{
tabId: TAB_ID,
worktreeId: WORKTREE_ID,
leafId: LEAF_ID,
paneRuntimeId: 1,
ptyId: PTY_ID
}
]
: []),
{
tabId: SECOND_TAB_ID,
worktreeId: WORKTREE_ID,
leafId,
paneRuntimeId: 2,
ptyId: SECOND_PTY_ID
}
]
})
}
export async function driveToLiveIdle(runtime: OrcaRuntimeService): Promise<void> {
await runtime.listTerminals()
runtime.onPtyData(PTY_ID, '\x1b]0;Codex working\x07', 1)
runtime.onPtyData(PTY_ID, '\x1b]0;Codex done\x07', 2)
await Promise.resolve()
}
export function pointerCount(write: ReturnType<typeof vi.fn>): number {
return write.mock.calls.filter(([, payload]) =>
String(payload).includes('orca orchestration check')
).length
}
export async function checkBoundMailbox(
runtime: OrcaRuntimeService,
options: MailboxCheckOptions = {}
): Promise<MailboxCheckResult> {
const response = await dispatchMailboxCheck(runtime, options)
expect(response.ok).toBe(true)
if (!response.ok) {
throw new Error(response.error.message)
}
return response.result as MailboxCheckResult
}
export async function dispatchMailboxCheck(
runtime: OrcaRuntimeService,
options: MailboxCheckOptions = {}
) {
const terminal = options.terminal ?? TERMINAL_HANDLE
return new RpcDispatcher({ runtime, methods: ORCHESTRATION_METHODS }).dispatch(
{
id: 'req-mailbox-consistency',
authToken: 'test-auth-token',
method: 'orchestration.check',
orchestrationContractVersion: ORCHESTRATION_CONTRACT_VERSION,
orchestrationCompatibilityEvidence: {
terminalHandle: terminal,
paneKey: options.paneKey ?? PANE_KEY,
launchToken: options.launchToken ?? LAUNCH_TOKEN
},
params: {
terminal,
...(options.types ? { types: options.types } : {}),
...(options.ack ? { ack: options.ack } : {}),
...(options.wait ? { wait: true, timeoutMs: 5_000 } : {})
}
},
{ signal: options.signal }
)
}
src/main/runtime/orchestration-mailbox-routing-races.test.ts
+334
View File
@@ -0,0 +1,334 @@
import { rmSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { afterEach, describe, expect, it, vi } from 'vitest'
import {
checkBoundMailbox,
createBoundRun,
createDatabase,
createRuntime,
dispatchMailboxCheck,
insertDirectRunMessage,
LEAF_ID,
PANE_KEY,
registerSecondPane,
SECOND_LAUNCH_TOKEN,
SECOND_PANE_KEY,
SECOND_TERMINAL_HANDLE,
sqliteFor,
temporaryDirectories,
TERMINAL_HANDLE
} from './orchestration-mailbox-notification-test-harness'
vi.mock('electron', () => ({
app: { getPath: vi.fn(() => tmpdir()), isPackaged: false },
BrowserWindow: { fromId: vi.fn(() => null) },
ipcMain: { on: vi.fn(), removeListener: vi.fn() },
webContents: { fromId: vi.fn(() => null) }
}))
describe('orchestration mailbox routing races', () => {
afterEach(() => {
for (const directory of temporaryDirectories.splice(0)) {
rmSync(directory, { recursive: true, force: true })
}
})
it('fences a Run consumer rebound during a routing-page yield', async () => {
const db = createDatabase('orca-mailbox-run-routing-fence-')
const harness = createRuntime(db)
const run = createBoundRun(db, 'Run routing fence')
for (let index = 0; index < 51; index += 1) {
insertDirectRunMessage(db, run.id, `Before rebind ${index}`)
}
const route = db.routeUnreadDirectMessagesToRunMailbox.bind(db)
let scheduled = false
vi.spyOn(db, 'routeUnreadDirectMessagesToRunMailbox').mockImplementation((...args) => {
const page = route(...args)
if (page.hasMore && !scheduled) {
scheduled = true
setImmediate(() => createBoundRun(db, 'Replacement Run'))
}
return page
})
const response = await dispatchMailboxCheck(harness.runtime)
expect(response).toMatchObject({ ok: false, error: { code: 'consumer_fenced' } })
expect(db.getUnreadMessages(`run:${run.id}`)).toHaveLength(51)
db.close()
})
it('fences a Dispatch completed during a routing-page yield', async () => {
const db = createDatabase('orca-mailbox-dispatch-routing-fence-')
const harness = createRuntime(db)
registerSecondPane(harness.runtime)
const run = db.createRun({
objective: 'Dispatch routing fence',
coordinatorHandle: SECOND_TERMINAL_HANDLE,
coordinatorPaneKey: SECOND_PANE_KEY
})
const task = db.createTask({ spec: 'Worker task', runId: run.id })
const dispatch = db.createDispatchContext(task.id, TERMINAL_HANDLE, PANE_KEY)
for (let index = 0; index < 51; index += 1) {
insertDirectRunMessage(db, run.id, `Before completion ${index}`)
}
const route = db.routeUnreadDirectMessagesToDispatchMailbox.bind(db)
let scheduled = false
vi.spyOn(db, 'routeUnreadDirectMessagesToDispatchMailbox').mockImplementation((...args) => {
const page = route(...args)
if (page.hasMore && !scheduled) {
scheduled = true
setImmediate(() => db.completeDispatch(dispatch.id))
}
return page
})
const response = await dispatchMailboxCheck(harness.runtime)
expect(response).toMatchObject({ ok: false, error: { code: 'dispatch_inactive' } })
expect(db.getUnreadMessages(`dispatch:${dispatch.id}`)).toHaveLength(0)
const first = await checkBoundMailbox(harness.runtime, {
terminal: SECOND_TERMINAL_HANDLE,
paneKey: SECOND_PANE_KEY,
launchToken: SECOND_LAUNCH_TOKEN
})
const second = await checkBoundMailbox(harness.runtime, {
ack: first.deliveryId!,
terminal: SECOND_TERMINAL_HANDLE,
paneKey: SECOND_PANE_KEY,
launchToken: SECOND_LAUNCH_TOKEN
})
const acknowledged = await checkBoundMailbox(harness.runtime, {
ack: second.deliveryId!,
terminal: SECOND_TERMINAL_HANDLE,
paneKey: SECOND_PANE_KEY,
launchToken: SECOND_LAUNCH_TOKEN
})
expect(first.count).toBe(50)
expect(second.count).toBe(1)
expect(acknowledged).toMatchObject({ count: 0, acknowledged: second.deliveryId })
db.close()
})
it('reconciles filtered-out mail when a Dispatch wait loses authority', async () => {
const db = createDatabase('orca-mailbox-dispatch-wait-fence-')
const harness = createRuntime(db)
registerSecondPane(harness.runtime)
const run = db.createRun({
objective: 'Dispatch wait fence',
coordinatorHandle: SECOND_TERMINAL_HANDLE,
coordinatorPaneKey: SECOND_PANE_KEY
})
const task = db.createTask({ spec: 'Waiting worker', runId: run.id })
const dispatch = db.createDispatchContext(task.id, TERMINAL_HANDLE, PANE_KEY)
const status = insertDirectRunMessage(db, run.id, 'Filtered-out status')
const controller = new AbortController()
const waiting = dispatchMailboxCheck(harness.runtime, {
wait: true,
types: 'question',
signal: controller.signal
})
const internals = harness.runtime as unknown as {
messageWaitersByHandle: Map<string, Set<unknown>>
}
await vi.waitFor(() => {
expect(internals.messageWaitersByHandle.has(`dispatch:${dispatch.id}`)).toBe(true)
})
db.completeDispatch(dispatch.id)
controller.abort()
const response = await waiting
expect(response).toMatchObject({ ok: false, error: { code: 'dispatch_inactive' } })
const checked = await checkBoundMailbox(harness.runtime, {
terminal: SECOND_TERMINAL_HANDLE,
paneKey: SECOND_PANE_KEY,
launchToken: SECOND_LAUNCH_TOKEN
})
expect(checked.messages).toEqual([expect.objectContaining({ id: status.id })])
const acknowledged = await checkBoundMailbox(harness.runtime, {
ack: checked.deliveryId!,
terminal: SECOND_TERMINAL_HANDLE,
paneKey: SECOND_PANE_KEY,
launchToken: SECOND_LAUNCH_TOKEN
})
expect(acknowledged).toMatchObject({ count: 0, acknowledged: checked.deliveryId })
db.close()
})
it('drains current and stored Run handles in one check', async () => {
const db = createDatabase('orca-mailbox-split-run-handles-')
const harness = createRuntime(db)
const run = db.createRun({
objective: 'Split Run handles',
coordinatorHandle: 'term_previous_coordinator',
coordinatorPaneKey: PANE_KEY
})
const current = insertDirectRunMessage(db, run.id, 'Current handle')
const previous = db.insertMessage({
from: 'term_worker',
to: 'term_previous_coordinator',
subject: 'Previous handle',
type: 'status',
runId: run.id
})
const checked = await checkBoundMailbox(harness.runtime)
expect(checked.messages).toEqual(
expect.arrayContaining([
expect.objectContaining({ id: current.id }),
expect.objectContaining({ id: previous.id })
])
)
db.close()
})
it('drains current and stored Dispatch handles in one check', async () => {
const db = createDatabase('orca-mailbox-split-dispatch-handles-')
const harness = createRuntime(db)
const run = db.createRun({
objective: 'Split Dispatch handles',
coordinatorHandle: 'term_coordinator',
coordinatorPaneKey:
'55555555-5555-4555-8555-555555555555:66666666-6666-4666-8666-666666666666'
})
const task = db.createTask({ spec: 'Worker task', runId: run.id })
db.createDispatchContext(task.id, 'term_previous_worker', PANE_KEY)
const current = insertDirectRunMessage(db, run.id, 'Current worker handle')
const previous = db.insertMessage({
from: 'term_coordinator',
to: 'term_previous_worker',
subject: 'Previous worker handle',
type: 'status',
runId: run.id
})
const checked = await checkBoundMailbox(harness.runtime)
expect(checked.messages).toEqual(
expect.arrayContaining([
expect.objectContaining({ id: current.id }),
expect.objectContaining({ id: previous.id })
])
)
db.close()
})
it('routes only the direct-mail snapshot captured when the check starts', async () => {
const db = createDatabase('orca-mailbox-routing-snapshot-')
const harness = createRuntime(db)
const run = createBoundRun(db, 'Routing snapshot')
for (let index = 0; index < 51; index += 1) {
insertDirectRunMessage(db, run.id, `Initial ${index}`)
}
const route = db.routeUnreadDirectMessagesToRunMailbox.bind(db)
let inserted = false
const routeSpy = vi
.spyOn(db, 'routeUnreadDirectMessagesToRunMailbox')
.mockImplementation((...args) => {
const page = route(...args)
if (page.hasMore && !inserted) {
inserted = true
for (let index = 0; index < 60; index += 1) {
insertDirectRunMessage(db, run.id, `Concurrent ${index}`)
}
}
return page
})
await checkBoundMailbox(harness.runtime)
expect(routeSpy).toHaveBeenCalledTimes(2)
expect(db.getUnreadMessages(TERMINAL_HANDLE)).toHaveLength(60)
db.close()
})
it('stops paged routing when the check is cancelled', async () => {
const db = createDatabase('orca-mailbox-routing-cancel-')
const harness = createRuntime(db)
const run = createBoundRun(db, 'Routing cancellation')
for (let index = 0; index < 101; index += 1) {
insertDirectRunMessage(db, run.id, `Before cancellation ${index}`)
}
const controller = new AbortController()
const route = db.routeUnreadDirectMessagesToRunMailbox.bind(db)
let scheduled = false
const routeSpy = vi
.spyOn(db, 'routeUnreadDirectMessagesToRunMailbox')
.mockImplementation((...args) => {
const page = route(...args)
if (!scheduled) {
scheduled = true
setImmediate(() => controller.abort())
}
return page
})
const response = await dispatchMailboxCheck(harness.runtime, {
signal: controller.signal
})
expect(response).toMatchObject({ ok: false, error: { code: 'runtime_error' } })
expect(routeSpy).toHaveBeenCalledOnce()
db.close()
})
it('preserves a later filtered reservation behind an unfiltered redrive', () => {
const db = createDatabase('orca-mailbox-reservation-union-')
const harness = createRuntime(db)
const internals = harness.runtime as unknown as {
parkMessageRedeliveryForMailbox: (
mailboxHandle: string,
reservedTypes: ReadonlySet<string> | undefined
) => void
parkedMessageRedeliveryTypesByMailboxHandle: Map<string, ReadonlySet<string> | null>
}
internals.parkMessageRedeliveryForMailbox('run:run_test', undefined)
internals.parkMessageRedeliveryForMailbox('run:run_test', new Set(['worker_done']))
expect(internals.parkedMessageRedeliveryTypesByMailboxHandle.get('run:run_test')).toEqual(
new Set(['worker_done'])
)
db.close()
})
it('uses a bounded indexed pane lookup for reminted Dispatch identity', () => {
const db = createDatabase('orca-mailbox-dispatch-pane-index-')
const run = db.createRun({
objective: 'Dispatch pane index',
coordinatorHandle: 'term_coordinator',
coordinatorPaneKey:
'55555555-5555-4555-8555-555555555555:66666666-6666-4666-8666-666666666666'
})
const task = db.createTask({ spec: 'Valid worker', runId: run.id })
const valid = db.createDispatchContext(task.id, 'term_old', PANE_KEY)
const collisionTask = db.createTask({ spec: 'Malformed collision', runId: run.id })
db.createDispatchContext(collisionTask.id, 'term_collision', `:${LEAF_ID}`)
expect(db.getActiveDispatchForIdentity('term_reminted', PANE_KEY)?.id).toBe(valid.id)
const plan = sqliteFor(db)
.prepare(
`EXPLAIN QUERY PLAN SELECT * FROM dispatch_contexts
WHERE assignee_pane_key IS NOT NULL
AND status IN ('pending', 'dispatched') AND instr(assignee_pane_key, ':') > 1
AND substr(assignee_pane_key, instr(assignee_pane_key, ':') + 1) = ?
ORDER BY rowid DESC LIMIT 1`
)
.all(LEAF_ID) as { detail: string }[]
expect(plan.map((row) => row.detail).join(' ')).toContain('idx_dispatch_assignee_pane_leaf')
const snapshotPlan = sqliteFor(db)
.prepare(
`EXPLAIN QUERY PLAN SELECT sequence FROM messages
WHERE run_id = ? AND to_handle = ? AND read = 0
AND delivery_contract = 'current_delivery'
ORDER BY sequence DESC LIMIT 1`
)
.all(run.id, TERMINAL_HANDLE) as { detail: string }[]
expect(snapshotPlan.map((row) => row.detail).join(' ')).toContain(
'idx_messages_delivery_contract'
)
db.close()
})
})
src/main/runtime/orchestration/db.ts
+136 -24
View File
@@ -93,6 +93,7 @@ export const ORCHESTRATION_DELIVERY_BATCH_LIMIT = 50
export type DirectMailboxRoutingPage = {
routedCount: number
hasMore: boolean
types: MessageType[]
}
export type ForeignDirectMailboxRoutingPage = DirectMailboxRoutingPage & {
@@ -1034,6 +1035,15 @@ export class OrchestrationDb {
CREATE INDEX IF NOT EXISTS idx_dispatch_assignee_pane_leaf
ON dispatch_contexts(${DISPATCH_PANE_KEY_MATCH_SUFFIX_SQL})
WHERE assignee_pane_key IS NOT NULL AND status IN ('pending', 'dispatched');
CREATE INDEX IF NOT EXISTS idx_dispatch_active_run_assignee_handle
ON dispatch_contexts(run_id, assignee_handle)
WHERE assignee_handle IS NOT NULL AND status IN ('pending', 'dispatched');
CREATE INDEX IF NOT EXISTS idx_dispatch_active_run_pane_leaf
ON dispatch_contexts(run_id, ${DISPATCH_PANE_KEY_MATCH_SUFFIX_SQL})
WHERE assignee_pane_key IS NOT NULL AND status IN ('pending', 'dispatched');
CREATE INDEX IF NOT EXISTS idx_dispatch_active_assignee_pane_key
ON dispatch_contexts(assignee_pane_key)
WHERE assignee_pane_key IS NOT NULL AND status IN ('pending', 'dispatched');
`)
this.createUndeliveredInboxIndexIfPossible()
@@ -1424,7 +1434,11 @@ export class OrchestrationDb {
}
this.db.exec(`
CREATE INDEX IF NOT EXISTS idx_messages_undelivered_inbox
ON messages(to_handle, read, delivered_at, sequence)
ON messages(to_handle, read, delivered_at, sequence);
CREATE INDEX IF NOT EXISTS idx_messages_undelivered_direct_run
ON messages(run_id, to_handle, sequence)
WHERE read = 0 AND delivered_at IS NULL
AND delivery_contract = 'current_delivery'
`)
}
@@ -3579,7 +3593,7 @@ export class OrchestrationDb {
return Boolean(
this.db
.prepare(
`SELECT 1 FROM messages
`SELECT 1 FROM messages INDEXED BY idx_messages_undelivered_direct_run
WHERE run_id = ? AND to_handle = ? AND read = 0 AND delivered_at IS NULL
AND delivery_contract = 'current_delivery'
LIMIT 1`
@@ -3596,23 +3610,48 @@ export class OrchestrationDb {
)
}
getLatestUnreadDirectMessageSequenceForRun(
runId: string,
directHandle: string
): number | undefined {
const row = this.db
.prepare(
`SELECT sequence FROM messages
WHERE run_id = ? AND to_handle = ? AND read = 0
AND delivery_contract = 'current_delivery'
ORDER BY sequence DESC LIMIT 1`
)
.get(runId, directHandle) as { sequence: number } | undefined
return row?.sequence
}
// Why: change mailbox ownership without changing unread or acknowledgment state.
private routeDirectMessagePage(
mailboxHandle: string,
runId: string,
directHandle: string
directHandle: string,
throughSequence?: number
): DirectMailboxRoutingPage {
const throughClause = throughSequence === undefined ? '' : ' AND sequence <= ?'
const params: (string | number)[] = [runId, directHandle]
if (throughSequence !== undefined) {
params.push(throughSequence)
}
params.push(ORCHESTRATION_DELIVERY_BATCH_LIMIT + 1)
const rows = this.db
.prepare(
`SELECT id FROM messages
`SELECT id, type FROM messages
WHERE run_id = ? AND to_handle = ? AND read = 0
AND delivery_contract = 'current_delivery'
AND delivery_contract = 'current_delivery'${throughClause}
ORDER BY sequence LIMIT ?`
)
.all(runId, directHandle, ORCHESTRATION_DELIVERY_BATCH_LIMIT + 1) as { id: string }[]
.all(...params) as {
id: string
type: MessageType
}[]
const page = rows.slice(0, ORCHESTRATION_DELIVERY_BATCH_LIMIT)
if (page.length === 0) {
return { routedCount: 0, hasMore: false }
return { routedCount: 0, hasMore: false, types: [] }
}
const placeholders = page.map(() => '?').join(',')
const result = this.db
@@ -3620,23 +3659,58 @@ export class OrchestrationDb {
.run(mailboxHandle, ...page.map((row) => row.id))
return {
routedCount: Number(result.changes),
hasMore: rows.length > ORCHESTRATION_DELIVERY_BATCH_LIMIT
hasMore: rows.length > ORCHESTRATION_DELIVERY_BATCH_LIMIT,
types: [...new Set(page.map((row) => row.type))]
}
}
routeUnreadDirectMessagesToRunMailbox(
runId: string,
directHandle: string
directHandle: string,
throughSequence?: number
): DirectMailboxRoutingPage {
return this.routeDirectMessagePage(`run:${runId}`, runId, directHandle)
return this.routeDirectMessagePage(`run:${runId}`, runId, directHandle, throughSequence)
}
routeUnreadDirectMessagesToDispatchMailbox(
dispatchId: string,
runId: string,
directHandle: string
directHandle: string,
throughSequence?: number
): DirectMailboxRoutingPage {
return this.routeDirectMessagePage(`dispatch:${dispatchId}`, runId, directHandle)
return this.routeDirectMessagePage(
`dispatch:${dispatchId}`,
runId,
directHandle,
throughSequence
)
}
routeUnreadDispatchMailboxToRunMailbox(
dispatchId: string,
runId: string
): { routedCount: number; types: MessageType[] } {
const dispatchMailbox = `dispatch:${dispatchId}`
this.db.exec('BEGIN IMMEDIATE')
try {
const rows = this.db
.prepare(
`SELECT DISTINCT type FROM messages
WHERE to_handle = ? AND read = 0 AND delivery_contract = 'current_delivery'`
)
.all(dispatchMailbox) as { type: MessageType }[]
const result = this.db
.prepare(
`UPDATE messages SET to_handle = ?
WHERE to_handle = ? AND read = 0 AND delivery_contract = 'current_delivery'`
)
.run(`run:${runId}`, dispatchMailbox)
this.db.exec('COMMIT')
return { routedCount: Number(result.changes), types: rows.map((row) => row.type) }
} catch (error) {
this.db.exec('ROLLBACK')
throw error
}
}
private findActiveDispatchForDirectMessageOwner(
@@ -3685,7 +3759,7 @@ export class OrchestrationDb {
}[]
const page = rows.slice(0, ORCHESTRATION_DELIVERY_BATCH_LIMIT)
if (page.length === 0) {
return { routedCount: 0, hasMore: false, mailboxes: [] }
return { routedCount: 0, hasMore: false, types: [], mailboxes: [] }
}
const runIds = [...new Set(page.map((row) => row.run_id))]
const dispatchByRun = new Map<string, DispatchContextRow>()
@@ -3723,6 +3797,7 @@ export class OrchestrationDb {
return {
routedCount: page.length,
hasMore: rows.length > ORCHESTRATION_DELIVERY_BATCH_LIMIT,
types: [...new Set(page.map((row) => row.type))],
mailboxes: [...byMailbox].map(([mailboxHandle, types]) => ({
mailboxHandle,
types: [...types]
@@ -3747,6 +3822,23 @@ export class OrchestrationDb {
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) {
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 delivery_contract = 'current_delivery'
AND id IN (${placeholders})`
)
.get(toHandle, ...batch) as { count: number }
matched += row.count
}
return matched === ids.length
}
getAllMessages(toHandle: string, limit = 20): MessageRow[] {
return exposeMessageListTimestamps(
this.db
@@ -3781,6 +3873,19 @@ export class OrchestrationDb {
}
}
markAsUndelivered(ids: string[]): void {
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(',')
this.db
.prepare(
`UPDATE messages SET delivered_at = NULL
WHERE read = 0 AND id IN (${placeholders})`
)
.run(...batch)
}
}
markAsReadAndDelivered(ids: string[]): void {
if (ids.length === 0) {
return
@@ -6896,21 +7001,28 @@ export class OrchestrationDb {
return undefined
}
const actives = this.db
const exactPane = this.db
.prepare(
`SELECT * FROM dispatch_contexts
WHERE assignee_pane_key = ? AND status IN ('pending', 'dispatched')
ORDER BY rowid DESC LIMIT 1`
)
.get(assigneePaneKey) as DispatchContextRow | undefined
if (exactPane) {
return exactPane
}
if (!parsePaneKey(assigneePaneKey)) {
return undefined
}
return this.db
.prepare(
`SELECT * FROM dispatch_contexts
WHERE assignee_pane_key IS NOT NULL
AND status IN ('pending', 'dispatched')
AND ${DISPATCH_PANE_KEY_MATCH_SUFFIX_SQL} = ?`
AND status IN ('pending', 'dispatched') AND instr(assignee_pane_key, ':') > 1
AND ${DISPATCH_PANE_KEY_MATCH_SUFFIX_SQL} = ?
ORDER BY rowid DESC LIMIT 1`
)
.all(paneKeyMatchSuffix(assigneePaneKey)) as DispatchContextRow[]
for (const row of actives) {
if (row.assignee_pane_key && isEquivalentPaneKey(row.assignee_pane_key, assigneePaneKey)) {
return row
}
}
return undefined
.get(paneKeyMatchSuffix(assigneePaneKey)) as DispatchContextRow | undefined
}
getLatestDispatchForTerminal(handle: string): DispatchContextRow | undefined {
src/main/runtime/rpc/methods/orchestration.ts
+105 -17
View File
@@ -1,5 +1,6 @@
/* eslint-disable max-lines -- Why: RPC method definitions co-locate param schemas with handlers; splitting by method would scatter the shared enums and Zod transforms without reducing complexity. */
import { z } from 'zod'
import { setImmediate as yieldToEventLoop } from 'node:timers/promises'
import { defineMethod, type RpcMethod } from '../core'
import { OptionalFiniteNumber, OptionalString, OptionalBoolean, requiredString } from '../schemas'
import {
@@ -43,6 +44,22 @@ const TASK_STATUSES: TaskStatus[] = [
'blocked'
]
async function routeAllDirectMessagePages(
routePage: () => { routedCount: number; hasMore: boolean },
signal?: AbortSignal
): Promise<void> {
while (true) {
if (signal?.aborted) {
throw new OrchestrationError('request_aborted', 'Mailbox routing was cancelled.')
}
const page = routePage()
if (!page.hasMore) {
return
}
await yieldToEventLoop()
}
}
function getLifecycleGroupRecipientError(type: 'worker_done' | 'heartbeat'): string {
return `${type} messages belong to one exact Dispatch and cannot target a group address.`
}
@@ -725,6 +742,16 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
const db = runtime.getOrchestrationDb()
const handle = params.terminal ?? 'unknown'
const typeFilter = parseMessageTypes(params.types)
const routeDirectSnapshot = async (
runId: string,
directHandle: string,
routePage: (throughSequence: number) => { routedCount: number; hasMore: boolean }
): Promise<void> => {
const throughSequence = db.getLatestUnreadDirectMessageSequenceForRun(runId, directHandle)
if (throughSequence !== undefined) {
await routeAllDirectMessagePages(() => routePage(throughSequence), signal)
}
}
// Why: a live runtime handle is authoritative; pane metadata is only the restart fallback.
const paneKey = runtime.getTerminalPaneKey(handle) ?? params.terminalPaneKey
@@ -741,13 +768,29 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
const generation = run.consumer_generation
const address = `run:${run.id}`
runtime.ensureOrchestrationFederationRelay(run.id)
const routedDirect = db.routeUnreadDirectMessagesToRunMailbox(run.id, handle)
if (
routedDirect.routedCount === 0 &&
run.coordinator_handle &&
run.coordinator_handle !== handle
) {
db.routeUnreadDirectMessagesToRunMailbox(run.id, run.coordinator_handle)
await routeDirectSnapshot(run.id, handle, (throughSequence) =>
db.routeUnreadDirectMessagesToRunMailbox(run.id, handle, throughSequence)
)
const coordinatorHandle = run.coordinator_handle
if (coordinatorHandle && coordinatorHandle !== handle) {
await routeDirectSnapshot(run.id, coordinatorHandle, (throughSequence) =>
db.routeUnreadDirectMessagesToRunMailbox(run.id, coordinatorHandle, throughSequence)
)
}
revalidateLegacyCoordinator?.()
const currentRun = resolveRunScope(runtime, {
runId: run.id,
callerTerminalHandle: handle,
callerPaneKey: paneKey ?? undefined,
requireCurrentConsumer: true,
legacyCoordinatorRunId,
callerEvidence: orchestrationCompatibilityEvidence
})
if (currentRun.consumer_generation !== generation) {
throw new OrchestrationError(
'consumer_fenced',
'This mailbox consumer was replaced while routing pending mail.'
)
}
const acknowledged = params.ack
@@ -917,24 +960,68 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
: undefined
if (workerMailbox) {
const address = `dispatch:${workerMailbox.dispatchId}`
if (activeDispatch) {
const routedDirect = db.routeUnreadDirectMessagesToDispatchMailbox(
activeDispatch.id,
activeDispatch.run_id,
handle
)
const revalidateWorkerMailbox = (): void => {
if (activeDispatch) {
const current = db.getActiveDispatchForIdentity(handle, paneKey ?? undefined)
if (current?.id === activeDispatch.id) {
return
}
} else if (remoteAttachment && paneKey) {
const current = db.findActiveRemoteAttachmentForPane(paneKey)
if (
current?.dispatch_id === remoteAttachment.dispatch_id &&
db.isRemoteAttachmentProcessCurrent({
dispatchId: current.dispatch_id,
paneKey,
processIncarnation: runtime.getTerminalProcessIncarnation(handle)
})
) {
return
}
}
const latestDispatch = db.getDispatchContextById(workerMailbox.dispatchId)
const owningRunId =
latestDispatch?.run_id ?? activeDispatch?.run_id ?? workerMailbox.runId
if (
routedDirect.routedCount === 0 &&
activeDispatch.assignee_handle &&
activeDispatch.assignee_handle !== handle
owningRunId &&
(!latestDispatch ||
(latestDispatch.status !== 'pending' && latestDispatch.status !== 'dispatched'))
) {
const routed = db.routeUnreadDispatchMailboxToRunMailbox(
workerMailbox.dispatchId,
owningRunId
)
for (const routedType of routed.types) {
runtime.notifyMessageArrived(`run:${owningRunId}`, routedType)
}
}
throw new OrchestrationError(
'dispatch_inactive',
`Dispatch ${workerMailbox.dispatchId} is no longer assigned to this worker.`
)
}
if (activeDispatch) {
await routeDirectSnapshot(activeDispatch.run_id, handle, (throughSequence) =>
db.routeUnreadDirectMessagesToDispatchMailbox(
activeDispatch.id,
activeDispatch.run_id,
activeDispatch.assignee_handle
handle,
throughSequence
)
)
const assigneeHandle = activeDispatch.assignee_handle
if (assigneeHandle && assigneeHandle !== handle) {
await routeDirectSnapshot(activeDispatch.run_id, assigneeHandle, (throughSequence) =>
db.routeUnreadDirectMessagesToDispatchMailbox(
activeDispatch.id,
activeDispatch.run_id,
assigneeHandle,
throughSequence
)
)
}
}
revalidateWorkerMailbox()
const showAll = params.all === true || (params.unread === false && params.peek !== true)
const messages = showAll
? db.getAllMessagesForHandle(address, 100, typeFilter)
@@ -958,6 +1045,7 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
timeoutMs: params.timeoutMs ?? undefined,
signal
})
revalidateWorkerMailbox()
if (waitResult === 'timed_out' || waitResult === 'cancelled') {
return {
...(workerMailbox.runId ? { runId: workerMailbox.runId } : {}),
src/main/runtime/terminal-send-stale-leaf-liveness.test.ts
+45 -8
View File
@@ -233,6 +233,7 @@ type StoredMessageRow = {
function makeOrchestrationDbStub(toHandle: () => string) {
const rows: StoredMessageRow[] = []
const runMailbox = 'run:run_test'
const markAsDelivered = vi.fn((ids: string[]) => {
for (const row of rows) {
if (ids.includes(row.id)) {
@@ -240,9 +241,18 @@ function makeOrchestrationDbStub(toHandle: () => string) {
}
}
})
const markAsUndelivered = vi.fn((ids: string[]) => {
for (const row of rows) {
if (ids.includes(row.id) && row.read === 0) {
row.delivered_at = null
}
}
})
return {
rows,
runMailbox,
markAsDelivered,
markAsUndelivered,
insert(subject: string, type: StoredMessageRow['type'] = 'status'): void {
rows.push({
id: `msg_${rows.length + 1}`,
@@ -268,10 +278,34 @@ function makeOrchestrationDbStub(toHandle: () => string) {
rows.filter((row) => row.to_handle === handle && row.read === 0 && !row.delivered_at),
getUndeliveredUnreadMailboxHandles: () => [toHandle()],
getActiveCoordinatorRun: () => null,
getCurrentRunForPane: () => undefined,
getCurrentRunForPane: () => ({ id: 'run_test' }),
getRun: () => ({ id: 'run_test', coordinator_handle: toHandle() }),
hasUndeliveredDirectMessageForRun: (runId: string, handle: string) =>
rows.some(
(row) =>
row.run_id === runId && row.to_handle === handle && row.read === 0 && !row.delivered_at
),
routeUnreadDirectMessagesToRunMailbox: (runId: string, handle: string) => {
const routed = rows.filter(
(row) => row.run_id === runId && row.to_handle === handle && row.read === 0
)
for (const row of routed) {
row.to_handle = runMailbox
}
return {
routedCount: routed.length,
hasMore: false,
types: [...new Set(routed.map((row) => row.type))]
}
},
areUnreadMessages: (handle: string, ids: string[]) =>
ids.every((id) =>
rows.some((row) => row.id === id && row.to_handle === handle && row.read === 0)
),
// Consulted by onPtyExit's dispatch-failure path.
getActiveDispatchForTerminal: () => null,
markAsDelivered,
markAsUndelivered,
close: () => {}
}
}
@@ -345,7 +379,7 @@ describe('push-on-idle orchestration delivery absence gate', () => {
const pulled: string[] = []
const checkResumed = runtime
.waitForMessage(handle, { typeFilter: ['worker_done'], timeoutMs: 60_000 })
.waitForMessage(stub.runMailbox, { typeFilter: ['worker_done'], timeoutMs: 60_000 })
.then(() => {
for (const row of stub.rows) {
if (row.type === 'worker_done' && row.read === 0) {
@@ -391,7 +425,7 @@ describe('push-on-idle orchestration delivery absence gate', () => {
})
})
const waitPromise = runtime.waitForMessage(handle, {
const waitPromise = runtime.waitForMessage(stub.runMailbox, {
typeFilter: ['worker_done'],
timeoutMs: 60_000
})
@@ -402,7 +436,7 @@ describe('push-on-idle orchestration delivery absence gate', () => {
// The reserving waiter goes away, then its type finally arrives — and the
// probe dedup drops this notify, so only the continuation can deliver it.
runtime.cancelMessageWaiters(handle)
runtime.cancelMessageWaiters(stub.runMailbox)
await expect(waitPromise).resolves.toBe('cancelled')
stub.insert('late completion', 'worker_done')
runtime.notifyMessageArrived(handle, 'worker_done')
@@ -570,7 +604,8 @@ describe('push-on-idle orchestration delivery absence gate', () => {
await vi.advanceTimersByTimeAsync(500)
expect(write.mock.calls.filter(([, data]) => data === '\r')).toHaveLength(0)
expect(stub.markAsDelivered).not.toHaveBeenCalled()
expect(stub.markAsDelivered).toHaveBeenCalledOnce()
expect(stub.markAsUndelivered).toHaveBeenCalledOnce()
expect(stub.rows[0].delivered_at).toBeNull()
// The replacement's own delivery starts a fresh flight and completes —
@@ -590,7 +625,7 @@ describe('push-on-idle orchestration delivery absence gate', () => {
expect(payloadWrites).toHaveLength(2)
await vi.advanceTimersByTimeAsync(500)
expect(write.mock.calls.filter(([, data]) => data === '\r')).toHaveLength(1)
expect(stub.markAsDelivered).toHaveBeenCalledOnce()
expect(stub.markAsDelivered).toHaveBeenCalledTimes(2)
expect(stub.rows[0].delivered_at).toEqual(expect.any(String))
} finally {
vi.useRealTimers()
@@ -625,7 +660,8 @@ describe('push-on-idle orchestration delivery absence gate', () => {
await vi.advanceTimersByTimeAsync(500)
expect(write.mock.calls.filter(([, data]) => data === '\r')).toHaveLength(0)
expect(stub.markAsDelivered).not.toHaveBeenCalled()
expect(stub.markAsDelivered).toHaveBeenCalledOnce()
expect(stub.markAsUndelivered).toHaveBeenCalledOnce()
// No stray settle flushed the parked trigger into the dead pty.
expect(write).toHaveBeenCalledTimes(1)
expect(stub.rows.every((row) => row.delivered_at === null)).toBe(true)
@@ -655,7 +691,8 @@ describe('push-on-idle orchestration delivery absence gate', () => {
await vi.advanceTimersByTimeAsync(500)
expect(write.mock.calls.filter(([, data]) => data === '\r')).toHaveLength(0)
expect(stub.markAsDelivered).not.toHaveBeenCalled()
expect(stub.markAsDelivered).toHaveBeenCalledOnce()
expect(stub.markAsUndelivered).toHaveBeenCalledOnce()
expect(stub.rows[0].delivered_at).toBeNull()
} finally {
vi.useRealTimers()