From 37a422cfa9a98852c54a44fb1692e706436f4003 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Fri, 21 Aug 2026 06:29:34 +0200 Subject: [PATCH 1/3] fix(factory): bound every await on the Relay placement path (#306) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The 5-minute spawn-ack timeout was real, configured and unreachable. The deadline was read at the top of `#awaitInvocation`'s poll loop, wrapped around an unbounded `commands.getInvocation`, so a read that never settled never returned control to the check. Production ran one readiness sweep for 62 minutes against that bound, holding its batch slots the whole time. Everything before the loop had no deadline at all — `#ensureMessaging`, `#ensureLifecycleAction` and `placement.spawn` all ran before one existed — and the deadline was computed inside `#awaitInvocation`, so time already spent on placement was free: an operation could cost the placement delay plus a fresh full ack budget. Anchor one budget per operation and thread it through every call, racing each against the remaining budget. The Relay surface takes no AbortSignal, so the race is the only bound available; it is the same technique the SDK's own `raceConfirmRead` uses. Also set `confirm: true` on the agent-spawn placement. An ack proves the engine accepted the dispatch, not that the node launched anything: a node on an obsolete broker acks and launches nothing, indistinguishable from a real spawn until the invocation is read back. `confirm` makes the SDK do that read, bounded, and `ack.confirmation` is consumed in place of a redundant re-poll — so a confirmed spawn costs fewer reads than before. `RelaySpawnAckTimeoutError` is deliberately left unclassified. Unlike the conditions on the #292 exemption list, a spawn-ack timeout is not per-item and not self-healing: it is evidence about the fleet, and retrying it against the same fleet times out again. A run of them is a pass-wide fault that should trip the fuse, not be exempted from it — classifying it would rebuild this outage as an endless series of five-minute sweeps that never abort and never alert. The fuse itself is untouched. Co-Authored-By: Claude Opus 5 Session-Id: 1e357a8b-0ac7-4606-992e-e5151f53afb6 --- src/fleet/relay-fleet-client.test.ts | 206 +++++++++++++++++++++++++++ src/fleet/relay-fleet-client.ts | 199 ++++++++++++++++++++++---- src/orchestrator/factory.test.ts | 81 +++++++++++ 3 files changed, 457 insertions(+), 29 deletions(-) diff --git a/src/fleet/relay-fleet-client.test.ts b/src/fleet/relay-fleet-client.test.ts index 76f936dc..43145f35 100644 --- a/src/fleet/relay-fleet-client.test.ts +++ b/src/fleet/relay-fleet-client.test.ts @@ -1115,3 +1115,209 @@ describe('RelayFleetClient', () => { expect(messaging.disconnected).toBe(1) }) }) + +/** + * #306 — every await on the placement path must be bounded. + * + * The 5-minute spawn-ack timeout existed before this suite and could never + * fire: the deadline was read at the top of the poll loop, around an unbounded + * `commands.getInvocation`, so a call that never settled never returned control + * to the check. Production ran a single readiness sweep for 62 minutes against + * that 5-minute bound with `consecutiveFailures: 0`. + * + * Each test below states which half it guards. The must-not-fire cases exist + * because the cheap way to pass the must-fire cases is an over-eager per-call + * timeout that kills healthy slow placements — which would trade a hang for an + * outage. + */ +describe('RelayFleetClient placement deadlines (#306)', () => { + const never = (): Promise => new Promise(() => {}) + const after = (ms: number, value: T): Promise => + new Promise((resolve) => setTimeout(() => resolve(value), ms)) + + const spawnInput = () => ({ + name: 'ar-1-impl', + capability: 'spawn:claude' as const, + node: 'self', + repo: 'AgentWorkforce/factory', + task: 'do work', + }) + + const pending = (invocationId: string): RelayActionInvocation => + ({ invocationId, actionName: 'spawn', status: 'dispatched' }) as RelayActionInvocation + + const completed = (invocationId: string): RelayActionInvocation => + ({ + invocationId, + actionName: 'spawn', + status: 'completed', + output: { agent_name: 'ar-1-impl', session_ref: 'session-1', pid: 7 }, + }) as RelayActionInvocation + + // MUST-FIRE. The exact production failure: a poll that never settles. + it('rejects within the ack budget when getInvocation never resolves', async () => { + const messaging = new FakeMessaging() + messaging.placementAck = { invocationId: 'inv-1', status: 'pending', placement: { node: 'node-a' } } + messaging.commands.getInvocation = never + const fleet = createClient(messaging, { spawnAckTimeoutMs: 120 }) + + const startedAt = Date.now() + await expect(fleet.spawn(spawnInput())).rejects.toThrow(/timed out/i) + expect(Date.now() - startedAt).toBeLessThan(2_000) + }) + + // MUST-FIRE. `placement.spawn` runs before any deadline existed at all. + it('rejects within the ack budget when placement.spawn never resolves', async () => { + const messaging = new FakeMessaging() + messaging.placement.spawn = never + const fleet = createClient(messaging, { spawnAckTimeoutMs: 120 }) + + const startedAt = Date.now() + await expect(fleet.spawn(spawnInput())).rejects.toThrow(/timed out/i) + expect(Date.now() - startedAt).toBeLessThan(2_000) + }) + + // MUST-FIRE. `#ensureLifecycleAction` is also awaited before the old deadline. + it('rejects within the ack budget when lifecycle-action registration never resolves', async () => { + const messaging = new FakeMessaging() + messaging.commands.register = never + const fleet = createClient(messaging, { spawnAckTimeoutMs: 120 }) + + const startedAt = Date.now() + await expect(fleet.spawn(spawnInput())).rejects.toThrow(/timed out/i) + expect(Date.now() - startedAt).toBeLessThan(2_000) + }) + + /** + * MUST-FIRE, and the one that pins requirement 2 specifically. + * + * `placement.spawn` completes — slowly, but it completes — and then every + * poll completes too. No single call is slow enough to trip a per-call + * timeout on its own. What must end this is one budget spanning the whole + * operation. + * + * Before the fix `#awaitInvocation` computed its own deadline on entry, so + * the time `placement.spawn` had already burned was free: the operation cost + * the spawn delay *plus* a fresh full ack budget. The assertion below is + * what separates a shared budget from a restarted one. + */ + it('carries one budget across placement.spawn and the poll loop', async () => { + const messaging = new FakeMessaging() + const spawnDelayMs = 500 + const budgetMs = 500 + messaging.placement.spawn = async (input) => { + messaging.placements.push(input) + return await after(spawnDelayMs, { + invocationId: 'inv-1', + actionName: 'spawn', + status: 'pending', + node: { name: 'node-a' }, + placement: { capability: input.capability, node: 'node-a', attempts: 1, queued: false }, + }) + } + messaging.commands.getInvocation = async () => await after(20, pending('inv-1')) + const fleet = createClient(messaging, { + spawnAckTimeoutMs: budgetMs, + sleep: (ms) => after(ms, undefined), + }) + + const startedAt = Date.now() + await expect(fleet.spawn(spawnInput())).rejects.toThrow(/timed out/i) + const elapsedMs = Date.now() - startedAt + + // A restarted budget costs spawnDelayMs + budgetMs (~1000ms); one shared + // budget cannot exceed budgetMs (~500ms). The threshold sits between them. + expect(elapsedMs).toBeLessThan(spawnDelayMs + budgetMs - 250) + }) + + /** + * MUST-NOT-FIRE. A healthy but slow placement: several real polls, each + * taking real time, all comfortably inside the overall budget. This is the + * test that forbids "fixing" #306 with a timeout short enough to break + * working spawns. + */ + it('still succeeds when slow-but-completing polls stay inside the overall budget', async () => { + const messaging = new FakeMessaging() + messaging.placementAck = { invocationId: 'inv-1', status: 'pending', placement: { node: 'node-a' } } + let polls = 0 + messaging.commands.getInvocation = async () => { + polls += 1 + return await after(40, polls < 4 ? pending('inv-1') : completed('inv-1')) + } + const fleet = createClient(messaging, { spawnAckTimeoutMs: 5_000, sleep: (ms) => after(ms, undefined) }) + + await expect(fleet.spawn(spawnInput())).resolves.toMatchObject({ + name: 'ar-1-impl', + sessionRef: 'session-1', + node: 'node-a', + }) + expect(polls).toBe(4) + }) + + // MUST-NOT-FIRE. A release that completes normally must not be timed out. + it('still releases normally when the release invocation completes', async () => { + const messaging = new FakeMessaging() + const fleet = createClient(messaging, { spawnAckTimeoutMs: 5_000 }) + + await expect(fleet.release('ar-1-impl', 'done')).resolves.toBeUndefined() + expect(messaging.invokes.map((invoke) => invoke.name)).toContain('release') + }) + + // MUST-FIRE on the release path, which `confirm` cannot reach: `release` + // goes through `commands.invoke`, not `placement.spawn`. + it('rejects within the ack budget when a release invocation never settles', async () => { + const messaging = new FakeMessaging() + messaging.commands.invoke = async (name: string) => ({ + invocationId: 'inv-rel', + actionName: name, + status: 'pending', + }) as RelayActionInvocationAck + messaging.commands.getInvocation = never + const fleet = createClient(messaging, { spawnAckTimeoutMs: 120 }) + + const startedAt = Date.now() + await expect(fleet.release('ar-1-impl', 'done')).rejects.toThrow(/timed out/i) + expect(Date.now() - startedAt).toBeLessThan(2_000) + }) + + /** + * #306 (1) — an ack alone proves the engine accepted the dispatch, not that + * the node launched anything. Ask the SDK to read the invocation back. + */ + it('requests placement confirmation with a budget drawn from the ack deadline', async () => { + const messaging = new FakeMessaging() + const fleet = createClient(messaging, { spawnAckTimeoutMs: 30_000, pollIntervalMs: 250 }) + + await fleet.spawn(spawnInput()) + + const placement = messaging.placements[0]! + expect(placement.confirm).toBe(true) + expect(placement.confirmPollIntervalMs).toBe(250) + expect(placement.confirmTimeoutMs).toBeGreaterThan(0) + expect(placement.confirmTimeoutMs).toBeLessThanOrEqual(30_000) + }) + + /** + * A confirmed placement already carries the terminal invocation, so polling + * for it again would double every spawn's round trips against the same + * budget. + */ + it('uses a confirmed invocation instead of polling for it again', async () => { + const messaging = new FakeMessaging() + let reads = 0 + messaging.commands.getInvocation = async (name: string, invocationId: string) => { + reads += 1 + return completed(invocationId) + } + const spawnPlacement = messaging.placement.spawn.bind(messaging.placement) + messaging.placement.spawn = async (input) => ({ + ...(await spawnPlacement(input)), + status: 'pending', + confirmation: completed('inv-confirmed'), + }) + const fleet = createClient(messaging) + + await expect(fleet.spawn(spawnInput())).resolves.toMatchObject({ sessionRef: 'session-1' }) + expect(reads).toBe(0) + }) +}) diff --git a/src/fleet/relay-fleet-client.ts b/src/fleet/relay-fleet-client.ts index 6d755264..e87722ec 100644 --- a/src/fleet/relay-fleet-client.ts +++ b/src/fleet/relay-fleet-client.ts @@ -83,6 +83,40 @@ const DEFAULT_EXIT_WATCH_INTERVAL_MS = 15_000 const DEFAULT_NODE_OFFLINE_GRACE_MS = 90_000 const DEFAULT_REGISTRATION_GRACE_MS = 60_000 +/** Distinguishes "the call outlived its budget" from any value the call returns. */ +const CALL_TIMED_OUT = Symbol('relay.placement.callTimedOut') + +type CallOutcome = { ok: true; value: T } | { ok: false; error: unknown } + +/** + * A placement call that outlived the operation's remaining budget (#306). + * + * Deliberately **not** registered in `isClassifiedPerItemDispatchFailure`. The + * classified conditions are per-item and self-healing — the unit goes back to + * the queue and the next pass dispatches it. This one is neither. A placement + * that cannot reach a terminal status inside the ack budget is evidence about + * the fleet, not about the work unit: the node is gone, wedged, or running a + * broker that acks without launching. Retrying that against the same fleet + * produces the same timeout, so a run of them is a pass-wide fault and should + * trip the #292 fuse rather than be exempted from it. + * + * Classifying it would rebuild the outage in slow motion — instead of one + * sweep hung forever, an unbounded series of five-minute sweeps that never + * abort, never alert, and never dispatch. The fuse is the alarm; this error is + * meant to reach it. + */ +export class RelaySpawnAckTimeoutError extends Error { + readonly operation: string + readonly timeoutMs: number + + constructor(operation: string, timeoutMs: number) { + super(`Relay placement timed out after ${timeoutMs}ms waiting for ${operation}`) + this.name = 'RelaySpawnAckTimeoutError' + this.operation = operation + this.timeoutMs = timeoutMs + } +} + export class RelayFleetClient implements FleetClient { readonly placementLocality = 'remote' as const readonly durableOwnership = true @@ -170,15 +204,24 @@ export class RelayFleetClient implements FleetClient { } async spawn(input: SpawnInput): Promise { - const messaging = await this.#ensureMessaging() + // One budget for the whole placement: bootstrap, lifecycle registration, + // the placement call and every poll after it share it (#306). Anchoring it + // here rather than inside `#awaitInvocation` is what stops the time already + // spent from being free. + const deadlineAtMs = this.#operationDeadline() + const messaging = await this.#withinDeadline('messaging bootstrap', deadlineAtMs, this.#ensureMessaging()) if (input.capability.startsWith('spawn:')) { - await this.#ensureLifecycleAction(messaging) + await this.#withinDeadline( + 'lifecycle action registration', + deadlineAtMs, + this.#ensureLifecycleAction(messaging), + ) // A transient startup registration failure may have torn down the first // subscription attempt. Re-arm it after the required action is durable so // the invocation cannot be accepted without a live Factory consumer. this.#ensureEventSubscription() } - const ack = await messaging.placement.spawn({ + const ack = await this.#withinDeadline('placement.spawn', deadlineAtMs, messaging.placement.spawn({ capability: input.capability, // 'self' from the orchestrator means "no placement preference": let the // engine pick the least-loaded eligible node. @@ -186,9 +229,22 @@ export class RelayFleetClient implements FleetClient { ...(input.repo ? { repo: input.repo } : {}), input: spawnActionInput(input), ...(this.#options.placementTtlMs !== undefined ? { ttlMs: this.#options.placementTtlMs } : {}), + // An ack proves the engine accepted the dispatch, not that the node + // launched anything: a node advertising `spawn:` on an obsolete + // broker acks and launches nothing, and that is indistinguishable from a + // real spawn until someone reads the invocation back. `confirm` makes the + // SDK do that read, bounded, and fail as `spawn_unconfirmed` instead of + // handing us an ack we would wait on forever (#306). + confirm: true, + confirmTimeoutMs: Math.max(1, deadlineAtMs - this.#now()), + confirmPollIntervalMs: this.#pollIntervalMs, log: this.#log, - }) - const invocation = await this.#awaitInvocation(ack.actionName || 'spawn', ack) + })) + // A confirmed placement already carries the terminal invocation. Polling + // for it again would spend the same budget twice over on a spawn that has + // already proven it launched. + const invocation = ack.confirmation + ?? await this.#awaitInvocation(ack.actionName || 'spawn', ack, deadlineAtMs) const result = spawnResultFromInvocation(input.name, input.sessionRef, invocation, ack) let acknowledgedNode: string try { @@ -244,20 +300,29 @@ export class RelayFleetClient implements FleetClient { } async release(name: string, reason?: string): Promise { - const messaging = await this.#ensureMessaging() - const ack = await messaging.commands.invoke('release', { + // `release` goes through `commands.invoke`, not `placement.spawn`, so the + // SDK's `confirm` cannot bound it — it needs the explicit budget (#306). + // An unbounded release is how the reaper's own teardown wedges. + const deadlineAtMs = this.#operationDeadline() + const messaging = await this.#withinDeadline('messaging bootstrap', deadlineAtMs, this.#ensureMessaging()) + const ack = await this.#withinDeadline('release invoke', deadlineAtMs, messaging.commands.invoke('release', { name, agent: name, ...(reason ? { reason } : {}), - }) - await this.#awaitInvocation(ack.actionName || 'release', ack) + })) + await this.#awaitInvocation(ack.actionName || 'release', ack, deadlineAtMs) this.#tracked.delete(name) this.#syncExitWatcher() } async createPreview(input: PreviewStartInput): Promise { - const messaging = await this.#ensureMessaging() - const ack = await messaging.placement.spawn({ + // Previews reach the same unbounded placement surface as agent spawns and + // need the same budget (#306). They keep the poll-based read-back rather + // than `confirm`: their failure semantics are the preview reaper's, not the + // dispatch fuse's, and bounding is what they were missing. + const deadlineAtMs = this.#operationDeadline() + const messaging = await this.#withinDeadline('messaging bootstrap', deadlineAtMs, this.#ensureMessaging()) + const ack = await this.#withinDeadline('preview placement.spawn', deadlineAtMs, messaging.placement.spawn({ capability: 'preview:tailscale-serve', ...(input.node && input.node !== 'self' ? { node: input.node } : {}), repo: input.repo, @@ -275,32 +340,34 @@ export class RelayFleetClient implements FleetClient { }, ...(this.#options.placementTtlMs !== undefined ? { ttlMs: this.#options.placementTtlMs } : {}), log: this.#log, - }) - const invocation = await this.#awaitInvocation(ack.actionName || 'preview:tailscale-serve', ack) + })) + const invocation = await this.#awaitInvocation(ack.actionName || 'preview:tailscale-serve', ack, deadlineAtMs) return previewReferenceFromInvocation(invocation, ack.placement?.node ?? ack.dispatchedNodeId) } async removePreview(preview: PreviewReference): Promise { - const messaging = await this.#ensureMessaging() - const ack = await messaging.placement.spawn({ + const deadlineAtMs = this.#operationDeadline() + const messaging = await this.#withinDeadline('messaging bootstrap', deadlineAtMs, this.#ensureMessaging()) + const ack = await this.#withinDeadline('preview placement.spawn', deadlineAtMs, messaging.placement.spawn({ capability: 'preview:tailscale-serve', ...(preview.node ? { node: preview.node } : {}), input: { operation: 'remove', preview }, ...(this.#options.placementTtlMs !== undefined ? { ttlMs: this.#options.placementTtlMs } : {}), log: this.#log, - }) - const invocation = await this.#awaitInvocation(ack.actionName || 'preview:tailscale-serve', ack) + })) + const invocation = await this.#awaitInvocation(ack.actionName || 'preview:tailscale-serve', ack, deadlineAtMs) return asRecord(invocation.output)?.removed === true } async reapPreviews(input: PreviewSweepInput): Promise { - const messaging = await this.#ensureMessaging() + const deadlineAtMs = this.#operationDeadline() + const messaging = await this.#withinDeadline('messaging bootstrap', deadlineAtMs, this.#ensureMessaging()) const nodes = (await this.roster()).nodes.filter((node) => node.live && node.capabilities.includes('preview:tailscale-serve'), ) const reports = await Promise.all(nodes.map(async (node): Promise => { try { - const ack = await messaging.placement.spawn({ + const ack = await this.#withinDeadline('preview placement.spawn', deadlineAtMs, messaging.placement.spawn({ capability: 'preview:tailscale-serve', node: node.name, input: { @@ -311,8 +378,8 @@ export class RelayFleetClient implements FleetClient { }, ...(this.#options.placementTtlMs !== undefined ? { ttlMs: this.#options.placementTtlMs } : {}), log: this.#log, - }) - const invocation = await this.#awaitInvocation(ack.actionName || 'preview:tailscale-serve', ack) + })) + const invocation = await this.#awaitInvocation(ack.actionName || 'preview:tailscale-serve', ack, deadlineAtMs) const output = asRecord(invocation.output) return { reaped: Array.isArray(output?.reaped) @@ -543,25 +610,99 @@ export class RelayFleetClient implements FleetClient { return registration.token } - async #awaitInvocation(actionName: string, ack: RelayActionInvocationAck): Promise { - const messaging = await this.#ensureMessaging() + /** The wall-clock instant one placement operation must be finished by. */ + #operationDeadline(): number { + return this.#now() + this.#spawnAckTimeoutMs + } + + /** + * Await `call`, or give up on it once the operation's budget is spent. + * + * The Relay messaging surface takes no `AbortSignal`, so racing the wait is + * the only bound available — it abandons the *wait*, not the call. That is + * the whole point of #306: a deadline read only between calls cannot bound + * any of them. `#awaitInvocation` checked `Date.now() > deadline` at the top + * of its poll loop around an unbounded `commands.getInvocation`, so a read + * that never settled never returned control to the check. Production ran one + * readiness sweep for 62 minutes against a 5-minute bound. + * + * `remainingMs` comes from the shared deadline rather than a per-call + * constant, so many polls cost one budget between them instead of a fresh + * budget each. Rejection is folded into the outcome value so a slow call we + * stopped waiting on cannot later surface as an unhandled rejection. + */ + async #withinDeadline(operation: string, deadlineAtMs: number, call: Promise): Promise { + const remainingMs = deadlineAtMs - this.#now() + if (remainingMs <= 0) { + void call.catch(() => {}) + throw new RelaySpawnAckTimeoutError(operation, this.#spawnAckTimeoutMs) + } + let timer: ReturnType | undefined + try { + const outcome = await Promise.race | typeof CALL_TIMED_OUT>([ + call.then( + (value) => ({ ok: true, value }) as const, + (error) => ({ ok: false, error }) as const, + ), + new Promise((resolve) => { + timer = setTimeout(() => resolve(CALL_TIMED_OUT), remainingMs) + timer.unref?.() + }), + ]) + if (outcome === CALL_TIMED_OUT) throw new RelaySpawnAckTimeoutError(operation, this.#spawnAckTimeoutMs) + if (outcome.ok) return outcome.value + throw outcome.error + } finally { + if (timer) clearTimeout(timer) + } + } + + /** + * Poll an invocation to a terminal status inside `deadlineAtMs`. + * + * Callers that already spent part of the budget — `spawn` burns some on + * bootstrap and placement — pass their own deadline so the total stays + * bounded. A caller that omits it is starting a fresh operation (`release`, + * the preview paths) and gets a full budget of its own. + */ + async #awaitInvocation( + actionName: string, + ack: RelayActionInvocationAck, + deadlineAtMs: number = this.#operationDeadline(), + ): Promise { + const messaging = await this.#withinDeadline( + `${actionName} messaging bootstrap`, + deadlineAtMs, + this.#ensureMessaging(), + ) let status = ack.status ?? 'pending' let invocation: RelayActionInvocation | undefined - const deadline = Date.now() + this.#spawnAckTimeoutMs while (!terminalStatuses.has(status)) { - if (Date.now() > deadline) { - throw new Error(`Timed out waiting for ${actionName} invocation ${ack.invocationId} to complete (last status: ${status})`) + if (this.#now() >= deadlineAtMs) { + throw new RelaySpawnAckTimeoutError( + `${actionName} invocation ${ack.invocationId} to complete (last status: ${status})`, + this.#spawnAckTimeoutMs, + ) } if (!openStatuses.has(status)) { throw new Error(`Unexpected ${actionName} invocation ${ack.invocationId} status: ${status}`) } - await this.#sleep(this.#pollIntervalMs) - invocation = await messaging.commands.getInvocation(actionName, ack.invocationId) + // Never sleep past the deadline: that only buys one more pointless read. + await this.#sleep(Math.max(0, Math.min(this.#pollIntervalMs, deadlineAtMs - this.#now()))) + invocation = await this.#withinDeadline( + `${actionName} invocation ${ack.invocationId}`, + deadlineAtMs, + messaging.commands.getInvocation(actionName, ack.invocationId), + ) status = invocation.status || 'pending' } - invocation ??= await messaging.commands.getInvocation(actionName, ack.invocationId) + invocation ??= await this.#withinDeadline( + `${actionName} invocation ${ack.invocationId}`, + deadlineAtMs, + messaging.commands.getInvocation(actionName, ack.invocationId), + ) if (status === 'failed' || status === 'denied') { throw new Error(`${actionName} invocation ${ack.invocationId} ${status}${invocation.error ? `: ${invocation.error}` : ''}`) } diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 55ef3aaa..35fee51e 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -29,6 +29,7 @@ import { type WorkflowRunnerInput, } from '../index' import { LatePlacementReleasedError, changeEventPath } from './factory' +import { RelaySpawnAckTimeoutError } from '../fleet/relay-fleet-client' import type { AgentWorktree, AgentWorktreeCleanupInspection, AgentWorktreeManager, AgentWorktreeRepository, ChangeEvent, EventPage, GithubConnectionRead, GithubConnectionWrite, GithubIssueStatus, GithubPublishPullRequestInput, GithubWriteback, LinearWriteback, PreviewReference, PreviewStartInput, ProviderSyncStatus, RosterEntry, SlackWriteback, SpawnInput, SpawnResult } from '../ports' import { FakeFleetClient, FakeMountClient, withDeadline } from '../testing' import type { CloseProbePrInput, GithubMergeGatePort, GithubMergeGateVerdict, GithubMergeInput, LinearIssue, VerificationGate, VerificationGateInput, VerificationVerdict } from '../index' @@ -791,6 +792,24 @@ class SpawnFailingFleetClient extends FakeFleetClient { } } +/** + * A fleet whose placement never reaches a terminal ack inside the budget — the + * #306 production failure, once `RelayFleetClient` is able to give up on it. + */ +class SpawnAckTimingOutFleetClient extends FakeFleetClient { + override readonly durableOwnership = true + /** Times out only the agents of these issue keys; others spawn normally. */ + readonly timeOutIssueKeys = new Set() + + override async spawn(input: SpawnInput): Promise { + const wedged = this.timeOutIssueKeys.size === 0 || + [...this.timeOutIssueKeys].some((key) => input.name.includes(`ar-${key}-`)) + if (!wedged) return await super.spawn(input) + this.spawns.push(input) + throw new RelaySpawnAckTimeoutError(`spawn invocation ${input.name} to complete`, 300_000) + } +} + class SpawnDeliveryFailingFleetClient extends FakeFleetClient { override readonly durableOwnership = true @@ -5504,6 +5523,68 @@ describe('FactoryLoop', () => { .rejects.toThrow(/unclassified dispatch failures without a successful dispatch/) expect(fleet.spawns).toEqual([]) }) + + /** + * #306. A spawn whose ack never reaches a terminal status now gives up on + * schedule instead of holding its batch slot forever. The pass must record + * that as a failure, free the slot, and carry on with the other units — + * the 62-minute production sweep did none of the three. + */ + it('releases the batch slot and counts the failure when a spawn ack times out (#306)', async () => { + const mount = twoReadyIssues() + const fleet = new SpawnAckTimingOutFleetClient() + fleet.timeOutIssueKeys.add('59') + const factory = createFactory(config({ issueSource: 'github', batchSize: 4 }), { + mount, + fleet, + triage: new StaticTriage(), + githubWriteback: new RecordingGithubWriteback(), + }) + + const report = await factory.runOnce() + + // Counted as an unexplained fault, not waved through as a known + // per-item condition: the fleet could not place, and that is a fact + // about the fleet. + expect(factory.status().counters.dispatchItemFailuresSkipped).toBe(1) + expect(factory.status().counters.dispatchItemsSkippedUndispatchable).toBeUndefined() + expect(report.skipped.map((skip) => skip.issue.key)).toEqual(['59']) + // The public reason stays the sanitized generic form: `contextualError` + // wraps the spawn failure, and `telemetryErrorClass` reads the outer + // class. The timeout is identified by its counter, the fuse and the + // operator log rather than by the run report — widening the report's + // classification is a separate change from bounding the call. + expect(report.skipped[0]?.reason).toBe('dispatch failed (Error)') + // The slot the wedged unit took is free again: the next unit dispatches + // inside the same pass rather than queueing behind a hang. + expect(report.dispatched.map((result) => result.issue.key)).toEqual(['60']) + expect(factory.status().dispatchCapacity.waiting).toBe(0) + }) + + /** + * #306, and the justification for leaving `RelaySpawnAckTimeoutError` + * unclassified. Unlike a late-placement release, a spawn-ack timeout is not + * self-healing: retrying it against the same fleet times out again. A run + * of them is a pass-wide fault and must reach the #292 fuse, or the outage + * simply returns in five-minute instalments — every sweep failing, no sweep + * ever complaining. + */ + it('lets repeated spawn-ack timeouts trip the #292 fuse (#306)', async () => { + const paths = [91, 92, 93, 94, 95, 96].map((number) => githubIssuePath('AgentWorkforce', 'pear', number)) + const mount = new FakeMountClient(Object.fromEntries( + paths.map((path, index) => [path, githubIssueFile(91 + index, { labels: ['factory', 'pear'] })]), + )) + const fleet = new SpawnAckTimingOutFleetClient() + const factory = createFactory(config({ issueSource: 'github', batchSize: 5 }), { + mount, + fleet, + triage: new StaticTriage(), + githubWriteback: new RecordingGithubWriteback(), + }) + + await expect(factory.runOnce()) + .rejects.toThrow(/unclassified dispatch failures without a successful dispatch/) + }) }) it('adopts a same-repo legacy PR and wakes its babysitter when REST metadata later becomes conflicting', async () => { From 7f3b55a5152af7b72c95e7996f8da0ae5a929680 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Fri, 21 Aug 2026 08:53:29 +0200 Subject: [PATCH 2/3] fix(factory): bound the preview roster read and check the budget before calling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three independent reviewers found the same gap: `reapPreviews` bounded its placement calls but awaited `roster()` outside the budget, and `roster()` is three unbounded reads (`agents.presence`, `agents.list`, `nodes.list`). `#reapPreviewOrphans` keeps the sweep promise in `#previewSweepInFlight` and schedules the next sweep only from its `.finally()`, so one stalled read stops preview cleanup permanently — the exact hang this branch removes everywhere else on the path. `#withinDeadline` also took an already-started promise, so an exhausted budget still issued the request before rejecting. On the placement path that request mutates: it can launch a worker this process has already decided it has no time to wait for, then abandon it. Take a thunk and invoke it only after the remaining-time check, so a spent budget refuses before anything is sent. That also moves the `confirmTimeoutMs` computation behind the check, where it belongs. Both are pinned by tests that fail before the change: the roster hang as a 5025ms harness timeout, and the eager call as a placement recorded against a budget that was already gone. Co-Authored-By: Claude Opus 5 Session-Id: 1e357a8b-0ac7-4606-992e-e5151f53afb6 --- src/fleet/relay-fleet-client.test.ts | 43 +++++++++++++++++++++++++ src/fleet/relay-fleet-client.ts | 47 +++++++++++++++++----------- 2 files changed, 72 insertions(+), 18 deletions(-) diff --git a/src/fleet/relay-fleet-client.test.ts b/src/fleet/relay-fleet-client.test.ts index 43145f35..9b251c9b 100644 --- a/src/fleet/relay-fleet-client.test.ts +++ b/src/fleet/relay-fleet-client.test.ts @@ -1297,6 +1297,49 @@ describe('RelayFleetClient placement deadlines (#306)', () => { expect(placement.confirmTimeoutMs).toBeLessThanOrEqual(30_000) }) + /** + * #307 review (codex, coderabbit, cubic — three independent finds). + * + * `reapPreviews` bounded its placement calls but awaited `roster()` outside + * the budget, and `roster()` is three unbounded reads. `#reapPreviewOrphans` + * keeps the sweep promise in `#previewSweepInFlight` and schedules the next + * sweep only from its `.finally()`, so one stalled read stops preview + * cleanup permanently — the exact hang this change removes elsewhere. + */ + it('rejects within the ack budget when the preview roster read never resolves', async () => { + const messaging = new FakeMessaging() + messaging.nodes.list = never + const fleet = createClient(messaging, { spawnAckTimeoutMs: 120 }) + + const startedAt = Date.now() + await expect(fleet.reapPreviews({ namespace: 'ns', activeOwners: [] })).rejects.toThrow(/timed out/i) + expect(Date.now() - startedAt).toBeLessThan(2_000) + }) + + /** + * #307 review (cubic). The budget must be checked *before* the request is + * made, not after. Taking an already-started promise meant an exhausted + * deadline still fired the call — a mutating one here — and then abandoned + * it, which is how a local timeout orphans a remote spawn. + * + * `now` is stepped so the budget is intact through bootstrap and lifecycle + * registration and spent by the time placement is reached. + */ + it('does not issue the placement call once the budget is already spent', async () => { + const messaging = new FakeMessaging() + let calls = 0 + const now = () => { + calls += 1 + return calls <= 3 ? 0 : 10_000 + } + const fleet = createClient(messaging, { spawnAckTimeoutMs: 1_000, now }) + + await expect(fleet.spawn(spawnInput())).rejects.toThrow(/timed out/i) + + // The mutating call was never made, so there is no remote spawn to orphan. + expect(messaging.placements).toEqual([]) + }) + /** * A confirmed placement already carries the terminal invocation, so polling * for it again would double every spawn's round trips against the same diff --git a/src/fleet/relay-fleet-client.ts b/src/fleet/relay-fleet-client.ts index e87722ec..fa381863 100644 --- a/src/fleet/relay-fleet-client.ts +++ b/src/fleet/relay-fleet-client.ts @@ -209,19 +209,19 @@ export class RelayFleetClient implements FleetClient { // here rather than inside `#awaitInvocation` is what stops the time already // spent from being free. const deadlineAtMs = this.#operationDeadline() - const messaging = await this.#withinDeadline('messaging bootstrap', deadlineAtMs, this.#ensureMessaging()) + const messaging = await this.#withinDeadline('messaging bootstrap', deadlineAtMs, () => this.#ensureMessaging()) if (input.capability.startsWith('spawn:')) { await this.#withinDeadline( 'lifecycle action registration', deadlineAtMs, - this.#ensureLifecycleAction(messaging), + () => this.#ensureLifecycleAction(messaging), ) // A transient startup registration failure may have torn down the first // subscription attempt. Re-arm it after the required action is durable so // the invocation cannot be accepted without a live Factory consumer. this.#ensureEventSubscription() } - const ack = await this.#withinDeadline('placement.spawn', deadlineAtMs, messaging.placement.spawn({ + const ack = await this.#withinDeadline('placement.spawn', deadlineAtMs, () => messaging.placement.spawn({ capability: input.capability, // 'self' from the orchestrator means "no placement preference": let the // engine pick the least-loaded eligible node. @@ -304,8 +304,8 @@ export class RelayFleetClient implements FleetClient { // SDK's `confirm` cannot bound it — it needs the explicit budget (#306). // An unbounded release is how the reaper's own teardown wedges. const deadlineAtMs = this.#operationDeadline() - const messaging = await this.#withinDeadline('messaging bootstrap', deadlineAtMs, this.#ensureMessaging()) - const ack = await this.#withinDeadline('release invoke', deadlineAtMs, messaging.commands.invoke('release', { + const messaging = await this.#withinDeadline('messaging bootstrap', deadlineAtMs, () => this.#ensureMessaging()) + const ack = await this.#withinDeadline('release invoke', deadlineAtMs, () => messaging.commands.invoke('release', { name, agent: name, ...(reason ? { reason } : {}), @@ -321,8 +321,8 @@ export class RelayFleetClient implements FleetClient { // than `confirm`: their failure semantics are the preview reaper's, not the // dispatch fuse's, and bounding is what they were missing. const deadlineAtMs = this.#operationDeadline() - const messaging = await this.#withinDeadline('messaging bootstrap', deadlineAtMs, this.#ensureMessaging()) - const ack = await this.#withinDeadline('preview placement.spawn', deadlineAtMs, messaging.placement.spawn({ + const messaging = await this.#withinDeadline('messaging bootstrap', deadlineAtMs, () => this.#ensureMessaging()) + const ack = await this.#withinDeadline('preview placement.spawn', deadlineAtMs, () => messaging.placement.spawn({ capability: 'preview:tailscale-serve', ...(input.node && input.node !== 'self' ? { node: input.node } : {}), repo: input.repo, @@ -347,8 +347,8 @@ export class RelayFleetClient implements FleetClient { async removePreview(preview: PreviewReference): Promise { const deadlineAtMs = this.#operationDeadline() - const messaging = await this.#withinDeadline('messaging bootstrap', deadlineAtMs, this.#ensureMessaging()) - const ack = await this.#withinDeadline('preview placement.spawn', deadlineAtMs, messaging.placement.spawn({ + const messaging = await this.#withinDeadline('messaging bootstrap', deadlineAtMs, () => this.#ensureMessaging()) + const ack = await this.#withinDeadline('preview placement.spawn', deadlineAtMs, () => messaging.placement.spawn({ capability: 'preview:tailscale-serve', ...(preview.node ? { node: preview.node } : {}), input: { operation: 'remove', preview }, @@ -361,13 +361,20 @@ export class RelayFleetClient implements FleetClient { async reapPreviews(input: PreviewSweepInput): Promise { const deadlineAtMs = this.#operationDeadline() - const messaging = await this.#withinDeadline('messaging bootstrap', deadlineAtMs, this.#ensureMessaging()) - const nodes = (await this.roster()).nodes.filter((node) => + const messaging = await this.#withinDeadline('messaging bootstrap', deadlineAtMs, () => this.#ensureMessaging()) + // The roster is three unbounded reads (`agents.presence`, `agents.list`, + // `nodes.list`). Leaving it outside the budget left exactly the hang this + // change removes everywhere else: `#reapPreviewOrphans` keeps the sweep + // promise in `#previewSweepInFlight` and schedules the next sweep only from + // its `.finally()`, so one stalled read stops preview cleanup for good + // (#307 review, codex/coderabbit/cubic). + const roster = await this.#withinDeadline('preview roster', deadlineAtMs, () => this.roster()) + const nodes = roster.nodes.filter((node) => node.live && node.capabilities.includes('preview:tailscale-serve'), ) const reports = await Promise.all(nodes.map(async (node): Promise => { try { - const ack = await this.#withinDeadline('preview placement.spawn', deadlineAtMs, messaging.placement.spawn({ + const ack = await this.#withinDeadline('preview placement.spawn', deadlineAtMs, () => messaging.placement.spawn({ capability: 'preview:tailscale-serve', node: node.name, input: { @@ -630,17 +637,21 @@ export class RelayFleetClient implements FleetClient { * constant, so many polls cost one budget between them instead of a fresh * budget each. Rejection is folded into the outcome value so a slow call we * stopped waiting on cannot later surface as an unhandled rejection. + * + * Takes a thunk rather than a promise so an exhausted budget refuses *before* + * the request is made. An already-started promise would mean the call had + * been sent — a mutating one, on the placement path — while this method was + * still deciding there was no time left to make it (#307 review, cubic). */ - async #withinDeadline(operation: string, deadlineAtMs: number, call: Promise): Promise { + async #withinDeadline(operation: string, deadlineAtMs: number, start: () => Promise): Promise { const remainingMs = deadlineAtMs - this.#now() if (remainingMs <= 0) { - void call.catch(() => {}) throw new RelaySpawnAckTimeoutError(operation, this.#spawnAckTimeoutMs) } let timer: ReturnType | undefined try { const outcome = await Promise.race | typeof CALL_TIMED_OUT>([ - call.then( + start().then( (value) => ({ ok: true, value }) as const, (error) => ({ ok: false, error }) as const, ), @@ -673,7 +684,7 @@ export class RelayFleetClient implements FleetClient { const messaging = await this.#withinDeadline( `${actionName} messaging bootstrap`, deadlineAtMs, - this.#ensureMessaging(), + () => this.#ensureMessaging(), ) let status = ack.status ?? 'pending' let invocation: RelayActionInvocation | undefined @@ -693,7 +704,7 @@ export class RelayFleetClient implements FleetClient { invocation = await this.#withinDeadline( `${actionName} invocation ${ack.invocationId}`, deadlineAtMs, - messaging.commands.getInvocation(actionName, ack.invocationId), + () => messaging.commands.getInvocation(actionName, ack.invocationId), ) status = invocation.status || 'pending' } @@ -701,7 +712,7 @@ export class RelayFleetClient implements FleetClient { invocation ??= await this.#withinDeadline( `${actionName} invocation ${ack.invocationId}`, deadlineAtMs, - messaging.commands.getInvocation(actionName, ack.invocationId), + () => messaging.commands.getInvocation(actionName, ack.invocationId), ) if (status === 'failed' || status === 'denied') { throw new Error(`${actionName} invocation ${ack.invocationId} ${status}${invocation.error ? `: ${invocation.error}` : ''}`) From 7b4b2770385d86b58ffd65c02ddb8e831875e392 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Fri, 21 Aug 2026 09:14:46 +0200 Subject: [PATCH 3/3] fix(factory): release placements Relay accepts after the local deadline MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bounding the wait does not cancel the call. Two paths could therefore leave a live worker that this process had already reported as failed — trading the infinite hang for a leaked agent, which is not a trade worth making. 1. An abandoned `placement.spawn`. The in-flight promise was dropped on timeout, so a placement Relay accepted a moment later launched a worker nothing was tracking. `#withinDeadline` now takes an `onAbandoned` hook that receives the still-in-flight call, and the spawn path uses it to release whatever the placement ultimately lands. 2. An acked placement whose invocation poll ran out of budget. This is the certain leak rather than the possible one: holding an ack means Relay accepted the placement, and we were throwing that ack away. Both now route through `#releaseLatePlacement`, which tracks the agent *before* attempting the release so a failed cleanup is retried by the reconciliation loop rather than forgotten — the same retain-and-retry the existing `unverified-placement` path uses. This is the client-side counterpart to #304's `LatePlacementReleasedError` / `#releaseOrphanedLatePlacement`. Tests fail before the change with `expected [] to have a length of 1`: no release was issued at all. A must-not-fire control pins that a placement which genuinely fails launches nothing and so is never released. Co-Authored-By: Claude Opus 5 Session-Id: 1e357a8b-0ac7-4606-992e-e5151f53afb6 --- src/fleet/relay-fleet-client.test.ts | 74 +++++++++++++++++++++ src/fleet/relay-fleet-client.ts | 99 +++++++++++++++++++++++++--- 2 files changed, 164 insertions(+), 9 deletions(-) diff --git a/src/fleet/relay-fleet-client.test.ts b/src/fleet/relay-fleet-client.test.ts index 9b251c9b..f35a108d 100644 --- a/src/fleet/relay-fleet-client.test.ts +++ b/src/fleet/relay-fleet-client.test.ts @@ -1316,6 +1316,80 @@ describe('RelayFleetClient placement deadlines (#306)', () => { expect(Date.now() - startedAt).toBeLessThan(2_000) }) + /** + * #307 review (cubic P1) — the finding that matters most, because getting it + * wrong trades an infinite hang for a leaked agent. + * + * Relay ACCEPTS the placement, but its response arrives after our local + * deadline. A worker is live on the fleet and this process has already + * reported the spawn as failed, so nothing downstream knows to release it. + * Abandoning the wait must not mean abandoning the worker — the same shape + * #304 handles one layer up with `LatePlacementReleasedError`. + */ + it('releases a placement Relay accepts after the local deadline expired', async () => { + const messaging = new FakeMessaging() + messaging.placement.spawn = async (input) => { + messaging.placements.push(input) + return await after(200, { + invocationId: 'inv-late', + actionName: 'spawn', + status: 'completed', + node: { name: 'node-a' } as RelayNode, + placement: { capability: input.capability, node: 'node-a', attempts: 1, queued: false }, + }) + } + const fleet = createClient(messaging, { spawnAckTimeoutMs: 60 }) + + await expect(fleet.spawn(spawnInput())).rejects.toThrow(/timed out/i) + + // The late placement lands after we gave up; the worker must be torn down. + await vi.waitFor(() => { + expect(messaging.invokes.filter((invoke) => invoke.name === 'release')).toHaveLength(1) + }, { timeout: 3_000 }) + const release = messaging.invokes.find((invoke) => invoke.name === 'release') + expect(release?.input).toMatchObject({ name: 'ar-1-impl', reason: 'late-placement-timeout' }) + // Released cleanly, so nothing is left retaining it. + expect(fleet.trackedAgents().has('ar-1-impl')).toBe(false) + }) + + /** + * The certain leak, as opposed to the possible one above: we already hold an + * ack, so Relay definitely accepted the placement. Running out of budget + * while polling the invocation must still release the worker. + */ + it('releases an acked placement when the invocation poll exhausts the budget', async () => { + const messaging = new FakeMessaging() + messaging.placementAck = { invocationId: 'inv-1', status: 'pending', placement: { node: 'node-a' } } + messaging.commands.getInvocation = never + const fleet = createClient(messaging, { spawnAckTimeoutMs: 100 }) + + await expect(fleet.spawn(spawnInput())).rejects.toThrow(/timed out/i) + + const releases = messaging.invokes.filter((invoke) => invoke.name === 'release') + expect(releases).toHaveLength(1) + expect(releases[0]?.input).toMatchObject({ name: 'ar-1-impl', reason: 'late-placement-timeout' }) + }) + + /** + * MUST-NOT-FIRE. A placement that genuinely failed launched nothing, so + * issuing a release would be a spurious teardown against a worker that never + * existed. + */ + it('does not release when the abandoned placement ultimately fails', async () => { + const messaging = new FakeMessaging() + messaging.placement.spawn = async () => { + await after(120, undefined) + throw new Error('placement rejected') + } + const fleet = createClient(messaging, { spawnAckTimeoutMs: 60 }) + + await expect(fleet.spawn(spawnInput())).rejects.toThrow(/timed out/i) + await after(300, undefined) + + expect(messaging.invokes.filter((invoke) => invoke.name === 'release')).toEqual([]) + expect(fleet.trackedAgents().has('ar-1-impl')).toBe(false) + }) + /** * #307 review (cubic). The budget must be checked *before* the request is * made, not after. Taking an already-started promise meant an exhausted diff --git a/src/fleet/relay-fleet-client.ts b/src/fleet/relay-fleet-client.ts index fa381863..90693f3f 100644 --- a/src/fleet/relay-fleet-client.ts +++ b/src/fleet/relay-fleet-client.ts @@ -82,6 +82,15 @@ const DEFAULT_EXIT_WATCH_INTERVAL_MS = 15_000 // synthesize a false exit. const DEFAULT_NODE_OFFLINE_GRACE_MS = 90_000 const DEFAULT_REGISTRATION_GRACE_MS = 60_000 +/** + * Compensating release for a placement Relay accepted after we stopped waiting. + * + * Giving up locally does not cancel a remote spawn: the engine may already have + * launched the worker. Without this the #306 bound would trade an infinite hang + * for a leaked agent — the same shape #304 fixed with `LatePlacementReleasedError` + * and `#releaseOrphanedLatePlacement` one layer up (#307 review, cubic). + */ +const LATE_PLACEMENT_RELEASE_REASON = 'late-placement-timeout' /** Distinguishes "the call outlived its budget" from any value the call returns. */ const CALL_TIMED_OUT = Symbol('relay.placement.callTimedOut') @@ -239,12 +248,27 @@ export class RelayFleetClient implements FleetClient { confirmTimeoutMs: Math.max(1, deadlineAtMs - this.#now()), confirmPollIntervalMs: this.#pollIntervalMs, log: this.#log, - })) + // Giving up on the wait does not cancel the placement. If Relay accepts + // it after we have already reported failure, a worker is live that + // nothing is tracking — so release it (#307 review, cubic). + }), (inFlight) => this.#releaseAbandonedPlacement(input.name, inFlight)) // A confirmed placement already carries the terminal invocation. Polling // for it again would spend the same budget twice over on a spawn that has // already proven it launched. - const invocation = ack.confirmation - ?? await this.#awaitInvocation(ack.actionName || 'spawn', ack, deadlineAtMs) + let invocation: RelayActionInvocation + try { + invocation = ack.confirmation + ?? await this.#awaitInvocation(ack.actionName || 'spawn', ack, deadlineAtMs) + } catch (error) { + // Holding an ack means Relay accepted the placement, so running out of + // budget while polling leaves a worker we asked for and never adopted. + // This is the certain leak; the abandoned-placement case above is the + // possible one. + if (error instanceof RelaySpawnAckTimeoutError) { + await this.#releaseLatePlacement(input.name, ack) + } + throw error + } const result = spawnResultFromInvocation(input.name, input.sessionRef, invocation, ack) let acknowledgedNode: string try { @@ -643,24 +667,40 @@ export class RelayFleetClient implements FleetClient { * been sent — a mutating one, on the placement path — while this method was * still deciding there was no time left to make it (#307 review, cubic). */ - async #withinDeadline(operation: string, deadlineAtMs: number, start: () => Promise): Promise { + async #withinDeadline( + operation: string, + deadlineAtMs: number, + start: () => Promise, + /** + * Called when the budget runs out with the call still in flight, receiving + * the abandoned work. Abandoning the *wait* does not cancel the *call*, so + * a mutating operation needs a way to clean up whatever it still lands. + */ + onAbandoned?: (inFlight: Promise>) => void, + ): Promise { const remainingMs = deadlineAtMs - this.#now() if (remainingMs <= 0) { throw new RelaySpawnAckTimeoutError(operation, this.#spawnAckTimeoutMs) } let timer: ReturnType | undefined try { + // Folded to an outcome once, so the abandoned handler shares this promise + // rather than attaching a second rejection path to the same call. + const inFlight: Promise> = start().then( + (value) => ({ ok: true, value }) as const, + (error) => ({ ok: false, error }) as const, + ) const outcome = await Promise.race | typeof CALL_TIMED_OUT>([ - start().then( - (value) => ({ ok: true, value }) as const, - (error) => ({ ok: false, error }) as const, - ), + inFlight, new Promise((resolve) => { timer = setTimeout(() => resolve(CALL_TIMED_OUT), remainingMs) timer.unref?.() }), ]) - if (outcome === CALL_TIMED_OUT) throw new RelaySpawnAckTimeoutError(operation, this.#spawnAckTimeoutMs) + if (outcome === CALL_TIMED_OUT) { + onAbandoned?.(inFlight) + throw new RelaySpawnAckTimeoutError(operation, this.#spawnAckTimeoutMs) + } if (outcome.ok) return outcome.value throw outcome.error } finally { @@ -668,6 +708,47 @@ export class RelayFleetClient implements FleetClient { } } + /** + * Release a worker whose placement Relay accepted after the local deadline. + * + * Tracked *before* the release is attempted so a failed cleanup is retried by + * the reconciliation loop instead of forgotten — the same retain-and-retry the + * `unverified-placement` path uses. Losing the release would leave a live + * worker outside Factory's lifecycle tracking. + */ + async #releaseLatePlacement(name: string, ack: RelayActionInvocationAck & { placement?: { node?: string } }): Promise { + const node = ack.placement?.node ?? ack.dispatchedNodeId + this.#track(name, { + invocationId: ack.invocationId, + ...(node ? { node } : {}), + pendingReleaseReason: LATE_PLACEMENT_RELEASE_REASON, + }) + this.#log(`Releasing ${name}: Relay accepted its placement after the local deadline expired`) + try { + await this.release(name, LATE_PLACEMENT_RELEASE_REASON) + } catch (error) { + // Retained above, so reconciliation retries the idempotent release. + this.#log(`Failed to release late placement ${name}: ${errorMessage(error)}`) + } + } + + /** + * Attach a compensating release to a placement we stopped waiting for. + * + * If the call ultimately fails, nothing was launched and there is nothing to + * release. If it succeeds, a worker exists that this process has already + * reported as failed, so it must be torn down. + */ + #releaseAbandonedPlacement( + name: string, + inFlight: Promise>, + ): void { + void inFlight.then(async (outcome) => { + if (!outcome.ok) return + await this.#releaseLatePlacement(name, outcome.value) + }) + } + /** * Poll an invocation to a terminal status inside `deadlineAtMs`. *