From 1e8de2c4568b955419783f77079387dcee527942 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Fri, 14 Aug 2026 17:40:30 +0200 Subject: [PATCH] fix(factory): reconcile readiness claims after restart --- src/config/schema.ts | 2 + src/orchestrator/factory.test.ts | 174 ++++++++++++++++- src/orchestrator/factory.ts | 322 ++++++++++++++++++++++++++++--- src/types.ts | 2 + 4 files changed, 467 insertions(+), 33 deletions(-) diff --git a/src/config/schema.ts b/src/config/schema.ts index eb36d2c..9cccc3d 100644 --- a/src/config/schema.ts +++ b/src/config/schema.ts @@ -40,6 +40,8 @@ const liveSubscriptionSchema = z.object({ pollIntervalMs: z.number().int().min(50).default(5_000), eventLimit: z.number().int().min(1).max(1_000).default(1_000), replaySkewMarginMs: z.number().int().min(0).default(60_000), + /** Independent source-of-truth sweep; live event watermarks remain a latency optimization. */ + reconcileIntervalMs: z.number().int().min(50).default(60_000), }).default({}) const dispatchSchema = z.object({ diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 1873e51..bddb2bf 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -3087,6 +3087,100 @@ describe('FactoryLoop', () => { } }) + it('releases a dead durable GitHub claim and redispatches the issue', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-dead-lifecycle-claim-')) + try { + class MissingDeadAgentsFleet extends RemoteLifecycleFleetClient { + override async release(name: string, reason?: string): Promise { + this.releases.push({ name, reason }) + throw new Error(`agent ${name} not found`) + } + } + const path = githubIssuePath('AgentWorkforce', 'pear', 242) + const content = githubIssueFile(242, { + title: '[factory-e2e] Release dead claims', + labels: ['factory', 'pear', 'factory:in-progress'], + }) + const mount = new FakeMountClient({ [path]: content }) + mount.setSubRoot('/linear/issues', 'absent') + const stateStore = new InMemoryStateStore({ batchSize: 4 }) + const fleet = new MissingDeadAgentsFleet() + const githubWriteback = new RecordingGithubWriteback() + const factory = createFactory(config({ + issueSource: 'github', + loop: { registryPath: join(root, 'registry.json') }, + }), { + mount, + fleet, + stateStore, + triage: new StaticTriage(), + githubWriteback, + probePrGhRunner: async () => ({ stdout: '[]' }), + }) + const issue = parseGithubFactoryIssue(path, content) + const decision = await factory.triageIssue(issue) + const staleSpecs = [ + { ...decision.implementers[0]!, name: 'ar-242-impl-pear' }, + { ...decision.reviewer, name: 'ar-242-review-pear' }, + ] + await stateStore.claimDispatchLifecycle( + 'factory-test', + issueKey(issue), + { + runId: 'crashed-run-242', + issue: { uuid: issue.uuid, key: issue.key, path: issue.path }, + decision, + dryRun: false, + // GitHub applies `factory:in-progress` immediately before this row + // advances to running, so dispatching is the narrow crash window. + phase: 'dispatching', + agents: staleSpecs.map((spec) => ({ + name: spec.name, + tracked: { + spec, + result: { name: spec.name, node: 'sf-mini', locality: 'remote' }, + }, + })), + invocationIds: ['factory:242:implementer', 'factory:242:reviewer'], + updatedAtMs: 0, + }, + 'dead-factory-owner', + 0, + 1, + ) + await stateStore.recordDispatchAttempt('factory-test', issueKey(issue), { + attempts: 1, + inFlight: true, + terminal: false, + backoffUntilMs: 0, + }) + + const report = await factory.runOnce() + + expect(report.dispatched.map((result) => result.issue.key)).toEqual(['242']) + expect(fleet.releases).toEqual([ + { name: 'ar-242-impl-pear', reason: 'orphaned-claim' }, + { name: 'ar-242-review-pear', reason: 'orphaned-claim' }, + ]) + expect(fleet.spawns.map((spawn) => spawn.name)).toEqual([ + 'ar-242-impl-pear', + 'ar-242-review-pear', + ]) + expect(githubWriteback.statuses).toEqual([ + { key: '242', status: 'ready' }, + { key: '242', status: 'in-progress' }, + ]) + expect(factory.status().counters.githubOrphanedLifecycleClaimsReleased).toBe(1) + expect(factory.status().counters.githubOrphanedLifecycleAgentReleaseFailures).toBe(2) + await expect(stateStore.getDispatchLifecycle('factory-test', issueKey(issue))).resolves.toMatchObject({ + phase: 'running', + runId: expect.not.stringMatching(/^crashed-run-242$/u), + }) + } finally { + await rm(root, { recursive: true, force: true }) + } + }) + it('consumes an orphan-recovery readiness exemption after a failed dispatch transition', async () => { const root = await mkdtemp(join(tmpdir(), 'factory-orphan-one-shot-')) try { @@ -3312,7 +3406,7 @@ describe('FactoryLoop', () => { expect(report.dispatched).toEqual([]) expect(report.skipped).toEqual([{ issue: { uuid: 'AgentWorkforce/pear#53', key: '53', path }, - reason: 'already tracked', + reason: 'active dispatch claim or live agent still owns the issue', }]) expect(fleet.spawns.map((spawn) => spawn.name)).toEqual(['ar-53-babysit-pear']) expect(fleet.spawns[0]).toMatchObject({ @@ -3470,7 +3564,7 @@ describe('FactoryLoop', () => { expect(report.dispatched).toEqual([]) expect(report.skipped).toEqual([expect.objectContaining({ issue: expect.objectContaining({ key: '53' }), - reason: 'live state is not ready-for-agent', + reason: 'matching open PR still owns the issue', })]) expect(fleet.spawns).toEqual([]) expect(factory.status().counters.githubOrphanedPullRequestsAdopted).toBeUndefined() @@ -3605,6 +3699,7 @@ describe('FactoryLoop', () => { const fleet = new FakeFleetClient() fleet.hydrateTracked([{ name: 'ar-55-impl-pear' }]) const githubWriteback = new RecordingGithubWriteback() + const infos: unknown[][] = [] const factory = createFactory(config({ issueSource: 'github', loop: { registryPath }, @@ -3614,6 +3709,7 @@ describe('FactoryLoop', () => { triage: new StaticTriage(), githubWriteback, probePrResolver: async () => undefined, + logger: { info: (...args: unknown[]) => infos.push(args) }, }) const report = await factory.runOnce() @@ -3621,6 +3717,14 @@ describe('FactoryLoop', () => { expect(report.dispatched).toEqual([]) expect(githubWriteback.statuses).toEqual([]) expect(factory.status().counters.githubOrphanRecoveriesBlockedActive).toBe(1) + expect(infos).toContainEqual([ + '[factory] readiness reconciliation skipped dispatch', + { + issue: '55', + path, + reason: 'active dispatch claim or live agent still owns the issue', + }, + ]) } finally { await rm(root, { recursive: true, force: true }) } @@ -3850,7 +3954,7 @@ describe('FactoryLoop', () => { expect(factory.status().counters.githubOrphanRecoveryPrProbeFailures).toBe(1) expect(report.skipped).toContainEqual(expect.objectContaining({ issue: expect.objectContaining({ key: '56' }), - reason: 'live state is not ready-for-agent', + reason: 'matching open-PR absence could not be verified', })) } finally { await rm(root, { recursive: true, force: true }) @@ -9280,6 +9384,69 @@ describe('FactoryLoop', () => { await factory.stop() }) + it('dispatches a GitHub issue labeled while Factory was stopped even when its event is below the startup watermark', async () => { + const path = githubIssueCompactPath('AgentWorkforce', 'pear', 240) + const mount = new CountingEventsMount({ + [path]: githubIssueFile(240, { + title: '[factory-e2e] Reconcile readiness independently of event watermarks', + labels: ['factory', 'pear'], + }), + }) + mount.setSubRoot('/linear/issues', 'absent') + // This is the label event the previous daemon consumed before stopping. + // The successor starts above it and therefore cannot rely on redelivery. + mount.emit(changeEvent(path, 'evt-before-restart-240')) + const fleet = new FakeFleetClient() + const factory = createFactory(config({ issueSource: 'github' }), { + mount, + fleet, + triage: new StaticTriage(), + githubWriteback: new RecordingGithubWriteback(), + }) + + await factory.start({ mode: 'live', liveSubscription: { transport: 'subscribe' } }) + + expect(fleet.spawns.map((spawn) => spawn.name)).toEqual([ + 'ar-240-impl-pear', + 'ar-240-review-pear', + ]) + expect(factory.status().counters.liveStartupBackfills).toBe(1) + expect(factory.status().counters.liveReplayEventsSuppressedByWatermark).toBeUndefined() + await factory.stop() + }) + + it('periodically dispatches a readiness-labeled GitHub issue without a live event', async () => { + const path = githubIssueCompactPath('AgentWorkforce', 'pear', 241) + const mount = new CountingEventsMount() + mount.setSubRoot('/linear/issues', 'absent') + const fleet = new FakeFleetClient() + const factory = createFactory(config({ issueSource: 'github' }), { + mount, + fleet, + triage: new StaticTriage(), + githubWriteback: new RecordingGithubWriteback(), + }) + + await factory.start({ + mode: 'live', + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 50 }, + }) + mount.files.set(path, { + content: githubIssueFile(241, { + title: '[factory-e2e] Periodically reconcile readiness', + labels: ['factory', 'pear'], + }), + }) + + await vi.waitFor(() => expect(fleet.spawns.map((spawn) => spawn.name)).toEqual([ + 'ar-241-impl-pear', + 'ar-241-review-pear', + ]), { timeout: 3_000 }) + expect(mount.getEventsCalls).toBe(0) + expect(factory.status().counters.readinessReconcileSweeps).toBeGreaterThanOrEqual(1) + await factory.stop() + }) + it('derives a repo-scoped subscription and startup backfill from the simple hoopsheet config', async () => { class ScopedStartupMount extends CountingEventsMount { readonly listTreePrefixes: string[] = [] @@ -20277,6 +20444,7 @@ describe('FactoryLoop PR babysitter', () => { pollIntervalMs: 50, eventLimit: 1, replaySkewMarginMs: 60_000, + reconcileIntervalMs: 60_000, } const factory = createFactory(babysitterConfig({ liveSubscription }), { mount, diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 13a25b0..f581edc 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -147,7 +147,9 @@ type GithubOrphanRecoveryContext = { activeIssueIdentities: Set onlineAgentNames: Set legacyUnownedAgentsByIssue: Map + orphanedLifecycleClaimsByIssue: Map } +type GithubOrphanRecoveryResult = { recovered: boolean; reason?: string } type BabysitterWakeKind = | 'pull-request-state' | 'review' @@ -531,6 +533,9 @@ export class FactoryLoop implements Factory { #liveHeartbeatRefresh?: Promise #liveHeartbeatLastWriteMs = 0 #stoppingHeartbeatRefreshActive = false + #readinessReconcileTimer?: ReturnType + #readinessReconcileInFlight?: Promise + #readinessReconcileIntervalMs = 60_000 readonly #liveEventQueue: ChangeEvent[] = [] #liveEventDrainScheduled = false #liveEventDrainActive = false @@ -901,6 +906,7 @@ export class FactoryLoop implements Factory { await this.#rearmSlackReplyWatchers() await this.#drainReadyClarificationWake() await this.#rearmGithubIssueCommentWatchers() + this.#scheduleReadinessReconcile() this.#scheduleCompletionSweep(0) return } catch (error) { @@ -957,8 +963,11 @@ export class FactoryLoop implements Factory { this.#dispatchLifecycleOwnershipWaitLogged.clear() if (this.#completionSweepTimer) clearTimeout(this.#completionSweepTimer) this.#completionSweepTimer = undefined + if (this.#readinessReconcileTimer) clearTimeout(this.#readinessReconcileTimer) + this.#readinessReconcileTimer = undefined if (this.#previewSweepTimer) clearTimeout(this.#previewSweepTimer) this.#previewSweepTimer = undefined + await this.#readinessReconcileInFlight await this.#previewSweepInFlight this.#stoppingHeartbeatRefreshActive = await this.#stopLiveHeartbeat('stopping') try { @@ -1112,6 +1121,7 @@ export class FactoryLoop implements Factory { ): Promise { const options = this.#liveOptions(overrides) this.#liveTransport = options.transport + this.#readinessReconcileIntervalMs = options.reconcileIntervalMs this.#liveConnectStartedAtMs = this.#clock.now() this.#liveReplaySkewMarginMs = options.replaySkewMarginMs const highWatermark = await this.#currentEventHighWatermark() @@ -1252,6 +1262,7 @@ export class FactoryLoop implements Factory { pollIntervalMs: overrides.pollIntervalMs ?? this.#config.liveSubscription.pollIntervalMs, eventLimit: overrides.eventLimit ?? this.#config.liveSubscription.eventLimit, replaySkewMarginMs: overrides.replaySkewMarginMs ?? this.#config.liveSubscription.replaySkewMarginMs, + reconcileIntervalMs: overrides.reconcileIntervalMs ?? this.#config.liveSubscription.reconcileIntervalMs, } } @@ -1279,6 +1290,48 @@ export class FactoryLoop implements Factory { } } + #scheduleReadinessReconcile(delayMs = this.#readinessReconcileIntervalMs): void { + if ( + !this.#started || + this.#stopping || + this.#readinessReconcileTimer || + this.#readinessReconcileInFlight + ) return + this.#readinessReconcileTimer = setTimeout(() => { + this.#readinessReconcileTimer = undefined + if (!this.#started || this.#stopping) return + const sweep = this.#reconcileReadyIssues() + this.#readinessReconcileInFlight = sweep + void sweep.finally(() => { + if (this.#readinessReconcileInFlight === sweep) { + this.#readinessReconcileInFlight = undefined + } + if (this.#started && !this.#stopping) this.#scheduleReadinessReconcile() + }) + }, delayMs) + this.#readinessReconcileTimer.unref?.() + } + + async #reconcileReadyIssues(): Promise { + this.#increment('readinessReconcileSweeps') + this.#logger.info?.('[factory] periodic readiness reconciliation started', { + intervalMs: this.#readinessReconcileIntervalMs, + }) + try { + const report = await this.runOnce() + this.#logger.info?.('[factory] periodic readiness reconciliation completed', { + candidates: report.pulled.length, + dispatched: report.dispatched.length, + skipped: report.skipped.length, + }) + } catch (error) { + this.#increment('readinessReconcileErrors') + this.#logger.warn?.('[factory] periodic readiness reconciliation failed; retry remains scheduled', { + error: describeError(error).errorMessage, + }) + } + } + #scheduleLivePoll(delayMs: number, options: FactoryLiveSubscriptionOptions): void { if (this.#livePollTimer || !this.#started) return this.#livePollTimer = setTimeout(() => { @@ -1755,6 +1808,14 @@ export class FactoryLoop implements Factory { const triaged: TriageDecision[] = [] const dispatched: DispatchResult[] = [] const skipped: IterationReport['skipped'] = [] + const recordSkip = (entry: IterationReport['skipped'][number]): void => { + skipped.push(entry) + this.#logger.info?.('[factory] readiness reconciliation skipped dispatch', { + issue: entry.issue.key, + path: entry.issue.path, + reason: entry.reason, + }) + } let lastReadyReadProgressAtMs = this.#clock.now() let readyIssueReads = 0 @@ -1806,7 +1867,6 @@ export class FactoryLoop implements Factory { const mayRecoverGithubOrphan = !wasReady && !dryRun && issueSource === 'github' && - Boolean(orphanRecovery) && Boolean(requiredLabel) && Boolean(labels?.has(requiredLabel)) && Boolean(labels?.has('factory:in-progress')) && @@ -1814,28 +1874,32 @@ export class FactoryLoop implements Factory { if (!mayRecoverGithubOrphan) { const dispatchBlock = await this.#dispatchBlockReason(issue) if (dispatchBlock) { - skipped.push({ issue: issueRef(issue), reason: dispatchBlock }) + recordSkip({ issue: issueRef(issue), reason: dispatchBlock }) continue } } + const orphanResult = mayRecoverGithubOrphan + ? await this.#reconcileOrphanedGithubInProgress(issue, orphanRecovery, dryRun) + : { recovered: false } + const recoveredOrphan = orphanResult.recovered const batch = await this.#batch() if (batch.isInFlight(issue) || batch.isQueued(issue)) { - skipped.push({ issue: issueRef(issue), reason: 'already tracked' }) + recordSkip({ issue: issueRef(issue), reason: orphanResult.reason ?? 'already tracked' }) continue } - - const recoveredOrphan = mayRecoverGithubOrphan && - await this.#reconcileOrphanedGithubInProgress(issue, orphanRecovery, dryRun) if (!wasReady && !recoveredOrphan) { if (mayRecoverGithubOrphan) { const dispatchBlock = await this.#dispatchBlockReason(issue) if (dispatchBlock) { - skipped.push({ issue: issueRef(issue), reason: dispatchBlock }) + recordSkip({ issue: issueRef(issue), reason: dispatchBlock }) continue } } - skipped.push({ issue: issueRef(issue), reason: 'live state is not ready-for-agent' }) + recordSkip({ + issue: issueRef(issue), + reason: orphanResult.reason ?? 'live state is not ready-for-agent', + }) continue } const recoveredIdentity = recoveredOrphan ? githubIssueRefIdentity(issueRef(issue)) : undefined @@ -1843,18 +1907,18 @@ export class FactoryLoop implements Factory { if (recoveredOrphan) { const dispatchBlock = await this.#dispatchBlockReason(issue) if (dispatchBlock) { - skipped.push({ issue: issueRef(issue), reason: dispatchBlock }) + recordSkip({ issue: issueRef(issue), reason: dispatchBlock }) continue } } if (!isInFactoryScope(issue, this.#config.safety)) { - skipped.push({ issue: issueRef(issue), reason: 'not factory-e2e scope' }) + recordSkip({ issue: issueRef(issue), reason: 'not factory-e2e scope' }) continue } if (!isDispatchableIssue(issue)) { - skipped.push({ issue: issueRef(issue), reason: 'not reconciled real Linear issue' }) + recordSkip({ issue: issueRef(issue), reason: 'not reconciled real Linear issue' }) continue } @@ -1865,7 +1929,7 @@ export class FactoryLoop implements Factory { result = await this.dispatch(decision, { dryRun }) } catch (error) { if (!(error instanceof LiveDispatchStateChangedError)) throw error - skipped.push({ issue: decision.issue, reason: 'live state changed during dispatch' }) + recordSkip({ issue: decision.issue, reason: 'live state changed during dispatch' }) this.#logger.info?.('[factory] skipped issue whose live state changed during dispatch', { issue: decision.issue.key, }) @@ -1877,7 +1941,7 @@ export class FactoryLoop implements Factory { : result.hold?.kind === 'dependency' ? `parked on dependencies: ${result.hold.blockers?.join(', ') ?? 'unresolved dependency'}` : 'queued or escalated' - skipped.push({ issue: decision.issue, reason }) + recordSkip({ issue: decision.issue, reason }) } else { dispatched.push(result) } @@ -1924,15 +1988,41 @@ export class FactoryLoop implements Factory { const onlineAgents = new Set(roster.agents.map((agent) => agent.name)) const activeIssueIdentities = new Set() const legacyUnownedAgentsByIssue = new Map() - for (const [, lifecycle] of lifecycles) { - if (isTerminalDispatchLifecycle(lifecycle)) continue - const identity = githubIssueRefIdentity(lifecycle.issue) - if (identity) activeIssueIdentities.add(identity) - } + const orphanedLifecycleClaimsByIssue = new Map() for (const [, waiting] of waitingClarifications) { const identity = githubIssueRefIdentity(waiting.issue) if (identity) activeIssueIdentities.add(identity) } + for (const [key, lifecycle] of lifecycles) { + const identity = githubIssueRefIdentity(lifecycle.issue) + if (!identity) continue + if (isTerminalDispatchLifecycle(lifecycle)) { + if (!activeIssueIdentities.has(identity)) { + orphanedLifecycleClaimsByIssue.set(identity, { key, lifecycle }) + } + continue + } + const activeAgents = lifecycle.agents.filter((agent) => agent.releasedAtMs === undefined) + const hasLiveAgent = activeAgents.some((agent) => onlineAgents.has(agent.name)) + const exitRecoveryActive = activeAgents.some((agent) => this.#agentExitsInFlight.has(agent.name)) + const dispatchCallActive = this.#dispatchInFlight.has(issueKey(lifecycle.issue)) + // The provider's `factory:in-progress` transition happens immediately + // before the durable lifecycle advances from dispatching to running. + // A crash can therefore leave any nonterminal phase behind while the + // external claim survives. The caller still verifies provider status, + // open-PR absence, a second roster, and the lifecycle lease before it + // releases this orphan-shaped row. + if ( + !activeIssueIdentities.has(identity) && + !hasLiveAgent && + !exitRecoveryActive && + !dispatchCallActive + ) { + orphanedLifecycleClaimsByIssue.set(identity, { key, lifecycle }) + } else { + activeIssueIdentities.add(identity) + } + } for (const agent of registry?.agents ?? []) { if (!onlineAgents.has(agent.name) || !agent.issue) continue const identity = githubIssueRefIdentity(agent.issue) @@ -1954,6 +2044,7 @@ export class FactoryLoop implements Factory { activeIssueIdentities, onlineAgentNames: onlineAgents, legacyUnownedAgentsByIssue, + orphanedLifecycleClaimsByIssue, } } catch (error) { this.#increment('githubOrphanRecoveryContextFailures') @@ -1968,8 +2059,10 @@ export class FactoryLoop implements Factory { issue: LinearIssue, context: GithubOrphanRecoveryContext | undefined, dryRun: boolean, - ): Promise { - if (dryRun || !context || !isGithubIssue(issue)) return false + ): Promise { + if (dryRun) return { recovered: false, reason: 'dry run does not release an in-progress claim' } + if (!context) return { recovered: false, reason: 'orphan-recovery safety context is unavailable' } + if (!isGithubIssue(issue)) return { recovered: false, reason: 'issue is not GitHub-native' } const labels = new Set(issue.labels.map((label) => label.trim().toLowerCase())) const required = this.#config.safety.requireLabel.trim().toLowerCase() if ( @@ -1977,7 +2070,7 @@ export class FactoryLoop implements Factory { !labels.has(required) || !labels.has('factory:in-progress') || labels.has('factory:human-review') - ) return false + ) return { recovered: false, reason: 'issue is not an orphan-recovery candidate' } const identity = githubIssueRefIdentity(issueRef(issue)) const legacyUnownedAgents = identity @@ -1993,13 +2086,13 @@ export class FactoryLoop implements Factory { ) ) { this.#increment('githubOrphanRecoveriesBlockedActive') - return false + return { recovered: false, reason: 'active dispatch claim or live agent still owns the issue' } } const getProviderStatus = this.#githubWriteback.getIssueStatus if (!getProviderStatus) { this.#increment('githubOrphanRecoveryStatusLookupUnavailable') - return false + return { recovered: false, reason: 'provider-authoritative issue status lookup is unavailable' } } let providerStatus: GithubIssueStatus | undefined try { @@ -2010,11 +2103,16 @@ export class FactoryLoop implements Factory { issue: issue.key, error: describeError(error).errorMessage, }) - return false + return { recovered: false, reason: 'provider-authoritative issue status could not be verified' } } if (!providerStatus || providerStatus === 'human-review') { this.#increment('githubOrphanRecoveriesBlockedProviderStatus') - return false + return { + recovered: false, + reason: providerStatus === 'human-review' + ? 'provider-authoritative issue status is human-review' + : 'provider-authoritative issue status is unavailable', + } } let openPr: ResolvedIssuePr | undefined @@ -2026,7 +2124,7 @@ export class FactoryLoop implements Factory { issue: issue.key, error: describeError(error).errorMessage, }) - return false + return { recovered: false, reason: 'matching open-PR absence could not be verified' } } if (openPr) { let adopted = false @@ -2049,7 +2147,12 @@ export class FactoryLoop implements Factory { repo: openPr.repo, prNumber: openPr.prNumber, }) - return false + return { + recovered: false, + reason: adopted + ? 'matching open PR was adopted' + : 'matching open PR still owns the issue', + } } // A pre-durable local Factory may have left live, registry-proven workers @@ -2058,7 +2161,12 @@ export class FactoryLoop implements Factory { // and workers instead of redispatching duplicate agents. if (legacyUnownedAgents.length > 0) { this.#increment('githubOrphanRecoveriesBlockedActive') - return false + return { recovered: false, reason: 'legacy registry-proven agents still own the issue' } + } + + const lifecycleClaim = context.orphanedLifecycleClaimsByIssue.get(identity) + if (lifecycleClaim && !await this.#releaseOrphanedGithubLifecycle(issue, identity, lifecycleClaim)) { + return { recovered: false, reason: 'orphaned durable lifecycle claim could not be safely released' } } try { @@ -2068,20 +2176,174 @@ export class FactoryLoop implements Factory { // A crashed dispatch may leave its durable attempt marked in-flight even // after every agent and lifecycle disappeared. Only clear that stale bit // after all provider, agent, lifecycle, and open-PR safety checks pass. - await this.#clearDispatchInFlight(issue) + const attempt = await this.#state.getDispatchAttempts(this.#workspaceId, issueStateKey(issue)) + if (attempt?.terminal) { + await this.#state.recordDispatchAttempt(this.#workspaceId, issueStateKey(issue), { + attempts: 0, + inFlight: false, + terminal: false, + backoffUntilMs: 0, + }) + } else { + await this.#clearDispatchInFlight(issue) + } this.#reconciledGithubInProgress.add(identity) this.#increment('githubOrphanedInProgressRecovered') this.#logger.warn?.('[factory] recovered orphaned GitHub in-progress issue for redispatch', { issue: issue.key, path: issue.path, }) - return true + return { recovered: true } } catch (error) { this.#increment('githubOrphanRecoveryWritebackFailures') this.#logger.warn?.('[factory] failed to clear orphaned GitHub lifecycle status; preserving in-progress issue', { issue: issue.key, error: describeError(error).errorMessage, }) + return { recovered: false, reason: 'orphaned provider lifecycle status could not be cleared' } + } + } + + async #releaseOrphanedGithubLifecycle( + issue: LinearIssue, + identity: string, + candidate: { key: string; lifecycle: DispatchLifecycle }, + ): Promise { + let key = candidate.key + let lifecycle = await this.#state.getDispatchLifecycle(this.#workspaceId, key) + if (!lifecycle) return true + if ( + lifecycle.runId !== candidate.lifecycle.runId || + githubIssueRefIdentity(lifecycle.issue) !== identity + ) { + this.#logger.info?.('[factory] preserved orphan-shaped GitHub claim because its durable lifecycle changed', { + issue: issue.key, + lifecycleKey: key, + }) + return false + } + + const activeAgents = lifecycle.agents.filter((agent) => agent.releasedAtMs === undefined) + const roster = await this.#fleet.roster() + const online = new Set(roster.agents.map((agent) => agent.name)) + if ( + activeAgents.some((agent) => online.has(agent.name)) || + roster.agents.some((agent) => githubAgentNameMatchesIssue(agent.name, issue)) + ) { + this.#increment('githubOrphanRecoveriesBlockedActive') + return false + } + + let epoch: number | undefined + if (!isTerminalDispatchLifecycle(lifecycle)) { + epoch = this.#dispatchLifecycleEpochs.get(key) + if (epoch !== undefined) { + const renewed = await this.#state.renewDispatchLifecycle( + this.#workspaceId, + key, + this.#dispatchLifecycleOwner, + epoch, + this.#clock.now(), + DISPATCH_LIFECYCLE_LEASE_MS, + ) + if (!renewed) { + this.#dispatchLifecycleEpochs.delete(key) + return false + } + } else { + const claim = await this.#state.claimDispatchLifecycle( + this.#workspaceId, + key, + lifecycle, + this.#dispatchLifecycleOwner, + this.#clock.now(), + DISPATCH_LIFECYCLE_LEASE_MS, + ) + if (!claim.acquired || !claim.lease) { + this.#logger.info?.('[factory] preserved orphan-shaped GitHub claim because another publisher still owns its lease', { + issue: issue.key, + owner: claim.lifecycle.lease?.owner, + leaseUntilMs: claim.lifecycle.lease?.leaseUntilMs, + }) + return false + } + key = claim.key ?? key + lifecycle = claim.lifecycle + epoch = claim.lease.epoch + this.#dispatchLifecycleEpochs.set(key, epoch) + } + } + + try { + for (const agent of lifecycle.agents.filter((entry) => entry.releasedAtMs === undefined)) { + try { + await this.#fleet.release(agent.name, 'orphaned-claim') + } catch (error) { + // The roster was checked twice before fencing this lifecycle, so a + // missing/dead worker is the expected crash-recovery shape. Do not + // let a control-plane "agent not found" response make the durable + // claim immortal; a failed fresh spawn remains visible to the normal + // dispatch retry machinery. + this.#increment('githubOrphanedLifecycleAgentReleaseFailures') + this.#logger.warn?.('[factory] dead claim agent release failed; clearing fenced lifecycle anyway', { + issue: issue.key, + agent: agent.name, + error: describeError(error).errorMessage, + }) + } + this.#fleet.markAgentTerminal?.(agent.name, 'orphaned-claim') + } + if (epoch !== undefined && !await this.#state.renewDispatchLifecycle( + this.#workspaceId, + key, + this.#dispatchLifecycleOwner, + epoch, + this.#clock.now(), + DISPATCH_LIFECYCLE_LEASE_MS, + )) { + this.#dispatchLifecycleEpochs.delete(key) + return false + } + + const retryTimer = this.#dispatchLifecycleRetryTimers.get(key) + if (retryTimer) clearTimeout(retryTimer) + this.#dispatchLifecycleRetryTimers.delete(key) + const batch = await this.#batch() + batch.abandon(lifecycle.issue) + await this.#state.clearDispatchLifecycle(this.#workspaceId, key) + await this.#state.clearBabysitterSession(this.#workspaceId, issueKey(lifecycle.issue)) + this.#dispatchLifecycleEpochs.delete(key) + this.#abandonedDispatchReasons.delete(key) + this.#resolveDispatchTerminalWaiters(lifecycle.issue) + await this.#writeInFlightRegistry().catch((error: unknown) => { + this.#logger.warn?.('[factory] failed to rewrite registry after orphaned claim release', { + issue: issue.key, + error: describeError(error).errorMessage, + }) + }) + this.#increment('githubOrphanedLifecycleClaimsReleased') + this.#logger.warn?.('[factory] released orphaned GitHub dispatch claim with no live agents', { + issue: issue.key, + lifecycleKey: key, + runId: lifecycle.runId, + agents: activeAgents.map((agent) => agent.name), + }) + return true + } catch (error) { + if (epoch !== undefined) { + await this.#state.releaseDispatchLifecycleLease( + this.#workspaceId, + key, + this.#dispatchLifecycleOwner, + epoch, + ).catch(() => undefined) + this.#dispatchLifecycleEpochs.delete(key) + } + this.#logger.warn?.('[factory] failed to release orphaned GitHub dispatch claim; preserving it', { + issue: issue.key, + lifecycleKey: key, + error: describeError(error).errorMessage, + }) return false } } diff --git a/src/types.ts b/src/types.ts index c28a363..407b769 100644 --- a/src/types.ts +++ b/src/types.ts @@ -94,6 +94,8 @@ export interface FactoryLiveSubscriptionOptions { pollIntervalMs: number eventLimit: number replaySkewMarginMs: number + /** Periodic source-of-truth readiness reconciliation, independent of event cursors/watermarks. */ + reconcileIntervalMs: number } /**