mirror of
https://github.com/stablyai/orca.git
synced 2026-10-18 08:03:22 +00:00
fix(orchestration): let a null-id worker_done complete the sender's own in-flight dispatch
A worker_done arriving without taskId/dispatchId was dropped with a warning before the pane/handle ownership check, so a worker using a custom harness (not the stock preamble) left its task stuck in 'dispatched' forever (#7429). Reconcile now falls back to the sender's sole active dispatch, adopting it only when the sender owns exactly one and passes hasLifecycleAuthority. Foreign panes, absent dispatches, and ambiguous multi-dispatch senders stay ignored, so no message that is ignored today starts completing a task it does not own. Co-authored-by: Orca <help@stably.ai>
This commit is contained in:
@@ -694,6 +694,28 @@ export class OrchestrationDb {
|
||||
return this.findActiveDispatchForAssignee(handle)
|
||||
}
|
||||
|
||||
// Why: attributing an id-less lifecycle message is only safe when the sender owns
|
||||
// exactly one active dispatch (#7429) — more than one match must not be guessed.
|
||||
getSoleActiveDispatchForSender(
|
||||
assigneeHandle: string,
|
||||
assigneePaneKey?: string | null
|
||||
): DispatchContextRow | undefined {
|
||||
const matches = (
|
||||
this.db
|
||||
.prepare("SELECT * FROM dispatch_contexts WHERE status IN ('pending', 'dispatched')")
|
||||
.all() as DispatchContextRow[]
|
||||
).filter(
|
||||
(row) =>
|
||||
row.assignee_handle === assigneeHandle ||
|
||||
Boolean(
|
||||
assigneePaneKey &&
|
||||
row.assignee_pane_key &&
|
||||
isEquivalentPaneKey(row.assignee_pane_key, assigneePaneKey)
|
||||
)
|
||||
)
|
||||
return matches.length === 1 ? matches[0] : undefined
|
||||
}
|
||||
|
||||
private findActiveDispatchForAssignee(
|
||||
assigneeHandle: string,
|
||||
assigneePaneKey?: string
|
||||
|
||||
@@ -409,4 +409,87 @@ describe('lifecycle reconciliation', () => {
|
||||
expect(db.getMessageById(lateHeartbeat.id)).toMatchObject({ read: 1 })
|
||||
expect(db.getMessageById(lateHeartbeat.id)?.delivered_at).not.toBeNull()
|
||||
})
|
||||
|
||||
it('completes a null-id worker_done sent from the pane that owns the sole dispatch', () => {
|
||||
db = new OrchestrationDb(':memory:')
|
||||
const task = db.createTask({ spec: 'work' })
|
||||
const dispatch = db.createDispatchContext(task.id, 'term_worker', `tab_w:${LEAF_A}`)
|
||||
const done = db.insertMessage({
|
||||
from: 'term_worker',
|
||||
to: 'term_coordinator',
|
||||
subject: 'Done',
|
||||
type: 'worker_done',
|
||||
payload: JSON.stringify({}),
|
||||
senderPaneKey: `tab_w:${LEAF_A}`
|
||||
})
|
||||
|
||||
expect(reconcileLifecycleMessage(db, done)).toEqual({
|
||||
action: 'completed',
|
||||
taskId: task.id,
|
||||
dispatchId: dispatch.id
|
||||
})
|
||||
expect(db.getTask(task.id)?.status).toBe('completed')
|
||||
expect(db.getDispatchContextById(dispatch.id)?.status).toBe('completed')
|
||||
})
|
||||
|
||||
it('ignores a null-id worker_done sent from a foreign pane and handle', () => {
|
||||
db = new OrchestrationDb(':memory:')
|
||||
const task = db.createTask({ spec: 'work' })
|
||||
const dispatch = db.createDispatchContext(task.id, 'term_owner', `tab_w:${LEAF_A}`)
|
||||
const logs: string[] = []
|
||||
const done = db.insertMessage({
|
||||
from: 'term_stranger',
|
||||
to: 'term_coordinator',
|
||||
subject: 'Done',
|
||||
type: 'worker_done',
|
||||
payload: JSON.stringify({}),
|
||||
senderPaneKey: `tab_w:${LEAF_B}`
|
||||
})
|
||||
|
||||
expect(reconcileLifecycleMessage(db, done, (line) => logs.push(line))).toEqual({
|
||||
action: 'ignored'
|
||||
})
|
||||
expect(db.getTask(task.id)?.status).toBe('dispatched')
|
||||
expect(db.getDispatchContextById(dispatch.id)?.status).toBe('dispatched')
|
||||
expect(logs.some((line) => line.includes('without taskId'))).toBe(true)
|
||||
})
|
||||
|
||||
it('ignores a null-id worker_done when the sender has no active dispatch', () => {
|
||||
db = new OrchestrationDb(':memory:')
|
||||
const task = db.createTask({ spec: 'work' })
|
||||
const dispatch = db.createDispatchContext(task.id, 'term_worker', `tab_w:${LEAF_A}`)
|
||||
db.updateTaskStatus(task.id, 'completed')
|
||||
const done = db.insertMessage({
|
||||
from: 'term_worker',
|
||||
to: 'term_coordinator',
|
||||
subject: 'Done again',
|
||||
type: 'worker_done',
|
||||
payload: JSON.stringify({}),
|
||||
senderPaneKey: `tab_w:${LEAF_A}`
|
||||
})
|
||||
|
||||
expect(reconcileLifecycleMessage(db, done)).toEqual({ action: 'ignored' })
|
||||
expect(db.getDispatchContextById(dispatch.id)?.status).toBe('completed')
|
||||
})
|
||||
|
||||
it('ignores a null-id worker_done when the sender matches two active dispatches', () => {
|
||||
db = new OrchestrationDb(':memory:')
|
||||
const ownTask = db.createTask({ spec: 'work' })
|
||||
const ownDispatch = db.createDispatchContext(ownTask.id, 'term_worker', `tab_w:${LEAF_A}`)
|
||||
const paneTask = db.createTask({ spec: 'other work' })
|
||||
const paneDispatch = db.createDispatchContext(paneTask.id, 'term_other', `tab_w:${LEAF_B}`)
|
||||
// Sender matches one dispatch by handle and another by pane, so neither can be attributed.
|
||||
const done = db.insertMessage({
|
||||
from: 'term_worker',
|
||||
to: 'term_coordinator',
|
||||
subject: 'Done',
|
||||
type: 'worker_done',
|
||||
payload: JSON.stringify({}),
|
||||
senderPaneKey: `tab_w:${LEAF_B}`
|
||||
})
|
||||
|
||||
expect(reconcileLifecycleMessage(db, done)).toEqual({ action: 'ignored' })
|
||||
expect(db.getDispatchContextById(ownDispatch.id)?.status).toBe('dispatched')
|
||||
expect(db.getDispatchContextById(paneDispatch.id)?.status).toBe('dispatched')
|
||||
})
|
||||
})
|
||||
|
||||
@@ -170,17 +170,11 @@ function reconcileWorkerDoneMessage(
|
||||
return persistedRejection
|
||||
}
|
||||
|
||||
const taskId = payload.taskId
|
||||
if (typeof taskId !== 'string' || taskId.length === 0) {
|
||||
onLog(`Warning: worker_done without taskId from ${msg.from_handle}`)
|
||||
return { action: 'ignored' }
|
||||
}
|
||||
|
||||
const dispatchId = payload.dispatchId
|
||||
if (typeof dispatchId !== 'string' || dispatchId.length === 0) {
|
||||
onLog(`Warning: worker_done without dispatchId from ${msg.from_handle}`)
|
||||
const target = resolveWorkerDoneTarget(db, msg, payload, onLog)
|
||||
if (!target) {
|
||||
return { action: 'ignored' }
|
||||
}
|
||||
const { taskId, dispatchId } = target
|
||||
|
||||
const task = db.getTask(taskId)
|
||||
if (!task) {
|
||||
@@ -239,6 +233,36 @@ function reconcileWorkerDoneMessage(
|
||||
return { action: 'completed', taskId, dispatchId }
|
||||
}
|
||||
|
||||
// Why: a worker_done from the pane that owns the sole in-flight dispatch is still
|
||||
// authoritative (#7429); the payload ids are a convenience, not the authority.
|
||||
function resolveWorkerDoneTarget(
|
||||
db: OrchestrationDb,
|
||||
msg: MessageRow,
|
||||
payload: Record<string, unknown>,
|
||||
onLog: LogFn
|
||||
): { taskId: string; dispatchId: string } | undefined {
|
||||
const payloadTaskId = payload.taskId
|
||||
const payloadDispatchId = payload.dispatchId
|
||||
const taskId = typeof payloadTaskId === 'string' && payloadTaskId.length > 0 ? payloadTaskId : ''
|
||||
const dispatchId =
|
||||
typeof payloadDispatchId === 'string' && payloadDispatchId.length > 0 ? payloadDispatchId : ''
|
||||
if (taskId && dispatchId) {
|
||||
return { taskId, dispatchId }
|
||||
}
|
||||
|
||||
const fallback = db.getSoleActiveDispatchForSender(msg.from_handle, msg.sender_pane_key)
|
||||
if (fallback && hasLifecycleAuthority(fallback, msg)) {
|
||||
return { taskId: taskId || fallback.task_id, dispatchId: dispatchId || fallback.id }
|
||||
}
|
||||
|
||||
onLog(
|
||||
taskId
|
||||
? `Warning: worker_done without dispatchId from ${msg.from_handle}`
|
||||
: `Warning: worker_done without taskId from ${msg.from_handle}`
|
||||
)
|
||||
return undefined
|
||||
}
|
||||
|
||||
function buildLifecycleAuthorityRejectionReason(
|
||||
dispatchId: string,
|
||||
dispatch: { assignee_handle: string | null; assignee_pane_key: string | null },
|
||||
|
||||
@@ -748,11 +748,26 @@ describe('orchestration RPC methods', () => {
|
||||
expect(db.getUnreadMessages('term_coord', ['worker_done'])).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('does not complete worker_done missing taskId or dispatchId', async () => {
|
||||
it('completes worker_done missing taskId or dispatchId from the assigned terminal', async () => {
|
||||
setup()
|
||||
const { task, dispatch } = createDispatchedTask()
|
||||
insertWorkerDone({ dispatchId: dispatch.id })
|
||||
insertWorkerDone({ taskId: task.id })
|
||||
|
||||
const result = (await call('orchestration.check', {
|
||||
terminal: 'term_coord',
|
||||
types: 'worker_done'
|
||||
})) as { count: number }
|
||||
|
||||
expect(result.count).toBe(1)
|
||||
expect(db.getTask(task.id)?.status).toBe('completed')
|
||||
expect(db.getDispatchContextById(dispatch.id)?.status).toBe('completed')
|
||||
})
|
||||
|
||||
it('does not complete worker_done missing taskId or dispatchId from another terminal', async () => {
|
||||
setup()
|
||||
const { task, dispatch } = createDispatchedTask()
|
||||
insertWorkerDone({ from: 'term_other', dispatchId: dispatch.id })
|
||||
insertWorkerDone({ from: 'term_other', taskId: task.id })
|
||||
|
||||
const result = (await call('orchestration.check', {
|
||||
terminal: 'term_coord',
|
||||
|
||||
Reference in New Issue
Block a user