mirror of
https://github.com/stablyai/orca.git
synced 2026-09-28 08:02:43 +00:00
test: pair orchestration federation against the two newest releases
#19689 shipped because no lane paired two builds over `orchestration.*`. The cross-version harness only covered the terminal stream and `agentSession.*`, and every federation "old peer" test wrote a protocol_version integer into a row and read it back with the same build, so a coordinator that omits a field and a host that starts requiring it were both invisible. Pairing against the newest tag alone would not have caught it either: the newest tag already contains the fix for anything broken and repaired inside the last release cycle. A worker host and the desktop that dispatches to it update on their own schedules, so N-1 is an ordinary peer. The suite therefore extracts the two newest stable tags and runs both skew directions against each. Each pairing drives one journey through the real coordinator modules — `startFederatedWorker` composes the attach params, `syncFederatedDispatch` drives the relay — against the peer build's real dispatcher over its real store, with a JSON round trip so an `undefined` field drops exactly as on the wire. What the old side has is read from its checkout; nothing about a release is written down. `pr-code-change-scope.mjs` did not route orchestration source at this job, so the lane would not have run on the PR that broke it; the federation source prefixes are added with cases in its test. The job's file list is now derived from the directory, so a new suite that is never named in the workflow fails instead of covering nothing.
This commit is contained in:
Jinwoo-H
@@ -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
|
||||
|
||||
@@ -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)
|
||||
})
|
||||
})
|
||||
@@ -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 = [
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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<string, BaselinePairings>()
|
||||
|
||||
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
|
||||
)
|
||||
})
|
||||
})
|
||||
})
|
||||
@@ -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<string, unknown> | 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<OrchestrationWireBuild['createStore']>
|
||||
|
||||
export type FederationHostPeer = {
|
||||
store: OrchestrationStoreHandle
|
||||
runtime: unknown
|
||||
missing: string[]
|
||||
call: (
|
||||
method: string,
|
||||
params: unknown,
|
||||
envelope: { requestId?: string; contractVersion?: number }
|
||||
) => Promise<RpcEnvelope>
|
||||
}
|
||||
|
||||
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<string, unknown>): { 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<T>(
|
||||
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<string, unknown> | 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 }
|
||||
}
|
||||
@@ -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<string, unknown> | null
|
||||
/** Fleet verdicts the coordinator derived, and the host errors it degraded to. */
|
||||
fleet: {
|
||||
observations: Record<string, unknown>
|
||||
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<string, unknown> {
|
||||
const result = call?.result
|
||||
return result && typeof result === 'object' && !Array.isArray(result)
|
||||
? (result as Record<string, unknown>)
|
||||
: {}
|
||||
}
|
||||
|
||||
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<string, unknown> {
|
||||
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<string, unknown>)
|
||||
.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<OrchestrationSkewRecord> {
|
||||
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<void>): Promise<void> => {
|
||||
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<string, unknown>
|
||||
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<Record<string, unknown> | 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
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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<RpcEnvelope>
|
||||
}
|
||||
|
||||
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<string, unknown>) => Promise<unknown>
|
||||
/** Coordinator-side caller of `orchestration.federationShow`. */
|
||||
callFederatedWorkerShow: (runtime: unknown, federated: unknown) => Promise<unknown>
|
||||
/** Coordinator-side caller of `orchestration.federationRead`. */
|
||||
readLegacyFederatedTerminal: (args: Record<string, unknown>) => Promise<unknown>
|
||||
/** Coordinator-side pull/ack/import loop. */
|
||||
syncFederatedDispatch: (runtime: unknown, dispatchId: string) => Promise<unknown>
|
||||
encodeControlMessage: (message: Record<string, unknown>) => 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<string, unknown>) => Promise<{
|
||||
observations: Map<string, Record<string, unknown>>
|
||||
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<T>(module: Record<string, unknown>, 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<Record<string, unknown> | null>
|
||||
|
||||
async function assembleBuild(args: {
|
||||
label: string
|
||||
revision: string
|
||||
load: ModuleLoader
|
||||
}): Promise<OrchestrationWireBuild> {
|
||||
const { label, revision, load } = args
|
||||
const required = async (key: ModuleKey): Promise<Record<string, unknown>> => {
|
||||
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<new (path: string) => OrchestrationStore>(store, 'OrchestrationDb', label)
|
||||
const RpcError = pick<new (code: string, message: string) => 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<ModuleKey, () => Promise<Record<string, unknown>>> = {
|
||||
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<string, unknown>
|
||||
}
|
||||
|
||||
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<OrchestrationWireBuild> {
|
||||
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)
|
||||
})
|
||||
}
|
||||
Reference in New Issue
Block a user