diff --git a/src/cli/fleet.test.ts b/src/cli/fleet.test.ts index a3ed48e7..09f1d01d 100644 --- a/src/cli/fleet.test.ts +++ b/src/cli/fleet.test.ts @@ -14,9 +14,9 @@ import type { FactoryIntegrationConnections, FactoryIntegrationProvider, FactoryPorts, - createFactory, } from '../index' import { + createFactory, FactoryConfigSchema, FileFactoryCloudEventOutbox, LiveDispatchStateChangedError, @@ -25,9 +25,10 @@ import { import { MountAuthScopeError, mountAuthRemediation } from '../mount/mount-auth-error' import { DocumentStateStore, FileStateStore } from '../state/file-state-store' import { FakeFleetClient, FakeMountClient, withDeadline } from '../testing' -import type { GithubConnectionRead, GithubConnectionWrite, GithubIssueLookup, LocalMountOptions, SpawnInput, SpawnResult } from '../ports' +import type { GithubConnectionRead, GithubConnectionWrite, GithubIssueLookup, GithubWriteback, LocalMountOptions, SpawnInput, SpawnResult } from '../ports' import type { HarnessDriverClientLike } from '../fleet/internal-fleet-client' import { factoryGithubIssueCommentDraftName } from '../github/writeback-paths' +import { GhCliGithubWriteback } from '../writeback/github' import { ensureLocalMount as runLocalMountPreflight } from '../mount/local-mount-preflight' import { formatLogArgs, installFactoryStopSignalHandlers, parseFleetCommand, parseGithubIssueSelector, parseGlobalOptions, reportFactoryVersionDrift, resolveBrokerConnectionPath, resolveFactoryBrokerConnectionPath, runFleetCli } from './fleet' @@ -1980,10 +1981,656 @@ describe('fleet CLI runtime', () => { } }) - // factory#319 regression guard. The self-race itself is exercised by - // "keeps relay dispatch ownership …" above, which failed ~50% of runs on - // main and is deterministic with the fix; this control pins the other half - // of the contract — that a park this dispatch did NOT make still aborts. + // factory#319. The state write becomes locally visible before provider + // confirmation returns. Pin the post-spawn read in that interval and test + // both outcomes: confirmation lets the completed dispatch converge; a + // rejected confirmation MUST NOT excuse the visible park. + it.each([ + { confirmation: 'acked' as const, expectedExit: 0 }, + { confirmation: 'failed' as const, expectedExit: 3 }, + ])('waits for its in-flight park confirmation ($confirmation)', async ({ confirmation, expectedExit }) => { + const root = await mkdtemp(join(tmpdir(), 'fleet-cli-park-inflight-')) + try { + const configPath = await writeConfig(root, { + loop: { + heartbeatPath: join(root, 'heartbeat.json'), + registryPath: join(root, 'registry.json'), + heartbeatStaleMs: 10_000, + }, + }) + const githubWrite: GithubConnectionWrite = { + publishPullRequest: async (input) => ({ + repo: input.repo, + number: 79, + url: 'https://github.com/AgentWorkforce/pear/pull/79', + headRef: input.headRef ?? 'unexpected-local-head', + }), + closePullRequest: async () => undefined, + } + class ControlledCompletingRemoteFleetClient extends FakeFleetClient { + override readonly placementLocality = 'remote' as const + implementerName?: string + exitEmitted = false + + override async spawn(input: SpawnInput): Promise { + const result = await super.spawn(input) + if (input.name.includes('-impl-')) this.implementerName = input.name + return { ...result, node: 'sf-mini', locality: 'remote' } + } + } + const fleet = new ControlledCompletingRemoteFleetClient() + // `setState` writes `stateId` at the top level of the record; the initial + // fixture carries it under `payload`. Read either. + const stateOf = (entry: { content: unknown } | undefined): string | undefined => { + const content = entry?.content as { stateId?: string; payload?: { stateId?: string } } | undefined + return content?.stateId ?? content?.payload?.stateId + } + + class ParkConfirmBlockingMount extends FakeMountClient { + parkWritten = false + releaseConfirm?: () => void + reReadSeen = false + + override async writeFile(path: string, content: unknown, opts?: { guarded?: boolean }): Promise { + await super.writeFile(path, content, opts) + if (path === issuePath && stateOf(this.files.get(path)) === TEST_STATE_IDS.humanReview) { + this.parkWritten = true + } + } + + override async readFile(path: string): Promise<{ content: unknown; revision?: string }> { + if (path === issuePath && fleet.implementerName && !fleet.exitEmitted) { + // This is the dispatch's post-spawn re-read. Start completion here + // instead of hoping a timer wins the race, then hold this read + // until completion has made the parked state visible. Completion's + // own reads pass through because `exitEmitted` is already true. + fleet.exitEmitted = true + fleet.emitAgentExit(fleet.implementerName, 'exited') + await vi.waitFor(() => expect(this.parkWritten).toBe(true), { timeout: 1_000 }) + this.reReadSeen = true + queueMicrotask(() => this.releaseConfirm?.()) + } + if (path === issuePath && this.parkWritten && confirmation === 'acked') { + if (!this.releaseConfirm) { + // First issue read after the park is `setState`'s own readback + // confirmation. Hold it, so `setState` cannot return. Bounded so + // a wrong assumption about ordering fails loudly rather than + // hanging the suite. + await new Promise((resolve) => { + this.releaseConfirm = resolve + setTimeout(resolve, 400) + }) + } + } + return super.readFile(path) + } + + override async confirmWrite(path: string): Promise<'acked' | 'pending' | 'failed' | 'timeout'> { + if (path === issuePath && this.parkWritten) return confirmation + return super.confirmWrite(path) + } + } + + const mount = new ParkConfirmBlockingMount({ + [issuePath]: issueFile, + '/github/repos/AgentWorkforce/pear/meta.json': { default_branch: 'main' }, + }, githubWrite) + const output = buffer() + + const code = await runFleetCli([ + 'dispatch', + 'AR-77', + '--backend', + 'relay', + '--config', + configPath, + ], { + fleet, + mount, + stdout: output, + stderr: buffer(), + probePrGhRunner: async () => ({ stdout: '[]' }), + }) + + // Both must hold or the test is not exercising the window at all. + expect(mount.parkWritten).toBe(true) + expect(mount.reReadSeen).toBe(true) + expect(code).toBe(expectedExit) + const releaseReasons = fleet.releases.map((release) => release.reason) + if (confirmation === 'acked') { + expect(releaseReasons).not.toContain('live dispatch state changed') + } else { + expect(releaseReasons).toContain('live dispatch state changed') + } + } finally { + await rm(root, { recursive: true, force: true }) + } + }) + + it.each([ + { receipt: 'explicit already-matched' as const, legacyVoid: false }, + { receipt: 'legacy void' as const, legacyVoid: true }, + ])('does not attribute a third-party GitHub park from an in-flight $receipt receipt', async ({ legacyVoid }) => { + const root = await mkdtemp(join(tmpdir(), 'fleet-cli-github-foreign-park-inflight-')) + try { + const configPath = await writeConfig(root, { + issueSource: 'github', + loop: { + heartbeatPath: join(root, 'heartbeat.json'), + registryPath: join(root, 'registry.json'), + heartbeatStaleMs: 10_000, + }, + }) + const githubPath = '/github/repos/AgentWorkforce__pear/issues/by-id/48.json' + const githubWrite: GithubConnectionWrite = { + publishPullRequest: async (input) => ({ + repo: input.repo, + number: 80, + url: 'https://github.com/AgentWorkforce/pear/pull/80', + headRef: input.headRef ?? 'unexpected-local-head', + }), + closePullRequest: async () => undefined, + } + class ControlledCompletingGithubFleetClient extends FakeFleetClient { + override readonly placementLocality = 'remote' as const + implementerName?: string + exitEmitted = false + + override async spawn(input: SpawnInput): Promise { + const result = await super.spawn(input) + if (input.name.includes('-impl-')) this.implementerName = input.name + return { ...result, node: 'sf-mini', locality: 'remote' } + } + } + const fleet = new ControlledCompletingGithubFleetClient() + let thirdPartyParked = false + let terminalLabelProvisioned = false + let terminalViews = 0 + let terminalEdits = 0 + let releaseConfirmation: (() => void) | undefined + + class ThirdPartyGithubParkMount extends FakeMountClient { + postSpawnReadSawPark = false + + parkAsThirdParty(): void { + const existing = this.files.get(githubPath) + const content = existing?.content as ReturnType + const labels = content.payload.labels + .filter((label) => label.name !== 'factory:in-progress') + labels.push({ name: 'factory:human-review' }) + this.files.set(githubPath, { + ...existing, + content: { + ...content, + payload: { ...content.payload, labels }, + }, + }) + thirdPartyParked = true + } + + override async readFile(path: string): Promise<{ content: unknown; revision?: string }> { + if (path === githubPath && fleet.implementerName && !fleet.exitEmitted) { + // This is dispatch's post-spawn read. Completion starts here and + // registers its fence before the CLI adapter's first label read. + fleet.exitEmitted = true + fleet.emitAgentExit(fleet.implementerName, 'exited') + await vi.waitFor(() => expect(thirdPartyParked).toBe(true), { timeout: 1_000 }) + const result = await super.readFile(path) + this.postSpawnReadSawPark = true + // Let the adapter's provider-confirmation read finish only after + // the dispatch has observed the foreign state while the fence is + // still registered. + setTimeout(() => releaseConfirmation?.(), 0) + return result + } + return super.readFile(path) + } + } + + const githubContent = githubIssueFile('pear') + githubContent.payload.title = '[factory-e2e] preserve foreign GitHub lifecycle ownership' + githubContent.payload.body = [ + 'Fix the completion race without attributing a third-party lifecycle change to Factory.', + '', + 'Acceptance criteria:', + '- A skipped GitHub CLI label edit never stamps Factory authorship.', + '- The live dispatch state-change guard still releases the remote agent.', + '- Add a deterministic integration regression for the provider-read ordering.', + ].join('\n') + const mount = new ThirdPartyGithubParkMount({ + [githubPath]: githubContent, + '/github/repos/AgentWorkforce/pear/meta.json': { default_branch: 'main' }, + }, githubWrite) + const providerLabels = new Set(['factory', 'pear']) + const githubWriteback = new GhCliGithubWriteback({ + runner: async (args) => { + if (args[0] === 'label' && args[1] === 'create') { + terminalLabelProvisioned = args[2] === 'factory:human-review' + return { stdout: '' } + } + if (args[0] === 'issue' && args[1] === 'view') { + if (terminalLabelProvisioned) { + terminalViews += 1 + if (terminalViews === 1) { + // The third party wins before the adapter's first read. The + // target is present and the previous label absent, so the CLI + // must skip `gh issue edit` and report already-matched. + providerLabels.delete('factory:in-progress') + providerLabels.add('factory:human-review') + mount.parkAsThirdParty() + } else if (terminalViews === 2 && !releaseConfirmation) { + await new Promise((resolve) => { + releaseConfirmation = resolve + setTimeout(resolve, 400) + }) + } + } + return { stdout: JSON.stringify({ labels: [...providerLabels].map((name) => ({ name })) }) } + } + if (args[0] === 'issue' && args[1] === 'edit') { + if (terminalLabelProvisioned) terminalEdits += 1 + const added = args[args.indexOf('--add-label') + 1] + const removed = args[args.indexOf('--remove-label') + 1] + if (args.includes('--add-label') && added) providerLabels.add(added) + if (args.includes('--remove-label') && removed) providerLabels.delete(removed) + } + return { stdout: '' } + }, + }) + const effectiveGithubWriteback: GithubWriteback = legacyVoid + ? { + getIssueStatus: async (target) => await githubWriteback.getIssueStatus(target), + postComment: async (target, body) => await githubWriteback.postComment(target, body), + setStatus: async (target, status) => { + await githubWriteback.setStatus(target, status) + }, + closeIssue: async (target, body) => await githubWriteback.closeIssue(target, body), + } + : githubWriteback + const output = buffer() + const errors = buffer() + + const code = await runFleetCli([ + 'dispatch', + '48', + '--backend', + 'relay', + '--config', + configPath, + ], { + fleet, + mount, + createFactory: (factoryConfig, ports) => createFactory(factoryConfig, { + ...ports, + githubWriteback: effectiveGithubWriteback, + }), + stdout: output, + stderr: errors, + probePrGhRunner: async () => ({ stdout: '[]' }), + }) + + expect(thirdPartyParked).toBe(true) + expect(mount.postSpawnReadSawPark).toBe(true) + expect(terminalEdits).toBe(0) + expect(code).toBe(3) + expect(errors.text()).toContain('Live state changed before writeback for 48') + expect(fleet.releases.map((release) => release.reason)).toContain('live dispatch state changed') + if (legacyVoid) { + expect(errors.text()).toContain('GitHub writeback returned no ownership receipt') + expect(errors.text()).toContain('"method":"setStatus"') + } else { + expect(errors.text()).not.toContain('GitHub writeback returned no ownership receipt') + } + } finally { + await rm(root, { recursive: true, force: true }) + } + }) + + it.each([ + { receipt: 'explicit acknowledged' as const, legacyVoid: false }, + { receipt: 'legacy void' as const, legacyVoid: true }, + ])('does not attribute a third-party GitHub close from an in-flight $receipt receipt', async ({ legacyVoid }) => { + const root = await mkdtemp(join(tmpdir(), 'fleet-cli-github-foreign-close-inflight-')) + try { + const configPath = await writeConfig(root, { + issueSource: 'github', + terminalState: 'done', + loop: { + heartbeatPath: join(root, 'heartbeat.json'), + registryPath: join(root, 'registry.json'), + heartbeatStaleMs: 10_000, + }, + }) + const githubPath = '/github/repos/AgentWorkforce__pear/issues/by-id/48.json' + const githubWrite: GithubConnectionWrite = { + publishPullRequest: async (input) => ({ + repo: input.repo, + number: 80, + url: 'https://github.com/AgentWorkforce/pear/pull/80', + headRef: input.headRef ?? 'unexpected-local-head', + }), + closePullRequest: async () => undefined, + } + class ControlledCompletingGithubFleetClient extends FakeFleetClient { + override readonly placementLocality = 'remote' as const + implementerName?: string + exitEmitted = false + + override async spawn(input: SpawnInput): Promise { + const result = await super.spawn(input) + if (input.name.includes('-impl-')) this.implementerName = input.name + return { ...result, node: 'sf-mini', locality: 'remote' } + } + } + const fleet = new ControlledCompletingGithubFleetClient() + let thirdPartyClosed = false + + class ThirdPartyGithubCloseMount extends FakeMountClient { + postSpawnReadSawClose = false + + closeAsThirdParty(): void { + const existing = this.files.get(githubPath) + const content = existing?.content as ReturnType + this.files.set(githubPath, { + ...existing, + content: { + ...content, + payload: { ...content.payload, state: 'closed' }, + }, + }) + thirdPartyClosed = true + } + + override async readFile(path: string): Promise<{ content: unknown; revision?: string }> { + if (path === githubPath && fleet.implementerName && !fleet.exitEmitted) { + // Pin the dispatch's post-spawn read behind completion. The + // provider close command returns idempotently after another actor + // creates the terminal event, so only the audit receipt can tell + // the dispatch that it must not claim ownership. + fleet.exitEmitted = true + fleet.emitAgentExit(fleet.implementerName, 'exited') + await vi.waitFor(() => expect(thirdPartyClosed).toBe(true), { timeout: 1_000 }) + this.postSpawnReadSawClose = true + } + return super.readFile(path) + } + } + + const githubContent = githubIssueFile('pear') + githubContent.payload.title = '[factory-e2e] preserve foreign GitHub close ownership' + githubContent.payload.body = [ + 'Keep terminal lifecycle ownership provider-authoritative during concurrent close operations.', + '', + 'Acceptance criteria:', + '- An idempotent close must not claim Factory ownership when another actor creates the close event.', + '- The post-spawn live-state guard must abort the dispatch.', + '- Remote agents must be released with the live-state-change reason.', + ].join('\n') + const mount = new ThirdPartyGithubCloseMount({ + [githubPath]: githubContent, + '/github/repos/AgentWorkforce/pear/meta.json': { default_branch: 'main' }, + '/github/repos/AgentWorkforce/pear/pulls/80/metadata.json': { + provider: 'github', + objectType: 'pull_request', + objectId: '80', + payload: { + number: 80, + title: '[factory-e2e] preserve foreign GitHub close ownership', + body: 'Fixes #48', + head_ref: 'factory/48-agentworkforce-pear', + isDraft: false, + state: 'MERGED', + merged: true, + }, + }, + }, githubWrite) + let providerState = 'OPEN' + const stateEvents: string[] = [] + let closeCommands = 0 + const githubWriteback = new GhCliGithubWriteback({ + runner: async (args) => { + if (args[0] === 'issue' && args[1] === 'view' && args.includes('state')) { + return { stdout: JSON.stringify({ state: providerState }) } + } + if (args[0] === 'api' && args[1] === 'user') return { stdout: 'factory-bot\n' } + if (args[0] === 'api' && args[1] === '--paginate') { + return { stdout: stateEvents.join('\n') } + } + if (args[0] === 'issue' && args[1] === 'close') { + closeCommands += 1 + providerState = 'CLOSED' + stateEvents.push('1\tclosed\tother-user') + mount.closeAsThirdParty() + } + return { stdout: '' } + }, + }) + const effectiveGithubWriteback: GithubWriteback = legacyVoid + ? { + getIssueStatus: async (target) => await githubWriteback.getIssueStatus(target), + postComment: async (target, body) => await githubWriteback.postComment(target, body), + setStatus: async (target, status) => await githubWriteback.setStatus(target, status), + closeIssue: async (target, body) => { + await githubWriteback.closeIssue(target, body) + }, + } + : githubWriteback + const output = buffer() + const errors = buffer() + + const code = await runFleetCli([ + 'dispatch', + '48', + '--backend', + 'relay', + '--config', + configPath, + ], { + fleet, + mount, + createFactory: (factoryConfig, ports) => createFactory(factoryConfig, { + ...ports, + githubWriteback: effectiveGithubWriteback, + }), + stdout: output, + stderr: errors, + probePrGhRunner: async () => ({ + stdout: JSON.stringify([{ + number: 80, + title: '[factory-e2e] preserve foreign GitHub close ownership', + body: 'Fixes #48', + headRefName: 'factory/48-agentworkforce-pear', + isDraft: false, + state: 'MERGED', + }]), + }), + }) + + expect(thirdPartyClosed, JSON.stringify({ + code, + output: output.text(), + errors: errors.text(), + releases: fleet.releases, + })).toBe(true) + expect(mount.postSpawnReadSawClose).toBe(true) + expect(closeCommands).toBe(1) + expect(code).toBe(3) + expect(errors.text()).toContain('Live state changed before writeback for 48') + expect(fleet.releases.map((release) => release.reason)).toContain('live dispatch state changed') + if (legacyVoid) { + expect(errors.text()).toContain('GitHub writeback returned no ownership receipt') + expect(errors.text()).toContain('"method":"closeIssue"') + } else { + expect(errors.text()).not.toContain('GitHub writeback returned no ownership receipt') + } + } finally { + await rm(root, { recursive: true, force: true }) + } + }) + + it('serializes terminal completion that starts after the ready read against the dispatch claim', async () => { + const root = await mkdtemp(join(tmpdir(), 'fleet-cli-claim-completion-fence-')) + let releaseClaim: (() => void) | undefined + try { + const configPath = await writeConfig(root, { + issueSource: 'github', + terminalState: 'human-review', + loop: { + heartbeatPath: join(root, 'heartbeat.json'), + registryPath: join(root, 'registry.json'), + heartbeatStaleMs: 10_000, + }, + }) + const githubPath = '/github/repos/AgentWorkforce__pear/issues/by-id/48.json' + class CompletingDuringClaimFleetClient extends FakeFleetClient { + override readonly placementLocality = 'remote' as const + implementerName?: string + exitEmitted = false + + override async spawn(input: SpawnInput): Promise { + const result = await super.spawn(input) + if (input.name.includes('-impl-')) this.implementerName = input.name + return { ...result, node: 'sf-mini', locality: 'remote' } + } + } + const fleet = new CompletingDuringClaimFleetClient() + let resolveCompletionRead!: () => void + const completionRead = new Promise((resolve) => { + resolveCompletionRead = resolve + }) + class CompletionObservedMount extends FakeMountClient { + completionReadSeen = false + + override async readFile(path: string): Promise<{ content: unknown; revision?: string }> { + if (path === githubPath && fleet.exitEmitted && !this.completionReadSeen) { + this.completionReadSeen = true + resolveCompletionRead() + } + return await super.readFile(path) + } + } + const githubWrite: GithubConnectionWrite = { + publishPullRequest: async (input) => ({ + repo: input.repo, + number: 80, + url: 'https://github.com/AgentWorkforce/pear/pull/80', + headRef: input.headRef ?? 'unexpected-local-head', + }), + closePullRequest: async () => undefined, + } + const githubContent = githubIssueFile('pear') + githubContent.payload.title = '[factory-e2e] serialize dispatch claim with terminal completion' + githubContent.payload.body = [ + 'Prevent a stale post-spawn ready snapshot from overwriting a concurrent terminal writeback.', + '', + 'Acceptance criteria:', + '- Completion that starts after the ready read waits for the dispatch claim write.', + '- The final provider state remains human review.', + '- The agent is released with the terminal completion reason.', + ].join('\n') + const mount = new CompletionObservedMount({ + [githubPath]: githubContent, + '/github/repos/AgentWorkforce/pear/meta.json': { default_branch: 'main' }, + }, githubWrite) + let resolveClaimStarted!: () => void + const claimStarted = new Promise((resolve) => { + resolveClaimStarted = resolve + }) + const claimRelease = new Promise((resolve) => { + releaseClaim = resolve + }) + const statusCalls: string[] = [] + let terminalStarted = false + const output = buffer() + const errors = buffer() + const setMountedStatus = (status: 'in-progress' | 'human-review'): void => { + const existing = mount.files.get(githubPath) + const content = existing?.content as ReturnType + const labels = content.payload.labels.filter((label) => + label.name !== 'factory:in-progress' && label.name !== 'factory:human-review') + labels.push({ name: `factory:${status}` }) + mount.files.set(githubPath, { + ...existing, + content: { ...content, payload: { ...content.payload, labels } }, + }) + } + const githubWriteback: GithubWriteback = { + getIssueStatus: async () => undefined, + postComment: async () => undefined, + setStatus: async (_issue, status) => { + statusCalls.push(`start:${status}`) + if (status === 'in-progress') { + resolveClaimStarted() + if (!fleet.implementerName) throw new Error('implementer was not spawned before the dispatch claim') + fleet.exitEmitted = true + fleet.emitAgentExit(fleet.implementerName, 'exited') + await claimRelease + } else if (status === 'human-review') { + terminalStarted = true + } + setMountedStatus(status) + statusCalls.push(`applied:${status}`) + return 'applied' + }, + closeIssue: async () => 'acknowledged', + } + + const run = runFleetCli([ + 'dispatch', + '48', + '--backend', + 'relay', + '--config', + configPath, + ], { + fleet, + mount, + createFactory: (factoryConfig, ports) => createFactory(factoryConfig, { + ...ports, + githubWriteback, + }), + stdout: output, + stderr: errors, + probePrGhRunner: async () => ({ stdout: '[]' }), + }) + + try { + await withDeadline(claimStarted, 1_000, 'dispatch claim did not start') + } catch (error) { + throw new Error(JSON.stringify({ + error: error instanceof Error ? error.message : String(error), + output: output.text(), + errors: errors.text(), + statusCalls, + spawns: fleet.spawns, + releases: fleet.releases, + })) + } + await withDeadline(completionRead, 1_000, 'completion did not reach its issue read') + await new Promise((resolve) => setTimeout(resolve, 10)) + const terminalStartedBeforeClaimReleased = terminalStarted + releaseClaim() + const code = await run + + expect(fleet.exitEmitted).toBe(true) + expect(mount.completionReadSeen).toBe(true) + expect(terminalStartedBeforeClaimReleased).toBe(false) + expect(statusCalls).toEqual([ + 'start:in-progress', + 'applied:in-progress', + 'start:human-review', + 'applied:human-review', + ]) + expect(code).toBe(0) + expect((mount.files.get(githubPath)?.content as ReturnType).payload.labels) + .toContainEqual({ name: 'factory:human-review' }) + expect(fleet.releases.map((release) => release.reason)).toContain('issue-human-review') + } finally { + releaseClaim?.() + await rm(root, { recursive: true, force: true }) + } + }) + it('still aborts when a third party parks the issue and this dispatch did not', async () => { const root = await mkdtemp(join(tmpdir(), 'fleet-cli-foreign-park-')) try { diff --git a/src/index.ts b/src/index.ts index 2f7e931d..3eb8b5ad 100644 --- a/src/index.ts +++ b/src/index.ts @@ -352,7 +352,9 @@ export type { SpawnResult, AgentUsage, GithubRead, + GithubIssueCloseWriteResult, GithubIssueStatus, + GithubStatusWriteResult, GithubWriteback, LinearWriteback, Logger, diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 8dcf790c..27bd6635 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -949,6 +949,22 @@ class RemoteLifecycleFleetClient extends FakeFleetClient { } } +class LaterSpawnHangingFleetClient extends RemoteLifecycleFleetClient { + readonly laterSpawnGate = Promise.withResolvers() + terminalExitEmitted = false + + override async spawn(input: SpawnInput): Promise { + if (input.name.includes('-review') && !this.terminalExitEmitted) { + const implementer = this.spawns.find((spawn) => spawn.name.includes('-impl-')) + if (!implementer) throw new Error('reviewer spawn started before the implementer placement') + this.terminalExitEmitted = true + this.emitAgentExit(implementer.name, 'issue-done') + await this.laterSpawnGate.promise + } + return await super.spawn(input) + } +} + class DurableRemoteLifecycleFleetClient extends RemoteLifecycleFleetClient { override readonly durableOwnership = true } @@ -10970,6 +10986,109 @@ describe('FactoryLoop', () => { } }) + it('settles post-spawn completion waits when a later spawn reaches the held-agent deadline', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-post-spawn-deadline-')) + const number = 1252 + const path = githubIssuePath('AgentWorkforce', 'pear', number) + const fleet = new LaterSpawnHangingFleetClient() + const githubWriteback = new RecordingGithubWriteback() + const factory = createFactory(config({ + issueSource: 'github', + mergePolicy: 'never', + terminalState: 'human-review', + dispatch: { agentHoldTimeoutMs: 1_000, agentlessHoldTimeoutMs: 60_000 }, + loop: { + heartbeatPath: join(root, 'heartbeat.json'), + registryPath: join(root, 'registry.json'), + }, + }), { + mount: new FakeMountClient({ + [path]: githubIssueFile(number, { labels: ['factory', 'pear'] }), + }), + fleet, + stateStore: new FileStateStore({ batchSize: 2, watchStatePath: join(root, 'state.json') }), + triage: new StaticTriage(), + githubWriteback, + }) + const run = factory.runOnce().catch((error: unknown) => error) + let stopped = false + try { + await vi.waitFor(() => expect(fleet.terminalExitEmitted).toBe(true), { timeout: 4_000 }) + await vi.waitFor(() => expect(githubWriteback.statuses).toContainEqual({ + key: String(number), + status: 'human-review', + }), { timeout: 4_000 }) + // Completion has written provider terminal state and is now waiting for + // the post-spawn observation, while the reviewer spawn never returns. + await vi.waitFor(() => expect(factory.status().counters.postSpawnWaitsSettledByAbandonment) + .toBe(1), { timeout: 8_000 }) + await vi.waitFor(() => expect(fleet.releases).toContainEqual({ + name: `ar-${number}-impl-pear`, + reason: 'held-past-deadline', + }), { timeout: 4_000 }) + + // The deadline settlement must have drained the completion exit before + // shutdown supplies its independent settlement backstop. + await withDeadline(factory.stop(), 2_000, 'stop remained blocked after held-agent abandonment') + stopped = true + expect(factory.status().counters.postSpawnWaitsSettledByStop).toBeUndefined() + + fleet.laterSpawnGate.resolve() + await withDeadline(run, 8_000, 'late reviewer spawn did not unwind after abandonment') + } finally { + fleet.laterSpawnGate.resolve() + if (!stopped) await factory.stop() + await rm(root, { recursive: true, force: true }) + } + }, 30_000) + + it('settles post-spawn completion waits before shutdown drains agent exits', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-post-spawn-stop-')) + const number = 1253 + const path = githubIssuePath('AgentWorkforce', 'pear', number) + const fleet = new LaterSpawnHangingFleetClient() + const githubWriteback = new RecordingGithubWriteback() + const factory = createFactory(config({ + issueSource: 'github', + mergePolicy: 'never', + terminalState: 'human-review', + dispatch: { agentHoldTimeoutMs: 60_000, agentlessHoldTimeoutMs: 60_000 }, + loop: { + heartbeatPath: join(root, 'heartbeat.json'), + registryPath: join(root, 'registry.json'), + }, + }), { + mount: new FakeMountClient({ + [path]: githubIssueFile(number, { labels: ['factory', 'pear'] }), + }), + fleet, + stateStore: new FileStateStore({ batchSize: 2, watchStatePath: join(root, 'state.json') }), + triage: new StaticTriage(), + githubWriteback, + }) + const run = factory.runOnce().catch((error: unknown) => error) + let stopped = false + try { + await vi.waitFor(() => expect(fleet.terminalExitEmitted).toBe(true), { timeout: 4_000 }) + await vi.waitFor(() => expect(githubWriteback.statuses).toContainEqual({ + key: String(number), + status: 'human-review', + }), { timeout: 4_000 }) + expect(factory.status().counters.postSpawnWaitsSettledByAbandonment).toBeUndefined() + + await withDeadline(factory.stop(), 2_000, 'stop remained blocked on a post-spawn completion wait') + stopped = true + expect(factory.status().counters.postSpawnWaitsSettledByStop).toBe(1) + + fleet.laterSpawnGate.resolve() + await withDeadline(run, 8_000, 'late reviewer spawn did not unwind after stop') + } finally { + fleet.laterSpawnGate.resolve() + if (!stopped) await factory.stop() + await rm(root, { recursive: true, force: true }) + } + }, 30_000) + // #303: a lifecycle that reached a slot-occupying phase and never had a // live agent placement has no `heldSinceAtMs`, so both halves of the // held-agent reaper skipped it. With batchSize 1 that one row held the only diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 29087793..f96ee73e 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -603,6 +603,19 @@ class ReadinessReconcileTimeoutError extends Error { } } +interface PostSpawnDispatchClaimFence { + completionAtWriteBoundary: boolean + claimStarted: boolean + accepted?: boolean + settled: Promise + settle(accepted: boolean): void +} + +interface PostSpawnIssueObservation { + settled: Promise + settle(accepted: boolean): void +} + const realClock: Clock = { now: () => Date.now(), sleep: (ms) => new Promise((resolve) => setTimeout(resolve, ms)), @@ -816,6 +829,16 @@ export class FactoryLoop implements Factory { #previewSweepTimer?: ReturnType #previewSweepInFlight?: Promise readonly #completionInFlight = new Set() + readonly #issueWritebackInFlight = new Map>() + // A fast completion can make terminal issue state visible while dispatch is + // still performing its post-spawn readiness read. Completion publishes its + // provider receipt, then waits for that read to classify the state as owned + // or foreign before it releases agents under a terminal success reason. + readonly #postSpawnIssueObservations = new Map() + // The post-spawn ready read and following dispatch claim form one local + // write boundary. A completion arriving first makes dispatch wait and + // re-read; one arriving after claim entry waits for the claim to finish. + readonly #postSpawnDispatchClaimFences = new Map() readonly #agentExitsInFlight = new Map>() #reconciledAgentExitsActive = 0 readonly #reconciledAgentExitWaiters: Array<() => void> = [] @@ -1303,6 +1326,15 @@ export class FactoryLoop implements Factory { async stop(): Promise { this.#started = false this.#stopping = true + // These waits are armed before every planned spawn returns. Resolve them + // before the first shutdown await: a fast first agent may be completing + // while a later spawn is hung, and #drainAgentExitsInFlight would otherwise + // wait forever on a resolver stranded inside that dispatch stack. + const postSpawnWaitKeys = new Set([ + ...this.#postSpawnIssueObservations.keys(), + ...this.#postSpawnDispatchClaimFences.keys(), + ]) + for (const key of postSpawnWaitKeys) this.#settlePostSpawnDispatchWaits(key, false, 'stop') if (this.#babysitterResourceDeliveryRetryTimer) clearTimeout(this.#babysitterResourceDeliveryRetryTimer) this.#babysitterResourceDeliveryRetryTimer = undefined if (this.#babysitterResourceSubscriptionRenewTimer) clearTimeout(this.#babysitterResourceSubscriptionRenewTimer) @@ -1370,6 +1402,8 @@ export class FactoryLoop implements Factory { this.#livePollInFlight = false this.#liveEventQueue.length = 0 this.#completionInFlight.clear() + this.#postSpawnIssueObservations.clear() + this.#postSpawnDispatchClaimFences.clear() this.#babysitterSpawned.clear() this.#babysitterPr.clear() this.#babysitterIssueRefs.clear() @@ -4552,6 +4586,48 @@ export class FactoryLoop implements Factory { if (!dryRun) await this.#ensureGithubAgentQuestionWatch(record, liveIssue) const spawnedForReaperHandoff: RegistryHandoffAgent[] = [] + const observationKey = issueKey(record.issue) + let resolvePostSpawnIssueObservation!: (accepted: boolean) => void + let postSpawnIssueObservationDidSettle = false + const postSpawnIssueObservationSettled = new Promise((resolve) => { + resolvePostSpawnIssueObservation = resolve + }) + let postSpawnIssueObservation!: PostSpawnIssueObservation + postSpawnIssueObservation = { + settled: postSpawnIssueObservationSettled, + settle: (accepted: boolean): void => { + if (postSpawnIssueObservationDidSettle) return + postSpawnIssueObservationDidSettle = true + resolvePostSpawnIssueObservation(accepted) + if (this.#postSpawnIssueObservations.get(observationKey) === postSpawnIssueObservation) { + this.#postSpawnIssueObservations.delete(observationKey) + } + }, + } + this.#postSpawnIssueObservations.set(observationKey, postSpawnIssueObservation) + let resolvePostSpawnDispatchClaim!: (accepted: boolean) => void + let postSpawnDispatchClaimSettled = false + const postSpawnDispatchClaim = new Promise((resolve) => { + resolvePostSpawnDispatchClaim = resolve + }) + let postSpawnDispatchClaimFence!: PostSpawnDispatchClaimFence + postSpawnDispatchClaimFence = { + completionAtWriteBoundary: false, + claimStarted: false, + settled: postSpawnDispatchClaim, + settle: (accepted: boolean): void => { + if (postSpawnDispatchClaimSettled) return + postSpawnDispatchClaimSettled = true + postSpawnDispatchClaimFence.accepted = accepted + resolvePostSpawnDispatchClaim(accepted) + if (this.#postSpawnDispatchClaimFences.get(observationKey) === postSpawnDispatchClaimFence) { + this.#postSpawnDispatchClaimFences.delete(observationKey) + } + }, + } + this.#postSpawnDispatchClaimFences.set(observationKey, postSpawnDispatchClaimFence) + const settlePostSpawnIssueObservation = postSpawnIssueObservation.settle + const settlePostSpawnDispatchClaim = postSpawnDispatchClaimFence.settle try { if (!dryRun) { const issue = await this.#readIssue(dispatchDecision.issue.path) @@ -4587,39 +4663,57 @@ export class FactoryLoop implements Factory { const comment = dispatchComment(dispatchDecision, agents) let implementingStateId: string | undefined if (!dryRun) { - const issue = await this.#readIssue(dispatchDecision.issue.path) - if (!issue || !this.#isIssueReady(issue)) { - // The agents spawned a few lines above can reach terminal before we - // get here. Their completion writeback parks the issue — Linear - // `humanReview`, or the GitHub human-review label — and stamps - // *this same record's* lifecycle on the way past. Re-reading the - // issue then shows "not ready", but the writer was us. - // - // Treating that as a foreign change is not a cosmetic misreport: the - // catch below classifies LiveDispatchStateChangedError as terminal - // and calls #releaseAndTerminateAgents, so a dispatch whose agents - // finished quickly tore down its own completed work and reported - // RETRYABLE to its supervisor (factory#319). + let issue = await this.#readIssue(dispatchDecision.issue.path) + if (postSpawnDispatchClaimFence.completionAtWriteBoundary) { + // Completion reached its terminal provider-write boundary before + // dispatch entered the claim boundary. Always re-read after its + // receipt, even when the first snapshot was ready: the terminal + // write may have started immediately after that snapshot. + const issueWriteback = this.#issueWritebackInFlight.get(issueKey(record.issue)) + if (issueWriteback) await issueWriteback + issue = await this.#readIssue(dispatchDecision.issue.path) + } else if ((!issue || !this.#isIssueReady(issue)) && record.issueWritebackConfirmedAtMs === undefined) { + // A very fast agent can finish while its own dispatch is still + // performing this post-spawn read. Do not guess authorship from the + // first visible state: an idempotent provider mutation cannot tell us + // whether Factory or another actor won the race. Instead, wait for + // this record's concurrent completion writeback to settle, then use + // only its provider-confirmed outcome (factory#319). // - // Only THIS dispatch's own confirmed issue writeback may excuse the - // change. Lifecycle phases are not enough: `publishing` is entered - // before the PR is published and `parking` before anything touches - // the issue, so a phase records local progress, not authorship — and - // a foreign park landing during those awaits would be misread as - // ours (codex review on #321). Losing the row to another owner is a - // different condition and is still caught where it always was, by - // #saveDispatchLifecycle returning false. + // The completion promise settles as soon as terminal issue writeback + // succeeds or fails, before Slack/release cleanup. If there is no + // completion in flight, this is an ordinary foreign state change and + // the existing abort remains immediate. + const issueWriteback = this.#issueWritebackInFlight.get(issueKey(record.issue)) + if (issueWriteback) { + await issueWriteback + issue = await this.#readIssue(dispatchDecision.issue.path) + } + } + if (!issue || !this.#isIssueReady(issue)) { + // Only a terminal writeback applied and confirmed by the completion + // already in flight for this exact record may excuse the change. The + // marker is stamped after provider acknowledgement/readback, never at + // the earlier locally-visible write boundary. A confirmed no-op does + // not establish authorship: another actor may have parked the issue + // before our adapter's first provider read. if (record.issueWritebackConfirmedAtMs === undefined) { throw new LiveDispatchStateChangedError(dispatchDecision.issue.key) } // The claim is moot and would be wrong to write: it would drag an // issue our own lifecycle has already parked back to `implementing`. } else { + // This assignment is synchronous with the preceding completion flag + // check. Completion either arrived first (the branch above) or will + // now observe claimStarted and wait; no await-sized gap remains. + postSpawnDispatchClaimFence.claimStarted = true implementingStateId = await this.#applyDispatchClaim(record, issue, comment) - record.issueWritebackConfirmedAtMs = this.#clock.now() + record.issueWritebackConfirmedAtMs ??= this.#clock.now() this.#emit('writeback-verified', { issue: dispatchDecision.issue, path: issue.path }) } } + settlePostSpawnDispatchClaim(true) + settlePostSpawnIssueObservation(true) const result = { issue: dispatchDecision.issue, @@ -4643,6 +4737,8 @@ export class FactoryLoop implements Factory { } return result } catch (error) { + settlePostSpawnDispatchClaim(false) + settlePostSpawnIssueObservation(!(error instanceof LiveDispatchStateChangedError)) // A spawn can fail after the broker accepted it but before its ack // reached Factory. Include every planned worktree agent, not only the // acknowledged spawns, so cleanup never races a name-only survivor. @@ -9633,6 +9729,11 @@ export class FactoryLoop implements Factory { // record leaves the batch. async #abandonStuckDispatch(record: InFlightIssue, reason: string): Promise { const key = issueKey(record.issue) + // A dispatch can be stuck inside a later spawn after an earlier agent has + // already started completion. Release both post-spawn waits before the + // first abandonment await so terminal processing cannot remain an + // absorbing promise after this lifecycle is reaped. + this.#settlePostSpawnDispatchWaits(key, false, 'abandonment') const heldPastDeadline = reason === HELD_PAST_DEADLINE_RELEASE_REASON const agentReleaseReason = heldPastDeadline ? HELD_PAST_DEADLINE_RELEASE_REASON : 'issue-abandoned' this.#abandonedDispatchReasons.set(key, reason) @@ -13891,6 +13992,21 @@ export class FactoryLoop implements Factory { return } this.#completionInFlight.add(completionKey) + const postSpawnIssueObservation = this.#postSpawnIssueObservations.get(completionKey) + const postSpawnDispatchClaimFence = this.#postSpawnDispatchClaimFences.get(completionKey) + let settleIssueWriteback!: () => void + const issueWritebackSettled = new Promise((resolve) => { + settleIssueWriteback = resolve + }) + let issueWritebackDidSettle = false + const settleIssueWritebackOnce = () => { + if (issueWritebackDidSettle) return + issueWritebackDidSettle = true + settleIssueWriteback() + if (this.#issueWritebackInFlight.get(completionKey) === issueWritebackSettled) { + this.#issueWritebackInFlight.delete(completionKey) + } + } let releaseReasonForRetry: string | undefined try { if (!await this.#assertDispatchLifecycleOwner(record)) return @@ -13940,29 +14056,77 @@ export class FactoryLoop implements Factory { const humanReview = configuredHumanReview || (githubIssue && !githubMerged) const statusLabel = humanReview ? 'In Human Review' : 'Done' if (issue) { + if (postSpawnDispatchClaimFence) { + // Publish boundary arrival before any await. Dispatch performs the + // paired synchronous claimStarted assignment after its ready read, + // so exactly one side wins: an earlier completion is awaited and + // re-read; a claim already entering its provider write finishes + // before this terminal write begins. + postSpawnDispatchClaimFence.completionAtWriteBoundary = true + if ( + (postSpawnDispatchClaimFence.claimStarted || postSpawnDispatchClaimFence.accepted !== undefined) && + !(postSpawnDispatchClaimFence.accepted ?? await postSpawnDispatchClaimFence.settled) + ) return + } + // Register only at the provider-write boundary. Work before this point + // (PR discovery/merge gating) has not changed the issue, so a foreign + // park during it must still abort immediately rather than waiting on a + // possibly long completion path. + this.#issueWritebackInFlight.set(completionKey, issueWritebackSettled) if (githubIssue) { if (humanReview) { - await this.#githubWriteback.setStatus(issue, 'human-review') + const statusWrite = await this.#githubWriteback.setStatus(issue, 'human-review') + if (statusWrite === undefined) this.#recordMissingGithubWritebackReceipt('setStatus') + // Only an explicit provider-proven transition establishes that + // this dispatch owns the visible park. Legacy void adapters and + // App acknowledgements remain deliberately untrusted. + if (statusWrite === 'applied') { + record.issueWritebackConfirmedAtMs ??= this.#clock.now() + } + // The lifecycle-state outcome is now known. Unblock the concurrent + // post-spawn read before the separate completion comment write. + settleIssueWritebackOnce() await this.#githubWriteback.postComment( issue, `Factory agents completed; this issue is awaiting human review. The pull request remains open.\n\nMerge policy: ${this.#config.mergePolicy}`, ) } else { - await this.#githubWriteback.closeIssue( + const closeWrite = await this.#githubWriteback.closeIssue( issue, 'Factory observed the linked pull request merge and completed this issue.', ) + if (closeWrite === undefined) this.#recordMissingGithubWritebackReceipt('closeIssue') + // A provider-confirmed, actor-attributed close is the only safe + // proof that this dispatch owns the visible terminal state. An + // idempotent no-op, legacy void adapter, or App acknowledgement + // may instead mean a third party won the close race. + if (closeWrite === 'applied') { + record.issueWritebackConfirmedAtMs ??= this.#clock.now() + } } } else { const targetState = humanReview ? this.#states.idFor(issueTeam, 'humanReview') : this.#states.idFor(issueTeam, 'done') await this.#linear.setState(issue, targetState) + record.issueWritebackConfirmedAtMs ??= this.#clock.now() await this.#recordCanonicalIssueState({ ...record.issue, stateId: targetState }) } - record.issueWritebackConfirmedAtMs = this.#clock.now() - this.#emit('writeback-verified', { issue: record.issue, path: issue.path }) + if (record.issueWritebackConfirmedAtMs !== undefined) { + this.#emit('writeback-verified', { issue: record.issue, path: issue.path }) + } + // Unblock a concurrent post-spawn read as soon as the issue writeback + // outcome is known. Completion still has dependency, Slack and release + // work to do; none determines whether the observed issue state is safe. + settleIssueWritebackOnce() + // If completion raced dispatch's post-spawn read, let that read consume + // the receipt before claiming terminal success. An unproven provider + // transition is a foreign live-state change: dispatch owns the + // abandonment and must release agents with that reason, not issue-done. + if (postSpawnIssueObservation && !await postSpawnIssueObservation.settled) return if (!humanReview) await this.#markDependencyTerminalAndReconcile(issue) + } else { + settleIssueWritebackOnce() } if (!await this.#saveDispatchLifecycle(record, 'writeback-applied')) return @@ -14025,6 +14189,10 @@ export class FactoryLoop implements Factory { if (releaseReasonForRetry) this.#scheduleReleaseRetry(record, releaseReasonForRetry) else this.#scheduleDispatchLifecycleRetry(record) } finally { + // Errors before or during provider confirmation leave the marker unset; + // the waiting dispatch will re-read and preserve the foreign-change + // abort. Always settle so a failed write cannot strand that dispatch. + settleIssueWritebackOnce() this.#completionInFlight.delete(completionKey) const stateKey = issueStateKey(record.issue) this.#probePrGhBackoffUntilMs.delete(stateKey) @@ -14041,6 +14209,31 @@ export class FactoryLoop implements Factory { } } + #recordMissingGithubWritebackReceipt(method: 'setStatus' | 'closeIssue'): void { + const implementation = this.#githubWriteback.constructor.name || 'anonymous GithubWriteback' + this.#increment('githubWritebackReceiptMissing') + this.#logger.warn?.('[factory] GitHub writeback returned no ownership receipt', { + implementation, + method, + }) + } + + #settlePostSpawnDispatchWaits( + key: string, + accepted: boolean, + source?: 'abandonment' | 'stop', + ): boolean { + const observation = this.#postSpawnIssueObservations.get(key) + const claimFence = this.#postSpawnDispatchClaimFences.get(key) + if (!observation && !claimFence) return false + // Capture both before either settlement deletes its own map entry. + observation?.settle(accepted) + claimFence?.settle(accepted) + if (source === 'abandonment') this.#increment('postSpawnWaitsSettledByAbandonment') + if (source === 'stop') this.#increment('postSpawnWaitsSettledByStop') + return true + } + #emit(event: FactoryEvent, payload: FactoryEventPayload): void { for (const listener of this.#listeners.get(event) ?? []) { try { diff --git a/src/ports/index.ts b/src/ports/index.ts index d4836146..8bd0acc0 100644 --- a/src/ports/index.ts +++ b/src/ports/index.ts @@ -41,7 +41,9 @@ export type { } from './fleet' export type { GithubRead, + GithubIssueCloseWriteResult, GithubIssueStatus, + GithubStatusWriteResult, GithubWriteback, LinearWriteback, SlackWriteback, diff --git a/src/ports/writeback.ts b/src/ports/writeback.ts index d28a27a9..92842f16 100644 --- a/src/ports/writeback.ts +++ b/src/ports/writeback.ts @@ -19,6 +19,22 @@ export interface GithubRead { export type GithubIssueStatus = 'ready' | 'in-progress' | 'human-review' +/** + * Whether a lifecycle call provably changed the provider's effective status. + * `applied` requires a provider audit event attributed to the writer's actor. + * `acknowledged` means the provider accepted an idempotent operation but its + * API did not prove who created the visible transition. `undefined` preserves + * compatibility with caller-supplied writebacks that predate this receipt. + */ +export type GithubStatusWriteResult = 'applied' | 'already-matched' | 'acknowledged' + +/** + * Whether closing an issue provably created the provider's visible transition. + * The same conservative receipt semantics as status writes apply: only an + * `applied` result may establish Factory ownership of the closed state. + */ +export type GithubIssueCloseWriteResult = 'applied' | 'already-matched' | 'acknowledged' + export interface GithubWriteback { /** Optional local-user PR publisher, implemented by the default `gh` writeback. */ publishPullRequest?(input: GithubPublishPullRequestInput): Promise @@ -29,6 +45,6 @@ export interface GithubWriteback { postComment(issue: LinearIssue, body: string): Promise /** Provider-authoritative lookup used to reconcile ambiguous comment writes. */ hasCommentMarker?(issue: LinearIssue, marker: string): Promise - setStatus(issue: LinearIssue, status: GithubIssueStatus): Promise - closeIssue(issue: LinearIssue, body: string): Promise + setStatus(issue: LinearIssue, status: GithubIssueStatus): Promise + closeIssue(issue: LinearIssue, body: string): Promise } diff --git a/src/writeback/github.ts b/src/writeback/github.ts index b1bb91f4..68d1581e 100644 --- a/src/writeback/github.ts +++ b/src/writeback/github.ts @@ -3,7 +3,12 @@ import { promisify } from 'node:util' import type { GithubConnectionWrite, MountClient } from '../ports' import type { GithubPublishPullRequestInput, GithubPublishPullRequestResult } from '../ports/mount' -import type { GithubIssueStatus, GithubWriteback } from '../ports/writeback' +import type { + GithubIssueCloseWriteResult, + GithubIssueStatus, + GithubStatusWriteResult, + GithubWriteback, +} from '../ports/writeback' import { defaultGhRunner, type GhRunner } from '../github/merge-gate' import type { LinearIssue, PrSummary } from '../types' import { asRecord, wrappedPayload } from './shared' @@ -63,6 +68,30 @@ export interface GhCliGithubWritebackConfig { gitRunner?: GhRunner } +interface GithubLabelEvent { + id: string + event: 'labeled' | 'unlabeled' + label: string + actor: string +} + +interface GithubLabelReceiptBaseline { + actor: string + eventIds: Set + statusLabels: Set +} + +interface GithubIssueStateEvent { + id: string + event: 'closed' | 'reopened' + actor: string +} + +interface GithubIssueCloseReceiptBaseline { + actor: string + eventIds: Set +} + type AppIssueConnectionWrite = GithubConnectionWrite & Required { + async setStatus(issue: LinearIssue, status: GithubIssueStatus): Promise { const ref = githubIssueRef(issue) if (status === 'ready') { for (const label of Object.values(FACTORY_GITHUB_STATUS_LABELS)) { @@ -111,7 +140,7 @@ export class AppGithubWriteback implements GithubWriteback { author: 'app', }) } - return + return 'acknowledged' } const target = FACTORY_GITHUB_STATUS_LABELS[status] const previous = FACTORY_GITHUB_STATUS_LABELS[status === 'in-progress' ? 'human-review' : 'in-progress'] @@ -134,9 +163,14 @@ export class AppGithubWriteback implements GithubWriteback { label: previous.name, author: 'app', }) + // Relayfile confirms that the operation was acknowledged, but the App + // writer has no provider-authoritative read/audit receipt proving whether + // this idempotent add created the visible label transition. Do not let the + // orchestrator claim authorship from acknowledgement alone. + return 'acknowledged' } - async closeIssue(issue: LinearIssue, body: string): Promise { + async closeIssue(issue: LinearIssue, body: string): Promise { const ref = githubIssueRef(issue) await this.postComment(issue, body) await this.#write.updateIssue({ @@ -145,6 +179,9 @@ export class AppGithubWriteback implements GithubWriteback { state: 'closed', author: 'app', }) + // Relayfile confirms acknowledgement, but this writer has no provider + // audit receipt that distinguishes our close from a concurrent actor's. + return 'acknowledged' } } @@ -298,24 +335,32 @@ export class GhCliGithubWriteback implements GithubWriteback { return result.stdout.includes(marker) } - async setStatus(issue: LinearIssue, status: GithubIssueStatus): Promise { + async setStatus(issue: LinearIssue, status: GithubIssueStatus): Promise { const ref = githubIssueRef(issue) if (status === 'ready') { const labels = await this.#issueLabels(ref) + const statusBefore = githubStatusFromLabels(labels) const editArgs = ['issue', 'edit', String(ref.number), '--repo', ref.repo] for (const label of Object.values(FACTORY_GITHUB_STATUS_LABELS)) { if (labels.has(label.name.toLowerCase())) { editArgs.push('--remove-label', label.name) } } - if (editArgs.length > 5) { + const editRequired = editArgs.length > 5 + const receiptBaseline = editRequired + ? await this.#labelReceiptBaseline(ref, labels).catch(() => undefined) + : undefined + if (editRequired) { await this.#run(editArgs) } const confirmed = await this.#issueLabels(ref) if (Object.values(FACTORY_GITHUB_STATUS_LABELS).some((label) => confirmed.has(label.name.toLowerCase()))) { throw new Error(`GitHub writeback did not confirm removal of Factory status labels on ${ref.repo}#${ref.number}`) } - return + if (!editRequired) return 'already-matched' + return await this.#hasAuthoredStatusTransition(ref, receiptBaseline, statusBefore, status) + ? 'applied' + : 'acknowledged' } const target = FACTORY_GITHUB_STATUS_LABELS[status] const previous = FACTORY_GITHUB_STATUS_LABELS[status === 'in-progress' ? 'human-review' : 'in-progress'] @@ -332,6 +377,7 @@ export class GhCliGithubWriteback implements GithubWriteback { '--force', ]) const labels = await this.#issueLabels(ref) + const statusBefore = githubStatusFromLabels(labels) const editArgs = [ 'issue', 'edit', @@ -345,12 +391,22 @@ export class GhCliGithubWriteback implements GithubWriteback { if (labels.has(previous.name.toLowerCase())) { editArgs.push('--remove-label', previous.name) } - if (editArgs.length > 5) { + const editRequired = editArgs.length > 5 + const receiptBaseline = editRequired + ? await this.#labelReceiptBaseline(ref, labels).catch(() => undefined) + : undefined + if (editRequired) { await this.#run(editArgs) } const confirmed = await this.#issueLabels(ref) if (confirmed.has(target.name.toLowerCase()) && !confirmed.has(previous.name.toLowerCase())) { - return + // Removing an obsolete label is a provider mutation, but it does not + // establish ownership when the requested effective status already won + // before our first read (notably human-review over in-progress). + if (statusBefore === status) return 'already-matched' + return await this.#hasAuthoredStatusTransition(ref, receiptBaseline, statusBefore, status) + ? 'applied' + : 'acknowledged' } throw new Error(`GitHub writeback did not confirm ${target.name} on ${ref.repo}#${ref.number}`) } @@ -376,9 +432,76 @@ export class GhCliGithubWriteback implements GithubWriteback { ) } - async closeIssue(issue: LinearIssue, body: string): Promise { + async #labelReceiptBaseline( + ref: { repo: string; number: number }, + labels: ReadonlySet, + ): Promise { + const actor = (await this.#run(['api', 'user', '--jq', '.login'])).stdout.trim().toLowerCase() + if (!actor) throw new Error('GitHub lifecycle receipt could not resolve the authenticated actor') + const events = await this.#issueLabelEvents(ref) + const statusLabelNames = new Set(Object.values(FACTORY_GITHUB_STATUS_LABELS).map((label) => label.name)) + return { + actor, + eventIds: new Set(events.map((event) => event.id)), + statusLabels: new Set([...labels].filter((label) => statusLabelNames.has(label))), + } + } + + async #hasAuthoredStatusTransition( + ref: { repo: string; number: number }, + baseline: GithubLabelReceiptBaseline | undefined, + from: GithubIssueStatus, + to: GithubIssueStatus, + ): Promise { + if (!baseline) return false + const expected = githubStatusTransitionEvent(from, to) + if (!expected) return false + const events = await this.#issueLabelEvents(ref).catch(() => []) + const statusLabels = new Set(Object.values(FACTORY_GITHUB_STATUS_LABELS).map((label) => label.name)) + const newStatusEvents = events.filter((event) => + !baseline.eventIds.has(event.id) && statusLabels.has(event.label), + ) + const effectiveLabels = new Set(baseline.statusLabels) + let definingEvent: GithubLabelEvent | undefined + for (const event of newStatusEvents) { + const before = githubStatusFromLabels(effectiveLabels) + if (event.event === 'labeled') effectiveLabels.add(event.label) + else effectiveLabels.delete(event.label) + const after = githubStatusFromLabels(effectiveLabels) + if (before !== to && after === to) definingEvent = event + } + if (githubStatusFromLabels(effectiveLabels) !== to) return false + // Attribute the event that last made the confirmed effective status true, + // not later status-label cleanup that leaves the effective status intact. + return definingEvent?.actor === baseline.actor + && definingEvent.event === expected.event + && definingEvent.label === expected.label + } + + async #issueLabelEvents(ref: { repo: string; number: number }): Promise { + const result = await this.#run([ + 'api', + '--paginate', + `repos/${ref.repo}/issues/${ref.number}/events`, + '--jq', + '.[] | select(.event == "labeled" or .event == "unlabeled") | [.id, .event, .label.name, .actor.login] | @tsv', + ]) + return result.stdout + .split('\n') + .map((line) => line.trim()) + .filter(Boolean) + .flatMap((line): GithubLabelEvent[] => { + const [id, event, label, actor] = line.split('\t') + if (!id || (event !== 'labeled' && event !== 'unlabeled') || !label || !actor) return [] + return [{ id, event, label: label.toLowerCase(), actor: actor.toLowerCase() }] + }) + } + + async closeIssue(issue: LinearIssue, body: string): Promise { const ref = githubIssueRef(issue) await this.postComment(issue, body) + if (await this.#issueState(ref) === 'closed') return 'already-matched' + const receiptBaseline = await this.#issueCloseReceiptBaseline(ref).catch(() => undefined) await this.#run([ 'issue', 'close', @@ -388,6 +511,77 @@ export class GhCliGithubWriteback implements GithubWriteback { '--reason', 'completed', ]) + if (await this.#issueState(ref) !== 'closed') { + throw new Error(`GitHub writeback did not confirm closed state on ${ref.repo}#${ref.number}`) + } + return await this.#hasAuthoredIssueClose(ref, receiptBaseline) + ? 'applied' + : 'acknowledged' + } + + async #issueState(ref: { repo: string; number: number }): Promise<'open' | 'closed'> { + const result = await this.#run([ + 'issue', + 'view', + String(ref.number), + '--repo', + ref.repo, + '--json', + 'state', + ]) + const parsed = JSON.parse(result.stdout) as { state?: unknown } + const state = stringValue(parsed.state)?.toLowerCase() + if (state === 'open' || state === 'closed') return state + throw new Error(`GitHub lifecycle read returned an unknown issue state on ${ref.repo}#${ref.number}`) + } + + async #issueCloseReceiptBaseline( + ref: { repo: string; number: number }, + ): Promise { + const actor = (await this.#run(['api', 'user', '--jq', '.login'])).stdout.trim().toLowerCase() + if (!actor) throw new Error('GitHub close receipt could not resolve the authenticated actor') + const events = await this.#issueStateEvents(ref) + return { actor, eventIds: new Set(events.map((event) => event.id)) } + } + + async #hasAuthoredIssueClose( + ref: { repo: string; number: number }, + baseline: GithubIssueCloseReceiptBaseline | undefined, + ): Promise { + if (!baseline) return false + const events = await this.#issueStateEvents(ref).catch(() => []) + const newEvents = events.filter((event) => !baseline.eventIds.has(event.id)) + let state: 'open' | 'closed' = 'open' + let definingClose: GithubIssueStateEvent | undefined + for (const event of newEvents) { + if (event.event === 'closed') { + if (state === 'open') definingClose = event + state = 'closed' + } else { + state = 'open' + definingClose = undefined + } + } + return state === 'closed' && definingClose?.actor === baseline.actor + } + + async #issueStateEvents(ref: { repo: string; number: number }): Promise { + const result = await this.#run([ + 'api', + '--paginate', + `repos/${ref.repo}/issues/${ref.number}/events`, + '--jq', + '.[] | select(.event == "closed" or .event == "reopened") | [.id, .event, .actor.login] | @tsv', + ]) + return result.stdout + .split('\n') + .map((line) => line.trim()) + .filter(Boolean) + .flatMap((line): GithubIssueStateEvent[] => { + const [id, event, actor] = line.split('\t') + if (!id || (event !== 'closed' && event !== 'reopened') || !actor) return [] + return [{ id, event, actor: actor.toLowerCase() }] + }) } async #gitValue(args: string[], description: string): Promise { @@ -401,6 +595,30 @@ export class GhCliGithubWriteback implements GithubWriteback { } } +const githubStatusFromLabels = (labels: Set): GithubIssueStatus => { + if (labels.has(FACTORY_GITHUB_STATUS_LABELS['human-review'].name.toLowerCase())) return 'human-review' + if (labels.has(FACTORY_GITHUB_STATUS_LABELS['in-progress'].name.toLowerCase())) return 'in-progress' + return 'ready' +} + +const githubStatusTransitionEvent = ( + from: GithubIssueStatus, + to: GithubIssueStatus, +): Pick | undefined => { + if (from === to) return undefined + if (to === 'human-review') { + return { event: 'labeled', label: FACTORY_GITHUB_STATUS_LABELS['human-review'].name } + } + if (to === 'in-progress') { + return from === 'human-review' + ? { event: 'unlabeled', label: FACTORY_GITHUB_STATUS_LABELS['human-review'].name } + : { event: 'labeled', label: FACTORY_GITHUB_STATUS_LABELS['in-progress'].name } + } + return from === 'human-review' + ? { event: 'unlabeled', label: FACTORY_GITHUB_STATUS_LABELS['human-review'].name } + : { event: 'unlabeled', label: FACTORY_GITHUB_STATUS_LABELS['in-progress'].name } +} + const defaultGitRunner: GhRunner = async (args) => { const { stdout, stderr } = await execFileAsync('git', args, { maxBuffer: 1024 * 1024 }) return { stdout, stderr } diff --git a/src/writeback/writeback.test.ts b/src/writeback/writeback.test.ts index 1e1d472d..c552aeaf 100644 --- a/src/writeback/writeback.test.ts +++ b/src/writeback/writeback.test.ts @@ -856,9 +856,9 @@ describe('AppGithubWriteback', () => { body: 'Fixes #221', })).resolves.toMatchObject({ number: 322, author: 'app' }) await app.postComment(appIssue, 'Factory dispatch for 221') - await app.setStatus(appIssue, 'human-review') - await app.setStatus(appIssue, 'ready') - await app.closeIssue(appIssue, 'Factory observed the linked PR merge.') + await expect(app.setStatus(appIssue, 'human-review')).resolves.toBe('acknowledged') + await expect(app.setStatus(appIssue, 'ready')).resolves.toBe('acknowledged') + await expect(app.closeIssue(appIssue, 'Factory observed the linked PR merge.')).resolves.toBe('acknowledged') expect(postIssueComment).toHaveBeenNthCalledWith(1, { repo: 'AgentWorkforce/factory', @@ -929,6 +929,21 @@ describe('GhCliGithubWriteback', () => { }, }, } + const authenticatedActorCall = ['api', 'user', '--jq', '.login'] + const issueLabelEventsCall = [ + 'api', + '--paginate', + 'repos/AgentWorkforce/factory/issues/48/events', + '--jq', + '.[] | select(.event == "labeled" or .event == "unlabeled") | [.id, .event, .label.name, .actor.login] | @tsv', + ] + const issueStateEventsCall = [ + 'api', + '--paginate', + 'repos/AgentWorkforce/factory/issues/48/events', + '--jq', + '.[] | select(.event == "closed" or .event == "reopened") | [.id, .event, .actor.login] | @tsv', + ] it('pushes a local branch and returns the gh-authenticated PR author', async () => { const ghCalls: string[][] = [] @@ -1042,45 +1057,65 @@ describe('GhCliGithubWriteback', () => { it('sets the first lifecycle status without removing an absent label, then transitions statuses', async () => { const calls: string[][] = [] const labels = new Set() + const events: string[] = [] + let nextEventId = 1 const github = new GhCliGithubWriteback({ runner: async (args) => { calls.push(args) + if (args[0] === 'api' && args[1] === 'user') return { stdout: 'factory-bot\n' } + if (args[0] === 'api' && args[1] === '--paginate') return { stdout: events.join('\n') } if (args[0] === 'issue' && args[1] === 'view') { return { stdout: JSON.stringify({ labels: [...labels].map((name) => ({ name })) }) } } if (args[0] === 'issue' && args[1] === 'edit') { const added = args[args.indexOf('--add-label') + 1] const removed = args[args.indexOf('--remove-label') + 1] - if (args.includes('--add-label') && added) labels.add(added) - if (args.includes('--remove-label') && removed) labels.delete(removed) + if (args.includes('--add-label') && added && !labels.has(added)) { + labels.add(added) + events.push(`${nextEventId++}\tlabeled\t${added}\tfactory-bot`) + } + if (args.includes('--remove-label') && removed && labels.has(removed)) { + labels.delete(removed) + events.push(`${nextEventId++}\tunlabeled\t${removed}\tfactory-bot`) + } } return { stdout: '' } }, }) await github.postComment(githubIssue, 'Factory dispatch for 48') - await github.setStatus(githubIssue, 'in-progress') - await github.setStatus(githubIssue, 'human-review') + await expect(github.setStatus(githubIssue, 'in-progress')).resolves.toBe('applied') + await expect(github.setStatus(githubIssue, 'human-review')).resolves.toBe('applied') expect(calls).toEqual([ ['issue', 'comment', '48', '--repo', 'AgentWorkforce/factory', '--body', 'Factory dispatch for 48'], ['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'], + authenticatedActorCall, + issueLabelEventsCall, ['issue', 'edit', '48', '--repo', 'AgentWorkforce/factory', '--add-label', 'factory:in-progress'], ['issue', 'view', '48', '--repo', 'AgentWorkforce/factory', '--json', 'labels'], + issueLabelEventsCall, ['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'], + authenticatedActorCall, + issueLabelEventsCall, ['issue', 'edit', '48', '--repo', 'AgentWorkforce/factory', '--add-label', 'factory:human-review', '--remove-label', 'factory:in-progress'], ['issue', 'view', '48', '--repo', 'AgentWorkforce/factory', '--json', 'labels'], + issueLabelEventsCall, ]) }) 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 events: string[] = [] + let nextEventId = 1 const github = new GhCliGithubWriteback({ runner: async (args) => { calls.push(args) + if (args[0] === 'api' && args[1] === 'user') return { stdout: 'factory-bot\n' } + if (args[0] === 'api' && args[1] === '--paginate') return { stdout: events.join('\n') } if (args[0] === 'issue' && args[1] === 'view') { return { stdout: JSON.stringify({ labels: [...labels].map((name) => ({ name })) }), @@ -1088,17 +1123,22 @@ describe('GhCliGithubWriteback', () => { } if (args[0] === 'issue' && args[1] === 'edit' && args.includes('--remove-label')) { args.forEach((arg, index) => { - if (arg === '--remove-label') labels.delete(args[index + 1]!) + const removed = args[index + 1]! + if (arg === '--remove-label' && labels.delete(removed)) { + events.push(`${nextEventId++}\tunlabeled\t${removed}\tfactory-bot`) + } }) } return { stdout: '' } }, }) - await github.setStatus(githubIssue, 'ready') + await expect(github.setStatus(githubIssue, 'ready')).resolves.toBe('applied') expect(calls).toEqual([ ['issue', 'view', '48', '--repo', 'AgentWorkforce/factory', '--json', 'labels'], + authenticatedActorCall, + issueLabelEventsCall, [ 'issue', 'edit', @@ -1111,9 +1151,169 @@ describe('GhCliGithubWriteback', () => { 'factory:human-review', ], ['issue', 'view', '48', '--repo', 'AgentWorkforce/factory', '--json', 'labels'], + issueLabelEventsCall, ]) }) + it('does not attribute a label add won by another actor between read and edit', async () => { + const labels = new Set(['factory:in-progress']) + const events: string[] = [] + const github = new GhCliGithubWriteback({ + runner: async (args) => { + if (args[0] === 'api' && args[1] === 'user') return { stdout: 'factory-bot\n' } + if (args[0] === 'api' && args[1] === '--paginate') return { stdout: events.join('\n') } + if (args[0] === 'issue' && args[1] === 'view') { + return { stdout: JSON.stringify({ labels: [...labels].map((name) => ({ name })) }) } + } + if (args[0] === 'issue' && args[1] === 'edit') { + labels.delete('factory:in-progress') + labels.add('factory:human-review') + events.push('1\tunlabeled\tfactory:in-progress\tother-user') + events.push('2\tlabeled\tfactory:human-review\tother-user') + } + return { stdout: '' } + }, + }) + + await expect(github.setStatus(githubIssue, 'human-review')).resolves.toBe('acknowledged') + }) + + it('does not attribute a label removal won by another actor between read and edit', async () => { + const labels = new Set(['factory:human-review']) + const events: string[] = [] + const github = new GhCliGithubWriteback({ + runner: async (args) => { + if (args[0] === 'api' && args[1] === 'user') return { stdout: 'factory-bot\n' } + if (args[0] === 'api' && args[1] === '--paginate') return { stdout: events.join('\n') } + if (args[0] === 'issue' && args[1] === 'view') { + return { stdout: JSON.stringify({ labels: [...labels].map((name) => ({ name })) }) } + } + if (args[0] === 'issue' && args[1] === 'edit') { + labels.delete('factory:human-review') + events.push('1\tunlabeled\tfactory:human-review\tother-user') + } + return { stdout: '' } + }, + }) + + await expect(github.setStatus(githubIssue, 'ready')).resolves.toBe('acknowledged') + }) + + it('does not attribute a park that another actor removes and recreates after the edit', async () => { + const labels = new Set(['factory:in-progress']) + const events: string[] = [] + const github = new GhCliGithubWriteback({ + runner: async (args) => { + if (args[0] === 'api' && args[1] === 'user') return { stdout: 'factory-bot\n' } + if (args[0] === 'api' && args[1] === '--paginate') return { stdout: events.join('\n') } + if (args[0] === 'issue' && args[1] === 'view') { + return { stdout: JSON.stringify({ labels: [...labels].map((name) => ({ name })) }) } + } + if (args[0] === 'issue' && args[1] === 'edit') { + labels.delete('factory:in-progress') + labels.add('factory:human-review') + events.push('1\tlabeled\tfactory:human-review\tfactory-bot') + events.push('2\tunlabeled\tfactory:in-progress\tfactory-bot') + labels.delete('factory:human-review') + labels.add('factory:human-review') + events.push('3\tunlabeled\tfactory:human-review\tother-user') + events.push('4\tlabeled\tfactory:human-review\tother-user') + } + return { stdout: '' } + }, + }) + + await expect(github.setStatus(githubIssue, 'human-review')).resolves.toBe('acknowledged') + }) + + it('attributes a park when another actor only removes the lower-priority stale label', async () => { + const labels = new Set(['factory:in-progress']) + const events: string[] = [] + const github = new GhCliGithubWriteback({ + runner: async (args) => { + if (args[0] === 'api' && args[1] === 'user') return { stdout: 'factory-bot\n' } + if (args[0] === 'api' && args[1] === '--paginate') return { stdout: events.join('\n') } + if (args[0] === 'issue' && args[1] === 'view') { + return { stdout: JSON.stringify({ labels: [...labels].map((name) => ({ name })) }) } + } + if (args[0] === 'issue' && args[1] === 'edit') { + labels.add('factory:human-review') + events.push('1\tlabeled\tfactory:human-review\tfactory-bot') + labels.delete('factory:in-progress') + events.push('2\tunlabeled\tfactory:in-progress\tother-user') + } + return { stdout: '' } + }, + }) + + await expect(github.setStatus(githubIssue, 'human-review')).resolves.toBe('applied') + }) + + it('does not attribute ready when another actor recreates the final removal', async () => { + const labels = new Set(['factory:human-review']) + const events: string[] = [] + const github = new GhCliGithubWriteback({ + runner: async (args) => { + if (args[0] === 'api' && args[1] === 'user') return { stdout: 'factory-bot\n' } + if (args[0] === 'api' && args[1] === '--paginate') return { stdout: events.join('\n') } + if (args[0] === 'issue' && args[1] === 'view') { + return { stdout: JSON.stringify({ labels: [...labels].map((name) => ({ name })) }) } + } + if (args[0] === 'issue' && args[1] === 'edit') { + labels.delete('factory:human-review') + events.push('1\tunlabeled\tfactory:human-review\tfactory-bot') + labels.add('factory:human-review') + labels.delete('factory:human-review') + events.push('2\tlabeled\tfactory:human-review\tother-user') + events.push('3\tunlabeled\tfactory:human-review\tother-user') + } + return { stdout: '' } + }, + }) + + await expect(github.setStatus(githubIssue, 'ready')).resolves.toBe('acknowledged') + }) + + it.each([ + { status: 'ready' as const, labels: [] }, + { status: 'in-progress' as const, labels: ['factory:in-progress'] }, + ])('does not issue a $status lifecycle edit when the state already matches', async ({ status, labels: initialLabels }) => { + const calls: string[][] = [] + const labels = new Set(initialLabels) + const github = new GhCliGithubWriteback({ + runner: async (args) => { + calls.push(args) + if (args[0] === 'issue' && args[1] === 'view') { + return { stdout: JSON.stringify({ labels: [...labels].map((name) => ({ name })) }) } + } + return { stdout: '' } + }, + }) + await expect(github.setStatus(githubIssue, status)).resolves.toBe('already-matched') + + expect(calls.some((args) => args[0] === 'issue' && args[1] === 'edit')).toBe(false) + }) + + it('does not claim a status transition for cleanup after human-review already won', async () => { + const calls: string[][] = [] + const labels = new Set(['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: [...labels].map((name) => ({ name })) }) } + } + if (args[0] === 'issue' && args[1] === 'edit' && args.includes('--remove-label')) { + labels.delete('factory:in-progress') + } + return { stdout: '' } + }, + }) + + await expect(github.setStatus(githubIssue, 'human-review')).resolves.toBe('already-matched') + expect(calls.some((args) => args[0] === 'issue' && args[1] === 'edit')).toBe(true) + }) + it('rejects an acknowledged lifecycle edit when provider read-back never shows the label', async () => { let edits = 0 const github = new GhCliGithubWriteback({ @@ -1146,18 +1346,76 @@ describe('GhCliGithubWriteback', () => { it('comments and closes the GitHub issue after merge', async () => { const calls: string[][] = [] + let state = 'OPEN' + const events: string[] = [] const github = new GhCliGithubWriteback({ runner: async (args) => { calls.push(args) + if (args[0] === 'issue' && args[1] === 'view' && args.includes('state')) { + return { stdout: JSON.stringify({ state }) } + } + if (args[0] === 'api' && args[1] === 'user') return { stdout: 'factory-bot\n' } + if (args[0] === 'api' && args[1] === '--paginate') return { stdout: events.join('\n') } + if (args[0] === 'issue' && args[1] === 'close') { + state = 'CLOSED' + events.push('1\tclosed\tfactory-bot') + } return { stdout: '' } }, }) - await github.closeIssue(githubIssue, 'Factory observed the linked pull request merge.') + await expect( + github.closeIssue(githubIssue, 'Factory observed the linked pull request merge.'), + ).resolves.toBe('applied') expect(calls).toEqual([ ['issue', 'comment', '48', '--repo', 'AgentWorkforce/factory', '--body', 'Factory observed the linked pull request merge.'], + ['issue', 'view', '48', '--repo', 'AgentWorkforce/factory', '--json', 'state'], + authenticatedActorCall, + issueStateEventsCall, ['issue', 'close', '48', '--repo', 'AgentWorkforce/factory', '--reason', 'completed'], + ['issue', 'view', '48', '--repo', 'AgentWorkforce/factory', '--json', 'state'], + issueStateEventsCall, + ]) + }) + + it('does not attribute an issue close won by another actor between read and command', async () => { + let state = 'OPEN' + const events: string[] = [] + const github = new GhCliGithubWriteback({ + runner: async (args) => { + if (args[0] === 'issue' && args[1] === 'view' && args.includes('state')) { + return { stdout: JSON.stringify({ state }) } + } + if (args[0] === 'api' && args[1] === 'user') return { stdout: 'factory-bot\n' } + if (args[0] === 'api' && args[1] === '--paginate') return { stdout: events.join('\n') } + if (args[0] === 'issue' && args[1] === 'close') { + state = 'CLOSED' + events.push('1\tclosed\tother-user') + } + return { stdout: '' } + }, + }) + + await expect(github.closeIssue(githubIssue, 'Factory completion.')).resolves.toBe('acknowledged') + }) + + it('reports an already-closed issue without issuing a close command', async () => { + const calls: string[][] = [] + const github = new GhCliGithubWriteback({ + runner: async (args) => { + calls.push(args) + if (args[0] === 'issue' && args[1] === 'view') { + return { stdout: JSON.stringify({ state: 'CLOSED' }) } + } + return { stdout: '' } + }, + }) + + await expect(github.closeIssue(githubIssue, 'Factory completion.')).resolves.toBe('already-matched') + expect(calls).toEqual([ + ['issue', 'comment', '48', '--repo', 'AgentWorkforce/factory', '--body', 'Factory completion.'], + ['issue', 'view', '48', '--repo', 'AgentWorkforce/factory', '--json', 'state'], ]) })