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_<dispatch>) 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 <sim@local>
2026-09-22 00:02:31 +00:00 · 2026-09-08 23:41:46 -07:00
parent 750e6ffada
commit addd9f3da7
10 changed files with 214 additions and 33 deletions
@@ -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

@@ -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 = '';
+  `)
+}
@@ -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,
@@ -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)
@@ -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,
@@ -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 })
+    }
  })
 })
@@ -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)
 }
@@ -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) => {
@@ -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)
+    }
+  })
+})
@@ -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'),