diff --git a/README.md b/README.md index 0583417..a8a2388 100644 --- a/README.md +++ b/README.md @@ -143,6 +143,19 @@ After init, add the `factory` label to an open issue and run a dry run below. | `factory featuremap check [--manifest ] [--base ]` | Validate the repository feature/test manifest and optionally report advisory drift for unchanged entries whose locations changed. | | `factory intake notion ` | Normalize ready specs from a read-only Notion mount into GitHub lifecycle issues or exact-path fleet work. Honors `--dry-run`. | +`factory status` includes `inFlightDispatches`, grouped by issue with agent +names and the provider-claim state (`pending`, `verified`, or `degraded`). This +view is read from Factory's local in-flight registry, so it remains available +when GitHub lifecycle writeback is the degraded subsystem. + +Dispatch lifecycle writes are claim-critical. Factory applies the +`factory:in-progress` label/state before the dispatch comment, confirms the +GitHub label by provider read-back, and retries either write three times. An +exhausted write is logged at error level as dead-lettered, recorded as a +degraded claim in the registry, and fails the dispatch instead of reporting a +clean dispatch with missing GitHub state. Durable lifecycle recovery repeats +the same label-and-comment claim without respawning acknowledged agents. + Global options work anywhere in the args: `--config `, `--dry-run`, `--backend `, and `--agent-exit-timeout `. The internal backend reuses a relay broker that's already running for your workspace, and diff --git a/src/cli/fleet.test.ts b/src/cli/fleet.test.ts index b4c9af3..f828c0c 100644 --- a/src/cli/fleet.test.ts +++ b/src/cli/fleet.test.ts @@ -2590,6 +2590,88 @@ describe('fleet CLI runtime', () => { } }) + it('lists registry-backed in-flight issues, agents, and degraded claims in factory status', async () => { + const root = await mkdtemp(join(tmpdir(), 'fleet-cli-registry-status-')) + try { + const heartbeatPath = join(root, 'heartbeat.json') + const registryPath = join(root, 'registry.json') + const configPath = await writeConfig(root, { + loop: { heartbeatPath, registryPath, heartbeatStaleMs: 10_000 }, + }) + const output = buffer() + const factory = { + status: vi.fn(() => ({ inFlight: [], queued: [], counters: {} })), + } as unknown as Factory + await writeFile(registryPath, JSON.stringify({ + pid: 4242, + heartbeatPath, + updatedAt: '2026-08-14T13:30:00.000Z', + updatedAtMs: Date.parse('2026-08-14T13:30:00.000Z'), + agents: [ + { + name: 'ar-242-impl-factory', + role: 'implementer', + issue: { uuid: 'AgentWorkforce/factory#242', key: '242', path: '/github/factory/242.json' }, + sessionRef: 'session-impl', + pids: [], + node: 'oslo-mini', + dispatchClaim: { + state: 'degraded', + write: 'GitHub dispatch comment', + attempts: 3, + maxAttempts: 3, + deadLettered: true, + error: 'GitHub comment write unavailable', + updatedAtMs: Date.parse('2026-08-14T13:29:00.000Z'), + }, + }, + { + name: 'ar-242-review', + role: 'reviewer', + issue: { uuid: 'AgentWorkforce/factory#242', key: '242', path: '/github/factory/242.json' }, + pids: [], + dispatchClaim: { + state: 'degraded', + write: 'GitHub dispatch comment', + attempts: 3, + maxAttempts: 3, + deadLettered: true, + error: 'GitHub comment write unavailable', + updatedAtMs: Date.parse('2026-08-14T13:29:00.000Z'), + }, + }, + ], + })) + + const code = await runFleetCli(['status', '--config', configPath], { + fleet: new FakeFleetClient(), + mount: new FakeMountClient(), + createFactory: () => factory, + stdout: output, + stderr: buffer(), + }) + + expect(code).toBe(0) + expect(JSON.parse(output.text())).toMatchObject({ + inFlightDispatches: [{ + issue: { key: '242' }, + agents: [ + { name: 'ar-242-impl-factory', role: 'implementer', sessionRef: 'session-impl', node: 'oslo-mini' }, + { name: 'ar-242-review', role: 'reviewer' }, + ], + claim: { + state: 'degraded', + write: 'GitHub dispatch comment', + attempts: 3, + deadLettered: true, + }, + }], + }) + } finally { + await rm(root, { recursive: true, force: true }) + } + }) + it('surfaces a stale registered workspace mirror in factory status', async () => { const root = await mkdtemp(join(tmpdir(), 'fleet-cli-stale-status-')) try { diff --git a/src/cli/fleet.ts b/src/cli/fleet.ts index 46fdc9e..9036c6d 100644 --- a/src/cli/fleet.ts +++ b/src/cli/fleet.ts @@ -34,6 +34,7 @@ import { reapFactoryOrphansOnce, reapFactoryEnvironmentsOnce, readFactoryLoopHeartbeat, + readFactoryInFlightRegistry, resolveFactoryStates, stateResolutionFromIds, standaloneBabysitterAgentName, @@ -42,6 +43,8 @@ import { type Capability, type Factory, type FactoryEventReporter, + type FactoryInFlightDispatchStatus, + type FactoryInFlightRegistry, type FactoryConfig, type IterationReport, type FleetBackend, @@ -733,7 +736,13 @@ async function runFactoryCommand( return 0 } if (command.action === 'status') { - writeJson(out, await factoryStatusWithMountHealth(factory, mount, config.loop.heartbeatPath, config.loop.heartbeatStaleMs)) + writeJson(out, await factoryStatusWithMountHealth( + factory, + mount, + config.loop.heartbeatPath, + config.loop.registryPath, + config.loop.heartbeatStaleMs, + )) return 0 } if (command.action === 'loop-status') { @@ -764,7 +773,13 @@ async function runFactoryCommand( const reports = await factory.runLoop({ dryRun: globals.dryRun }) writeJson(out, { reports, - status: await factoryStatusWithMountHealth(factory, mount, config.loop.heartbeatPath, config.loop.heartbeatStaleMs), + status: await factoryStatusWithMountHealth( + factory, + mount, + config.loop.heartbeatPath, + config.loop.registryPath, + config.loop.heartbeatStaleMs, + ), }) } finally { removeSignalHandlers() @@ -1051,6 +1066,7 @@ async function factoryStatusWithMountHealth( factory: Factory, mount: MountClient, heartbeatPath: string, + registryPath: string, heartbeatStaleMs: number, ): Promise & { localMountDegraded?: boolean @@ -1065,7 +1081,16 @@ async function factoryStatusWithMountHealth( } }> { const status = factory.status() - const heartbeat = await readFactoryLoopHeartbeat(heartbeatPath) + const [heartbeat, registry] = await Promise.all([ + readFactoryLoopHeartbeat(heartbeatPath), + readFactoryInFlightRegistry(registryPath), + ]) + const registryDispatches = registry?.heartbeatPath && registry.heartbeatPath !== heartbeatPath + ? [] + : inFlightDispatchesFromRegistry(registry) + const observableStatus = registryDispatches.length > 0 + ? { ...status, inFlightDispatches: registryDispatches } + : status const liveness = checkFactoryLoopLiveness(heartbeat, { staleMs: heartbeatStaleMs }) const eventListener = liveness.ok ? heartbeat?.eventListener ?? { @@ -1077,9 +1102,9 @@ async function factoryStatusWithMountHealth( reason: liveness.reason, } const health = mount.getLocalMountHealth?.() - if (!health) return { ...status, eventListener } + if (!health) return { ...observableStatus, eventListener } return { - ...status, + ...observableStatus, eventListener, localMountDegraded: health.degraded, ...(health.reason ? { localMountDegradedReason: health.reason } : {}), @@ -1093,6 +1118,49 @@ async function factoryStatusWithMountHealth( } } +function inFlightDispatchesFromRegistry( + registry: FactoryInFlightRegistry | undefined, +): FactoryInFlightDispatchStatus[] { + if (!registry) return [] + const grouped = new Map() + const claimPriority = { verified: 0, pending: 1, degraded: 2 } as const + + for (const agent of registry.agents) { + if (!agent.issue) continue + const key = `${agent.issue.key}\u0000${agent.issue.uuid}\u0000${agent.issue.path}` + const claim = agent.dispatchClaim ?? { + state: 'pending' as const, + updatedAtMs: registry.updatedAtMs, + } + const existing = grouped.get(key) + const entry = existing ?? { + issue: { ...agent.issue }, + agents: [], + claim: { ...claim }, + } + if (claimPriority[claim.state] > claimPriority[entry.claim.state]) { + entry.claim = { ...claim } + } + if (!entry.agents.some((candidate) => candidate.name === agent.name)) { + entry.agents.push({ + name: agent.name, + ...(agent.role ? { role: agent.role } : {}), + ...(agent.sessionRef ? { sessionRef: agent.sessionRef } : {}), + ...(agent.invocationId ? { invocationId: agent.invocationId } : {}), + ...(agent.node ? { node: agent.node } : {}), + }) + } + grouped.set(key, entry) + } + + return [...grouped.values()] + .map((entry) => ({ + ...entry, + agents: entry.agents.sort((left, right) => left.name.localeCompare(right.name)), + })) + .sort((left, right) => left.issue.key.localeCompare(right.issue.key)) +} + function writeMountRefreshSummary( refreshedStaleMounts: readonly RefreshedStaleMount[], stderr: Pick, @@ -1872,6 +1940,7 @@ async function issueProjectionStatus( factory, mount, config.loop.heartbeatPath, + config.loop.registryPath, config.loop.heartbeatStaleMs, ) const githubConnection = mount.integrationConnections diff --git a/src/index.ts b/src/index.ts index 17ea22a..62b4c22 100644 --- a/src/index.ts +++ b/src/index.ts @@ -430,6 +430,8 @@ export type { DispatchResult, Factory, FactoryEventPayload, + FactoryDispatchClaimStatus, + FactoryInFlightDispatchStatus, FactoryInFlightRegistry, FactoryInFlightRegistryAgent, FactoryInFlightRegistryProcess, diff --git a/src/orchestrator/batch-tracker.ts b/src/orchestrator/batch-tracker.ts index 0f71cba..21360d8 100644 --- a/src/orchestrator/batch-tracker.ts +++ b/src/orchestrator/batch-tracker.ts @@ -1,5 +1,5 @@ import type { AgentSpec, SpawnResult } from '../ports' -import type { DispatchResult, IssueRef, TriageDecision } from '../types' +import type { DispatchResult, FactoryDispatchClaimStatus, IssueRef, TriageDecision } from '../types' import { githubRepositoriesMatch } from '../github/repo-identity' export interface TrackedAgent { @@ -17,6 +17,7 @@ export interface InFlightIssue { agents: Map invocationIds: Set result?: DispatchResult + dispatchClaim?: FactoryDispatchClaimStatus } export interface QueuedIssue { diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 1873e51..d5d8064 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -346,7 +346,11 @@ class ProviderHumanReviewGithubWriteback extends RecordingGithubWriteback { } class FailingGithubCommentWriteback extends RecordingGithubWriteback { - override async postComment(): Promise { + override async postComment(issue: LinearIssue, body: string): Promise { + if (body.startsWith('Factory dispatch for ')) { + await super.postComment(issue, body) + return + } throw new Error('GitHub comment writeback unavailable') } } @@ -2325,6 +2329,141 @@ describe('FactoryLoop', () => { expect(mergeGate.merges).toEqual([]) }) + it('retries a failed GitHub dispatch claim, dead-letters it loudly, and preserves registry visibility', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-github-dispatch-writeback-')) + const path = githubIssuePath('AgentWorkforce', 'pear', 242) + const registryPath = join(root, 'registry.json') + const heartbeatPath = join(root, 'heartbeat.json') + const mount = new FakeMountClient({ + [path]: githubIssueFile(242, { labels: ['factory'] }), + }) + const githubWriteback = new RecordingGithubWriteback() + let labelAttempts = 0 + githubWriteback.setStatus = async (_issue, status) => { + if (status !== 'in-progress') return + labelAttempts += 1 + throw new Error('GitHub label write unavailable') + } + const errors: unknown[][] = [] + const factory = createFactory(config({ + issueSource: 'github', + loop: { registryPath, heartbeatPath }, + }), { + mount, + fleet: new FakeFleetClient(), + triage: new StaticTriage(), + githubWriteback, + clock: new ManualClock(), + logger: { + warn: () => {}, + error: (...args: unknown[]) => errors.push(args), + }, + }) + + try { + const decision = await factory.triageIssue(parseGithubFactoryIssue(path, githubIssueFile(242, { labels: ['factory'] }))) + await expect(factory.dispatch(decision)).rejects.toThrow('GitHub label write unavailable') + + expect(labelAttempts).toBe(3) + expect(githubWriteback.comments).toEqual([]) + expect(factory.status().counters).toMatchObject({ + dispatchWritebackFailures: 3, + dispatchWritebackRetries: 2, + dispatchWritebackDeadLetters: 1, + }) + expect(errors.filter(([message]) => message === '[factory] dispatch writeback failed; retrying')) + .toHaveLength(2) + expect(errors).toContainEqual([ + '[factory] dispatch writeback dead-lettered after retries exhausted', + expect.objectContaining({ + issue: '242', + write: 'GitHub label factory:in-progress', + attempts: 3, + error: 'GitHub label write unavailable', + }), + ]) + + const registry = await readFactoryInFlightRegistry(registryPath) + expect(registry?.agents.map((agent) => ({ + name: agent.name, + issue: agent.issue?.key, + claim: agent.dispatchClaim, + }))).toEqual([ + { + name: 'ar-242-impl-pear', + issue: '242', + claim: expect.objectContaining({ + state: 'degraded', + write: 'GitHub label factory:in-progress', + attempts: 3, + maxAttempts: 3, + deadLettered: true, + error: 'GitHub label write unavailable', + }), + }, + { + name: 'ar-242-review-pear', + issue: '242', + claim: expect.objectContaining({ + state: 'degraded', + deadLettered: true, + }), + }, + ]) + } finally { + await factory.stop() + await rm(root, { recursive: true, force: true }) + } + }) + + it('retries a transient GitHub dispatch comment instead of silently skipping it', async () => { + const path = githubIssuePath('AgentWorkforce', 'pear', 243) + const issue = githubIssueFile(243, { labels: ['factory'] }) + const mount = new FakeMountClient({ [path]: issue }) + const githubWriteback = new RecordingGithubWriteback() + const originalPostComment = githubWriteback.postComment.bind(githubWriteback) + let commentAttempts = 0 + githubWriteback.postComment = async (target, body) => { + commentAttempts += 1 + if (commentAttempts < 3) throw new Error('transient GitHub comment failure') + await originalPostComment(target, body) + } + const errors: unknown[][] = [] + const factory = createFactory(config({ issueSource: 'github' }), { + mount, + fleet: new FakeFleetClient(), + triage: new StaticTriage(), + githubWriteback, + clock: new ManualClock(), + logger: { + warn: () => {}, + error: (...args: unknown[]) => errors.push(args), + }, + }) + + await expect(factory.dispatch(await factory.triageIssue(parseGithubFactoryIssue(path, issue)))) + .resolves.toMatchObject({ issue: { key: '243' } }) + + expect(commentAttempts).toBe(3) + expect(githubWriteback.statuses).toEqual([{ key: '243', status: 'in-progress' }]) + expect(githubWriteback.comments).toEqual([ + { key: '243', body: expect.stringContaining('Factory dispatch for 243') }, + ]) + expect(errors.filter(([message]) => message === '[factory] dispatch writeback failed; retrying')) + .toHaveLength(2) + expect(factory.status()).toMatchObject({ + inFlightDispatches: [{ + issue: { key: '243' }, + claim: { state: 'verified' }, + }], + counters: { + dispatchWritebackFailures: 2, + dispatchWritebackRetries: 2, + }, + }) + await factory.stop() + }) + it('dispatches a dependency chain in order and promotes the next issue after its blocker closes', async () => { const blockerPath = githubIssuePath('AgentWorkforce', 'pear', 128) const firstDependentPath = githubIssuePath('AgentWorkforce', 'pear', 131) @@ -12185,12 +12324,14 @@ describe('FactoryLoop', () => { }) }) - it('logs and continues when best-effort dispatch comment writeback fails', async () => { + it('retries and fails the dispatch when its Linear claim comment cannot be recorded', async () => { const mount = new FakeMountClient({ [issuePath(25)]: issueFile(25) }) const fleet = new FakeFleetClient() - const warnings: unknown[] = [] + const errors: unknown[] = [] + let commentAttempts = 0 const linear: LinearWriteback = { async postComment() { + commentAttempts += 1 throw new Error('unsupported Linear writeback path') }, async setState(issue, stateId) { @@ -12208,23 +12349,25 @@ describe('FactoryLoop', () => { fleet, triage: new StaticTriage(), linear, + clock: new ManualClock(), logger: { - warn: (...args: unknown[]) => warnings.push(args), - error: () => {}, + warn: () => {}, + error: (...args: unknown[]) => errors.push(args), }, }) const decision = await factory.triageIssue(parseLinearIssue(issuePath(25), issueFile(25))) - await expect(factory.dispatch(decision)).resolves.toMatchObject({ - issue: { key: 'AR-25' }, - stateId: implementing, - }) - expect(warnings[0]).toEqual([ - '[factory] comment writeback skipped', + await expect(factory.dispatch(decision)).rejects.toThrow('unsupported Linear writeback path') + expect(commentAttempts).toBe(3) + expect(errors.filter((entry) => (entry as unknown[])[0] === '[factory] dispatch writeback failed; retrying')) + .toHaveLength(2) + expect(errors).toContainEqual([ + '[factory] dispatch writeback dead-lettered after retries exhausted', expect.objectContaining({ - name: 'Error', - message: 'unsupported Linear writeback path', - stack: expect.stringContaining('Error: unsupported Linear writeback path'), + issue: 'AR-25', + write: 'Linear dispatch comment', + attempts: 3, + error: 'unsupported Linear writeback path', }), ]) expect(mount.writes).toContainEqual({ path: issuePath(25), content: { stateId: implementing } }) diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 13a25b0..f3f74d7 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -76,6 +76,8 @@ import type { FactoryLoopRunOptions, FactoryLoopHeartbeat, FactoryLoopLiveness, + FactoryDispatchClaimStatus, + FactoryInFlightDispatchStatus, FactoryInFlightRegistry, FactoryInFlightRegistryAgent, IssueRef, @@ -333,6 +335,8 @@ const STOP_TEARDOWN_TIMEOUT_MS = 2_500 const DISPATCH_LIFECYCLE_LEASE_MS = 5 * 60_000 const DISPATCH_LIFECYCLE_RENEW_MS = 60_000 const DISPATCH_LIFECYCLE_RETRY_MS = 1_000 +const DISPATCH_WRITEBACK_MAX_ATTEMPTS = 3 +const DISPATCH_WRITEBACK_RETRY_MS = 250 const STARTUP_AGENT_EXIT_DRAIN_TIMEOUT_MS = 30_000 const RECONCILED_AGENT_EXIT_CONCURRENCY = 4 const SLACK_EVENT_WATERMARK_CACHE_MS = 60_000 @@ -501,6 +505,7 @@ export class FactoryLoop implements Factory { readonly #abandonedDispatchReasons = new Map() readonly #dispatchLifecycleCapacityWaitLogged = new Set() readonly #dispatchLifecycleOwnershipWaitLogged = new Set() + readonly #dispatchClaimStatuses = new Map() readonly #localReleaseCheckpoints = new Map>() #dispatchLifecycleRenewTimer?: ReturnType #clarificationSweepTimer?: ReturnType @@ -2702,6 +2707,13 @@ export class FactoryLoop implements Factory { } agents.push({ name: spawned.name, role: spec.role }) } + if (!dryRun) { + record.dispatchClaim = { + state: 'pending', + updatedAtMs: this.#clock.now(), + } + this.#dispatchClaimStatuses.set(issueKey(record.issue), record.dispatchClaim) + } await this.#writeInFlightRegistry() const comment = dispatchComment(dispatchDecision, agents) @@ -2711,17 +2723,7 @@ export class FactoryLoop implements Factory { if (!issue || !this.#isIssueReady(issue)) { throw new LiveDispatchStateChangedError(dispatchDecision.issue.key) } - try { - await this.#postIssueComment(issue, comment) - } catch (error) { - this.#logger.warn?.('[factory] comment writeback skipped', error) - } - if (isGithubIssue(issue)) { - await this.#githubWriteback.setStatus(issue, 'in-progress') - } else { - implementingStateId = this.#states.idFor(issue.team, 'agentImplementing') - await this.#linear.setState(issue, implementingStateId) - } + implementingStateId = await this.#applyDispatchClaim(record, issue, comment) this.#emit('writeback-verified', { issue: dispatchDecision.issue, path: issue.path }) } @@ -2846,8 +2848,27 @@ export class FactoryLoop implements Factory { status(): FactoryStatus { const batch = this.#batchView + const inFlightDispatches: FactoryInFlightDispatchStatus[] = batch?.inFlight + .filter((record) => !record.dryRun) + .map((record) => ({ + issue: { ...record.issue }, + agents: [...record.agents].map(([name, tracked]) => ({ + name, + role: tracked.spec.role, + ...(tracked.sessionRef ? { sessionRef: tracked.sessionRef } : {}), + ...(tracked.spec.invocationId ? { invocationId: tracked.spec.invocationId } : {}), + ...(tracked.result?.node ? { node: tracked.result.node } : {}), + })), + claim: { + ...(record.dispatchClaim ?? this.#dispatchClaimStatuses.get(issueKey(record.issue)) ?? { + state: 'pending' as const, + updatedAtMs: this.#clock.now(), + }), + }, + })) ?? [] return { inFlight: batch?.inFlight.map((record) => record.issue) ?? [], + ...(inFlightDispatches.length > 0 ? { inFlightDispatches } : {}), queued: batch?.queued.map((queued) => queued.issue) ?? [], parked: batch?.parked.map((parked) => ({ issue: parked.issue, @@ -4130,16 +4151,21 @@ export class FactoryLoop implements Factory { const spawned = await this.#spawnAgent(record, spec, record.dryRun) agents.push({ name: spawned.name, role: spec.role }) } + const comment = dispatchComment(record.decision, agents) + let implementingStateId: string | undefined + if (!record.dryRun) { + record.dispatchClaim = { + state: 'pending', + updatedAtMs: this.#clock.now(), + } + this.#dispatchClaimStatuses.set(issueKey(record.issue), record.dispatchClaim) + } await this.#writeInFlightRegistry() if (!record.dryRun) { const issue = liveIssue ?? await this.#readIssue(record.issue.path) if (!issue) throw new Error(`Unable to recover durable dispatch ${record.issue.key}: issue is no longer readable`) await this.#ensureGithubAgentQuestionWatch(record, issue) - if (isGithubIssue(issue)) { - await this.#githubWriteback.setStatus(issue, 'in-progress') - } else { - await this.#linear.setState(issue, this.#states.idFor(issue.team, 'agentImplementing')) - } + implementingStateId = await this.#applyDispatchClaim(record, issue, comment) } const recoveredPreviews = uniquePreviewReferences([ ...dispatchSpecs(record.decision).map((spec) => spec.preview), @@ -4148,7 +4174,8 @@ export class FactoryLoop implements Factory { record.result ??= { issue: record.issue, agents, - comments: [dispatchComment(record.decision, agents)], + comments: [comment], + stateId: implementingStateId, ...(recoveredPreviews.length > 0 ? { previews: recoveredPreviews } : {}), dryRun: record.dryRun, } @@ -4476,6 +4503,116 @@ export class FactoryLoop implements Factory { await this.#linear.postComment(issue, body) } + async #applyDispatchClaim( + record: InFlightIssue, + issue: LinearIssue, + comment: string, + ): Promise { + let implementingStateId: string | undefined + if (isGithubIssue(issue)) { + await this.#retryDispatchWriteback(record, issue, 'GitHub label factory:in-progress', async () => { + await this.#githubWriteback.setStatus(issue, 'in-progress') + }) + + const commentApplied = this.#githubWriteback.hasCommentMarker + ? async (): Promise => this.#githubWriteback.hasCommentMarker!(issue, comment) + : undefined + await this.#retryDispatchWriteback( + record, + issue, + 'GitHub dispatch comment', + async () => this.#githubWriteback.postComment(issue, comment), + commentApplied, + ) + } else { + implementingStateId = this.#states.idFor(issue.team, 'agentImplementing') + await this.#retryDispatchWriteback(record, issue, `Linear state ${implementingStateId}`, async () => { + await this.#linear.setState(issue, implementingStateId!) + }) + await this.#retryDispatchWriteback(record, issue, 'Linear dispatch comment', async () => { + await this.#linear.postComment(issue, comment) + }) + } + + record.dispatchClaim = { + state: 'verified', + updatedAtMs: this.#clock.now(), + } + this.#dispatchClaimStatuses.set(issueKey(record.issue), record.dispatchClaim) + await this.#writeDispatchClaimRegistry(record.issue) + return implementingStateId + } + + async #retryDispatchWriteback( + record: InFlightIssue, + issue: LinearIssue, + write: string, + apply: () => Promise, + isApplied?: () => Promise, + ): Promise { + let lastError: unknown + for (let attempt = 1; attempt <= DISPATCH_WRITEBACK_MAX_ATTEMPTS; attempt += 1) { + try { + if (isApplied && await isApplied()) return + await apply() + if (isApplied && !await isApplied()) { + throw new Error(`${write} returned without a provider read-back acknowledgement`) + } + return + } catch (error) { + lastError = error + const deadLettered = attempt === DISPATCH_WRITEBACK_MAX_ATTEMPTS + this.#increment('dispatchWritebackFailures') + record.dispatchClaim = { + state: 'degraded', + write, + attempts: attempt, + maxAttempts: DISPATCH_WRITEBACK_MAX_ATTEMPTS, + error: describeError(error).errorMessage, + ...(deadLettered ? { deadLettered: true } : {}), + updatedAtMs: this.#clock.now(), + } + this.#dispatchClaimStatuses.set(issueKey(record.issue), record.dispatchClaim) + await this.#writeDispatchClaimRegistry(record.issue) + + if (deadLettered) { + this.#increment('dispatchWritebackDeadLetters') + this.#logger.error?.('[factory] dispatch writeback dead-lettered after retries exhausted', { + issue: issue.key, + write, + attempts: attempt, + error: describeError(error).errorMessage, + }) + throw error + } + + this.#increment('dispatchWritebackRetries') + this.#logger.error?.('[factory] dispatch writeback failed; retrying', { + issue: issue.key, + write, + attempt, + maxAttempts: DISPATCH_WRITEBACK_MAX_ATTEMPTS, + retryMs: DISPATCH_WRITEBACK_RETRY_MS, + error: describeError(error).errorMessage, + }) + await this.#clock.sleep(DISPATCH_WRITEBACK_RETRY_MS) + } + } + throw lastError + } + + async #writeDispatchClaimRegistry(issue: IssueRef): Promise { + try { + await this.#writeInFlightRegistry() + } catch (error) { + this.#increment('dispatchClaimRegistryWriteFailures') + this.#logger.error?.('[factory] failed to persist dispatch claim visibility', { + issue: issue.key, + error: describeError(error).errorMessage, + }) + } + } + // Probes the GitHub issue sub-root at most once and caches the verdict so // repeated iterations skip listTree calls when the mount is absent. async #ensureGithubIngestionReady(): Promise { @@ -5775,6 +5912,7 @@ export class FactoryLoop implements Factory { } } const fleetTracked = this.#fleet.trackedAgents?.().get(agentName) + const dispatchClaim = this.#dispatchClaimStatuses.get(issueKey(issue)) agents.push({ name: agentName, role: tracked.spec.role, @@ -5784,12 +5922,16 @@ export class FactoryLoop implements Factory { processes, ...(fleetTracked?.invocationId ? { invocationId: fleetTracked.invocationId } : {}), ...(fleetTracked?.node ? { node: fleetTracked.node } : {}), + ...(dispatchClaim ? { dispatchClaim: { ...dispatchClaim } } : {}), }) } if (!empty) { for (const record of (await this.#batch()).inFlight) { if (record.dryRun) continue + if (record.dispatchClaim) { + this.#dispatchClaimStatuses.set(issueKey(record.issue), record.dispatchClaim) + } for (const [agentName, tracked] of record.agents) { await appendAgent(record.issue, agentName, tracked) } @@ -16110,6 +16252,7 @@ const lifecycleFromInFlightRecord = ( agents: [...record.agents].map(([name, tracked]) => ({ name, tracked: cloneTrackedAgent(tracked) })), invocationIds: [...record.invocationIds], result: record.result ? structuredClone(record.result) : undefined, + ...(record.dispatchClaim ? { dispatchClaim: { ...record.dispatchClaim } } : {}), ...(pullRequests.length > 0 ? { pullRequests: pullRequests.map((receipt) => ({ ...receipt })) } : {}), ...(pullRequest ? { pullRequest: { ...pullRequest } } : {}), ...(releaseReason ? { releaseReason } : {}), @@ -16124,6 +16267,7 @@ const inFlightRecordFromLifecycle = (lifecycle: DispatchLifecycle): InFlightIssu agents: new Map(lifecycle.agents.map((agent) => [agent.name, cloneTrackedAgent(agent.tracked)])), invocationIds: new Set(lifecycle.invocationIds), result: lifecycle.result ? structuredClone(lifecycle.result) : undefined, + ...(lifecycle.dispatchClaim ? { dispatchClaim: { ...lifecycle.dispatchClaim } } : {}), }) const dispatchResultFromLifecycle = (lifecycle: DispatchLifecycle): DispatchResult => diff --git a/src/ports/state.ts b/src/ports/state.ts index 7a0e22e..df9234a 100644 --- a/src/ports/state.ts +++ b/src/ports/state.ts @@ -7,7 +7,7 @@ import type { QueuedIssue, TrackedAgent, } from '../orchestrator/batch-tracker' -import type { IssueRef, TriageDecision } from '../types' +import type { FactoryDispatchClaimStatus, IssueRef, TriageDecision } from '../types' import type { RunCostTotal } from '../cost/ledger' export type CriticalRecord = { issue: IssueRef; input: SendInput } @@ -198,6 +198,8 @@ export type DispatchLifecycle = { agents: DispatchLifecycleAgent[] invocationIds: string[] result?: import('../types').DispatchResult + /** Durable provider claim status retained across dispatch-owner restarts. */ + dispatchClaim?: FactoryDispatchClaimStatus /** All repository-specific PR receipts for team dispatches. `pullRequest` remains the primary receipt for compatibility. */ pullRequests?: import('./mount').GithubPublishPullRequestResult[] pullRequest?: import('./mount').GithubPublishPullRequestResult diff --git a/src/types.ts b/src/types.ts index c28a363..3587f00 100644 --- a/src/types.ts +++ b/src/types.ts @@ -137,6 +137,30 @@ export interface FactoryInFlightRegistryAgent { // Remote (relay-backend) placement facts; pids are meaningless off-machine. invocationId?: string node?: string + /** Durable-claim visibility independent of the provider writeback surface. */ + dispatchClaim?: FactoryDispatchClaimStatus +} + +export interface FactoryDispatchClaimStatus { + state: 'pending' | 'verified' | 'degraded' + write?: string + attempts?: number + maxAttempts?: number + error?: string + deadLettered?: boolean + updatedAtMs: number +} + +export interface FactoryInFlightDispatchStatus { + issue: IssueRef + agents: Array<{ + name: string + role?: AgentSpec['role'] + sessionRef?: string + invocationId?: string + node?: string + }> + claim: FactoryDispatchClaimStatus } export interface FactoryInFlightRegistryProcess { @@ -210,6 +234,8 @@ export interface DispatchResult { export interface FactoryStatus { inFlight: IssueRef[] + /** Registry-backed issue/agent ownership, including degraded GitHub claims. */ + inFlightDispatches?: FactoryInFlightDispatchStatus[] queued: IssueRef[] parked?: Array<{ issue: IssueRef diff --git a/src/writeback/github.ts b/src/writeback/github.ts index 02f5ff9..390e63c 100644 --- a/src/writeback/github.ts +++ b/src/writeback/github.ts @@ -223,6 +223,10 @@ export class GhCliGithubWriteback implements GithubWriteback { if (editArgs.length > 5) { await this.#run(editArgs) } + const confirmed = await this.#issueLabels(ref) + if (confirmed.has(inProgress.name.toLowerCase())) { + throw new Error(`GitHub writeback did not confirm removal of ${inProgress.name} on ${ref.repo}#${ref.number}`) + } return } const target = STATUS_LABELS[status] @@ -256,6 +260,11 @@ export class GhCliGithubWriteback implements GithubWriteback { if (editArgs.length > 5) { await this.#run(editArgs) } + const confirmed = await this.#issueLabels(ref) + if (confirmed.has(target.name.toLowerCase()) && !confirmed.has(previous.name.toLowerCase())) { + return + } + throw new Error(`GitHub writeback did not confirm ${target.name} on ${ref.repo}#${ref.number}`) } async #issueLabels(ref: { repo: string; number: number }): Promise> { diff --git a/src/writeback/writeback.test.ts b/src/writeback/writeback.test.ts index edca822..95eb882 100644 --- a/src/writeback/writeback.test.ts +++ b/src/writeback/writeback.test.ts @@ -936,24 +936,28 @@ describe('GhCliGithubWriteback', () => { ['label', 'create', 'factory:in-progress', '--repo', 'AgentWorkforce/factory', '--color', '1d76db', '--description', 'Factory agents are working on this issue.', '--force'], ['issue', 'view', '48', '--repo', 'AgentWorkforce/factory', '--json', 'labels'], ['issue', 'edit', '48', '--repo', 'AgentWorkforce/factory', '--add-label', 'factory:in-progress'], + ['issue', 'view', '48', '--repo', 'AgentWorkforce/factory', '--json', 'labels'], ['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'], ]) }) it('clears stale lifecycle labels when returning an orphaned issue to ready', async () => { const calls: string[][] = [] + const labels = new Set(['factory-ready', 'factory:in-progress', 'factory:human-review']) const github = new GhCliGithubWriteback({ runner: async (args) => { calls.push(args) if (args[0] === 'issue' && args[1] === 'view') { return { - stdout: JSON.stringify({ - labels: [{ name: 'factory-ready' }, { name: 'factory:in-progress' }, { name: 'factory:human-review' }], - }), + stdout: JSON.stringify({ labels: [...labels].map((name) => ({ name })) }), } } + if (args[0] === 'issue' && args[1] === 'edit' && args.includes('--remove-label')) { + labels.delete(args[args.indexOf('--remove-label') + 1]!) + } return { stdout: '' } }, }) @@ -963,9 +967,28 @@ describe('GhCliGithubWriteback', () => { expect(calls).toEqual([ ['issue', 'view', '48', '--repo', 'AgentWorkforce/factory', '--json', 'labels'], ['issue', 'edit', '48', '--repo', 'AgentWorkforce/factory', '--remove-label', 'factory:in-progress'], + ['issue', 'view', '48', '--repo', 'AgentWorkforce/factory', '--json', 'labels'], ]) }) + it('rejects an acknowledged lifecycle edit when provider read-back never shows the label', async () => { + let edits = 0 + const github = new GhCliGithubWriteback({ + runner: async (args) => { + if (args[0] === 'issue' && args[1] === 'view') { + return { stdout: JSON.stringify({ labels: [] }) } + } + if (args[0] === 'issue' && args[1] === 'edit') edits += 1 + return { stdout: '' } + }, + }) + + await expect(github.setStatus(githubIssue, 'in-progress')).rejects.toThrow( + 'GitHub writeback did not confirm factory:in-progress on AgentWorkforce/factory#48', + ) + expect(edits).toBe(1) + }) + it('treats provider human-review status as authoritative over a stale in-progress label', async () => { const github = new GhCliGithubWriteback({ runner: async () => ({