mirror of
https://github.com/stablyai/orca.git
synced 2026-10-26 08:02:44 +00:00
Prevent same-pane handles from stealing assignee during restart
When a handle remints after a restart, nested agents or other processes inheriting the pane key must not steal the assignee binding while the original worker is still live. Rebind is a self-heal, not an ownership transfer. Gate rebind on handle liveness via getTerminalPaneKey so strict crash detection remains intact for the genuine worker.
This commit is contained in:
@@ -281,7 +281,11 @@ export class Coordinator {
|
||||
}
|
||||
|
||||
private handleLifecycleMessage(msg: MessageRow): void {
|
||||
const result = reconcileLifecycleMessage(this.db, msg, this.opts.onLog)
|
||||
const result = reconcileLifecycleMessage(this.db, msg, this.opts.onLog, {
|
||||
// Why: a handle that still resolves to a pane is live; lightweight fakes
|
||||
// without getTerminalPaneKey fail open to the rebind self-heal.
|
||||
isAssigneeHandleLive: (handle) => this.runtime.getTerminalPaneKey?.(handle) != null
|
||||
})
|
||||
if (result.action === 'completed') {
|
||||
if (!this.state.completedTasks.includes(result.taskId)) {
|
||||
this.state.completedTasks.push(result.taskId)
|
||||
|
||||
@@ -45,6 +45,12 @@ export type LifecycleRejectionResult = {
|
||||
|
||||
type LogFn = (msg: string) => void
|
||||
|
||||
export type LifecycleReconcileOptions = {
|
||||
// Why: gates assignee rebind on the stored handle being dead. Optional so
|
||||
// DB-only callers/fakes keep working; absent means fail-open to rebind.
|
||||
isAssigneeHandleLive?: (handle: string) => boolean
|
||||
}
|
||||
|
||||
const noopLog: LogFn = () => {}
|
||||
|
||||
function parseObjectPayload(msg: MessageRow, onInvalidJson: () => void): Record<string, unknown> {
|
||||
@@ -85,13 +91,14 @@ function getPersistedLifecycleRejection(
|
||||
export function reconcileLifecycleMessage(
|
||||
db: OrchestrationDb,
|
||||
msg: MessageRow,
|
||||
onLog: LogFn = noopLog
|
||||
onLog: LogFn = noopLog,
|
||||
opts?: LifecycleReconcileOptions
|
||||
): LifecycleReconciliationResult {
|
||||
switch (msg.type) {
|
||||
case 'worker_done':
|
||||
return reconcileWorkerDoneMessage(db, msg, onLog)
|
||||
case 'heartbeat':
|
||||
return reconcileHeartbeatMessage(db, msg, onLog)
|
||||
return reconcileHeartbeatMessage(db, msg, onLog, opts)
|
||||
case 'status':
|
||||
case 'dispatch':
|
||||
case 'merge_ready':
|
||||
@@ -105,7 +112,8 @@ export function reconcileLifecycleMessage(
|
||||
function reconcileHeartbeatMessage(
|
||||
db: OrchestrationDb,
|
||||
msg: MessageRow,
|
||||
onLog: LogFn
|
||||
onLog: LogFn,
|
||||
opts?: LifecycleReconcileOptions
|
||||
): LifecycleReconciliationResult {
|
||||
if (!msg.payload) {
|
||||
onLog(`Heartbeat from ${msg.from_handle} missing payload; ignored`)
|
||||
@@ -150,8 +158,18 @@ function reconcileHeartbeatMessage(
|
||||
// dispatch, so re-anchor the assignee here. This is the durable self-heal
|
||||
// after a restart remints the worker's handle — handle-based resolution and
|
||||
// strict crash detection work again from the first post-restart heartbeat,
|
||||
// with no in-memory tracking.
|
||||
db.rebindDispatchAssignee(dispatchId, msg.from_handle, msg.sender_pane_key ?? undefined)
|
||||
// with no in-memory tracking. Rebind is a self-heal, not an ownership
|
||||
// transfer: while the stored assignee handle is still live, a same-pane
|
||||
// sender under a different handle (e.g. a nested agent inheriting
|
||||
// ORCA_PANE_KEY) must not steal the binding — that would silence strict
|
||||
// crash detection for the genuine worker.
|
||||
const assigneeStillLiveElsewhere =
|
||||
msg.from_handle !== dispatch.assignee_handle &&
|
||||
dispatch.assignee_handle != null &&
|
||||
opts?.isAssigneeHandleLive?.(dispatch.assignee_handle) === true
|
||||
if (!assigneeStillLiveElsewhere) {
|
||||
db.rebindDispatchAssignee(dispatchId, msg.from_handle, msg.sender_pane_key ?? undefined)
|
||||
}
|
||||
|
||||
// Why: dispatchId-specific writes let the DB ignore late heartbeats for
|
||||
// completed/failed retries without masking a newer hung dispatch.
|
||||
|
||||
@@ -87,6 +87,54 @@ describe('orchestration lineage survives a restart handle remint', () => {
|
||||
expect(d.getActiveDispatchForTerminal('term_before_restart')).toBeUndefined()
|
||||
})
|
||||
|
||||
it('does not let a same-pane sender steal the assignee while it is still live', () => {
|
||||
// Why: rebind is a restart self-heal, not an ownership transfer. A nested
|
||||
// agent inheriting the pane key must not silence strict crash detection
|
||||
// for the genuine worker by rebinding assignee_handle to itself.
|
||||
const d = createDb()
|
||||
const task = d.createTask({ spec: 'work' })
|
||||
const ctx = d.createDispatchContext(task.id, 'term_worker', `tab_worker:${LEAF_B}`)
|
||||
|
||||
const heartbeat = d.insertMessage({
|
||||
from: 'term_impostor',
|
||||
to: 'coordinator',
|
||||
subject: 'hb',
|
||||
type: 'heartbeat',
|
||||
payload: JSON.stringify({ dispatchId: ctx.id }),
|
||||
senderPaneKey: `tab_worker:${LEAF_B}`
|
||||
})
|
||||
const result = reconcileLifecycleMessage(d, heartbeat, undefined, {
|
||||
isAssigneeHandleLive: (handle) => handle === 'term_worker'
|
||||
})
|
||||
|
||||
// Pane authority still records liveness (pre-existing v6 behavior)…
|
||||
expect(result.action).toBe('heartbeat_recorded')
|
||||
// …but the binding stays with the live worker.
|
||||
expect(d.getDispatchContextById(ctx.id)?.assignee_handle).toBe('term_worker')
|
||||
expect(d.getActiveDispatchForTerminal('term_worker')?.id).toBe(ctx.id)
|
||||
})
|
||||
|
||||
it('still rebinds after a restart when the stored assignee handle is dead', () => {
|
||||
const d = createDb()
|
||||
const task = d.createTask({ spec: 'work' })
|
||||
const ctx = d.createDispatchContext(task.id, 'term_before_restart', `tab_before:${LEAF_B}`)
|
||||
|
||||
const heartbeat = d.insertMessage({
|
||||
from: 'term_after_restart',
|
||||
to: 'coordinator',
|
||||
subject: 'hb',
|
||||
type: 'heartbeat',
|
||||
payload: JSON.stringify({ dispatchId: ctx.id }),
|
||||
senderPaneKey: `tab_after:${LEAF_B}`
|
||||
})
|
||||
const result = reconcileLifecycleMessage(d, heartbeat, undefined, {
|
||||
isAssigneeHandleLive: () => false
|
||||
})
|
||||
|
||||
expect(result.action).toBe('heartbeat_recorded')
|
||||
expect(d.getDispatchContextById(ctx.id)?.assignee_handle).toBe('term_after_restart')
|
||||
})
|
||||
|
||||
it('does not rebind a closed dispatch from a late heartbeat', () => {
|
||||
// Why: rebind is scoped to live dispatches — a straggler heartbeat after
|
||||
// completion must not resurrect assignee identity on a done row.
|
||||
|
||||
@@ -231,7 +231,9 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
|
||||
// immediately dispatch to the same terminal, which fails if the lock
|
||||
// is still held.
|
||||
if (msg.type === 'worker_done' || msg.type === 'heartbeat') {
|
||||
const reconciled = reconcileLifecycleMessage(db, msg)
|
||||
const reconciled = reconcileLifecycleMessage(db, msg, undefined, {
|
||||
isAssigneeHandleLive: (handle) => runtime.getTerminalPaneKey(handle) != null
|
||||
})
|
||||
// Why: a suppressed message is already read; waking a `check --wait`
|
||||
// waiter for it would return an empty result before the deadline.
|
||||
if (reconciled.action === 'suppressed') {
|
||||
@@ -319,7 +321,9 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
|
||||
// the coordinator loop sees them, but unread `check` is still an
|
||||
// authoritative read path for worker_done/heartbeat.
|
||||
visibleMessages = messages.map((message) => {
|
||||
const reconciled = reconcileLifecycleMessage(db, message)
|
||||
const reconciled = reconcileLifecycleMessage(db, message, undefined, {
|
||||
isAssigneeHandleLive: (handle) => runtime.getTerminalPaneKey(handle) != null
|
||||
})
|
||||
return reconciled.action === 'rejected'
|
||||
? (db.getMessageById(message.id) ?? message)
|
||||
: message
|
||||
|
||||
Reference in New Issue
Block a user