mirror of
https://github.com/stablyai/orca.git
synced 2026-10-22 08:03:37 +00:00
fix(orchestration): preserve active sibling dispatches
This commit is contained in:
@@ -48,8 +48,8 @@ Slack, GitHub comments, or any other channel to reach a human during the run.
|
||||
# Ask the coordinator a question and block until it answers.
|
||||
#
|
||||
# BEHAVIOR RULE #1 (MUST NOT VIOLATE):
|
||||
# NEVER use AskUserQuestion; use \`orca orchestration ask\` or send
|
||||
# --type decision_gate. AskUserQuestion opens a local TUI prompt that the
|
||||
# NEVER use AskUserQuestion; use \`orca orchestration ask\`.
|
||||
# AskUserQuestion opens a local TUI prompt that the
|
||||
# coordinator cannot see and cannot answer — your session will hang forever
|
||||
# waiting on a human. Every interactive question goes through \`ask\` below.
|
||||
#
|
||||
|
||||
@@ -121,6 +121,76 @@ describe('Task/Dispatch lifecycle guards', () => {
|
||||
expectCapability(database, worker, false)
|
||||
})
|
||||
|
||||
it('keeps a Task dispatched when missing-terminal recovery leaves another worker active', () => {
|
||||
const database = createDatabase()
|
||||
const task = database.createTask({ spec: 'legacy missing-terminal split' })
|
||||
const missing = startWorker(database, task.id, 'missing')
|
||||
sqliteFor(database).prepare("UPDATE tasks SET status = 'ready' WHERE id = ?").run(task.id)
|
||||
const live = startWorker(database, task.id, 'live')
|
||||
|
||||
database.reconcileMissingWorkerTerminal(missing.dispatchId, 'terminal missing')
|
||||
|
||||
expect(database.getTask(task.id)?.status).toBe('dispatched')
|
||||
expect(database.getDispatchContextById(missing.dispatchId)).toMatchObject({
|
||||
status: 'failed',
|
||||
capability_revoked_at: expect.any(String)
|
||||
})
|
||||
expect(database.getWorkerDispatch(missing.dispatchId)?.state).toBe('abandoned')
|
||||
expect(database.getDispatchContextById(live.dispatchId)?.status).toBe('dispatched')
|
||||
expect(database.getWorkerDispatch(live.dispatchId)?.state).toBe('ready')
|
||||
expectCapability(database, missing, false)
|
||||
expectCapability(database, live, true)
|
||||
|
||||
database.reconcileMissingWorkerTerminal(live.dispatchId, 'second terminal missing')
|
||||
expect(database.getTask(task.id)?.status).toBe('ready')
|
||||
expectCapability(database, live, false)
|
||||
})
|
||||
|
||||
it.each(['local', 'federated'] as const)(
|
||||
'keeps a Task dispatched when a %s worker start fails beside a live worker',
|
||||
(kind) => {
|
||||
const database = createDatabase()
|
||||
const task = database.createTask({ spec: `${kind} split start failure` })
|
||||
const failed = database.createStartingWorkerDispatch({
|
||||
taskId: task.id,
|
||||
startOptions: {},
|
||||
...(kind === 'federated'
|
||||
? {
|
||||
federation: {
|
||||
environmentId: 'server-1',
|
||||
environmentName: 'worker server',
|
||||
peerFingerprint: 'peer-1',
|
||||
protocolVersion: 3
|
||||
}
|
||||
}
|
||||
: {})
|
||||
})
|
||||
sqliteFor(database).prepare("UPDATE tasks SET status = 'ready' WHERE id = ?").run(task.id)
|
||||
const live = startWorker(database, task.id, `${kind}_live`)
|
||||
|
||||
if (kind === 'local') {
|
||||
database.failWorkerStart(failed.dispatch.id, 'start_failed', 'worker failed to start')
|
||||
} else {
|
||||
database.reconcileFederatedWorkerStart({
|
||||
dispatchId: failed.dispatch.id,
|
||||
state: 'failed',
|
||||
stage: 'start_failed',
|
||||
lastError: 'worker failed to start'
|
||||
})
|
||||
}
|
||||
|
||||
expect(database.getTask(task.id)?.status).toBe('dispatched')
|
||||
expect(database.getDispatchContextById(failed.dispatch.id)).toMatchObject({
|
||||
status: 'failed',
|
||||
capability_revoked_at: expect.any(String)
|
||||
})
|
||||
expect(database.getWorkerDispatch(failed.dispatch.id)?.state).toBe('failed')
|
||||
expect(database.getDispatchContextById(live.dispatchId)?.status).toBe('dispatched')
|
||||
expect(database.getWorkerDispatch(live.dispatchId)?.state).toBe('ready')
|
||||
expectCapability(database, live, true)
|
||||
}
|
||||
)
|
||||
|
||||
it('rejects gate creation while a supervised worker remains active', () => {
|
||||
const database = createDatabase()
|
||||
const task = database.createTask({ spec: 'worker gate guard' })
|
||||
|
||||
@@ -5125,7 +5125,13 @@ export class OrchestrationDb {
|
||||
)
|
||||
.run(stage, reason, dispatchId)
|
||||
this.db
|
||||
.prepare("UPDATE tasks SET status = 'failed', completed_at = datetime('now') WHERE id = ?")
|
||||
.prepare(
|
||||
`UPDATE tasks SET status = 'failed', completed_at = datetime('now')
|
||||
WHERE id = ? AND NOT EXISTS (
|
||||
SELECT 1 FROM dispatch_contexts
|
||||
WHERE task_id = tasks.id AND status IN ('pending', 'dispatched')
|
||||
)`
|
||||
)
|
||||
.run(dispatch.task_id)
|
||||
this.closeQuestionsForDispatch(dispatchId)
|
||||
this.db.exec('COMMIT')
|
||||
@@ -5250,7 +5256,12 @@ export class OrchestrationDb {
|
||||
.run(reason, params.dispatchId)
|
||||
this.db
|
||||
.prepare(
|
||||
"UPDATE tasks SET status = 'failed', completed_at = datetime('now') WHERE id = ? AND status IN ('blocked', 'dispatched')"
|
||||
`UPDATE tasks SET status = 'failed', completed_at = datetime('now')
|
||||
WHERE id = ? AND status IN ('blocked', 'dispatched')
|
||||
AND NOT EXISTS (
|
||||
SELECT 1 FROM dispatch_contexts
|
||||
WHERE task_id = tasks.id AND status IN ('pending', 'dispatched')
|
||||
)`
|
||||
)
|
||||
.run(dispatch.task_id)
|
||||
this.closeQuestionsForDispatch(params.dispatchId)
|
||||
@@ -5317,7 +5328,11 @@ export class OrchestrationDb {
|
||||
.prepare(
|
||||
`UPDATE tasks
|
||||
SET status = ?, completed_at = CASE WHEN ? = 'failed' THEN datetime('now') ELSE NULL END
|
||||
WHERE id = ? AND status IN ('dispatched', 'blocked')`
|
||||
WHERE id = ? AND status IN ('dispatched', 'blocked')
|
||||
AND NOT EXISTS (
|
||||
SELECT 1 FROM dispatch_contexts
|
||||
WHERE task_id = tasks.id AND status IN ('pending', 'dispatched')
|
||||
)`
|
||||
)
|
||||
.run(taskStatus, taskStatus, dispatch.task_id)
|
||||
}
|
||||
|
||||
@@ -96,6 +96,7 @@ describe('buildDispatchPreamble', () => {
|
||||
expect(result).toMatch(/orchestration ask --from term_worker/)
|
||||
expect(result).toContain('--question')
|
||||
expect(result).toContain('--timeout-ms 600000')
|
||||
expect(result).not.toContain('--type decision_gate')
|
||||
// Why: the exact phrase is asserted so the rule can't be trimmed away by
|
||||
// accident. BEHAVIOR RULE #1 is the only place AskUserQuestion appears.
|
||||
expect(result).toContain('BEHAVIOR RULE #1')
|
||||
|
||||
@@ -104,8 +104,8 @@ Slack, GitHub comments, or any other channel to reach a human during the run.
|
||||
# Ask the coordinator a question and block until it answers.
|
||||
#
|
||||
# BEHAVIOR RULE #1 (MUST NOT VIOLATE):
|
||||
# NEVER use AskUserQuestion; use \`${cli} orchestration ask\` or send
|
||||
# --type decision_gate. AskUserQuestion opens a local TUI prompt that the
|
||||
# NEVER use AskUserQuestion; use \`${cli} orchestration ask\`.
|
||||
# AskUserQuestion opens a local TUI prompt that the
|
||||
# coordinator cannot see and cannot answer — your session will hang forever
|
||||
# waiting on a human. Every interactive question goes through \`ask\` below.
|
||||
#
|
||||
|
||||
Reference in New Issue
Block a user