From 543f910448eeb286dd5d41fac4aa08df96bdc058 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Sat, 22 Aug 2026 15:16:15 +0200 Subject: [PATCH 01/14] fix(factory): stamp the writeback marker at the write, not one await later (#319) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CodeRabbit on #321, against the phase-based version: `#handleAgentExit` can call `#completeIssue` while the record is still `dispatching`, so the terminal write becomes visible before `#saveDispatchLifecycle(record, 'writeback-applied')` updates the phase — and the post-spawn re-read would still tear down completed work. The marker introduced in b6760c9 already closes that window, since it is stamped long before the phase save. This closes the remainder: the marker now lands immediately after each write rather than after a further await, so no interleaving exists in which the writeback is observable but unattributed. `??=` keeps it first-write-wins. Not taken: CodeRabbit also suggested stamping self-owned lifecycle state *before* the terminal writeback. That is exactly what codex's P1 on this PR forbids — claiming authorship of a write not yet made would let a foreign park landing in that window be excused as ours. Confirm-after-write is the correct order, and the asymmetry decides it: tearing down our own finished work is bad but bounded, while ignoring somebody else's park is not. The interleaving test CodeRabbit asks for needs the deterministic harness tracked in #320. Co-Authored-By: Claude Opus 5 Session-Id: fb75fef3-5243-40ab-b0b7-9082ab820310 Session-Id: fb75fef3-5243-40ab-b0b7-9082ab820310 --- src/orchestrator/factory.ts | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 29087793..9673e7cd 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -4616,7 +4616,7 @@ export class FactoryLoop implements Factory { // issue our own lifecycle has already parked back to `implementing`. } else { 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 }) } } @@ -13943,6 +13943,10 @@ export class FactoryLoop implements Factory { if (githubIssue) { if (humanReview) { await this.#githubWriteback.setStatus(issue, 'human-review') + // Confirmed the instant the write lands. The post-spawn re-read + // can observe this state during any later await in this method, so + // the marker must not trail the write by one (CodeRabbit on #321). + record.issueWritebackConfirmedAtMs ??= this.#clock.now() 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}`, @@ -13952,15 +13956,17 @@ export class FactoryLoop implements Factory { issue, 'Factory observed the linked pull request merge and completed this issue.', ) + 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() + record.issueWritebackConfirmedAtMs ??= this.#clock.now() this.#emit('writeback-verified', { issue: record.issue, path: issue.path }) if (!humanReview) await this.#markDependencyTerminalAndReconcile(issue) } From 684fa133a4d0d36de5feb80f5c271ab48513fe4c Mon Sep 17 00:00:00 2001 From: Khaliq Date: Sat, 22 Aug 2026 16:01:41 +0200 Subject: [PATCH 02/14] fix(factory): attribute a park at the write, not when the writeback returns (#319) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit codex P1 on #322, and the scope is larger than the "8 lines in one file" this PR opened with. A writeback call does more than change state. `GithubWriteback.setStatus` adds the target label, then clears the previous one; `LinearWriteback .setState` writes the state, then awaits a readback confirmation. The issue becomes observably not-ready when the *first* of those lands, but `issueWritebackConfirmedAtMs` was stamped only once the call returned. In between, the post-spawn re-read sees a parked issue with no marker, calls this dispatch's own write foreign, and abandons — terminating agents that had already finished. codex found this on the GitHub path. The same shape is on the Linear path, which is the one the original flake actually exercises: `#isIssueReady` is `stateId === readyForAgent`, so the issue reads parked the moment `mount.writeFile` resolves, while `setState` keeps awaiting `confirmWriteback`. Add an optional `WritebackApplyHooks.onApplied`, invoked synchronously the instant the state-defining write resolves — after the target-label add and before the previous-label removal, after `writeFile` and before the readback — and stamp the marker from it. Optional, so other implementations stay compatible; the post-call stamp remains as a backstop. `onApplied` fires only after the write resolves, never before, so the marker still cannot be claimed for a write that has not landed. That was the point of the earlier codex P1 and it is preserved: this moves the stamp earlier in the writeback, not before the write. Tests. The trailing await is an injectable seam, so the must-fire that was impossible on #321 is now writable and deterministic: hold `setState`'s readback open and the dispatch's re-read is pinned inside the window. It fails 5/5 against main with `expected 3 to be +0` — the production signature — and passes here. Paired with two hook-contract tests: `onApplied` lands between the add and the removal, and never fires when the write rejects. 389 tests pass. Flake rate 12 runs, 12 pass / 0 fail. Co-Authored-By: Claude Opus 5 Session-Id: fb75fef3-5243-40ab-b0b7-9082ab820310 --- src/cli/fleet.test.ts | 105 ++++++++++++++++++++++++++++++++ src/orchestrator/factory.ts | 20 ++++-- src/ports/index.ts | 1 + src/ports/writeback.ts | 18 +++++- src/writeback/github.ts | 13 +++- src/writeback/linear.ts | 8 ++- src/writeback/writeback.test.ts | 61 +++++++++++++++++++ 7 files changed, 214 insertions(+), 12 deletions(-) diff --git a/src/cli/fleet.test.ts b/src/cli/fleet.test.ts index a3ed48e7..6a4de060 100644 --- a/src/cli/fleet.test.ts +++ b/src/cli/fleet.test.ts @@ -1984,6 +1984,111 @@ describe('fleet CLI runtime', () => { // "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 MUST-FIRE. Deterministic, and it only became writable once + // codex pointed out that the writeback keeps awaiting AFTER the state is + // visible: that trailing await is an injectable seam. `setState` writes the + // parked state, then awaits its readback confirmation before returning, so + // holding that readback open pins the dispatch's post-spawn re-read inside + // the window every single run. + // + // Against a build that stamps the marker only when `setState` RETURNS, the + // re-read sees a parked issue with no marker, calls its own write foreign, + // and abandons — tearing down agents that had already finished. Exit 3. + it('does not abandon when its own park is visible but the writeback has not returned', async () => { + 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, + } + // `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 && this.parkWritten) { + 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) + }) + } else if (!this.reReadSeen) { + // Second is the dispatch's post-spawn re-read: the one under + // test. Let it observe the parked issue, then unblock setState. + this.reReadSeen = true + queueMicrotask(() => this.releaseConfirm?.()) + } + } + return super.readFile(path) + } + } + + const mount = new ParkConfirmBlockingMount({ + [issuePath]: issueFile, + '/github/repos/AgentWorkforce/pear/meta.json': { default_branch: 'main' }, + }, githubWrite) + const fleet = new CompletingRemoteFleetClient() + 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) + // Against main this is 3 — FACTORY_EXIT.RETRYABLE — the same + // `expected 3 to be +0` signature as the original flake. + expect(code).toBe(0) + // Abandoning releases with this reason; completing must not. + expect(fleet.releases.map((release) => release.reason)).not.toContain('live dispatch state changed') + } finally { + 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/orchestrator/factory.ts b/src/orchestrator/factory.ts index 9673e7cd..83a8eea3 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -13942,10 +13942,15 @@ export class FactoryLoop implements Factory { if (issue) { if (githubIssue) { if (humanReview) { - await this.#githubWriteback.setStatus(issue, 'human-review') - // Confirmed the instant the write lands. The post-spawn re-read - // can observe this state during any later await in this method, so - // the marker must not trail the write by one (CodeRabbit on #321). + // Stamped from inside setStatus, the instant the human-review label + // lands: `#isIssueReady` returns false on that label alone, while + // setStatus still has its previous-label removal outstanding. A + // stamp after setStatus returns leaves the issue readable as + // not-ready with the marker unset — the self-abandonment window + // (codex review on #322). + await this.#githubWriteback.setStatus(issue, 'human-review', { + onApplied: () => { record.issueWritebackConfirmedAtMs ??= this.#clock.now() }, + }) record.issueWritebackConfirmedAtMs ??= this.#clock.now() await this.#githubWriteback.postComment( issue, @@ -13962,7 +13967,12 @@ export class FactoryLoop implements Factory { const targetState = humanReview ? this.#states.idFor(issueTeam, 'humanReview') : this.#states.idFor(issueTeam, 'done') - await this.#linear.setState(issue, targetState) + // Same window on the Linear path, and this is the one the original + // flake exercises: `setState` makes the state visible on writeFile, + // then keeps awaiting its readback confirmation before returning. + await this.#linear.setState(issue, targetState, { + onApplied: () => { record.issueWritebackConfirmedAtMs ??= this.#clock.now() }, + }) record.issueWritebackConfirmedAtMs ??= this.#clock.now() await this.#recordCanonicalIssueState({ ...record.issue, stateId: targetState }) } diff --git a/src/ports/index.ts b/src/ports/index.ts index d4836146..01adb15d 100644 --- a/src/ports/index.ts +++ b/src/ports/index.ts @@ -45,6 +45,7 @@ export type { GithubWriteback, LinearWriteback, SlackWriteback, + WritebackApplyHooks, } from './writeback' export type { Clock, diff --git a/src/ports/writeback.ts b/src/ports/writeback.ts index d28a27a9..528be61a 100644 --- a/src/ports/writeback.ts +++ b/src/ports/writeback.ts @@ -1,8 +1,22 @@ import type { GithubPublishPullRequestInput, GithubPublishPullRequestResult } from './mount' import type { LinearIssue, PrSummary } from '../types' +/** + * Notification that the *state-defining* write has landed. + * + * A writeback call does more than change state: it confirms readbacks, clears + * the previous label, posts comments. The issue becomes observably not-ready + * the moment the state write resolves, which is well before the call returns — + * so a caller that needs to know "is this state change mine?" cannot wait for + * the promise. `onApplied` fires synchronously at that instant, before any + * confirmation or cleanup await (factory#319). + */ +export interface WritebackApplyHooks { + onApplied?: () => void +} + export interface LinearWriteback { - setState(issue: LinearIssue, stateId: string): Promise + setState(issue: LinearIssue, stateId: string, hooks?: WritebackApplyHooks): Promise postComment(issue: LinearIssue, body: string): Promise createIssue(payload: Record): Promise<{ path: string }> verify(issue: LinearIssue, expect: { stateId?: string; commentName?: string }): Promise @@ -29,6 +43,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 + setStatus(issue: LinearIssue, status: GithubIssueStatus, hooks?: WritebackApplyHooks): Promise closeIssue(issue: LinearIssue, body: string): Promise } diff --git a/src/writeback/github.ts b/src/writeback/github.ts index b1bb91f4..7e594c31 100644 --- a/src/writeback/github.ts +++ b/src/writeback/github.ts @@ -1,7 +1,7 @@ import { execFile } from 'node:child_process' import { promisify } from 'node:util' -import type { GithubConnectionWrite, MountClient } from '../ports' +import type { GithubConnectionWrite, MountClient, WritebackApplyHooks } from '../ports' import type { GithubPublishPullRequestInput, GithubPublishPullRequestResult } from '../ports/mount' import type { GithubIssueStatus, GithubWriteback } from '../ports/writeback' import { defaultGhRunner, type GhRunner } from '../github/merge-gate' @@ -99,7 +99,7 @@ export class AppGithubWriteback implements GithubWriteback { }) } - async setStatus(issue: LinearIssue, status: GithubIssueStatus): Promise { + async setStatus(issue: LinearIssue, status: GithubIssueStatus, hooks?: WritebackApplyHooks): Promise { const ref = githubIssueRef(issue) if (status === 'ready') { for (const label of Object.values(FACTORY_GITHUB_STATUS_LABELS)) { @@ -111,6 +111,7 @@ export class AppGithubWriteback implements GithubWriteback { author: 'app', }) } + hooks?.onApplied?.() return } const target = FACTORY_GITHUB_STATUS_LABELS[status] @@ -127,6 +128,9 @@ export class AppGithubWriteback implements GithubWriteback { label: target.name, author: 'app', }) + // The target label alone decides readiness, so the status change is live + // here — the removal below is cleanup. Attribute it now (factory#319). + hooks?.onApplied?.() await this.#write.mutateIssueLabel({ repo: ref.repo, number: ref.number, @@ -298,7 +302,7 @@ export class GhCliGithubWriteback implements GithubWriteback { return result.stdout.includes(marker) } - async setStatus(issue: LinearIssue, status: GithubIssueStatus): Promise { + async setStatus(issue: LinearIssue, status: GithubIssueStatus, hooks?: WritebackApplyHooks): Promise { const ref = githubIssueRef(issue) if (status === 'ready') { const labels = await this.#issueLabels(ref) @@ -311,6 +315,7 @@ export class GhCliGithubWriteback implements GithubWriteback { if (editArgs.length > 5) { await this.#run(editArgs) } + hooks?.onApplied?.() 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}`) @@ -348,6 +353,8 @@ export class GhCliGithubWriteback implements GithubWriteback { if (editArgs.length > 5) { await this.#run(editArgs) } + // The label edit has landed; `#issueLabels` below only verifies it. + hooks?.onApplied?.() const confirmed = await this.#issueLabels(ref) if (confirmed.has(target.name.toLowerCase()) && !confirmed.has(previous.name.toLowerCase())) { return diff --git a/src/writeback/linear.ts b/src/writeback/linear.ts index 610c951f..74f25209 100644 --- a/src/writeback/linear.ts +++ b/src/writeback/linear.ts @@ -1,5 +1,5 @@ import { linearByIdPath, linearByUuidPath, linearCommentPath, linearIssuePath } from '../constants/linear' -import type { MountClient } from '../ports' +import type { MountClient, WritebackApplyHooks } from '../ports' import type { Logger } from '../ports/system' import { assertInFactoryScope, isInFactoryScope } from '../safety/factory-scope' import type { LinearIssue } from '../types' @@ -257,7 +257,7 @@ export const MountLinearWriteback = ( } const adapter = { - async setState(issue: LinearIssue, stateId: string): Promise { + async setState(issue: LinearIssue, stateId: string, hooks?: WritebackApplyHooks): Promise { const path = issuePath(issue) const canonical = await canonicalForIssue(issue) assertInFactoryScope(scopeIssueFromPayload(canonical.payload, issue.key), safety) @@ -265,6 +265,10 @@ export const MountLinearWriteback = ( ...canonical.writable, stateId, }, { guarded: true }) + // The issue reads as this state from here on; the readback below only + // confirms it. Anything that needs to attribute the change must be told + // now, not when this call returns (factory#319). + hooks?.onApplied?.() updateCanonicalState(path, issue, canonical, stateId) await confirmWriteback(mount, path, () => verifyStateReadback(mount, issue, stateId), logger, readbackConfirm) }, diff --git a/src/writeback/writeback.test.ts b/src/writeback/writeback.test.ts index 1e1d472d..6380a232 100644 --- a/src/writeback/writeback.test.ts +++ b/src/writeback/writeback.test.ts @@ -826,6 +826,67 @@ describe('AppGithubWriteback', () => { }, } + // factory#319: `onApplied` exists so a caller can attribute a state change + // the instant it lands. Two properties matter, and they pull opposite ways — + // fire early enough to close the self-abandonment window, but never for a + // write that did not happen. + it('fires onApplied as soon as the status label lands, before the previous label is cleared', async () => { + const order: string[] = [] + const connection: GithubConnectionWrite = { + publishPullRequest: async (input) => ({ + repo: input.repo, + number: 1, + url: 'u', + headRef: input.headRef ?? 'h', + }), + closePullRequest: async () => undefined, + postIssueComment: async () => undefined, + ensureRepositoryLabel: async () => undefined, + mutateIssueLabel: vi.fn(async (input: { operation: string; label: string }) => { + order.push(`${input.operation}:${input.label}`) + }), + updateIssue: async () => undefined, + } + const app = new AppGithubWriteback(connection) + + await app.setStatus(appIssue, 'human-review', { onApplied: () => order.push('applied') }) + + // The target label alone decides readiness, so `applied` must land between + // the add and the cleanup removal — not after both. + expect(order).toEqual([ + 'add:factory:human-review', + 'applied', + 'remove:factory:in-progress', + ]) + }) + + it('never fires onApplied when the status write itself rejects', async () => { + const connection: GithubConnectionWrite = { + publishPullRequest: async (input) => ({ + repo: input.repo, + number: 1, + url: 'u', + headRef: input.headRef ?? 'h', + }), + closePullRequest: async () => undefined, + postIssueComment: async () => undefined, + ensureRepositoryLabel: async () => undefined, + mutateIssueLabel: async (input: { operation: string }) => { + if (input.operation === 'add') throw new Error('label add rejected') + }, + updateIssue: async () => undefined, + } + const app = new AppGithubWriteback(connection) + const onApplied = vi.fn() + + await expect(app.setStatus(appIssue, 'human-review', { onApplied })) + .rejects.toThrow(/label add rejected/) + + // Claiming authorship of a write that never landed is the failure mode on + // the other side: it would let a foreign change be excused as ours. + expect(onApplied).not.toHaveBeenCalled() + }) + it('delegates PRs and lifecycle writes to the app connection without exposing read methods', async () => { const publishPullRequest: GithubConnectionWrite['publishPullRequest'] = vi.fn(async (input) => ({ repo: input.repo, From f13a1194b6a54592ee7d18913b56d9c5a3f862f7 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Sun, 23 Aug 2026 17:01:36 +0200 Subject: [PATCH 03/14] fix(factory): do not attribute skipped GitHub status edits (#319) --- src/writeback/github.ts | 6 +++--- src/writeback/writeback.test.ts | 23 +++++++++++++++++++++++ 2 files changed, 26 insertions(+), 3 deletions(-) diff --git a/src/writeback/github.ts b/src/writeback/github.ts index 7e594c31..10525a54 100644 --- a/src/writeback/github.ts +++ b/src/writeback/github.ts @@ -314,8 +314,8 @@ export class GhCliGithubWriteback implements GithubWriteback { } if (editArgs.length > 5) { await this.#run(editArgs) + hooks?.onApplied?.() } - hooks?.onApplied?.() 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}`) @@ -352,9 +352,9 @@ export class GhCliGithubWriteback implements GithubWriteback { } if (editArgs.length > 5) { await this.#run(editArgs) + // The label edit has landed; `#issueLabels` below only verifies it. + hooks?.onApplied?.() } - // The label edit has landed; `#issueLabels` below only verifies it. - hooks?.onApplied?.() const confirmed = await this.#issueLabels(ref) if (confirmed.has(target.name.toLowerCase()) && !confirmed.has(previous.name.toLowerCase())) { return diff --git a/src/writeback/writeback.test.ts b/src/writeback/writeback.test.ts index 6380a232..0ce785ae 100644 --- a/src/writeback/writeback.test.ts +++ b/src/writeback/writeback.test.ts @@ -1175,6 +1175,29 @@ describe('GhCliGithubWriteback', () => { ]) }) + it.each([ + { status: 'ready' as const, labels: [] }, + { status: 'in-progress' as const, labels: ['factory:in-progress'] }, + ])('never fires onApplied when the $status lifecycle edit is skipped', 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: '' } + }, + }) + const onApplied = vi.fn() + + await github.setStatus(githubIssue, status, { onApplied }) + + expect(calls.some((args) => args[0] === 'issue' && args[1] === 'edit')).toBe(false) + expect(onApplied).not.toHaveBeenCalled() + }) + it('rejects an acknowledged lifecycle edit when provider read-back never shows the label', async () => { let edits = 0 const github = new GhCliGithubWriteback({ From 047aa190c489dff82e6d750e65729e4456286316 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Sun, 23 Aug 2026 17:11:31 +0200 Subject: [PATCH 04/14] test(factory): pin the post-spawn park window (#319) --- src/cli/fleet.test.ts | 29 +++++++++++++++++++++++------ 1 file changed, 23 insertions(+), 6 deletions(-) diff --git a/src/cli/fleet.test.ts b/src/cli/fleet.test.ts index 6a4de060..840dbfc9 100644 --- a/src/cli/fleet.test.ts +++ b/src/cli/fleet.test.ts @@ -2013,6 +2013,18 @@ describe('fleet CLI runtime', () => { }), 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 => { @@ -2033,6 +2045,17 @@ describe('fleet CLI runtime', () => { } 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) { if (!this.releaseConfirm) { // First issue read after the park is `setState`'s own readback @@ -2043,11 +2066,6 @@ describe('fleet CLI runtime', () => { this.releaseConfirm = resolve setTimeout(resolve, 400) }) - } else if (!this.reReadSeen) { - // Second is the dispatch's post-spawn re-read: the one under - // test. Let it observe the parked issue, then unblock setState. - this.reReadSeen = true - queueMicrotask(() => this.releaseConfirm?.()) } } return super.readFile(path) @@ -2058,7 +2076,6 @@ describe('fleet CLI runtime', () => { [issuePath]: issueFile, '/github/repos/AgentWorkforce/pear/meta.json': { default_branch: 'main' }, }, githubWrite) - const fleet = new CompletingRemoteFleetClient() const output = buffer() const code = await runFleetCli([ From e80cc6c9dc6a31095ab9622bf799f70ff1bf91c2 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Sun, 23 Aug 2026 17:24:57 +0200 Subject: [PATCH 05/14] fix(factory): await terminal writeback confirmation (#319) --- src/cli/fleet.test.ts | 42 +++++++-------- src/orchestrator/factory.ts | 93 ++++++++++++++++++++------------- src/ports/index.ts | 1 - src/ports/writeback.ts | 18 +------ src/writeback/github.ts | 13 ++--- src/writeback/linear.ts | 8 +-- src/writeback/writeback.test.ts | 68 +----------------------- 7 files changed, 87 insertions(+), 156 deletions(-) diff --git a/src/cli/fleet.test.ts b/src/cli/fleet.test.ts index 840dbfc9..fbd59036 100644 --- a/src/cli/fleet.test.ts +++ b/src/cli/fleet.test.ts @@ -1980,21 +1980,14 @@ 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 MUST-FIRE. Deterministic, and it only became writable once - // codex pointed out that the writeback keeps awaiting AFTER the state is - // visible: that trailing await is an injectable seam. `setState` writes the - // parked state, then awaits its readback confirmation before returning, so - // holding that readback open pins the dispatch's post-spawn re-read inside - // the window every single run. - // - // Against a build that stamps the marker only when `setState` RETURNS, the - // re-read sees a parked issue with no marker, calls its own write foreign, - // and abandons — tearing down agents that had already finished. Exit 3. - it('does not abandon when its own park is visible but the writeback has not returned', async () => { + // 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, { @@ -2056,7 +2049,7 @@ describe('fleet CLI runtime', () => { this.reReadSeen = true queueMicrotask(() => this.releaseConfirm?.()) } - if (path === issuePath && this.parkWritten) { + 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 @@ -2070,6 +2063,11 @@ describe('fleet CLI runtime', () => { } 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({ @@ -2096,11 +2094,13 @@ describe('fleet CLI runtime', () => { // Both must hold or the test is not exercising the window at all. expect(mount.parkWritten).toBe(true) expect(mount.reReadSeen).toBe(true) - // Against main this is 3 — FACTORY_EXIT.RETRYABLE — the same - // `expected 3 to be +0` signature as the original flake. - expect(code).toBe(0) - // Abandoning releases with this reason; completing must not. - expect(fleet.releases.map((release) => release.reason)).not.toContain('live dispatch state changed') + 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 }) } diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 83a8eea3..47ab622a 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -816,6 +816,7 @@ export class FactoryLoop implements Factory { #previewSweepTimer?: ReturnType #previewSweepInFlight?: Promise readonly #completionInFlight = new Set() + readonly #issueWritebackInFlight = new Map>() readonly #agentExitsInFlight = new Map>() #reconciledAgentExitsActive = 0 readonly #reconciledAgentExitWaiters: Array<() => void> = [] @@ -4587,28 +4588,33 @@ 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 ((!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 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 skipped/idempotent edit + // is safe here because the completion deliberately converged on and + // confirmed the desired terminal state; it does not claim which + // actor first created that state. if (record.issueWritebackConfirmedAtMs === undefined) { throw new LiveDispatchStateChangedError(dispatchDecision.issue.key) } @@ -13891,6 +13897,19 @@ export class FactoryLoop implements Factory { return } this.#completionInFlight.add(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,17 +13959,14 @@ export class FactoryLoop implements Factory { const humanReview = configuredHumanReview || (githubIssue && !githubMerged) const statusLabel = humanReview ? 'In Human Review' : 'Done' if (issue) { + // 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) { - // Stamped from inside setStatus, the instant the human-review label - // lands: `#isIssueReady` returns false on that label alone, while - // setStatus still has its previous-label removal outstanding. A - // stamp after setStatus returns leaves the issue readable as - // not-ready with the marker unset — the self-abandonment window - // (codex review on #322). - await this.#githubWriteback.setStatus(issue, 'human-review', { - onApplied: () => { record.issueWritebackConfirmedAtMs ??= this.#clock.now() }, - }) + await this.#githubWriteback.setStatus(issue, 'human-review') record.issueWritebackConfirmedAtMs ??= this.#clock.now() await this.#githubWriteback.postComment( issue, @@ -13967,18 +13983,19 @@ export class FactoryLoop implements Factory { const targetState = humanReview ? this.#states.idFor(issueTeam, 'humanReview') : this.#states.idFor(issueTeam, 'done') - // Same window on the Linear path, and this is the one the original - // flake exercises: `setState` makes the state visible on writeFile, - // then keeps awaiting its readback confirmation before returning. - await this.#linear.setState(issue, targetState, { - onApplied: () => { record.issueWritebackConfirmedAtMs ??= this.#clock.now() }, - }) + 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 }) + // 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 (!humanReview) await this.#markDependencyTerminalAndReconcile(issue) + } else { + settleIssueWritebackOnce() } if (!await this.#saveDispatchLifecycle(record, 'writeback-applied')) return @@ -14041,6 +14058,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) diff --git a/src/ports/index.ts b/src/ports/index.ts index 01adb15d..d4836146 100644 --- a/src/ports/index.ts +++ b/src/ports/index.ts @@ -45,7 +45,6 @@ export type { GithubWriteback, LinearWriteback, SlackWriteback, - WritebackApplyHooks, } from './writeback' export type { Clock, diff --git a/src/ports/writeback.ts b/src/ports/writeback.ts index 528be61a..d28a27a9 100644 --- a/src/ports/writeback.ts +++ b/src/ports/writeback.ts @@ -1,22 +1,8 @@ import type { GithubPublishPullRequestInput, GithubPublishPullRequestResult } from './mount' import type { LinearIssue, PrSummary } from '../types' -/** - * Notification that the *state-defining* write has landed. - * - * A writeback call does more than change state: it confirms readbacks, clears - * the previous label, posts comments. The issue becomes observably not-ready - * the moment the state write resolves, which is well before the call returns — - * so a caller that needs to know "is this state change mine?" cannot wait for - * the promise. `onApplied` fires synchronously at that instant, before any - * confirmation or cleanup await (factory#319). - */ -export interface WritebackApplyHooks { - onApplied?: () => void -} - export interface LinearWriteback { - setState(issue: LinearIssue, stateId: string, hooks?: WritebackApplyHooks): Promise + setState(issue: LinearIssue, stateId: string): Promise postComment(issue: LinearIssue, body: string): Promise createIssue(payload: Record): Promise<{ path: string }> verify(issue: LinearIssue, expect: { stateId?: string; commentName?: string }): Promise @@ -43,6 +29,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, hooks?: WritebackApplyHooks): 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 10525a54..b1bb91f4 100644 --- a/src/writeback/github.ts +++ b/src/writeback/github.ts @@ -1,7 +1,7 @@ import { execFile } from 'node:child_process' import { promisify } from 'node:util' -import type { GithubConnectionWrite, MountClient, WritebackApplyHooks } from '../ports' +import type { GithubConnectionWrite, MountClient } from '../ports' import type { GithubPublishPullRequestInput, GithubPublishPullRequestResult } from '../ports/mount' import type { GithubIssueStatus, GithubWriteback } from '../ports/writeback' import { defaultGhRunner, type GhRunner } from '../github/merge-gate' @@ -99,7 +99,7 @@ export class AppGithubWriteback implements GithubWriteback { }) } - async setStatus(issue: LinearIssue, status: GithubIssueStatus, hooks?: WritebackApplyHooks): Promise { + 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 +111,6 @@ export class AppGithubWriteback implements GithubWriteback { author: 'app', }) } - hooks?.onApplied?.() return } const target = FACTORY_GITHUB_STATUS_LABELS[status] @@ -128,9 +127,6 @@ export class AppGithubWriteback implements GithubWriteback { label: target.name, author: 'app', }) - // The target label alone decides readiness, so the status change is live - // here — the removal below is cleanup. Attribute it now (factory#319). - hooks?.onApplied?.() await this.#write.mutateIssueLabel({ repo: ref.repo, number: ref.number, @@ -302,7 +298,7 @@ export class GhCliGithubWriteback implements GithubWriteback { return result.stdout.includes(marker) } - async setStatus(issue: LinearIssue, status: GithubIssueStatus, hooks?: WritebackApplyHooks): Promise { + async setStatus(issue: LinearIssue, status: GithubIssueStatus): Promise { const ref = githubIssueRef(issue) if (status === 'ready') { const labels = await this.#issueLabels(ref) @@ -314,7 +310,6 @@ export class GhCliGithubWriteback implements GithubWriteback { } if (editArgs.length > 5) { await this.#run(editArgs) - hooks?.onApplied?.() } const confirmed = await this.#issueLabels(ref) if (Object.values(FACTORY_GITHUB_STATUS_LABELS).some((label) => confirmed.has(label.name.toLowerCase()))) { @@ -352,8 +347,6 @@ export class GhCliGithubWriteback implements GithubWriteback { } if (editArgs.length > 5) { await this.#run(editArgs) - // The label edit has landed; `#issueLabels` below only verifies it. - hooks?.onApplied?.() } const confirmed = await this.#issueLabels(ref) if (confirmed.has(target.name.toLowerCase()) && !confirmed.has(previous.name.toLowerCase())) { diff --git a/src/writeback/linear.ts b/src/writeback/linear.ts index 74f25209..610c951f 100644 --- a/src/writeback/linear.ts +++ b/src/writeback/linear.ts @@ -1,5 +1,5 @@ import { linearByIdPath, linearByUuidPath, linearCommentPath, linearIssuePath } from '../constants/linear' -import type { MountClient, WritebackApplyHooks } from '../ports' +import type { MountClient } from '../ports' import type { Logger } from '../ports/system' import { assertInFactoryScope, isInFactoryScope } from '../safety/factory-scope' import type { LinearIssue } from '../types' @@ -257,7 +257,7 @@ export const MountLinearWriteback = ( } const adapter = { - async setState(issue: LinearIssue, stateId: string, hooks?: WritebackApplyHooks): Promise { + async setState(issue: LinearIssue, stateId: string): Promise { const path = issuePath(issue) const canonical = await canonicalForIssue(issue) assertInFactoryScope(scopeIssueFromPayload(canonical.payload, issue.key), safety) @@ -265,10 +265,6 @@ export const MountLinearWriteback = ( ...canonical.writable, stateId, }, { guarded: true }) - // The issue reads as this state from here on; the readback below only - // confirms it. Anything that needs to attribute the change must be told - // now, not when this call returns (factory#319). - hooks?.onApplied?.() updateCanonicalState(path, issue, canonical, stateId) await confirmWriteback(mount, path, () => verifyStateReadback(mount, issue, stateId), logger, readbackConfirm) }, diff --git a/src/writeback/writeback.test.ts b/src/writeback/writeback.test.ts index 0ce785ae..bac5d249 100644 --- a/src/writeback/writeback.test.ts +++ b/src/writeback/writeback.test.ts @@ -826,67 +826,6 @@ describe('AppGithubWriteback', () => { }, } - // factory#319: `onApplied` exists so a caller can attribute a state change - // the instant it lands. Two properties matter, and they pull opposite ways — - // fire early enough to close the self-abandonment window, but never for a - // write that did not happen. - it('fires onApplied as soon as the status label lands, before the previous label is cleared', async () => { - const order: string[] = [] - const connection: GithubConnectionWrite = { - publishPullRequest: async (input) => ({ - repo: input.repo, - number: 1, - url: 'u', - headRef: input.headRef ?? 'h', - }), - closePullRequest: async () => undefined, - postIssueComment: async () => undefined, - ensureRepositoryLabel: async () => undefined, - mutateIssueLabel: vi.fn(async (input: { operation: string; label: string }) => { - order.push(`${input.operation}:${input.label}`) - }), - updateIssue: async () => undefined, - } - const app = new AppGithubWriteback(connection) - - await app.setStatus(appIssue, 'human-review', { onApplied: () => order.push('applied') }) - - // The target label alone decides readiness, so `applied` must land between - // the add and the cleanup removal — not after both. - expect(order).toEqual([ - 'add:factory:human-review', - 'applied', - 'remove:factory:in-progress', - ]) - }) - - it('never fires onApplied when the status write itself rejects', async () => { - const connection: GithubConnectionWrite = { - publishPullRequest: async (input) => ({ - repo: input.repo, - number: 1, - url: 'u', - headRef: input.headRef ?? 'h', - }), - closePullRequest: async () => undefined, - postIssueComment: async () => undefined, - ensureRepositoryLabel: async () => undefined, - mutateIssueLabel: async (input: { operation: string }) => { - if (input.operation === 'add') throw new Error('label add rejected') - }, - updateIssue: async () => undefined, - } - const app = new AppGithubWriteback(connection) - const onApplied = vi.fn() - - await expect(app.setStatus(appIssue, 'human-review', { onApplied })) - .rejects.toThrow(/label add rejected/) - - // Claiming authorship of a write that never landed is the failure mode on - // the other side: it would let a foreign change be excused as ours. - expect(onApplied).not.toHaveBeenCalled() - }) - it('delegates PRs and lifecycle writes to the app connection without exposing read methods', async () => { const publishPullRequest: GithubConnectionWrite['publishPullRequest'] = vi.fn(async (input) => ({ repo: input.repo, @@ -1178,7 +1117,7 @@ describe('GhCliGithubWriteback', () => { it.each([ { status: 'ready' as const, labels: [] }, { status: 'in-progress' as const, labels: ['factory:in-progress'] }, - ])('never fires onApplied when the $status lifecycle edit is skipped', async ({ status, labels: initialLabels }) => { + ])('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({ @@ -1190,12 +1129,9 @@ describe('GhCliGithubWriteback', () => { return { stdout: '' } }, }) - const onApplied = vi.fn() - - await github.setStatus(githubIssue, status, { onApplied }) + await github.setStatus(githubIssue, status) expect(calls.some((args) => args[0] === 'issue' && args[1] === 'edit')).toBe(false) - expect(onApplied).not.toHaveBeenCalled() }) it('rejects an acknowledged lifecycle edit when provider read-back never shows the label', async () => { From 2dc54efe822a1d6bccc3693064bf435ff1b1e524 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Sun, 23 Aug 2026 17:51:38 +0200 Subject: [PATCH 06/14] fix(factory): preserve authorship across no-op writeback (#319) --- src/cli/fleet.test.ts | 163 +++++++++++++++++++++++++++++++- src/index.ts | 1 + src/orchestrator/factory.ts | 27 +++--- src/ports/index.ts | 1 + src/ports/writeback.ts | 9 +- src/writeback/github.ts | 19 ++-- src/writeback/writeback.test.ts | 8 +- 7 files changed, 203 insertions(+), 25 deletions(-) diff --git a/src/cli/fleet.test.ts b/src/cli/fleet.test.ts index fbd59036..550990a3 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, @@ -28,6 +28,7 @@ import { FakeFleetClient, FakeMountClient, withDeadline } from '../testing' import type { GithubConnectionRead, GithubConnectionWrite, GithubIssueLookup, 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' @@ -2106,6 +2107,166 @@ describe('fleet CLI runtime', () => { } }) + it('does not attribute a third-party GitHub park when the in-flight CLI write is a confirmed no-op', async () => { + 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 output = buffer() + const errors = buffer() + + const code = await runFleetCli([ + 'dispatch', + '48', + '--backend', + 'relay', + '--config', + configPath, + ], { + fleet, + mount, + createFactory: (factoryConfig, ports) => createFactory(factoryConfig, { + ...ports, + githubWriteback, + }), + 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') + } finally { + 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..1e8a1d61 100644 --- a/src/index.ts +++ b/src/index.ts @@ -353,6 +353,7 @@ export type { AgentUsage, GithubRead, GithubIssueStatus, + GithubStatusWriteResult, GithubWriteback, LinearWriteback, Logger, diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 47ab622a..1fa96f73 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -4608,13 +4608,12 @@ export class FactoryLoop implements Factory { } } if (!issue || !this.#isIssueReady(issue)) { - // Only a terminal writeback 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 skipped/idempotent edit - // is safe here because the completion deliberately converged on and - // confirmed the desired terminal state; it does not claim which - // actor first created that state. + // 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) } @@ -13966,8 +13965,13 @@ export class FactoryLoop implements Factory { this.#issueWritebackInFlight.set(completionKey, issueWritebackSettled) if (githubIssue) { if (humanReview) { - await this.#githubWriteback.setStatus(issue, 'human-review') - record.issueWritebackConfirmedAtMs ??= this.#clock.now() + const statusWrite = await this.#githubWriteback.setStatus(issue, 'human-review') + if (statusWrite !== 'already-matched') { + 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}`, @@ -13987,8 +13991,9 @@ export class FactoryLoop implements Factory { 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. diff --git a/src/ports/index.ts b/src/ports/index.ts index d4836146..4a5671e0 100644 --- a/src/ports/index.ts +++ b/src/ports/index.ts @@ -42,6 +42,7 @@ export type { export type { GithubRead, GithubIssueStatus, + GithubStatusWriteResult, GithubWriteback, LinearWriteback, SlackWriteback, diff --git a/src/ports/writeback.ts b/src/ports/writeback.ts index d28a27a9..839ded95 100644 --- a/src/ports/writeback.ts +++ b/src/ports/writeback.ts @@ -19,6 +19,13 @@ export interface GithubRead { export type GithubIssueStatus = 'ready' | 'in-progress' | 'human-review' +/** + * Whether a lifecycle call issued a provider mutation. `undefined` preserves + * compatibility with caller-supplied writebacks that predate this receipt; + * built-in adapters must return an explicit result. + */ +export type GithubStatusWriteResult = 'applied' | 'already-matched' + export interface GithubWriteback { /** Optional local-user PR publisher, implemented by the default `gh` writeback. */ publishPullRequest?(input: GithubPublishPullRequestInput): Promise @@ -29,6 +36,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 + 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..5fd7c98a 100644 --- a/src/writeback/github.ts +++ b/src/writeback/github.ts @@ -3,7 +3,7 @@ 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 { 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' @@ -99,7 +99,7 @@ export class AppGithubWriteback implements GithubWriteback { }) } - async setStatus(issue: LinearIssue, status: GithubIssueStatus): Promise { + 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 +111,7 @@ export class AppGithubWriteback implements GithubWriteback { author: 'app', }) } - return + return 'applied' } const target = FACTORY_GITHUB_STATUS_LABELS[status] const previous = FACTORY_GITHUB_STATUS_LABELS[status === 'in-progress' ? 'human-review' : 'in-progress'] @@ -134,6 +134,7 @@ export class AppGithubWriteback implements GithubWriteback { label: previous.name, author: 'app', }) + return 'applied' } async closeIssue(issue: LinearIssue, body: string): Promise { @@ -298,7 +299,7 @@ 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) @@ -308,14 +309,15 @@ export class GhCliGithubWriteback implements GithubWriteback { editArgs.push('--remove-label', label.name) } } - if (editArgs.length > 5) { + const applied = editArgs.length > 5 + if (applied) { 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 + return applied ? 'applied' : 'already-matched' } const target = FACTORY_GITHUB_STATUS_LABELS[status] const previous = FACTORY_GITHUB_STATUS_LABELS[status === 'in-progress' ? 'human-review' : 'in-progress'] @@ -345,12 +347,13 @@ export class GhCliGithubWriteback implements GithubWriteback { if (labels.has(previous.name.toLowerCase())) { editArgs.push('--remove-label', previous.name) } - if (editArgs.length > 5) { + const applied = editArgs.length > 5 + if (applied) { await this.#run(editArgs) } const confirmed = await this.#issueLabels(ref) if (confirmed.has(target.name.toLowerCase()) && !confirmed.has(previous.name.toLowerCase())) { - return + return applied ? 'applied' : 'already-matched' } throw new Error(`GitHub writeback did not confirm ${target.name} on ${ref.repo}#${ref.number}`) } diff --git a/src/writeback/writeback.test.ts b/src/writeback/writeback.test.ts index bac5d249..2f07d379 100644 --- a/src/writeback/writeback.test.ts +++ b/src/writeback/writeback.test.ts @@ -1059,8 +1059,8 @@ describe('GhCliGithubWriteback', () => { }) 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'], @@ -1095,7 +1095,7 @@ describe('GhCliGithubWriteback', () => { }, }) - 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'], @@ -1129,7 +1129,7 @@ describe('GhCliGithubWriteback', () => { return { stdout: '' } }, }) - await github.setStatus(githubIssue, status) + await expect(github.setStatus(githubIssue, status)).resolves.toBe('already-matched') expect(calls.some((args) => args[0] === 'issue' && args[1] === 'edit')).toBe(false) }) From 9b09add813d839569b610f47526dd8b70eb819da Mon Sep 17 00:00:00 2001 From: Khaliq Date: Sun, 23 Aug 2026 18:07:18 +0200 Subject: [PATCH 07/14] fix(factory): require proven GitHub status transition (#319) --- src/cli/fleet.test.ts | 17 +++++++++++++++-- src/orchestrator/factory.ts | 5 ++++- src/ports/writeback.ts | 9 +++++---- src/writeback/github.ts | 20 +++++++++++++++++--- src/writeback/writeback.test.ts | 24 ++++++++++++++++++++++-- 5 files changed, 63 insertions(+), 12 deletions(-) diff --git a/src/cli/fleet.test.ts b/src/cli/fleet.test.ts index 550990a3..377058c0 100644 --- a/src/cli/fleet.test.ts +++ b/src/cli/fleet.test.ts @@ -2107,7 +2107,10 @@ describe('fleet CLI runtime', () => { } }) - it('does not attribute a third-party GitHub park when the in-flight CLI write is a confirmed no-op', async () => { + 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, { @@ -2234,6 +2237,16 @@ describe('fleet CLI runtime', () => { 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() @@ -2249,7 +2262,7 @@ describe('fleet CLI runtime', () => { mount, createFactory: (factoryConfig, ports) => createFactory(factoryConfig, { ...ports, - githubWriteback, + githubWriteback: effectiveGithubWriteback, }), stdout: output, stderr: errors, diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 1fa96f73..414e6cb5 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -13966,7 +13966,10 @@ export class FactoryLoop implements Factory { if (githubIssue) { if (humanReview) { const statusWrite = await this.#githubWriteback.setStatus(issue, 'human-review') - if (statusWrite !== 'already-matched') { + // 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 diff --git a/src/ports/writeback.ts b/src/ports/writeback.ts index 839ded95..7ef6fcf7 100644 --- a/src/ports/writeback.ts +++ b/src/ports/writeback.ts @@ -20,11 +20,12 @@ export interface GithubRead { export type GithubIssueStatus = 'ready' | 'in-progress' | 'human-review' /** - * Whether a lifecycle call issued a provider mutation. `undefined` preserves - * compatibility with caller-supplied writebacks that predate this receipt; - * built-in adapters must return an explicit result. + * Whether a lifecycle call provably changed the provider's effective status. + * `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' +export type GithubStatusWriteResult = 'applied' | 'already-matched' | 'acknowledged' export interface GithubWriteback { /** Optional local-user PR publisher, implemented by the default `gh` writeback. */ diff --git a/src/writeback/github.ts b/src/writeback/github.ts index 5fd7c98a..f995fa26 100644 --- a/src/writeback/github.ts +++ b/src/writeback/github.ts @@ -111,7 +111,7 @@ export class AppGithubWriteback implements GithubWriteback { author: 'app', }) } - return 'applied' + return 'acknowledged' } const target = FACTORY_GITHUB_STATUS_LABELS[status] const previous = FACTORY_GITHUB_STATUS_LABELS[status === 'in-progress' ? 'human-review' : 'in-progress'] @@ -134,7 +134,11 @@ export class AppGithubWriteback implements GithubWriteback { label: previous.name, author: 'app', }) - return 'applied' + // 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 { @@ -334,6 +338,7 @@ export class GhCliGithubWriteback implements GithubWriteback { '--force', ]) const labels = await this.#issueLabels(ref) + const statusBefore = githubStatusFromLabels(labels) const editArgs = [ 'issue', 'edit', @@ -353,7 +358,10 @@ export class GhCliGithubWriteback implements GithubWriteback { } const confirmed = await this.#issueLabels(ref) if (confirmed.has(target.name.toLowerCase()) && !confirmed.has(previous.name.toLowerCase())) { - return applied ? 'applied' : 'already-matched' + // 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). + return statusBefore === status ? 'already-matched' : 'applied' } throw new Error(`GitHub writeback did not confirm ${target.name} on ${ref.repo}#${ref.number}`) } @@ -404,6 +412,12 @@ 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 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 2f07d379..b3c5e8ed 100644 --- a/src/writeback/writeback.test.ts +++ b/src/writeback/writeback.test.ts @@ -856,8 +856,8 @@ 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 expect(app.setStatus(appIssue, 'human-review')).resolves.toBe('acknowledged') + await expect(app.setStatus(appIssue, 'ready')).resolves.toBe('acknowledged') await app.closeIssue(appIssue, 'Factory observed the linked PR merge.') expect(postIssueComment).toHaveBeenNthCalledWith(1, { @@ -1134,6 +1134,26 @@ describe('GhCliGithubWriteback', () => { 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({ From b5dbe7c534d5d442f0f6ef8335c09fbf9f54b30b Mon Sep 17 00:00:00 2001 From: Khaliq Date: Sun, 23 Aug 2026 18:26:47 +0200 Subject: [PATCH 08/14] fix(writeback): require provider authorship receipt (#319) --- src/cli/fleet.test.ts | 2 +- src/ports/writeback.ts | 1 + src/writeback/github.ts | 99 +++++++++++++++++++++++++++++++-- src/writeback/writeback.test.ts | 84 +++++++++++++++++++++++++++- 4 files changed, 176 insertions(+), 10 deletions(-) diff --git a/src/cli/fleet.test.ts b/src/cli/fleet.test.ts index 377058c0..b4209f31 100644 --- a/src/cli/fleet.test.ts +++ b/src/cli/fleet.test.ts @@ -25,7 +25,7 @@ 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' diff --git a/src/ports/writeback.ts b/src/ports/writeback.ts index 7ef6fcf7..b1501bf7 100644 --- a/src/ports/writeback.ts +++ b/src/ports/writeback.ts @@ -21,6 +21,7 @@ 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. diff --git a/src/writeback/github.ts b/src/writeback/github.ts index f995fa26..5500967a 100644 --- a/src/writeback/github.ts +++ b/src/writeback/github.ts @@ -63,6 +63,18 @@ export interface GhCliGithubWritebackConfig { gitRunner?: GhRunner } +interface GithubLabelEvent { + id: string + event: 'labeled' | 'unlabeled' + label: string + actor: string +} + +interface GithubLabelReceiptBaseline { + actor: string + eventIds: Set +} + type AppIssueConnectionWrite = GithubConnectionWrite & Required 5 - if (applied) { + const editRequired = editArgs.length > 5 + const receiptBaseline = editRequired + ? await this.#labelReceiptBaseline(ref).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 applied ? 'applied' : 'already-matched' + 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'] @@ -352,8 +371,11 @@ export class GhCliGithubWriteback implements GithubWriteback { if (labels.has(previous.name.toLowerCase())) { editArgs.push('--remove-label', previous.name) } - const applied = editArgs.length > 5 - if (applied) { + const editRequired = editArgs.length > 5 + const receiptBaseline = editRequired + ? await this.#labelReceiptBaseline(ref).catch(() => undefined) + : undefined + if (editRequired) { await this.#run(editArgs) } const confirmed = await this.#issueLabels(ref) @@ -361,7 +383,10 @@ export class GhCliGithubWriteback implements GithubWriteback { // 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). - return statusBefore === status ? 'already-matched' : 'applied' + 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}`) } @@ -387,6 +412,50 @@ export class GhCliGithubWriteback implements GithubWriteback { ) } + async #labelReceiptBaseline(ref: { repo: string; number: number }): 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) + return { actor, eventIds: new Set(events.map((event) => event.id)) } + } + + 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(() => []) + return events.some((event) => + !baseline.eventIds.has(event.id) + && event.actor === baseline.actor + && event.event === expected.event + && event.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) @@ -418,6 +487,24 @@ const githubStatusFromLabels = (labels: Set): GithubIssueStatus => { 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 b3c5e8ed..642fdbf6 100644 --- a/src/writeback/writeback.test.ts +++ b/src/writeback/writeback.test.ts @@ -929,6 +929,14 @@ 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', + ] it('pushes a local branch and returns the gh-authenticated PR author', async () => { const ghCalls: string[][] = [] @@ -1042,17 +1050,27 @@ 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: '' } }, @@ -1066,21 +1084,31 @@ describe('GhCliGithubWriteback', () => { ['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,7 +1116,10 @@ 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: '' } @@ -1099,6 +1130,8 @@ describe('GhCliGithubWriteback', () => { expect(calls).toEqual([ ['issue', 'view', '48', '--repo', 'AgentWorkforce/factory', '--json', 'labels'], + authenticatedActorCall, + issueLabelEventsCall, [ 'issue', 'edit', @@ -1111,9 +1144,54 @@ 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.each([ { status: 'ready' as const, labels: [] }, { status: 'in-progress' as const, labels: ['factory:in-progress'] }, From 2fae2c675a351dd50ed5b9181f8e3d1f65f6b6ed Mon Sep 17 00:00:00 2001 From: Khaliq Date: Sun, 23 Aug 2026 18:44:22 +0200 Subject: [PATCH 09/14] fix(writeback): require final authored status transition --- src/writeback/github.ts | 21 +++++++++---- src/writeback/writeback.test.ts | 52 +++++++++++++++++++++++++++++++++ 2 files changed, 68 insertions(+), 5 deletions(-) diff --git a/src/writeback/github.ts b/src/writeback/github.ts index 5500967a..6b37ecc9 100644 --- a/src/writeback/github.ts +++ b/src/writeback/github.ts @@ -429,12 +429,23 @@ export class GhCliGithubWriteback implements GithubWriteback { const expected = githubStatusTransitionEvent(from, to) if (!expected) return false const events = await this.#issueLabelEvents(ref).catch(() => []) - return events.some((event) => - !baseline.eventIds.has(event.id) - && event.actor === baseline.actor - && event.event === expected.event - && event.label === expected.label, + 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), ) + let definingIndex = -1 + for (let index = newStatusEvents.length - 1; index >= 0; index -= 1) { + const event = newStatusEvents[index] + if (event?.event === expected.event && event.label === expected.label) { + definingIndex = index + break + } + } + if (definingIndex < 0 || newStatusEvents[definingIndex]?.actor !== baseline.actor) return false + // The matching event is not enough if another actor later rewrites any + // Factory status label before final-state confirmation. In that case the + // visible status was recreated/superseded by the later actor. + return newStatusEvents.slice(definingIndex + 1).every((event) => event.actor === baseline.actor) } async #issueLabelEvents(ref: { repo: string; number: number }): Promise { diff --git a/src/writeback/writeback.test.ts b/src/writeback/writeback.test.ts index 642fdbf6..c02c723d 100644 --- a/src/writeback/writeback.test.ts +++ b/src/writeback/writeback.test.ts @@ -1192,6 +1192,58 @@ describe('GhCliGithubWriteback', () => { 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('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'] }, From f89353ee9654b6ed95a8736375ba299786401fac Mon Sep 17 00:00:00 2001 From: Khaliq Date: Sun, 23 Aug 2026 18:52:36 +0200 Subject: [PATCH 10/14] fix(writeback): attribute effective final status transition --- src/writeback/github.ts | 43 +++++++++++++++++++++------------ src/writeback/writeback.test.ts | 23 ++++++++++++++++++ 2 files changed, 50 insertions(+), 16 deletions(-) diff --git a/src/writeback/github.ts b/src/writeback/github.ts index 6b37ecc9..ddbfdeda 100644 --- a/src/writeback/github.ts +++ b/src/writeback/github.ts @@ -73,6 +73,7 @@ interface GithubLabelEvent { interface GithubLabelReceiptBaseline { actor: string eventIds: Set + statusLabels: Set } type AppIssueConnectionWrite = GithubConnectionWrite & Required 5 const receiptBaseline = editRequired - ? await this.#labelReceiptBaseline(ref).catch(() => undefined) + ? await this.#labelReceiptBaseline(ref, labels).catch(() => undefined) : undefined if (editRequired) { await this.#run(editArgs) @@ -373,7 +374,7 @@ export class GhCliGithubWriteback implements GithubWriteback { } const editRequired = editArgs.length > 5 const receiptBaseline = editRequired - ? await this.#labelReceiptBaseline(ref).catch(() => undefined) + ? await this.#labelReceiptBaseline(ref, labels).catch(() => undefined) : undefined if (editRequired) { await this.#run(editArgs) @@ -412,11 +413,19 @@ export class GhCliGithubWriteback implements GithubWriteback { ) } - async #labelReceiptBaseline(ref: { repo: string; number: number }): 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) - return { actor, eventIds: new Set(events.map((event) => event.id)) } + 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( @@ -433,19 +442,21 @@ export class GhCliGithubWriteback implements GithubWriteback { const newStatusEvents = events.filter((event) => !baseline.eventIds.has(event.id) && statusLabels.has(event.label), ) - let definingIndex = -1 - for (let index = newStatusEvents.length - 1; index >= 0; index -= 1) { - const event = newStatusEvents[index] - if (event?.event === expected.event && event.label === expected.label) { - definingIndex = index - break - } + 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 (definingIndex < 0 || newStatusEvents[definingIndex]?.actor !== baseline.actor) return false - // The matching event is not enough if another actor later rewrites any - // Factory status label before final-state confirmation. In that case the - // visible status was recreated/superseded by the later actor. - return newStatusEvents.slice(definingIndex + 1).every((event) => event.actor === baseline.actor) + 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 { diff --git a/src/writeback/writeback.test.ts b/src/writeback/writeback.test.ts index c02c723d..b66bece9 100644 --- a/src/writeback/writeback.test.ts +++ b/src/writeback/writeback.test.ts @@ -1219,6 +1219,29 @@ describe('GhCliGithubWriteback', () => { 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[] = [] From 882ceab4f5eba9b33d1324fac33145dc1119632a Mon Sep 17 00:00:00 2001 From: Khaliq Date: Sun, 23 Aug 2026 20:14:43 +0200 Subject: [PATCH 11/14] fix(writeback): prove GitHub close ownership --- src/cli/fleet.test.ts | 176 ++++++++++++++++++++++++++++++++ src/index.ts | 1 + src/orchestrator/factory.ts | 10 +- src/ports/index.ts | 1 + src/ports/writeback.ts | 9 +- src/writeback/github.ts | 98 +++++++++++++++++- src/writeback/writeback.test.ts | 69 ++++++++++++- 7 files changed, 356 insertions(+), 8 deletions(-) diff --git a/src/cli/fleet.test.ts b/src/cli/fleet.test.ts index b4209f31..646f3e51 100644 --- a/src/cli/fleet.test.ts +++ b/src/cli/fleet.test.ts @@ -2280,6 +2280,182 @@ describe('fleet CLI runtime', () => { } }) + 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') + } finally { + 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 1e8a1d61..3eb8b5ad 100644 --- a/src/index.ts +++ b/src/index.ts @@ -352,6 +352,7 @@ export type { SpawnResult, AgentUsage, GithubRead, + GithubIssueCloseWriteResult, GithubIssueStatus, GithubStatusWriteResult, GithubWriteback, diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 414e6cb5..8c4ad816 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -13980,11 +13980,17 @@ export class FactoryLoop implements Factory { `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.', ) - record.issueWritebackConfirmedAtMs ??= this.#clock.now() + // 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 diff --git a/src/ports/index.ts b/src/ports/index.ts index 4a5671e0..8bd0acc0 100644 --- a/src/ports/index.ts +++ b/src/ports/index.ts @@ -41,6 +41,7 @@ export type { } from './fleet' export type { GithubRead, + GithubIssueCloseWriteResult, GithubIssueStatus, GithubStatusWriteResult, GithubWriteback, diff --git a/src/ports/writeback.ts b/src/ports/writeback.ts index b1501bf7..92842f16 100644 --- a/src/ports/writeback.ts +++ b/src/ports/writeback.ts @@ -28,6 +28,13 @@ export type GithubIssueStatus = 'ready' | 'in-progress' | 'human-review' */ 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 @@ -39,5 +46,5 @@ export interface GithubWriteback { /** 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 + closeIssue(issue: LinearIssue, body: string): Promise } diff --git a/src/writeback/github.ts b/src/writeback/github.ts index ddbfdeda..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, GithubStatusWriteResult, 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' @@ -76,6 +81,17 @@ interface GithubLabelReceiptBaseline { statusLabels: Set } +interface GithubIssueStateEvent { + id: string + event: 'closed' | 'reopened' + actor: string +} + +interface GithubIssueCloseReceiptBaseline { + actor: string + eventIds: Set +} + type AppIssueConnectionWrite = GithubConnectionWrite & Required { + async closeIssue(issue: LinearIssue, body: string): Promise { const ref = githubIssueRef(issue) await this.postComment(issue, body) await this.#write.updateIssue({ @@ -163,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' } } @@ -478,9 +497,11 @@ export class GhCliGithubWriteback implements GithubWriteback { }) } - async closeIssue(issue: LinearIssue, body: string): Promise { + 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', @@ -490,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 { diff --git a/src/writeback/writeback.test.ts b/src/writeback/writeback.test.ts index b66bece9..c552aeaf 100644 --- a/src/writeback/writeback.test.ts +++ b/src/writeback/writeback.test.ts @@ -858,7 +858,7 @@ describe('AppGithubWriteback', () => { await app.postComment(appIssue, 'Factory dispatch for 221') await expect(app.setStatus(appIssue, 'human-review')).resolves.toBe('acknowledged') await expect(app.setStatus(appIssue, 'ready')).resolves.toBe('acknowledged') - await app.closeIssue(appIssue, 'Factory observed the linked PR merge.') + await expect(app.closeIssue(appIssue, 'Factory observed the linked PR merge.')).resolves.toBe('acknowledged') expect(postIssueComment).toHaveBeenNthCalledWith(1, { repo: 'AgentWorkforce/factory', @@ -937,6 +937,13 @@ describe('GhCliGithubWriteback', () => { '--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[][] = [] @@ -1339,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'], ]) }) From 33b0aa4dfdedcac513d7efbf8aafb7a5735ad4ef Mon Sep 17 00:00:00 2001 From: Khaliq Date: Sun, 23 Aug 2026 20:30:22 +0200 Subject: [PATCH 12/14] fix(factory): serialize terminal ownership observation --- src/cli/fleet.test.ts | 12 +++++++++++ src/orchestrator/factory.ts | 40 +++++++++++++++++++++++++++++++++++++ 2 files changed, 52 insertions(+) diff --git a/src/cli/fleet.test.ts b/src/cli/fleet.test.ts index 646f3e51..6b6e788d 100644 --- a/src/cli/fleet.test.ts +++ b/src/cli/fleet.test.ts @@ -2275,6 +2275,12 @@ describe('fleet CLI runtime', () => { 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 }) } @@ -2451,6 +2457,12 @@ describe('fleet CLI runtime', () => { 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 }) } diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 8c4ad816..0daa4084 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -817,6 +817,11 @@ export class FactoryLoop implements Factory { #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>() readonly #agentExitsInFlight = new Map>() #reconciledAgentExitsActive = 0 readonly #reconciledAgentExitWaiters: Array<() => void> = [] @@ -1371,6 +1376,7 @@ export class FactoryLoop implements Factory { this.#livePollInFlight = false this.#liveEventQueue.length = 0 this.#completionInFlight.clear() + this.#postSpawnIssueObservations.clear() this.#babysitterSpawned.clear() this.#babysitterPr.clear() this.#babysitterIssueRefs.clear() @@ -4553,6 +4559,21 @@ 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 postSpawnIssueObservationSettled = false + const postSpawnIssueObservation = new Promise((resolve) => { + resolvePostSpawnIssueObservation = resolve + }) + this.#postSpawnIssueObservations.set(observationKey, postSpawnIssueObservation) + const settlePostSpawnIssueObservation = (accepted: boolean): void => { + if (postSpawnIssueObservationSettled) return + postSpawnIssueObservationSettled = true + resolvePostSpawnIssueObservation(accepted) + if (this.#postSpawnIssueObservations.get(observationKey) === postSpawnIssueObservation) { + this.#postSpawnIssueObservations.delete(observationKey) + } + } try { if (!dryRun) { const issue = await this.#readIssue(dispatchDecision.issue.path) @@ -4625,6 +4646,7 @@ export class FactoryLoop implements Factory { this.#emit('writeback-verified', { issue: dispatchDecision.issue, path: issue.path }) } } + settlePostSpawnIssueObservation(true) const result = { issue: dispatchDecision.issue, @@ -4648,6 +4670,7 @@ export class FactoryLoop implements Factory { } return result } catch (error) { + 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. @@ -13896,6 +13919,7 @@ export class FactoryLoop implements Factory { return } this.#completionInFlight.add(completionKey) + const postSpawnIssueObservation = this.#postSpawnIssueObservations.get(completionKey) let settleIssueWriteback!: () => void const issueWritebackSettled = new Promise((resolve) => { settleIssueWriteback = resolve @@ -13966,6 +13990,7 @@ export class FactoryLoop implements Factory { if (githubIssue) { if (humanReview) { 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. @@ -13984,6 +14009,7 @@ export class FactoryLoop implements Factory { 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 @@ -14007,6 +14033,11 @@ export class FactoryLoop implements Factory { // 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) return if (!humanReview) await this.#markDependencyTerminalAndReconcile(issue) } else { settleIssueWritebackOnce() @@ -14092,6 +14123,15 @@ 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, + }) + } + #emit(event: FactoryEvent, payload: FactoryEventPayload): void { for (const listener of this.#listeners.get(event) ?? []) { try { From 09d12a54cf369e5d03671ae0790fd8f9a10fb857 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Sun, 23 Aug 2026 20:51:55 +0200 Subject: [PATCH 13/14] fix(factory): serialize dispatch claim with completion --- src/cli/fleet.test.ts | 163 ++++++++++++++++++++++++++++++++++++ src/orchestrator/factory.ts | 61 +++++++++++++- 2 files changed, 223 insertions(+), 1 deletion(-) diff --git a/src/cli/fleet.test.ts b/src/cli/fleet.test.ts index 6b6e788d..09f1d01d 100644 --- a/src/cli/fleet.test.ts +++ b/src/cli/fleet.test.ts @@ -2468,6 +2468,169 @@ describe('fleet CLI runtime', () => { } }) + 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/orchestrator/factory.ts b/src/orchestrator/factory.ts index 0daa4084..1bc0b3c6 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -603,6 +603,13 @@ class ReadinessReconcileTimeoutError extends Error { } } +interface PostSpawnDispatchClaimFence { + completionAtWriteBoundary: boolean + claimStarted: boolean + accepted?: boolean + settled: Promise +} + const realClock: Clock = { now: () => Date.now(), sleep: (ms) => new Promise((resolve) => setTimeout(resolve, ms)), @@ -822,6 +829,10 @@ export class FactoryLoop implements Factory { // 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> = [] @@ -1377,6 +1388,7 @@ export class FactoryLoop implements Factory { this.#liveEventQueue.length = 0 this.#completionInFlight.clear() this.#postSpawnIssueObservations.clear() + this.#postSpawnDispatchClaimFences.clear() this.#babysitterSpawned.clear() this.#babysitterPr.clear() this.#babysitterIssueRefs.clear() @@ -4566,6 +4578,17 @@ export class FactoryLoop implements Factory { resolvePostSpawnIssueObservation = resolve }) this.#postSpawnIssueObservations.set(observationKey, postSpawnIssueObservation) + let resolvePostSpawnDispatchClaim!: (accepted: boolean) => void + let postSpawnDispatchClaimSettled = false + const postSpawnDispatchClaim = new Promise((resolve) => { + resolvePostSpawnDispatchClaim = resolve + }) + const postSpawnDispatchClaimFence: PostSpawnDispatchClaimFence = { + completionAtWriteBoundary: false, + claimStarted: false, + settled: postSpawnDispatchClaim, + } + this.#postSpawnDispatchClaimFences.set(observationKey, postSpawnDispatchClaimFence) const settlePostSpawnIssueObservation = (accepted: boolean): void => { if (postSpawnIssueObservationSettled) return postSpawnIssueObservationSettled = true @@ -4574,6 +4597,15 @@ export class FactoryLoop implements Factory { this.#postSpawnIssueObservations.delete(observationKey) } } + const settlePostSpawnDispatchClaim = (accepted: boolean): void => { + if (postSpawnDispatchClaimSettled) return + postSpawnDispatchClaimSettled = true + postSpawnDispatchClaimFence.accepted = accepted + resolvePostSpawnDispatchClaim(accepted) + if (this.#postSpawnDispatchClaimFences.get(observationKey) === postSpawnDispatchClaimFence) { + this.#postSpawnDispatchClaimFences.delete(observationKey) + } + } try { if (!dryRun) { const issue = await this.#readIssue(dispatchDecision.issue.path) @@ -4610,7 +4642,15 @@ export class FactoryLoop implements Factory { let implementingStateId: string | undefined if (!dryRun) { let issue = await this.#readIssue(dispatchDecision.issue.path) - if ((!issue || !this.#isIssueReady(issue)) && record.issueWritebackConfirmedAtMs === undefined) { + 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 @@ -4641,11 +4681,16 @@ export class FactoryLoop implements Factory { // 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() this.#emit('writeback-verified', { issue: dispatchDecision.issue, path: issue.path }) } } + settlePostSpawnDispatchClaim(true) settlePostSpawnIssueObservation(true) const result = { @@ -4670,6 +4715,7 @@ 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 @@ -13920,6 +13966,7 @@ export class FactoryLoop implements Factory { } 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 @@ -13982,6 +14029,18 @@ 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 From 88cfd41009073dafd827eb4c361b279caf325df6 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Sun, 23 Aug 2026 21:11:25 +0200 Subject: [PATCH 14/14] fix(factory): settle post-spawn waits on teardown --- src/orchestrator/factory.test.ts | 119 +++++++++++++++++++++++++++++++ src/orchestrator/factory.ts | 87 ++++++++++++++++------ 2 files changed, 184 insertions(+), 22 deletions(-) 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 1bc0b3c6..f96ee73e 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -608,6 +608,12 @@ interface PostSpawnDispatchClaimFence { claimStarted: boolean accepted?: boolean settled: Promise + settle(accepted: boolean): void +} + +interface PostSpawnIssueObservation { + settled: Promise + settle(accepted: boolean): void } const realClock: Clock = { @@ -828,7 +834,7 @@ export class FactoryLoop implements Factory { // 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>() + 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. @@ -1320,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) @@ -4573,39 +4588,46 @@ export class FactoryLoop implements Factory { const spawnedForReaperHandoff: RegistryHandoffAgent[] = [] const observationKey = issueKey(record.issue) let resolvePostSpawnIssueObservation!: (accepted: boolean) => void - let postSpawnIssueObservationSettled = false - const postSpawnIssueObservation = new Promise((resolve) => { + 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 }) - const postSpawnDispatchClaimFence: PostSpawnDispatchClaimFence = { + 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 = (accepted: boolean): void => { - if (postSpawnIssueObservationSettled) return - postSpawnIssueObservationSettled = true - resolvePostSpawnIssueObservation(accepted) - if (this.#postSpawnIssueObservations.get(observationKey) === postSpawnIssueObservation) { - this.#postSpawnIssueObservations.delete(observationKey) - } - } - const settlePostSpawnDispatchClaim = (accepted: boolean): void => { - if (postSpawnDispatchClaimSettled) return - postSpawnDispatchClaimSettled = true - postSpawnDispatchClaimFence.accepted = accepted - resolvePostSpawnDispatchClaim(accepted) - if (this.#postSpawnDispatchClaimFences.get(observationKey) === postSpawnDispatchClaimFence) { - this.#postSpawnDispatchClaimFences.delete(observationKey) - } - } + const settlePostSpawnIssueObservation = postSpawnIssueObservation.settle + const settlePostSpawnDispatchClaim = postSpawnDispatchClaimFence.settle try { if (!dryRun) { const issue = await this.#readIssue(dispatchDecision.issue.path) @@ -9707,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) @@ -14096,7 +14123,7 @@ export class FactoryLoop implements Factory { // 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) return + if (postSpawnIssueObservation && !await postSpawnIssueObservation.settled) return if (!humanReview) await this.#markDependencyTerminalAndReconcile(issue) } else { settleIssueWritebackOnce() @@ -14191,6 +14218,22 @@ export class FactoryLoop implements Factory { }) } + #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 {