From f32d2af34fc2daf6489cd4f4fe47ae4c99a3350b Mon Sep 17 00:00:00 2001 From: kjgbot Date: Sun, 6 Sep 2026 19:53:01 +0200 Subject: [PATCH 1/2] fix(babysitter): defer unreadable PRs and release unplaced claims Session-Id: 01a077c7-5a1f-7a50-9b0f-e5b918dcc472 --- src/fleet/internal-fleet-client.test.ts | 12 +- src/fleet/internal-fleet-client.ts | 13 +- src/fleet/relay-fleet-client.test.ts | 3 +- src/fleet/relay-fleet-client.ts | 54 +++-- src/orchestrator/factory.test.ts | 251 +++++++++++++++++++++--- src/orchestrator/factory.ts | 101 ++++++++-- src/ports/fleet.ts | 8 + src/ports/index.ts | 2 +- 8 files changed, 377 insertions(+), 67 deletions(-) diff --git a/src/fleet/internal-fleet-client.test.ts b/src/fleet/internal-fleet-client.test.ts index fa343c3e..70280198 100644 --- a/src/fleet/internal-fleet-client.test.ts +++ b/src/fleet/internal-fleet-client.test.ts @@ -1,4 +1,5 @@ import { describe, expect, it, vi } from 'vitest' +import { FleetSpawnNotCreatedError } from '../ports/fleet' import { mkdir, mkdtemp, rm, symlink, writeFile } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' @@ -288,6 +289,15 @@ describe('InternalFleetClient', () => { ) }) + it('preserves uncertainty when the broker rejects a spawn without placement evidence', async () => { + const client = new FakeHarnessDriverClient() + const error = new Error('spawn acknowledgement unavailable') + vi.spyOn(client, 'spawnPty').mockRejectedValueOnce(error) + const fleet = new InternalFleetClient({ client, resolveAgentRelayMcpCommand: () => undefined }) + await expect(fleet.spawn({ name: 'ar-1-impl', capability: 'spawn:codex' })).rejects.toBe(error) + await fleet.dispose() + }) + it('fails closed when an identity proof cannot be installed', async () => { const fleet = new InternalFleetClient({ client: new FakeHarnessDriverClient(), @@ -298,7 +308,7 @@ describe('InternalFleetClient', () => { name: 'ar-1-impl', capability: 'spawn:codex', identityKey: 'factory:dispatch:v1:github:agentworkforce/factory#1:implementer', - })).rejects.toThrow(/identity proof cannot be installed/) + })).rejects.toBeInstanceOf(FleetSpawnNotCreatedError) }) it('resolves the Factory-extended Agent Relay MCP command from the packed entrypoint', async () => { diff --git a/src/fleet/internal-fleet-client.ts b/src/fleet/internal-fleet-client.ts index 0a9cee9b..05d0a24b 100644 --- a/src/fleet/internal-fleet-client.ts +++ b/src/fleet/internal-fleet-client.ts @@ -10,7 +10,7 @@ import type { BrokerEvent, ListAgent, SendMessageInput, SpawnPtyInput } from '@a import type { PreviewConfig } from '../config/schema' import type { AgentMessage, AgentPidResolution, AgentUsage, Capability, FleetClient, FleetTrackedAgent, PreviewReference, PreviewStartInput, PreviewSweepInput, PreviewSweepResult, RosterEntry, SendInput, SpawnInput, SpawnResult, TeammateAgent, TeammateQuery } from '../ports/fleet' -import { FleetDeliveryRejectedError } from '../ports/fleet' +import { FleetDeliveryRejectedError, FleetSpawnNotCreatedError } from '../ports/fleet' import type { Logger } from '../ports/system' import { normalizeLogger } from '../logging' import { TailscalePreviewManager, type PreviewManager } from '../node/tailscale-preview' @@ -265,6 +265,16 @@ export class InternalFleetClient implements FleetClient { } async spawn(input: SpawnInput): Promise { + let placementAttempted = false + try { + return await this.#spawn(input, () => { placementAttempted = true }) + } catch (error) { + if (!placementAttempted) throw new FleetSpawnNotCreatedError(error) + throw error + } + } + + async #spawn(input: SpawnInput, onPlacementAttempt: () => void): Promise { assertSelfNode(input.node) // The broker has no delivery-target registration event. Subscribe before // spawn so worker_ready can act as a bounded re-send trigger if the child's @@ -288,6 +298,7 @@ export class InternalFleetClient implements FleetClient { this.#tracked.set(input.name, { invocationId: input.invocationId }) let handle: SpawnedHandleLike try { + onPlacementAttempt() handle = await this.#callBroker('spawnPty', (client) => client.spawnPty(spawnInput)) } catch (error) { this.#trackAgentExit(input.name) diff --git a/src/fleet/relay-fleet-client.test.ts b/src/fleet/relay-fleet-client.test.ts index b070dd4a..e0e048fc 100644 --- a/src/fleet/relay-fleet-client.test.ts +++ b/src/fleet/relay-fleet-client.test.ts @@ -1,4 +1,5 @@ import { describe, expect, it, vi } from 'vitest' +import { FleetSpawnNotCreatedError } from '../ports/fleet' import { describeControlPlaneError } from './control-plane-circuit' import { FactoryAgentRegistrationError, MAX_REGISTRATION_ATTEMPTS, ReadOnlyFleetIdentityError, RelayFleetClient, type RelayClientFactoryOptions, type RelayClientLike } from './relay-fleet-client' @@ -332,7 +333,7 @@ describe('RelayFleetClient', () => { name: 'ar-6-impl', capability: 'spawn:codex', repo: 'AgentWorkforce/factory', - })).rejects.toThrow(/provisionSandbox returned no nodeName/) + })).rejects.toBeInstanceOf(FleetSpawnNotCreatedError) expect(messaging.placements).toHaveLength(0) }) diff --git a/src/fleet/relay-fleet-client.ts b/src/fleet/relay-fleet-client.ts index d3ed324c..f419ea02 100644 --- a/src/fleet/relay-fleet-client.ts +++ b/src/fleet/relay-fleet-client.ts @@ -5,6 +5,7 @@ import { AgentRelay } from '@agent-relay/sdk' import { describeControlPlaneError } from './control-plane-circuit' import { resolveRelayAgentToken, resolveRelayWorkspaceKey } from './relay-workspace-key' +import { FleetSpawnNotCreatedError } from '../ports/fleet' import type { AgentLifecycleSignal, AgentMessage, AgentUsage, Capability, FleetClient, FleetConnectStatus, NodeCapability, PreviewReference, PreviewStartInput, PreviewSweepInput, PreviewSweepResult, RosterEntry, SendInput, SpawnInput, SpawnResult, TeammateAgent, TeammateQuery } from '../ports/fleet' import { RelaycastTeammateDirectory, type TeammateDirectory } from './teammates' import type { @@ -368,6 +369,16 @@ export class RelayFleetClient implements FleetClient { } async spawn(input: SpawnInput): Promise { + let placementAttempted = false + try { + return await this.#spawn(input, () => { placementAttempted = true }) + } catch (error) { + if (!placementAttempted) throw new FleetSpawnNotCreatedError(error) + throw error + } + } + + async #spawn(input: SpawnInput, onPlacementAttempt: () => void): Promise { // 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 @@ -432,26 +443,29 @@ export class RelayFleetClient implements FleetClient { // means "no placement preference". const resolvedNode = sandboxTargetNode ?? (input.node && input.node !== 'self' ? input.node : undefined) - const ack = await this.#withinDeadline('placement.spawn', deadlineAtMs, () => messaging.placement.spawn({ - capability: input.capability, - ...(resolvedNode ? { node: resolvedNode } : {}), - ...(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, - // 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)) + const ack = await this.#withinDeadline('placement.spawn', deadlineAtMs, () => { + onPlacementAttempt() + return messaging.placement.spawn({ + capability: input.capability, + ...(resolvedNode ? { node: resolvedNode } : {}), + ...(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, + // 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. diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 99d3e14b..83aed1ea 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -30,6 +30,7 @@ import { } from '../index' import { LatePlacementReleasedError, changeEventPath, defaultMergeGate } from './factory' import type { GhRunner } from '../github' +import { FleetSpawnNotCreatedError } from '../ports/fleet' import { RelaySpawnAckTimeoutError } from '../fleet/relay-fleet-client' import { RelayfileOperationTimeoutError } from '../mount/relayfile-operation-timeout' import type { AgentSpec, AgentWorktree, AgentWorktreeCleanupInspection, AgentWorktreeManager, AgentWorktreeRepository, ChangeEvent, EventPage, GithubConnectionRead, GithubConnectionWrite, GithubIssueStatus, GithubIssueCloseWriteResult, GithubPublishPullRequestInput, GithubStatusClaimReceipt, GithubStatusWriteResult, GithubWriteback, LinearWriteback, PreviewReference, PreviewStartInput, ProviderSyncStatus, RosterEntry, SandboxPushInput, SandboxPushResult, SlackWriteback, SpawnInput, SpawnResult } from '../ports' @@ -6810,6 +6811,7 @@ describe('FactoryLoop', () => { body: 'No issue number is required outside the verified legacy head.', state: 'open', draft: false, + labels: [], html_url: 'https://github.com/AgentWorkforce/pear/pull/153', head: { ref: '53-legacy-schedule-repair', @@ -6913,6 +6915,9 @@ describe('FactoryLoop', () => { const mount = new FakeMountClient({ [path]: githubIssueFile(53, { labels: ['factory', 'pear', 'factory:in-progress'] }), }) + mount.files.set('/github/repos/AgentWorkforce/pear/pulls/153/metadata.json', { content: { + number: 153, state: 'open', draft: false, labels: [], head_ref: 'factory/53-agentworkforce-pear-proof', + } }) const fleet = new RemoteLifecycleFleetClient() const worktrees = new RecordingWorktreeManager() const factory = createFactory(config({ @@ -7041,6 +7046,9 @@ describe('FactoryLoop', () => { updatedAt: '2026-07-18T11:00:00.000Z', }), }) + mount.files.set('/github/repos/AgentWorkforce/pear/pulls/153/metadata.json', { content: { + number: 153, state: 'open', draft: false, labels: [], head_ref: '53-agentworkforce-pear-proof', + } }) const firstFleet = new RemoteLifecycleFleetClient() const worktrees = new RecordingWorktreeManager() const githubWriteback = new RecordingGithubWriteback() @@ -10806,6 +10814,7 @@ describe('FactoryLoop', () => { head_ref: branch, state: 'open', draft: false, + labels: [], } }) fleet.emitAgentExit('ar-591-impl-pear', 'reconciled-missing') @@ -10857,6 +10866,7 @@ describe('FactoryLoop', () => { head_ref: branch, state: 'open', draft: false, + labels: [], } }) fleet.emitAgentExit('ar-597-impl-pear', 'reconciled-missing') @@ -11292,6 +11302,7 @@ describe('FactoryLoop', () => { number: 1186, state: 'open', draft: false, + labels: [], head_ref: branch, url: 'https://github.com/AgentWorkforce/pear/pull/1186', } }) @@ -20263,6 +20274,7 @@ describe('FactoryLoop', () => { url: `https://github.com/${input.repo}/pull/${prNumber}`, state: 'open', draft: false, + labels: [], }, }) return { @@ -21691,6 +21703,7 @@ describe('FactoryLoop', () => { head_ref: branch, state: 'open', draft: false, + labels: [], } }) fleet.emitAgentExit('ar-93-impl-pear', 'crash') @@ -28082,7 +28095,7 @@ describe('FactoryLoop PR babysitter', () => { // open/draft/merged, never calls gh. const seedPrMeta = (mount: FakeMountClient, repo: string, n: number, payload: Record) => { mount.files.set(`/github/repos/${repo}/pulls/${n}/metadata.json`, { - content: { number: n, head_ref: `ar-${n}-fix`, url: `https://github.com/${repo}/pull/${n}`, ...payload }, + content: { state: 'open', draft: false, labels: [], number: n, head_ref: `ar-${n}-fix`, url: `https://github.com/${repo}/pull/${n}`, ...payload }, }) } @@ -28630,6 +28643,7 @@ describe('FactoryLoop PR babysitter', () => { for (const repo of arrivalOrder) { const path = `/github/repos/${repo}/pulls/${number}/metadata.json` mount.files.set(path, { content: { + labels: [], number, state: 'open', head_ref: `factory/${number}`, @@ -28744,6 +28758,157 @@ describe('FactoryLoop PR babysitter', () => { } }) + it.each(['absent', 'unreadable', 'malformed'])('defers factory-created activation when PR metadata is %s and retries after recovery', async (mode) => { + const issue = realIssueFile(401, ready, { title: 'Real metadata recovery' }) + const mount = new FakeMountClient({ [issuePath(401)]: issue }) + const path = '/github/repos/AgentWorkforce/pear/pulls/401/metadata.json' + if (mode !== 'absent') mount.files.set(path, { content: mode === 'malformed' ? {} : { + number: 401, state: 'open', draft: false, labels: [], + } }) + const read = mount.readFile.bind(mount) + let unavailable = true + vi.spyOn(mount, 'readFile').mockImplementation(async (candidate) => { + if (mode === 'unreadable' && unavailable && candidate === path) throw new Error('mount unavailable') + return read(candidate) + }) + const fleet = new FakeFleetClient() + const stateStore = new InMemoryStateStore({ batchSize: 2 }) + const claim = vi.spyOn(stateStore, 'markRunning') + const factory = createFactory(babysitterConfig(), { + mount, fleet, stateStore, triage: new StaticTriage(), + probePrResolver: async () => ({ repo: 'AgentWorkforce/pear', prNumber: 401 }), + }) + try { + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(401), issue))) + fleet.emitAgentExit('ar-401-impl-pear', 'worker_exited') + await vi.waitFor(() => expect((factory.status().counters.babysitterActivationDeferred ?? 0) + + fleet.spawns.filter((spawn) => spawn.name.includes('babysit')).length).toBeGreaterThan(0)) + expect(fleet.spawns.filter((spawn) => spawn.name.includes('babysit'))).toEqual([]) + expect(claim).not.toHaveBeenCalled() + unavailable = false + seedPrMeta(mount, 'AgentWorkforce/pear', 401, { state: 'open', draft: false, labels: [] }) + fleet.emitAgentExit('ar-401-impl-pear', 'worker_exited') + await vi.waitFor(() => expect(fleet.spawns.filter((spawn) => spawn.name.includes('babysit'))).toHaveLength(1)) + } finally { + await factory.stop() + } + }) + + it.each(['roster', 'spawn-preflight'])('releases the factory-created claim after a %s failure and retries', async (mode) => { + const issue = realIssueFile(401, ready, { title: 'Real pre-placement retry' }) + const mount = new FakeMountClient({ [issuePath(401)]: issue }) + seedPrMeta(mount, 'AgentWorkforce/pear', 401, { state: 'open', draft: false, labels: [] }) + const fleet = new FakeFleetClient() + const stateStore = new InMemoryStateStore({ batchSize: 2 }) + const mark = stateStore.markRunning.bind(stateStore) + const roster = vi.spyOn(fleet, 'roster') + vi.spyOn(stateStore, 'markRunning').mockImplementationOnce(async (...args) => { + const claim = await mark(...args) + if (mode === 'roster') roster.mockRejectedValueOnce(new Error('temporary roster failure')) + else vi.spyOn(fleet, 'spawn').mockRejectedValueOnce(new FleetSpawnNotCreatedError(new Error('temporary spawn preflight failure'))) + return claim + }) + const factory = createFactory(babysitterConfig(), { + mount, fleet, stateStore, triage: new StaticTriage(), + probePrResolver: async () => ({ repo: 'AgentWorkforce/pear', prNumber: 401 }), + }) + try { + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(401), issue))) + fleet.emitAgentExit('ar-401-impl-pear', 'worker_exited') + await vi.waitFor(() => expect(factory.status().counters.babysitterSpawnFailures).toBe(1)) + expect(fleet.spawns.filter((spawn) => spawn.name.includes('babysit'))).toEqual([]) + expect(await stateStore.getBabysitterGeneration('factory-test', 'factory-created:agentworkforce/pear#401')).toBeUndefined() + fleet.emitAgentExit('ar-401-impl-pear', 'worker_exited') + await vi.waitFor(() => expect(fleet.spawns.filter((spawn) => spawn.name.includes('babysit'))).toHaveLength(1)) + expect(await stateStore.getBabysitterGeneration('factory-test', 'factory-created:agentworkforce/pear#401')).toBeDefined() + } finally { + await factory.stop() + } + }) + + it.each(['uncertain-spawn', 'session-persist', 'injection', 'release-error', 'release-denied'])('retains and identifies the factory-created claim after %s failure', async (mode) => { + const issue = realIssueFile(401, ready, { title: 'Real uncertain placement' }) + const mount = new FakeMountClient({ [issuePath(401)]: issue }) + seedPrMeta(mount, 'AgentWorkforce/pear', 401, { state: 'open', draft: false, labels: [] }) + const fleet = mode === 'injection' ? new RemoteLifecycleFleetClient() : new FakeFleetClient() + const stateStore = new InMemoryStateStore({ batchSize: 2 }) + const spawn = fleet.spawn.bind(fleet) + const calls = vi.spyOn(fleet, 'spawn').mockImplementation(async (input) => { + if (mode.startsWith('release-') && input.name.includes('babysit')) throw new FleetSpawnNotCreatedError(new Error('spawn preflight failed')) + if (mode === 'uncertain-spawn' && input.name.includes('babysit')) throw new Error('placement acknowledgement lost') + return spawn(input) + }) + if (mode === 'session-persist') vi.spyOn(stateStore, 'setBabysitterSession').mockRejectedValueOnce(new Error('session storage unavailable')) + if (mode === 'release-error') vi.spyOn(stateStore, 'clearBabysitterGeneration').mockRejectedValue(new Error('claim storage unavailable')) + if (mode === 'release-denied') vi.spyOn(stateStore, 'clearBabysitterGeneration').mockResolvedValue(false) + const warn = vi.fn() + const factory = createFactory(babysitterConfig(), { + mount, fleet, stateStore, logger: { warn }, triage: new StaticTriage(), + probePrResolver: async () => ({ repo: 'AgentWorkforce/pear', prNumber: 401 }), + }) + try { + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(401), issue))) + if (mode === 'injection') vi.spyOn(fleet, 'waitForInjected').mockRejectedValueOnce(new Error('transport acknowledgement unavailable')) + fleet.emitAgentExit('ar-401-impl-pear', 'worker_exited') + await vi.waitFor(() => expect(factory.status().counters.babysitterSpawnFailures).toBe(1)) + expect(await stateStore.getBabysitterGeneration('factory-test', 'factory-created:agentworkforce/pear#401')).toBeDefined() + expect(warn).toHaveBeenCalledWith('[factory] babysitter activation claim retained after failure', expect.objectContaining({ + repo: 'AgentWorkforce/pear', prNumber: 401, + ownershipKey: 'factory-created:agentworkforce/pear#401', + generationId: expect.any(String), + })) + fleet.emitAgentExit('ar-401-impl-pear', 'worker_exited') + await vi.waitFor(async () => { + if (mode === 'uncertain-spawn' || mode.startsWith('release-')) expect(factory.status().counters.babysitterActivationClaimRejected).toBeGreaterThan(0) + else expect(await stateStore.listBabysitterSessions('factory-test')).toHaveLength(1) + }) + expect(calls.mock.calls.filter(([input]) => input.name.includes('babysit'))).toHaveLength(1) + } finally { + await factory.stop() + } + }) + + it('does not replay an uncertain babysitter plan during durable dispatch recovery', async () => { + const root = await mkdtemp(join(tmpdir(), 'babysitter-uncertain-recovery-')) + const issue = realIssueFile(401, ready, { title: 'Real uncertain dispatch recovery' }) + const mount = new FakeMountClient({ [issuePath(401)]: issue }) + seedPrMeta(mount, 'AgentWorkforce/pear', 401, { state: 'open', draft: false, labels: [] }) + class DurableFleet extends FakeFleetClient { readonly durableOwnership = true } + const firstFleet = new DurableFleet() + const realSpawn = firstFleet.spawn.bind(firstFleet) + vi.spyOn(firstFleet, 'spawn').mockImplementation(async (input) => { + if (input.name.includes('babysit')) throw new Error('placement acknowledgement lost') + return realSpawn(input) + }) + const state = () => new FileStateStore({ batchSize: 2, watchStatePath: join(root, 'state.json') }) + const first = createFactory(babysitterConfig(), { + mount, fleet: firstFleet, stateStore: state(), triage: new StaticTriage(), + probePrResolver: async () => ({ repo: 'AgentWorkforce/pear', prNumber: 401 }), + }) + const nextFleet = new DurableFleet() + // Hydration restores observation metadata; it cannot prove a live worker. + vi.spyOn(nextFleet, 'hydrateTracked').mockImplementation(() => {}) + const next = createFactory(babysitterConfig(), { + mount, fleet: nextFleet, stateStore: state(), triage: new StaticTriage(), dispatchLifecycleRetryMs: 10, + probePrResolver: async () => ({ repo: 'AgentWorkforce/pear', prNumber: 401 }), + }) + try { + await first.dispatch(await first.triageIssue(parseLinearIssue(issuePath(401), issue))) + firstFleet.emitAgentExit('ar-401-impl-pear', 'worker_exited') + await vi.waitFor(() => expect(first.status().counters.babysitterSpawnFailures).toBe(1)) + expect((await state().listDispatchLifecycles('factory-test'))[0]?.[1].phase).toBe('dispatching') + await first.stop() + await next.start({ mode: 'dispatch-owner' }) + await vi.waitFor(() => expect((next.status().counters.babysitterActivationClaimRejected ?? 0) + + nextFleet.spawns.filter((input) => input.name.includes('babysit')).length).toBeGreaterThan(0)) + expect(nextFleet.spawns.filter((input) => input.name.includes('babysit'))).toEqual([]) + } finally { + await first.stop() + await next.stop() + await rm(root, { recursive: true, force: true }) + } + }) + it.each(['denied', 'unavailable', 'legacy-session'])('fails closed when the PR claim is %s', async (result) => { const issue = realIssueFile(401, ready, { title: 'Real babysitter durable claim' }) const mount = new FakeMountClient({ [issuePath(401)]: issue }) @@ -28836,6 +29001,7 @@ describe('FactoryLoop PR babysitter', () => { const issue = realIssueFile(401, ready, { title: 'Real babysitter spawn' }) const mount = new FakeMountClient({ [issuePath(401)]: issue }) const fleet = new FakeFleetClient() + seedPrMeta(mount, 'AgentWorkforce/pear', 401, { state: 'open', draft: false, labels: [] }) const factory = createFactory(babysitterConfig(), { mount, fleet, @@ -28879,6 +29045,7 @@ describe('FactoryLoop PR babysitter', () => { const issue = realIssueFile(402, ready, { title: 'Real remote preview babysitter handoff' }) const mount = new FakeMountClient({ [issuePath(402)]: issue }) const fleet = new RemotePreviewFleetClient() + seedPrMeta(mount, 'AgentWorkforce/pear', 402, { state: 'open', draft: false, labels: [] }) const factory = createFactory(babysitterConfig({ preview: { provider: 'tailscale-serve', @@ -28913,6 +29080,7 @@ describe('FactoryLoop PR babysitter', () => { fleet.setSessionRef('ar-404-impl-pear', 'session-ar-404-impl-pear') const slack = new RecordingSlack() const stateStore = new InMemoryStateStore({ batchSize: 10 }) + seedPrMeta(mount, 'AgentWorkforce/pear', 404, { state: 'open', draft: false, labels: [] }) const factory = createFactory(babysitterConfig({ slack: slackConfig() }), { mount, fleet, @@ -28959,6 +29127,7 @@ describe('FactoryLoop PR babysitter', () => { const fleet = new FakeFleetClient() const slack = new RecordingSlack() const stateStore = new InMemoryStateStore({ batchSize: 10 }) + seedPrMeta(mount, 'AgentWorkforce/pear', 405, { state: 'open', draft: false, labels: [] }) const factory = createFactory(babysitterConfig({ slack: slackConfig() }), { mount, fleet, @@ -29015,6 +29184,7 @@ describe('FactoryLoop PR babysitter', () => { const firstFleet = new RemoteLifecycleFleetClient() firstFleet.setSessionRef('ar-406-impl-pear', 'session-ar-406-impl-pear') firstFleet.setSessionRef('ar-406-babysit', 'session-ar-406-babysit') + seedPrMeta(mount, 'AgentWorkforce/pear', 406, { state: 'open', draft: false, labels: [] }) const first = createFactory(factoryConfig, { mount, fleet: firstFleet, @@ -29847,6 +30017,7 @@ describe('FactoryLoop PR babysitter', () => { const mount = new FakeMountClient({ [path]: issueFile, '/github/repos/AgentWorkforce__pear/pulls/by-id/5.json': prFile(5, { + labels: [], title: 'Unrelated observability work', body: 'tsc, eslint, and 52 tests all pass.', head_ref: 'claude/unrelated-observability', @@ -29880,6 +30051,7 @@ describe('FactoryLoop PR babysitter', () => { const issue = realIssueFile(403, ready, { title: 'Real babysitter idempotent' }) const mount = new FakeMountClient({ [issuePath(403)]: issue }) const fleet = new FakeFleetClient() + seedPrMeta(mount, 'AgentWorkforce/pear', 403, { state: 'open', draft: false, labels: [] }) const factory = createFactory(babysitterConfig(), { mount, fleet, @@ -29911,7 +30083,7 @@ describe('FactoryLoop PR babysitter', () => { try { await factory.start({ mode: 'live', liveSubscription: { transport: 'subscribe' } }) await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(number), issue))) - mount.files.set(prPath, { content: { number, state: 'open', draft: false, head_ref: `ar-${number}-fix` } }) + mount.files.set(prPath, { content: { labels: [], number, state: 'open', draft: false, head_ref: `ar-${number}-fix` } }) mount.emit(changeEvent(prPath, 'durable-pr-open')) await flush() await vi.waitFor(() => expect(fleet.spawns.map((spawn) => spawn.name)).toContain(`ar-${number}-babysit`)) @@ -29969,7 +30141,7 @@ describe('FactoryLoop PR babysitter', () => { // A duplicate PR observation renews the same server identity rather than // making another subscription record. There is no claim, so no wake. const renamedPrPath = `/github/repos/AgentWorkforce/pear/pulls/${number}__renamed-after-review/metadata.json` - mount.files.set(renamedPrPath, { content: { number, state: 'open', draft: false, head_ref: `ar-${number}-fix` } }) + mount.files.set(renamedPrPath, { content: { labels: [], number, state: 'open', draft: false, head_ref: `ar-${number}-fix` } }) mount.emit(changeEvent(renamedPrPath, 'durable-pr-repeat')) await vi.waitFor(() => expect(subscriptions.createCalls).toHaveLength(2)) expect(subscriptions.records).toHaveLength(1) @@ -30034,7 +30206,7 @@ describe('FactoryLoop PR babysitter', () => { try { await factory.start({ mode: 'live', liveSubscription: { transport: 'subscribe' } }) await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(number), issue))) - mount.files.set(prPath, { content: { number, state: 'open', draft: false, head_ref: `ar-${number}-fix` } }) + mount.files.set(prPath, { content: { labels: [], number, state: 'open', draft: false, head_ref: `ar-${number}-fix` } }) mount.emit(changeEvent(prPath, 'fallback-pr-open')) await vi.waitFor(() => expect(fleet.spawns.map((spawn) => spawn.name)).toContain(`ar-${number}-babysit`)) @@ -30126,7 +30298,7 @@ describe('FactoryLoop PR babysitter', () => { const subscription = subscriptions.records[0]! subscriptions.claims = [subscriptions.claimFor(subscription, 'pull_request.closed', 'delivery-close')] - mount.files.set(prPath, { content: { number, state: 'closed', draft: false, head_ref: `ar-${number}-fix` } }) + mount.files.set(prPath, { content: { labels: [], number, state: 'closed', draft: false, head_ref: `ar-${number}-fix` } }) mount.emit(changeEvent(prPath, 'terminal-close')) await vi.waitFor(() => expect(subscriptions.accepted.map((entry) => entry.deliveryId)).toEqual(['delivery-close'])) @@ -30176,7 +30348,7 @@ describe('FactoryLoop PR babysitter', () => { try { await first.start({ mode: 'live', liveSubscription: { transport: 'subscribe' } }) await first.dispatch(await first.triageIssue(parseLinearIssue(issuePath(number), issue))) - mount.files.set(prPath, { content: { number, state: 'open', draft: false, head_ref: `ar-${number}-fix` } }) + mount.files.set(prPath, { content: { labels: [], number, state: 'open', draft: false, head_ref: `ar-${number}-fix` } }) mount.emit(changeEvent(prPath, 'acceptance-restart-open')) await flush() await vi.waitFor(() => expect(subscriptions.records).toHaveLength(1)) @@ -30265,7 +30437,7 @@ describe('FactoryLoop PR babysitter', () => { try { await factory.start({ mode: 'live', liveSubscription: { transport: 'subscribe' } }) await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(number), issue))) - mount.files.set(prPath, { content: { number, state: 'open', draft: false, head_ref: `ar-${number}-fix` } }) + mount.files.set(prPath, { content: { labels: [], number, state: 'open', draft: false, head_ref: `ar-${number}-fix` } }) mount.emit(changeEvent(prPath, 'transient-pr-open')) await vi.waitFor(() => expect(subscriptions.records).toHaveLength(1)) const subscription = subscriptions.records[0]! @@ -30810,7 +30982,7 @@ describe('FactoryLoop PR babysitter', () => { const prPath = '/github/repos/AgentWorkforce/pear/pulls/404/metadata.json' mount.files.set(prPath, { - content: { number: 404, state: 'open', head_ref: 'ar-404-fix', isDraft: false, url: 'https://github.com/AgentWorkforce/pear/pull/404' }, + content: { labels: [], number: 404, state: 'open', head_ref: 'ar-404-fix', isDraft: false, url: 'https://github.com/AgentWorkforce/pear/pull/404' }, }) mount.emit(changeEvent(prPath, 'pr-404-open')) @@ -30850,7 +31022,7 @@ describe('FactoryLoop PR babysitter', () => { await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(2404), issue))) mount.files.set(prPath, { - content: { number: 2404, state: 'open', head_ref: 'ar-2404-fix', isDraft: false, url: 'https://github.com/AgentWorkforce/pear/pull/2404' }, + content: { labels: [], number: 2404, state: 'open', head_ref: 'ar-2404-fix', isDraft: false, url: 'https://github.com/AgentWorkforce/pear/pull/2404' }, }) mount.emit(changeEvent(prPath, 'pr-2404-open')) @@ -30898,7 +31070,7 @@ describe('FactoryLoop PR babysitter', () => { await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(2405), issue))) mount.files.set(prPath, { - content: { number: 2405, state: 'open', head_ref: 'ar-2405-fix', isDraft: false, url: 'https://github.com/AgentWorkforce/pear/pull/2405' }, + content: { labels: [], number: 2405, state: 'open', head_ref: 'ar-2405-fix', isDraft: false, url: 'https://github.com/AgentWorkforce/pear/pull/2405' }, }) mount.emit(changeEvent(prPath, 'pr-2405-open')) @@ -31006,6 +31178,7 @@ describe('FactoryLoop PR babysitter', () => { const ownPr = '/github/repos/AgentWorkforce/pear/pulls/420/metadata.json' mount.files.set(ownPr, { content: { + labels: [], number: 420, state: 'open', head_ref: 'ar-420-fix', @@ -31069,7 +31242,7 @@ describe('FactoryLoop PR babysitter', () => { // same-number PR in another repo can never replace exact ownership. const otherPr = '/github/repos/AgentWorkforce/pear/pulls/421/metadata.json' mount.files.set(otherPr, { - content: { number: 421, state: 'open', head_ref: 'unrelated', body: 'Fixes AR-420' }, + content: { labels: [], number: 421, state: 'open', head_ref: 'unrelated', body: 'Fixes AR-420' }, }) mount.emit(changeEvent(otherPr, 'pr-421-malicious-reference')) mount.files.set('/github/repos/AgentWorkforce/hoopsheet/comments/9999.json', { content: { @@ -31202,6 +31375,7 @@ describe('FactoryLoop PR babysitter', () => { }) }) const prPath = '/github/repos/AgentWorkforce/pear/pulls/436/metadata.json' mount.files.set(prPath, { content: { + labels: [], number: 436, state: 'open', head_ref: 'ar-436-fix', @@ -31308,6 +31482,7 @@ describe('FactoryLoop PR babysitter', () => { objectType: 'pull_request', objectId: '435', payload: { + labels: [], number: 435, state: 'open', draft: false, @@ -31340,6 +31515,7 @@ describe('FactoryLoop PR babysitter', () => { objectType: 'pull_request', objectId: '435', payload: { + labels: [], number: 435, state: 'open', draft: false, @@ -31402,6 +31578,7 @@ describe('FactoryLoop PR babysitter', () => { await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(421), issue))) const prPath = '/github/repos/AgentWorkforce/pear/pulls/421/metadata.json' mount.files.set(prPath, { content: { + labels: [], number: 421, state: 'open', head_ref: 'ar-421-fix', @@ -31463,6 +31640,7 @@ describe('FactoryLoop PR babysitter', () => { await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(424), issue))) const prPath = '/github/repos/AgentWorkforce/pear/pulls/424/metadata.json' mount.files.set(prPath, { content: { + labels: [], number: 424, state: 'open', head_ref: 'ar-424-fix', @@ -31530,6 +31708,7 @@ describe('FactoryLoop PR babysitter', () => { await factory.dispatch(await factory.triageIssue(parsedIssue)) const prPath = '/github/repos/AgentWorkforce/pear/pulls/426/metadata.json' mount.files.set(prPath, { content: { + labels: [], number: 426, state: 'open', head_ref: 'ar-426-fix', @@ -31622,6 +31801,7 @@ describe('FactoryLoop PR babysitter', () => { await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(425), issue))) const prPath = '/github/repos/AgentWorkforce/pear/pulls/425/metadata.json' mount.files.set(prPath, { content: { + labels: [], number: 425, state: 'open', head_ref: 'ar-425-fix', @@ -31702,6 +31882,7 @@ describe('FactoryLoop PR babysitter', () => { await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(423), issue))) const prPath = '/github/repos/AgentWorkforce/pear/pulls/423/metadata.json' mount.files.set(prPath, { content: { + labels: [], number: 423, state: 'open', head_ref: 'ar-423-fix', @@ -31759,7 +31940,7 @@ describe('FactoryLoop PR babysitter', () => { await factory.start({ mode: 'live', liveSubscription: { transport: 'subscribe' } }) try { await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(422), issue))) - mount.files.set(prPath, { content: { number: 422, state: 'open', head_ref: 'ar-422-fix', draft: false } }) + mount.files.set(prPath, { content: { labels: [], number: 422, state: 'open', head_ref: 'ar-422-fix', draft: false } }) mount.emit(changeEvent(prPath, 'pr-422-open')) await vi.waitFor(() => expect(fleet.spawns.map((spawn) => spawn.name)).toContain('ar-422-babysit')) const inputsBefore = fleet.inputs.length @@ -31816,7 +31997,7 @@ describe('FactoryLoop PR babysitter', () => { try { await factory.runOnce() mount.files.set(prPath, { - content: { number, state: 'open', head_ref: `factory/${number}`, draft: false }, + content: { labels: [], number, state: 'open', head_ref: `factory/${number}`, draft: false }, }) mount.emit(changeEvent(prPath, 'github-pr-critical-open')) await vi.waitFor(() => expect(fleet.spawns.map((spawn) => spawn.name)).toContain('ar-31-babysit-pear')) @@ -31871,7 +32052,7 @@ describe('FactoryLoop PR babysitter', () => { await factory.start({ mode: 'live', liveSubscription: { transport: 'subscribe' } }) try { await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(427), issue))) - mount.files.set(prPath, { content: { number: 427, state: 'open', head_ref: 'ar-427-fix', draft: false } }) + mount.files.set(prPath, { content: { labels: [], number: 427, state: 'open', head_ref: 'ar-427-fix', draft: false } }) mount.emit(changeEvent(prPath, 'pr-427-open')) await vi.waitFor(() => expect(fleet.spawns.map((spawn) => spawn.name)).toContain('ar-427-babysit')) const inputsBefore = fleet.inputs.length @@ -31945,7 +32126,7 @@ describe('FactoryLoop PR babysitter', () => { await factory.start({ mode: 'live', liveSubscription: { transport: 'subscribe' } }) try { await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(426), issue))) - mount.files.set(prPath, { content: { number: 426, state: 'open', head_ref: 'ar-426-fix', draft: false } }) + mount.files.set(prPath, { content: { labels: [], number: 426, state: 'open', head_ref: 'ar-426-fix', draft: false } }) mount.emit(changeEvent(prPath, 'pr-426-open')) await vi.waitFor(() => expect(fleet.spawns.map((spawn) => spawn.name)).toContain('ar-426-babysit')) @@ -32016,7 +32197,7 @@ describe('FactoryLoop PR babysitter', () => { await factory.start({ mode: 'live', liveSubscription: { transport: 'subscribe' } }) try { await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(433), issue))) - mount.files.set(prPath, { content: { number: 433, state: 'open', head_ref: 'ar-433-fix', draft: false } }) + mount.files.set(prPath, { content: { labels: [], number: 433, state: 'open', head_ref: 'ar-433-fix', draft: false } }) mount.emit(changeEvent(prPath, 'pr-433-open')) await vi.waitFor(() => expect(fleet.spawns.map((spawn) => spawn.name)).toContain('ar-433-babysit')) mount.emit(changeEvent('/github/repos/AgentWorkforce/pear/pulls/433/comments/9903.json', 'comment-9903')) @@ -32084,7 +32265,7 @@ describe('FactoryLoop PR babysitter', () => { await factory.start({ mode: 'live', liveSubscription: { transport: 'subscribe' } }) try { await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(428), issue))) - mount.files.set(prPath, { content: { number: 428, state: 'open', head_ref: 'ar-428-fix', draft: false } }) + mount.files.set(prPath, { content: { labels: [], number: 428, state: 'open', head_ref: 'ar-428-fix', draft: false } }) mount.emit(changeEvent(prPath, 'pr-428-open')) await vi.waitFor(() => expect(fleet.spawns.map((spawn) => spawn.name)).toContain('ar-428-babysit')) const inputsBefore = fleet.inputs.length @@ -32129,6 +32310,7 @@ describe('FactoryLoop PR babysitter', () => { const commentPath = '/github/repos/AgentWorkforce/pear/comments/9301.json' const mount = new FakeMountClient({ [issuePath(423)]: issue }) const firstFleet = new FakeFleetClient() + seedPrMeta(mount, 'AgentWorkforce/pear', 423, { state: 'open', draft: false }) const first = createFactory(babysitterConfig(), { mount, fleet: firstFleet, @@ -32160,7 +32342,7 @@ describe('FactoryLoop PR babysitter', () => { mount.emit(changeEvent(commentPath, 'pre-restart-comment')) await vi.waitFor(() => expect(first.status().counters.babysitterEventsQueued).toBe(1)) expect(firstFleet.messages.filter((message) => message.text.startsWith(' { restartedFleet.messages.filter((message) => message.text.startsWith(' expect( restartedFleet.messages.filter((message) => message.text.startsWith(' { const mount = new FakeMountClient({ [issue.path]: githubIssueFile(52, { labels: ['factory', 'pear', 'factory:in-progress'] }), '/github/repos/AgentWorkforce__pear/pulls/by-id/5.json': prFile(5, { + labels: [], title: 'Unrelated observability work', body: 'tsc, eslint, and 52 tests all pass.', head_ref: 'claude/unrelated-observability', @@ -32318,7 +32501,7 @@ describe('FactoryLoop PR babysitter', () => { await factory.start({ mode: 'live', liveSubscription }) try { await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(424), issue))) - mount.files.set(prPath, { content: { number: 424, state: 'open', head_ref: 'ar-424-fix', draft: false } }) + mount.files.set(prPath, { content: { labels: [], number: 424, state: 'open', head_ref: 'ar-424-fix', draft: false } }) mount.emit(changeEvent(prPath, 'poll-pr-424-open')) await vi.waitFor(() => expect(fleet.spawns.map((spawn) => spawn.name)).toContain('ar-424-babysit')) @@ -32377,7 +32560,7 @@ describe('FactoryLoop PR babysitter', () => { await factory.start({ mode: 'live', liveSubscription: { transport: 'subscribe' } }) try { await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(425), issue))) - mount.files.set(prPath, { content: { number: 425, state: 'open', head_ref: 'ar-425-fix', draft: false } }) + mount.files.set(prPath, { content: { labels: [], number: 425, state: 'open', head_ref: 'ar-425-fix', draft: false } }) mount.emit(changeEvent(prPath, 'pr-425-open')) await vi.waitFor(() => expect(fleet.spawns.map((spawn) => spawn.name)).toContain('ar-425-babysit')) mount.emit(changeEvent('/github/repos/AgentWorkforce/pear/pulls/425/comments/9501.json', 'comment-9501')) @@ -32421,7 +32604,7 @@ describe('FactoryLoop PR babysitter', () => { await factory.start({ mode: 'live', liveSubscription: { transport: 'subscribe' } }) try { await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(429), issue))) - mount.files.set(prPath, { content: { number: 429, state: 'open', head_ref: 'ar-429-fix', draft: false } }) + mount.files.set(prPath, { content: { labels: [], number: 429, state: 'open', head_ref: 'ar-429-fix', draft: false } }) mount.emit(changeEvent(prPath, 'pr-429-open')) await vi.waitFor(() => expect(fleet.spawns.map((spawn) => spawn.name)).toContain('ar-429-babysit')) @@ -32449,14 +32632,14 @@ describe('FactoryLoop PR babysitter', () => { await factory.start({ mode: 'live', liveSubscription: { transport: 'subscribe' } }) try { await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(427), issue))) - mount.files.set(prPath, { content: { number: 427, state: 'open', head_ref: 'ar-427-fix', draft: false } }) + mount.files.set(prPath, { content: { labels: [], number: 427, state: 'open', head_ref: 'ar-427-fix', draft: false } }) mount.emit(changeEvent(prPath, 'pr-427-open')) await vi.waitFor(() => expect(fleet.spawns.map((spawn) => spawn.name)).toContain('ar-427-babysit')) const inputsBefore = fleet.inputs.length mount.emit(changeEvent('/github/repos/AgentWorkforce/pear/pulls/427/comments/9701.json', 'comment-9701')) await vi.waitFor(() => expect(factory.status().counters.babysitterEventsQueued).toBe(1)) - mount.files.set(prPath, { content: { number: 427, state: 'closed', merged: false, head_ref: 'renamed' } }) + mount.files.set(prPath, { content: { labels: [], number: 427, state: 'closed', merged: false, head_ref: 'renamed' } }) mount.emit(changeEvent(prPath, 'pr-427-closed')) await new Promise((resolve) => setTimeout(resolve, 900)) @@ -32492,14 +32675,14 @@ describe('FactoryLoop PR babysitter', () => { await factory.start({ mode: 'live', liveSubscription: { transport: 'subscribe' } }) try { await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(432), issue))) - mount.files.set(prPath, { content: { number: 432, state: 'open', head_ref: 'ar-432-fix', draft: false } }) + mount.files.set(prPath, { content: { labels: [], number: 432, state: 'open', head_ref: 'ar-432-fix', draft: false } }) mount.emit(changeEvent(prPath, 'pr-432-open')) await vi.waitFor(() => expect(fleet.spawns.map((spawn) => spawn.name)).toContain('ar-432-babysit')) const inputsBefore = fleet.inputs.length mount.emit(changeEvent('/github/repos/AgentWorkforce/pear/pulls/432/comments/9902.json', 'comment-9902')) await vi.waitFor(() => expect(fleet.pendingWake).toBeDefined(), { timeout: 3_000 }) - mount.files.set(prPath, { content: { number: 432, state: 'closed', merged: false, head_ref: 'renamed' } }) + mount.files.set(prPath, { content: { labels: [], number: 432, state: 'closed', merged: false, head_ref: 'renamed' } }) mount.emit(changeEvent(prPath, 'pr-432-closed')) await vi.waitFor(async () => expect(await stateStore.listBabysitterSessions('factory-test')).toEqual([])) @@ -32535,6 +32718,7 @@ describe('FactoryLoop PR babysitter', () => { const prPath = '/github/repos/AgentWorkforce/pear/pulls/408/metadata.json' mount.files.set(prPath, { content: { + labels: [], number: 408, state: 'open', head_ref: 'feature/fix-ci', @@ -32566,7 +32750,7 @@ describe('FactoryLoop PR babysitter', () => { await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(430), issue))) const prPath = '/github/repos/AgentWorkforce/pear/pulls/430/metadata.json' mount.files.set(prPath, { - content: { number: 431, state: 'open', head_ref: 'ar-430-fix', draft: false }, + content: { labels: [], number: 431, state: 'open', head_ref: 'ar-430-fix', draft: false }, }) mount.emit(changeEvent(prPath, 'pr-number-mismatch')) @@ -32595,7 +32779,7 @@ describe('FactoryLoop PR babysitter', () => { // Flat __/pulls/by-id/.json shape (githubPullRoot layout). const prPath = '/github/repos/AgentWorkforce__pear/pulls/by-id/409.json' mount.files.set(prPath, { - content: { number: 409, state: 'open', head_ref: 'ar-409-fix', isDraft: false }, + content: { labels: [], number: 409, state: 'open', head_ref: 'ar-409-fix', isDraft: false }, }) mount.emit(changeEvent(prPath, 'pr-409-open')) @@ -32618,6 +32802,7 @@ describe('FactoryLoop PR babysitter', () => { [issuePath(412)]: related, [issuePath(413)]: dependent, [prPath]: prFile(410, { + labels: [], title: 'Real merged after review', body: 'Linear: AR-412', head_ref: 'ar-410-fix', @@ -32640,6 +32825,7 @@ describe('FactoryLoop PR babysitter', () => { ])) mount.files.set(prPath, { content: prFile(410, { + labels: [], title: 'Real merged after review', body: 'Linear: AR-412', head_ref: 'ar-410-fix', @@ -32681,6 +32867,7 @@ describe('FactoryLoop PR babysitter', () => { labels: ['factory', 'factory:human-review'], }), [prPath]: prFile(414, { + labels: [], head_ref: 'issue-414-fix', state: 'MERGED', merged: true, @@ -32722,6 +32909,7 @@ describe('FactoryLoop PR babysitter', () => { labels: ['factory', 'factory:human-review'], }), [cloudPrPath]: prFile(2891, { + labels: [], title: 'fix(relayauth): dormant emergency source mint gate', // The URL-form of a cloud#139 reference reproduces the observed // false positive: `containsExplicitIssueReference` matched any @@ -32776,6 +32964,7 @@ describe('FactoryLoop PR babysitter', () => { labels: ['factory', 'factory:human-review'], }), [factoryPrPath]: prFile(250, { + labels: [], title: 'fix: address factory#222', body: 'Fixes #222', head_ref: 'factory/222-fix', @@ -32834,6 +33023,7 @@ describe('FactoryLoop PR babysitter', () => { labels: ['factory', 'factory:human-review'], }), [prPath]: prFile(172, { + labels: [], title: 'chore: connection diagnostic', body: 'Diagnostic for #155. No credential behaviour changes — nothing about token minting,' + ' refresh, or connection selection moves here.', @@ -32891,6 +33081,7 @@ describe('FactoryLoop PR babysitter', () => { labels: ['factory', 'factory:human-review'], }), [prPath]: prFile(174, { + labels: [], title: 'chore: unrelated cleanup', body: 'Background context lives in #155.', head_ref: 'chore/unrelated-cleanup', @@ -32936,6 +33127,7 @@ describe('FactoryLoop PR babysitter', () => { labels: ['factory', 'incident', 'factory:human-review'], }), [prPath]: prFile(173, { + labels: [], title: 'fix: repoint installation', body: 'Fixes #155', head_ref: 'factory/155-repoint', @@ -32976,6 +33168,7 @@ describe('FactoryLoop PR babysitter', () => { const mount = new FakeMountClient({ [issuePath(411)]: issue, [prPath]: prFile(411, { + labels: [], title: 'Real merged before ready', body: 'Linear: AR-411', head_ref: 'ar-411-fix', @@ -33015,6 +33208,7 @@ describe('FactoryLoop PR babysitter', () => { const mount = new FakeMountClient({ [issuePath(415)]: issue, [prPath]: prFile(415, { + labels: [], title: 'Real merged but disclaimed', body: 'Linear: AR-415 — this does not fix AR-415, it is groundwork only.', head_ref: 'ar-415-fix', @@ -33055,6 +33249,7 @@ describe('FactoryLoop PR babysitter', () => { const mount = new FakeMountClient({ [issuePath(416)]: issue, [prPath]: prFile(416, { + labels: [], title: 'Real incident in flight', body: 'Linear: AR-416', head_ref: 'ar-416-fix', diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index a06568e8..1c190291 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -81,6 +81,7 @@ import type { TerminalDispatchLifecyclePhase, WaitingClarification, } from '../ports/state' +import { FleetSpawnNotCreatedError } from '../ports/fleet' import type { Clock, Logger } from '../ports/system' import type { AgentWorktree, AgentWorktreeManager, AgentWorktreeRepository } from '../ports/worktree' import { factoryWorktreeIssueSlug, factoryWorktreePath } from '../git/agent-worktree' @@ -9029,6 +9030,18 @@ export class FactoryLoop implements Factory { specs.push(tracked.spec) } for (const spec of specs) { + if (spec.role === 'babysitter' && spec.ownedPullRequest) { + // A persisted plan is not placement evidence. Recovery must pass the + // same PR snapshot and generation gates as the original handoff. + const owned = spec.ownedPullRequest + // A live roster entry can close the lost-ack gap without placement. + // A missing entry cannot authorize replay of the persisted intent. + await this.#spawnAgent(record, spec, record.dryRun, undefined, true) + await this.#ensureBabysitter(record, { repo: owned.repo, prNumber: owned.number, path: owned.path }) + const tracked = record.agents.get(spec.name) + if (tracked?.result) agents.push({ name: tracked.result.name, role: spec.role }) + continue + } const spawned = await this.#spawnAgent(record, spec, record.dryRun) agents.push({ name: spawned.name, role: spec.role }) } @@ -11583,7 +11596,10 @@ export class FactoryLoop implements Factory { await writeJsonFileAtomically(path, registry) } - async #spawnAgent(record: InFlightIssue, spec: AgentSpec, dryRun: boolean): Promise<{ name: string }> { + async #spawnAgent(record: InFlightIssue, spec: AgentSpec, dryRun: boolean, + placement?: { status: 'not-created' | 'uncertain' | 'created' }, + adoptOnly = false, + ): Promise<{ name: string }> { const batch = await this.#batch() const invocationId = batch.invocationIdFor(record.issue, spec) const existing = record.agents.get(spec.name) @@ -11591,11 +11607,13 @@ export class FactoryLoop implements Factory { // Answering with its old spawn result here would report a synthetic success // for a process that no longer exists. if (existing?.result && existing.releasedAtMs === undefined) { + if (placement) placement.status = 'created' this.#scheduleHeldAgentDeadline(record) return { name: existing.result?.name ?? spec.name } } if (!batch.shouldSpawn(record, invocationId)) { + if (placement) placement.status = 'created' return { name: spec.name } } @@ -11612,6 +11630,7 @@ export class FactoryLoop implements Factory { } const rosterAgent = roster.agents.find((agent) => agent.name === spec.name) if (rosterAgent) { + if (placement) placement.status = 'created' if (this.#fleet.placementLocality === 'remote') { const host = rosterAgent.node ? roster.nodes.find((node) => @@ -11639,6 +11658,8 @@ export class FactoryLoop implements Factory { return { name: spec.name } } + if (adoptOnly) return { name: spec.name } + if (this.#fleet.placementLocality === 'remote') { const loads = new Map() for (const agent of roster.agents) { @@ -11658,6 +11679,9 @@ export class FactoryLoop implements Factory { await this.#prepareAgentWorktree(record, spec) let result try { + // Crossing the placement boundary can create a worker even if its ack + // never arrives. An ordinary rejection is not evidence of non-placement. + if (placement) placement.status = 'uncertain' result = await this.#fleet.spawn({ name: spec.name, capability: spec.capability, @@ -11675,6 +11699,7 @@ export class FactoryLoop implements Factory { channel: spec.channel, }) } catch (error) { + if (placement && error instanceof FleetSpawnNotCreatedError) placement.status = 'not-created' const wrapped = contextualError( `Dispatch spawn failed for ${record.issue.key}/${spec.name} (${spec.capability}) cwd=${spec.clonePath ?? 'default'}`, error, @@ -11685,6 +11710,7 @@ export class FactoryLoop implements Factory { : 'agent_spawn_failed' as const, }) } + if (placement) placement.status = 'created' // The never-placed deadline can fire while this spawn is in flight — that // is the whole point of arming it before the first await, and it makes a // late `spawn` result newly reachable (#303 review, cubic). By now the @@ -11946,6 +11972,13 @@ export class FactoryLoop implements Factory { if (tracingReconciledExit) this.#logger.info?.('[factory] reconciled agent exit question replay completed', { issue: record.issue.key, name }) const exiting = record.agents.get(name) + if (exiting?.spec.role === 'babysitter' && !exiting.result && exiting.spec.ownedPullRequest) { + // Roster reconciliation can report a planned name as missing. That is + // not an exited worker and must not enter the ordinary restart path. + const owned = exiting.spec.ownedPullRequest + await this.#ensureBabysitter(record, { repo: owned.repo, prNumber: owned.number, path: owned.path }) + return + } if (exiting) await this.#reportAgent(record, exiting, 'agent.exited', { releaseReason: reason }) if (tracingReconciledExit) this.#logger.info?.('[factory] reconciled agent exit telemetry completed', { issue: record.issue.key, name }) @@ -17407,7 +17440,7 @@ export class FactoryLoop implements Factory { } const wantedPr = githubPrIdentity(prRef.repo, prRef.prNumber) const trackedBabysitter = [...record.agents.entries()].find(([, agent]) => - agent.spec.role === 'babysitter' && + agent.spec.role === 'babysitter' && agent.result && agent.releasedAtMs === undefined && githubPrIdentity(agent.spec.ownedPullRequest?.repo ?? '', agent.spec.ownedPullRequest?.number ?? 0) === wantedPr) if (trackedBabysitter) { const [trackedName, tracked] = trackedBabysitter @@ -17436,11 +17469,21 @@ export class FactoryLoop implements Factory { const spawnFinished = new Promise((resolve) => { finishSpawn = resolve }) this.#babysitterSpawnInFlight.set(babysitterKey, spawnFinished) + const ownershipKey = `factory-created:${prIdentity}` + let generationId: string | undefined + let claimAttempted = false + const placement: { status: 'not-created' | 'uncertain' | 'created' } = { status: 'not-created' } try { - const snapshot = await this.#readPrSnapshot(prRef) - if (snapshot && (this.#babysitterActivationExcluded(prRef.repo, prRef.prNumber, snapshot.labels) || - prMetaShowsMerged(snapshot) || snapshot.draft || - (snapshot.state && snapshot.state.toUpperCase() !== 'OPEN'))) { + const snapshot = await this.#readPrSnapshot(prRef, { forActivation: true }) + if (!snapshot) { + this.#increment('babysitterActivationDeferred') + if (this.#usesDurableDispatchLifecycle()) this.#scheduleDispatchLifecycleRetry(record) + this.#logger.info?.('[factory] babysitter activation deferred until PR metadata is readable', { + repo: prRef.repo, prNumber: prRef.prNumber, + }) + } + if (!snapshot || this.#babysitterActivationExcluded(prRef.repo, prRef.prNumber, snapshot.labels) || + prMetaShowsMerged(snapshot) || snapshot.draft || snapshot.state?.toUpperCase() !== 'OPEN') { this.#babysitterSpawned.delete(babysitterKey) this.#babysitterPr.delete(babysitterKey) this.#babysitterIssueRefs.delete(babysitterKey) @@ -17528,9 +17571,9 @@ export class FactoryLoop implements Factory { }) // A PR is the work unit, regardless of how many issue records point at it. - // Persist before placement and never force/clear this claim on timeout, - // exit or failed acknowledgement: none proves that a worker was not - // created. Existing tracked placements recover above without re-dispatch. + // Persist before placement. Release only with positive evidence that no + // worker was created; timeout, exit and missing acknowledgements cannot + // establish that. Tracked placements recover above without re-dispatch. // An uncertain, untracked placement requires operator reconciliation. const priorSessions = await this.#state.listBabysitterSessions(this.#workspaceId) if (priorSessions.some(([, session]) => @@ -17541,8 +17584,9 @@ export class FactoryLoop implements Factory { this.#babysitterIssueRefs.delete(babysitterKey) return } + claimAttempted = true const claim = await this.#state.markRunning( - this.#workspaceId, `factory-created:${prIdentity}`, spec.name, + this.#workspaceId, ownershipKey, spec.name, this.#clock.now(), DISPATCH_LIFECYCLE_LEASE_MS, ) if (!claim) { @@ -17552,11 +17596,12 @@ export class FactoryLoop implements Factory { this.#babysitterIssueRefs.delete(babysitterKey) return } + generationId = claim.generationId const spawned = await this.#spawnAgent(record, { ...spec, task, ownedPullRequest: { repo: prRef.repo, number: prRef.prNumber, path: prRef.path }, - }, false) + }, false, placement) const tracked = record.agents.get(spawned.name) this.#babysitterPr.set(babysitterKey, { repo: prRef.repo, @@ -17612,13 +17657,36 @@ export class FactoryLoop implements Factory { await this.#state.recordCritical(this.#workspaceId, ack.eventId, { issue: record.issue, input }) } } catch (error) { - // Allow a later event to retry the spawn. + let claimReleased = false + if (generationId && placement.status === 'not-created') { + try { + // Compare-and-delete only this generation, never another owner's. + claimReleased = await this.#state.clearBabysitterGeneration(this.#workspaceId, ownershipKey, generationId) + } catch (releaseError) { + this.#logger.warn?.('[factory] babysitter activation claim release failed', { + repo: prRef.repo, prNumber: prRef.prNumber, ownershipKey, generationId, + error: describeError(releaseError).errorMessage, + }) + } + } + if (claimAttempted && !claimReleased) { + this.#logger.warn?.('[factory] babysitter activation claim retained after failure', { + repo: prRef.repo, prNumber: prRef.prNumber, ownershipKey, generationId, + placement: placement.status, + reason: !generationId ? 'claim-acknowledgement-unknown' + : placement.status === 'not-created' ? 'claim-release-unconfirmed' : 'worker-may-exist', + error: describeError(error).errorMessage, + }) + } + // Allow another event to retry admission. The durable claim still fences + // uncertain placement; a plan without a result is never an adopted worker. this.#babysitterSpawned.delete(babysitterKey) this.#babysitterPr.delete(babysitterKey) this.#babysitterIssueRefs.delete(babysitterKey) - if (await this.#assertIssueDispatchLifecycleOwner(record.issue)) { + if (claimReleased && await this.#assertIssueDispatchLifecycleOwner(record.issue)) { await this.#state.clearBabysitterSession(this.#workspaceId, babysitterKey) } + if (claimReleased && this.#usesDurableDispatchLifecycle()) this.#scheduleDispatchLifecycleRetry(record) this.#increment('babysitterSpawnFailures') this.#error(error, record.issue) } finally { @@ -17824,13 +17892,16 @@ export class FactoryLoop implements Factory { return await this.#readPrSnapshot(ref) } - async #readPrSnapshot(ref: Pick): Promise { + async #readPrSnapshot(ref: Pick, + options?: { forActivation?: boolean }, + ): Promise { const discoveredPaths = await this.#pullMetaPathsFor(ref.repo, ref.prNumber) const candidatePaths = [...new Set([ref.path, ...discoveredPaths].filter((path): path is string => Boolean(path)))] for (const path of candidatePaths) { try { const snapshot = parsePullSnapshot((await this.#mount.readFile(path)).content, ref.prNumber) - if (snapshot) { + if (snapshot && (!options?.forActivation || + (snapshot.state !== undefined && snapshot.draft !== undefined && snapshot.labels !== undefined))) { return snapshot } } catch { diff --git a/src/ports/fleet.ts b/src/ports/fleet.ts index 2ba1820f..3ac90127 100644 --- a/src/ports/fleet.ts +++ b/src/ports/fleet.ts @@ -151,6 +151,14 @@ export type SendInput = { mode?: 'wait' | 'steer' } +/** Positive adapter evidence that spawn failed before any worker placement. */ +export class FleetSpawnNotCreatedError extends Error { + constructor(cause: unknown) { + super(cause instanceof Error ? cause.message : String(cause), { cause }) + this.name = 'FleetSpawnNotCreatedError' + } +} + /** * Positive transport evidence that a correlated message cannot be delivered. * Unlike a delivery-confirmation timeout, this makes an uncorrelated retry safe. diff --git a/src/ports/index.ts b/src/ports/index.ts index b23029eb..56bb5a2e 100644 --- a/src/ports/index.ts +++ b/src/ports/index.ts @@ -21,7 +21,7 @@ export type { SubscribeOptions, Subscription, } from './mount' -export { FleetDeliveryRejectedError } from './fleet' +export { FleetDeliveryRejectedError, FleetSpawnNotCreatedError } from './fleet' export type { A2aSkill, AgentLifecycleSignal, From 5fad4da0941ac8a2a8c6482c9bed016976a75000 Mon Sep 17 00:00:00 2001 From: kjgbot Date: Sun, 6 Sep 2026 21:38:19 +0200 Subject: [PATCH 2/2] fix(fleet): keep the pre-placement error class and share one roster read MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `FleetSpawnNotCreatedError` wrapped every pre-placement failure and stamped its own name over the cause's. The bit it adds — "no worker was created" — has one reader, and that reader uses `instanceof`; every other reader of a dispatch failure reads a class NAME off the OUTERMOST error. So the wrapper was erasing `FactoryAgentRegistrationError`, `ReadOnlyFleetIdentityError` and `RelaySpawnAckTimeoutError` from `perItemDispatchSkipReason`'s operator sentence, from the hosted orchestrator's `errorClass`, and from the head of the #355 class-name vocabulary — collapsing an enrolment refusal, a read-only identity and a bootstrap ack timeout into one indistinguishable bucket, on exactly the surfaces an operator has during an outage. The wrapper now adopts the cause's class name as its own (and exposes it as `causeClass`), guarded by `isTelemetryErrorClassName` so a dependency-controlled `name` still cannot choose what crosses a telemetry boundary. `instanceof`, and therefore `placement.status = 'not-created'`, is unchanged. Durable recovery of a factory-created babysitter also ran two roster-based admission passes back to back: an adoption pass that only no-ops when no live roster entry exists, then `#ensureBabysitter`'s real admission repeating the same `retryOnTimeout(roster(), { attempts: 3, delayMs: 2000 })`. The adoption pass now hands its read forward through `#admissionRoster`, keyed by PR identity because a reconciled agent exit can reach that admission first. The hand-off is single-use and age-bounded, so a slow gate chain re-reads rather than placing against a roster that has stopped describing the fleet; the durable `factory-created:` claim, not adoption, remains what keeps one PR to one worker. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_017Ld4S9gUGzTjhVhbtX9cTd Session-Id: fe8515ad-c2bc-4f9a-9b39-88b2ba12f73e --- src/fleet/relay-fleet-client.test.ts | 97 +++++++++++++++++++++++- src/orchestrator/factory.test.ts | 85 +++++++++++++++++++++ src/orchestrator/factory.ts | 109 +++++++++++++++++++++++---- src/ports/fleet.ts | 38 +++++++++- 4 files changed, 313 insertions(+), 16 deletions(-) diff --git a/src/fleet/relay-fleet-client.test.ts b/src/fleet/relay-fleet-client.test.ts index e0e048fc..524e5ed4 100644 --- a/src/fleet/relay-fleet-client.test.ts +++ b/src/fleet/relay-fleet-client.test.ts @@ -2,8 +2,10 @@ import { describe, expect, it, vi } from 'vitest' import { FleetSpawnNotCreatedError } from '../ports/fleet' import { describeControlPlaneError } from './control-plane-circuit' -import { FactoryAgentRegistrationError, MAX_REGISTRATION_ATTEMPTS, ReadOnlyFleetIdentityError, RelayFleetClient, type RelayClientFactoryOptions, type RelayClientLike } from './relay-fleet-client' +import { FactoryAgentRegistrationError, MAX_REGISTRATION_ATTEMPTS, ReadOnlyFleetIdentityError, RelayFleetClient, RelaySpawnAckTimeoutError, type RelayClientFactoryOptions, type RelayClientLike } from './relay-fleet-client' import { runFleetCli } from '../cli/fleet' +import { telemetryErrorClass } from '../observability/error-class' +import { factoryDispatchFailureReasonCodeForErrorClass } from '../orchestrator/dispatch-failure-reason' import type { RelayActionInvocation, @@ -2521,3 +2523,96 @@ describe('registration failures survive redaction with their cause', () => { expect(rendered).not.toContain('https://') }) }) + +/** + * A pre-placement wrapper must not erase the class that names the failure. + * + * `FleetSpawnNotCreatedError` exists to carry one extra bit — "no worker was + * created" — to the one reader that needs it (`placement.status`, decided by + * `instanceof`). Every other reader of a dispatch failure reads the OUTERMOST + * error's class name: `perItemDispatchSkipReason` renders + * `dispatch failed ()`, the hosted orchestrator publishes + * `errorClass: telemetryErrorClass(error)`, and the #355 vocabulary maps five + * allowlisted class names onto failure codes. + * + * Collapsing every pre-placement failure to one class name would make an + * enrolment failure indistinguishable from a sandbox-provision refusal on + * exactly the surfaces an operator has during an outage. So the wrapper takes + * the cause's class name as its own — guarded by the very allowlist that + * publishes it, so a dependency-controlled `name` still cannot choose what + * crosses the boundary. + */ +describe('FleetSpawnNotCreatedError class preservation', () => { + const preserved = [ + { + cause: () => new FactoryAgentRegistrationError( + 'ar-350-impl-factory', + 'MAX_ATTEMPTS', + 'Remote agent did not register with the fleet before the startup deadline', + ), + name: 'FactoryAgentRegistrationError', + code: 'agent-registration-failed', + }, + { + cause: () => new ReadOnlyFleetIdentityError('factory-cloud'), + name: 'ReadOnlyFleetIdentityError', + code: 'fleet-identity-read-only', + }, + { + cause: () => new RelaySpawnAckTimeoutError('spawn invocation ar-350-impl-factory', 300_000), + name: 'RelaySpawnAckTimeoutError', + code: 'spawn-ack-timeout', + }, + ] as const + + it.each(preserved)('keeps $name readable on the outermost error', ({ cause, name, code }) => { + const wrapped = new FleetSpawnNotCreatedError(cause()) + + // The not-created bit still reaches its only reader. + expect(wrapped).toBeInstanceOf(FleetSpawnNotCreatedError) + // And every class-name reader still sees the failure that actually happened. + expect(wrapped.name).toBe(name) + expect(wrapped.causeClass).toBe(name) + expect(telemetryErrorClass(wrapped)).toBe(name) + expect(factoryDispatchFailureReasonCodeForErrorClass(telemetryErrorClass(wrapped))).toBe(code) + }) + + it('distinguishes the three pre-placement classes from one another', () => { + const names = preserved.map(({ cause }) => telemetryErrorClass(new FleetSpawnNotCreatedError(cause()))) + expect(new Set(names).size).toBe(names.length) + }) + + it('falls back to its own class name when the cause names no admissible class', () => { + const hostile = new Error('placement refused') + hostile.name = 'at_live_abcdef0123456789' + + for (const cause of [hostile, 'not an error', undefined]) { + const wrapped = new FleetSpawnNotCreatedError(cause) + expect(wrapped.name).toBe('FleetSpawnNotCreatedError') + expect(wrapped.causeClass).toBe('FleetSpawnNotCreatedError') + expect(telemetryErrorClass(wrapped)).toBe('FleetSpawnNotCreatedError') + } + }) + + it('surfaces a read-only identity refusal from spawn under its own class name', async () => { + const messaging = new FakeMessaging() + const fleet = new RelayFleetClient({ + workspaceKey: 'rk_live_test', + readOnly: true, + env: {}, + sleep: immediateSleep, + pollIntervalMs: 0, + createRelay: () => ({ messaging: messaging.asMessaging() }), + }) + + const outcome = await fleet.spawn({ + name: 'ar-350-impl-factory', + capability: 'spawn:codex', + repo: 'AgentWorkforce/factory', + }).then(() => 'resolved' as const, (error: unknown) => error) + + expect(messaging.placements).toEqual([]) + expect(outcome).toBeInstanceOf(FleetSpawnNotCreatedError) + expect(telemetryErrorClass(outcome)).toBe('ReadOnlyFleetIdentityError') + }) +}) diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 83aed1ea..4db694b4 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -28909,6 +28909,91 @@ describe('FactoryLoop PR babysitter', () => { } }) + /** + * Recovery admits a factory-created babysitter through TWO calls into + * `#spawnAgent`: an adoptOnly pass that can only adopt a live roster entry, + * then `#ensureBabysitter`'s real admission. Each of those reads the roster + * through `retryOnTimeout(..., { attempts: 3, delayMs: 2000 })`, so a + * recovery that has to place a worker used to pay for the same roster read + * twice — the second one asking a question the first had just answered. + * + * Counted rather than asserted structurally, because the cost IS the point: + * everything the successor does before the babysitter is placed is fixed, + * so one read per admission pass is directly visible in the total. + */ + it('admits a recovered factory-created babysitter with one roster pass', async () => { + const root = await mkdtemp(join(tmpdir(), 'babysitter-recovery-roster-')) + const issue = realIssueFile(401, ready, { title: 'Real recovery admission cost' }) + const mount = new FakeMountClient({ [issuePath(401)]: issue }) + seedPrMeta(mount, 'AgentWorkforce/pear', 401, { state: 'open', draft: false, labels: [] }) + class DurableFleet extends FakeFleetClient { readonly durableOwnership = true } + const firstFleet = new DurableFleet() + const realSpawn = firstFleet.spawn.bind(firstFleet) + // Positive non-placement evidence, so the first process releases its claim + // and recovery has to re-admit rather than fence itself out. + vi.spyOn(firstFleet, 'spawn').mockImplementation(async (input) => { + if (input.name.includes('babysit')) throw new FleetSpawnNotCreatedError(new Error('spawn preflight failed')) + return realSpawn(input) + }) + const state = () => new FileStateStore({ batchSize: 2, watchStatePath: join(root, 'state.json') }) + const first = createFactory(babysitterConfig(), { + mount, fleet: firstFleet, stateStore: state(), triage: new StaticTriage(), + probePrResolver: async () => ({ repo: 'AgentWorkforce/pear', prNumber: 401 }), + }) + const nextFleet = new DurableFleet() + // Hydration restores observation metadata; it cannot prove a live worker. + vi.spyOn(nextFleet, 'hydrateTracked').mockImplementation(() => {}) + const nextState = state() + // Counted up to the placement, because after it the successor's ordinary + // loops read the roster on their own schedule and the number stops being + // about admission. Everything before it is fixed: the successor's startup + // and restore make ROSTER_READS_BEFORE_ADMISSION reads, and the recovery + // pass adds one per `#spawnAgent` that reaches the fleet. + const ROSTER_READS_BEFORE_ADMISSION = 2 + const mark = nextState.markRunning.bind(nextState) + vi.spyOn(nextState, 'markRunning').mockImplementation(async (...args) => { + // Recovery's adoption pass and a reconciled agent exit both reach the + // babysitter admission, and either can get there first. Holding the PR + // claim back one beat pins the interleaving under test — adoption + // finishes, then admission runs — so what is measured is the cost of + // that pair, not which of them the event loop happened to schedule. + if (args[1] === 'factory-created:agentworkforce/pear#401') { + await new Promise((resolve) => setTimeout(resolve, 25)) + } + return mark(...args) + }) + let counting = true + let rosterReads = 0 + const readRoster = nextFleet.roster.bind(nextFleet) + vi.spyOn(nextFleet, 'roster').mockImplementation(async () => { + if (counting) rosterReads += 1 + return readRoster() + }) + const nextSpawn = nextFleet.spawn.bind(nextFleet) + vi.spyOn(nextFleet, 'spawn').mockImplementation(async (input) => { + if (input.name.includes('babysit')) counting = false + return nextSpawn(input) + }) + const next = createFactory(babysitterConfig(), { + mount, fleet: nextFleet, stateStore: nextState, triage: new StaticTriage(), dispatchLifecycleRetryMs: 10, + probePrResolver: async () => ({ repo: 'AgentWorkforce/pear', prNumber: 401 }), + }) + try { + await first.dispatch(await first.triageIssue(parseLinearIssue(issuePath(401), issue))) + firstFleet.emitAgentExit('ar-401-impl-pear', 'worker_exited') + await vi.waitFor(() => expect(first.status().counters.babysitterSpawnFailures).toBe(1)) + expect((await state().listDispatchLifecycles('factory-test'))[0]?.[1].phase).toBe('dispatching') + await first.stop() + await next.start({ mode: 'dispatch-owner' }) + await vi.waitFor(() => expect(nextFleet.spawns.filter((input) => input.name.includes('babysit'))).toHaveLength(1)) + expect(rosterReads).toBe(ROSTER_READS_BEFORE_ADMISSION + 1) + } finally { + await first.stop() + await next.stop() + await rm(root, { recursive: true, force: true }) + } + }) + it.each(['denied', 'unavailable', 'legacy-session'])('fails closed when the PR claim is %s', async (result) => { const issue = realIssueFile(401, ready, { title: 'Real babysitter durable claim' }) const mount = new FakeMountClient({ [issuePath(401)]: issue }) diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 1c190291..7ca1948b 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -518,6 +518,25 @@ 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 + +/** + * A roster read handed from one admission pass to the next one behind it. + * + * Mutable on purpose, and for the same reason the `placement` holder threaded + * through `#spawnAgent` is: the two passes are separated by an entire gate + * chain, so the answer has to travel in something the callee can fill and the + * next callee can empty. Read and cleared by `#admissionRoster`. + */ +type RosterHandoff = { entry?: RosterEntry; readAtMs?: number } + +/** + * How stale a handed-over roster read may be before it is re-read. + * + * Short enough that the snapshot still describes the fleet a placement is + * about to be made against, long enough to cover the durable reads and the + * claim that sit between the adoption pass and the admission it feeds. + */ +const ROSTER_HANDOFF_MAX_AGE_MS = 5_000 const REMOTE_AGENT_REGISTRATION_POLL_MS = 500 /** * Ceiling on the durable capacity-wait re-arm (#303). @@ -1333,6 +1352,11 @@ export class FactoryLoop implements Factory { // webhooks / agent-exit safety nets don't respawn it while multi-repository issues retain one // owner per PR. readonly #babysitterSpawned = new Set() + /** + * Roster reads a recovery adoption pass left for the admission behind it, + * keyed by PR identity. Set and cleared around one recovery branch only. + */ + readonly #babysitterAdmissionRoster = new Map() readonly #babysitterSpawnInFlight = new Map>() // Composite issue + PR identity -> the open PR the babysitter is shepherding, including the // webhook-fed mount path so readiness can re-read PR meta without a gh call. @@ -9036,8 +9060,19 @@ export class FactoryLoop implements Factory { const owned = spec.ownedPullRequest // A live roster entry can close the lost-ack gap without placement. // A missing entry cannot authorize replay of the persisted intent. - await this.#spawnAgent(record, spec, record.dryRun, undefined, true) - await this.#ensureBabysitter(record, { repo: owned.repo, prNumber: owned.number, path: owned.path }) + // One roster read serves both admission passes. The adoption attempt + // fills the hand-off; whichever `#ensureBabysitter` admits this PR next + // consumes it. Keyed by the PR rather than threaded through the call, + // because a reconciled agent exit can reach that admission first. + const roster: RosterHandoff = {} + const ownedIdentity = githubPrIdentity(owned.repo, owned.number) + if (ownedIdentity) this.#babysitterAdmissionRoster.set(ownedIdentity, roster) + try { + await this.#spawnAgent(record, spec, record.dryRun, { adoptOnly: true, roster }) + await this.#ensureBabysitter(record, { repo: owned.repo, prNumber: owned.number, path: owned.path }) + } finally { + if (ownedIdentity) this.#babysitterAdmissionRoster.delete(ownedIdentity) + } const tracked = record.agents.get(spec.name) if (tracked?.result) agents.push({ name: tracked.result.name, role: spec.role }) continue @@ -11596,10 +11631,57 @@ export class FactoryLoop implements Factory { await writeJsonFileAtomically(path, registry) } - async #spawnAgent(record: InFlightIssue, spec: AgentSpec, dryRun: boolean, - placement?: { status: 'not-created' | 'uncertain' | 'created' }, - adoptOnly = false, - ): Promise<{ name: string }> { + /** + * The roster this admission pass decides on, reusing a handed-over read. + * + * Durable recovery admits a factory-created babysitter through two + * `#spawnAgent` calls in a row — an adoption pass that can only adopt a live + * roster entry, then `#ensureBabysitter`'s real admission — and each read + * costs a `retryOnTimeout` chain of up to three attempts two seconds apart. + * The second call asks the question the first has just answered, so the + * first hands its answer forward instead of paying for it again. + * + * The hand-off is single-use and age-bounded, and both properties matter. + * Single-use stops a snapshot leaking into a later, unrelated placement; the + * age bound stops a slow gate chain between the two passes from choosing a + * node out of a roster that has stopped describing the fleet. When either + * says no, this reads the roster exactly as it always did. + * + * What it does not lean on: adoption is not what keeps one PR to one worker. + * The durable `factory-created:` claim taken between the two passes is, + * and reusing a snapshot cannot weaken it. + */ + async #admissionRoster(record: InFlightIssue, handoff?: RosterHandoff): Promise { + const handed = handoff?.entry + const handedAtMs = handoff?.readAtMs + if (handoff) { + handoff.entry = undefined + handoff.readAtMs = undefined + } + if (handed && handedAtMs !== undefined && this.#clock.now() - handedAtMs <= ROSTER_HANDOFF_MAX_AGE_MS) { + return handed + } + let roster: RosterEntry + try { + roster = await retryOnTimeout(() => this.#fleet.roster(), { attempts: 3, delayMs: 2000 }) + } catch (error) { + throw contextualError(`Dispatch roster lookup failed for ${record.issue.key}`, error) + } + if (handoff) { + handoff.entry = roster + handoff.readAtMs = this.#clock.now() + } + return roster + } + + async #spawnAgent(record: InFlightIssue, spec: AgentSpec, dryRun: boolean, options: { + placement?: { status: 'not-created' | 'uncertain' | 'created' } + /** Adopt a live roster entry if there is one; never place a new worker. */ + adoptOnly?: boolean + /** Roster read shared with the admission pass that follows this one. */ + roster?: RosterHandoff + } = {}): Promise<{ name: string }> { + const { placement, adoptOnly = false } = options const batch = await this.#batch() const invocationId = batch.invocationIdFor(record.issue, spec) const existing = record.agents.get(spec.name) @@ -11622,12 +11704,7 @@ export class FactoryLoop implements Factory { return { name: spec.name } } - let roster - try { - roster = await retryOnTimeout(() => this.#fleet.roster(), { attempts: 3, delayMs: 2000 }) - } catch (error) { - throw contextualError(`Dispatch roster lookup failed for ${record.issue.key}`, error) - } + const roster = await this.#admissionRoster(record, options.roster) const rosterAgent = roster.agents.find((agent) => agent.name === spec.name) if (rosterAgent) { if (placement) placement.status = 'created' @@ -17597,11 +17674,17 @@ export class FactoryLoop implements Factory { return } generationId = claim.generationId + const handedRoster = this.#babysitterAdmissionRoster.get(prIdentity) const spawned = await this.#spawnAgent(record, { ...spec, task, ownedPullRequest: { repo: prRef.repo, number: prRef.prNumber, path: prRef.path }, - }, false, placement) + }, false, { + placement, + // Consumed when a recovery adoption pass read the roster for this PR + // moments ago; absent, this reads it exactly as it always did. + ...(handedRoster ? { roster: handedRoster } : {}), + }) const tracked = record.agents.get(spawned.name) this.#babysitterPr.set(babysitterKey, { repo: prRef.repo, diff --git a/src/ports/fleet.ts b/src/ports/fleet.ts index 3ac90127..fe128306 100644 --- a/src/ports/fleet.ts +++ b/src/ports/fleet.ts @@ -1,3 +1,5 @@ +import { isTelemetryErrorClassName } from '../observability/error-class' + // `workflow:run` is a fleet node capability. The node-side Phase 4 handler // invokes the Relayflows SDK in that node's repo checkout; the // factory only emits the workflow path and inputs through the relay fleet. @@ -151,11 +153,43 @@ export type SendInput = { mode?: 'wait' | 'steer' } -/** Positive adapter evidence that spawn failed before any worker placement. */ +/** + * Positive adapter evidence that spawn failed before any worker placement. + * + * The extra bit this carries — "no worker was created" — has exactly one + * reader, and it reads it by `instanceof`. Every *other* reader of a dispatch + * failure reads a class NAME off the outermost error: `perItemDispatchSkipReason` + * renders `dispatch failed ()`, the hosted orchestrator publishes + * `errorClass`, and the #355 vocabulary maps allowlisted class names onto + * failure codes. A wrapper that stamped its own name over the cause's would + * therefore buy one bit by destroying the identity of the failure on every + * surface an operator actually has: an enrolment refusal + * (`FactoryAgentRegistrationError`), a read-only identity + * (`ReadOnlyFleetIdentityError`) and a bootstrap ack timeout + * (`RelaySpawnAckTimeoutError`) would all read as one indistinguishable + * "pre-placement failure". + * + * So the wrapper adopts the cause's class name as its own and keeps the bit in + * its type. `isTelemetryErrorClassName` is the guard: `name` is writable and a + * cause may come from a dependency, so only a string the publication allowlist + * would admit anyway may be adopted — anything else falls back to this class's + * own name. Nothing crosses a boundary here that `telemetryErrorClass` would + * not have let across one frame further down the cause chain. + */ export class FleetSpawnNotCreatedError extends Error { + /** + * The class name of the pre-placement cause, or this class's own when the + * cause named none the telemetry allowlist admits. Always equal to `name`; + * exposed so a reader that wants the cause's identity does not have to know + * that `name` was reassigned. + */ + readonly causeClass: string + constructor(cause: unknown) { super(cause instanceof Error ? cause.message : String(cause), { cause }) - this.name = 'FleetSpawnNotCreatedError' + const causeName = cause instanceof Error ? cause.name : '' + this.causeClass = isTelemetryErrorClassName(causeName) ? causeName : 'FleetSpawnNotCreatedError' + this.name = this.causeClass } }