mirror of
https://github.com/stablyai/orca.git
synced 2026-10-23 00:03:52 +00:00
test(orchestration): close worker-exit escalation coverage gaps from review
Review of #15235 found the suite could go green on real regressions. A mutation sweep over the fix now kills all seven mutations it was checked against: - The check --wait test was self-fulfilling: waitForMessage resolves off the notification alone and never reads the row, so a notify with no message behind it passed. It now asserts the message is actually in the Run mailbox. - The oracle graded counts only, so corrupted escalation content stayed green. It now pins from/to/run_id/type/priority/subject/delivery_contract/payload. - Nothing proved routing follows the dying worker's own Run. A second, newer Run is now live in the isolation test, which is what any "latest active run" lookup would wrongly pick — the exact shape of the bug being fixed. - Addressing the coordinator handle instead of run:<id> survived every test, because trg_messages_route_coordinator_mail reroutes handle-addressed mail with a run_id. The two only diverge once the Run has no bound coordinator, so an unbound-Run case now covers it. - Added supervised worker_dispatches, circuit-breaker, and missing-owning-Run fallback cases, none of which were exercised. Also carry dispatchId in the escalation payload: applyEscalationToDispatch rejects an escalation lacking an exact Dispatch binding, and a coordinator reading the message needs to know which Dispatch died.
This commit is contained in:
@@ -16844,8 +16844,11 @@ export class OrcaRuntimeService {
|
||||
subject: `Agent exited unexpectedly (code ${exitCode})`,
|
||||
type: 'escalation',
|
||||
priority: 'high',
|
||||
// Why: applyEscalationToDispatch rejects an escalation without an exact Dispatch
|
||||
// binding, and a coordinator reading this needs to know which Dispatch died.
|
||||
payload: JSON.stringify({
|
||||
taskId: dispatch.task_id,
|
||||
dispatchId: dispatch.id,
|
||||
exitCode,
|
||||
handle
|
||||
}),
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import { OrcaRuntimeService } from '../orca-runtime'
|
||||
import { LEGACY_RUN_ID, OrchestrationDb } from './db'
|
||||
import { DISPATCH_CIRCUIT_BREAK_FAILURES } from './db/dispatch-context/dispatch-circuit-breaker'
|
||||
import { makePaneKey } from '../../../shared/stable-pane-id'
|
||||
import { getDefaultWorkspaceSession } from '../../../shared/constants'
|
||||
import type { WorkspaceSessionState } from '../../../shared/workspace-session-state-types'
|
||||
@@ -109,6 +110,9 @@ async function gradeWorkerExit(
|
||||
escalationsToCoordinatorHandle: number
|
||||
escalationsInRunMailbox: number
|
||||
totalUnreadForCoordinator: number
|
||||
escalation: Record<string, unknown> | null
|
||||
workerHandle: string
|
||||
runId: string
|
||||
}> {
|
||||
const { runtime, workerHandle, coordinatorHandle } = makeRuntimeWithTwoPanes()
|
||||
const db = new OrchestrationDb(':memory:')
|
||||
@@ -131,14 +135,31 @@ async function gradeWorkerExit(
|
||||
runtime.onPtyExit(WORKER_PTY_ID, 137)
|
||||
await settle()
|
||||
|
||||
const direct = db.getUnreadMessages(coordinatorHandle, ['escalation'])
|
||||
const runMailbox = db.getUnreadRunMailbox(runId, 100, ['escalation'])
|
||||
const escalation = runMailbox[0] ?? direct[0]
|
||||
return {
|
||||
dispatchStatus: db.getDispatchContextById(dispatch.id)?.status,
|
||||
escalationsToCoordinatorHandle: db.getUnreadMessages(coordinatorHandle, ['escalation'])
|
||||
.length,
|
||||
escalationsInRunMailbox: db.getUnreadRunMailbox(runId, 100, ['escalation']).length,
|
||||
escalationsToCoordinatorHandle: direct.length,
|
||||
escalationsInRunMailbox: runMailbox.length,
|
||||
// The ticket's literal claim: the coordinator receives no message of any kind.
|
||||
totalUnreadForCoordinator:
|
||||
db.getUnreadMessages(coordinatorHandle).length + db.getUnreadRunMailbox(runId, 100).length
|
||||
db.getUnreadMessages(coordinatorHandle).length + db.getUnreadRunMailbox(runId, 100).length,
|
||||
// Counts alone stay green while the escalation's content rots.
|
||||
escalation: escalation
|
||||
? {
|
||||
from: escalation.from_handle,
|
||||
to: escalation.to_handle,
|
||||
runId: escalation.run_id,
|
||||
type: escalation.type,
|
||||
priority: escalation.priority,
|
||||
subject: escalation.subject,
|
||||
deliveryContract: escalation.delivery_contract,
|
||||
payload: JSON.parse(escalation.payload ?? 'null')
|
||||
}
|
||||
: null,
|
||||
workerHandle,
|
||||
runId
|
||||
}
|
||||
} finally {
|
||||
db.close()
|
||||
@@ -147,30 +168,54 @@ async function gradeWorkerExit(
|
||||
|
||||
describe('STA-4604 worker PTY exit escalation reaches the coordinator', () => {
|
||||
it('delivers the escalation to a lightweight Run mailbox', async () => {
|
||||
await expect(gradeWorkerExit('lightweight-run')).resolves.toEqual({
|
||||
const graded = await gradeWorkerExit('lightweight-run')
|
||||
expect(graded).toMatchObject({
|
||||
dispatchStatus: 'failed',
|
||||
// Addressed run:<id> — the only address getOrCreateRunDelivery / getUnreadRunMailbox read.
|
||||
escalationsToCoordinatorHandle: 0,
|
||||
escalationsInRunMailbox: 1,
|
||||
totalUnreadForCoordinator: 1
|
||||
})
|
||||
expect(graded.escalation).toEqual({
|
||||
from: graded.workerHandle,
|
||||
to: `run:${graded.runId}`,
|
||||
runId: graded.runId,
|
||||
type: 'escalation',
|
||||
priority: 'high',
|
||||
subject: 'Agent exited unexpectedly (code 137)',
|
||||
deliveryContract: 'current_delivery',
|
||||
payload: {
|
||||
taskId: expect.any(String),
|
||||
dispatchId: expect.any(String),
|
||||
exitCode: 137,
|
||||
handle: graded.workerHandle
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
it('keeps legacy coordinator_runs delivery on the coordinator handle', async () => {
|
||||
await expect(gradeWorkerExit('legacy-run')).resolves.toEqual({
|
||||
const graded = await gradeWorkerExit('legacy-run')
|
||||
expect(graded).toMatchObject({
|
||||
dispatchStatus: 'failed',
|
||||
escalationsToCoordinatorHandle: 1,
|
||||
escalationsInRunMailbox: 0,
|
||||
totalUnreadForCoordinator: 1
|
||||
})
|
||||
expect(graded.escalation).toMatchObject({
|
||||
to: expect.stringMatching(/^term_/),
|
||||
runId: LEGACY_RUN_ID,
|
||||
type: 'escalation',
|
||||
subject: 'Agent exited unexpectedly (code 137)'
|
||||
})
|
||||
})
|
||||
|
||||
it('stays silent for a legacy dispatch with no active coordinator', async () => {
|
||||
await expect(gradeWorkerExit('legacy-run-no-coordinator')).resolves.toEqual({
|
||||
await expect(gradeWorkerExit('legacy-run-no-coordinator')).resolves.toMatchObject({
|
||||
dispatchStatus: 'failed',
|
||||
escalationsToCoordinatorHandle: 0,
|
||||
escalationsInRunMailbox: 0,
|
||||
totalUnreadForCoordinator: 0
|
||||
totalUnreadForCoordinator: 0,
|
||||
escalation: null
|
||||
})
|
||||
})
|
||||
|
||||
@@ -194,9 +239,182 @@ describe('STA-4604 worker PTY exit escalation reaches the coordinator', () => {
|
||||
runtime.onPtyExit(WORKER_PTY_ID, 137)
|
||||
|
||||
await expect(waiting).resolves.toBe('notified')
|
||||
// Why: waitForMessage resolves off the notification alone, so without this the test
|
||||
// would pass on a notify with no message behind it.
|
||||
expect(db.getUnreadRunMailbox(run.id, 100, ['escalation'])).toEqual([
|
||||
expect.objectContaining({
|
||||
run_id: run.id,
|
||||
to_handle: `run:${run.id}`,
|
||||
from_handle: workerHandle,
|
||||
type: 'escalation'
|
||||
})
|
||||
])
|
||||
await settle()
|
||||
} finally {
|
||||
db.close()
|
||||
}
|
||||
})
|
||||
|
||||
it('escalates only into the dying worker own Run when several Runs are live', async () => {
|
||||
const { runtime, workerHandle, coordinatorHandle } = makeRuntimeWithTwoPanes()
|
||||
const db = new OrchestrationDb(':memory:')
|
||||
try {
|
||||
// A second, more recently created Run is the trap: any "latest active run" lookup picks it.
|
||||
const ownRun = db.createRun({
|
||||
objective: 'run that owns the dying worker',
|
||||
coordinatorHandle,
|
||||
coordinatorPaneKey: COORDINATOR_PANE_KEY
|
||||
})
|
||||
const otherRun = db.createRun({
|
||||
objective: 'unrelated newer run',
|
||||
coordinatorHandle: 'term_other_coordinator',
|
||||
coordinatorPaneKey: makePaneKey('tab-other', '33333333-3333-4333-8333-333333333333')
|
||||
})
|
||||
const task = db.createTask({ spec: 'owned work', runId: ownRun.id })
|
||||
db.createDispatchContext(task.id, workerHandle, WORKER_PANE_KEY)
|
||||
runtime.setOrchestrationDb(db as never)
|
||||
|
||||
runtime.onPtyExit(WORKER_PTY_ID, 137)
|
||||
await settle()
|
||||
|
||||
expect({
|
||||
own: db.getUnreadRunMailbox(ownRun.id, 100, ['escalation']).length,
|
||||
other: db.getUnreadRunMailbox(otherRun.id, 100, ['escalation']).length
|
||||
}).toEqual({ own: 1, other: 0 })
|
||||
} finally {
|
||||
db.close()
|
||||
}
|
||||
})
|
||||
|
||||
it('escalates for a supervised worker and settles its worker_dispatches row', async () => {
|
||||
const { runtime, workerHandle, coordinatorHandle } = makeRuntimeWithTwoPanes()
|
||||
const db = new OrchestrationDb(':memory:')
|
||||
try {
|
||||
const run = db.createRun({
|
||||
objective: 'supervised worker run',
|
||||
coordinatorHandle,
|
||||
coordinatorPaneKey: COORDINATOR_PANE_KEY
|
||||
})
|
||||
const task = db.createTask({ spec: 'supervised work', runId: run.id })
|
||||
const started = db.createStartingWorkerDispatch({ taskId: task.id, startOptions: {} })
|
||||
db.prepareStartingWorkerAuthority({
|
||||
dispatchId: started.dispatch.id,
|
||||
handle: workerHandle,
|
||||
paneKey: WORKER_PANE_KEY,
|
||||
processIncarnation: 'inc-1',
|
||||
worktreeId: WORKTREE_ID,
|
||||
effects: [],
|
||||
setupState: 'skipped'
|
||||
})
|
||||
db.markWorkerDispatchReady(started.dispatch.id)
|
||||
runtime.setOrchestrationDb(db as never)
|
||||
|
||||
runtime.onPtyExit(WORKER_PTY_ID, 137)
|
||||
await settle()
|
||||
|
||||
expect({
|
||||
dispatch: db.getDispatchContextById(started.dispatch.id)?.status,
|
||||
workerState: db.getWorkerDispatch(started.dispatch.id)?.state,
|
||||
workerStage: db.getWorkerDispatch(started.dispatch.id)?.stage,
|
||||
escalations: db.getUnreadRunMailbox(run.id, 100, ['escalation']).length
|
||||
}).toEqual({
|
||||
dispatch: 'failed',
|
||||
workerState: 'failed',
|
||||
workerStage: 'process_exited',
|
||||
escalations: 1
|
||||
})
|
||||
} finally {
|
||||
db.close()
|
||||
}
|
||||
})
|
||||
|
||||
it('still escalates on the failure that breaks the circuit', async () => {
|
||||
const { runtime, workerHandle, coordinatorHandle } = makeRuntimeWithTwoPanes()
|
||||
const db = new OrchestrationDb(':memory:')
|
||||
try {
|
||||
const run = db.createRun({
|
||||
objective: 'circuit breaker run',
|
||||
coordinatorHandle,
|
||||
coordinatorPaneKey: COORDINATOR_PANE_KEY
|
||||
})
|
||||
const task = db.createTask({ spec: 'repeatedly failing work', runId: run.id })
|
||||
// Burn the breaker down to its last life so this exit is the one that trips it.
|
||||
for (let attempt = 1; attempt < DISPATCH_CIRCUIT_BREAK_FAILURES; attempt += 1) {
|
||||
const previous = db.createDispatchContext(task.id, workerHandle, WORKER_PANE_KEY)
|
||||
db.failDispatch(previous.id, `attempt ${attempt}`, { workerProcessExited: true })
|
||||
}
|
||||
const dispatch = db.createDispatchContext(task.id, workerHandle, WORKER_PANE_KEY)
|
||||
runtime.setOrchestrationDb(db as never)
|
||||
|
||||
runtime.onPtyExit(WORKER_PTY_ID, 137)
|
||||
await settle()
|
||||
|
||||
expect({
|
||||
dispatch: db.getDispatchContextById(dispatch.id)?.status,
|
||||
escalations: db.getUnreadRunMailbox(run.id, 100, ['escalation']).length
|
||||
}).toEqual({ dispatch: 'circuit_broken', escalations: 1 })
|
||||
} finally {
|
||||
db.close()
|
||||
}
|
||||
})
|
||||
|
||||
it('falls back to the legacy gate when the dispatch owning Run row is gone', async () => {
|
||||
const { runtime, workerHandle, coordinatorHandle } = makeRuntimeWithTwoPanes()
|
||||
const insertMessage = vi.fn((message: { to: string }) => ({
|
||||
...message,
|
||||
to_handle: message.to,
|
||||
type: 'escalation'
|
||||
}))
|
||||
runtime.setOrchestrationDb({
|
||||
getActiveDispatchForTerminal: (handle: string) =>
|
||||
handle === workerHandle
|
||||
? { id: 'ctx-orphan', run_id: 'run-that-no-longer-exists', task_id: 'task-orphan' }
|
||||
: undefined,
|
||||
failDispatch: vi.fn(),
|
||||
getRun: () => undefined,
|
||||
getActiveCoordinatorRun: () => ({ id: 'run-legacy', coordinator_handle: coordinatorHandle }),
|
||||
insertMessage
|
||||
} as never)
|
||||
|
||||
runtime.onPtyExit(WORKER_PTY_ID, 137)
|
||||
|
||||
expect(insertMessage).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ to: coordinatorHandle, type: 'escalation' })
|
||||
)
|
||||
// An orphaned dispatch has no Run mailbox to address, so it must not invent one.
|
||||
expect(insertMessage.mock.calls[0]?.[0]).not.toHaveProperty('runId')
|
||||
})
|
||||
|
||||
it('still reaches the Run mailbox when the Run has no bound coordinator', async () => {
|
||||
const { runtime, workerHandle, coordinatorHandle } = makeRuntimeWithTwoPanes()
|
||||
const db = new OrchestrationDb(':memory:')
|
||||
try {
|
||||
const run = db.createRun({
|
||||
objective: 'run whose coordinator later unbinds',
|
||||
coordinatorHandle,
|
||||
coordinatorPaneKey: COORDINATOR_PANE_KEY
|
||||
})
|
||||
const task = db.createTask({ spec: 'work outliving its coordinator', runId: run.id })
|
||||
db.createDispatchContext(task.id, workerHandle, WORKER_PANE_KEY)
|
||||
// Rebinding the pane to a newer Run clears the old Run's coordinator_handle.
|
||||
db.createRun({
|
||||
objective: 'newer run on the same coordinator pane',
|
||||
coordinatorHandle,
|
||||
coordinatorPaneKey: COORDINATOR_PANE_KEY
|
||||
})
|
||||
expect(db.getRun(run.id)?.coordinator_handle).toBeNull()
|
||||
runtime.setOrchestrationDb(db as never)
|
||||
|
||||
runtime.onPtyExit(WORKER_PTY_ID, 137)
|
||||
await settle()
|
||||
|
||||
// Why: with no coordinator handle to address, only the durable run:<id> mailbox
|
||||
// keeps the escalation reachable for whoever rebinds next.
|
||||
expect(db.getUnreadRunMailbox(run.id, 100, ['escalation'])).toEqual([
|
||||
expect.objectContaining({ to_handle: `run:${run.id}`, run_id: run.id })
|
||||
])
|
||||
} finally {
|
||||
db.close()
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user