From e17dbe2cbb478227679552cc6052aba4c06c8d5c Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Thu, 27 Aug 2026 04:28:32 -0700 Subject: [PATCH] fix(orchestration): link worker-created sub-runs --- .../orchestration/db/contract-constants.ts | 4 +- .../db/dispatch-context/dispatch-lookup.ts | 32 ++++++++++- .../orchestration/db/runs/run-create.ts | 26 ++++++++- .../db/schema/create-core-tables-sql.ts | 1 + .../db/schema/migrate-v13-v30.ts | 3 + .../orchestration-schema-version-skew.ts | 3 +- src/main/runtime/orchestration/types.ts | 2 + .../rpc/methods/orchestration-runs.test.ts | 56 +++++++++++++++++++ .../runtime/rpc/methods/orchestration-runs.ts | 9 ++- 9 files changed, 128 insertions(+), 8 deletions(-) diff --git a/src/main/runtime/orchestration/db/contract-constants.ts b/src/main/runtime/orchestration/db/contract-constants.ts index 56390138f0a..f26f05b96dc 100644 --- a/src/main/runtime/orchestration/db/contract-constants.ts +++ b/src/main/runtime/orchestration/db/contract-constants.ts @@ -6,5 +6,5 @@ export const LEGACY_RUN_ID = ORCHESTRATION_LEGACY_RUN_ID export const LEGACY_CONTRACT_VERSION = 0 export const CURRENT_CONTRACT_VERSION = ORCHESTRATION_CONTRACT_VERSION -// Schema versions: v2 'heartbeat'+last_heartbeat_at, v3 delivered_at, v4 task-creator terminal, v5 task_title/display_name, v6 pane identity, v7 lightweight Runs, v8 crash-safe Run deliveries, v9 durable question threads, v10 Dispatch capabilities, v11 durable mutation receipts, v12 composed worker state, v18 post-v6 version-skew repair, v19 adopted legacy Runs and compatibility receipts, v20 legacy question backfill, v21 legacy scheduler-loss provenance, v22 dispatch assignee lookup, v23 worker terminal resource ownership, v24 creator-incarnation authority, v25 active Dispatch handle lookup, v26 indexed mutation receipt capacity, v27 durable federation acknowledgments, v28 durable local mutation caller identity. -export const SCHEMA_VERSION = 30 +// Schema versions: v2 'heartbeat'+last_heartbeat_at, v3 delivered_at, v4 task-creator terminal, v5 task_title/display_name, v6 pane identity, v7 lightweight Runs, v8 crash-safe Run deliveries, v9 durable question threads, v10 Dispatch capabilities, v11 durable mutation receipts, v12 composed worker state, v18 post-v6 version-skew repair, v19 adopted legacy Runs and compatibility receipts, v20 legacy question backfill, v21 legacy scheduler-loss provenance, v22 dispatch assignee lookup, v23 worker terminal resource ownership, v24 creator-incarnation authority, v25 active Dispatch handle lookup, v26 indexed mutation receipt capacity, v27 durable federation acknowledgments, v28 durable local mutation caller identity, v31 sub-Run parent dispatch provenance. +export const SCHEMA_VERSION = 31 diff --git a/src/main/runtime/orchestration/db/dispatch-context/dispatch-lookup.ts b/src/main/runtime/orchestration/db/dispatch-context/dispatch-lookup.ts index ae1cd4f45d1..814ec3c8d31 100644 --- a/src/main/runtime/orchestration/db/dispatch-context/dispatch-lookup.ts +++ b/src/main/runtime/orchestration/db/dispatch-context/dispatch-lookup.ts @@ -1,5 +1,5 @@ import { parsePaneKey } from '../../../../../shared/stable-pane-id' -import type { DispatchContextRow } from '../../types' +import type { DispatchContextRow, RemoteDispatchAttachmentRow } from '../../types' import { DISPATCH_PANE_KEY_MATCH_SUFFIX_SQL, isEquivalentPaneKey, @@ -17,6 +17,34 @@ export function getActiveDispatchForTerminal( return this.findActiveDispatchForAssignee(handle, assigneePaneKey) } +/** Resolve the origin id across local Dispatch rows and live federated attachments. */ +export function getActiveDispatchIdForTerminal( + this: OrchestrationDb, + handle: string, + assigneePaneKey?: string, + processIncarnation?: string +): string | undefined { + const local = this.findActiveDispatchForAssignee(handle, assigneePaneKey) + if (local) { + return local.id + } + if (!assigneePaneKey || !processIncarnation || !parsePaneKey(assigneePaneKey)) { + return undefined + } + const rows = this.db + .prepare( + `SELECT * FROM remote_dispatch_attachments + WHERE process_incarnation = ? AND pane_key IS NOT NULL + AND state IN ('starting', 'ready', 'start_unknown', 'stopping', 'stop_unknown')` + ) + .all(processIncarnation) as RemoteDispatchAttachmentRow[] + const matches = rows.filter( + (row) => row.pane_key !== null && isEquivalentPaneKey(row.pane_key, assigneePaneKey) + ) + // Ambiguous identity is fail-closed: never attribute a sub-Run to an arbitrary parent. + return matches.length === 1 ? matches[0].dispatch_id : undefined +} + /** * Cheap "are there any dispatch rows at all" probe. When false, no terminal * can have an active or recent-completed dispatch, so orchestration-context @@ -169,6 +197,7 @@ export function getLatestDispatchForTerminal( export type DispatchLookupMethods = { getActiveDispatchForTerminal: typeof getActiveDispatchForTerminal + getActiveDispatchIdForTerminal: typeof getActiveDispatchIdForTerminal hasAnyDispatchContexts: typeof hasAnyDispatchContexts getActiveDispatchForIdentity: typeof getActiveDispatchForIdentity getActiveDispatchMailboxOwners: typeof getActiveDispatchMailboxOwners @@ -180,6 +209,7 @@ export type DispatchLookupMethods = { export function attachDispatchLookup(ctor: { prototype: object }): void { Object.assign(ctor.prototype, { getActiveDispatchForTerminal, + getActiveDispatchIdForTerminal, hasAnyDispatchContexts, getActiveDispatchForIdentity, getActiveDispatchMailboxOwners, diff --git a/src/main/runtime/orchestration/db/runs/run-create.ts b/src/main/runtime/orchestration/db/runs/run-create.ts index 3f99e018936..242ca4a9de4 100644 --- a/src/main/runtime/orchestration/db/runs/run-create.ts +++ b/src/main/runtime/orchestration/db/runs/run-create.ts @@ -10,20 +10,40 @@ export function createRun( objective: string coordinatorHandle: string coordinatorPaneKey: string + /** The active worker Dispatch that created this sub-Run, if any. */ + parentDispatchId?: string } ): RunRow { const id = generateId('run') this.db.exec('BEGIN IMMEDIATE') try { + // Re-check inside the write transaction so a completed/revoked Dispatch is never + // recorded as the origin of a Run after its lifecycle authority has ended. + const parentDispatchId = params.parentDispatchId + ? (( + this.db + .prepare( + `SELECT id FROM dispatch_contexts + WHERE id = ? AND status IN ('pending', 'dispatched')` + ) + .get(params.parentDispatchId) as { id: string } | undefined + )?.id ?? null) + : null this.unbindOtherRunsForPane(params.coordinatorPaneKey) this.db .prepare( `INSERT INTO runs ( id, objective, coordinator_handle, coordinator_pane_key, - consumer_generation, legacy - ) VALUES (?, ?, ?, ?, 1, 0)` + parent_dispatch_id, consumer_generation, legacy + ) VALUES (?, ?, ?, ?, ?, 1, 0)` + ) + .run( + id, + params.objective, + params.coordinatorHandle, + params.coordinatorPaneKey, + parentDispatchId ) - .run(id, params.objective, params.coordinatorHandle, params.coordinatorPaneKey) this.rememberRunCoordinatorHandle(id, params.coordinatorHandle) this.db.exec('COMMIT') } catch (error) { diff --git a/src/main/runtime/orchestration/db/schema/create-core-tables-sql.ts b/src/main/runtime/orchestration/db/schema/create-core-tables-sql.ts index 1b3172edf31..c03c0cd7331 100644 --- a/src/main/runtime/orchestration/db/schema/create-core-tables-sql.ts +++ b/src/main/runtime/orchestration/db/schema/create-core-tables-sql.ts @@ -6,6 +6,7 @@ CREATE TABLE IF NOT EXISTS runs ( id TEXT PRIMARY KEY, objective TEXT NOT NULL, home_database TEXT NOT NULL DEFAULT 'this_database', + parent_dispatch_id TEXT, coordinator_handle TEXT, coordinator_pane_key TEXT, consumer_generation INTEGER NOT NULL DEFAULT 0, diff --git a/src/main/runtime/orchestration/db/schema/migrate-v13-v30.ts b/src/main/runtime/orchestration/db/schema/migrate-v13-v30.ts index 654395bcd2c..0419cc81946 100644 --- a/src/main/runtime/orchestration/db/schema/migrate-v13-v30.ts +++ b/src/main/runtime/orchestration/db/schema/migrate-v13-v30.ts @@ -183,6 +183,9 @@ export function applySchemaMigrationsV13ToV30(this: OrchestrationDb, current: nu AND pane_key IS NOT NULL; `) } + if (current < 31 && !this.hasColumn('runs', 'parent_dispatch_id')) { + this.db.exec('ALTER TABLE runs ADD COLUMN parent_dispatch_id TEXT') + } this.db.exec(` CREATE INDEX IF NOT EXISTS idx_dispatch_assignee_pane_leaf ON dispatch_contexts(${DISPATCH_PANE_KEY_MATCH_SUFFIX_SQL}) diff --git a/src/main/runtime/orchestration/orchestration-schema-version-skew.ts b/src/main/runtime/orchestration/orchestration-schema-version-skew.ts index 95ecbe6b034..59434ce609c 100644 --- a/src/main/runtime/orchestration/orchestration-schema-version-skew.ts +++ b/src/main/runtime/orchestration/orchestration-schema-version-skew.ts @@ -28,7 +28,8 @@ const POST_V6_COLUMNS = [ const VERSIONED_POST_V6_COLUMNS = [ { version: 27, table: 'federated_dispatches', column: 'to_home_acknowledged_sequence' }, { version: 30, table: 'dispatch_contexts', column: 'depth' }, - { version: 30, table: 'remote_dispatch_attachments', column: 'depth' } + { version: 30, table: 'remote_dispatch_attachments', column: 'depth' }, + { version: 31, table: 'runs', column: 'parent_dispatch_id' } ] as const const POST_V6_INDEXES = [ diff --git a/src/main/runtime/orchestration/types.ts b/src/main/runtime/orchestration/types.ts index b34f69e3c22..f300682defa 100644 --- a/src/main/runtime/orchestration/types.ts +++ b/src/main/runtime/orchestration/types.ts @@ -44,6 +44,8 @@ export type RunRow = { id: string objective: string home_database: string + /** Dispatch that created this Run, when a worker starts a sub-Run. */ + parent_dispatch_id: string | null coordinator_handle: string | null coordinator_pane_key: string | null consumer_generation: number diff --git a/src/main/runtime/rpc/methods/orchestration-runs.test.ts b/src/main/runtime/rpc/methods/orchestration-runs.test.ts index 4afc18f4b24..475591b95b3 100644 --- a/src/main/runtime/rpc/methods/orchestration-runs.test.ts +++ b/src/main/runtime/rpc/methods/orchestration-runs.test.ts @@ -86,6 +86,62 @@ describe('orchestration RPC methods', () => { expect(current.run?.id).toBe(created.run.id) }) + it('links a worker-created sub-Run to its active origin Dispatch', async () => { + setup(false) + const workerPane = 'tab_worker:bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb' + vi.spyOn(runtime, 'getTerminalPaneKey').mockImplementation((handle) => + handle === 'term_coord' ? coordinatorPaneKey : workerPane + ) + const root = db.createRun({ + objective: 'supervise worker', + coordinatorHandle: 'term_coord', + coordinatorPaneKey + }) + const task = db.createTask({ spec: 'worker task', runId: root.id }) + const dispatch = db.createDispatchContext({ + taskId: task.id, + assigneeHandle: 'term_worker', + assigneePaneKey: workerPane, + creator: { kind: 'system' }, + maxDepth: 2 + }) + + const created = (await call('orchestration.runCreate', { + objective: 'worker sub-run', + from: 'term_worker' + })) as { run: { id: string; parent_dispatch_id: string | null } } + + expect(created.run.parent_dispatch_id).toBe(dispatch.id) + expect(db.listRuns().runs.find((run) => run.id === created.run.id)).toMatchObject({ + parent_dispatch_id: dispatch.id + }) + }) + + it('does not retain a parent link when the origin Dispatch is no longer active', () => { + setup(false) + const run = db.createRun({ + objective: 'root', + coordinatorHandle: 'term_coord', + coordinatorPaneKey + }) + const task = db.createTask({ spec: 'worker task', runId: run.id }) + const dispatch = db.createDispatchContext({ + taskId: task.id, + assigneeHandle: 'term_worker', + assigneePaneKey: 'tab_worker:bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb', + creator: { kind: 'system' }, + maxDepth: 2 + }) + db.completeDispatch(dispatch.id) + const subRun = db.createRun({ + objective: 'late retry', + coordinatorHandle: 'term_worker', + coordinatorPaneKey: 'tab_worker:bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb', + parentDispatchId: dispatch.id + }) + expect(subRun.parent_dispatch_id).toBeNull() + }) + it('requires runtime-observed stable pane identity for binding', async () => { setup(false) vi.spyOn(runtime, 'getTerminalPaneKey').mockReturnValue(null) diff --git a/src/main/runtime/rpc/methods/orchestration-runs.ts b/src/main/runtime/rpc/methods/orchestration-runs.ts index 7938bc240f1..eb33ebc9462 100644 --- a/src/main/runtime/rpc/methods/orchestration-runs.ts +++ b/src/main/runtime/rpc/methods/orchestration-runs.ts @@ -38,10 +38,17 @@ export const ORCHESTRATION_RUN_METHODS: RpcMethod[] = [ }) const db = runtime.getOrchestrationDb() const priorRun = db.getCurrentRunForPane(paneKey) + const authority = runtime.getOrchestrationDispatchAuthority(params.from) + const parentDispatchId = db.getActiveDispatchIdForTerminal( + params.from, + paneKey, + authority?.processIncarnation ?? undefined + ) const run = db.createRun({ objective: params.objective, coordinatorHandle: params.from, - coordinatorPaneKey: paneKey + coordinatorPaneKey: paneKey, + parentDispatchId }) runtime.cancelMessageWaiters(params.from) if (priorRun) {