diff --git a/.github/workflows/pr.yml b/.github/workflows/pr.yml index 268b6ad66e3..f01f79a2f85 100644 --- a/.github/workflows/pr.yml +++ b/.github/workflows/pr.yml @@ -654,10 +654,12 @@ jobs: run: >- pnpm exec vitest run --config config/vitest.config.ts tests/e2e/cross-version-wire/release-checkout.unit.test.ts + tests/e2e/cross-version-wire/published-field-shape.unit.test.ts tests/e2e/cross-version-wire/cross-version-browser-placement.unit.test.ts tests/e2e/cross-version-wire/cross-version-terminal-wire.unit.test.ts tests/e2e/cross-version-wire/reported-lossy-initial-snapshot.unit.test.ts tests/e2e/cross-version-wire/cross-version-agent-session-wire.unit.test.ts + tests/e2e/cross-version-wire/cross-version-orchestration-wire.unit.test.ts managed_hook_node18: name: managed hooks on Node 18 diff --git a/config/scripts/cross-version-wire-lane-contract.test.mjs b/config/scripts/cross-version-wire-lane-contract.test.mjs new file mode 100644 index 00000000000..18c1ed7dcb6 --- /dev/null +++ b/config/scripts/cross-version-wire-lane-contract.test.mjs @@ -0,0 +1,46 @@ +import { readdirSync, readFileSync } from 'node:fs' +import { join, resolve } from 'node:path' +import { describe, expect, it } from 'vitest' +import { parse as parseYaml } from 'yaml' + +const projectDir = resolve(import.meta.dirname, '../..') +const prWorkflow = parseYaml(readFileSync(join(projectDir, '.github/workflows/pr.yml'), 'utf8')) + +describe('cross-version wire lane contract', () => { + const crossVersionStep = prWorkflow.jobs['cross-version-wire'].steps.find( + (step) => step.name === 'Old/new client and server compatibility journeys' + ) + const crossVersionDir = 'tests/e2e/cross-version-wire' + + it('runs every suite in the cross-version directory', () => { + // Why derive from the directory rather than pin a list: a new suite that is + // never named in the job is a lane that silently covers nothing, which is how + // orchestration went uncovered while the harness itself was healthy. + const onDisk = readdirSync(join(projectDir, crossVersionDir)) + .filter((name) => name.endsWith('.unit.test.ts')) + .map((name) => `${crossVersionDir}/${name}`) + .sort() + const listed = crossVersionStep.run + .split(/\s+/) + .filter((token) => token.startsWith(`${crossVersionDir}/`)) + .sort() + expect(onDisk.length).toBeGreaterThan(0) + expect(listed).toEqual(onDisk) + }) + + it('covers the orchestration federation surface', () => { + // The suite exists because #19689 shipped: a released coordinator lost its Run + // id on federationAttachStart and no lane paired two builds over that method. + expect(readdirSync(join(projectDir, crossVersionDir))).toContain( + 'cross-version-orchestration-wire.unit.test.ts' + ) + expect(crossVersionStep.run).toContain( + `${crossVersionDir}/cross-version-orchestration-wire.unit.test.ts` + ) + // Full history: the suite extracts two release tags, and a shallow clone has none. + const checkout = prWorkflow.jobs['cross-version-wire'].steps.find( + (step) => step.uses?.startsWith('actions/checkout') + ) + expect(checkout.with['fetch-depth']).toBe(0) + }) +}) diff --git a/config/scripts/pr-code-change-scope.mjs b/config/scripts/pr-code-change-scope.mjs index 8e88aa16df5..34b4cfb8468 100644 --- a/config/scripts/pr-code-change-scope.mjs +++ b/config/scripts/pr-code-change-scope.mjs @@ -124,7 +124,15 @@ const CROSS_VERSION_WIRE_PREFIXES = [ 'src/main/runtime/rpc/methods/session-tabs.ts', 'src/main/runtime/rpc/methods/structured-agent-session', 'src/main/runtime/rpc/methods/terminal', - 'src/renderer/src/runtime/remote-runtime-terminal-multiplexer' + 'src/renderer/src/runtime/remote-runtime-terminal-multiplexer', + // Federation pairs a desktop with a worker host on its own release, so both the + // coordinator-side callers and the host-side methods are cross-version surface. + 'src/main/runtime/orchestration/federation', + 'src/main/runtime/rpc/methods/orchestration/federation/', + 'src/main/runtime/rpc/methods/orchestration/worker/worker-observation', + 'src/main/runtime/rpc/methods/orchestration/worker/worker-legacy-federated-read', + 'src/main/runtime/rpc/orchestration-contract-fence', + 'src/main/runtime/rpc/orchestration-mutation-executor' ] const MANAGED_HOOK_PREFIXES = [ diff --git a/config/scripts/pr-code-change-scope.test.mjs b/config/scripts/pr-code-change-scope.test.mjs index e622dd8603a..89d00c0905a 100644 --- a/config/scripts/pr-code-change-scope.test.mjs +++ b/config/scripts/pr-code-change-scope.test.mjs @@ -277,7 +277,26 @@ describe('per-job path classification', () => { 'src/main/runtime/rpc/methods/structured-agent-session-hold.ts', 'src/main/runtime/rpc/methods/structured-agent-session-schemas.ts', 'src/main/runtime/rpc/methods/terminal.ts', - 'src/renderer/src/runtime/remote-runtime-terminal-multiplexer.ts' + 'src/renderer/src/runtime/remote-runtime-terminal-multiplexer.ts', + // Orchestration federation: the coordinator side that composes what goes on + // the wire, and the host side that validates and publishes the answer. Both + // pair against a peer on a different release, so both must select the lane — + // #19542 changed only host-side files and this job never ran. + 'src/main/runtime/orchestration/federation-sync.ts', + 'src/main/runtime/orchestration/federation-sync-capability.ts', + 'src/main/runtime/orchestration/federation-control-message.ts', + 'src/main/runtime/orchestration/federation-lifecycle-settlement.ts', + 'src/main/runtime/rpc/methods/orchestration/federation/federation.ts', + 'src/main/runtime/rpc/methods/orchestration/federation/federation-start-schema.ts', + 'src/main/runtime/rpc/methods/orchestration/federation/federation-control.ts', + 'src/main/runtime/rpc/methods/orchestration/federation/federation-relay.ts', + 'src/main/runtime/rpc/methods/orchestration/federation/federated-worker-start.ts', + 'src/main/runtime/rpc/methods/orchestration/federation/federated-fleet-snapshot.ts', + 'src/main/runtime/rpc/methods/orchestration/federation/federated-attach-receipt.ts', + 'src/main/runtime/rpc/methods/orchestration/worker/worker-observation.ts', + 'src/main/runtime/rpc/methods/orchestration/worker/worker-legacy-federated-read.ts', + 'src/main/runtime/rpc/orchestration-contract-fence.ts', + 'src/main/runtime/rpc/orchestration-mutation-executor.ts' ]) { expectClassification([file], { 'cross-version-wire': true, diff --git a/docs/reference/remote-wire-compatibility.md b/docs/reference/remote-wire-compatibility.md index 13741c03789..62d5f0e47d4 100644 --- a/docs/reference/remote-wire-compatibility.md +++ b/docs/reference/remote-wire-compatibility.md @@ -154,10 +154,41 @@ Run it with: pnpm exec vitest run --config config/vitest.config.ts tests/e2e/cross-version-wire/cross-version-agent-session-wire.unit.test.ts ``` -The harness covers the terminal stream and the structured agent-session surface. It does -**not** cover the session-tab sync channel, legacy agent-session publications, file or Git -RPCs, mobile/E2EE framing, or the relay transport. A change on those paths still needs its -own reasoning against the three rules above. +`tests/e2e/cross-version-wire/cross-version-orchestration-wire.unit.test.ts` pairs the +orchestration federation surface: `federationAttachStart`, `federationFleetSnapshot`, +`federationShow`, `federationRead`, and the `federationPull`/`federationAck`/ +`federationImport` relay loop. A coordinator build's own `startFederatedWorker` composes +the attach params and its own sync loop drives the relay, so the suite fails on what a +real release sends rather than on a hand-written payload. + +It pairs against the **two** newest release tags, not one. A worker host and the desktop +that dispatches to it update on their own schedules, so a coordinator one release behind +is an ordinary peer for weeks. The newest tag already contains everything fixed during +the last cycle: pairing only against it cannot see a break introduced and repaired inside +that window, which is how a v1.4.198 coordinator losing its Run id on +`federationAttachStart` reached users (#19689). `resolveBaselineReleaseRefs(2)` in +`release-checkout.ts` derives both refs from `git tag`; no version is written down. + +Run it with: + +```bash +pnpm exec vitest run --config config/vitest.config.ts tests/e2e/cross-version-wire/cross-version-orchestration-wire.unit.test.ts +``` + +It fails when a released coordinator's params are refused by the current host's schema +(Rule 2 — this is #19689's class), when the current host stops publishing a field a +released peer's receipt parser reads (Rule 3, compared as dotted paths so a removal +inside `setup`, `launch`, `attachment` or `observation` cannot hide), when an +`orchestration.*` method name a released peer still calls disappears, or when a +coordinator throws instead of degrading against a host that lacks a method. Whether the +old side has a method or a caller module at all is read from its checkout — never +asserted as a literal. + +The harness covers the terminal stream, the structured agent-session surface, and +orchestration federation. It does **not** cover the session-tab sync channel, legacy +agent-session publications, file or Git RPCs, mobile/E2EE framing, or the relay +transport. A change on those paths still needs its own reasoning against the three rules +above. ## Worked example: `agentWait` on terminal and worker reads diff --git a/tests/e2e/cross-version-wire/cross-version-orchestration-wire.unit.test.ts b/tests/e2e/cross-version-wire/cross-version-orchestration-wire.unit.test.ts new file mode 100644 index 00000000000..d89ff212260 --- /dev/null +++ b/tests/e2e/cross-version-wire/cross-version-orchestration-wire.unit.test.ts @@ -0,0 +1,311 @@ +// Cross-version coverage for the orchestration federation surface, paired the same +// way the terminal and agent-session harnesses are: current code against real +// published releases, in both skew directions, over one scripted journey. +// +// Two baselines rather than one. A worker host and the desktop that dispatches to it +// update on their own schedules, so a coordinator one release behind is an ordinary +// peer for weeks. The newest tag already contains everything fixed during the last +// cycle — pairing only against it cannot see a break introduced and repaired inside +// that window, which is exactly how the v1.4.198 coordinator losing its Run id on +// `federationAttachStart` reached users (#19689). +// +// Nothing below writes down what a release has. Every param is composed by that +// build's own coordinator module, every response by that build's own dispatcher, and +// every "the old side lacks X" is read from the extracted checkout. + +import { beforeAll, describe, expect, it } from 'vitest' +import { + findCall, + publishedBy, + publishedFieldPaths, + runOrchestrationSkewJourney, + type OrchestrationSkewRecord +} from './orchestration-skew-journey' +import { comparePublishedFields } from './published-field-shape' +import { resolveBaselineReleaseRefs } from './release-checkout' +import { + loadOrchestrationWireBuild, + WORKING_TREE, + type OrchestrationWireBuild +} from './versioned-orchestration-wire' + +// Why: a cold CI run extracts two release checkouts before the first pairing. +const SUITE_TIMEOUT_MS = 300_000 + +/** A desktop pinned at the prior release is a normal federation peer for weeks. */ +const BASELINE_COUNT = 2 + +const ATTACH = 'orchestration.federationAttachStart' +const FLEET_SNAPSHOT = 'orchestration.federationFleetSnapshot' +const SHOW = 'orchestration.federationShow' +const READ = 'orchestration.federationRead' +const PULL = 'orchestration.federationPull' +const IMPORT = 'orchestration.federationImport' + +/** Every method whose published payload a coordinator decodes on this journey. */ +const PUBLISHED_METHODS = [ATTACH, SHOW, READ, PULL, IMPORT, FLEET_SNAPSHOT] + +/** + * Runtime seams the RPC dispatcher touches on every method, federated or not: + * client-hosted browser routing and feature telemetry. They are named here so any + * *other* gap in the host stub fails loudly instead of returning undefined and + * reading as a wire break. + */ +const DISPATCHER_PLUMBING = new Set(['routeClientHostedBrowserRpc', 'recordFeatureInteraction']) + +type BaselinePairings = { + ref: string + build: OrchestrationWireBuild + /** Old coordinator against the working tree host. */ + oldClient: OrchestrationSkewRecord + /** Working tree coordinator against the old host. */ + oldHost: OrchestrationSkewRecord + /** The old build talking to itself: what that release publishes today. */ + reference: OrchestrationSkewRecord +} + +// Resolved at collection time so each baseline gets its own named describe block; +// a lane that quietly ran one pairing instead of two is the failure to avoid. +const BASELINE_REFS = resolveBaselineReleaseRefs(BASELINE_COUNT) + +let current: OrchestrationWireBuild +let currentReference: OrchestrationSkewRecord +const pairings = new Map() + +function pairingFor(ref: string): BaselinePairings { + const pairing = pairings.get(ref) + if (!pairing) { + throw new Error(`Cross-version orchestration pairing for ${ref} was never built`) + } + return pairing +} + +beforeAll(async () => { + current = await loadOrchestrationWireBuild(WORKING_TREE) + currentReference = await runOrchestrationSkewJourney({ + clientBuild: current, + hostBuild: current + }) + for (const ref of BASELINE_REFS) { + const build = await loadOrchestrationWireBuild(ref) + pairings.set(ref, { + ref, + build, + oldClient: await runOrchestrationSkewJourney({ clientBuild: build, hostBuild: current }), + oldHost: await runOrchestrationSkewJourney({ clientBuild: current, hostBuild: build }), + reference: await runOrchestrationSkewJourney({ clientBuild: build, hostBuild: build }) + }) + } +}, SUITE_TIMEOUT_MS) + +function orchestrationMethods(build: OrchestrationWireBuild): string[] { + return build.methodNames.filter((name) => name.startsWith('orchestration.')) +} + +function describeRecord(record: OrchestrationSkewRecord): string { + return `${record.clientLabel} coordinator -> ${record.hostLabel} host` +} + +function stallDetail(record: OrchestrationSkewRecord): string { + return JSON.stringify({ completed: record.completed, stepErrors: record.stepErrors }) +} + +describe('cross-version orchestration federation wire', () => { + it( + 'skews current code against two real published releases', + () => { + expect(BASELINE_REFS.length).toBe(BASELINE_COUNT) + expect(new Set(BASELINE_REFS).size).toBe(BASELINE_COUNT) + for (const pairing of pairings.values()) { + expect(pairing.build.revision).toMatch(/^[0-9a-f]{40}$/) + expect(pairing.build.revision).not.toBe(current.revision) + // Anti-vacuous: an empty registry would make every "this release does not + // register X" claim below meaningless. + expect(pairing.build.methodNames).toContain('terminal.create') + expect(orchestrationMethods(pairing.build).length).toBeGreaterThan(0) + } + expect(new Set([...pairings.values()].map((entry) => entry.build.revision)).size).toBe( + BASELINE_COUNT + ) + }, + SUITE_TIMEOUT_MS + ) + + it('reaches the whole journey when both sides are current code', () => { + expect(currentReference.stepErrors, stallDetail(currentReference)).toEqual([]) + expect(currentReference.completed).toEqual([ + 'attach-start', + 'fleet-snapshot', + 'worker-show', + 'worker-read', + 'relay-sync' + ]) + }) + + it('keeps one orchestration contract version across every paired build', () => { + // A bump fences every peer that has not updated, on every orchestration + // mutation at once. That is a deliberate flag day, never a side effect. + for (const pairing of pairings.values()) { + expect( + pairing.build.orchestrationContractVersion, + `${pairing.ref} would be fenced out of every orchestration mutation` + ).toBe(current.orchestrationContractVersion) + } + }) + + describe.each(BASELINE_REFS)('against %s', (ref) => { + it('never drops an orchestration method name a released peer may still call', () => { + const pairing = pairingFor(ref) + // A renamed or removed verb is Rule 3 for an `orca` CLI or coordinator that + // has not updated: it calls the old name and gets method_not_found. + const removed = orchestrationMethods(pairing.build).filter( + (name) => !current.methodNames.includes(name) + ) + expect(removed, `${pairing.ref} registers methods current code dropped`).toEqual([]) + }) + + it('registers the attach method on both sides rather than skipping the pairing', () => { + const pairing = pairingFor(ref) + // A supported peer without the attach method is the bug, not a reason to skip: + // the coordinator would have nothing to degrade to. + expect(pairing.build.methodNames).toContain(ATTACH) + expect(current.methodNames).toContain(ATTACH) + }) + + it('starts a federated worker from the old coordinator against the new host', () => { + const pairing = pairingFor(ref) + const record = pairing.oldClient + const attach = findCall(record, ATTACH) + expect(attach, `${describeRecord(record)} never called ${ATTACH}`).toBeDefined() + expect( + attach?.error, + `${describeRecord(record)} was refused: ${JSON.stringify(attach?.error)}` + ).toBeNull() + expect(record.startReceipt?.state, stallDetail(record)).toBe('ready') + expect(record.completed).toContain('attach-start') + // The host must bind the attachment to a Run whatever the coordinator sent, + // or every later control-mail import fails its `requireRun`. + expect(record.hostHomeRunId, 'the new host bound no home Run').toBeTruthy() + }) + + it('starts a federated worker from the new coordinator against the old host', () => { + const pairing = pairingFor(ref) + const record = pairing.oldHost + const attach = findCall(record, ATTACH) + expect( + attach?.error, + `${describeRecord(record)} was refused: ${JSON.stringify(attach?.error)}` + ).toBeNull() + expect(record.startReceipt?.state, stallDetail(record)).toBe('ready') + }) + + it('has each build accept the other build’s outgoing attach params', () => { + const pairing = pairingFor(ref) + // Read from what the builds really sent, so neither side's required-field + // list is written down here. + const fromOld = findCall(pairing.oldClient, ATTACH)?.params + const fromNew = findCall(pairing.oldHost, ATTACH)?.params + expect(fromOld).toBeTruthy() + expect(fromNew).toBeTruthy() + expect(() => current.parseAttachStartParams(fromOld)).not.toThrow() + expect(() => pairing.build.parseAttachStartParams(fromNew)).not.toThrow() + }) + + it.each(PUBLISHED_METHODS)('still publishes every %s field the old peer reads', (method) => { + const pairing = pairingFor(ref) + // Rule 3, both sides read from a pairing rather than from a list: what the + // release publishes to a coordinator of its own version, against what current + // code publishes to a coordinator of its own version. + const older = publishedBy(pairing.reference, method) + const newer = publishedBy(currentReference, method) + if (Object.keys(older).length === 0) { + // The release never published this method at all; its absence is covered + // by the degradation test below, not by a field comparison. + expect(findCall(pairing.reference, method)?.result ?? null).toBeNull() + return + } + expect(Object.keys(newer).length, `current code published no ${method}`).toBeGreaterThan(0) + const skew = comparePublishedFields({ + older: publishedFieldPaths(older), + newer: publishedFieldPaths(newer) + }) + expect(skew.removed, `${method} dropped fields a ${pairing.ref} peer reads`).toEqual([]) + }) + + it('degrades instead of throwing when the old host has no fleet snapshot', () => { + const pairing = pairingFor(ref) + const record = pairing.oldHost + const hostHasMethod = pairing.build.methodNames.includes(FLEET_SNAPSHOT) + expect(record.fleet, stallDetail(record)).not.toBeNull() + if (hostHasMethod) { + expect(record.fleet?.errorCodes).toEqual([]) + expect(Object.keys(record.fleet?.observations ?? {}).length).toBe(1) + return + } + // The coordinator must turn a missing method into a named degradation, not a + // thrown call and not a fabricated verdict. + expect(record.fleet?.errorCodes).toEqual(['capability_unsupported']) + expect(record.fleet?.observations).toEqual({}) + expect(record.completed).toContain('fleet-snapshot') + }) + + it('reads and shows the worker across the skew in both directions', () => { + const pairing = pairingFor(ref) + for (const record of [pairing.oldClient, pairing.oldHost]) { + expect(record.completed, `${describeRecord(record)}: ${stallDetail(record)}`).toContain( + 'worker-show' + ) + expect(record.completed, `${describeRecord(record)}: ${stallDetail(record)}`).toContain( + 'worker-read' + ) + } + }) + + it('relays mail both ways across the skew', () => { + const pairing = pairingFor(ref) + for (const record of [pairing.oldClient, pairing.oldHost]) { + expect(record.completed, `${describeRecord(record)}: ${stallDetail(record)}`).toContain( + 'relay-sync' + ) + // The worker's report was pulled and acknowledged... + expect(record.syncResult, describeRecord(record)).toMatchObject({ + imported: 1, + acknowledgedThrough: 1 + }) + // ...and the coordinator's control mail reached the worker's mailbox, + // which the host can only do once it resolves the attachment's home Run. + expect( + record.controlMailImported, + `${describeRecord(record)} imported no coordinator control mail` + ).toBe(1) + } + }) + + it('exercises the whole federation surface rather than a stubbed-out one', () => { + const pairing = pairingFor(ref) + for (const record of [pairing.oldClient, pairing.oldHost, pairing.reference]) { + const gaps = [ + ...record.missingHostRuntimeMethods, + ...record.missingClientRuntimeMethods + ].filter((name) => !DISPATCHER_PLUMBING.has(name)) + expect(gaps, `${describeRecord(record)} needed unstubbed runtime methods`).toEqual([]) + } + }) + + it('only ever loses a step the old build genuinely cannot take', () => { + const pairing = pairingFor(ref) + // A coordinator that predates a caller cannot make that call; anything else + // failing is a skew break rather than a version fact. + const clientHasFleetCaller = pairing.build.readFederatedFleetSnapshots !== null + const expectedOldClientSteps = clientHasFleetCaller + ? currentReference.completed + : currentReference.completed.filter((step) => step !== 'fleet-snapshot') + expect(pairing.oldClient.completed, stallDetail(pairing.oldClient)).toEqual( + expectedOldClientSteps + ) + expect(pairing.oldHost.completed, stallDetail(pairing.oldHost)).toEqual( + currentReference.completed + ) + }) + }) +}) diff --git a/tests/e2e/cross-version-wire/orchestration-federation-peers.ts b/tests/e2e/cross-version-wire/orchestration-federation-peers.ts new file mode 100644 index 00000000000..c58ad1e91b0 --- /dev/null +++ b/tests/e2e/cross-version-wire/orchestration-federation-peers.ts @@ -0,0 +1,244 @@ +import type { OrchestrationWireBuild, RpcEnvelope } from './versioned-orchestration-wire' + +/** + * The two ends of a federation pairing, each built from one release. + * + * The host is that build's real RPC dispatcher over that build's real orchestration + * store; the Run home is that build's real coordinator-side callers whose + * `callOrchestrationWorkerServer` forwards to the peer's dispatcher over a JSON + * round trip. Nothing between them is hand-written, so a pairing fails on what + * those builds really send and publish. + */ + +export const FEDERATION_FIXTURES = { + environmentId: 'environment_worker_host', + environmentName: 'worker-host', + hostPeerFingerprint: 'worker_host_peer_fingerprint', + homePeerFingerprint: 'run_home_peer_fingerprint', + worktreeSelector: 'id:repo::remote-worktree', + worktreeId: 'repo::remote-worktree', + workerTerminal: 'term_remote_worker', + coordinatorTerminal: 'term_coord', + coordinatorPaneKey: 'tab_coord:aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa', + workerPaneKey: 'tab_worker:bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb', + workerIncarnation: 'worker_host_epoch:pty:1', + taskSpec: 'Cross-version federation journey' +} as const + +export type WireCall = { + method: string + /** Params exactly as they left the coordinator, after the JSON round trip. */ + params: Record | null + /** Result exactly as the host published it, or null when the call was refused. */ + result: unknown + error: { code: string; message: string } | null +} + +export type OrchestrationStoreHandle = ReturnType + +export type FederationHostPeer = { + store: OrchestrationStoreHandle + runtime: unknown + missing: string[] + call: ( + method: string, + params: unknown, + envelope: { requestId?: string; contractVersion?: number } + ) => Promise +} + +export type FederationHomePeer = { + store: OrchestrationStoreHandle + runtime: unknown + missing: string[] + runId: string + taskId: string +} + +/** Record every method a build asked for that the stub lacks, by name, so a gap + * fails naming what to add here instead of as a TypeError that reads like a break. */ +function stubRuntime(methods: Record): { runtime: unknown; missing: string[] } { + const missing: string[] = [] + const runtime = new Proxy(methods, { + get(target, property, receiver) { + if (typeof property === 'string' && !(property in target)) { + if (!missing.includes(property)) { + missing.push(property) + } + return () => undefined + } + return Reflect.get(target, property, receiver) + } + }) + return { runtime, missing } +} + +export function storeCall( + store: OrchestrationStoreHandle, + method: string, + ...args: unknown[] +): T { + const fn = store[method] + if (typeof fn !== 'function') { + throw new Error(`Orchestration store has no ${method}; the journey needs updating.`) + } + return (fn as (...rest: unknown[]) => T).apply(store, args) +} + +/** The worker host: this build's real dispatcher over this build's real store. */ +export function createFederationHostPeer(build: OrchestrationWireBuild): FederationHostPeer { + const store = build.createStore() + const terminal = { + handle: FEDERATION_FIXTURES.workerTerminal, + worktreeId: FEDERATION_FIXTURES.worktreeId, + status: 'running', + connected: true + } + const { runtime, missing } = stubRuntime({ + getRuntimeId: () => `worker_host_epoch:${build.label}`, + // Capabilities come from the host build's own source, so a coordinator's + // negotiation sees what that release could really advertise. + getStatus: () => ({ + runtimeId: `worker_host_epoch:${build.label}`, + protocolVersion: build.protocolVersion, + capabilities: build.capabilities + }), + getOrchestrationDb: () => store, + getNestedWorkerMaxDepth: () => 3, + validateOrchestrationAgentLauncher: () => {}, + showManagedTerminalWorkspace: async () => ({ + id: FEDERATION_FIXTURES.worktreeId, + repoId: 'repo' + }), + showTerminal: async () => terminal, + isTerminalRunningAgent: async () => true, + listTerminals: async () => ({ terminals: [terminal], totalCount: 1, truncated: false }), + waitForTerminal: async () => ({ + handle: FEDERATION_FIXTURES.workerTerminal, + condition: 'tui-idle', + satisfied: true, + status: 'running', + exitCode: null + }), + getOrchestrationDispatchAuthority: () => ({ + paneKey: FEDERATION_FIXTURES.workerPaneKey, + processIncarnation: FEDERATION_FIXTURES.workerIncarnation, + hostScope: { kind: 'local' } + }), + getTerminalPaneKey: () => FEDERATION_FIXTURES.workerPaneKey, + getTerminalProcessIncarnation: () => FEDERATION_FIXTURES.workerIncarnation, + getTerminalLivenessVerdict: () => ({ + status: 'live', + ptyIds: [FEDERATION_FIXTURES.workerTerminal] + }), + getTerminalOrchestrationCliCommand: () => 'orca', + sendTerminalAgentPrompt: async () => ({ + handle: FEDERATION_FIXTURES.workerTerminal, + accepted: true, + bytesWritten: 1 + }), + readTerminal: async () => ({ + handle: FEDERATION_FIXTURES.workerTerminal, + status: 'running', + entries: [{ cursor: 1, text: 'remote worker output' }], + nextCursor: '1', + limited: false + }), + notifyMessageArrived: () => {} + }) + const dispatcher = build.createDispatcher(runtime) + return { + store, + runtime, + missing, + call: (method, params, envelope) => + dispatcher.dispatch( + { + id: `${build.label}-${method}`, + method, + params, + ...(envelope.requestId ? { orchestrationRequestId: envelope.requestId } : {}), + ...(envelope.contractVersion === undefined + ? {} + : { orchestrationContractVersion: envelope.contractVersion }) + }, + { authenticatedCallerFingerprint: FEDERATION_FIXTURES.homePeerFingerprint } + ) + } +} + +/** The Run home: this build's coordinator-side callers over its own store. */ +export function createFederationHomePeer( + build: OrchestrationWireBuild, + host: FederationHostPeer, + calls: WireCall[] +): FederationHomePeer { + const store = build.createStore() + const run = storeCall<{ id: string }>(store, 'createRun', { + objective: FEDERATION_FIXTURES.taskSpec, + coordinatorHandle: FEDERATION_FIXTURES.coordinatorTerminal, + coordinatorPaneKey: FEDERATION_FIXTURES.coordinatorPaneKey + }) + const task = storeCall<{ id: string }>(store, 'createTask', { + spec: FEDERATION_FIXTURES.taskSpec, + runId: run.id, + createdByTerminalHandle: FEDERATION_FIXTURES.coordinatorTerminal, + createdByPaneKey: FEDERATION_FIXTURES.coordinatorPaneKey + }) + const { runtime, missing } = stubRuntime({ + getRuntimeId: () => `run_home_epoch:${build.label}`, + getOrchestrationDb: () => store, + getNestedWorkerMaxDepth: () => 3, + validateOrchestrationAgentLauncher: () => {}, + resolveOrchestrationWorkerServer: () => ({ + environmentId: FEDERATION_FIXTURES.environmentId, + name: FEDERATION_FIXTURES.environmentName, + peerFingerprint: FEDERATION_FIXTURES.hostPeerFingerprint, + pairingRevision: 1 + }), + getOrchestrationDispatchAuthority: () => ({ + paneKey: FEDERATION_FIXTURES.coordinatorPaneKey, + processIncarnation: 'run_home_epoch:pty:1' + }), + getTerminalPaneKey: () => FEDERATION_FIXTURES.coordinatorPaneKey, + getTerminalProcessIncarnation: () => 'run_home_epoch:pty:1', + ensureOrchestrationFederationRelay: () => {}, + stopOrchestrationFederationRelay: () => {}, + syncOrchestrationFederatedDispatchAfterCurrent: async () => undefined, + notifyMessageArrived: () => {}, + callOrchestrationWorkerServer: async ( + _environmentId: string, + method: string, + params: unknown, + _timeoutMs?: number, + meta?: { orchestrationRequestId?: string } + ) => { + // The wire, exactly: an `undefined` field never reaches the far side. + const wireParams = + params === undefined ? undefined : (JSON.parse(JSON.stringify(params)) as unknown) + const reply = await host.call(method, wireParams, { + requestId: meta?.orchestrationRequestId, + // The transport stamps the coordinator build's own contract version on + // every orchestration call; the host fences anything it does not know. + ...(method.startsWith('orchestration.') + ? { contractVersion: build.orchestrationContractVersion } + : {}) + }) + const entry: WireCall = { + method, + params: (wireParams as Record | undefined) ?? null, + result: reply.ok ? (reply.result ?? null) : null, + error: reply.ok ? null : (reply.error ?? { code: 'unknown', message: 'no error payload' }) + } + calls.push(entry) + if (!reply.ok) { + throw build.createRpcError( + entry.error?.code ?? 'unknown', + entry.error?.message ?? 'The worker server refused the call.' + ) + } + return reply.result + } + }) + return { store, runtime, missing, runId: run.id, taskId: task.id } +} diff --git a/tests/e2e/cross-version-wire/orchestration-skew-journey.ts b/tests/e2e/cross-version-wire/orchestration-skew-journey.ts new file mode 100644 index 00000000000..6fa27674574 --- /dev/null +++ b/tests/e2e/cross-version-wire/orchestration-skew-journey.ts @@ -0,0 +1,299 @@ +import { + createFederationHomePeer, + createFederationHostPeer, + FEDERATION_FIXTURES, + storeCall, + type WireCall +} from './orchestration-federation-peers' +import type { OrchestrationWireBuild } from './versioned-orchestration-wire' + +/** + * One scripted federation journey, run byte-identically for every coordinator/host + * version pairing: a coordinator starts a remote worker, watches the fleet, reads + * and shows it, and relays mail both ways. + * + * Every step ends on a recorded fact — a wire call, a persisted row, a receipt the + * coordinator's own parser accepted — never on elapsed time. + */ + +export const JOURNEY_STEPS = [ + 'attach-start', + 'fleet-snapshot', + 'worker-show', + 'worker-read', + 'relay-sync' +] as const + +export type JourneyStep = (typeof JOURNEY_STEPS)[number] + +const WORKER_REPORT_SUBJECT = 'worker status' +const CONTROL_MAIL_SUBJECT = 'coordinator control mail' +const READINESS_TIMEOUT_MS = 60_000 + +export type OrchestrationSkewRecord = { + clientLabel: string + hostLabel: string + clientRevision: string + hostRevision: string + /** Steps that actually completed, in order. The liveness oracle. */ + completed: JourneyStep[] + /** Every request/response pair the journey put on the wire, in order. */ + calls: WireCall[] + /** What `startFederatedWorker` handed back to the coordinator. */ + startReceipt: Record | null + /** Fleet verdicts the coordinator derived, and the host errors it degraded to. */ + fleet: { + observations: Record + errorCodes: string[] + } | null + /** The Run the host bound the attachment to. A coordinator that sends none still + * needs one here, or every later control-mail import fails `requireRun`. */ + hostHomeRunId: string | null + /** How many relay items the host reported importing from the coordinator. */ + controlMailImported: number + syncResult: { imported: number; acknowledgedThrough: number } | null + missingHostRuntimeMethods: string[] + missingClientRuntimeMethods: string[] + stepErrors: { step: JourneyStep; message: string }[] +} + +/** A pairing that never advanced past a step, carrying the partial record. */ +export class OrchestrationSkewStall extends Error { + readonly step: JourneyStep + readonly record: OrchestrationSkewRecord + + constructor(step: JourneyStep, detail: string, record: OrchestrationSkewRecord) { + super(`Cross-version orchestration journey stalled at ${step}: ${detail}`) + this.name = 'OrchestrationSkewStall' + this.step = step + this.record = record + } +} + +function published(call: WireCall | undefined): Record { + const result = call?.result + return result && typeof result === 'object' && !Array.isArray(result) + ? (result as Record) + : {} +} + +export function findCall(record: OrchestrationSkewRecord, method: string): WireCall | undefined { + return record.calls.find((call) => call.method === method) +} + +export function publishedBy( + record: OrchestrationSkewRecord, + method: string +): Record { + return published(findCall(record, method)) +} + +/** + * Dotted names for every field in a published payload, nested objects included. + * + * A federation receipt carries fields a coordinator reads inside `setup`, `launch`, + * `attachment` and `observation`; comparing only the top level would let a removal + * one level down pass. Arrays are walked through their first element, because every + * item in these payloads is the same shape. + */ +export function publishedFieldPaths(payload: unknown, prefix = ''): string[] { + if (Array.isArray(payload)) { + return payload.length === 0 ? [] : publishedFieldPaths(payload[0], `${prefix}[]`) + } + if (!payload || typeof payload !== 'object') { + return prefix ? [prefix] : [] + } + return Object.entries(payload as Record) + .flatMap(([key, value]) => { + const path = prefix ? `${prefix}.${key}` : key + const nested = publishedFieldPaths(value, path) + return nested.length === 0 ? [path] : nested + }) + .sort() +} + +/** + * Drive one federation journey for a coordinator build against a host build. + * + * Steps a build genuinely cannot take — a coordinator with no fleet-snapshot caller, + * a host with no such method — are recorded rather than thrown, so the suite can + * assert the degradation instead of losing the rest of the journey. + */ +export async function runOrchestrationSkewJourney(args: { + clientBuild: OrchestrationWireBuild + hostBuild: OrchestrationWireBuild +}): Promise { + const { clientBuild, hostBuild } = args + const calls: WireCall[] = [] + const host = createFederationHostPeer(hostBuild) + const client = createFederationHomePeer(clientBuild, host, calls) + const record: OrchestrationSkewRecord = { + clientLabel: clientBuild.label, + hostLabel: hostBuild.label, + clientRevision: clientBuild.revision, + hostRevision: hostBuild.revision, + completed: [], + calls, + startReceipt: null, + fleet: null, + hostHomeRunId: null, + controlMailImported: 0, + syncResult: null, + missingHostRuntimeMethods: host.missing, + missingClientRuntimeMethods: client.missing, + stepErrors: [] + } + + const step = async (name: JourneyStep, run: () => Promise): Promise => { + try { + await run() + record.completed.push(name) + } catch (error) { + record.stepErrors.push({ + step: name, + message: error instanceof Error ? error.message : String(error) + }) + } + } + + try { + // 1. worker-start against a federated target. The coordinator build composes + // every attach param itself; the host validates with its own schema. + const receipt = (await clientBuild.startFederatedWorker({ + params: { + on: FEDERATION_FIXTURES.environmentId, + worktree: FEDERATION_FIXTURES.worktreeSelector, + terminal: FEDERATION_FIXTURES.workerTerminal, + spec: FEDERATION_FIXTURES.taskSpec, + from: FEDERATION_FIXTURES.coordinatorTerminal, + timeoutMs: READINESS_TIMEOUT_MS + }, + runtime: client.runtime, + db: client.store, + runId: client.runId, + task: { id: client.taskId, spec: FEDERATION_FIXTURES.taskSpec, status: 'pending' }, + orchestrationMutation: { + callerFingerprint: FEDERATION_FIXTURES.homePeerFingerprint, + requestId: `attach_${clientBuild.label}_${hostBuild.label}`, + method: 'orchestration.workerStart', + payloadHash: 'cross_version_attach_payload' + } + })) as Record + record.startReceipt = receipt + const attach = findCall(record, 'orchestration.federationAttachStart') + if (!attach) { + throw new OrchestrationSkewStall( + 'attach-start', + 'the coordinator never called orchestration.federationAttachStart', + record + ) + } + if (attach.error) { + record.stepErrors.push({ + step: 'attach-start', + message: `${attach.error.code}: ${attach.error.message}` + }) + } else { + record.completed.push('attach-start') + } + const dispatchId = String(receipt.dispatchId ?? '') + if (!dispatchId) { + return record + } + record.hostHomeRunId = + storeCall<{ home_run_id?: string } | undefined>( + host.store, + 'getRemoteDispatchAttachment', + dispatchId + )?.home_run_id ?? null + const federated = storeCall | undefined>( + client.store, + 'getFederatedDispatch', + dispatchId + ) + if (!federated) { + return record + } + + // 2. The fleet sweep the coordinator runs behind `worker-list`. + await step('fleet-snapshot', async () => { + const read = clientBuild.readFederatedFleetSnapshots + if (!read) { + throw new Error(`${clientBuild.label} has no fleet-snapshot caller`) + } + const snapshot = await read({ + runtime: client.runtime, + db: client.store, + dispatchIds: [dispatchId] + }) + record.fleet = { + observations: Object.fromEntries(snapshot.observations), + errorCodes: snapshot.errors.map((error) => error.code).sort() + } + }) + + // 3/4. The read paths behind `worker-show` and `worker-read --include-remote`. + await step('worker-show', async () => { + await clientBuild.callFederatedWorkerShow(client.runtime, federated) + }) + await step('worker-read', async () => { + await clientBuild.readLegacyFederatedTerminal({ + runtime: client.runtime, + server: { + environmentId: FEDERATION_FIXTURES.environmentId, + name: FEDERATION_FIXTURES.environmentName, + peerFingerprint: FEDERATION_FIXTURES.hostPeerFingerprint, + pairingRevision: 1 + }, + federated, + workerState: 'ready', + dispatchId, + source: undefined, + cursor: undefined, + limit: 10 + }) + }) + + // 5. Mail both ways over the relay: the worker's report is pulled and + // acknowledged, and the coordinator's control mail is imported. + await step('relay-sync', async () => { + storeCall(host.store, 'enqueueFederationRelay', { + dispatchId, + direction: 'to_home', + kind: 'message', + payload: JSON.stringify({ + from: `dispatch:${dispatchId}`, + subject: WORKER_REPORT_SUBJECT, + body: 'halfway through', + type: 'status', + priority: 'normal' + }) + }) + storeCall(client.store, 'enqueueFederationRelay', { + dispatchId, + direction: 'to_worker', + kind: 'control_message', + payload: clientBuild.encodeControlMessage({ + from: `run:${client.runId}`, + subject: CONTROL_MAIL_SUBJECT, + body: 'keep going', + type: 'status', + priority: 'normal', + threadId: null, + payload: null + }) + }) + record.syncResult = (await clientBuild.syncFederatedDispatch( + client.runtime, + dispatchId + )) as OrchestrationSkewRecord['syncResult'] + const imported = publishedBy(record, 'orchestration.federationImport').imported + record.controlMailImported = typeof imported === 'number' ? imported : 0 + }) + } finally { + host.store.close() + client.store.close() + } + + return record +} diff --git a/tests/e2e/cross-version-wire/release-checkout.ts b/tests/e2e/cross-version-wire/release-checkout.ts index eafee3802be..27c93457418 100644 --- a/tests/e2e/cross-version-wire/release-checkout.ts +++ b/tests/e2e/cross-version-wire/release-checkout.ts @@ -112,9 +112,22 @@ function compareReleaseTags(a: string, b: string): number { * the exact failure this harness exists to prevent. */ export function resolveBaselineReleaseRef(): string { + return resolveBaselineReleaseRefs(1)[0] as string +} + +/** + * The newest `count` release points, newest first. A surface whose peers routinely + * sit a release behind — federation pairs a desktop with a worker host that updates + * on its own schedule — has to be checked against N-1 as well as N, because N alone + * already contains the fix for anything broken during the last cycle. + * + * Returns fewer than `count` only when the repository has fewer stable tags; an + * explicit {@link BASELINE_REF_ENV} pins exactly one ref and wins outright. + */ +export function resolveBaselineReleaseRefs(count: number): string[] { const override = process.env[BASELINE_REF_ENV]?.trim() if (override) { - return override + return [override] } let tags: string[] try { @@ -125,24 +138,30 @@ export function resolveBaselineReleaseRef(): string { `Run it inside a git checkout, or pin a ref with ${BASELINE_REF_ENV}.` ) } - const latest = selectLatestStableReleaseTag(tags) - if (!latest) { + const selected = selectStableReleaseTags(tags, count) + if (selected.length === 0) { throw new Error( `Cross-version harness found no stable desktop release tags matching vX.Y.Z (saw ${tags.length} tag(s) total). ` + 'CI checkouts default to a shallow clone with no tags: use `actions/checkout` with `fetch-depth: 0`, ' + `or pin a ref with ${BASELINE_REF_ENV}.` ) } - return latest + return selected } export function selectLatestStableReleaseTag(tags: string[]): string | null { - return ( - tags - .filter((tag) => STABLE_DESKTOP_RELEASE_TAG.test(tag)) - .sort(compareReleaseTags) - .at(-1) ?? null - ) + return selectStableReleaseTags(tags, 1)[0] ?? null +} + +/** The newest `count` stable desktop release tags, newest first. */ +export function selectStableReleaseTags(tags: string[], count: number): string[] { + if (count <= 0) { + return [] + } + return tags + .filter((tag) => STABLE_DESKTOP_RELEASE_TAG.test(tag)) + .sort((a, b) => compareReleaseTags(b, a)) + .slice(0, count) } function resolveCommit(ref: string): string { diff --git a/tests/e2e/cross-version-wire/release-checkout.unit.test.ts b/tests/e2e/cross-version-wire/release-checkout.unit.test.ts index 106e2ea778e..7e7e1fb1240 100644 --- a/tests/e2e/cross-version-wire/release-checkout.unit.test.ts +++ b/tests/e2e/cross-version-wire/release-checkout.unit.test.ts @@ -20,6 +20,10 @@ import { importReleaseCheckoutModule, materializeReleaseCheckout, REPO_ROOT, + resolveBaselineReleaseRef, + resolveBaselineReleaseRefs, + selectLatestStableReleaseTag, + selectStableReleaseTags, type CheckoutLockOptions, type CheckoutStagingContext, type ReleaseCheckout @@ -260,6 +264,47 @@ afterEach(() => { } }) +describe('stable release tag selection', () => { + it('ignores legacy, mobile, and prerelease tags and orders newest first', () => { + expect( + selectStableReleaseTags( + ['v799', 'mobile-v9.0.0', 'v1.4.177-rc.3', 'v1.4.175', 'v1.4.176', 'v1.4.180'], + 2 + ) + ).toEqual(['v1.4.180', 'v1.4.176']) + }) + + it('orders by version component, not lexically', () => { + expect(selectStableReleaseTags(['v1.4.9', 'v1.4.10', 'v1.4.100'], 3)).toEqual([ + 'v1.4.100', + 'v1.4.10', + 'v1.4.9' + ]) + }) + + it('returns every stable tag it has when fewer exist than asked for', () => { + expect(selectStableReleaseTags(['v1.4.175', 'v1.4.177-rc.3'], 2)).toEqual(['v1.4.175']) + expect(selectStableReleaseTags([], 2)).toEqual([]) + expect(selectStableReleaseTags(['v1.4.175'], 0)).toEqual([]) + }) + + it('keeps the single-baseline selector agreeing with the newest of the list', () => { + const tags = ['v1.4.175', 'v1.4.176', 'v1.4.180'] + expect(selectLatestStableReleaseTag(tags)).toBe(selectStableReleaseTags(tags, 2)[0]) + expect(selectLatestStableReleaseTag([])).toBeNull() + }) + + it('resolves real repository tags without hard-coding a version', () => { + const refs = resolveBaselineReleaseRefs(2) + expect(refs.length).toBeGreaterThan(0) + for (const ref of refs) { + expect(ref).toMatch(/^v\d+\.\d+\.\d+$/) + } + expect(new Set(refs).size).toBe(refs.length) + expect(refs[0]).toBe(resolveBaselineReleaseRef()) + }) +}) + describe('release checkout materialization', () => { it('single-flights concurrent consumers of one release identity', async () => { const cacheRoot = temporaryCacheRoot() diff --git a/tests/e2e/cross-version-wire/versioned-orchestration-wire.ts b/tests/e2e/cross-version-wire/versioned-orchestration-wire.ts new file mode 100644 index 00000000000..39239be6034 --- /dev/null +++ b/tests/e2e/cross-version-wire/versioned-orchestration-wire.ts @@ -0,0 +1,278 @@ +import { access } from 'node:fs/promises' +import { constants } from 'node:fs' +import { join } from 'node:path' +import { + importReleaseCheckoutModule, + materializeReleaseCheckout, + type ReleaseCheckout +} from './release-checkout' + +/** + * One build's orchestration federation surface: the host dispatcher it registers, + * the coordinator-side callers that compose what goes on the wire, and the store + * both sides persist into. Everything here is read from the build, so "this + * release does not have X" is a fact about a real checkout rather than a list. + */ + +export const WORKING_TREE = 'working-tree' as const + +/** Modules whose path moved when orchestration RPC methods were foldered. */ +const MODULE_ALIASES = { + dispatcher: ['/src/main/runtime/rpc/dispatcher.ts'], + methodRegistry: ['/src/main/runtime/rpc/methods/index.ts'], + protocol: ['/src/shared/protocol-version.ts'], + db: ['/src/main/runtime/orchestration/db.ts'], + orchestrationError: ['/src/main/runtime/orchestration/orchestration-error.ts'], + federationSync: ['/src/main/runtime/orchestration/federation-sync.ts'], + controlMessage: ['/src/main/runtime/orchestration/federation-control-message.ts'], + federatedWorkerStart: [ + '/src/main/runtime/rpc/methods/orchestration/federation/federated-worker-start.ts', + '/src/main/runtime/rpc/methods/orchestration-federated-worker-start.ts' + ], + workerObservation: [ + '/src/main/runtime/rpc/methods/orchestration/worker/worker-observation.ts', + '/src/main/runtime/rpc/methods/orchestration-worker-observation.ts' + ], + legacyFederatedRead: [ + '/src/main/runtime/rpc/methods/orchestration/worker/worker-legacy-federated-read.ts', + '/src/main/runtime/rpc/methods/orchestration-worker-legacy-federated-read.ts' + ], + attachStartSchema: [ + '/src/main/runtime/rpc/methods/orchestration/federation/federation-start-schema.ts', + '/src/main/runtime/rpc/methods/orchestration-federation-start-schema.ts' + ], + /** Absent from builds that predate the fleet snapshot; absence is the finding. */ + fleetSnapshot: [ + '/src/main/runtime/rpc/methods/orchestration/federation/federated-fleet-snapshot.ts', + '/src/main/runtime/rpc/methods/orchestration-federated-fleet-snapshot.ts' + ] +} as const + +type ModuleKey = keyof typeof MODULE_ALIASES + +export type RpcEnvelope = { + id: string + ok: boolean + result?: unknown + error?: { code: string; message: string } +} + +export type OrchestrationDispatcher = { + dispatch: ( + request: { + id: string + method: string + params?: unknown + orchestrationRequestId?: string + orchestrationContractVersion?: number + }, + options?: { authenticatedCallerFingerprint?: string } + ) => Promise +} + +export type OrchestrationStore = { + close: () => void + [method: string]: unknown +} + +export type OrchestrationWireBuild = { + /** Human label used in test names and failure messages. */ + label: string + /** `working-tree` for current code, otherwise the resolved release commit. */ + revision: string + /** Capability strings this build defines; a host cannot advertise more. */ + capabilities: readonly string[] + protocolVersion: number + /** Stamped on every outgoing `orchestration.*` call; a host fences a mismatch. */ + orchestrationContractVersion: number + /** Every RPC method name this build registers, read from its own registry. */ + methodNames: readonly string[] + /** A dispatcher over the method set this build really ships. */ + createDispatcher: (runtime: unknown) => OrchestrationDispatcher + /** In-memory store of this build's schema, migrated by this build's code. */ + createStore: () => OrchestrationStore + /** This build's error class, so a peer's `instanceof` narrowing still fires. */ + createRpcError: (code: string, message: string) => Error + /** Coordinator-side composer of `orchestration.federationAttachStart` params. */ + startFederatedWorker: (args: Record) => Promise + /** Coordinator-side caller of `orchestration.federationShow`. */ + callFederatedWorkerShow: (runtime: unknown, federated: unknown) => Promise + /** Coordinator-side caller of `orchestration.federationRead`. */ + readLegacyFederatedTerminal: (args: Record) => Promise + /** Coordinator-side pull/ack/import loop. */ + syncFederatedDispatch: (runtime: unknown, dispatchId: string) => Promise + encodeControlMessage: (message: Record) => string + /** The host-side zod schema for attach params, used to cross-validate a peer's send. */ + parseAttachStartParams: (params: unknown) => unknown + /** null when this build has no fleet-snapshot caller at all. */ + readFederatedFleetSnapshots: + | ((args: Record) => Promise<{ + observations: Map> + errors: { code: string; dispatchIds: string[] }[] + }>) + | null +} + +function registeredMethodNames(methods: readonly unknown[]): string[] { + return methods + .flatMap((method) => { + if (!method || typeof method !== 'object') { + return [] + } + const name = Reflect.get(method, 'name') + return typeof name === 'string' ? [name] : [] + }) + .sort() +} + +function pick(module: Record, name: string, label: string): T { + const value = module[name] + if (typeof value !== 'function') { + throw new Error(`Build ${label} publishes no ${name}; the federation surface moved.`) + } + return value as T +} + +type ModuleLoader = (key: ModuleKey) => Promise | null> + +async function assembleBuild(args: { + label: string + revision: string + load: ModuleLoader +}): Promise { + const { label, revision, load } = args + const required = async (key: ModuleKey): Promise> => { + const module = await load(key) + if (!module) { + throw new Error( + `Build ${label} has no module for ${key}; add its path to MODULE_ALIASES rather than skipping the pairing.` + ) + } + return module + } + const [ + protocol, + dispatcher, + methodRegistry, + store, + errors, + sync, + controlMessage, + workerStart, + observation, + legacyRead, + schema, + fleet + ] = await Promise.all([ + required('protocol'), + required('dispatcher'), + required('methodRegistry'), + required('db'), + required('orchestrationError'), + required('federationSync'), + required('controlMessage'), + required('federatedWorkerStart'), + required('workerObservation'), + required('legacyFederatedRead'), + required('attachStartSchema'), + load('fleetSnapshot') + ]) + + const capabilities = protocol.RUNTIME_CAPABILITIES + if (!Array.isArray(capabilities) || capabilities.length === 0) { + throw new Error(`Build ${label} declares no RUNTIME_CAPABILITIES to negotiate with`) + } + const RpcDispatcher = pick< + new (options: { runtime: unknown; methods: unknown[] }) => OrchestrationDispatcher + >(dispatcher, 'RpcDispatcher', label) + const Store = pick OrchestrationStore>(store, 'OrchestrationDb', label) + const RpcError = pick Error>( + errors, + 'OrchestrationError', + label + ) + const attachSchema = schema.FederationAttachStartParams as { parse: (value: unknown) => unknown } + if (typeof attachSchema?.parse !== 'function') { + throw new Error(`Build ${label} publishes no FederationAttachStartParams schema`) + } + const methods = methodRegistry.ALL_RPC_METHODS as unknown[] + + return { + label, + revision, + capabilities: capabilities as readonly string[], + protocolVersion: protocol.RUNTIME_PROTOCOL_VERSION as number, + orchestrationContractVersion: protocol.ORCHESTRATION_CONTRACT_VERSION as number, + methodNames: registeredMethodNames(methods), + createDispatcher: (runtime) => new RpcDispatcher({ runtime, methods }), + createStore: () => new Store(':memory:'), + createRpcError: (code, message) => new RpcError(code, message), + startFederatedWorker: pick(workerStart, 'startFederatedWorker', label), + callFederatedWorkerShow: pick(observation, 'callFederatedWorkerShow', label), + readLegacyFederatedTerminal: pick(legacyRead, 'readLegacyFederatedTerminal', label), + syncFederatedDispatch: pick(sync, 'syncFederatedDispatch', label), + encodeControlMessage: pick(controlMessage, 'encodeFederatedControlMessage', label), + parseAttachStartParams: (params) => attachSchema.parse(params), + readFederatedFleetSnapshots: fleet ? pick(fleet, 'readFederatedFleetSnapshots', label) : null + } +} + +function workingTreeLoader(): ModuleLoader { + // Static specifiers so the working tree is compiled by the test runner's own + // module graph; a dynamic path here would resolve against the checkout cache. + const modules: Record Promise>> = { + protocol: () => import('../../../src/shared/protocol-version'), + dispatcher: () => import('../../../src/main/runtime/rpc/dispatcher'), + methodRegistry: () => import('../../../src/main/runtime/rpc/methods'), + db: () => import('../../../src/main/runtime/orchestration/db'), + orchestrationError: () => import('../../../src/main/runtime/orchestration/orchestration-error'), + federationSync: () => import('../../../src/main/runtime/orchestration/federation-sync'), + controlMessage: () => + import('../../../src/main/runtime/orchestration/federation-control-message'), + federatedWorkerStart: () => + import('../../../src/main/runtime/rpc/methods/orchestration/federation/federated-worker-start'), + workerObservation: () => + import('../../../src/main/runtime/rpc/methods/orchestration/worker/worker-observation'), + legacyFederatedRead: () => + import('../../../src/main/runtime/rpc/methods/orchestration/worker/worker-legacy-federated-read'), + attachStartSchema: () => + import('../../../src/main/runtime/rpc/methods/orchestration/federation/federation-start-schema'), + fleetSnapshot: () => + import('../../../src/main/runtime/rpc/methods/orchestration/federation/federated-fleet-snapshot') + } + return async (key) => (await modules[key]()) as Record +} + +function releaseLoader(checkout: ReleaseCheckout): ModuleLoader { + return async (key) => { + for (const candidate of MODULE_ALIASES[key]) { + try { + await access(join(checkout.root, candidate.slice(1)), constants.F_OK) + } catch { + continue + } + return await importReleaseCheckoutModule(checkout, candidate) + } + return null + } +} + +/** + * Load the orchestration federation surface for one build. `WORKING_TREE` imports + * current source; any other value is a git ref extracted into a cached checkout. + */ +export async function loadOrchestrationWireBuild(ref: string): Promise { + if (ref === WORKING_TREE) { + return assembleBuild({ + label: WORKING_TREE, + revision: WORKING_TREE, + load: workingTreeLoader() + }) + } + const checkout = await materializeReleaseCheckout(ref) + return assembleBuild({ + label: checkout.ref, + revision: checkout.commit, + load: releaseLoader(checkout) + }) +}