From 2ff3d4dd6dccd30d897f11141e2145ac8f2f5ed3 Mon Sep 17 00:00:00 2001 From: Jinwoo-H Date: Sun, 16 Aug 2026 22:36:08 -0400 Subject: [PATCH] fix(orchestration): fence live worker lifecycle transitions --- src/main/runtime/orca-runtime.ts | 2 +- .../coordinator-decision-gates.test.ts | 69 ++++++ .../coordinator-decision-gates.ts | 23 +- .../coordinator-escalation-triage.ts | 8 + .../db-task-dispatch-lifecycle-guards.test.ts | 202 ++++++++++++++++++ .../db-task-dispatch-races.test.ts | 4 +- src/main/runtime/orchestration/db.ts | 192 ++++++++++++++--- 7 files changed, 461 insertions(+), 39 deletions(-) create mode 100644 src/main/runtime/orchestration/coordinator-decision-gates.test.ts create mode 100644 src/main/runtime/orchestration/db-task-dispatch-lifecycle-guards.test.ts diff --git a/src/main/runtime/orca-runtime.ts b/src/main/runtime/orca-runtime.ts index 69b7c342438..ce49bc46b4e 100644 --- a/src/main/runtime/orca-runtime.ts +++ b/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 diff --git a/src/main/runtime/orchestration/coordinator-decision-gates.test.ts b/src/main/runtime/orchestration/coordinator-decision-gates.test.ts new file mode 100644 index 00000000000..6e3bd3f8df1 --- /dev/null +++ b/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') + }) +}) diff --git a/src/main/runtime/orchestration/coordinator-decision-gates.ts b/src/main/runtime/orchestration/coordinator-decision-gates.ts index fa9fa3b73d9..4d3170296d5 100644 --- a/src/main/runtime/orchestration/coordinator-decision-gates.ts +++ b/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`) } diff --git a/src/main/runtime/orchestration/coordinator-escalation-triage.ts b/src/main/runtime/orchestration/coordinator-escalation-triage.ts index f37fd654399..9b687449719 100644 --- a/src/main/runtime/orchestration/coordinator-escalation-triage.ts +++ b/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') { diff --git a/src/main/runtime/orchestration/db-task-dispatch-lifecycle-guards.test.ts b/src/main/runtime/orchestration/db-task-dispatch-lifecycle-guards.test.ts new file mode 100644 index 00000000000..d246712634f --- /dev/null +++ b/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 +} diff --git a/src/main/runtime/orchestration/db-task-dispatch-races.test.ts b/src/main/runtime/orchestration/db-task-dispatch-races.test.ts index 613e3a3db34..c9c765f9757 100644 --- a/src/main/runtime/orchestration/db-task-dispatch-races.test.ts +++ b/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 }) diff --git a/src/main/runtime/orchestration/db.ts b/src/main/runtime/orchestration/db.ts index cbd0add8349..898c897f929 100644 --- a/src/main/runtime/orchestration/db.ts +++ b/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 | 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[] + 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 | 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 {