diff --git a/src/cli/fleet.test.ts b/src/cli/fleet.test.ts index b4c9af3..a60e152 100644 --- a/src/cli/fleet.test.ts +++ b/src/cli/fleet.test.ts @@ -2590,6 +2590,60 @@ describe('fleet CLI runtime', () => { } }) + it('surfaces live daemon counters from the heartbeat in top-level status', async () => { + const root = await mkdtemp(join(tmpdir(), 'fleet-cli-daemon-counters-')) + try { + const heartbeatPath = join(root, 'heartbeat.json') + const configPath = await writeConfig(root, { loop: { heartbeatPath, heartbeatStaleMs: 10_000 } }) + const now = Date.now() + await writeFile(heartbeatPath, `${JSON.stringify({ + pid: process.pid, + status: 'running', + iteration: 0, + maxIterations: 0, + updatedAt: new Date(now).toISOString(), + updatedAtMs: now, + eventListener: { state: 'subscribed' }, + counters: { + babysitterEventsIgnoredUnownedPr: 7, + babysitterFlatEventsUnreadable: 2, + }, + })}\n`, 'utf8') + const output = buffer() + const factory = { + start: vi.fn(), + stop: vi.fn(), + runLoop: vi.fn(async () => []), + runOnce: vi.fn(), + status: vi.fn(() => ({ inFlight: [], queued: [], counters: { localStatusRead: 1 } })), + triageIssue: vi.fn(), + dispatch: vi.fn(), + on: vi.fn(), + dispose: vi.fn(), + } as unknown as Factory + + const code = await runFleetCli(['status', '--config', configPath], { + fleet: new FakeFleetClient(), + mount: new FakeMountClient(), + createFactory: () => factory, + stdout: output, + stderr: buffer(), + }) + + expect(code).toBe(0) + expect(JSON.parse(output.text())).toMatchObject({ + eventListener: { state: 'subscribed' }, + counters: { + babysitterEventsIgnoredUnownedPr: 7, + babysitterFlatEventsUnreadable: 2, + localStatusRead: 1, + }, + }) + } finally { + await rm(root, { recursive: true, force: true }) + } + }) + it('surfaces a stale registered workspace mirror in factory status', async () => { const root = await mkdtemp(join(tmpdir(), 'fleet-cli-stale-status-')) try { @@ -3135,7 +3189,12 @@ describe('fleet CLI runtime', () => { expect(result.reports).toHaveLength(2) expect(result.status.counters.loopIdle).toBe(1) const heartbeat = JSON.parse(await readFile(heartbeatPath, 'utf8')) - expect(heartbeat).toMatchObject({ status: 'idle', iteration: 2, maxIterations: 2 }) + expect(heartbeat).toMatchObject({ + status: 'idle', + iteration: 2, + maxIterations: 2, + counters: { loopIdle: 1 }, + }) const statusOut = buffer() const statusCode = await runFleetCli([ diff --git a/src/cli/fleet.ts b/src/cli/fleet.ts index 46fdc9e..8a6e701 100644 --- a/src/cli/fleet.ts +++ b/src/cli/fleet.ts @@ -1064,8 +1064,16 @@ async function factoryStatusWithMountHealth( root?: string } }> { - const status = factory.status() + const processStatus = factory.status() const heartbeat = await readFactoryLoopHeartbeat(heartbeatPath) + // `factory status` runs in a fresh CLI process, so its in-memory Factory has + // no knowledge of counters accumulated by the live daemon. The heartbeat is + // the daemon-owned status handoff; local counters win only when this helper + // is used in the same process as active work. + const status = { + ...processStatus, + counters: { ...(heartbeat?.counters ?? {}), ...processStatus.counters }, + } const liveness = checkFactoryLoopLiveness(heartbeat, { staleMs: heartbeatStaleMs }) const eventListener = liveness.ok ? heartbeat?.eventListener ?? { diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 1873e51..c5c2469 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -18959,6 +18959,196 @@ describe('FactoryLoop PR babysitter', () => { } }) + it('retries a transient PR-open snapshot read and still spawns the babysitter', async () => { + const issue = realIssueFile(418, ready, { title: 'Real transient babysitter snapshot' }) + const prPath = '/github/repos/AgentWorkforce/pear/pulls/418/metadata.json' + class FailOncePrSnapshotMount extends FakeMountClient { + failed = false + + override async readFile(path: string) { + if (path === prPath && !this.failed) { + this.failed = true + this.reads.push(path) + throw new Error('fetch failed') + } + return super.readFile(path) + } + } + const mount = new FailOncePrSnapshotMount({ [issuePath(418)]: issue }) + const fleet = new FakeFleetClient() + const warnings: unknown[][] = [] + const factory = createFactory(babysitterConfig(), { + mount, + fleet, + triage: new StaticTriage(), + clock: { now: Date.now, sleep: async () => undefined }, + logger: { + warn: (...args: unknown[]) => warnings.push(args), + error: () => undefined, + }, + }) + + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(418), issue))) + await factory.start({ mode: 'live', liveSubscription: { transport: 'subscribe' } }) + try { + await vi.waitFor(() => expect(factory.status().counters.babysitterReconcileRuns).toBeGreaterThan(0)) + mount.files.set(prPath, { + content: { + number: 418, + state: 'open', + head_ref: 'factory/ar-418-transient-read', + isDraft: false, + url: 'https://github.com/AgentWorkforce/pear/pull/418', + }, + }) + mount.emit(changeEvent(prPath, 'pr-418-open')) + + await vi.waitFor(() => expect(fleet.spawns.map((spawn) => spawn.name)).toContain('ar-418-babysit')) + expect(mount.reads.filter((path) => path === prPath)).toHaveLength(2) + expect(factory.status().counters.babysitterPrSnapshotReadFailures).toBe(1) + expect(factory.status().counters.babysitterPrSnapshotReadRetries).toBe(1) + expect(factory.status().counters.babysitterPrSnapshotReadDeadLetters).toBeUndefined() + expect(warnings.some((warning) => warning[0] === '[factory] babysitter could not read PR snapshot; retrying')).toBe(true) + } finally { + await factory.stop() + } + }) + + it('dead-letters exhausted PR snapshot reads at error level for periodic redrive', async () => { + const issue = realIssueFile(419, ready, { title: 'Real exhausted babysitter snapshot' }) + const prPath = '/github/repos/AgentWorkforce/pear/pulls/419/metadata.json' + class UnreadablePrSnapshotMount extends FakeMountClient { + override async readFile(path: string) { + if (path === prPath) { + this.reads.push(path) + throw new Error('fetch failed') + } + return super.readFile(path) + } + } + const mount = new UnreadablePrSnapshotMount({ [issuePath(419)]: issue }) + const fleet = new FakeFleetClient() + const warnings: unknown[][] = [] + const errors: unknown[][] = [] + const factory = createFactory(babysitterConfig(), { + mount, + fleet, + triage: new StaticTriage(), + clock: { now: Date.now, sleep: async () => undefined }, + logger: { + warn: (...args: unknown[]) => warnings.push(args), + error: (...args: unknown[]) => errors.push(args), + }, + }) + + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(419), issue))) + await factory.start({ mode: 'live', liveSubscription: { transport: 'subscribe' } }) + try { + await vi.waitFor(() => expect(factory.status().counters.babysitterReconcileRuns).toBeGreaterThan(0)) + mount.files.set(prPath, { + content: { number: 419, state: 'open', head_ref: 'factory/ar-419-unreadable', isDraft: false }, + }) + mount.emit(changeEvent(prPath, 'pr-419-open')) + + await vi.waitFor(() => expect(factory.status().counters.babysitterPrSnapshotReadDeadLetters).toBe(1)) + expect(mount.reads.filter((path) => path === prPath)).toHaveLength(5) + expect(factory.status().counters.babysitterPrSnapshotReadFailures).toBe(5) + expect(factory.status().counters.babysitterPrSnapshotReadRetries).toBe(4) + expect(warnings.filter((warning) => warning[0] === '[factory] babysitter could not read PR snapshot; retrying')).toHaveLength(4) + expect(errors.some((error) => error[0] === '[factory] babysitter PR snapshot read retries exhausted; dead-lettered for reconcile')).toBe(true) + expect(fleet.spawns.map((spawn) => spawn.name)).not.toContain('ar-419-babysit') + } finally { + await factory.stop() + } + }) + + it('adopts an ownerless open Factory PR during the periodic reconcile sweep', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-babysitter-orphan-reconcile-')) + const path = githubIssuePath('AgentWorkforce', 'cloud', 3021) + const issue = githubIssueFile(3021, { + repo: 'cloud', + labels: ['factory'], + title: '[factory] Fix failing CI checks on PR #3017', + }) + const prPath = '/github/repos/AgentWorkforce/cloud/pulls/3024/metadata.json' + const prUrl = 'https://github.com/AgentWorkforce/cloud/pull/3024' + const mount = new FakeMountClient({ + [path]: issue, + [prPath]: { + number: 3024, + title: '3021: [factory] Fix 2 failing CI checks on PR #3017 (deployment objective metadata)', + body: '', + state: 'open', + head_ref: 'factory/3022-chief-org-live-population', + isDraft: false, + isCrossRepository: false, + url: prUrl, + }, + }) + mount.setSubRoot('/linear/issues', 'absent') + const fleet = new FakeFleetClient() + const factory = createFactory(babysitterConfig({ + issueSource: 'github', + repos: { + byLabel: { cloud: 'AgentWorkforce/cloud' }, + clonePaths: { 'AgentWorkforce/cloud': '/work/cloud' }, + default: 'AgentWorkforce/cloud', + }, + safety: { requireLabel: 'factory' }, + }), { + mount, + fleet, + triage: new StaticTriage(), + }) + try { + await factory.dispatch(await factory.triageIssue(parseGithubFactoryIssue(path, issue))) + + await factory.runLoop({ + maxIterations: 1, + heartbeatPath: join(root, 'heartbeat.json'), + registryPath: join(root, 'registry.json'), + }) + + const babysitter = fleet.spawns.find((spawn) => spawn.name.includes('3021-babysit')) + expect(babysitter?.repo).toBe('AgentWorkforce/cloud') + expect(babysitter?.task).toContain(prUrl) + expect(factory.status().counters.babysitterOrphanedPrsDetected).toBe(1) + expect(factory.status().counters.babysitterOrphanedPrsAdopted).toBe(1) + expect(factory.status().counters.babysitterOrphanedPrAdoptionFailures).toBeUndefined() + } finally { + await factory.stop() + await rm(root, { recursive: true, force: true }) + } + }) + + it('warns for the first unowned PR event and escalates repeated occurrences', async () => { + const mount = new FakeMountClient() + const warnings: unknown[][] = [] + const errors: unknown[][] = [] + const factory = createFactory(babysitterConfig(), { + mount, + fleet: new FakeFleetClient(), + triage: new StaticTriage(), + logger: { + warn: (...args: unknown[]) => warnings.push(args), + error: (...args: unknown[]) => errors.push(args), + }, + }) + await factory.start({ mode: 'live', liveSubscription: { transport: 'subscribe' } }) + try { + mount.emit(changeEvent('/github/repos/AgentWorkforce/pear/pulls/499/comments/1.json', 'unowned-499-1')) + mount.emit(changeEvent('/github/repos/AgentWorkforce/pear/pulls/499/comments/2.json', 'unowned-499-2')) + mount.emit(changeEvent('/github/repos/AgentWorkforce/pear/pulls/499/comments/3.json', 'unowned-499-3')) + + await vi.waitFor(() => expect(factory.status().counters.babysitterEventsIgnoredUnownedPr).toBe(3)) + expect(warnings.some((warning) => warning[0] === '[factory] ignored unowned PR event for babysitter routing; periodic reconcile will attempt adoption')).toBe(true) + expect(errors.some((error) => error[0] === '[factory] repeated unowned PR events have no babysitter; reconcile has not adopted the PR')).toBe(true) + expect(factory.status().counters.babysitterEventsIgnoredUnownedPrEscalations).toBe(1) + } finally { + await factory.stop() + } + }) + it('routes and coalesces only the owned PR review/check/comment events with metadata-only fencing', async () => { const issue = realIssueFile(420, ready, { title: 'Real babysitter event routing' }) const mount = new FakeMountClient({ [issuePath(420)]: issue }) @@ -20493,7 +20683,15 @@ describe('FactoryLoop PR babysitter', () => { mount, fleet, triage: new StaticTriage(), - probePrResolver: async () => ({ repo: 'AgentWorkforce/pear', prNumber: 408 }), + probePrResolver: async () => ({ + repo: 'AgentWorkforce/pear', + prNumber: 408, + matchScore: 10, + draft: false, + state: 'OPEN', + headRef: 'factory/unrelated-work', + crossRepository: false, + }), }) await factory.start({ mode: 'live', liveSubscription: { transport: 'subscribe' } }) diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 13a25b0..9d70e0a 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -131,6 +131,8 @@ type TerminationRoots = { pids: number[]; status: AgentPidResolution['status'] } type ResolvedIssuePr = { repo: string prNumber: number + /** Resolver evidence: branch=30, title=20, explicit body reference=10. */ + matchScore?: number draft?: boolean headRef?: string headRepo?: string @@ -164,6 +166,15 @@ type BabysitterResourceSubscription = Pick< 'subscriptionId' | 'provider' | 'resourceRef' | 'subscriberId' | 'ownerId' | 'expiresAt' > & { terminal?: boolean } type BabysitterPendingDeliveryClaim = { deliveryId: string; claimToken: string } +type BabysitterPrSnapshotDeadLetter = { + path: string + repo: string + prNumber: number + attempts: number + exhaustions: number + error: string + failedAtMs: number +} type BabysitterPrRef = { repo: string prNumber: number @@ -292,6 +303,11 @@ const INJECTION_RETRY_ATTEMPT_TIMEOUT_MS = 15_000 const INJECTION_MAX_ATTEMPTS = 6 const BABYSITTER_EVENT_COALESCE_MS = 750 const BABYSITTER_EVENT_RETRY_MS = 1_000 +const BABYSITTER_PR_SNAPSHOT_READ_MAX_ATTEMPTS = 5 +const BABYSITTER_PR_SNAPSHOT_READ_RETRY_BASE_MS = 250 +const BABYSITTER_PR_SNAPSHOT_DEAD_LETTER_LIMIT = 200 +const BABYSITTER_UNOWNED_EVENT_ESCALATION_COUNT = 3 +const BABYSITTER_UNOWNED_EVENT_OBSERVATION_LIMIT = 500 const BABYSITTER_SUBSCRIPTION_TTL_SECONDS = 60 * 60 // Relayfile receives provider-native GitHub events, not the materialized file // changes that the legacy local router consumed. `closed` is separately @@ -557,6 +573,9 @@ export class FactoryLoop implements Factory { // owner per PR. readonly #babysitterSpawned = new Set() readonly #babysitterSpawnInFlight = new Map>() + readonly #babysitterPrSnapshotReads = new Map>() + readonly #babysitterPrSnapshotDeadLetters = new Map() + readonly #babysitterUnownedEventOccurrences = 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. readonly #babysitterPr = new Map() @@ -997,6 +1016,9 @@ export class FactoryLoop implements Factory { this.#liveEventQueue.length = 0 this.#completionInFlight.clear() this.#babysitterSpawned.clear() + this.#babysitterPrSnapshotReads.clear() + this.#babysitterPrSnapshotDeadLetters.clear() + this.#babysitterUnownedEventOccurrences.clear() this.#babysitterPr.clear() this.#babysitterIssueRefs.clear() this.#babysitterSubscriptionOwners.clear() @@ -1591,18 +1613,19 @@ export class FactoryLoop implements Factory { // while the event loop was busy (for example during a large startup pull). // Keep reconciliation active even when babysitters own PR completion. await this.#fleet.reconcileTrackedAgents?.() - // When the babysitter owns PR-open, completion is driven by PR webhooks + - // the babysitter's readiness signal (see #handlePrChange / #handleAgentExit), - // not this polling sweep. Disabling it here is what makes the babysitter path - // webhook-driven rather than polled. - if (this.#config.babysitter.enabled) { - return - } if (this.#completionSweepActive) { return } this.#completionSweepActive = true try { + // Babysitter readiness remains webhook-driven, but PR ownership cannot + // be. A transient miss of the one PR-open event must be recoverable from + // current open-PR state, so this same timer adopts only ownerless PRs and + // never evaluates merge readiness itself. + if (this.#config.babysitter.enabled) { + await this.#reconcileOrphanedBabysitters(reason) + return + } const batch = await this.#batch() const records = batch.inFlight .filter((record) => !record.dryRun && !this.#completionInFlight.has(issueKey(record.issue))) @@ -1662,6 +1685,147 @@ export class FactoryLoop implements Factory { } } + async #reconcileOrphanedBabysitters(reason: 'live-timer' | 'run-loop'): Promise { + this.#increment('babysitterReconcileRuns') + + // Redrive exhausted event paths first. This preserves the exact webhook + // context when the mount recovers, while the issue/PR scan below remains an + // independent restart-safe path if the one-shot event is gone for good. + await Promise.all([...this.#babysitterPrSnapshotDeadLetters.values()] + .slice(0, COMPLETION_SWEEP_BATCH_SIZE) + .map(async (deadLetter) => { + if (this.#stopping) return + this.#increment('babysitterPrSnapshotDeadLetterRedriveAttempts') + await this.#handlePrChange(deadLetter.path) + })) + + const batch = await this.#batch() + const records = batch.inFlight.filter((record) => !record.dryRun) + for (let index = 0; index < records.length; index += COMPLETION_SWEEP_BATCH_SIZE) { + await Promise.all(records.slice(index, index + COMPLETION_SWEEP_BATCH_SIZE).map(async (record) => { + try { + await this.#reconcileOrphanedBabysitterForIssue(record, batch, reason) + } catch (error) { + this.#increment('babysitterReconcileFailures') + this.#logger.error?.('[factory] babysitter orphan reconcile failed', { + issue: record.issue.key, + reason, + error: describeError(error).errorMessage, + }) + } + })) + await this.#refreshLiveHeartbeatIfDue() + if (index + COMPLETION_SWEEP_BATCH_SIZE < records.length) await liveEventYield() + } + } + + async #reconcileOrphanedBabysitterForIssue( + record: InFlightIssue, + batch: BatchSnapshot, + reason: 'live-timer' | 'run-loop', + ): Promise { + if (batch.getIssue(record.issue) !== record) return + const expectedRepoOwners = new Set(record.decision.implementers.map((implementer) => implementer.repo.toLowerCase())).size || 1 + const currentOwners = [...this.#babysitterPr.entries()].filter(([key, ref]) => { + const ownerIssue = this.#babysitterIssueRefs.get(key) + return Boolean(ref.agentName && ownerIssue && issueKey(ownerIssue) === issueKey(record.issue)) + }) + if (currentOwners.length >= expectedRepoOwners) return + + const issue = await this.#readIssue(record.issue.path) + if (!issue || !isInFactoryScope(issue, this.#config.safety)) return + const pr = await this.#openPrForIssue(issue) + if (!pr) return + const exactFactoryBranch = Boolean(pr.headRef && factoryBranchMatchesIssue(pr.headRef, issue.key)) + // Some app-created branches can carry the originating work-unit number + // rather than the mirrored issue number (cloud#3024 is a production + // example). The resolver may still prove the issue through the PR title. + // Accept that strong association only for same-repository `factory/` + // branches; an explicit body mention alone remains insufficient. + const factoryTitleMatch = Boolean( + pr.headRef?.toLowerCase().startsWith('factory/') && + pr.crossRepository !== true && + (pr.matchScore ?? 0) >= 20, + ) + if ( + pr.draft !== false || + pr.state?.trim().toUpperCase() !== 'OPEN' || + !pr.headRef || + (!exactFactoryBranch && !factoryTitleMatch) + ) return + if (pr.path) { + const pathParts = githubPullPathParts(pr.path) + if (!pathParts || githubPrIdentity(`${pathParts.owner}/${pathParts.repo}`, pathParts.number) !== githubPrIdentity(pr.repo, pr.prNumber)) { + this.#increment('babysitterReconcileInvalidPrIdentity') + this.#logger.warn?.('[factory] skipped babysitter orphan reconcile for PR metadata with inconsistent path identity', { + issue: record.issue.key, + repo: pr.repo, + prNumber: pr.prNumber, + path: pr.path, + }) + return + } + } + if (await this.#babysitterOwnerFor(pr.repo, pr.prNumber)) return + + const sameRepoOwner = currentOwners.find(([, ref]) => ref.repo.toLowerCase() === pr.repo.toLowerCase()) + if (sameRepoOwner) return + + this.#increment('babysitterOrphanedPrsDetected') + this.#logger.warn?.('[factory] detected open Factory PR without a babysitter; adopting during reconcile', { + issue: record.issue.key, + repo: pr.repo, + prNumber: pr.prNumber, + reason, + }) + await this.#ensureBabysitter(record, { + repo: pr.repo, + prNumber: pr.prNumber, + url: pr.url, + path: pr.path, + headRef: pr.headRef, + }) + const owner = await this.#babysitterOwnerFor(pr.repo, pr.prNumber) + if (!owner) { + this.#increment('babysitterOrphanedPrAdoptionFailures') + this.#logger.error?.('[factory] open Factory PR remains ownerless after babysitter reconcile', { + issue: record.issue.key, + repo: pr.repo, + prNumber: pr.prNumber, + reason, + }) + void this.#report({ + type: 'factory.anomaly', + level: 'error', + attributes: { + component: 'babysitter', + operation: 'reconcile_pr_owner', + errorCode: 'orphan_adoption_failed', + retryable: true, + count: 1, + }, + }) + return + } + + const identity = githubPrIdentity(pr.repo, pr.prNumber) + if (!identity) return + this.#babysitterUnownedEventOccurrences.delete(identity) + for (const [path, deadLetter] of this.#babysitterPrSnapshotDeadLetters) { + if (githubPrIdentity(deadLetter.repo, deadLetter.prNumber) !== identity) continue + this.#babysitterPrSnapshotDeadLetters.delete(path) + this.#increment('babysitterPrSnapshotReadDeadLettersRecoveredByReconcile') + } + this.#increment('babysitterOrphanedPrsAdopted') + this.#logger.info?.('[factory] orphaned open Factory PR adopted by babysitter reconcile', { + issue: record.issue.key, + repo: pr.repo, + prNumber: pr.prNumber, + babysitter: owner.ref.agentName, + reason, + }) + } + async #completionPrForIssue(issue: LinearIssue): Promise { if (this.#customProbePrResolver) { return this.#probePrResolver(issue) @@ -5312,6 +5476,7 @@ export class FactoryLoop implements Factory { updatedAtMs, registryPath, eventListener: this.#eventListenerStatus(), + counters: { ...this.#counters }, } await mkdir(dirname(path), { recursive: true }) await writeFile(path, `${JSON.stringify(heartbeat, null, 2)}\n`, 'utf8') @@ -9761,9 +9926,38 @@ export class FactoryLoop implements Factory { const owner = await this.#babysitterOwnerFor(`${event.owner}/${event.repo}`, target.prNumber) if (!owner) { this.#increment('babysitterEventsIgnoredUnownedPr') - this.#logger.debug?.('[factory] ignored unowned PR event for babysitter routing', { ...event, prNumber: target.prNumber }) + const identity = githubPrIdentity(`${event.owner}/${event.repo}`, target.prNumber) + if (!identity) continue + const occurrences = (this.#babysitterUnownedEventOccurrences.get(identity) ?? 0) + 1 + this.#babysitterUnownedEventOccurrences.delete(identity) + this.#babysitterUnownedEventOccurrences.set(identity, occurrences) + if (this.#babysitterUnownedEventOccurrences.size > BABYSITTER_UNOWNED_EVENT_OBSERVATION_LIMIT) { + const oldest = this.#babysitterUnownedEventOccurrences.keys().next().value + if (oldest !== undefined) this.#babysitterUnownedEventOccurrences.delete(oldest) + } + const details = { ...event, prNumber: target.prNumber, occurrences } + if (occurrences === BABYSITTER_UNOWNED_EVENT_ESCALATION_COUNT || occurrences % 10 === 0) { + this.#increment('babysitterEventsIgnoredUnownedPrEscalations') + this.#logger.error?.('[factory] repeated unowned PR events have no babysitter; reconcile has not adopted the PR', details) + void this.#report({ + type: 'factory.anomaly', + level: 'error', + attributes: { + component: 'babysitter', + operation: 'route_pr_event', + errorCode: 'unowned_pr', + count: occurrences, + }, + }) + } else if (occurrences === 1) { + this.#logger.warn?.('[factory] ignored unowned PR event for babysitter routing; periodic reconcile will attempt adoption', details) + } else { + this.#logger.debug?.('[factory] ignored repeated unowned PR event while awaiting babysitter reconcile', details) + } continue } + const identity = githubPrIdentity(`${event.owner}/${event.repo}`, target.prNumber) + if (identity) this.#babysitterUnownedEventOccurrences.delete(identity) if ( this.#mount.resourceSubscriptions && !this.#babysitterResourceSubscriptionUnavailable && @@ -10239,13 +10433,7 @@ export class FactoryLoop implements Factory { return } - let snapshot: PullSnapshot | undefined - try { - snapshot = parsePullSnapshot((await this.#mount.readFile(path)).content, parts.number) - } catch (error) { - this.#logger.debug?.('[factory] babysitter could not read PR snapshot', { path, error: describeError(error).errorMessage }) - return - } + const snapshot = await this.#readBabysitterPrSnapshot(path, parts) if (!snapshot) { return } @@ -10352,6 +10540,105 @@ export class FactoryLoop implements Factory { } } + async #readBabysitterPrSnapshot( + path: string, + parts: { owner: string; repo: string; number: number }, + ): Promise { + const inFlight = this.#babysitterPrSnapshotReads.get(path) + if (inFlight) return inFlight + + const read = this.#readBabysitterPrSnapshotWithRetry(path, parts) + this.#babysitterPrSnapshotReads.set(path, read) + try { + return await read + } finally { + if (this.#babysitterPrSnapshotReads.get(path) === read) { + this.#babysitterPrSnapshotReads.delete(path) + } + } + } + + async #readBabysitterPrSnapshotWithRetry( + path: string, + parts: { owner: string; repo: string; number: number }, + ): Promise { + let lastError = 'unknown PR snapshot read failure' + for (let attempt = 1; attempt <= BABYSITTER_PR_SNAPSHOT_READ_MAX_ATTEMPTS; attempt += 1) { + try { + const snapshot = parsePullSnapshot((await this.#mount.readFile(path)).content, parts.number) + if (!snapshot) throw new Error('PR snapshot did not match its mount path identity') + const deadLetter = this.#babysitterPrSnapshotDeadLetters.get(path) + if (deadLetter) { + this.#babysitterPrSnapshotDeadLetters.delete(path) + this.#increment('babysitterPrSnapshotReadDeadLettersRecovered') + this.#logger.info?.('[factory] recovered dead-lettered PR snapshot', { + path, + repo: deadLetter.repo, + prNumber: deadLetter.prNumber, + priorExhaustions: deadLetter.exhaustions, + }) + } + return snapshot + } catch (error) { + lastError = describeError(error).errorMessage + this.#increment('babysitterPrSnapshotReadFailures') + if (this.#stopping) return undefined + if (attempt < BABYSITTER_PR_SNAPSHOT_READ_MAX_ATTEMPTS) { + const retryDelayMs = BABYSITTER_PR_SNAPSHOT_READ_RETRY_BASE_MS * 2 ** (attempt - 1) + this.#increment('babysitterPrSnapshotReadRetries') + this.#logger.warn?.('[factory] babysitter could not read PR snapshot; retrying', { + path, + attempt, + maxAttempts: BABYSITTER_PR_SNAPSHOT_READ_MAX_ATTEMPTS, + retryDelayMs, + error: lastError, + }) + await this.#clock.sleep(retryDelayMs) + continue + } + } + } + + const previous = this.#babysitterPrSnapshotDeadLetters.get(path) + const deadLetter: BabysitterPrSnapshotDeadLetter = { + path, + repo: `${parts.owner}/${parts.repo}`, + prNumber: parts.number, + attempts: BABYSITTER_PR_SNAPSHOT_READ_MAX_ATTEMPTS, + exhaustions: (previous?.exhaustions ?? 0) + 1, + error: lastError, + failedAtMs: this.#clock.now(), + } + this.#babysitterPrSnapshotDeadLetters.delete(path) + this.#babysitterPrSnapshotDeadLetters.set(path, deadLetter) + if (this.#babysitterPrSnapshotDeadLetters.size > BABYSITTER_PR_SNAPSHOT_DEAD_LETTER_LIMIT) { + const oldest = this.#babysitterPrSnapshotDeadLetters.keys().next().value + if (oldest !== undefined) this.#babysitterPrSnapshotDeadLetters.delete(oldest) + } + this.#increment(previous ? 'babysitterPrSnapshotReadDeadLetterRedrives' : 'babysitterPrSnapshotReadDeadLetters') + this.#logger.error?.('[factory] babysitter PR snapshot read retries exhausted; dead-lettered for reconcile', { + path, + repo: deadLetter.repo, + prNumber: deadLetter.prNumber, + attempts: deadLetter.attempts, + exhaustions: deadLetter.exhaustions, + error: deadLetter.error, + }) + void this.#report({ + type: 'factory.anomaly', + level: 'error', + attributes: { + component: 'babysitter', + operation: 'read_pr_snapshot', + errorCode: 'snapshot_read_exhausted', + retryable: true, + attempt: deadLetter.attempts, + count: 1, + }, + }) + return undefined + } + #inFlightIssueForPrSnapshot(snapshot: PullSnapshot, batch: BatchSnapshot, eventRepo: string): InFlightIssue | undefined { let best: { record: InFlightIssue; score: number } | undefined let ambiguous = false @@ -14747,6 +15034,7 @@ const resolveIssuePrFromMount = async ( candidates.push({ repo, prNumber: pr.number, + matchScore: score, draft: pr.draft, headRef: pr.headRef, headRepo: pr.headRepo, @@ -14833,6 +15121,7 @@ const resolveIssuePrFromGh = async ( candidates.push({ repo, prNumber: pr.number, + matchScore: score, draft: pr.draft, headRef: pr.headRef, headRepo: pr.headRepo, diff --git a/src/types.ts b/src/types.ts index c28a363..8337848 100644 --- a/src/types.ts +++ b/src/types.ts @@ -125,6 +125,8 @@ export interface FactoryLoopHeartbeat { updatedAtMs: number registryPath?: string eventListener?: FactoryEventListenerStatus + /** Daemon-owned operational counters surfaced to out-of-process status clients. */ + counters?: Record } export interface FactoryInFlightRegistryAgent {