diff --git a/.agentworkforce/features/critical-paths.md b/.agentworkforce/features/critical-paths.md index 22e43c4..4b3712e 100644 --- a/.agentworkforce/features/critical-paths.md +++ b/.agentworkforce/features/critical-paths.md @@ -56,6 +56,8 @@ For GitHub-native issues, every collision-prone role is repository-qualified and Batch capacity must queue excess work and promote the next issue when the current record completes. Dispatch failures must back off and stop at `dispatch.maxAttempts`. Local internal dispatch expands supported home-relative clone paths, infers cwd only for one remote-matching repository, and fails before spawn when a configured checkout is missing, non-git, or not its worktree root. +Critical briefing delivery is fail-closed. A confirmed delivery failure retries the persisted message, but exhausted initial delivery or reinjection must atomically persist an abort intent for the exact dispatch run, fence side effects to its lease owner, release the complete team, publish a correlated issue-visible abort, and only then restore Ready for Agent plus retry eligibility. Partial release or writeback failures retain the in-progress fence and batch slot for durable cleanup takeover; stale failures from older runs cannot abort a newer team. The lifecycle contract spans `src/orchestrator/factory.ts`, `src/orchestrator/batch-tracker.ts`, `src/ports/state.ts`, `src/state/file-state-store.ts`, and provider writeback implementations. + **What breaks if this fails:** agent execution, batch fairness, retry safety, issue lifecycle accuracy. --- @@ -237,7 +239,7 @@ Internal execution must reuse an operator-owned broker when present, otherwise s | Repo-qualified identity + composite issue key | Equal-number issues share agents, PR caches, or lifecycle state | `src/triage/agent-names.ts`, `src/orchestrator/factory.ts` | | Clone expansion/inference/preflight | Local work spawns in a missing or wrong checkout | `src/config/schema.ts`, `src/config/local-clone-paths.ts`, `src/cli/fleet.ts` | | Live high-water + replay suppression | Missed or duplicate dispatch | `src/orchestrator/factory.ts`, `src/subscriptions/event-client.ts` | -| Critical message confirmation | Agents start without receiving tasks | `src/orchestrator/factory.ts`, `src/fleet/internal-fleet-client.ts` | +| Critical message confirmation + abort cleanup | Agents start without receiving tasks, or dirty lifecycle state blocks a safe retry | `src/orchestrator/factory.ts`, `src/fleet/internal-fleet-client.ts`, `src/ports/state.ts`, `src/state/file-state-store.ts`, `src/writeback/github.ts` | | Durable relay lifecycle + owner epoch | Duplicate remote team/PR, leaked capacity, or stalled release | `src/orchestrator/factory.ts`, `src/ports/state.ts`, `src/state/file-state-store.ts` | | Remote head publication + receipt recovery | In-flight issue stalls or publishes a stale/local branch | `src/orchestrator/factory.ts`, `src/mount/relayfile-github-connection-write.ts` | | Canonical babysitter routing + durable wake | Wrong PR wakes, event loss, or duplicate branch mutation | `src/orchestrator/factory.ts`, `src/ports/state.ts`, `src/state/file-state-store.ts` | diff --git a/.agentworkforce/features/manifest.yaml b/.agentworkforce/features/manifest.yaml index 54757e9..20a355a 100644 --- a/.agentworkforce/features/manifest.yaml +++ b/.agentworkforce/features/manifest.yaml @@ -555,8 +555,8 @@ categories: - id: dispatch-critical-delivery-retry name: Critical Delivery Retry api: FleetClient.onDeliveryFailed() - description: Persist critical messages by event ID and retry them when broker delivery fails - location: src/orchestrator/factory.ts, src/state/in-memory-state-store.ts + description: Persist critical messages by event ID, retry confirmed failures, and on exhausted initial delivery or reinjection durably fence and release the full team before an idempotent issue-visible abort restores retryable source state + location: src/orchestrator/factory.ts, src/orchestrator/batch-tracker.ts, src/ports/state.ts, src/state/in-memory-state-store.ts, src/state/file-state-store.ts, src/writeback/github.ts verify_tier: 4 - id: dispatch-agent-resume diff --git a/.agentworkforce/features/verify/procedures.md b/.agentworkforce/features/verify/procedures.md index bda35ec..af4a024 100644 --- a/.agentworkforce/features/verify/procedures.md +++ b/.agentworkforce/features/verify/procedures.md @@ -241,6 +241,7 @@ Against a disposable fixture/provider issue, verify: 1. implementer(s) and reviewer/workflow spawn once; 2. task delivery receives `delivery_injected` acknowledgement before input submission; 3. a simulated delivery failure retries the persisted critical message; + exhausted initial delivery and exhausted reinjection must converge on one owner-fenced abort that releases the full team, emits at most one correlated `cleanup-pending` marker and at most one final `aborted` marker, restores source readiness only after cleanup acknowledgement, and survives retries, release/writeback failure, plus remote-owner restart without duplicating either marker or respawning workers; 4. an interrupted session resumes once; 5. batch overflow queues and completion promotes the next issue; 6. registry data contains agent/session/PID or remote invocation/node identity; diff --git a/src/orchestrator/batch-tracker.ts b/src/orchestrator/batch-tracker.ts index 21a985a..259b9a8 100644 --- a/src/orchestrator/batch-tracker.ts +++ b/src/orchestrator/batch-tracker.ts @@ -8,6 +8,8 @@ export interface TrackedAgent { } export interface InFlightIssue { + /** Exact durable dispatch attempt; absent for local-only dispatches. */ + runId?: string issue: IssueRef decision: TriageDecision dryRun: boolean @@ -181,6 +183,7 @@ export class BatchTracker { const existing = this.#inFlight.get(key) if (existing) return existing const restored: InFlightIssue = { + runId: record.runId, issue: { ...record.issue }, decision: structuredClone(record.decision), dryRun: record.dryRun, diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index d5db5d9..deb1e3f 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -25,7 +25,7 @@ import { type WorkflowRunnerInput, } from '../index' import { changeEventPath } from './factory' -import type { ChangeEvent, EventPage, GithubConnectionWrite, GithubIssueStatus, GithubPublishPullRequestInput, GithubWriteback, LinearWriteback, ProviderSyncStatus, SlackWriteback, SpawnInput, SpawnResult } from '../ports' +import type { ChangeEvent, DispatchLifecycle, EventPage, GithubConnectionWrite, GithubIssueStatus, GithubPublishPullRequestInput, GithubWriteback, LinearWriteback, ProviderSyncStatus, SlackWriteback, SpawnInput, SpawnResult } from '../ports' import { FakeFleetClient, FakeMountClient } from '../testing' import type { CloseProbePrInput, GithubMergeGatePort, GithubMergeGateVerdict, GithubMergeInput, LinearIssue } from '../index' import { BatchTracker, issueKey } from './batch-tracker' @@ -238,7 +238,9 @@ class StaticTriage implements TriageEngine { class RecordingGithubWriteback implements GithubWriteback { readonly comments: Array<{ key: string; body: string }> = [] readonly statuses: Array<{ key: string; status: GithubIssueStatus }> = [] + readonly clearedStatuses: string[] = [] readonly closes: Array<{ key: string; body: string }> = [] + beforeClearStatus?: (issue: LinearIssue) => void async postComment(issue: LinearIssue, body: string): Promise { this.comments.push({ key: issue.key, body }) @@ -248,6 +250,11 @@ class RecordingGithubWriteback implements GithubWriteback { this.statuses.push({ key: issue.key, status }) } + async clearStatus(issue: LinearIssue): Promise { + this.beforeClearStatus?.(issue) + this.clearedStatuses.push(issue.key) + } + async closeIssue(issue: LinearIssue, body: string): Promise { this.closes.push({ key: issue.key, body }) } @@ -259,6 +266,22 @@ class FailingGithubCommentWriteback extends RecordingGithubWriteback { } } +class TransientGithubClearStatusWriteback extends RecordingGithubWriteback { + clearAttempts = 0 + + override async clearStatus(issue: LinearIssue): Promise { + this.clearAttempts += 1 + if (this.clearAttempts === 1) throw new Error('GitHub status clear unavailable') + await super.clearStatus(issue) + } +} + +class ProviderAwareTransientGithubClearStatusWriteback extends TransientGithubClearStatusWriteback { + async hasCommentMarker(issue: LinearIssue, marker: string): Promise { + return this.comments.some((comment) => comment.key === issue.key && comment.body.includes(marker)) + } +} + class ImmediateGithubReplyWriteback extends RecordingGithubWriteback { constructor( private readonly mount: FakeMountClient, @@ -592,6 +615,174 @@ class TransientRemoteReleaseFleetClient extends RemoteLifecycleFleetClient { } } +class RemoteInitialInjectionAndReleaseFailureFleetClient extends RemoteLifecycleFleetClient { + override async waitForInjected( + input: Parameters[0], + _opts?: Parameters[1], + ): ReturnType { + this.messages.push(input) + throw new Error(`recipient unavailable: ${input.to}`) + } + + override async release(name: string): Promise { + throw new Error(`relay release unavailable for ${name}`) + } +} + +class RemoteInitialInjectionFailureFleetClient extends RemoteLifecycleFleetClient { + override async waitForInjected( + input: Parameters[0], + _opts?: Parameters[1], + ): ReturnType { + this.messages.push(input) + throw new Error(`recipient unavailable: ${input.to}`) + } +} + +class RemoteReinjectionFailingFleetClient extends RemoteLifecycleFleetClient { + readonly attemptsByTarget = new Map() + + constructor(readonly failingTarget: string) { + super() + } + + override async waitForInjected( + input: Parameters[0], + opts?: Parameters[1], + ): ReturnType { + const attempts = (this.attemptsByTarget.get(input.to) ?? 0) + 1 + this.attemptsByTarget.set(input.to, attempts) + if (input.to === this.failingTarget && attempts > 1) { + this.messages.push(input) + throw new Error(`recipient unavailable: ${input.to}`) + } + return await super.waitForInjected(input, opts) + } +} + +class RejectFirstAbortingSaveStateStore extends InMemoryStateStore { + abortingSaveRejections = 0 + + override async saveDispatchLifecycle( + ...args: Parameters + ): Promise { + const [, key, owner, epoch, , lifecycle] = args + if (lifecycle.phase === 'aborting' && this.abortingSaveRejections === 0) { + this.abortingSaveRejections += 1 + await super.releaseDispatchLifecycleLease(args[0], key, owner, epoch) + return false + } + return await super.saveDispatchLifecycle(...args) + } +} + +class RejectFirstAbandonedSaveStateStore extends InMemoryStateStore { + abandonedSaveRejections = 0 + dispatchAttemptInFlightAtRejection?: boolean + + override async saveDispatchLifecycle( + ...args: Parameters + ): Promise { + const [, key, owner, epoch, , lifecycle] = args + if (lifecycle.phase === 'abandoned' && this.abandonedSaveRejections === 0) { + this.abandonedSaveRejections += 1 + this.dispatchAttemptInFlightAtRejection = ( + await this.getDispatchAttempts(args[0], lifecycle.issue.key) ?? + await this.getDispatchAttempts(args[0], issueKey(lifecycle.issue)) + )?.inFlight + await super.releaseDispatchLifecycleLease(args[0], key, owner, epoch) + return false + } + return await super.saveDispatchLifecycle(...args) + } +} + +class ThrowFirstAbortIntentStateStore extends InMemoryStateStore { + abortIntentFailures = 0 + + override async beginCriticalDeliveryAbort( + ...args: Parameters + ) { + if (this.abortIntentFailures === 0) { + this.abortIntentFailures += 1 + throw new Error('transient abort intent persistence failure') + } + return await super.beginCriticalDeliveryAbort(...args) + } +} + +class BlockingThrowFirstAbortIntentStateStore extends InMemoryStateStore { + readonly abortIntentStarted: Promise + #signalAbortIntentStarted!: () => void + readonly #abortIntentAllowed: Promise + #allowAbortIntent!: () => void + abortIntentFailures = 0 + abortIntentAttempts = 0 + + constructor(options: ConstructorParameters[0]) { + super(options) + this.abortIntentStarted = new Promise((resolve) => { + this.#signalAbortIntentStarted = resolve + }) + this.#abortIntentAllowed = new Promise((resolve) => { + this.#allowAbortIntent = resolve + }) + } + + allowAbortIntent(): void { + this.#allowAbortIntent() + } + + override async beginCriticalDeliveryAbort( + ...args: Parameters + ) { + this.abortIntentAttempts += 1 + if (this.abortIntentFailures === 0) { + this.#signalAbortIntentStarted() + await this.#abortIntentAllowed + this.abortIntentFailures += 1 + throw new Error('transient abort intent persistence failure during stop') + } + return await super.beginCriticalDeliveryAbort(...args) + } +} + +class BlockingCriticalAbortFleetClient extends RemoteInitialInjectionFailureFleetClient { + readonly abortReleaseStarted: Promise + #signalAbortReleaseStarted!: () => void + readonly #abortReleaseAllowed: Promise + #allowAbortRelease!: () => void + #blocked = false + disposeCalls = 0 + + constructor() { + super() + this.abortReleaseStarted = new Promise((resolve) => { + this.#signalAbortReleaseStarted = resolve + }) + this.#abortReleaseAllowed = new Promise((resolve) => { + this.#allowAbortRelease = resolve + }) + } + + allowAbortRelease(): void { + this.#allowAbortRelease() + } + + override async release(name: string, reason?: string): Promise { + if (!this.#blocked && reason?.startsWith('critical-delivery-failed:')) { + this.#blocked = true + this.#signalAbortReleaseStarted() + await this.#abortReleaseAllowed + } + await super.release(name, reason) + } + + override async dispose(): Promise { + this.disposeCalls += 1 + } +} + class BlockingClarificationReleaseFleetClient extends RemoteLifecycleFleetClient { readonly clarificationReleaseStarted: Promise #signalClarificationReleaseStarted!: () => void @@ -831,6 +1022,46 @@ class LagThenInjectedFleetClient extends FakeFleetClient { } } +class ReinjectionFailingFleetClient extends FakeFleetClient { + readonly attemptsByTarget = new Map() + + constructor(readonly failingTarget: string) { + super() + } + + override async waitForInjected( + input: Parameters[0], + opts?: Parameters[1], + ): ReturnType { + const attempts = (this.attemptsByTarget.get(input.to) ?? 0) + 1 + this.attemptsByTarget.set(input.to, attempts) + if (input.to === this.failingTarget && attempts > 1) { + this.messages.push(input) + throw new Error(`recipient unavailable: ${input.to}`) + } + return await super.waitForInjected(input, opts) + } +} + +class TransientAbortReleaseFleetClient extends ReinjectionFailingFleetClient { + releaseFailures = 0 + + override async release(name: string, reason?: string): Promise { + if (reason?.startsWith('critical-delivery-failed:') && this.releaseFailures === 0) { + this.releaseFailures += 1 + throw new Error(`transient release failure for ${name}`) + } + await super.release(name, reason) + } +} + +class ExitDuringAbortFleetClient extends ReinjectionFailingFleetClient { + override async release(name: string, reason?: string): Promise { + this.emitAgentExit(name, 'release-exit-race') + await super.release(name, reason) + } +} + class UnresolvedPidFleetClient extends FakeFleetClient { async resolveAgentPid(_name: string): Promise<{ status: 'unresolved' }> { return { status: 'unresolved' } @@ -4442,7 +4673,7 @@ describe('FactoryLoop', () => { expect(reports[1]?.error).toBeUndefined() expect(factory.status().inFlight).toEqual([]) expect(factory.status().counters.loopIterationFailures).toBe(1) - expect(factory.status().counters.loopDispatchFailureHandoffsReaped).toBe(2) + expect(factory.status().counters.criticalDeliveryAbortsCompleted).toBe(1) expect([...alive]).toEqual([]) expect(killed.map((entry) => entry.pid)).toEqual(expect.arrayContaining([7_602, 7_601, 7_604, 7_603])) const heartbeat = await readFactoryLoopHeartbeat(heartbeatPath) @@ -4519,7 +4750,7 @@ describe('FactoryLoop', () => { const firstFailedKill = killed.findIndex((entry) => failedPids.has(entry.pid) && entry.signal !== 0) expect(firstFailedKill).toBeGreaterThanOrEqual(0) expect(firstHealthyKill).toBeGreaterThan(firstFailedKill) - expect(factory.status().counters.loopDispatchFailureHandoffsReaped).toBe(2) + expect(factory.status().counters.criticalDeliveryAbortsCompleted).toBe(1) } finally { await rm(root, { recursive: true, force: true }) } @@ -7211,6 +7442,8 @@ describe('FactoryLoop', () => { expect(errors).toHaveLength(1) expect(errors[0]).toMatchObject({ issue: { key: 'AR-8' } }) + expect(fleet.releases).toEqual([]) + expect(factory.status().inFlight).toMatchObject([{ key: 'AR-8' }]) await factory.stop() }) @@ -7366,6 +7599,737 @@ describe('FactoryLoop', () => { expect(fleet.deliveryEvents.at(-1)).toEqual({ kind: 'input', name: 'ar-64-review', data: '\r' }) }) + it('aborts the whole team and restores Linear readiness when the initial implementer briefing cannot be delivered', async () => { + const mount = new FakeMountClient({ [issuePath(68)]: issueFile(68) }) + const clock = new ManualClock() + const fleet = new InjectFailingPidFleetClient([]) + const factory = createFactory(config(), { mount, fleet, triage: new StaticTriage(), clock }) + + await expect(factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(68), issueFile(68))))) + .rejects.toThrow('Critical task injection failed to ar-68-impl-pear') + + expect(fleet.releases).toEqual([ + { name: 'ar-68-impl-pear', reason: 'critical-delivery-failed:ar-68-impl-pear' }, + { name: 'ar-68-review', reason: 'critical-delivery-failed:ar-68-impl-pear' }, + ]) + expect(factory.status().inFlight).toEqual([]) + expect(factory.status().counters.criticalDeliveryAbortsCompleted).toBe(1) + expect(mount.writes + .filter((write) => 'stateId' in (write.content as Record)) + .map((write) => (write.content as { stateId: string }).stateId)).toEqual([implementing, ready]) + expect(mount.writes.some((write) => JSON.stringify(write.content).includes('factory-critical-delivery-abort'))).toBe(true) + }) + + it('aborts both workers when the initial reviewer briefing fails after the implementer was injected', async () => { + const mount = new FakeMountClient({ [issuePath(69)]: issueFile(69) }) + const fleet = new SelectiveInjectFailingPidFleetClient([], 'ar-69-review') + const factory = createFactory(config(), { mount, fleet, triage: new StaticTriage(), clock: new ManualClock() }) + + await expect(factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(69), issueFile(69))))) + .rejects.toThrow('Critical task injection failed to ar-69-review') + + expect(fleet.inputs).toEqual([{ name: 'ar-69-impl-pear', data: '\r' }]) + expect(fleet.releases.map((release) => release.name)).toEqual(['ar-69-impl-pear', 'ar-69-review']) + expect(factory.status().inFlight).toEqual([]) + }) + + it('aborts exactly once when confirmed critical reinjection exhausts and ignores a duplicate delivery failure', async () => { + const mount = new FakeMountClient({ [issuePath(70)]: issueFile(70) }) + const fleet = new ReinjectionFailingFleetClient('ar-70-impl-pear') + const factory = createFactory(config(), { mount, fleet, triage: new StaticTriage(), clock: new ManualClock() }) + + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(70), issueFile(70)))) + fleet.emitDeliveryFailed({ to: 'ar-70-impl-pear', msgId: 'fake-1', reason: 'dead-lettered' }) + fleet.emitDeliveryFailed({ to: 'ar-70-impl-pear', msgId: 'fake-1', reason: 'duplicate' }) + + await vi.waitFor(() => expect(factory.status().counters.criticalDeliveryAbortsCompleted).toBe(1)) + expect(fleet.releases).toEqual([ + { name: 'ar-70-impl-pear', reason: 'critical-delivery-failed:ar-70-impl-pear' }, + { name: 'ar-70-review', reason: 'critical-delivery-failed:ar-70-impl-pear' }, + ]) + expect(factory.status().inFlight).toEqual([]) + expect(fleet.resumes).toEqual([]) + }) + + it('aborts the complete team when confirmed reviewer reinjection exhausts', async () => { + const mount = new FakeMountClient({ [issuePath(75)]: issueFile(75) }) + const fleet = new ReinjectionFailingFleetClient('ar-75-review') + const factory = createFactory(config(), { mount, fleet, triage: new StaticTriage(), clock: new ManualClock() }) + + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(75), issueFile(75)))) + fleet.emitDeliveryFailed({ to: 'ar-75-review', msgId: 'fake-2', reason: 'dead-lettered' }) + + await vi.waitFor(() => expect(factory.status().counters.criticalDeliveryAbortsCompleted).toBe(1)) + expect(fleet.releases.map((release) => release.name)).toEqual(['ar-75-impl-pear', 'ar-75-review']) + expect(factory.status().inFlight).toEqual([]) + }) + + it('suppresses release-driven agent exits so abort cleanup cannot restart the team', async () => { + const mount = new FakeMountClient({ [issuePath(76)]: issueFile(76) }) + const fleet = new ExitDuringAbortFleetClient('ar-76-impl-pear') + const factory = createFactory(config(), { mount, fleet, triage: new StaticTriage(), clock: new ManualClock() }) + + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(76), issueFile(76)))) + fleet.emitDeliveryFailed({ to: 'ar-76-impl-pear', msgId: 'fake-1', reason: 'dead-lettered' }) + + await vi.waitFor(() => expect(factory.status().counters.criticalDeliveryAbortsCompleted).toBe(1)) + expect(fleet.resumes).toEqual([]) + expect(fleet.spawns.map((spawn) => spawn.name)).toEqual(['ar-76-impl-pear', 'ar-76-review']) + expect(factory.status().counters.criticalDeliveryAbortExitsSuppressed).toBe(2) + }) + + it('publishes a visible GitHub abort and removes the in-progress lifecycle label only after team release', async () => { + const path = githubIssuePath('AgentWorkforce', 'pear', 71) + const file = githubIssueFile(71, { labels: ['factory'] }) + const mount = new FakeMountClient({ [path]: file }) + const fleet = new InjectFailingPidFleetClient([]) + const githubWriteback = new RecordingGithubWriteback() + githubWriteback.beforeClearStatus = () => { + expect(fleet.releases.map((release) => release.name)).toEqual([ + 'ar-71-impl-pear', + 'ar-71-review-pear', + ]) + } + const factory = createFactory(config({ issueSource: 'github' }), { + mount, + fleet, + triage: new StaticTriage(), + githubWriteback, + clock: new ManualClock(), + }) + + const issue = parseGithubFactoryIssue(path, file) + await expect(factory.dispatch(await factory.triageIssue(issue))) + .rejects.toThrow('Critical task injection failed to ar-71-impl-pear') + + expect(fleet.releases.map((release) => release.name)).toEqual(['ar-71-impl-pear', 'ar-71-review-pear']) + expect(githubWriteback.statuses).toEqual([{ key: '71', status: 'in-progress' }]) + expect(githubWriteback.clearedStatuses).toEqual(['71']) + expect(githubWriteback.comments.at(-1)?.body).toContain('Factory aborted this dispatch') + expect(githubWriteback.comments.at(-1)?.body).toContain('factory-critical-delivery-abort') + }) + + it('retains the abort fence until a transient team release failure is acknowledged', async () => { + const mount = new FakeMountClient({ [issuePath(72)]: issueFile(72) }) + const fleet = new TransientAbortReleaseFleetClient('ar-72-impl-pear') + const factory = createFactory(config(), { mount, fleet, triage: new StaticTriage(), clock: new ManualClock() }) + + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(72), issueFile(72)))) + fleet.emitDeliveryFailed({ to: 'ar-72-impl-pear', msgId: 'fake-1', reason: 'dead-lettered' }) + + await vi.waitFor(() => expect(fleet.releaseFailures).toBe(1)) + expect(factory.status().inFlight).toMatchObject([{ key: 'AR-72' }]) + expect(mount.writes + .filter((write) => 'stateId' in (write.content as Record)) + .map((write) => (write.content as { stateId: string }).stateId)).toEqual([implementing]) + + await vi.waitFor(() => expect(factory.status().counters.criticalDeliveryAbortsCompleted).toBe(1), { timeout: 3_000 }) + expect(factory.status().inFlight).toEqual([]) + expect(mount.writes + .filter((write) => 'stateId' in (write.content as Record)) + .map((write) => (write.content as { stateId: string }).stateId)).toEqual([implementing, ready]) + }) + + it('reconciles an idempotent GitHub abort notice across restart before retrying failed status publication', async () => { + const path = githubIssuePath('AgentWorkforce', 'pear', 73) + const file = githubIssueFile(73, { labels: ['factory'] }) + const mount = new FakeMountClient({ [path]: file }) + const stateStore = new InMemoryStateStore({ batchSize: 2 }) + const firstFleet = new RemoteInitialInjectionFailureFleetClient() + const githubWriteback = new ProviderAwareTransientGithubClearStatusWriteback() + const firstFactory = createFactory(config({ issueSource: 'github' }), { + mount, + fleet: firstFleet, + triage: new StaticTriage(), + githubWriteback, + stateStore, + clock: new ManualClock(), + }) + + const issue = parseGithubFactoryIssue(path, file) + const decision = await firstFactory.triageIssue(issue) + await expect(firstFactory.dispatch(decision)) + .rejects.toThrow('Critical task injection failed to ar-73-impl-pear') + expect(firstFactory.status().inFlight).toMatchObject([{ key: '73' }]) + await expect(stateStore.getDispatchLifecycle('factory-test', issueKey(decision.issue))).resolves.toMatchObject({ + phase: 'aborting', + }) + expect(githubWriteback.comments.filter((comment) => comment.body.includes('factory-critical-delivery-abort'))).toHaveLength(1) + expect(githubWriteback.clearAttempts).toBe(1) + await firstFactory.stop() + + const takeoverFleet = new RemoteLifecycleFleetClient() + const takeover = createFactory(config({ issueSource: 'github' }), { + mount, + fleet: takeoverFleet, + triage: new StaticTriage(), + githubWriteback, + stateStore, + clock: new ManualClock(), + }) + await takeover.start({ mode: 'dispatch-owner' }) + + await vi.waitFor(async () => { + await expect(stateStore.getDispatchLifecycle('factory-test', issueKey(decision.issue))).resolves.toMatchObject({ + phase: 'abandoned', + }) + }, { timeout: 3_000 }) + expect(githubWriteback.clearAttempts).toBe(2) + expect(githubWriteback.comments.filter((comment) => comment.body.includes('factory-critical-delivery-abort'))).toHaveLength(1) + expect(githubWriteback.clearedStatuses).toEqual(['73']) + expect(takeoverFleet.spawns).toEqual([]) + expect(takeoverFleet.resumes).toEqual([]) + expect(takeoverFleet.releases).toEqual([]) + await takeover.stop() + }) + + it('persists remote abort cleanup for restart takeover instead of resuming an unbriefed team', async () => { + const mount = new FakeMountClient({ [issuePath(74)]: issueFile(74) }) + const stateStore = new InMemoryStateStore({ batchSize: 2 }) + const firstFleet = new RemoteInitialInjectionAndReleaseFailureFleetClient() + const firstFactory = createFactory(config(), { + mount, + fleet: firstFleet, + triage: new StaticTriage(), + stateStore, + clock: new ManualClock(), + }) + const decision = await firstFactory.triageIssue(parseLinearIssue(issuePath(74), issueFile(74))) + + await expect(firstFactory.dispatch(decision)).rejects.toThrow('Critical task injection failed') + await expect(stateStore.getDispatchLifecycle('factory-test', issueKey(decision.issue))).resolves.toMatchObject({ + phase: 'aborting', + releaseReason: 'critical-delivery-failed:ar-74-impl-pear', + }) + expect(firstFactory.status().inFlight).toMatchObject([{ key: 'AR-74' }]) + await firstFactory.stop() + + const takeoverFleet = new RemoteLifecycleFleetClient() + const takeover = createFactory(config(), { + mount, + fleet: takeoverFleet, + triage: new StaticTriage(), + stateStore, + clock: new ManualClock(), + }) + await takeover.start({ mode: 'dispatch-owner' }) + + await vi.waitFor(async () => { + await expect(stateStore.getDispatchLifecycle('factory-test', issueKey(decision.issue))).resolves.toMatchObject({ + phase: 'abandoned', + }) + }, { timeout: 3_000 }) + expect(takeoverFleet.spawns).toEqual([]) + expect(takeoverFleet.resumes).toEqual([]) + expect(takeoverFleet.releases.map((release) => release.name).sort()).toEqual([ + 'ar-74-impl-pear', + 'ar-74-review', + ]) + expect(mount.writes + .filter((write) => 'stateId' in (write.content as Record)) + .map((write) => (write.content as { stateId: string }).stateId)).toEqual([implementing, ready]) + await takeover.stop() + }) + + it('keeps exact-run abort intent durable when the first aborting save loses its lease, then takeover cleans up', async () => { + const mount = new FakeMountClient({ [issuePath(77)]: issueFile(77) }) + const stateStore = new RejectFirstAbortingSaveStateStore({ batchSize: 2 }) + const firstFleet = new RemoteInitialInjectionFailureFleetClient() + const firstFactory = createFactory(config(), { + mount, + fleet: firstFleet, + triage: new StaticTriage(), + stateStore, + clock: new ManualClock(), + }) + const decision = await firstFactory.triageIssue(parseLinearIssue(issuePath(77), issueFile(77))) + + await expect(firstFactory.dispatch(decision)).rejects.toThrow('Critical task injection failed') + const aborting = await stateStore.getDispatchLifecycle('factory-test', issueKey(decision.issue)) + expect(aborting).toMatchObject({ + phase: 'aborting', + releaseReason: 'critical-delivery-failed:ar-77-impl-pear', + }) + expect(stateStore.abortingSaveRejections).toBe(1) + expect(firstFleet.releases.map((release) => release.name)).toEqual(['ar-77-impl-pear']) + expect(mount.writes + .filter((write) => 'stateId' in (write.content as Record)) + .map((write) => (write.content as { stateId: string }).stateId)).toEqual([implementing]) + firstFleet.emitAgentExit('ar-77-review', 'late-after-lease-loss') + await flush() + expect(firstFactory.status().counters.criticalDeliveryAbortExitsSuppressed).toBeUndefined() + expect(firstFleet.resumes).toEqual([]) + await firstFactory.stop() + + const takeoverFleet = new RemoteLifecycleFleetClient() + const takeover = createFactory(config(), { + mount, + fleet: takeoverFleet, + triage: new StaticTriage(), + stateStore, + clock: new ManualClock(), + }) + await takeover.start({ mode: 'dispatch-owner' }) + + await vi.waitFor(async () => { + await expect(stateStore.getDispatchLifecycle('factory-test', issueKey(decision.issue))).resolves.toMatchObject({ + runId: aborting?.runId, + phase: 'abandoned', + }) + }, { timeout: 3_000 }) + expect(takeoverFleet.spawns).toEqual([]) + expect(takeoverFleet.resumes).toEqual([]) + expect(takeoverFleet.releases.map((release) => release.name).sort()).toEqual([ + 'ar-77-impl-pear', + 'ar-77-review', + ]) + expect(mount.writes + .filter((write) => 'stateId' in (write.content as Record)) + .map((write) => (write.content as { stateId: string }).stateId)).toEqual([implementing, ready]) + await takeover.stop() + }) + + it('retains consumed async failure evidence across abort-save lease loss and takeover', async () => { + const mount = new FakeMountClient({ [issuePath(78)]: issueFile(78) }) + const stateStore = new RejectFirstAbortingSaveStateStore({ batchSize: 2 }) + const firstFleet = new RemoteReinjectionFailingFleetClient('ar-78-impl-pear') + const firstFactory = createFactory(config(), { + mount, + fleet: firstFleet, + triage: new StaticTriage(), + stateStore, + clock: new ManualClock(), + }) + const decision = await firstFactory.triageIssue(parseLinearIssue(issuePath(78), issueFile(78))) + + await firstFactory.dispatch(decision) + firstFleet.emitDeliveryFailed({ to: 'ar-78-impl-pear', msgId: 'fake-1', reason: 'dead-lettered' }) + await vi.waitFor(async () => { + await expect(stateStore.getDispatchLifecycle('factory-test', issueKey(decision.issue))).resolves.toMatchObject({ + phase: 'aborting', + releaseReason: 'critical-delivery-failed:ar-78-impl-pear', + }) + expect(stateStore.abortingSaveRejections).toBe(1) + }) + expect(mount.writes + .filter((write) => 'stateId' in (write.content as Record)) + .map((write) => (write.content as { stateId: string }).stateId)).toEqual([implementing]) + await firstFactory.stop() + + const takeoverFleet = new RemoteLifecycleFleetClient() + const takeover = createFactory(config(), { + mount, + fleet: takeoverFleet, + triage: new StaticTriage(), + stateStore, + clock: new ManualClock(), + }) + await takeover.start({ mode: 'dispatch-owner' }) + + await vi.waitFor(async () => { + await expect(stateStore.getDispatchLifecycle('factory-test', issueKey(decision.issue))).resolves.toMatchObject({ + phase: 'abandoned', + }) + }, { timeout: 3_000 }) + expect(takeoverFleet.spawns).toEqual([]) + expect(takeoverFleet.resumes).toEqual([]) + expect(takeoverFleet.releases.map((release) => release.name).sort()).toEqual([ + 'ar-78-impl-pear', + 'ar-78-review', + ]) + expect(mount.writes + .filter((write) => 'stateId' in (write.content as Record)) + .map((write) => (write.content as { stateId: string }).stateId)).toEqual([implementing, ready]) + await takeover.stop() + }) + + it('retries an unexpected abort failure while retaining the fail-closed exit fence', async () => { + const mount = new FakeMountClient({ [issuePath(98)]: issueFile(98) }) + const stateStore = new ThrowFirstAbortIntentStateStore({ batchSize: 2 }) + const fleet = new RemoteInitialInjectionFailureFleetClient() + const factory = createFactory(config(), { + mount, + fleet, + triage: new StaticTriage(), + stateStore, + clock: new ManualClock(), + }) + const decision = await factory.triageIssue(parseLinearIssue(issuePath(98), issueFile(98))) + + await expect(factory.dispatch(decision)).rejects.toThrow('transient abort intent persistence failure') + expect(stateStore.abortIntentFailures).toBe(1) + expect(factory.status().counters.criticalDeliveryAbortUnexpectedFailures).toBe(1) + fleet.emitAgentExit('ar-98-review', 'exit-during-abort-retry') + await flush() + expect(factory.status().counters.criticalDeliveryAbortExitsSuppressed).toBe(1) + expect(fleet.resumes).toEqual([]) + + await vi.waitFor(() => expect(factory.status().counters.criticalDeliveryAbortsCompleted).toBe(1), { timeout: 3_000 }) + await expect(stateStore.getDispatchLifecycle('factory-test', issueKey(decision.issue))).resolves.toMatchObject({ + phase: 'abandoned', + }) + expect(factory.status().inFlight).toEqual([]) + await factory.stop() + }) + + it('drains an active critical abort before fleet disposal without starting queued work', async () => { + const mount = new FakeMountClient({ + [issuePath(100)]: issueFile(100), + [issuePath(101)]: issueFile(101), + }) + const stateStore = new InMemoryStateStore({ batchSize: 1 }) + const fleet = new BlockingCriticalAbortFleetClient() + const factory = createFactory(config({ batchSize: 1 }), { + mount, + fleet, + triage: new StaticTriage(), + stateStore, + clock: new ManualClock(), + }) + const first = await factory.triageIssue(parseLinearIssue(issuePath(100), issueFile(100))) + const queued = await factory.triageIssue(parseLinearIssue(issuePath(101), issueFile(101))) + const firstDispatch = factory.dispatch(first).catch((error: unknown) => error) + + await fleet.abortReleaseStarted + await expect(factory.dispatch(queued)).resolves.toMatchObject({ + issue: { key: 'AR-101' }, + agents: [], + }) + const stopping = factory.stop() + await flush() + expect(fleet.disposeCalls).toBe(0) + + fleet.allowAbortRelease() + await expect(firstDispatch).resolves.toBeInstanceOf(Error) + await stopping + + expect(fleet.disposeCalls).toBe(1) + expect(fleet.spawns.map((spawn) => spawn.name)).toEqual(['ar-100-impl-pear', 'ar-100-review']) + await expect(stateStore.getDispatchLifecycle('factory-test', issueKey(first.issue))).resolves.toMatchObject({ + phase: 'abandoned', + }) + await expect(stateStore.getDispatchLifecycle('factory-test', issueKey(queued.issue))).resolves.toMatchObject({ + phase: 'queued', + }) + }) + + it('preserves consumed async abort evidence when the first intent write fails during stop', async () => { + const mount = new FakeMountClient({ [issuePath(105)]: issueFile(105) }) + const stateStore = new BlockingThrowFirstAbortIntentStateStore({ batchSize: 2 }) + const firstFleet = new RemoteReinjectionFailingFleetClient('ar-105-impl-pear') + const firstFactory = createFactory(config(), { + mount, + fleet: firstFleet, + triage: new StaticTriage(), + stateStore, + clock: new ManualClock(), + }) + const decision = await firstFactory.triageIssue(parseLinearIssue(issuePath(105), issueFile(105))) + + await firstFactory.dispatch(decision) + firstFleet.emitDeliveryFailed({ + to: 'ar-105-impl-pear', + msgId: 'fake-1', + reason: 'dead-lettered-during-stop', + }) + await stateStore.abortIntentStarted + + let stopSettled = false + const stopping = firstFactory.stop().finally(() => { stopSettled = true }) + await flush() + expect(stopSettled).toBe(false) + stateStore.allowAbortIntent() + await stopping + + expect(stateStore.abortIntentFailures).toBe(1) + expect(stateStore.abortIntentAttempts).toBe(2) + expect(firstFactory.status().counters.criticalDeliveryAbortStopIntentsPreserved).toBe(1) + await expect(stateStore.consumeCritical('factory-test', 'fake-1')).resolves.toBeUndefined() + await expect(stateStore.getDispatchLifecycle('factory-test', issueKey(decision.issue))).resolves.toMatchObject({ + runId: expect.any(String), + phase: 'aborting', + releaseReason: 'critical-delivery-failed:ar-105-impl-pear', + }) + expect(firstFleet.releases).toEqual([]) + expect(firstFleet.resumes).toEqual([]) + expect(mount.writes + .filter((write) => 'stateId' in (write.content as Record)) + .map((write) => (write.content as { stateId: string }).stateId)).toEqual([implementing]) + + const takeoverFleet = new RemoteLifecycleFleetClient() + const takeover = createFactory(config(), { + mount, + fleet: takeoverFleet, + triage: new StaticTriage(), + stateStore, + clock: new ManualClock(), + }) + await takeover.start({ mode: 'dispatch-owner' }) + + await vi.waitFor(async () => { + await expect(stateStore.getDispatchLifecycle('factory-test', issueKey(decision.issue))).resolves.toMatchObject({ + phase: 'abandoned', + }) + }, { timeout: 3_000 }) + expect(takeoverFleet.spawns).toEqual([]) + expect(takeoverFleet.resumes).toEqual([]) + expect(takeoverFleet.releases.map((release) => release.name).sort()).toEqual([ + 'ar-105-impl-pear', + 'ar-105-review', + ]) + expect(mount.writes + .filter((write) => 'stateId' in (write.content as Record)) + .map((write) => (write.content as { stateId: string }).stateId)).toEqual([implementing, ready]) + await takeover.stop() + }) + + it('preserves consumed abort evidence when stop cancels an already-scheduled retry', async () => { + const mount = new FakeMountClient({ [issuePath(106)]: issueFile(106) }) + const stateStore = new ThrowFirstAbortIntentStateStore({ batchSize: 2 }) + const firstFleet = new RemoteReinjectionFailingFleetClient('ar-106-impl-pear') + const firstFactory = createFactory(config(), { + mount, + fleet: firstFleet, + triage: new StaticTriage(), + stateStore, + clock: new ManualClock(), + }) + const decision = await firstFactory.triageIssue(parseLinearIssue(issuePath(106), issueFile(106))) + + await firstFactory.dispatch(decision) + firstFleet.emitDeliveryFailed({ + to: 'ar-106-impl-pear', + msgId: 'fake-1', + reason: 'dead-lettered-before-stop', + }) + await vi.waitFor(() => expect(stateStore.abortIntentFailures).toBe(1)) + await expect(stateStore.getDispatchLifecycle('factory-test', issueKey(decision.issue))).resolves.toMatchObject({ + phase: 'running', + }) + + await firstFactory.stop() + expect(firstFactory.status().counters.criticalDeliveryAbortStopIntentsPreserved).toBe(1) + await expect(stateStore.getDispatchLifecycle('factory-test', issueKey(decision.issue))).resolves.toMatchObject({ + phase: 'aborting', + releaseReason: 'critical-delivery-failed:ar-106-impl-pear', + }) + expect(firstFleet.releases).toEqual([]) + + const takeoverFleet = new RemoteLifecycleFleetClient() + const takeover = createFactory(config(), { + mount, + fleet: takeoverFleet, + triage: new StaticTriage(), + stateStore, + clock: new ManualClock(), + }) + await takeover.start({ mode: 'dispatch-owner' }) + await vi.waitFor(async () => { + await expect(stateStore.getDispatchLifecycle('factory-test', issueKey(decision.issue))).resolves.toMatchObject({ + phase: 'abandoned', + }) + }, { timeout: 3_000 }) + expect(takeoverFleet.spawns).toEqual([]) + expect(takeoverFleet.resumes).toEqual([]) + expect(takeoverFleet.releases.map((release) => release.name).sort()).toEqual([ + 'ar-106-impl-pear', + 'ar-106-review', + ]) + await takeover.stop() + }) + + it('releases the local exit fence when the final abandoned save loses ownership', async () => { + const mount = new FakeMountClient({ [issuePath(99)]: issueFile(99) }) + const stateStore = new RejectFirstAbandonedSaveStateStore({ batchSize: 2 }) + const firstFleet = new RemoteInitialInjectionFailureFleetClient() + const firstFactory = createFactory(config(), { + mount, + fleet: firstFleet, + triage: new StaticTriage(), + stateStore, + clock: new ManualClock(), + }) + const decision = await firstFactory.triageIssue(parseLinearIssue(issuePath(99), issueFile(99))) + + await expect(firstFactory.dispatch(decision)).rejects.toThrow('Critical task injection failed') + expect(stateStore.abandonedSaveRejections).toBe(1) + expect(stateStore.dispatchAttemptInFlightAtRejection).toBe(false) + await expect(stateStore.getDispatchLifecycle('factory-test', issueKey(decision.issue))).resolves.toMatchObject({ + phase: 'aborting', + releaseReason: 'critical-delivery-failed:ar-99-impl-pear', + }) + expect(mount.writes + .filter((write) => 'stateId' in (write.content as Record)) + .map((write) => (write.content as { stateId: string }).stateId)).toEqual([implementing, ready]) + firstFleet.emitAgentExit('ar-99-review', 'late-after-final-save-lease-loss') + await flush() + expect(firstFactory.status().counters.criticalDeliveryAbortExitsSuppressed).toBeUndefined() + expect(firstFleet.resumes).toEqual([]) + await firstFactory.stop() + + const takeoverFleet = new RemoteLifecycleFleetClient() + const takeover = createFactory(config(), { + mount, + fleet: takeoverFleet, + triage: new StaticTriage(), + stateStore, + clock: new ManualClock(), + }) + await takeover.start({ mode: 'dispatch-owner' }) + await vi.waitFor(async () => { + await expect(stateStore.getDispatchLifecycle('factory-test', issueKey(decision.issue))).resolves.toMatchObject({ + phase: 'abandoned', + }) + }, { timeout: 3_000 }) + expect(takeoverFleet.spawns).toEqual([]) + expect(takeoverFleet.resumes).toEqual([]) + expect(takeoverFleet.releases).toEqual([]) + await takeover.stop() + }) + + it('ignores a delayed critical failure record from an older dispatch run', async () => { + const mount = new FakeMountClient({ [issuePath(79)]: issueFile(79) }) + const stateStore = new InMemoryStateStore({ batchSize: 2 }) + const fleet = new RemoteLifecycleFleetClient() + const factory = createFactory(config(), { + mount, + fleet, + triage: new StaticTriage(), + stateStore, + clock: new ManualClock(), + }) + const decision = await factory.triageIssue(parseLinearIssue(issuePath(79), issueFile(79))) + + await factory.dispatch(decision) + await stateStore.recordCritical('factory-test', 'stale-critical-event', { + issue: decision.issue, + input: fleet.messages[0]!, + runId: 'older-dispatch-run', + }) + fleet.emitDeliveryFailed({ + to: 'ar-79-impl-pear', + msgId: 'stale-critical-event', + reason: 'delayed-old-run-event', + }) + await vi.waitFor(() => expect(factory.status().counters.criticalDeliveryFailuresIgnoredStaleRun).toBe(1)) + + expect(fleet.releases).toEqual([]) + await expect(stateStore.getDispatchLifecycle('factory-test', issueKey(decision.issue))).resolves.toMatchObject({ + phase: 'running', + }) + await factory.stop() + }) + + it.each([ + { phase: 'complete' as const, seededBatch: false }, + { phase: 'abandoned' as const, seededBatch: false }, + { phase: 'complete' as const, seededBatch: true }, + ])( + 'does not mutate batch capacity for a late exact-run failure after durable $phase (seeded=$seededBatch)', + async ({ phase, seededBatch }) => { + const issueNumber = phase === 'complete' ? (seededBatch ? 102 : 96) : 97 + const mount = new FakeMountClient({ + [issuePath(issueNumber)]: issueFile(issueNumber), + ...(seededBatch ? { + [issuePath(103)]: issueFile(103), + [issuePath(104)]: issueFile(104), + } : {}), + }) + const stateStore = new InMemoryStateStore({ batchSize: seededBatch ? 1 : 2 }) + const fleet = new RemoteLifecycleFleetClient() + const factory = createFactory(config({ batchSize: seededBatch ? 1 : 2 }), { + mount, + fleet, + triage: new StaticTriage(), + stateStore, + clock: new ManualClock(), + }) + const decision = await factory.triageIssue(parseLinearIssue(issuePath(issueNumber), issueFile(issueNumber))) + const runId = `terminal-run-${phase}` + const specs = [...decision.implementers, decision.reviewer] + const lifecycle: DispatchLifecycle = { + runId, + issue: decision.issue, + decision, + dryRun: false, + phase: 'running', + agents: specs.map((spec) => ({ + name: spec.name, + tracked: { + spec, + result: { name: spec.name, node: 'sf-mini', locality: 'remote' }, + }, + })), + invocationIds: [], + updatedAtMs: 0, + } + const key = issueKey(decision.issue) + const claim = await stateStore.claimDispatchLifecycle( + 'factory-test', + key, + lifecycle, + 'completed-owner', + 0, + 10_000, + ) + expect(claim).toMatchObject({ acquired: true, lease: { owner: 'completed-owner', epoch: 1 } }) + await expect(stateStore.saveDispatchLifecycle( + 'factory-test', + key, + 'completed-owner', + 1, + 1, + { ...claim.lifecycle, phase }, + )).resolves.toBe(true) + const target = decision.implementers[0]!.name + await stateStore.recordCritical('factory-test', `late-${phase}`, { + issue: decision.issue, + input: { to: target, text: 'late terminal briefing', from: 'factory', data: { issue: decision.issue } }, + runId, + }) + const stateBatch = await stateStore.getBatch('factory-test') + if (seededBatch) { + const occupied = await factory.triageIssue(parseLinearIssue(issuePath(103), issueFile(103))) + const queued = await factory.triageIssue(parseLinearIssue(issuePath(104), issueFile(104))) + expect(stateBatch.start(occupied, false)).toBeDefined() + expect(stateBatch.queue(queued, false)).toBe(true) + } + + await factory.start({ mode: 'dispatch-owner' }) + fleet.emitDeliveryFailed({ to: target, msgId: `late-${phase}`, reason: 'late-terminal-event' }) + await vi.waitFor(() => expect(factory.status().counters.criticalDeliveryFailuresIgnoredTerminalRun).toBe(1)) + + await expect(stateStore.getDispatchLifecycle('factory-test', key)).resolves.toMatchObject({ runId, phase }) + await expect(stateStore.consumeCritical('factory-test', `late-${phase}`)).resolves.toBeUndefined() + const batch = await stateStore.getBatch('factory-test') + expect(batch.size).toBe(seededBatch ? 1 : 0) + expect(batch.inFlight.map((record) => record.issue.key)).toEqual(seededBatch ? ['AR-103'] : []) + expect(batch.queued.map((record) => record.issue.key)).toEqual(seededBatch ? ['AR-104'] : []) + expect(batch.canStart()).toBe(!seededBatch) + expect(factory.status().inFlight.map((issue) => issue.key)).toEqual(seededBatch ? ['AR-103'] : []) + expect(fleet.spawns).toEqual([]) + expect(fleet.resumes).toEqual([]) + expect(fleet.releases).toEqual([]) + expect(fleet.messages).toEqual([]) + expect(fleet.inputs).toEqual([]) + expect(mount.writes).toEqual([]) + + fleet.emitDeliveryFailed({ to: target, msgId: `late-${phase}`, reason: 'duplicate-terminal-event' }) + await new Promise((resolve) => setTimeout(resolve, 1_100)) + await expect(stateStore.getDispatchLifecycle('factory-test', key)).resolves.toMatchObject({ runId, phase }) + expect(factory.status().counters.criticalDeliveryFailuresIgnoredTerminalRun).toBe(1) + expect(factory.status().inFlight.map((issue) => issue.key)).toEqual(seededBatch ? ['AR-103'] : []) + expect((await stateStore.getBatch('factory-test')).size).toBe(seededBatch ? 1 : 0) + expect((await stateStore.getBatch('factory-test')).queued.map((record) => record.issue.key)) + .toEqual(seededBatch ? ['AR-104'] : []) + expect(fleet.spawns).toEqual([]) + expect(fleet.resumes).toEqual([]) + expect(fleet.releases).toEqual([]) + expect(mount.writes).toEqual([]) + await factory.stop() + }, + ) + it('emits error and rejects when writeback verification fails', async () => { const mount = new FakeMountClient({ [issuePath(9)]: issueFile(9) }) mount.setConfirmWrite(issuePath(9), 'failed') diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 8c07b34..9340e0a 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -156,6 +156,12 @@ class ClarificationWakeLeaseLostError extends Error {} class ClarificationQuestionDeliveryLeaseLostError extends Error {} class GithubEscalationReconciliationUnavailableError extends Error {} class GithubEscalationPostAmbiguousError extends Error {} +class CriticalTaskInjectionError extends Error { + constructor(readonly target: string, cause: unknown) { + super(`Critical task injection failed to ${target}: ${describeError(cause).errorMessage}`, { cause }) + this.name = 'CriticalTaskInjectionError' + } +} type GithubEscalationReconciliation = 'found' | 'absent' | 'unavailable' class ClarificationWakeStoppedError extends Error {} @@ -306,6 +312,16 @@ export class FactoryLoop implements Factory { readonly #dispatchTerminalWaiters = new Map void>>() readonly #dispatchLifecycleRetryTimers = new Map>() readonly #dispatchLifecycleDrives = new Set>() + readonly #deliveryFailureInFlight = new Set>() + readonly #criticalDeliveryAborts = new Map>() + readonly #criticalDeliveryAborting = new Set() + readonly #criticalDeliveryAbortRetryTimers = new Map>() + readonly #criticalDeliveryAbortRetryInputs = new Map() + readonly #criticalDeliveryNotices = new Set() #dispatchLifecycleRenewTimer?: ReturnType #clarificationSweepTimer?: ReturnType #clarificationSweepDueAtMs?: number @@ -650,15 +666,35 @@ export class FactoryLoop implements Factory { async stop(): Promise { this.#started = false this.#stopping = true + // Stop accepting new failure callbacks before draining callbacks that may + // already be between critical-record consumption and abort registration. + this.#offDeliveryFailed?.() + this.#offDeliveryFailed = undefined if (this.#dispatchLifecycleRenewTimer) clearInterval(this.#dispatchLifecycleRenewTimer) this.#dispatchLifecycleRenewTimer = undefined for (const timer of this.#dispatchLifecycleRetryTimers.values()) clearTimeout(timer) this.#dispatchLifecycleRetryTimers.clear() + const pendingCriticalDeliveryAbortRetries = [...this.#criticalDeliveryAbortRetryInputs.values()] + for (const timer of this.#criticalDeliveryAbortRetryTimers.values()) clearTimeout(timer) + this.#criticalDeliveryAbortRetryTimers.clear() if (this.#completionSweepTimer) clearTimeout(this.#completionSweepTimer) this.#completionSweepTimer = undefined this.#stoppingHeartbeatRefreshActive = await this.#stopLiveHeartbeat('stopping') try { await Promise.allSettled([...this.#dispatchLifecycleDrives]) + while (this.#deliveryFailureInFlight.size > 0) { + await Promise.allSettled([...this.#deliveryFailureInFlight]) + } + // A delivery-failure callback can be releasing a team while stop begins. + // Let that fenced cleanup settle before disposing the fleet; retry timers + // are already disabled by #stopping, so this drain cannot create new work. + while (this.#criticalDeliveryAborts.size > 0) { + await Promise.allSettled([...this.#criticalDeliveryAborts.values()]) + } + for (const pending of pendingCriticalDeliveryAbortRetries) { + await this.#preserveCriticalDeliveryAbortIntentForStop(pending.record, pending.target) + } + this.#criticalDeliveryAbortRetryInputs.clear() // Fence every source of new clarification work before touching the fleet. // A wake already past the fence is allowed to unwind, and is awaited // without a timeout so it can never race fleet disposal. @@ -709,10 +745,8 @@ export class FactoryLoop implements Factory { await this.#state.clearSlackThreads(this.#workspaceId) this.#slackWatcherStarts.clear() this.#offAgentExit?.() - this.#offDeliveryFailed?.() this.#offAgentMessage?.() this.#offAgentExit = undefined - this.#offDeliveryFailed = undefined this.#offAgentMessage = undefined await this.#fleet.dispose() } finally { @@ -1714,11 +1748,18 @@ export class FactoryLoop implements Factory { } let dispatchDecision = labelDispatch.decision + let dispatchRunId: string | undefined // A valid label resolution clears any prior failure notice so a later // regression posts a fresh, actionable comment instead of being deduped. this.#labelDispatchFailures.delete(issueStateKey(dispatchDecision.issue)) if (!dryRun && this.#fleet.placementLocality === 'remote') { + const previousLifecycle = await this.#state.getDispatchLifecycle(this.#workspaceId, issueKey(dispatchDecision.issue)) + if (previousLifecycle && isRetryableCriticalDeliveryAbort(previousLifecycle)) { + await this.#state.clearDispatchLifecycle(this.#workspaceId, issueKey(dispatchDecision.issue)) + this.#dispatchLifecycleEpochs.delete(issueKey(dispatchDecision.issue)) + } const lifecycleClaim = await this.#claimDispatchLifecycle(dispatchDecision, dryRun) + dispatchRunId = lifecycleClaim.lifecycle.runId dispatchDecision = structuredClone(lifecycleClaim.lifecycle.decision) if (lifecycleClaim.lifecycle.phase === 'waiting-for-human') { return lifecycleClaim.lifecycle.result ?? { issue: dispatchDecision.issue, agents: [], dryRun } @@ -1743,6 +1784,7 @@ export class FactoryLoop implements Factory { this.#emit('issue-queued', { issue: dispatchDecision.issue }) return { issue: dispatchDecision.issue, agents: [], dryRun } } + record.runId ??= dispatchRunId if (record.result) { return record.result @@ -1809,6 +1851,11 @@ export class FactoryLoop implements Factory { } return result } catch (error) { + if (error instanceof CriticalTaskInjectionError) { + await this.#abortCriticalDelivery(record, error.target, error) + this.#error(error, decision.issue) + throw error + } await this.#persistDispatchFailureReaperHandoff(record, spawnedForReaperHandoff) await this.#recordDispatchFailure(decision.issue) const failedState = await this.#state.getDispatchAttempts(this.#workspaceId, decision.issue.key) @@ -1851,7 +1898,14 @@ export class FactoryLoop implements Factory { } if (!this.#offDeliveryFailed) { this.#offDeliveryFailed = this.#fleet.onDeliveryFailed?.((info) => { - void this.#handleDeliveryFailed(info) + const handling = this.#handleDeliveryFailed(info).catch((error) => { + this.#logger.warn?.('[factory] delivery failure handler failed', { + target: info.to, + error: describeError(error).errorMessage, + }) + }) + this.#deliveryFailureInFlight.add(handling) + void handling.finally(() => this.#deliveryFailureInFlight.delete(handling)) }) } if (!this.#offAgentMessage) { @@ -2008,9 +2062,15 @@ export class FactoryLoop implements Factory { return false } const previous = await this.#state.getDispatchLifecycle(this.#workspaceId, key) + if (!record.runId || previous?.runId !== record.runId) { + this.#dispatchLifecycleEpochs.delete(key) + this.#increment('dispatchLifecycleRunFencesRejected') + this.#scheduleDispatchLifecycleRetry(record) + return false + } const lifecycle = lifecycleFromInFlightRecord( record, - previous?.runId ?? randomUUID(), + record.runId, phase, this.#clock.now(), pullRequest ?? previous?.pullRequest, @@ -2140,6 +2200,11 @@ export class FactoryLoop implements Factory { await this.#resumeDurableDispatch(record) return } + if (lifecycle.phase === 'aborting') { + const target = criticalDeliveryAbortTarget(lifecycle.releaseReason) ?? 'unknown agent' + await this.#abortCriticalDelivery(record, target, new Error('critical briefing delivery failed')) + return + } if (lifecycle.phase === 'publishing') { const implementer = [...record.agents.values()].find((agent) => agent.spec.role === 'implementer') if (!implementer) throw new Error(`durable dispatch ${record.issue.key} has no implementer to publish`) @@ -3239,6 +3304,10 @@ export class FactoryLoop implements Factory { if (!record) { return } + if (this.#criticalDeliveryAborting.has(issueKey(record.issue))) { + this.#increment('criticalDeliveryAbortExitsSuppressed') + return + } if (!await this.#assertDispatchLifecycleOwner(record)) { this.#logger.warn?.('[factory] ignored agent exit after durable lifecycle ownership was lost', { issue: record.issue.key, @@ -3790,21 +3859,317 @@ export class FactoryLoop implements Factory { async #handleDeliveryFailed(info: { to: string; msgId?: string; reason?: string }): Promise { const critical = await this.#state.consumeCritical(this.#workspaceId, info.msgId ?? '') - const record = (await this.#batch()).getIssueByAgent(info.to) + const batch = await this.#batch() + let record = batch.getIssueByAgent(info.to) + let terminalCriticalRun = false + if (!record && critical?.runId && this.#fleet.placementLocality === 'remote') { + const lifecycle = await this.#state.getDispatchLifecycle(this.#workspaceId, issueKey(critical.issue)) + if (lifecycle?.runId === critical.runId) { + if (isTerminalDispatchLifecycle(lifecycle)) terminalCriticalRun = true + // Reconstruct only the cleanup input. Do not speculatively consume a + // batch slot: the lifecycle can become terminal before the atomic + // abort-intent transition, and a rejected intent must leave no local + // in-flight record behind. + else record = inFlightRecordFromLifecycle(lifecycle) + } + } const issue = critical?.issue ?? record?.issue const error = new Error(`Critical delivery failed to ${info.to}${info.reason ? `: ${info.reason}` : ''}`) this.#error(error, issue) if (critical && this.#fleet.waitForInjected) { + if (terminalCriticalRun) { + this.#increment('criticalDeliveryFailuresIgnoredTerminalRun') + return + } + if (this.#fleet.placementLocality === 'remote' && ( + !critical.runId || !record?.runId || critical.runId !== record.runId + )) { + this.#increment('criticalDeliveryFailuresIgnoredStaleRun') + return + } + if (record && !await this.#assertDispatchLifecycleOwner(record)) { + // The message has already failed and was consumed. Persist the exact + // run's fatal intent even though this process lost its side-effect + // lease, so the successor cannot hydrate/resume an unbriefed team. + await this.#abortCriticalDelivery(record, info.to, error) + return + } try { const ack = await this.#waitForInjectedAndSubmit(critical.input) await this.#state.recordCritical(this.#workspaceId, ack.eventId, critical) } catch (retryError) { this.#error(retryError, critical.issue) + if (record) { + await this.#abortCriticalDelivery(record, info.to, retryError) + } + } + } + } + + async #abortCriticalDelivery(record: InFlightIssue, target: string, cause: unknown): Promise { + const key = issueKey(record.issue) + const existing = this.#criticalDeliveryAborts.get(key) + if (existing) return await existing + + // Fence exit/resume handling synchronously, before the first ownership or + // persistence await. A release-driven exit must never restart an unbriefed + // worker while the team-wide abort is in progress. + this.#criticalDeliveryAborting.add(key) + const abort = this.#doAbortCriticalDelivery(record, target, cause) + this.#criticalDeliveryAborts.set(key, abort) + try { + await abort + } catch (error) { + // Keep the exit/resume fence fail-closed, but guarantee unexpected + // failures have a convergence path that will eventually clear it. + this.#increment('criticalDeliveryAbortUnexpectedFailures') + if (this.#stopping && await this.#preserveCriticalDeliveryAbortIntentForStop(record, target)) { + // Shutdown needs only the exact-run fatal intent to be durable. A + // successor can finish fleet release and provider publication without + // making stop depend on eventual control-plane/provider recovery. + this.#criticalDeliveryAborting.delete(key) + return } + this.#scheduleCriticalDeliveryAbortRetry(record, target, cause) + throw error + } finally { + if (this.#criticalDeliveryAborts.get(key) === abort) this.#criticalDeliveryAborts.delete(key) } } + async #doAbortCriticalDelivery(record: InFlightIssue, target: string, cause: unknown): Promise { + const key = issueKey(record.issue) + const reason = `critical-delivery-failed:${target}` + const agents = [...record.agents] + const released = new Set() + if (this.#fleet.placementLocality === 'remote') { + if (!record.runId) { + this.#criticalDeliveryAborting.delete(key) + this.#increment('criticalDeliveryAbortsIgnoredMissingRun') + return + } + const lifecycle = await this.#state.beginCriticalDeliveryAbort( + this.#workspaceId, + key, + record.runId, + this.#clock.now(), + reason, + ) + if (!lifecycle) { + this.#criticalDeliveryAborting.delete(key) + this.#increment('criticalDeliveryAbortsIgnoredStaleRun') + return + } + for (const agent of lifecycle.agents) { + if (agent.releasedAtMs !== undefined) released.add(agent.name) + } + } + if (!await this.#assertDispatchLifecycleOwner(record)) { + this.#criticalDeliveryAborting.delete(key) + this.#increment('criticalDeliveryAbortsDeferredToOwner') + return + } + if (this.#fleet.placementLocality !== 'remote') { + await this.#persistDispatchFailureReaperHandoff(record, agents.map(([name, tracked]) => ({ + issue: record.issue, + name, + tracked: cloneTrackedAgent(tracked), + persistedAtMs: this.#clock.now(), + }))) + } + + for (const [agentName] of agents) { + this.#fleet.markAgentTerminal?.(agentName, reason) + } + + const failed = new Set() + for (const agent of agents) { + if (released.has(agent[0])) continue + const releaseFailed = await this.#releaseAndTerminateAgents([agent], reason, 'completion') + if (releaseFailed.length > 0) { + failed.add(agent[0]) + continue + } + released.add(agent[0]) + if (this.#fleet.placementLocality === 'remote' && !await this.#saveDispatchLifecycle( + record, + 'aborting', + undefined, + reason, + released, + )) { + // The durable abort intent remains authoritative, but this process no + // longer owns cleanup side effects. Its exit handlers are separately + // owner-fenced, so release the local suppression fence for takeover. + this.#criticalDeliveryAborting.delete(key) + return + } + } + + if (failed.size > 0) { + this.#increment('criticalDeliveryAbortReleaseRetries') + try { + await this.#publishCriticalDeliveryAbortNotice(record, target, cause, true) + } catch (error) { + this.#logger.warn?.('[factory] critical delivery abort notice failed while cleanup remained pending', { + issue: record.issue.key, + target, + error: describeError(error).errorMessage, + }) + } + await this.#writeInFlightRegistry() + this.#scheduleCriticalDeliveryAbortRetry(record, target, cause) + return + } + + // Publishing the operator-visible failure and restoring provider readiness + // are part of the abort transaction. Until both succeed, keep the batch + // slot and dispatch-attempt fence held even though the workers are gone. + try { + await this.#publishCriticalDeliveryAbortNotice(record, target, cause, false) + } catch (error) { + this.#increment('criticalDeliveryAbortPublicationRetries') + this.#logger.warn?.('[factory] critical delivery abort publication failed; retaining abort fence', { + issue: record.issue.key, + target, + error: describeError(error).errorMessage, + }) + this.#scheduleCriticalDeliveryAbortRetry(record, target, cause) + return + } + + // Drop the dispatch-attempt fence before the terminal lifecycle save. The + // still-authoritative `aborting` lifecycle prevents redispatch until that + // save succeeds, while a crash immediately after `abandoned` can no longer + // strand an in-flight attempt forever. + await this.#recordDispatchFailure(record.issue) + await this.#state.clearCriticalForIssue(this.#workspaceId, record.issue) + await this.#stopSlackWatcher(record.issue) + await this.#stopGithubIssueCommentWatcherForIssue(record.issue) + if (this.#fleet.placementLocality === 'remote') { + if (!await this.#saveDispatchLifecycle(record, 'abandoned', undefined, reason, released)) { + this.#criticalDeliveryAborting.delete(key) + return + } + this.#resolveDispatchTerminalWaiters(record.issue) + } + + const next = (await this.#batch()).complete(record.issue) + for (const [agentName] of agents) { + await this.#state.clearFailureHandoff(this.#workspaceId, registryHandoffKey(record.issue, agentName)) + } + await this.#writeInFlightRegistry() + this.#criticalDeliveryAborting.delete(key) + this.#increment('criticalDeliveryAbortsCompleted') + if (next) { + if (this.#stopping) { + (await this.#batch()).queue(next.decision, next.dryRun) + } else { + await this.dispatch(next.decision, { dryRun: next.dryRun }) + } + } + } + + async #preserveCriticalDeliveryAbortIntentForStop(record: InFlightIssue, target: string): Promise { + if (this.#fleet.placementLocality !== 'remote' || !record.runId) return false + const key = issueKey(record.issue) + const reason = `critical-delivery-failed:${target}` + let failures = 0 + // Once an exact critical record has been consumed, shutdown cannot safely + // release the lifecycle lease until a takeover-visible abort intent exists. + // Retry only this atomic state transition; full cleanup/provider recovery + // deliberately remains successor work. + while (this.#stopping) { + try { + const lifecycle = await this.#state.beginCriticalDeliveryAbort( + this.#workspaceId, + key, + record.runId, + this.#clock.now(), + reason, + ) + if (lifecycle) { + this.#increment('criticalDeliveryAbortStopIntentsPreserved') + return true + } + const current = await this.#state.getDispatchLifecycle(this.#workspaceId, key) + if (!current || current.runId !== record.runId || isTerminalDispatchLifecycle(current)) { + this.#increment('criticalDeliveryAbortStopIntentsObsolete') + return true + } + } catch (intentError) { + failures += 1 + this.#increment('criticalDeliveryAbortStopIntentRetries') + this.#logger.warn?.('[factory] retrying critical delivery abort intent during shutdown', { + issue: record.issue.key, + target, + failures, + error: describeError(intentError).errorMessage, + }) + } + await this.#clock.sleep(DISPATCH_LIFECYCLE_RETRY_MS) + } + return false + } + + #scheduleCriticalDeliveryAbortRetry(record: InFlightIssue, target: string, cause: unknown): void { + const key = issueKey(record.issue) + if (this.#stopping || this.#criticalDeliveryAbortRetryTimers.has(key)) return + this.#criticalDeliveryAbortRetryInputs.set(key, { record, target, cause }) + const timer = setTimeout(() => { + this.#criticalDeliveryAbortRetryTimers.delete(key) + this.#criticalDeliveryAbortRetryInputs.delete(key) + void this.#abortCriticalDelivery(record, target, cause).catch((error) => { + this.#logger.warn?.('[factory] critical delivery abort retry failed', { + issue: record.issue.key, + target, + error: describeError(error).errorMessage, + }) + this.#scheduleCriticalDeliveryAbortRetry(record, target, cause) + }) + }, DISPATCH_LIFECYCLE_RETRY_MS) + timer.unref?.() + this.#criticalDeliveryAbortRetryTimers.set(key, timer) + } + + async #publishCriticalDeliveryAbortNotice( + record: InFlightIssue, + target: string, + cause: unknown, + cleanupPending: boolean, + ): Promise { + const issue = await this.#readIssue(record.issue.path) + if (!issue) throw new Error(`Unable to publish critical delivery failure for unreadable issue ${record.issue.key}`) + const attempt = (await this.#state.getDispatchAttempts(this.#workspaceId, issueStateKey(record.issue)))?.attempts ?? 0 + const state = cleanupPending ? 'cleanup-pending' : 'aborted' + const marker = criticalDeliveryAbortMarker(record.issue, attempt, state) + const detail = describeError(cause).errorMessage + const body = cleanupPending + ? `${marker}\nFactory could not deliver the critical briefing to \`${target}\` for ${record.issue.key}. The dispatch is aborting, but one or more workers have not yet acknowledged release. Factory is retaining the in-progress fence and will retry cleanup; do not retry this issue yet.\n\nDelivery error: ${detail}` + : `${marker}\nFactory aborted this dispatch because the critical briefing could not be delivered to \`${target}\`. Every spawned worker was released before the issue can be made retryable, so no agent can continue on an incomplete task. Factory is returning the issue to Ready for Agent as this abort finalizes; retry after fleet delivery is healthy.\n\nDelivery error: ${detail}` + + if (!this.#criticalDeliveryNotices.has(marker)) { + if (isGithubIssue(issue) && this.#githubWriteback.hasCommentMarker) { + if (!await this.#githubWriteback.hasCommentMarker(issue, marker)) { + await this.#githubWriteback.postComment(issue, body) + } + } else { + await this.#postIssueComment(issue, body) + } + this.#criticalDeliveryNotices.add(marker) + } + if (cleanupPending) return + if (isGithubIssue(issue)) { + await this.#githubWriteback.clearStatus(issue) + } else { + const readyStateId = this.#states.idFor(issue.team, 'readyForAgent') + await this.#linear.setState(issue, readyStateId) + await this.#recordCanonicalIssueState({ ...record.issue, stateId: readyStateId }) + } + this.#emit('writeback-verified', { issue: record.issue, path: issue.path }) + } + async #handleAgentMessage(message: AgentMessage): Promise { const babysitterCritical = parseBabysitterCriticalSignal(message) if (babysitterCritical) { @@ -5027,8 +5392,12 @@ export class FactoryLoop implements Factory { from: 'factory', data: { issue: record.issue }, } - const ack = await this.#waitForInjectedAndSubmit(input) - await this.#state.recordCritical(this.#workspaceId, ack.eventId, { issue: record.issue, input }) + try { + const ack = await this.#waitForInjectedAndSubmit(input) + await this.#state.recordCritical(this.#workspaceId, ack.eventId, { issue: record.issue, input, runId: record.runId }) + } catch (error) { + throw new CriticalTaskInjectionError(input.to, error) + } } async #sendImplementerTask(record: InFlightIssue): Promise { @@ -5064,8 +5433,12 @@ export class FactoryLoop implements Factory { from: 'factory', data: { issue: record.issue }, } - const ack = await this.#waitForInjectedAndSubmit(input) - await this.#state.recordCritical(this.#workspaceId, ack.eventId, { issue: record.issue, input }) + try { + const ack = await this.#waitForInjectedAndSubmit(input) + await this.#state.recordCritical(this.#workspaceId, ack.eventId, { issue: record.issue, input, runId: record.runId }) + } catch (error) { + throw new CriticalTaskInjectionError(input.to, error) + } } } @@ -5923,7 +6296,7 @@ export class FactoryLoop implements Factory { data: { issue: record.issue }, } const ack = await this.#waitForInjectedAndSubmit(input) - await this.#state.recordCritical(this.#workspaceId, ack.eventId, { issue: record.issue, input }) + await this.#state.recordCritical(this.#workspaceId, ack.eventId, { issue: record.issue, input, runId: record.runId }) } } catch (error) { // Allow a later event to retry the spawn. @@ -9743,6 +10116,22 @@ const durableBabysitterTrackedAgent = (session: BabysitterSessionState): Tracked const isTerminalDispatchLifecycle = (lifecycle: DispatchLifecycle): boolean => lifecycle.phase === 'complete' || lifecycle.phase === 'abandoned' +const CRITICAL_DELIVERY_ABORT_REASON_PREFIX = 'critical-delivery-failed:' + +const criticalDeliveryAbortTarget = (reason?: string): string | undefined => + reason?.startsWith(CRITICAL_DELIVERY_ABORT_REASON_PREFIX) + ? reason.slice(CRITICAL_DELIVERY_ABORT_REASON_PREFIX.length) + : undefined + +const isRetryableCriticalDeliveryAbort = (lifecycle: DispatchLifecycle): boolean => + lifecycle.phase === 'abandoned' && criticalDeliveryAbortTarget(lifecycle.releaseReason) !== undefined + +const criticalDeliveryAbortMarker = ( + issue: IssueRef, + attempt: number, + state: 'cleanup-pending' | 'aborted', +): string => `` + const lifecycleFromInFlightRecord = ( record: InFlightIssue, runId: string, @@ -9765,6 +10154,7 @@ const lifecycleFromInFlightRecord = ( }) const inFlightRecordFromLifecycle = (lifecycle: DispatchLifecycle): InFlightIssue => ({ + runId: lifecycle.runId, issue: { ...lifecycle.issue }, decision: structuredClone(lifecycle.decision), dryRun: lifecycle.dryRun, diff --git a/src/ports/state.ts b/src/ports/state.ts index 5eb8fd1..57542af 100644 --- a/src/ports/state.ts +++ b/src/ports/state.ts @@ -2,7 +2,7 @@ import type { SendInput, SpawnResult } from './fleet' import type { InFlightIssue, QueuedIssue, TrackedAgent } from '../orchestrator/batch-tracker' import type { IssueRef, TriageDecision } from '../types' -export type CriticalRecord = { issue: IssueRef; input: SendInput } +export type CriticalRecord = { issue: IssueRef; input: SendInput; runId?: string } export type ClarificationReply = { id: string @@ -77,6 +77,7 @@ export type DispatchLifecyclePhase = | 'queued' | 'dispatching' | 'retryable' + | 'aborting' | 'running' | 'parking' | 'waiting-for-human' @@ -207,12 +208,26 @@ export interface StateStore { nowMs: number, lifecycle: DispatchLifecycle, ): Promise + /** + * Atomically records a fatal delivery failure for one exact dispatch run. + * This deliberately does not require or transfer the cleanup lease: a stale + * owner may preserve the fail-closed intent, but only the lease holder may + * perform fleet or provider side effects. + */ + beginCriticalDeliveryAbort( + workspaceId: string, + key: string, + runId: string, + nowMs: number, + releaseReason: string, + ): Promise getDispatchLifecycle(workspaceId: string, key: string): Promise listDispatchLifecycles(workspaceId: string): Promise> clearDispatchLifecycle(workspaceId: string, key: string): Promise recordCritical(workspaceId: string, key: string, value: CriticalRecord): Promise consumeCritical(workspaceId: string, key: string): Promise + clearCriticalForIssue(workspaceId: string, issue: IssueRef): Promise isResumed(workspaceId: string, exitKey: string): Promise markResumed(workspaceId: string, exitKey: string): Promise diff --git a/src/ports/writeback.ts b/src/ports/writeback.ts index d59fd37..5c44beb 100644 --- a/src/ports/writeback.ts +++ b/src/ports/writeback.ts @@ -23,5 +23,7 @@ export interface GithubWriteback { /** Provider-authoritative lookup used to reconcile ambiguous comment writes. */ hasCommentMarker?(issue: LinearIssue, marker: string): Promise setStatus(issue: LinearIssue, status: GithubIssueStatus): Promise + /** Remove Factory lifecycle labels after work is safely aborted before implementation starts. */ + clearStatus(issue: LinearIssue): Promise closeIssue(issue: LinearIssue, body: string): Promise } diff --git a/src/state/file-state-store.test.ts b/src/state/file-state-store.test.ts index 243b3c5..04f2c83 100644 --- a/src/state/file-state-store.test.ts +++ b/src/state/file-state-store.test.ts @@ -52,6 +52,51 @@ describe('FileStateStore', () => { } }) + it('atomically persists abort intent for only the exact dispatch run without transferring its lease', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-lifecycle-abort-')) + try { + const watchStatePath = join(root, 'state.json') + const store = new FileStateStore({ batchSize: 2, watchStatePath }) + const key = 'AR-86:uuid-86:/linear/issues/AR-86.json' + const seed = dispatchLifecycle(86) + const claim = await store.claimDispatchLifecycle('workspace-1', key, seed, 'owner-a', 1_000, 5_000) + + await expect(store.beginCriticalDeliveryAbort( + 'workspace-1', + key, + 'older-run', + 1_001, + 'critical-delivery-failed:stale-agent', + )).resolves.toBeUndefined() + await expect(store.getDispatchLifecycle('workspace-1', key)).resolves.toMatchObject({ + runId: seed.runId, + phase: 'dispatching', + lease: claim.lease, + }) + + await expect(store.beginCriticalDeliveryAbort( + 'workspace-1', + key, + seed.runId, + 1_002, + 'critical-delivery-failed:ar-86-impl-factory', + )).resolves.toMatchObject({ + runId: seed.runId, + phase: 'aborting', + releaseReason: 'critical-delivery-failed:ar-86-impl-factory', + lease: claim.lease, + }) + await expect(store.saveDispatchLifecycle('workspace-1', key, 'owner-a', 1, 1_003, { + ...seed, + phase: 'complete', + })).resolves.toBe(false) + await expect(new FileStateStore({ batchSize: 2, watchStatePath }).getDispatchLifecycle('workspace-1', key)) + .resolves.toMatchObject({ runId: seed.runId, phase: 'aborting', lease: claim.lease }) + } finally { + await rm(root, { recursive: true, force: true }) + } + }) + it('restores and clears babysitter ownership plus pending wake state in a fresh process-equivalent store', async () => { const root = await mkdtemp(join(tmpdir(), 'factory-file-state-babysitter-')) try { diff --git a/src/state/file-state-store.ts b/src/state/file-state-store.ts index 7efb645..32cc647 100644 --- a/src/state/file-state-store.ts +++ b/src/state/file-state-store.ts @@ -168,7 +168,14 @@ export class FileStateStore extends InMemoryStateStore { const document = await this.#loadFromDisk() const workspace = document.workspaces[workspaceId] const current = workspace?.dispatchLifecycles[key] - if (!current?.lease || current.lease.owner !== owner || current.lease.epoch !== epoch || current.lease.leaseUntilMs <= nowMs) { + if ( + !current?.lease || + current.runId !== lifecycle.runId || + current.lease.owner !== owner || + current.lease.epoch !== epoch || + current.lease.leaseUntilMs <= nowMs || + (current.phase === 'aborting' && lifecycle.phase !== 'aborting' && lifecycle.phase !== 'abandoned') + ) { return false } const next = cloneLifecycle(lifecycle) @@ -180,6 +187,30 @@ export class FileStateStore extends InMemoryStateStore { })) } + override async beginCriticalDeliveryAbort( + workspaceId: string, + key: string, + runId: string, + nowMs: number, + releaseReason: string, + ): Promise { + return await this.#exclusive(async () => this.#withMutationLock(async () => { + const document = await this.#loadFromDisk() + const lifecycle = document.workspaces[workspaceId]?.dispatchLifecycles[key] + if ( + !lifecycle || + lifecycle.runId !== runId || + lifecycle.phase === 'complete' || + lifecycle.phase === 'abandoned' + ) return undefined + lifecycle.phase = 'aborting' + lifecycle.releaseReason ??= releaseReason + lifecycle.updatedAtMs = nowMs + await this.#persist(document) + return cloneLifecycle(lifecycle) + })) + } + override async getDispatchLifecycle(workspaceId: string, key: string): Promise { return await this.#exclusive(async () => { const lifecycle = (await this.#loadFromDisk()).workspaces[workspaceId]?.dispatchLifecycles[key] diff --git a/src/state/in-memory-state-store.ts b/src/state/in-memory-state-store.ts index e7bab58..64b3043 100644 --- a/src/state/in-memory-state-store.ts +++ b/src/state/in-memory-state-store.ts @@ -150,7 +150,14 @@ export class InMemoryStateStore implements StateStore { lifecycle: DispatchLifecycle, ): Promise { const current = this.#workspace(workspaceId).dispatchLifecycles.get(key) - if (!current?.lease || current.lease.owner !== owner || current.lease.epoch !== epoch || current.lease.leaseUntilMs <= nowMs) { + if ( + !current?.lease || + current.runId !== lifecycle.runId || + current.lease.owner !== owner || + current.lease.epoch !== epoch || + current.lease.leaseUntilMs <= nowMs || + (current.phase === 'aborting' && lifecycle.phase !== 'aborting' && lifecycle.phase !== 'abandoned') + ) { return false } const next = cloneDispatchLifecycle(lifecycle) @@ -160,6 +167,26 @@ export class InMemoryStateStore implements StateStore { return true } + async beginCriticalDeliveryAbort( + workspaceId: string, + key: string, + runId: string, + nowMs: number, + releaseReason: string, + ): Promise { + const lifecycle = this.#workspace(workspaceId).dispatchLifecycles.get(key) + if ( + !lifecycle || + lifecycle.runId !== runId || + lifecycle.phase === 'complete' || + lifecycle.phase === 'abandoned' + ) return undefined + lifecycle.phase = 'aborting' + lifecycle.releaseReason ??= releaseReason + lifecycle.updatedAtMs = nowMs + return cloneDispatchLifecycle(lifecycle) + } + async getDispatchLifecycle(workspaceId: string, key: string): Promise { const lifecycle = this.#workspace(workspaceId).dispatchLifecycles.get(key) return lifecycle ? cloneDispatchLifecycle(lifecycle) : undefined @@ -186,6 +213,19 @@ export class InMemoryStateStore implements StateStore { return critical } + async clearCriticalForIssue(workspaceId: string, issue: CriticalRecord['issue']): Promise { + const criticalMessages = this.#workspace(workspaceId).criticalMessages + for (const [key, critical] of criticalMessages) { + if ( + critical.issue.key === issue.key && + critical.issue.uuid === issue.uuid && + critical.issue.path === issue.path + ) { + criticalMessages.delete(key) + } + } + } + async isResumed(workspaceId: string, exitKey: string): Promise { return this.#workspace(workspaceId).resumedExitKeys.has(exitKey) } diff --git a/src/writeback/github.ts b/src/writeback/github.ts index 58a9920..315ae5f 100644 --- a/src/writeback/github.ts +++ b/src/writeback/github.ts @@ -131,6 +131,23 @@ export class GhCliGithubWriteback implements GithubWriteback { } } + async clearStatus(issue: LinearIssue): Promise { + const ref = githubIssueRef(issue) + const labels = await this.#issueLabels(ref) + const lifecycleLabels = Object.values(STATUS_LABELS) + .map((status) => status.name) + .filter((name) => labels.has(name.toLowerCase())) + if (lifecycleLabels.length === 0) return + await this.#run([ + 'issue', + 'edit', + String(ref.number), + '--repo', + ref.repo, + ...lifecycleLabels.flatMap((name) => ['--remove-label', name]), + ]) + } + async #issueLabels(ref: { repo: string; number: number }): Promise> { const result = await this.#run([ 'issue', diff --git a/src/writeback/writeback.test.ts b/src/writeback/writeback.test.ts index db0d75f..4f7591b 100644 --- a/src/writeback/writeback.test.ts +++ b/src/writeback/writeback.test.ts @@ -796,6 +796,7 @@ describe('GhCliGithubWriteback', () => { await github.postComment(githubIssue, 'Factory dispatch for 48') await github.setStatus(githubIssue, 'in-progress') await github.setStatus(githubIssue, 'human-review') + await github.clearStatus(githubIssue) expect(calls).toEqual([ ['issue', 'comment', '48', '--repo', 'AgentWorkforce/factory', '--body', 'Factory dispatch for 48'], @@ -805,6 +806,8 @@ describe('GhCliGithubWriteback', () => { ['label', 'create', 'factory:human-review', '--repo', 'AgentWorkforce/factory', '--color', 'fbca04', '--description', 'Factory work is ready for human review.', '--force'], ['issue', 'view', '48', '--repo', 'AgentWorkforce/factory', '--json', 'labels'], ['issue', 'edit', '48', '--repo', 'AgentWorkforce/factory', '--add-label', 'factory:human-review', '--remove-label', 'factory:in-progress'], + ['issue', 'view', '48', '--repo', 'AgentWorkforce/factory', '--json', 'labels'], + ['issue', 'edit', '48', '--repo', 'AgentWorkforce/factory', '--remove-label', 'factory:human-review'], ]) })