fix(orchestration): link worker-created sub-runs

This commit is contained in:
Brennan Benson
2026-08-27 04:28:32 -07:00
parent 026389a3bc
commit e17dbe2cbb
9 changed files with 128 additions and 8 deletions
@@ -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
@@ -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,
@@ -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) {
@@ -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,
@@ -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})
@@ -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 = [
+2
View File
@@ -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
@@ -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)
@@ -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) {