From addd9f3da7de4ddeb72a4905405fce9bafdd369f Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Tue, 8 Sep 2026 23:41:46 -0700 Subject: [PATCH] fix(orchestration): accept a missing Run id from older federation coordinators (#19689) * fix(orchestration): accept v1.4.198 coordinators on federationAttachStart #19542 made runId required on the attach RPC and backfilled existing attachments with '', on the premise that federation was unreleased. It shipped in v1.4.198, so a v1.4.198 coordinator got 'Missing Run ID' from an upgraded worker host, and every pre-upgrade attachment lost its mailbox because home_run_id='' matches no Run. - runId is optional on the wire; an absent id mints a per-attachment stub Run (run_federated_) through the same INSERT OR IGNORE path. - migrate-v40 backfills existing attachments with the stub and inserts the stub Runs, so in-flight workers keep reporting back. - create SQL gives home_run_id a DEFAULT '' so a rolled-back v1.4.198 host can still insert into a v1.4.199-created table. Stub Runs never carry run_legacy_local, and #19542's attachment-mailbox exclusion in the skew probe is untouched, so no adoption replay path is reintroduced (probe test added). * fix(orchestration): repair empty federated home Run ids on every open A host on v1.4.199 that rolls back to v1.4.198, attaches workers (rows land with home_run_id='' via the DEFAULT), then upgrades again never re-runs the v40 backfill because user_version is already 40, so those attachments stay without a Run and their control mail is refused. Lift the two idempotent set-based statements out of migrate-v40 into backfillFederatedStubHomeRuns and run it from both the v40 migration and the OrchestrationDb constructor after migrate(), matching the existing on-open rememberCurrentRunCoordinatorHandles repair. Tests: reopen a v40 file DB holding a v1.4.198-shaped '' row through the constructor and assert the stub Run and mailbox; pin the wire schema accepting a v1.4.198 request with no runId ('' -> undefined, whitespace passes the schema and is refused at the DB layer). --------- Co-authored-by: Merge Sim --- .../orchestration/db/contract-constants.ts | 6 + .../federated-stub-home-run-backfill.ts | 16 +++ .../remote-dispatch-attachment-create.ts | 13 ++- .../orchestration/db/orchestration-db.ts | 2 + .../db/schema/create-graph-tables-sql.ts | 3 +- .../federated-home-run-migration.test.ts | 103 ++++++++++++++++-- .../orchestration/db/schema/migrate-v40.ts | 14 ++- ...chestration-federated-legacy-probe.test.ts | 46 ++++++-- .../federation-start-schema.test.ts | 41 +++++++ .../federation/federation-start-schema.ts | 3 +- 10 files changed, 214 insertions(+), 33 deletions(-) create mode 100644 src/main/runtime/orchestration/db/federation/federated-stub-home-run-backfill.ts create mode 100644 src/main/runtime/rpc/methods/orchestration/federation/federation-start-schema.test.ts diff --git a/src/main/runtime/orchestration/db/contract-constants.ts b/src/main/runtime/orchestration/db/contract-constants.ts index 47e0cd6165a..0a2e27f000f 100644 --- a/src/main/runtime/orchestration/db/contract-constants.ts +++ b/src/main/runtime/orchestration/db/contract-constants.ts @@ -3,6 +3,12 @@ import { ORCHESTRATION_CONTRACT_VERSION } from '../../../../shared/protocol-vers export const LEGACY_RUN_ID = ORCHESTRATION_LEGACY_RUN_ID +// Why: a v1.4.198 coordinator sends no Run id, so its remote workers file mail under a per-attachment stub Run. +export const FEDERATED_STUB_HOME_RUN_ID_PREFIX = 'run_federated_' +export function federatedStubHomeRunId(dispatchId: string): string { + return `${FEDERATED_STUB_HOME_RUN_ID_PREFIX}${dispatchId}` +} + export const LEGACY_CONTRACT_VERSION = 0 export const CURRENT_CONTRACT_VERSION = ORCHESTRATION_CONTRACT_VERSION diff --git a/src/main/runtime/orchestration/db/federation/federated-stub-home-run-backfill.ts b/src/main/runtime/orchestration/db/federation/federated-stub-home-run-backfill.ts new file mode 100644 index 00000000000..47c596e260d --- /dev/null +++ b/src/main/runtime/orchestration/db/federation/federated-stub-home-run-backfill.ts @@ -0,0 +1,16 @@ +import type Database from '../../../../sqlite/sync-database' +import { FEDERATED_STUB_HOME_RUN_ID_PREFIX } from '../contract-constants' + +// Why: a rolled-back v1.4.198 host inserts attachments with home_run_id='' after user_version is +// already 40, so this idempotent repair runs on every open, not only inside the v40 migration. +export function backfillFederatedStubHomeRuns(db: Database.Database): void { + db.exec(` + INSERT OR IGNORE INTO runs (id, objective, home_database, consumer_generation, legacy) + SELECT '${FEDERATED_STUB_HOME_RUN_ID_PREFIX}' || dispatch_id, + 'Coordinated from ' || home_peer_fingerprint, 'remote', 0, 0 + FROM remote_dispatch_attachments WHERE home_run_id = ''; + UPDATE remote_dispatch_attachments + SET home_run_id = '${FEDERATED_STUB_HOME_RUN_ID_PREFIX}' || dispatch_id + WHERE home_run_id = ''; + `) +} diff --git a/src/main/runtime/orchestration/db/federation/remote-dispatch-attachment-create.ts b/src/main/runtime/orchestration/db/federation/remote-dispatch-attachment-create.ts index d927cf96838..5a6e89b65ca 100644 --- a/src/main/runtime/orchestration/db/federation/remote-dispatch-attachment-create.ts +++ b/src/main/runtime/orchestration/db/federation/remote-dispatch-attachment-create.ts @@ -2,13 +2,15 @@ import type { WorkerDispatchState, RemoteDispatchAttachmentRow } from '../../typ import { OrchestrationError } from '../../orchestration-error' import { ensureMutationReceiptCapacity } from '../../mutation-receipt-capacity' import type { OrchestrationDb } from '../orchestration-db' +import { federatedStubHomeRunId } from '../contract-constants' import { insertRemoteDispatchAttachmentRow } from '../dispatch-row-writer' export function createRemoteDispatchAttachment( this: OrchestrationDb, params: { dispatchId: string - runId: string + /** Absent from a v1.4.198 coordinator; replaced by a per-attachment stub Run. */ + runId?: string taskId: string homePeerFingerprint: string protocolVersion: number @@ -44,7 +46,8 @@ export function createRemoteDispatchAttachment( `Remote attachment request ${params.mutationReceipt.requestId} already exists.` ) } - if (!params.runId?.trim()) { + const runId = params.runId ?? federatedStubHomeRunId(params.dispatchId) + if (!runId.trim()) { throw new OrchestrationError('invalid_argument', 'Missing Run ID') } this.db @@ -52,8 +55,8 @@ export function createRemoteDispatchAttachment( `INSERT OR IGNORE INTO runs (id, objective, home_database, consumer_generation, legacy) VALUES (?, ?, 'remote', 0, 0)` ) - .run(params.runId, `Coordinated from ${params.homePeerFingerprint}`) - this.requireRun(params.runId) + .run(runId, `Coordinated from ${params.homePeerFingerprint}`) + this.requireRun(runId) ensureMutationReceiptCapacity(this.db) this.db .prepare( @@ -70,7 +73,7 @@ export function createRemoteDispatchAttachment( ) insertRemoteDispatchAttachmentRow(this.db, { dispatchId: params.dispatchId, - runId: params.runId, + runId, taskId: params.taskId, homePeerFingerprint: params.homePeerFingerprint, protocolVersion: params.protocolVersion, diff --git a/src/main/runtime/orchestration/db/orchestration-db.ts b/src/main/runtime/orchestration/db/orchestration-db.ts index 2a970841a38..1ce52e96c4a 100644 --- a/src/main/runtime/orchestration/db/orchestration-db.ts +++ b/src/main/runtime/orchestration/db/orchestration-db.ts @@ -1,6 +1,7 @@ import Database from '../../../sqlite/sync-database' import { attachOrchestrationDbMethods } from './attach-orchestration-db-methods' import { hardenOrchestrationDatabaseFiles } from './database-file-permissions' +import { backfillFederatedStubHomeRuns } from './federation/federated-stub-home-run-backfill' import type { OrchestrationDbMethods } from './orchestration-db-methods' import { createCoordinatorMailRoutingTrigger, @@ -28,6 +29,7 @@ class OrchestrationDbCore { this.db.pragma('busy_timeout = 5000') createTables.call(this as unknown as OrchestrationDb) migrate.call(this as unknown as OrchestrationDb) + backfillFederatedStubHomeRuns(this.db) createCoordinatorMailRoutingTrigger.call(this as unknown as OrchestrationDb) rememberCurrentRunCoordinatorHandles.call(this as unknown as OrchestrationDb) hardenOrchestrationDatabaseFiles(dbPath) diff --git a/src/main/runtime/orchestration/db/schema/create-graph-tables-sql.ts b/src/main/runtime/orchestration/db/schema/create-graph-tables-sql.ts index 0897cb852de..d674a5298e7 100644 --- a/src/main/runtime/orchestration/db/schema/create-graph-tables-sql.ts +++ b/src/main/runtime/orchestration/db/schema/create-graph-tables-sql.ts @@ -38,7 +38,8 @@ CREATE TABLE IF NOT EXISTS federated_dispatches ( ); CREATE TABLE IF NOT EXISTS remote_dispatch_attachments ( - home_run_id TEXT NOT NULL, + -- DEFAULT: a rolled-back v1.4.198 host still inserts here without a home Run. + home_run_id TEXT NOT NULL DEFAULT '', dispatch_id TEXT PRIMARY KEY, task_id TEXT NOT NULL, home_peer_fingerprint TEXT NOT NULL, diff --git a/src/main/runtime/orchestration/db/schema/federated-home-run-migration.test.ts b/src/main/runtime/orchestration/db/schema/federated-home-run-migration.test.ts index b970435223a..4fcd78634f3 100644 --- a/src/main/runtime/orchestration/db/schema/federated-home-run-migration.test.ts +++ b/src/main/runtime/orchestration/db/schema/federated-home-run-migration.test.ts @@ -1,26 +1,105 @@ -import { afterEach, describe, expect, it } from 'vitest' +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import { ORCHESTRATION_CONTRACT_VERSION } from '../../../../../shared/protocol-version' import { OrchestrationDb } from '../orchestration-db' +import { SCHEMA_VERSION, federatedStubHomeRunId } from '../contract-constants' import { migrateV40 } from './migrate-v40' import { importFederatedControlMessage } from '../../federation-control-message' describe('federated home Run migration', () => { - const db = new OrchestrationDb(':memory:') + let db: OrchestrationDb + beforeEach(() => { + db = new OrchestrationDb(':memory:') + }) afterEach(() => db.close()) - it('adds the home Run column and refuses mail for a development placeholder', () => { + function importInstruction(target: OrchestrationDb, dispatchId: string, messageId: string): void { + expect( + importFederatedControlMessage(target, { + dispatchId, + messageId, + payload: JSON.stringify({ from: 'home', subject: 'Instruction', body: '', type: 'status' }) + }) + ).toEqual({ imported: true, type: 'status' }) + } + + function attachWithoutRunId(target: OrchestrationDb, dispatchId: string, runId?: string): void { + target.createRemoteDispatchAttachment({ + dispatchId, + runId, + taskId: `task_${dispatchId}`, + homePeerFingerprint: 'home', + protocolVersion: ORCHESTRATION_CONTRACT_VERSION, + runtimeEpoch: 'epoch', + mutationReceipt: { + callerFingerprint: 'home', + requestId: `request_${dispatchId}`, + method: 'orchestration.federationAttachStart', + payloadHash: `payload_${dispatchId}` + } + }) + } + + it('backfills a pre-upgrade attachment with a stub home Run that keeps its mailbox', () => { db.db.exec('ALTER TABLE remote_dispatch_attachments DROP COLUMN home_run_id') db.db.exec(`INSERT INTO remote_dispatch_attachments (dispatch_id, task_id, home_peer_fingerprint, runtime_epoch) VALUES ('ctx_old', 'task_old', 'home', 'epoch')`) migrateV40.call(db, 39) - expect(db.getRemoteDispatchAttachment('ctx_old')?.home_run_id).toBe('') - expect(() => - importFederatedControlMessage(db, { - dispatchId: 'ctx_old', - messageId: 'message_old', - payload: JSON.stringify({ from: 'home', subject: 'Instruction', body: '', type: 'message' }) - }) - ).toThrow('Run not found:') - expect(db.getMessageById('message_old')).toBeUndefined() + const stubRunId = federatedStubHomeRunId('ctx_old') + expect(db.getRemoteDispatchAttachment('ctx_old')?.home_run_id).toBe(stubRunId) + expect(db.getRunRaw(stubRunId)).toMatchObject({ home_database: 'remote', legacy: 0 }) + importInstruction(db, 'ctx_old', 'message_old') + expect(db.getMessageById('message_old')?.run_id).toBe(stubRunId) + }) + + it('mints a stub home Run when a v1.4.198 coordinator attaches without a Run id', () => { + attachWithoutRunId(db, 'ctx_legacy_home') + const stubRunId = federatedStubHomeRunId('ctx_legacy_home') + expect(db.getRemoteDispatchAttachment('ctx_legacy_home')?.home_run_id).toBe(stubRunId) + importInstruction(db, 'ctx_legacy_home', 'message_legacy_home') + expect(db.getMessageById('message_legacy_home')?.run_id).toBe(stubRunId) + }) + + it('rejects a whitespace-only Run id instead of minting a stub', () => { + expect(() => attachWithoutRunId(db, 'ctx_blank', ' ')).toThrow('Missing Run ID') + expect(db.getRemoteDispatchAttachment('ctx_blank')).toBeUndefined() + }) + + it('repairs rows a rolled-back v1.4.198 host inserted after user_version reached 40', () => { + const dir = mkdtempSync(join(tmpdir(), 'orca-federated-home-run-')) + const dbPath = join(dir, 'orchestration.db') + try { + const upgraded = new OrchestrationDb(dbPath) + expect(upgraded.db.pragma('user_version', { simple: true })).toBe(SCHEMA_VERSION) + // v1.4.198's insert shape: no home_run_id column, so the DEFAULT '' lands. + upgraded.db.exec(`INSERT INTO remote_dispatch_attachments + (dispatch_id, task_id, home_peer_fingerprint, runtime_epoch) + VALUES ('ctx_rolled_back', 'task_rolled_back', 'home', 'epoch')`) + expect(upgraded.getRemoteDispatchAttachment('ctx_rolled_back')?.home_run_id).toBe('') + expect(() => + importFederatedControlMessage(upgraded, { + dispatchId: 'ctx_rolled_back', + messageId: 'message_refused', + payload: JSON.stringify({ from: 'home', subject: 'x', body: '', type: 'status' }) + }) + ).toThrow(/Run not found/) + upgraded.close() + + const reopened = new OrchestrationDb(dbPath) + try { + const stubRunId = federatedStubHomeRunId('ctx_rolled_back') + expect(reopened.getRemoteDispatchAttachment('ctx_rolled_back')?.home_run_id).toBe(stubRunId) + expect(reopened.getRunRaw(stubRunId)).toMatchObject({ home_database: 'remote', legacy: 0 }) + importInstruction(reopened, 'ctx_rolled_back', 'message_rolled_back') + expect(reopened.getMessageById('message_rolled_back')?.run_id).toBe(stubRunId) + } finally { + reopened.close() + } + } finally { + rmSync(dir, { recursive: true, force: true }) + } }) }) diff --git a/src/main/runtime/orchestration/db/schema/migrate-v40.ts b/src/main/runtime/orchestration/db/schema/migrate-v40.ts index 50ef46f82cc..e3cd46ca30d 100644 --- a/src/main/runtime/orchestration/db/schema/migrate-v40.ts +++ b/src/main/runtime/orchestration/db/schema/migrate-v40.ts @@ -1,11 +1,15 @@ import type { OrchestrationDb } from '../orchestration-db' +import { backfillFederatedStubHomeRuns } from '../federation/federated-stub-home-run-backfill' export function migrateV40(this: OrchestrationDb, current: number): void { - if (current >= 40 || this.hasColumn('remote_dispatch_attachments', 'home_run_id')) { + if (current >= 40) { return } - // Federation is unreleased; any development-only rows fail Run validation until reattached. - this.db.exec( - "ALTER TABLE remote_dispatch_attachments ADD COLUMN home_run_id TEXT NOT NULL DEFAULT ''" - ) + if (!this.hasColumn('remote_dispatch_attachments', 'home_run_id')) { + this.db.exec( + "ALTER TABLE remote_dispatch_attachments ADD COLUMN home_run_id TEXT NOT NULL DEFAULT ''" + ) + } + // Why: workers attached by v1.4.198 keep a mailbox; without a Run their control mail is refused. + backfillFederatedStubHomeRuns(this.db) } diff --git a/src/main/runtime/orchestration/orchestration-federated-legacy-probe.test.ts b/src/main/runtime/orchestration/orchestration-federated-legacy-probe.test.ts index ad08a7383f0..24d18f0baef 100644 --- a/src/main/runtime/orchestration/orchestration-federated-legacy-probe.test.ts +++ b/src/main/runtime/orchestration/orchestration-federated-legacy-probe.test.ts @@ -3,7 +3,7 @@ import { tmpdir } from 'node:os' import { join } from 'node:path' import { afterEach, describe, expect, it } from 'vitest' import { LEGACY_RUN_ID, OrchestrationDb } from './db' -import { SCHEMA_VERSION } from './db/contract-constants' +import { federatedStubHomeRunId, SCHEMA_VERSION } from './db/contract-constants' import { resolveOrchestrationMigrationStartVersion } from './orchestration-schema-version-skew' describe('federated mailbox legacy-adoption probe', () => { @@ -17,15 +17,22 @@ describe('federated mailbox legacy-adoption probe', () => { } }) - function seedMailbox(handle: string, kind: 'message' | 'delivery'): string { + function seedMailbox( + handle: string, + kind: 'message' | 'delivery', + homeRunId = 'run_home', + mailRunId = LEGACY_RUN_ID + ): string { directory = mkdtempSync(join(tmpdir(), 'orca-federated-legacy-probe-')) const path = join(directory, 'orchestration.db') db = new OrchestrationDb(path) - db.db.exec(` - INSERT INTO remote_dispatch_attachments ( - dispatch_id, task_id, home_peer_fingerprint, home_run_id, runtime_epoch, state - ) VALUES ('ctx_remote', 'task_remote', 'peer_home', 'run_home', 'epoch', 'ready'); - `) + db.db + .prepare( + `INSERT INTO remote_dispatch_attachments ( + dispatch_id, task_id, home_peer_fingerprint, home_run_id, runtime_epoch, state + ) VALUES ('ctx_remote', 'task_remote', 'peer_home', ?, 'epoch', 'ready')` + ) + .run(homeRunId) if (kind === 'message') { db.db .prepare( @@ -33,14 +40,14 @@ describe('federated mailbox legacy-adoption probe', () => { id, run_id, delivery_contract, from_handle, to_handle, subject, type ) VALUES ('msg_probe', ?, 'current_delivery', 'term_home', ?, 'continue', 'dispatch')` ) - .run(LEGACY_RUN_ID, handle) + .run(mailRunId, handle) } else { db.db .prepare( `INSERT INTO deliveries (id, run_id, mailbox_handle, consumer_generation, message_ids) VALUES ('delivery_probe', ?, ?, 0, '[]')` ) - .run(LEGACY_RUN_ID, handle) + .run(mailRunId, handle) } return path } @@ -70,6 +77,27 @@ describe('federated mailbox legacy-adoption probe', () => { } ) + it.each(['message', 'delivery'] as const)( + 'does not treat a stub-home-Run attachment %s as pre-Runs evidence', + (kind) => { + const stubRunId = federatedStubHomeRunId('ctx_remote') + const path = seedMailbox('dispatch:ctx_remote', kind, stubRunId, stubRunId) + db!.db + .prepare( + `INSERT INTO runs (id, objective, home_database, consumer_generation, legacy) + VALUES (?, 'Coordinated from peer_home', 'remote', 0, 0)` + ) + .run(stubRunId) + expect( + resolveOrchestrationMigrationStartVersion(db!.db, SCHEMA_VERSION, SCHEMA_VERSION) + ).toBe(SCHEMA_VERSION) + db!.close() + db = new OrchestrationDb(path) + expect(db.getLegacyAdoption()).toBeUndefined() + expect(db.getRemoteDispatchAttachment('ctx_remote')?.home_run_id).toBe(stubRunId) + } + ) + it.each(['message', 'delivery'] as const)( 'still replays adoption for a genuine legacy %s', (kind) => { diff --git a/src/main/runtime/rpc/methods/orchestration/federation/federation-start-schema.test.ts b/src/main/runtime/rpc/methods/orchestration/federation/federation-start-schema.test.ts new file mode 100644 index 00000000000..f00e5102d6d --- /dev/null +++ b/src/main/runtime/rpc/methods/orchestration/federation/federation-start-schema.test.ts @@ -0,0 +1,41 @@ +import { describe, expect, it } from 'vitest' +import { FederationAttachStartParams } from './federation-start-schema' + +// The request shape a v1.4.198 coordinator sends: no runId field at all. +const legacyRequest = { + dispatchId: 'ctx_legacy', + taskId: 'task_legacy', + taskSpec: 'Do the thing', + protocolVersion: 3, + worktree: 'feature-branch' +} + +describe('FederationAttachStartParams', () => { + it('parses a v1.4.198 request that carries no runId', () => { + const result = FederationAttachStartParams.safeParse(legacyRequest) + expect(result.success, result.success ? undefined : JSON.stringify(result.error.issues)).toBe( + true + ) + expect(result.success && result.data.runId).toBeUndefined() + }) + + it('keeps a v1.4.199 runId verbatim', () => { + const result = FederationAttachStartParams.parse({ ...legacyRequest, runId: 'run_home' }) + expect(result.runId).toBe('run_home') + }) + + // Pins OptionalString: '' and non-strings drop to undefined (a stub Run is minted downstream); + // whitespace-only passes the schema and is refused by createRemoteDispatchAttachment. + it('maps an empty or non-string runId to undefined but passes whitespace through', () => { + expect(FederationAttachStartParams.parse({ ...legacyRequest, runId: '' }).runId).toBeUndefined() + expect(FederationAttachStartParams.parse({ ...legacyRequest, runId: 7 }).runId).toBeUndefined() + expect(FederationAttachStartParams.parse({ ...legacyRequest, runId: ' ' }).runId).toBe(' ') + }) + + it('still requires the dispatch, task, spec, and worktree fields', () => { + for (const field of ['dispatchId', 'taskId', 'taskSpec', 'worktree'] as const) { + const { [field]: _dropped, ...rest } = legacyRequest + expect(FederationAttachStartParams.safeParse(rest).success, field).toBe(false) + } + }) +}) diff --git a/src/main/runtime/rpc/methods/orchestration/federation/federation-start-schema.ts b/src/main/runtime/rpc/methods/orchestration/federation/federation-start-schema.ts index 1e7257df27d..84ed57d58cc 100644 --- a/src/main/runtime/rpc/methods/orchestration/federation/federation-start-schema.ts +++ b/src/main/runtime/rpc/methods/orchestration/federation/federation-start-schema.ts @@ -3,7 +3,8 @@ import { OptionalFiniteNumber, OptionalString, requiredString } from '../../../s import { OptionalWorkerLaunchPreference } from '../worker/worker-start-schema' export const FederationAttachStartParams = z.object({ - runId: requiredString('Missing Run ID'), + /** Omitted by v1.4.198 coordinators; the worker host then mints a stub home Run. */ + runId: OptionalString, dispatchId: requiredString('Missing Dispatch ID'), taskId: requiredString('Missing Task ID'), taskSpec: requiredString('Missing Task spec'),