fix(orchestration): fence live worker lifecycle transitions

src/main/runtime/orca-runtime.ts

@@ -16482,7 +16482,7 @@ export class OrcaRuntimeService {
     }
 
     const errorContext = `Agent exited with code ${exitCode}`
-    this._orchestrationDb.failDispatch(dispatch.id, errorContext)
+    this._orchestrationDb.failDispatch(dispatch.id, errorContext, { workerProcessExited: true })
 
     // Why: create an escalation message so the coordinator is notified about
     // the unexpected exit on its next check cycle, even if the circuit breaker

src/main/runtime/orchestration/coordinator-decision-gates.test.ts

@@ -0,0 +1,69 @@
+import { afterEach, describe, expect, it } from 'vitest'
+import { openDecisionGateFromMessage } from './coordinator-decision-gates'
+import { OrchestrationDb } from './db'
+
+describe('coordinator decision-gate authority', () => {
+  let db: OrchestrationDb
+
+  afterEach(() => {
+    db?.close()
+  })
+
+  it('opens a gate only for the sender-owned active Dispatch', () => {
+    db = new OrchestrationDb(':memory:')
+    const task = db.createTask({ spec: 'owned gate target' })
+    const dispatch = db.createDispatchContext(task.id, 'term_owner', 'tab_owner:leaf_owner')
+    const logs: string[] = []
+
+    openDecisionGateFromMessage(
+      db,
+      db.insertMessage({
+        from: 'term_owner',
+        to: 'term_coordinator',
+        subject: 'Need approval',
+        type: 'decision_gate',
+        senderPaneKey: 'tab_owner:leaf_owner',
+        payload: JSON.stringify({ taskId: task.id, question: 'Proceed?' })
+      }),
+      (message) => logs.push(message)
+    )
+
+    expect(db.listGates({ taskId: task.id })).toHaveLength(1)
+    expect(db.getTask(task.id)?.status).toBe('blocked')
+    expect(db.getDispatchContextById(dispatch.id)?.status).toBe('completed')
+    expect(logs.at(-1)).toBe(`Task ${task.id} blocked on decision gate`)
+  })
+
+  it('rejects a gate targeting another active Dispatch without mutating either Task', () => {
+    db = new OrchestrationDb(':memory:')
+    const attackerTask = db.createTask({ spec: 'attacker assignment' })
+    const attacker = db.createDispatchContext(
+      attackerTask.id,
+      'term_attacker',
+      'tab_attacker:leaf_attacker'
+    )
+    const victimTask = db.createTask({ spec: 'victim assignment' })
+    const victim = db.createDispatchContext(victimTask.id, 'term_victim', 'tab_victim:leaf_victim')
+    const logs: string[] = []
+
+    openDecisionGateFromMessage(
+      db,
+      db.insertMessage({
+        from: 'term_attacker',
+        to: 'term_coordinator',
+        subject: 'Block the victim',
+        type: 'decision_gate',
+        senderPaneKey: 'tab_attacker:leaf_attacker',
+        payload: JSON.stringify({ taskId: victimTask.id, question: 'Stop?' })
+      }),
+      (message) => logs.push(message)
+    )
+
+    expect(db.listGates({ taskId: victimTask.id })).toHaveLength(0)
+    expect(db.getTask(attackerTask.id)?.status).toBe('dispatched')
+    expect(db.getTask(victimTask.id)?.status).toBe('dispatched')
+    expect(db.getDispatchContextById(attacker.id)?.status).toBe('dispatched')
+    expect(db.getDispatchContextById(victim.id)?.status).toBe('dispatched')
+    expect(logs.at(-1)).toContain('Rejected decision gate from term_attacker')
+  })
+})

src/main/runtime/orchestration/coordinator-decision-gates.ts

@@ -1,5 +1,6 @@
 /** Decision-gate handling: opening a gate from a worker message and keeping gated tasks blocked. */
 import type { OrchestrationDb } from './db'
