From bc39fb3ba83d9056a534891d1b40cac25501bc31 Mon Sep 17 00:00:00 2001 From: Miya Date: Wed, 26 Aug 2026 15:31:20 +0200 Subject: [PATCH 1/4] fix(orchestrator): verify remote agent placement Session-Id: 01a03e13-5b3b-7473-a480-53403c8df263 Session-Id: 01a03e13-5b3b-7473-a480-53403c8df263 --- src/fleet/relay-fleet-client.test.ts | 29 ++++ src/fleet/relay-fleet-client.ts | 13 ++ src/orchestrator/factory.test.ts | 169 +++++++++++++++++++- src/orchestrator/factory.ts | 225 +++++++++++++++++++++++++-- src/ports/fleet.ts | 6 + src/ports/state.ts | 5 + src/state/file-state-store.test.ts | 32 ++++ src/state/file-state-store.ts | 19 +++ src/state/in-memory-state-store.ts | 12 ++ 9 files changed, 498 insertions(+), 12 deletions(-) diff --git a/src/fleet/relay-fleet-client.test.ts b/src/fleet/relay-fleet-client.test.ts index 73a336b7..b9679219 100644 --- a/src/fleet/relay-fleet-client.test.ts +++ b/src/fleet/relay-fleet-client.test.ts @@ -815,6 +815,35 @@ describe('RelayFleetClient', () => { expect(messaging.agentPresenceCalls).toBe(1) }) + it('confirms registration only when presence and a live capable host agree', async () => { + const messaging = new FakeMessaging() + messaging.agentRows = [ + { name: 'ar-1-impl', status: 'online', node: 'alpha' }, + { name: 'ar-hostless-impl', status: 'online' }, + ] + messaging.nodeRows = [ + { name: 'alpha', status: 'online', capabilities: [{ name: 'spawn:codex' }] }, + { name: 'beta', status: 'offline', capabilities: [{ name: 'spawn:codex' }] }, + ] + const fleet = createClient(messaging) + + await expect(fleet.isAgentRegistered({ + name: 'ar-1-impl', + node: 'alpha', + capability: 'spawn:codex', + })).resolves.toBe(true) + await expect(fleet.isAgentRegistered({ + name: 'ar-1-impl', + node: 'beta', + capability: 'spawn:codex', + })).resolves.toBe(false) + await expect(fleet.isAgentRegistered({ + name: 'ar-hostless-impl', + node: 'alpha', + capability: 'spawn:codex', + })).resolves.toBe(false) + }) + it('sends DMs and channel messages through the agent-scoped surface', async () => { const messaging = new FakeMessaging() const fleet = createClient(messaging) diff --git a/src/fleet/relay-fleet-client.ts b/src/fleet/relay-fleet-client.ts index bb93b27a..24d39bf6 100644 --- a/src/fleet/relay-fleet-client.ts +++ b/src/fleet/relay-fleet-client.ts @@ -574,6 +574,19 @@ export class RelayFleetClient implements FleetClient { } } + async isAgentRegistered(input: { + name: string + node: string + capability: Capability + }): Promise { + const roster = await this.roster() + const agent = roster.agents.find((candidate) => + candidate.name === input.name && candidate.node === input.node) + if (!agent) return false + return roster.nodes.some((node) => + node.name === input.node && node.live && node.capabilities.includes(input.capability)) + } + async discoverTeammates(query: TeammateQuery): Promise { this.#teammateDirectory ??= this.#createTeammateDirectory() return await this.#teammateDirectory.discover(query) diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 42d34106..89ace5c2 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -1073,7 +1073,18 @@ class RemoteLifecycleFleetClient extends FakeFleetClient { override async roster() { const roster = await super.roster() - return { ...roster, agents: roster.agents.map((agent) => ({ ...agent, node: 'sf-mini' })) } + return { + agents: roster.agents.map((agent) => ({ ...agent, node: 'sf-mini' })), + nodes: [{ + name: 'sf-mini', + capabilities: ['spawn:codex' as const, 'spawn:claude' as const, 'workflow:run' as const], + live: true, + }], + } + } + + async isAgentRegistered(): Promise { + return true } override async reconcileTrackedAgents(): Promise { @@ -1084,6 +1095,49 @@ class RemoteLifecycleFleetClient extends FakeFleetClient { } } +class NodeDiscoveryRemoteFleetClient extends RemoteLifecycleFleetClient { + nodes: RosterEntry['nodes'] = [ + { name: 'sf-mini', capabilities: ['spawn:codex', 'spawn:claude'], live: true }, + ] + readonly unregistered = new Set() + + override async spawn(input: SpawnInput): Promise { + const result = await super.spawn(input) + return { + ...result, + node: input.node && input.node !== 'self' ? input.node : result.node, + locality: 'remote', + } + } + + override async roster(): Promise { + const roster = await super.roster() + return { + agents: roster.agents + .filter((agent) => !this.unregistered.has(agent.name)) + .map((agent) => ({ + ...agent, + node: this.spawns.findLast((spawn) => spawn.name === agent.name)?.node === 'self' + ? agent.node + : this.spawns.findLast((spawn) => spawn.name === agent.name)?.node ?? agent.node, + })), + nodes: structuredClone(this.nodes), + } + } + + override async isAgentRegistered(input: { + name: string + node: string + capability: 'spawn:codex' | 'spawn:claude' | 'workflow:run' + }): Promise { + if (this.unregistered.has(input.name)) return false + const roster = await this.roster() + return roster.agents.some((agent) => agent.name === input.name && agent.node === input.node) && + roster.nodes.some((node) => + node.name === input.node && node.live && node.capabilities.includes(input.capability)) + } +} + /** * Fires a hook from inside `#finishDurableRelease`'s agent release: after * completion recorded the terminal role at the provider write, and before the @@ -3115,6 +3169,119 @@ describe('waitForDispatchTerminal', () => { }) }) +describe('remote fleet node discovery and registration admission', () => { + it('skips an offline capable node and pins every spawn to a live capable node', async () => { + const path = issuePath(390) + const issue = issueFile(390) + const fleet = new NodeDiscoveryRemoteFleetClient() + fleet.nodes = [ + { name: 'chief-broker', capabilities: ['spawn:codex', 'spawn:claude'], live: false }, + { name: 'sf-mini', capabilities: ['spawn:codex', 'spawn:claude'], live: true }, + ] + const factory = createFactory(config(), { + mount: new FakeMountClient({ [path]: issue }), + fleet, + stateStore: new InMemoryStateStore({ batchSize: 2 }), + triage: new StaticTriage(), + }) + + try { + const decision = await factory.triageIssue(parseLinearIssue(path, issue)) + await expect(factory.dispatch(decision)).resolves.toMatchObject({ + agents: [ + { name: 'ar-390-impl-pear', role: 'implementer' }, + { name: 'ar-390-review', role: 'reviewer' }, + ], + }) + + expect(fleet.spawns.map(({ name, node }) => ({ name, node }))).toEqual([ + { name: 'ar-390-impl-pear', node: 'sf-mini' }, + { name: 'ar-390-review', node: 'sf-mini' }, + ]) + } finally { + await factory.stop() + } + }) + + it('refuses when no live capable node exists before writing a durable lifecycle claim', async () => { + const path = issuePath(391) + const issue = issueFile(391) + const fleet = new NodeDiscoveryRemoteFleetClient() + fleet.nodes = [ + { name: 'chief-broker', capabilities: ['spawn:codex', 'spawn:claude'], live: false }, + { name: 'sf-mini', capabilities: ['spawn:codex'], live: false }, + ] + const stateStore = new InMemoryStateStore({ batchSize: 1 }) + const factory = createFactory(config({ batchSize: 1 }), { + mount: new FakeMountClient({ [path]: issue }), + fleet, + stateStore, + triage: new StaticTriage(), + }) + + try { + const decision = await factory.triageIssue(parseLinearIssue(path, issue)) + await expect(factory.dispatch(decision)).rejects.toThrow( + 'no live fleet node advertises spawn:codex', + ) + + expect(fleet.spawns).toEqual([]) + await expect(stateStore.listDispatchLifecycles('factory-test')).resolves.toEqual([]) + } finally { + await factory.stop() + } + }) + + it('rolls back an unregistered remote spawn and frees the slot for the next issue', async () => { + const firstPath = issuePath(392) + const secondPath = issuePath(393) + const firstIssue = issueFile(392) + const secondIssue = issueFile(393) + const fleet = new NodeDiscoveryRemoteFleetClient() + fleet.unregistered.add('ar-392-impl-pear') + const stateStore = new InMemoryStateStore({ batchSize: 1 }) + const clock = new ManualClock() + const factory = createFactory(config({ batchSize: 1 }), { + mount: new FakeMountClient({ [firstPath]: firstIssue, [secondPath]: secondIssue }), + fleet, + stateStore, + triage: new StaticTriage(), + clock, + }) + + try { + const first = await factory.triageIssue(parseLinearIssue(firstPath, firstIssue)) + await expect(factory.dispatch(first)).rejects.toThrow( + 'did not register with the fleet before the startup deadline', + ) + expect(clock.value).toBe(30_000) + + expect(fleet.releases).toContainEqual({ + name: 'ar-392-impl-pear', + reason: 'spawn-registration-timeout', + }) + await expect(stateStore.getDispatchLifecycle( + 'factory-test', + dispatchIssueIdentity(first.issue), + )).resolves.toBeUndefined() + + const second = await factory.triageIssue(parseLinearIssue(secondPath, secondIssue)) + await expect(factory.dispatch(second)).resolves.toMatchObject({ + agents: [ + { name: 'ar-393-impl-pear', role: 'implementer' }, + { name: 'ar-393-review', role: 'reviewer' }, + ], + }) + await expect(stateStore.getDispatchLifecycle( + 'factory-test', + dispatchIssueIdentity(second.issue), + )).resolves.toMatchObject({ phase: 'running' }) + } finally { + await factory.stop() + } + }) +}) + describe('FactoryLoop', () => { it('sweeps preview orphans on daemon startup using durable active issue owners', async () => { const mount = new FakeMountClient() diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 3cb5125e..dd7a0115 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -45,6 +45,7 @@ import type { MountClient, ProviderSyncStatus, PreviewReference, + RosterEntry, SlackWriteback, SpawnResult, Subscription, @@ -456,6 +457,8 @@ const STOP_REJECTED_DISPATCH_DRAIN_TIMEOUT_MS = 2_500 const DISPATCH_LIFECYCLE_LEASE_MS = 5 * 60_000 const DISPATCH_LIFECYCLE_RENEW_MS = 60_000 const DISPATCH_LIFECYCLE_RETRY_MS = 1_000 +const REMOTE_AGENT_REGISTRATION_TIMEOUT_MS = 30_000 +const REMOTE_AGENT_REGISTRATION_POLL_MS = 500 /** * Ceiling on the durable capacity-wait re-arm (#303). * @@ -654,6 +657,24 @@ class DispatchLifecycleOwnedElsewhereError extends Error { } } +class FleetPlacementUnavailableError extends Error { + constructor(readonly capability: Capability) { + super(`Refusing remote dispatch: no live fleet node advertises ${capability}`) + this.name = 'FleetPlacementUnavailableError' + } +} + +class RemoteAgentRegistrationTimeoutError extends Error { + constructor( + readonly agentName: string, + readonly cleanupConfirmed: boolean, + readonly cleanupError?: unknown, + ) { + super(`Remote agent ${agentName} did not register with the fleet before the startup deadline`) + this.name = 'RemoteAgentRegistrationTimeoutError' + } +} + /** * The durable dispatch-lifecycle claim was refused for one work unit: its * record is already terminal, or another publisher currently holds the lease. @@ -3203,10 +3224,11 @@ export class FactoryLoop implements Factory { } } - async #assertFleetControlPlaneAvailable(): Promise { + async #assertFleetControlPlaneAvailable(): Promise { try { - await this.#fleet.roster() + const roster = await this.#fleet.roster() this.#increment('fleetControlPlaneProbeSuccesses') + return roster } catch (error) { const health = this.#fleetControlPlane.status() this.#increment('fleetControlPlaneProbeFailures') @@ -5616,7 +5638,10 @@ export class FactoryLoop implements Factory { // consuming a dispatch attempt. The mutation proxy probes again at the // actual spawn/resume boundary so a later control-plane fault still fails // closed. - if (!dryRun) await this.#assertFleetControlPlaneAvailable() + const admissionRoster = !dryRun ? await this.#assertFleetControlPlaneAvailable() : undefined + if (admissionRoster && this.#fleet.placementLocality === 'remote') { + dispatchDecision = decisionWithVerifiedRemotePlacements(dispatchDecision, admissionRoster) + } const durableDispatch = !dryRun && this.#usesDurableDispatchLifecycle() // Local dispatches need the same deterministic branch identity as remote // ones. Without it, every worker starts in the configured shared checkout @@ -6033,6 +6058,15 @@ export class FactoryLoop implements Factory { throw error } settlePostSpawnIssueObservation(!(error instanceof LiveDispatchStateChangedError)) + if ( + (error instanceof FleetPlacementUnavailableError || + (error instanceof RemoteAgentRegistrationTimeoutError && error.cleanupConfirmed)) && + await this.#rollbackUnregisteredRemoteDispatch(record, spawnedForReaperHandoff) + ) { + this.#increment('remoteDispatchAdmissionRollbacks') + this.#error(error, decision.issue) + throw error + } // A spawn can fail after the broker accepted it but before its ack // reached Factory. Include every planned worktree agent, not only the // acknowledged spawns, so cleanup never races a name-only survivor. @@ -10616,14 +10650,6 @@ export class FactoryLoop implements Factory { return { name: spec.name } } - // Persist intent before the remote side effect. If the owner crashes after - // the spawn ack but before recording its result, takeover retries the same - // deterministic invocation id instead of inventing a second worker. - batch.recordPlanned(record, { ...spec, invocationId }) - if (!await this.#saveDispatchLifecycle(record, 'dispatching')) { - throw new Error(`Dispatch lifecycle ownership lost before spawning ${spec.name}`) - } - let roster try { roster = await retryOnTimeout(() => this.#fleet.roster(), { attempts: 3, delayMs: 2000 }) @@ -10632,6 +10658,16 @@ export class FactoryLoop implements Factory { } const rosterAgent = roster.agents.find((agent) => agent.name === spec.name) if (rosterAgent) { + if (this.#fleet.placementLocality === 'remote') { + const host = rosterAgent.node + ? roster.nodes.find((node) => + node.name === rosterAgent.node && node.live && node.capabilities.includes(spec.capability)) + : undefined + if (!host) { + throw new FleetPlacementUnavailableError(spec.capability) + } + spec = { ...spec, node: host.name } + } const trackedPlacement = this.#fleet.trackedAgents?.().get(spec.name) record.heldSinceAtMs ??= this.#clock.now() batch.recordSpawn(record, spec, invocationId, { @@ -10649,6 +10685,22 @@ export class FactoryLoop implements Factory { return { name: spec.name } } + if (this.#fleet.placementLocality === 'remote') { + const loads = new Map() + for (const agent of roster.agents) { + if (agent.node) loads.set(agent.node, (loads.get(agent.node) ?? 0) + 1) + } + spec = { ...spec, node: liveFleetNodeForSpec(spec, roster, loads) } + } + + // Persist intent before the remote side effect. If the owner crashes after + // the spawn ack but before recording its result, takeover retries the same + // deterministic invocation id instead of inventing a second worker. + batch.recordPlanned(record, { ...spec, invocationId }) + if (!await this.#saveDispatchLifecycle(record, 'dispatching')) { + throw new Error(`Dispatch lifecycle ownership lost before spawning ${spec.name}`) + } + await this.#prepareAgentWorktree(record, spec) let result try { @@ -10690,6 +10742,24 @@ export class FactoryLoop implements Factory { await this.#releaseOrphanedLatePlacement(record, spec, result) throw new LatePlacementReleasedError(record.issue.key, result.name ?? spec.name) } + if (this.#fleet.placementLocality === 'remote') { + const registered = await this.#awaitRemoteAgentRegistration(result.name, spec.capability, result.node) + if (!registered) { + try { + await this.#fleet.release(result.name, 'spawn-registration-timeout') + this.#fleet.markAgentTerminal?.(result.name, 'spawn-registration-timeout') + throw new RemoteAgentRegistrationTimeoutError(result.name, true) + } catch (error) { + if (error instanceof RemoteAgentRegistrationTimeoutError) throw error + // Cleanup is unconfirmed. Persist the placement so a successor can + // retry the release; forgetting it would be worse than retaining a + // nonterminal lifecycle for a worker that may still be alive. + batch.recordSpawn(record, spec, invocationId, result) + await this.#saveDispatchLifecycle(record, 'dispatching') + throw new RemoteAgentRegistrationTimeoutError(result.name, false, error) + } + } + } record.heldSinceAtMs ??= this.#clock.now() batch.recordSpawn(record, spec, invocationId, result) if (!await this.#saveDispatchLifecycle(record, 'dispatching')) { @@ -10701,6 +10771,87 @@ export class FactoryLoop implements Factory { return { name: result.name } } + async #awaitRemoteAgentRegistration( + name: string, + capability: Capability, + expectedNode: string | undefined, + ): Promise { + const deadlineAtMs = this.#clock.now() + REMOTE_AGENT_REGISTRATION_TIMEOUT_MS + do { + try { + if (expectedNode) { + if (this.#fleet.isAgentRegistered) { + if (await this.#fleet.isAgentRegistered({ name, node: expectedNode, capability })) return true + } else { + const roster = await this.#fleet.roster() + const agent = roster.agents.find((candidate) => candidate.name === name && candidate.node === expectedNode) + const node = agent + ? roster.nodes.find((candidate) => + candidate.name === expectedNode && candidate.live && candidate.capabilities.includes(capability)) + : undefined + if (agent && node) return true + } + } + } catch (error) { + this.#logger.warn?.('[factory] remote agent registration probe failed; retrying within startup bound', { + agent: name, + error: describeError(error).errorMessage, + }) + } + const remainingMs = deadlineAtMs - this.#clock.now() + if (remainingMs <= 0) return false + await this.#clock.sleep(Math.min(REMOTE_AGENT_REGISTRATION_POLL_MS, remainingMs)) + } while (this.#clock.now() <= deadlineAtMs) + return false + } + + async #rollbackUnregisteredRemoteDispatch( + record: InFlightIssue, + acknowledged: RegistryHandoffAgent[], + ): Promise { + const handoffs = this.#dispatchFailureHandoffs(record, acknowledged) + if ( + handoffs.some((handoff) => handoff.worktree) && + !await this.#teardownFailedDispatchWorktrees( + handoffs, + 'spawn-registration-timeout', + { skipNeverPlacedAgents: true }, + ) + ) return false + try { + await this.#teardownPreviews(record) + } catch (error) { + this.#logger.warn?.('[factory] retained unregistered spawn lifecycle after preview rollback failed', { + issue: record.issue.key, + error: describeError(error).errorMessage, + }) + return false + } + + const key = dispatchLifecycleKey(record.issue) + const epoch = this.#dispatchLifecycleEpochs.get(key) + const lifecycle = await this.#state.getDispatchLifecycle(this.#workspaceId, key) + const lease = lifecycle?.lease + if ( + epoch === undefined || + !lease || + lease.owner !== this.#dispatchLifecycleOwner || + lease.epoch !== epoch || + !await this.#state.clearClaimedDispatchLifecycle(this.#workspaceId, key, lease) + ) return false + + const retryTimer = this.#dispatchLifecycleRetryTimers.get(key) + if (retryTimer) clearTimeout(retryTimer) + this.#dispatchLifecycleRetryTimers.delete(key) + this.#dispatchLifecycleEpochs.delete(key) + this.#abandonedDispatchReasons.delete(key) + const batch = await this.#batch() + batch.abandon(record.issue) + await this.#writeInFlightRegistry() + this.#resetDispatchCapacityBackoff() + return true + } + /** * Is this process still the owner of a lifecycle that is not already done? * @@ -20038,6 +20189,58 @@ function dispatchSpecs(decision: TriageDecision): AgentSpec[] { return [...decision.implementers, decision.reviewer] } +function liveFleetNodeForSpec( + spec: AgentSpec, + roster: RosterEntry, + assignedLoads: Map, +): string { + const eligible = roster.nodes.filter((node) => node.live && node.capabilities.includes(spec.capability)) + if (eligible.length === 0) throw new FleetPlacementUnavailableError(spec.capability) + + const explicitlyRequested = spec.node && spec.node !== 'self' + ? eligible.find((node) => node.name === spec.node) + : undefined + const selected = explicitlyRequested ?? [...eligible].sort((left, right) => { + const loadDifference = (assignedLoads.get(left.name) ?? 0) - (assignedLoads.get(right.name) ?? 0) + return loadDifference || left.name.localeCompare(right.name) + })[0]! + assignedLoads.set(selected.name, (assignedLoads.get(selected.name) ?? 0) + 1) + return selected.name +} + +/** + * Resolve every remote dispatch spec against one canonical roster snapshot. + * + * A configured node is a preference, not an entitlement: if it is offline or + * no longer advertises the required capability, placement is re-selected from + * the live fleet. This runs before durable lifecycle creation, so an empty + * eligible set cannot leave a claim or consume a batch slot. + */ +function decisionWithVerifiedRemotePlacements( + decision: TriageDecision, + roster: RosterEntry, +): TriageDecision { + const assignedLoads = new Map() + for (const agent of roster.agents) { + if (agent.node) assignedLoads.set(agent.node, (assignedLoads.get(agent.node) ?? 0) + 1) + } + const place = (spec: AgentSpec): AgentSpec => ({ + ...spec, + node: liveFleetNodeForSpec(spec, roster, assignedLoads), + }) + if (decision.scope === 'workflow') { + return { + ...structuredClone(decision), + ...(decision.workflow ? { workflow: place(decision.workflow) } : {}), + } + } + return { + ...structuredClone(decision), + implementers: decision.implementers.map(place), + reviewer: place(decision.reviewer), + } +} + function dispatchSessionOwner(decision: TriageDecision): string | undefined { for (const spec of dispatchSpecs(decision)) { const sessionOwner = spec.principal?.trim() || spec.owner?.trim() diff --git a/src/ports/fleet.ts b/src/ports/fleet.ts index eee9795b..8cfa0d61 100644 --- a/src/ports/fleet.ts +++ b/src/ports/fleet.ts @@ -230,6 +230,12 @@ export interface FleetClient { }): Promise release(name: string, reason?: string): Promise roster(): Promise + /** One bounded-poll sample proving a spawned remote identity is broker-visible on its expected host. */ + isAgentRegistered?(input: { + name: string + node: string + capability: Capability + }): Promise /** Find addressable teammate agents by their published A2A cards. */ discoverTeammates(query: TeammateQuery): Promise /** diff --git a/src/ports/state.ts b/src/ports/state.ts index e3590286..aaee7d9e 100644 --- a/src/ports/state.ts +++ b/src/ports/state.ts @@ -529,6 +529,11 @@ export interface StateStore { key: string, expectedLease: DispatchLifecycleLease | undefined, ): Promise + clearClaimedDispatchLifecycle( + workspaceId: string, + key: string, + expectedLease: DispatchLifecycleLease, + ): Promise clearDispatchLifecycle(workspaceId: string, key: string): Promise recordCritical(workspaceId: string, key: string, value: CriticalRecord): Promise diff --git a/src/state/file-state-store.test.ts b/src/state/file-state-store.test.ts index 7e9403e8..30ac4831 100644 --- a/src/state/file-state-store.test.ts +++ b/src/state/file-state-store.test.ts @@ -1019,6 +1019,38 @@ describe('FileStateStore', () => { } }) + it('fences claimed lifecycle rollback against a concurrent lease takeover', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-lifecycle-claimed-clear-')) + try { + const stores = [ + new FileStateStore({ batchSize: 2, watchStatePath: join(root, 'state.json') }), + new InMemoryStateStore({ batchSize: 2 }), + ] + for (const [index, store] of stores.entries()) { + const workspace = `workspace-${index}` + const seed = dispatchLifecycle(895 + index) + const key = dispatchIssueIdentity(seed.issue) + const initial = await store.claimDispatchLifecycle( + workspace, key, seed, 'owner-a', 1_000, 5_000, + ) + expect(initial.lease).toBeDefined() + const takeover = await store.claimDispatchLifecycle( + workspace, key, seed, 'owner-b', 6_001, 5_000, + ) + expect(takeover).toMatchObject({ acquired: true, lease: { owner: 'owner-b', epoch: 2 } }) + + expect(await store.clearClaimedDispatchLifecycle(workspace, key, initial.lease!)).toBe(false) + expect(await store.getDispatchLifecycle(workspace, key)).toMatchObject({ + lease: { owner: 'owner-b', epoch: 2 }, + }) + expect(await store.clearClaimedDispatchLifecycle(workspace, key, takeover.lease!)).toBe(true) + expect(await store.getDispatchLifecycle(workspace, key)).toBeUndefined() + } + } finally { + await rm(root, { recursive: true, force: true }) + } + }) + it('does not count a bare-repo lifecycle after its canonical PR is handed to a babysitter', async () => { const root = await mkdtemp(join(tmpdir(), 'factory-lifecycle-bare-repo-handoff-')) try { diff --git a/src/state/file-state-store.ts b/src/state/file-state-store.ts index 7adadf9f..99fc6fa6 100644 --- a/src/state/file-state-store.ts +++ b/src/state/file-state-store.ts @@ -410,6 +410,25 @@ export class DocumentStateStore extends InMemoryStateStore { })) } + override async clearClaimedDispatchLifecycle( + workspaceId: string, + key: string, + expectedLease: NonNullable, + ): Promise { + return await this.#exclusive(async () => this.#withMutationLock(async () => { + const document = await this.#loadFromDisk() + const workspace = document.workspaces[workspaceId] + const lifecycle = workspace?.dispatchLifecycles[key] + if (!workspace || !lifecycle || !dispatchLifecycleLeaseMatches(lifecycle.lease, expectedLease)) { + return false + } + delete workspace.dispatchLifecycles[key] + if (workspaceIsEmpty(workspace)) delete document.workspaces[workspaceId] + await this.#persist(document) + return true + })) + } + override async clearDispatchLifecycle(workspaceId: string, key: string): Promise { await this.#exclusive(async () => this.#withMutationLock(async () => { const document = await this.#loadFromDisk() diff --git a/src/state/in-memory-state-store.ts b/src/state/in-memory-state-store.ts index 3b0a7559..23ee9b2c 100644 --- a/src/state/in-memory-state-store.ts +++ b/src/state/in-memory-state-store.ts @@ -343,6 +343,18 @@ export class InMemoryStateStore implements StateStore { return true } + async clearClaimedDispatchLifecycle( + workspaceId: string, + key: string, + expectedLease: NonNullable, + ): Promise { + const lifecycles = this.#workspace(workspaceId).dispatchLifecycles + const lifecycle = lifecycles.get(key) + if (!lifecycle || !dispatchLifecycleLeaseMatches(lifecycle.lease, expectedLease)) return false + lifecycles.delete(key) + return true + } + async clearDispatchLifecycle(workspaceId: string, key: string): Promise { this.#workspace(workspaceId).dispatchLifecycles.delete(key) } From a4b2984922de63683e2b7e5c153a30db3ecd0a2d Mon Sep 17 00:00:00 2001 From: Miya Date: Wed, 26 Aug 2026 15:40:31 +0200 Subject: [PATCH 2/4] fix(orchestrator): reap partial remote teams Session-Id: 01a03e13-5b3b-7473-a480-53403c8df263 Session-Id: 01a03e13-5b3b-7473-a480-53403c8df263 --- src/fleet/relay-fleet-client.test.ts | 3 ++- src/orchestrator/factory.test.ts | 8 ++++++-- src/orchestrator/factory.ts | 24 ++++++++++++++++++++---- 3 files changed, 28 insertions(+), 7 deletions(-) diff --git a/src/fleet/relay-fleet-client.test.ts b/src/fleet/relay-fleet-client.test.ts index b9679219..417e3d4c 100644 --- a/src/fleet/relay-fleet-client.test.ts +++ b/src/fleet/relay-fleet-client.test.ts @@ -819,6 +819,7 @@ describe('RelayFleetClient', () => { const messaging = new FakeMessaging() messaging.agentRows = [ { name: 'ar-1-impl', status: 'online', node: 'alpha' }, + { name: 'ar-offline-host-impl', status: 'online', node: 'beta' }, { name: 'ar-hostless-impl', status: 'online' }, ] messaging.nodeRows = [ @@ -833,7 +834,7 @@ describe('RelayFleetClient', () => { capability: 'spawn:codex', })).resolves.toBe(true) await expect(fleet.isAgentRegistered({ - name: 'ar-1-impl', + name: 'ar-offline-host-impl', node: 'beta', capability: 'spawn:codex', })).resolves.toBe(false) diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 89ace5c2..aa7b488f 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -1105,7 +1105,7 @@ class NodeDiscoveryRemoteFleetClient extends RemoteLifecycleFleetClient { const result = await super.spawn(input) return { ...result, - node: input.node && input.node !== 'self' ? input.node : result.node, + node: input.node ?? result.node, locality: 'remote', } } @@ -3238,7 +3238,7 @@ describe('remote fleet node discovery and registration admission', () => { const firstIssue = issueFile(392) const secondIssue = issueFile(393) const fleet = new NodeDiscoveryRemoteFleetClient() - fleet.unregistered.add('ar-392-impl-pear') + fleet.unregistered.add('ar-392-review') const stateStore = new InMemoryStateStore({ batchSize: 1 }) const clock = new ManualClock() const factory = createFactory(config({ batchSize: 1 }), { @@ -3256,6 +3256,10 @@ describe('remote fleet node discovery and registration admission', () => { ) expect(clock.value).toBe(30_000) + expect(fleet.releases).toContainEqual({ + name: 'ar-392-review', + reason: 'spawn-registration-timeout', + }) expect(fleet.releases).toContainEqual({ name: 'ar-392-impl-pear', reason: 'spawn-registration-timeout', diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index dd7a0115..3ecd7c33 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -10810,14 +10810,30 @@ export class FactoryLoop implements Factory { acknowledged: RegistryHandoffAgent[], ): Promise { const handoffs = this.#dispatchFailureHandoffs(record, acknowledged) - if ( - handoffs.some((handoff) => handoff.worktree) && - !await this.#teardownFailedDispatchWorktrees( + await this.#persistDispatchFailureReaperHandoff(record, handoffs) + const hasWorktrees = handoffs.some((handoff) => handoff.worktree) + if (hasWorktrees) { + if (!await this.#teardownFailedDispatchWorktrees( handoffs, 'spawn-registration-timeout', { skipNeverPlacedAgents: true }, + )) return false + } else { + const failed = await this.#releaseAndTerminateAgents( + handoffs + .filter((handoff) => handoff.tracked.result !== undefined) + .map((handoff) => [handoff.name, handoff.tracked]), + 'spawn-registration-timeout', + 'completion', ) - ) return false + if (failed.length > 0) return false + for (const handoff of handoffs) { + await this.#state.clearFailureHandoff( + this.#workspaceId, + registryHandoffKey(handoff.issue, handoff.name), + ) + } + } try { await this.#teardownPreviews(record) } catch (error) { From a912e03e652b0383f60c52199fb7d56277217f5d Mon Sep 17 00:00:00 2001 From: Miya Date: Wed, 26 Aug 2026 15:49:18 +0200 Subject: [PATCH 3/4] test(orchestrator): advertise preview placement node Session-Id: 01a03e13-5b3b-7473-a480-53403c8df263 --- src/orchestrator/factory.test.ts | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index aa7b488f..214bd939 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -27137,6 +27137,17 @@ describe('FactoryLoop PR babysitter', () => { it('pins a remote babysitter to the preview node and gives it the live URL', async () => { class RemotePreviewFleetClient extends RemoteLifecycleFleetClient { + override async roster(): Promise { + const roster = await super.roster() + return { + ...roster, + nodes: [ + ...roster.nodes, + { name: 'preview-node', capabilities: ['spawn:claude'], live: true }, + ], + } + } + override async createPreview(input: PreviewStartInput): Promise { return { ...await super.createPreview(input), node: 'preview-node' } } From 9bce29ab9c87ac9270507c6d373c404dd872232c Mon Sep 17 00:00:00 2001 From: Miya Date: Wed, 26 Aug 2026 15:59:48 +0200 Subject: [PATCH 4/4] test(cli): model remote fleet registration Session-Id: 01a03e13-5b3b-7473-a480-53403c8df263 --- src/cli/fleet.test.ts | 20 +++++++++++++++++--- 1 file changed, 17 insertions(+), 3 deletions(-) diff --git a/src/cli/fleet.test.ts b/src/cli/fleet.test.ts index 935c58cb..91fb34f2 100644 --- a/src/cli/fleet.test.ts +++ b/src/cli/fleet.test.ts @@ -127,8 +127,23 @@ const issueFile = { * flag would put the timer in scope for tests whose whole point is that no * timer decides the ordering (#346 review, cubic). */ -class ControlledCompletingRemoteFleetClient extends FakeFleetClient { +class CompletingRemoteFleetBase extends FakeFleetClient { override readonly placementLocality = 'remote' as const + + override async roster() { + const roster = await super.roster() + return { + agents: roster.agents.map((agent) => ({ ...agent, node: 'sf-mini' })), + nodes: [{ + name: 'sf-mini', + capabilities: ['spawn:codex' as const, 'spawn:claude' as const, 'workflow:run' as const], + live: true, + }], + } + } +} + +class ControlledCompletingRemoteFleetClient extends CompletingRemoteFleetBase { implementerName?: string exitEmitted = false @@ -139,8 +154,7 @@ class ControlledCompletingRemoteFleetClient extends FakeFleetClient { } } -class CompletingRemoteFleetClient extends FakeFleetClient { - override readonly placementLocality = 'remote' as const +class CompletingRemoteFleetClient extends CompletingRemoteFleetBase { readonly lifecycleOrder: string[] = [] override async spawn(input: SpawnInput): Promise {