mirror of
https://github.com/stablyai/orca.git
synced 2026-10-31 00:57:45 +00:00
fix(orchestration): keep a settled worker report when startup fails late (#26762)
When a worker reported its outcome (succeeded or failed) before the worker-start path finished, a late startup error such as a rejected prompt delivery overwrote that settled outcome with a start failure and tore down the live worker session. The start-failure path now checks for a settled report first and returns a ready receipt carrying the reported workerOutcome without teardown; otherwise it records the failure before awaiting teardown so no report can slip in between. Co-authored-by: Neil <neil@stably.ai>
This commit is contained in:
co-authored by
Neil
parent
232ca39297
commit
8a42aa87c6
@@ -30,12 +30,6 @@ vi.mock('./orchestration/worker/worker-setup-gate', () => ({
|
||||
persistWorkerReadinessStage: () => {},
|
||||
persistWorkerSetupWaitOutcome: () => {}
|
||||
}))
|
||||
vi.mock('./orchestration/worker/worker-start-receipt', () => ({
|
||||
failWorkerStartWithReceipt: (args: { failedStage: string }) => ({
|
||||
state: 'failed',
|
||||
stage: args.failedStage
|
||||
})
|
||||
}))
|
||||
vi.mock('./orchestration/runs/dispatch-creator', () => ({
|
||||
resolveDispatchCreator: () => ({ kind: 'terminal', handle: 'term_c' })
|
||||
}))
|
||||
@@ -97,14 +91,22 @@ function fakes() {
|
||||
getTerminalPaneKey: vi.fn(() => 'pane_1'),
|
||||
retireStructuredAgentSessionTabFromSnapshot
|
||||
} as unknown as OrcaRuntimeService
|
||||
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the start path calls only these three methods.
|
||||
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the failing start path calls only these methods.
|
||||
const db = {
|
||||
createStartingWorkerDispatch: () => ({
|
||||
dispatch: { id: 'd_fail', depth: 0 },
|
||||
task: { id: 't1', spec: 'do the thing' }
|
||||
}),
|
||||
recordWorkerStage: () => {},
|
||||
prepareStartingWorkerAuthority: () => {}
|
||||
prepareStartingWorkerAuthority: () => {},
|
||||
getWorkerDispatch: () => undefined,
|
||||
failWorkerStart: (_dispatchId: string, stage: string) => ({
|
||||
state: 'failed',
|
||||
stage,
|
||||
effects: '[]',
|
||||
residual_resources: '[]'
|
||||
}),
|
||||
getWorkerTerminalResourceByOwner: () => undefined
|
||||
} as unknown as OrchestrationDb
|
||||
return { runtime, db, retireStructuredAgentSessionTabFromSnapshot }
|
||||
}
|
||||
|
||||
@@ -30,7 +30,6 @@ import { assertExplicitWorkerUsable } from './explicit-worker-terminal-validatio
|
||||
import { resolveDispatchAssigneeParty } from '../../../../orchestration/orchestration-party'
|
||||
import { CHAT_WORKER_AUTHORITY, chatAssigneeOf, chatWorkerMode } from '../chat-assignee-admission'
|
||||
import { recordCreatedWorkerTerminalCustody } from './created-worker-terminal-custody'
|
||||
import { tearDownFailedWorkerStart } from './failed-worker-start-teardown'
|
||||
import { releaseRetriedWorkerTerminal } from './retried-worker-terminal-release'
|
||||
import { requireWorkerAuthority, type WorkerEffect } from './worker-topology'
|
||||
import { prepareLocalWorkerStart } from './worker-start-validation'
|
||||
@@ -293,12 +292,9 @@ export async function startLocalWorker(args: {
|
||||
}
|
||||
})
|
||||
} catch (error) {
|
||||
await tearDownFailedWorkerStart({
|
||||
return await failWorkerStartWithReceipt({
|
||||
runtime,
|
||||
structuredSession: placed?.structuredSession ?? null,
|
||||
dispatchId: started.dispatch.id
|
||||
})
|
||||
return failWorkerStartWithReceipt({
|
||||
db,
|
||||
runId: run.id,
|
||||
taskId: task.id,
|
||||
|
||||
+43
@@ -13,6 +13,7 @@ import { OrchestrationDb } from '../../../../orchestration/db'
|
||||
import type { RpcRequest } from '../../../core'
|
||||
import { RpcDispatcher } from '../../../dispatcher'
|
||||
import { ORCHESTRATION_METHODS } from '../../orchestration'
|
||||
import * as FailedStartTeardown from './failed-worker-start-teardown'
|
||||
|
||||
vi.mock('../../../../../git/worktree', () => ({
|
||||
listWorktrees: vi.fn().mockResolvedValue([
|
||||
@@ -182,6 +183,7 @@ describe('orchestration worker-start prompt contract', () => {
|
||||
rmSync(root, { recursive: true, force: true })
|
||||
}
|
||||
vi.useRealTimers()
|
||||
vi.restoreAllMocks()
|
||||
})
|
||||
|
||||
it('durably accepts exactly one submitted and started turn', async () => {
|
||||
@@ -265,6 +267,47 @@ describe('orchestration worker-start prompt contract', () => {
|
||||
)
|
||||
})
|
||||
|
||||
it.each([
|
||||
['succeeded', false],
|
||||
['failed', true]
|
||||
] as const)(
|
||||
'preserves an early %s report when prompt delivery then rejects (unknown=%s)',
|
||||
async (outcome, unknown) => {
|
||||
vi.useFakeTimers()
|
||||
const teardown = vi.spyOn(FailedStartTeardown, 'tearDownFailedWorkerStart')
|
||||
const harness = await createPromptContractHarness('swallowed')
|
||||
vi.spyOn(harness.runtime, 'sendTerminalAgentPrompt').mockImplementation(async () => {
|
||||
const dispatch = harness.db.findActiveDispatchForAssignee(harness.handle)
|
||||
harness.db.settleWorkerReport({
|
||||
taskId: harness.taskId,
|
||||
dispatchId: dispatch!.id,
|
||||
outcome,
|
||||
result: 'Finished before the prompt receipt'
|
||||
})
|
||||
throw Object.assign(
|
||||
new Error('terminal_handle_stale'),
|
||||
unknown ? { code: 'operation_unknown' } : {}
|
||||
)
|
||||
})
|
||||
const pending = harness.dispatcher.dispatch(harness.request)
|
||||
await vi.runAllTimersAsync()
|
||||
const response = await pending
|
||||
expect(response).toMatchObject({
|
||||
ok: true,
|
||||
result: { state: 'ready', stage: 'settled', workerOutcome: outcome }
|
||||
})
|
||||
expect(response.ok && response.result).not.toHaveProperty('lastError')
|
||||
expect(teardown).not.toHaveBeenCalled()
|
||||
expect(harness.db.getTask(harness.taskId)?.status).toBe(
|
||||
outcome === 'succeeded' ? 'completed' : 'failed'
|
||||
)
|
||||
expect(await harness.dispatcher.dispatch({ ...harness.request, id: 'retry' })).toMatchObject({
|
||||
ok: true,
|
||||
result: { state: 'ready', workerOutcome: outcome, mutation: { replayed: true } }
|
||||
})
|
||||
}
|
||||
)
|
||||
|
||||
it('retains accepted authority when the observation binding becomes stale', async () => {
|
||||
vi.useFakeTimers()
|
||||
const harness = await createPromptContractHarness('swallowed')
|
||||
|
||||
+2
-5
@@ -7,6 +7,7 @@ import type { OrchestrationDb } from '../../../../orchestration/db'
|
||||
import type { RunRow, TaskRow } from '../../../../orchestration/types'
|
||||
import type { WorkerStartModeReceipt } from '../../orchestration-worker-start-mode'
|
||||
import { deliverWorkerDispatchPreamble } from './deliver-worker-dispatch-preamble'
|
||||
import { reportedWorkerOutcome } from './worker-start-receipt'
|
||||
import { chatAssigneeSessionId } from '../../../../orchestration/chat-assignee'
|
||||
import type { OrchestrationWorkerLaunchReceipt } from './worker-launch-preferences'
|
||||
import {
|
||||
@@ -147,11 +148,7 @@ export async function deliverAndSettleWorkerStartReadiness(args: {
|
||||
const worker = alreadySettled
|
||||
? currentWorker
|
||||
: db.markWorkerDispatchReady(args.dispatchId, effects)
|
||||
// A completed task proves start succeeded; older callers use only 'ready' as start success.
|
||||
const reportedOutcome =
|
||||
worker.stage === 'settled' && (worker.state === 'succeeded' || worker.state === 'failed')
|
||||
? worker.state
|
||||
: undefined
|
||||
const reportedOutcome = reportedWorkerOutcome(worker)
|
||||
return {
|
||||
runId: run.id,
|
||||
taskId: task.id,
|
||||
|
||||
@@ -1,10 +1,19 @@
|
||||
import type { OrchestrationDb } from '../../../../orchestration/db'
|
||||
import { isUnknownWorkerStartOutcome, type WorkerSetupReceipt } from './worker-topology'
|
||||
import type { WorkerDispatchRow } from '../../../../orchestration/types'
|
||||
import type { OrcaRuntimeService } from '../../../../orca-runtime'
|
||||
import { tearDownFailedWorkerStart } from './failed-worker-start-teardown'
|
||||
import {
|
||||
isUnknownWorkerStartOutcome,
|
||||
type createStructuredWorkerSessionForWorktree,
|
||||
type WorkerSetupReceipt
|
||||
} from './worker-topology'
|
||||
import type { OrchestrationWorkerLaunchReceipt } from './worker-launch-preferences'
|
||||
import type { WorkerStartModeReceipt } from '../../orchestration-worker-start-mode'
|
||||
import { isStructuredWorkerHandle } from '../../../../../../shared/structured-worker-handle'
|
||||
|
||||
export function failWorkerStartWithReceipt(args: {
|
||||
type WorkerStartFailureArgs = {
|
||||
runtime: OrcaRuntimeService
|
||||
structuredSession: Awaited<ReturnType<typeof createStructuredWorkerSessionForWorktree>> | null
|
||||
db: OrchestrationDb
|
||||
runId: string
|
||||
taskId: string
|
||||
@@ -14,7 +23,57 @@ export function failWorkerStartWithReceipt(args: {
|
||||
setup: WorkerSetupReceipt
|
||||
launch: OrchestrationWorkerLaunchReceipt
|
||||
mode: WorkerStartModeReceipt
|
||||
}): unknown {
|
||||
}
|
||||
|
||||
/** A worker report settles the Task; that proves the start, so older callers still see 'ready'. */
|
||||
export function reportedWorkerOutcome(
|
||||
worker: WorkerDispatchRow | undefined
|
||||
): 'succeeded' | 'failed' | undefined {
|
||||
return worker?.stage === 'settled' && (worker.state === 'succeeded' || worker.state === 'failed')
|
||||
? worker.state
|
||||
: undefined
|
||||
}
|
||||
|
||||
function settledWorkerStartReceipt(args: WorkerStartFailureArgs): unknown {
|
||||
const worker = args.db.getWorkerDispatch(args.dispatchId)
|
||||
const workerOutcome = reportedWorkerOutcome(worker)
|
||||
if (!worker || !workerOutcome) {
|
||||
return undefined
|
||||
}
|
||||
return {
|
||||
runId: args.runId,
|
||||
taskId: args.taskId,
|
||||
dispatchId: args.dispatchId,
|
||||
state: 'ready',
|
||||
stage: worker.stage,
|
||||
workerOutcome,
|
||||
setup: args.setup,
|
||||
launch: args.launch,
|
||||
mode: args.mode,
|
||||
effects: JSON.parse(worker.effects),
|
||||
residualResources: []
|
||||
}
|
||||
}
|
||||
|
||||
export async function failWorkerStartWithReceipt(args: WorkerStartFailureArgs): Promise<unknown> {
|
||||
// A report that settled the Task outranks a late startup error, and keeps its session alive.
|
||||
const settled = settledWorkerStartReceipt(args)
|
||||
if (settled) {
|
||||
return settled
|
||||
}
|
||||
try {
|
||||
// Fail before awaiting teardown, so no report can settle between that check and this transition.
|
||||
return failedWorkerStartReceipt(args)
|
||||
} finally {
|
||||
await tearDownFailedWorkerStart({
|
||||
runtime: args.runtime,
|
||||
structuredSession: args.structuredSession,
|
||||
dispatchId: args.dispatchId
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
function failedWorkerStartReceipt(args: WorkerStartFailureArgs): unknown {
|
||||
const reason = args.error instanceof Error ? args.error.message : String(args.error)
|
||||
const unknown = isUnknownWorkerStartOutcome(args.error, args.failedStage)
|
||||
const worker = unknown
|
||||
|
||||
Reference in New Issue
Block a user