+import { OrchestrationError } from './orchestration-error'
 import type { MessageRow } from './types'
 
 export function openDecisionGateFromMessage(
@@ -23,11 +24,23 @@ export function openDecisionGateFromMessage(
     return
   }
 
-  db.createGate({
-    taskId: payload.taskId,
-    question: payload.question,
-    options: payload.options
-  })
+  try {
+    db.createGate({
+      taskId: payload.taskId,
+      question: payload.question,
+      options: payload.options,
+      requester: { handle: msg.from_handle, paneKey: msg.sender_pane_key }
+    })
+  } catch (error) {
+    if (
+      error instanceof OrchestrationError &&
+      (error.code === 'consumer_fenced' || error.code === 'task_not_startable')
+    ) {
+      onLog(`Rejected decision gate from ${msg.from_handle}: ${error.message}`)
+      return
+    }
+    throw error
+  }
 
   onLog(`Task ${payload.taskId} blocked on decision gate`)
 }

src/main/runtime/orchestration/coordinator-escalation-triage.ts

@@ -32,6 +32,14 @@ export function applyEscalationToDispatch(
     return null
   }
 
+  const worker = db.getWorkerDispatch(dispatch.id)
+  if (worker && !['failed', 'succeeded', 'stopped', 'abandoned'].includes(worker.state)) {
+    onLog(
+      `Task ${taskId} remains dispatched until supervised worker ${dispatch.id} stops or reports.`
+    )
+    return null
+  }
+
   // Why: fail the dispatch to increment the circuit breaker; under threshold the task returns to 'pending' for re-dispatch next tick.
   const updated = db.failDispatch(dispatch.id, msg.subject)
   if (updated?.status === 'circuit_broken') {

src/main/runtime/orchestration/db-task-dispatch-lifecycle-guards.test.ts

@@ -0,0 +1,202 @@
+import { mkdtempSync, rmSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { afterEach, describe, expect, it } from 'vitest'
+import type Database from '../../sqlite/sync-database'
+import { OrchestrationDb } from './db'
+
+type WorkerFixture = {
+  dispatchId: string
+  capability: string
+  handle: string
+  paneKey: string
+  processIncarnation: string
+}
+
+let db: OrchestrationDb | undefined
+let dir: string | undefined
+
+afterEach(() => {
+  db?.close()
+  if (dir) {
+    rmSync(dir, { recursive: true, force: true })
+  }
+  db = undefined
+  dir = undefined
+})
+
+describe('Task/Dispatch lifecycle guards', () => {
+  it('rejects a worker report while another supervised Dispatch is active', () => {
+    const database = createDatabase()
+    const task = database.createTask({ spec: 'legacy supervised split' })
+    const first = startWorker(database, task.id, 'first')
+    sqliteFor(database).prepare("UPDATE tasks SET status = 'ready' WHERE id = ?").run(task.id)
+    const second = startWorker(database, task.id, 'second')
+
+    expect(
+      database.settleWorkerReport({
+        taskId: task.id,
+        dispatchId: second.dispatchId,
+        outcome: 'succeeded',
+        result: 'must wait for first'
+      })
+    ).toMatchObject({ action: 'rejected', code: 'inactive_dispatch' })
+    expect(database.getTask(task.id)?.status).toBe('dispatched')
+    expect(database.getDispatchContextById(first.dispatchId)?.status).toBe('dispatched')
+    expect(database.getDispatchContextById(second.dispatchId)?.status).toBe('dispatched')
+    expect(database.getWorkerDispatch(first.dispatchId)?.state).toBe('ready')
+    expect(database.getWorkerDispatch(second.dispatchId)?.state).toBe('ready')
+    expectCapability(database, first, true)
+    expectCapability(database, second, true)
+  })
+
+  it.each(['succeeded', 'failed'] as const)(
+    'settles context-only legacy siblings after a %s worker report',
+    (outcome) => {
+      const database = createDatabase()
+      const task = database.createTask({ spec: 'legacy mixed split' })
+      const contextOnly = database.createDispatchContext(task.id, 'term_context')
+      sqliteFor(database).prepare("UPDATE tasks SET status = 'ready' WHERE id = ?").run(task.id)
+      const worker = startWorker(database, task.id, 'reporter')
+
+      expect(
+        database.settleWorkerReport({
+          taskId: task.id,
+          dispatchId: worker.dispatchId,
+          outcome,
+          result: `${outcome} result`
+        })
+      ).toEqual({ action: 'settled', outcome, duplicate: false })
+      const expectedStatus = outcome === 'succeeded' ? 'completed' : 'failed'
+      expect(database.getTask(task.id)?.status).toBe(expectedStatus)
+      expect(database.getDispatchContextById(contextOnly.id)).toMatchObject({
+        status: expectedStatus,
+        capability_revoked_at: expect.any(String)
+      })
+      expect(database.getActiveDispatchForTerminal('term_context')).toBeUndefined()
+      expect(() =>
+        database.createDispatchContext(
+          database.createTask({ spec: 'later context work' }).id,
+          'term_context'
+        )
+      ).not.toThrow()
+    }
+  )
+
+  it('rejects generic failure while a supervised worker remains active', () => {
+    const database = createDatabase()
+    const task = database.createTask({ spec: 'supervised failure guard' })
+    const worker = startWorker(database, task.id, 'guarded')
+
+    expect(() => database.failDispatch(worker.dispatchId, 'unsafe retry')).toThrowError(
+      expect.objectContaining({
+        code: 'task_not_startable',
+        data: { dispatchId: worker.dispatchId }
+      })
+    )
+    expect(database.getTask(task.id)?.status).toBe('dispatched')
+    expect(database.getDispatchContextById(worker.dispatchId)).toMatchObject({
+      status: 'dispatched',
+      failure_count: 0,
+      capability_revoked_at: null
+    })
+    expect(database.getWorkerDispatch(worker.dispatchId)?.state).toBe('ready')
+    expectCapability(database, worker, true)
+  })
+
+  it('atomically settles worker state when a proven process exit fails its Dispatch', () => {
+    const database = createDatabase()
+    const task = database.createTask({ spec: 'exited worker' })
+    const worker = startWorker(database, task.id, 'exited')
+
+    expect(
+      database.failDispatch(worker.dispatchId, 'process exited', { workerProcessExited: true })
+    ).toMatchObject({ status: 'failed', failure_count: 1 })
+    expect(database.getTask(task.id)?.status).toBe('ready')
+    expect(database.getWorkerDispatch(worker.dispatchId)).toMatchObject({
+      state: 'failed',
+      stage: 'process_exited',
+      last_error: 'process exited'
+    })
+    expectCapability(database, worker, false)
+  })
+
+  it('rejects gate creation while a supervised worker remains active', () => {
+    const database = createDatabase()
+    const task = database.createTask({ spec: 'worker gate guard' })
+    const worker = startWorker(database, task.id, 'gate')
+
+    expect(() => database.createGate({ taskId: task.id, question: 'Proceed?' })).toThrowError(
+      expect.objectContaining({
+        code: 'task_not_startable',
+        data: { taskId: task.id, dispatchId: worker.dispatchId }
+      })
+    )
+    expect(database.listGates({ taskId: task.id })).toHaveLength(0)
+    expect(database.getTask(task.id)?.status).toBe('dispatched')
+    expect(database.getDispatchContextById(worker.dispatchId)?.status).toBe('dispatched')
+    expect(database.getWorkerDispatch(worker.dispatchId)?.state).toBe('ready')
+    expectCapability(database, worker, true)
+  })
+
+  it('rolls back gate resolution when an active Dispatch blocks readiness', () => {
+    const database = createDatabase()
+    const task = database.createTask({ spec: 'corrupt gated task' })
+    const gate = database.createGate({ taskId: task.id, question: 'Proceed?' })
+    sqliteFor(database).prepare("UPDATE tasks SET status = 'ready' WHERE id = ?").run(task.id)
+    const dispatch = database.createDispatchContext(task.id, 'term_worker')
+    sqliteFor(database).prepare("UPDATE tasks SET status = 'blocked' WHERE id = ?").run(task.id)
+
+    expect(() => database.resolveGate(gate.id, 'yes')).toThrowError(
+      expect.objectContaining({ code: 'task_not_startable' })
+    )
+    expect(database.getGate(gate.id)).toMatchObject({
+      status: 'pending',
+      resolution: null,
+      resolved_at: null
+    })
+    expect(database.getTask(task.id)?.status).toBe('blocked')
+    expect(database.getDispatchContextById(dispatch.id)?.status).toBe('dispatched')
+  })
+})
+
+function createDatabase(): OrchestrationDb {
+  dir = mkdtempSync(join(tmpdir(), 'orca-task-dispatch-lifecycle-'))
+  db = new OrchestrationDb(join(dir, 'orchestration.db'))
+  return db
+}
+
+function startWorker(database: OrchestrationDb, taskId: string, name: string): WorkerFixture {
+  const started = database.createStartingWorkerDispatch({ taskId, startOptions: {} })
+  const paneSuffix = name.length.toString(16).padStart(12, '0')
+  const paneKey = `tab_${name}:aaaaaaaa-aaaa-4aaa-8aaa-${paneSuffix}`
+  const processIncarnation = `${name}:1`
+  const handle = `term_${name}`
+  const capability = database.prepareStartingWorkerAuthority({
+    dispatchId: started.dispatch.id,
+    handle,
+    paneKey,
+    processIncarnation,
+    worktreeId: `repo::${name}`,
+    effects: [],
+    setupState: 'not_applicable',
+    terminalOwnership: 'created'
+  })
+  database.markWorkerDispatchReady(started.dispatch.id)
+  return { dispatchId: started.dispatch.id, capability, handle, paneKey, processIncarnation }
+}
+
+function expectCapability(database: OrchestrationDb, worker: WorkerFixture, valid: boolean): void {
+  expect(
+    database.verifyDispatchCapability({
+      dispatchId: worker.dispatchId,
+      capability: worker.capability,
+      paneKey: worker.paneKey,
+      processIncarnation: worker.processIncarnation
+    }).valid
+  ).toBe(valid)
+}
+
+function sqliteFor(database: OrchestrationDb): Database.Database {
+  return (database as unknown as { db: Database.Database }).db
+}

src/main/runtime/orchestration/db-task-dispatch-races.test.ts

@@ -89,7 +89,9 @@ describe('Task/Dispatch concurrency', () => {
       return prepare(sql)
     })
 
-    expect(first.db.failDispatch(started.dispatch.id, 'stale failure')).toMatchObject({
+    expect(
+      first.db.failDispatch(started.dispatch.id, 'stale failure', { workerProcessExited: true })
+    ).toMatchObject({
       status: 'completed',
       failure_count: 0
     })

src/main/runtime/orchestration/db.ts

@@ -7590,6 +7590,31 @@ export class OrchestrationDb {
       }
     }
 
+    const conflictingWorker = this.db
+      .prepare(
+        `SELECT active.id
+         FROM dispatch_contexts active
+         JOIN worker_dispatches worker ON worker.dispatch_id = active.id
+         WHERE active.task_id = ? AND active.id != ?
+           AND active.status IN ('pending', 'dispatched')
+           AND worker.state NOT IN ('failed', 'succeeded', 'stopped', 'abandoned')
+         ORDER BY active.rowid DESC LIMIT 1`
+      )
+      .get(params.taskId, params.dispatchId) as Pick<DispatchContextRow, 'id'> | undefined
+    if (conflictingWorker) {
+      return {
+        action: 'rejected',
+        code: 'inactive_dispatch',
+        reason: `Task ${params.taskId} still has active supervised Dispatch ${conflictingWorker.id}; stop or settle it before completing ${params.dispatchId}.`
+      }
+    }
+    const siblingDispatchIds = this.db
+      .prepare(
+        `SELECT id FROM dispatch_contexts
+         WHERE task_id = ? AND id != ? AND status IN ('pending', 'dispatched')`
+      )
+      .all(params.taskId, params.dispatchId) as Pick<DispatchContextRow, 'id'>[]
+
     this.db.exec('SAVEPOINT settle_worker_report')
     const dispatchUpdate = this.db
       .prepare(
@@ -7623,7 +7648,15 @@ export class OrchestrationDb {
          WHERE dispatch_id = ? AND state = 'ready'`
       )
       .run(params.outcome === 'succeeded' ? 'succeeded' : 'failed', params.dispatchId)
+    this.settleActiveDispatchesForTask(
+      params.taskId,
+      expectedDispatchStatus,
+      params.outcome === 'failed' ? params.result : undefined
+    )
     this.closeQuestionsForDispatch(params.dispatchId)
+    for (const sibling of siblingDispatchIds) {
+      this.closeQuestionsForDispatch(sibling.id)
+    }
     if (params.outcome === 'succeeded') {
       this.promoteReadyTasks(params.taskId)
     }
@@ -7662,7 +7695,11 @@ export class OrchestrationDb {
       .all(thresholdIso, thresholdIso) as DispatchContextRow[]
   }
 
-  failDispatch(ctxId: string, error: string): DispatchContextRow | undefined {
+  failDispatch(
+    ctxId: string,
+    error: string,
+    options: { workerProcessExited?: boolean } = {}
+  ): DispatchContextRow | undefined {
     this.db.exec('SAVEPOINT fail_dispatch')
     try {
       const updated = this.db
@@ -7672,17 +7709,47 @@ export class OrchestrationDb {
                failure_count = failure_count + 1, last_failure = ?,
                completed_at = COALESCE(completed_at, datetime('now')),
                capability_revoked_at = COALESCE(capability_revoked_at, datetime('now'))
-           WHERE id = ? AND status IN ('pending', 'dispatched')`
+           WHERE id = ? AND status IN ('pending', 'dispatched')
+             AND (? = 1 OR NOT EXISTS (
+               SELECT 1 FROM worker_dispatches worker
+               WHERE worker.dispatch_id = dispatch_contexts.id
+                 AND worker.state NOT IN ('failed', 'succeeded', 'stopped', 'abandoned')
+             ))`
         )
-        .run(error, ctxId)
+        .run(error, ctxId, options.workerProcessExited ? 1 : 0)
       const ctx = this.db.prepare('SELECT * FROM dispatch_contexts WHERE id = ?').get(ctxId) as
         | DispatchContextRow
         | undefined
+      const worker = this.getWorkerDispatch(ctxId)
       if (updated.changes !== 1 || !ctx) {
+        if (
+          ctx &&
+          worker &&
+          !['failed', 'succeeded', 'stopped', 'abandoned'].includes(worker.state) &&
+          !options.workerProcessExited
+        ) {
+          throw new OrchestrationError(
+            'task_not_startable',
+            `Dispatch ${ctxId} has an active supervised worker; stop it or settle its report first.`,
+            { dispatchId: ctxId }
+          )
+        }
         this.db.exec('RELEASE fail_dispatch')
         return ctx
       }
 
+      if (worker && options.workerProcessExited) {
+        this.db
+          .prepare(
+            `UPDATE worker_dispatches
+             SET state = 'failed', stage = 'process_exited', last_error = ?,
+                 updated_at = datetime('now')
+             WHERE dispatch_id = ?
+               AND state NOT IN ('failed', 'succeeded', 'stopped', 'abandoned')`
+          )
+          .run(error, ctxId)
+      }
+
       // Why: back to 'ready' not 'pending' — 'pending' would strand it since promoteReadyTasks only runs when a dep completes.
       const taskStatus: TaskStatus = ctx.status === 'circuit_broken' ? 'failed' : 'ready'
       this.db
@@ -7705,25 +7772,78 @@ export class OrchestrationDb {
 
   // ── Decision Gates ──
 
-  createGate(gate: { taskId: string; question: string; options?: string[] }): DecisionGateRow {
-    const id = generateId('gate')
-    const optionsJson = JSON.stringify(gate.options ?? [])
-    this.db
-      .prepare(
-        'INSERT INTO decision_gates (id, run_id, task_id, question, options) VALUES (?, ?, ?, ?, ?)'
-      )
-      .run(
-        id,
-        this.getTask(gate.taskId)?.run_id ?? LEGACY_RUN_ID,
-        gate.taskId,
-        gate.question,
-        optionsJson
-      )
+  createGate(gate: {
+    taskId: string
+    question: string
+    options?: string[]
+    requester?: { handle: string; paneKey?: string | null }
+  }): DecisionGateRow {
+    this.db.exec('SAVEPOINT create_gate')
+    try {
+      const active = this.db
+        .prepare(
+          `SELECT * FROM dispatch_contexts
+           WHERE task_id = ? AND status IN ('pending', 'dispatched')
+           ORDER BY rowid DESC LIMIT 1`
+        )
+        .get(gate.taskId) as DispatchContextRow | undefined
+      if (
+        gate.requester &&
+        (!active ||
+          active.assignee_handle !== gate.requester.handle ||
+          (gate.requester.paneKey &&
+            active.assignee_pane_key &&
+            !isEquivalentPaneKey(active.assignee_pane_key, gate.requester.paneKey)))
+      ) {
+        throw new OrchestrationError(
+          'consumer_fenced',
+          `Terminal ${gate.requester.handle} does not own the active Dispatch for Task ${gate.taskId}.`,
+          { taskId: gate.taskId, dispatchId: active?.id }
+        )
+      }
+      const activeWorker = this.db
+        .prepare(
+          `SELECT active.id
+           FROM dispatch_contexts active
+           JOIN worker_dispatches worker ON worker.dispatch_id = active.id
+           WHERE active.task_id = ? AND active.status IN ('pending', 'dispatched')
+             AND worker.state NOT IN ('failed', 'succeeded', 'stopped', 'abandoned')
+           ORDER BY active.rowid DESC LIMIT 1`
+        )
+        .get(gate.taskId) as Pick<DispatchContextRow, 'id'> | undefined
+      if (activeWorker) {
+        throw new OrchestrationError(
+          'task_not_startable',
+          `Task ${gate.taskId} cannot open a gate while supervised Dispatch ${activeWorker.id} is active; stop or settle its worker first.`,
+          { taskId: gate.taskId, dispatchId: activeWorker.id }
+        )
+      }
 
-    this.completeActiveDispatchesForTask(gate.taskId)
-    this.db.prepare("UPDATE tasks SET status = 'blocked' WHERE id = ?").run(gate.taskId)
-
-    return this.db.prepare('SELECT * FROM decision_gates WHERE id = ?').get(id) as DecisionGateRow
+      const id = generateId('gate')
+      const optionsJson = JSON.stringify(gate.options ?? [])
+      this.db
+        .prepare(
+          'INSERT INTO decision_gates (id, run_id, task_id, question, options) VALUES (?, ?, ?, ?, ?)'
+        )
+        .run(
+          id,
+          this.getTask(gate.taskId)?.run_id ?? LEGACY_RUN_ID,
+          gate.taskId,
+          gate.question,
+          optionsJson
+        )
+      this.completeActiveDispatchesForTask(gate.taskId)
+      this.db.prepare("UPDATE tasks SET status = 'blocked' WHERE id = ?").run(gate.taskId)
+      const created = this.db.prepare('SELECT * FROM decision_gates WHERE id = ?').get(id) as
+        | DecisionGateRow
+        | undefined
+      this.db.exec('RELEASE create_gate')
+      return created as DecisionGateRow
+    } catch (error) {
+      this.db.exec('ROLLBACK TO create_gate')
+      this.db.exec('RELEASE create_gate')
+      throw error
+    }
   }
 
   resolveGate(gateId: string, resolution: string): DecisionGateRow | undefined {
@@ -7734,18 +7854,26 @@ export class OrchestrationDb {
       return undefined
     }
 
-    this.db
-      .prepare(
-        "UPDATE decision_gates SET status = 'resolved', resolution = ?, resolved_at = datetime('now') WHERE id = ?"
-      )
-      .run(resolution, gateId)
+    this.db.exec('SAVEPOINT resolve_gate')
+    try {
+      this.db
+        .prepare(
+          "UPDATE decision_gates SET status = 'resolved', resolution = ?, resolved_at = datetime('now') WHERE id = ?"
+        )
+        .run(resolution, gateId)
 
-    // Why: set to 'ready' (not the previous status) so the coordinator re-dispatches the worker with the resolution context.
-    this.db.prepare("UPDATE tasks SET status = 'ready' WHERE id = ?").run(gate.task_id)
-
-    return this.db.prepare('SELECT * FROM decision_gates WHERE id = ?').get(gateId) as
-      | DecisionGateRow
-      | undefined
+      // Why: ready lets the coordinator re-dispatch with the resolution context.
+      this.updateTaskStatus(gate.task_id, 'ready')
+      const resolved = this.db.prepare('SELECT * FROM decision_gates WHERE id = ?').get(gateId) as
+        | DecisionGateRow
+        | undefined
+      this.db.exec('RELEASE resolve_gate')
+      return resolved
+    } catch (error) {
+      this.db.exec('ROLLBACK TO resolve_gate')
+      this.db.exec('RELEASE resolve_gate')
+      throw error
+    }
   }
 
   timeoutGate(gateId: string): DecisionGateRow | undefined {