From cd6f0a92fedc47b995dd3cbdb185bba851a38483 Mon Sep 17 00:00:00 2001 From: Miya Date: Thu, 16 Jul 2026 23:08:42 +0200 Subject: [PATCH 1/4] Release agents while awaiting clarification --- src/config/schema.test.ts | 1 + src/config/schema.ts | 4 + src/dispatch/templates.test.ts | 17 +- src/dispatch/templates.ts | 11 +- src/orchestrator/factory.test.ts | 562 +++++++++++++++++++++++- src/orchestrator/factory.ts | 666 ++++++++++++++++++++++++++++- src/ports/state.ts | 53 +++ src/state/file-state-store.test.ts | 100 ++++- src/state/file-state-store.ts | 324 +++++++++++++- src/state/in-memory-state-store.ts | 150 +++++++ 10 files changed, 1840 insertions(+), 48 deletions(-) diff --git a/src/config/schema.test.ts b/src/config/schema.test.ts index 4592fe8..c926ca8 100644 --- a/src/config/schema.test.ts +++ b/src/config/schema.test.ts @@ -36,6 +36,7 @@ describe('FactoryConfigSchema', () => { channel: 'C123', style: 'threaded-summarized', botUserId: 'U0B2596R7EZ', + stakeholderUserIds: [], staleAfterMs: 10 * 60_000, }) expect(parsed.mergePolicy).toBe('never') diff --git a/src/config/schema.ts b/src/config/schema.ts index 9af82e9..3031de2 100644 --- a/src/config/schema.ts +++ b/src/config/schema.ts @@ -91,6 +91,10 @@ const slackSchema = z.object({ channel: z.string(), style: z.literal('threaded-summarized').default('threaded-summarized'), botUserId: z.string().default('U0B2596R7EZ'), + // Slack user IDs to mention when an agent needs a human decision. Keeping + // this explicit avoids guessing that a Linear assignee name is also a Slack + // identity, while still making parked questions immediately actionable. + stakeholderUserIds: z.array(z.string().min(1)).default([]), staleAfterMs: z.number().int().min(1_000).default(10 * 60_000), }).optional() diff --git a/src/dispatch/templates.test.ts b/src/dispatch/templates.test.ts index b93614f..0c36dc6 100644 --- a/src/dispatch/templates.test.ts +++ b/src/dispatch/templates.test.ts @@ -34,7 +34,7 @@ describe('renderAgentTask', () => { expect(task).toContain('Factory will open the PR targeting the repository default branch through the connected GitHub workspace.') expect(task).toContain('Do not run `gh pr create` or require local GitHub CLI authentication.') expect(task).toContain('Factory will hand the opened PR to reviewer `ar-123-review`.') - expect(task).toContain('write to the .integrations mount path so the factory can relay it to the issue thread.') + expect(task).toContain('DM `factory` with `[factory-needs-input]`') expect(task).toContain('DM `broker` when fully done.') expect(task).toContain('Do NOT auto-merge.') expect(task).toContain('Merge policy: never') @@ -238,7 +238,7 @@ describe('renderAgentTask', () => { expect(task).not.toMatch(/\n\n(.integrations|Connected integrations)/m) }) - it('directs agent questions to .integrations writeback instead of relay DM', () => { + it('makes the needs-input marker an explicit release boundary', () => { const task = renderAgentTask({ issue, route: { repo: 'pear', clonePath: '/tmp/pear' }, @@ -248,14 +248,11 @@ describe('renderAgentTask', () => { slackDispatchThread: { channel: 'C123', threadId: '169.000', mountRoot: '/work/.integrations' }, }) - // The Slack-thread writeback replaces the old relay DM pattern. - expect(task).toContain('write your question to this issue\'s Slack dispatch thread via the .integrations mount') - expect(task).toContain('Write path: /work/.integrations/slack/channels/C123/messages/169_000/replies/question.json') - expect(task).toContain('Write a JSON object with a "text" field') - expect(task).toContain('Continue with safe reversible work while waiting for a reply.') - // No relay DM or legacy patterns. + expect(task).toContain('DM `factory` with `[factory-needs-input]`') + expect(task).toContain('/work/.integrations/slack/channels/C123/messages/169_000/replies/question.json') + expect(task).toContain('Do not wait or poll') + expect(task).toContain('release the whole team and resume it from session memory only after the first human reply') + expect(task).toContain('cold-start the team with the issue, question, reply, branch, and PR context') expect(task).not.toContain('FACTORY_NEEDS_INPUT') - expect(task).not.toContain('DM `broker` with') - expect(task).not.toContain('[factory-needs-input]') }) }) diff --git a/src/dispatch/templates.ts b/src/dispatch/templates.ts index 5345430..4cafae4 100644 --- a/src/dispatch/templates.ts +++ b/src/dispatch/templates.ts @@ -90,7 +90,7 @@ export function renderAgentTask(input: RenderAgentTaskInput): string { 'When implementation is complete, finish your session normally; Factory will open the PR targeting the repository default branch through the connected GitHub workspace.', 'Do not run `gh pr create` or require local GitHub CLI authentication.', `Factory will hand the opened PR to reviewer \`${input.reviewerName}\`.`, - 'If blocked and you need human input, write to the .integrations mount path so the factory can relay it to the issue thread.', + 'If blocked and you need human input, DM `factory` with `[factory-needs-input]`, the issue key, and one concrete question. Factory will relay it to the issue conversation.', 'DM `broker` when fully done.', 'Do NOT auto-merge.', mergePolicyLine(input.config.mergePolicy), @@ -98,13 +98,12 @@ export function renderAgentTask(input: RenderAgentTaskInput): string { const questionInstructions = input.slackDispatchThread ? [ '', - 'If you are blocked or need a human answer mid-task, write your question to this issue\'s Slack dispatch thread via the .integrations mount.', + 'If you are blocked or need a human answer mid-task, finish any safe reversible work first, then DM `factory` with `[factory-needs-input]`, the issue key, and one concrete question.', // Absolute path: the agent runs in its repo clone, not the daemon cwd // where .integrations lives, so a relative path would be unreachable. - `Write path: ${input.slackDispatchThread.mountRoot}/slack/channels/${input.slackDispatchThread.channel}/messages/${input.slackDispatchThread.threadId.replaceAll('.', '_')}/replies/question.json`, - 'Write a JSON object with a "text" field containing your question.', - 'The human\'s reply will be delivered to you as an `` system message injected into your session — wait for it, do not poll.', - 'Continue with safe reversible work while waiting for a reply.', + `Factory will post the question to the Slack thread represented at ${input.slackDispatchThread.mountRoot}/slack/channels/${input.slackDispatchThread.channel}/messages/${input.slackDispatchThread.threadId.replaceAll('.', '_')}/replies/question.json.`, + 'After sending the marker, stop work and finish your session normally. Do not wait or poll: Factory will release the whole team and resume it from session memory only after the first human reply.', + 'If session resume is unavailable, Factory will cold-start the team with the issue, question, reply, branch, and PR context so work can be re-hydrated explicitly.', ] : [] diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index c6c5da2..bc21e13 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -175,6 +175,7 @@ const slackConfig = (channel = 'C0FACTORY__factory-e2e') => ({ channel, style: 'threaded-summarized' as const, botUserId: 'U0B2596R7EZ', + stakeholderUserIds: [] as string[], staleAfterMs: 10 * 60_000, }) @@ -410,6 +411,33 @@ class ReleaseFailingFleetClient extends FakeFleetClient { } } +class DoubleReleaseFailingUnresolvedFleetClient extends FakeFleetClient { + readonly releaseAttempts: Array<{ name: string; reason?: string }> = [] + failReleases = true + + override async release(name: string, reason?: string): Promise { + this.releaseAttempts.push({ name, reason }) + if (this.failReleases) throw new Error(`control plane unavailable for ${name}`) + await super.release(name, reason) + } + + async resolveAgentPid(_name: string): Promise<{ status: 'unresolved' }> { + return { status: 'unresolved' } + } +} + +class RefuseFirstClarificationParkStateStore extends InMemoryStateStore { + refusalsRemaining = 1 + + override async markClarificationParked(workspaceId: string, issueKey: string, parkedAtMs: number) { + if (this.refusalsRemaining > 0) { + this.refusalsRemaining -= 1 + return undefined + } + return super.markClarificationParked(workspaceId, issueKey, parkedAtMs) + } +} + class ScriptedGithubMergeGate implements GithubMergeGatePort { readonly checks: Array<{ repo: string; number: number; expectedHeadSha?: string }> = [] readonly merges: GithubMergeInput[] = [] @@ -939,6 +967,21 @@ class CloudWritebackFakeMountClient extends FakeMountClient { } } +class HumanReplyDuringQuestionMountClient extends CloudWritebackFakeMountClient { + override async writeFile(path: string, content: unknown, opts?: { guarded?: boolean }): Promise { + await super.writeFile(path, content, opts) + const text = typeof content === 'object' && content !== null && 'text' in content + ? String((content as { text?: unknown }).text ?? '') + : '' + if (!text.includes('needs input')) return + emitSlackReply(this, slackReplyFixturePath('C0FACTORY__factory-e2e', this.threadTs, 'human-immediate'), 'human-immediate', { + text: 'The immediate answer must survive the post-to-park transition.', + user: 'U123', + user_is_bot: false, + }) + } +} + class ConfirmRecordingSlackMountClient extends CloudWritebackFakeMountClient { readonly confirmedPaths: string[] = [] @@ -948,12 +991,41 @@ class ConfirmRecordingSlackMountClient extends CloudWritebackFakeMountClient { } } -class FailFirstSlackReplyMountClient extends CloudWritebackFakeMountClient { - failedReply = false +class BlockingSlackQuestionMountClient extends ConfirmRecordingSlackMountClient { + readonly questionWriteStarted: Promise + #signalQuestionWriteStarted!: () => void + #releaseQuestionWrite!: () => void + readonly #questionWriteReleased: Promise + #blocked = false + + constructor(initialFiles: Record = {}) { + super(initialFiles) + this.questionWriteStarted = new Promise((resolve) => { this.#signalQuestionWriteStarted = resolve }) + this.#questionWriteReleased = new Promise((resolve) => { this.#releaseQuestionWrite = resolve }) + } + + releaseQuestionWrite(): void { + this.#releaseQuestionWrite() + } override async confirmWrite(path: string, opts?: { timeoutMs?: number }): Promise<'acked' | 'pending' | 'failed' | 'timeout'> { - if (!this.failedReply && path.includes('/replies/')) { - this.failedReply = true + if (!this.#blocked && path.includes('/replies/')) { + this.#blocked = true + this.#signalQuestionWriteStarted() + await this.#questionWriteReleased + } + return super.confirmWrite(path, opts) + } +} + +class FailNextSlackReplyMountClient extends CloudWritebackFakeMountClient { + failNextReply = false + failedReplies = 0 + + override async confirmWrite(path: string, opts?: { timeoutMs?: number }): Promise<'acked' | 'pending' | 'failed' | 'timeout'> { + if (this.failNextReply && path.includes('/replies/')) { + this.failNextReply = false + this.failedReplies += 1 return 'failed' } return super.confirmWrite(path, opts) @@ -8313,6 +8385,8 @@ describe('FactoryLoop', () => { it('routes a mid-task agent question to the Slack dispatch thread and returns the human answer via sendInput', async () => { const mount = new ConfirmRecordingSlackMountClient({ [issuePath(36)]: issueFile(36) }) const fleet = new FakeFleetClient() + fleet.setSessionRef('ar-36-impl-pear', 'session-ar-36-impl-pear') + fleet.setSessionRef('ar-36-review', 'session-ar-36-review') const factory = createFactory(config({ slack: slackConfig() }), { mount, fleet, @@ -8320,14 +8394,14 @@ describe('FactoryLoop', () => { }) await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(36), issueFile(36)))) + mount.files.set(issuePath(36), { content: issueFile(36, implementing) }) fleet.emitAgentMessage({ from: 'ar-36-impl-pear', target: 'broker', body: 'FACTORY_NEEDS_INPUT\nIssue: AR-36\nQuestion: Which retry helper should I use?', eventId: 'agent-question-36', }) - await flush() - await flush() + await vi.waitFor(() => expect(factory.status().counters.agentQuestionTeamsReleased).toBe(1)) expect(slackReplyWrites(mount)).toEqual([ expect.objectContaining({ @@ -8338,6 +8412,11 @@ describe('FactoryLoop', () => { }), ]) expect(factory.status().counters.agentQuestionsPostedToSlack).toBe(1) + expect(fleet.releases).toEqual([ + { name: 'ar-36-impl-pear', reason: 'waiting-for-human' }, + { name: 'ar-36-review', reason: 'waiting-for-human' }, + ]) + expect(factory.status().inFlight).toEqual([]) expect(slackAnswerInputs(fleet)).toEqual([]) emitSlackReply(mount, slackReplyFixturePath('C0FACTORY__factory-e2e', mount.threadTs, 'bot-question-echo'), 'bot-question-echo', { @@ -8354,13 +8433,341 @@ describe('FactoryLoop', () => { user: 'U123', user_is_bot: false, }) - await flush() - await flush() + await vi.waitFor(() => expect(factory.status().counters.clarificationTeamsWoken).toBe(1)) expect(slackAnswerInputs(fleet)).toEqual([ { name: 'ar-36-impl-pear', data: '\nHuman reply in the Slack thread:\nUse the shared retry helper in factory.ts.\n\r' }, + { name: 'ar-36-review', data: '\nHuman reply in the Slack thread:\nUse the shared retry helper in factory.ts.\n\r' }, + ]) + expect(fleet.resumes).toEqual([ + { name: 'ar-36-impl-pear', sessionRef: 'session-ar-36-impl-pear', node: 'self', capability: 'spawn:codex' }, + { name: 'ar-36-review', sessionRef: 'session-ar-36-review', node: 'self', capability: 'spawn:claude' }, ]) + expect(factory.status().inFlight.map((issue) => issue.key)).toEqual(['AR-36']) expect(fleet.messages).toHaveLength(2) + + emitSlackReply(mount, slackReplyFixturePath('C0FACTORY__factory-e2e', mount.threadTs, 'human-noise-36'), 'human-noise-36', { + text: 'One more thought that must not trigger a second wake.', + user: 'U456', + user_is_bot: false, + }) + await flush() + await flush() + expect(fleet.resumes).toHaveLength(2) + }) + + it('tags configured Slack stakeholders when an agent needs input', async () => { + const mount = new ConfirmRecordingSlackMountClient({ [issuePath(44)]: issueFile(44) }) + const fleet = new FakeFleetClient() + const factory = createFactory(config({ + slack: { ...slackConfig(), stakeholderUserIds: ['UOWNER', 'ULEAD'] }, + }), { + mount, + fleet, + triage: new StaticTriage(), + }) + + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(44), issueFile(44)))) + fleet.emitAgentMessage({ + from: 'ar-44-impl-pear', + target: 'factory', + body: '[factory-needs-input] Which owner should approve this?', + eventId: 'agent-question-44', + }) + await vi.waitFor(() => expect(factory.status().counters.agentQuestionTeamsReleased).toBe(1)) + + expect(slackReplyWrites(mount).at(-1)?.content.text).toBe( + '<@UOWNER> <@ULEAD>\nAR-44: ar-44-impl-pear needs input.\nQuestion: Which owner should approve this?', + ) + }) + + it('atomically keeps the first clarification when concurrent agent questions arrive', async () => { + const mount = new BlockingSlackQuestionMountClient({ [issuePath(49)]: issueFile(49) }) + const fleet = new FakeFleetClient() + const stateStore = new InMemoryStateStore({ batchSize: 2 }) + const factory = createFactory(config({ slack: slackConfig() }), { + mount, + fleet, + stateStore, + triage: new StaticTriage(), + }) + + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(49), issueFile(49)))) + fleet.emitAgentMessage({ + from: 'ar-49-impl-pear', + target: 'factory', + body: '[factory-needs-input] Preserve this original question.', + eventId: 'agent-question-49-first', + }) + await mount.questionWriteStarted + + fleet.emitAgentMessage({ + from: 'ar-49-review', + target: 'factory', + body: '[factory-needs-input] This concurrent question must not overwrite progress.', + eventId: 'agent-question-49-second', + }) + await vi.waitFor(() => expect(factory.status().counters.agentQuestionClarificationAlreadyReserved).toBe(1)) + mount.releaseQuestionWrite() + await vi.waitFor(() => expect(factory.status().counters.agentQuestionTeamsReleased).toBe(1)) + + const saved = await stateStore.listWaitingClarifications('factory-test') + expect(saved).toHaveLength(1) + expect(saved[0]?.[1]).toMatchObject({ + askerName: 'ar-49-impl-pear', + question: 'Preserve this original question.', + releasedAgents: ['ar-49-impl-pear', 'ar-49-review'], + }) + expect(slackReplyWrites(mount)).toHaveLength(1) + expect(fleet.releases).toHaveLength(2) + }) + + it('retries a release-complete clarification when the durable parked transition is refused once', async () => { + const mount = new ConfirmRecordingSlackMountClient({ [issuePath(50)]: issueFile(50) }) + const fleet = new FakeFleetClient() + const stateStore = new RefuseFirstClarificationParkStateStore({ batchSize: 2 }) + const factory = createFactory(config({ slack: slackConfig() }), { + mount, + fleet, + stateStore, + triage: new StaticTriage(), + }) + + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(50), issueFile(50)))) + fleet.emitAgentMessage({ + from: 'ar-50-impl-pear', + target: 'factory', + body: '[factory-needs-input] Retry the durable park.', + eventId: 'agent-question-50', + }) + await vi.waitFor(() => expect(factory.status().counters.clarificationParkReleasePending).toBe(1)) + + const pending = (await stateStore.listWaitingClarifications('factory-test'))[0]?.[1] + expect(pending?.releasedAgents).toEqual(['ar-50-impl-pear', 'ar-50-review']) + expect(pending?.parkedAtMs).toBeUndefined() + expect(factory.status().counters.agentQuestionTeamsReleased).toBeUndefined() + expect(factory.status().inFlight).toEqual([]) + + mount.files.set(issuePath(50), { content: issueFile(50, implementing) }) + await factory.runLoop({ maxIterations: 1 }) + + const recovered = (await stateStore.listWaitingClarifications('factory-test'))[0]?.[1] + expect(recovered?.parkedAtMs).toBeTypeOf('number') + expect(factory.status().counters.clarificationParksRecovered).toBe(1) + expect(factory.status().inFlight).toEqual([]) + }) + + it('keeps a clarification release-pending when both release attempts fail and no pid can be resolved', async () => { + const mount = new ConfirmRecordingSlackMountClient({ [issuePath(45)]: issueFile(45) }) + const fleet = new DoubleReleaseFailingUnresolvedFleetClient() + const stateStore = new InMemoryStateStore({ batchSize: 2 }) + const factory = createFactory(config({ slack: slackConfig() }), { + mount, + fleet, + stateStore, + triage: new StaticTriage(), + }) + + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(45), issueFile(45)))) + fleet.emitAgentMessage({ + from: 'ar-45-impl-pear', + target: 'factory', + body: '[factory-needs-input] Should this wait?', + eventId: 'agent-question-45', + }) + await vi.waitFor(() => expect(factory.status().counters.clarificationParkReleasePending).toBe(1)) + + const waiting = (await stateStore.listWaitingClarifications('factory-test'))[0]?.[1] + expect(fleet.releaseAttempts.slice(0, 2)).toEqual([ + { name: 'ar-45-impl-pear', reason: 'waiting-for-human' }, + { name: 'ar-45-impl-pear', reason: 'waiting-for-human' }, + ]) + expect(waiting?.releasedAgents).toBeUndefined() + expect(waiting?.parkedAtMs).toBeUndefined() + expect(factory.status().counters.agentQuestionTeamsReleased).toBeUndefined() + expect(factory.status().inFlight.map((issue) => issue.key)).toEqual(['AR-45']) + + fleet.failReleases = false + mount.files.set(issuePath(45), { content: issueFile(45, implementing) }) + await factory.runLoop({ maxIterations: 1 }) + const retried = (await stateStore.listWaitingClarifications('factory-test'))[0]?.[1] + expect(retried?.releasedAgents).toEqual(['ar-45-impl-pear', 'ar-45-review']) + expect(retried?.parkedAtMs).toBeTypeOf('number') + expect(factory.status().counters.clarificationParksRecovered).toBe(1) + expect(factory.status().inFlight).toEqual([]) + }) + + it.each([ + [46, planning], + [48, 'cancelled-state-id'], + ] as const)('cancels a clarification wake when Linear issue %s moves to %s', async (number, nextStateId) => { + const mount = new ConfirmRecordingSlackMountClient({ [issuePath(number)]: issueFile(number) }) + const fleet = new FakeFleetClient() + const factory = createFactory(config({ slack: slackConfig() }), { + mount, + fleet, + triage: new StaticTriage(), + }) + + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(number), issueFile(number)))) + mount.files.set(issuePath(number), { content: issueFile(number, implementing) }) + fleet.emitAgentMessage({ + from: `ar-${number}-impl-pear`, + target: 'factory', + body: '[factory-needs-input] Should this wake later?', + eventId: `agent-question-${number}`, + }) + await vi.waitFor(() => expect(factory.status().counters.agentQuestionTeamsReleased).toBe(1)) + mount.files.set(issuePath(number), { content: issueFile(number, nextStateId) }) + + emitSlackReply(mount, slackReplyFixturePath('C0FACTORY__factory-e2e', mount.threadTs, `human-answer-${number}`), `human-answer-${number}`, { + text: 'Do not resume after replanning.', + user: 'U123', + user_is_bot: false, + }) + await vi.waitFor(() => expect(factory.status().counters.clarificationWakesCancelledStaleIssue).toBe(1)) + + expect(fleet.resumes).toEqual([]) + expect(fleet.spawns).toHaveLength(2) + }) + + it('cancels a clarification wake when a GitHub issue loses factory scope', async () => { + const path = githubIssuePath('AgentWorkforce', 'pear', 64) + const mount = new ConfirmRecordingSlackMountClient({ + [path]: githubIssueFile(64, { labels: ['factory'] }), + }) + mount.setSubRoot('/linear/issues', 'absent') + const fleet = new FakeFleetClient() + const factory = createFactory(config({ issueSource: 'github', slack: slackConfig() }), { + mount, + fleet, + triage: new StaticTriage(), + githubWriteback: new RecordingGithubWriteback(), + }) + + await factory.dispatch(await factory.triageIssue(parseGithubFactoryIssue(path, githubIssueFile(64, { labels: ['factory'] })))) + mount.files.set(path, { content: githubIssueFile(64, { labels: ['factory', 'factory:in-progress'] }) }) + fleet.emitAgentMessage({ + from: 'ar-64-impl-pear', + target: 'factory', + body: '[factory-needs-input] Is this still in scope?', + eventId: 'agent-question-64', + }) + await vi.waitFor(() => expect(factory.status().counters.agentQuestionTeamsReleased).toBe(1)) + mount.files.set(path, { content: githubIssueFile(64, { labels: ['factory:in-progress'] }) }) + + emitSlackReply(mount, slackReplyFixturePath('C0FACTORY__factory-e2e', mount.threadTs, 'human-answer-64'), 'human-answer-64', { + text: 'This should not wake out of scope.', + user: 'U123', + user_is_bot: false, + }) + await vi.waitFor(() => expect(factory.status().counters.clarificationWakesCancelledStaleIssue).toBe(1)) + + expect(fleet.resumes).toEqual([]) + expect(fleet.spawns).toHaveLength(2) + }) + + it('periodically escalates a clarification that remains unanswered for seven days', async () => { + const clock = new ManualClock() + const mount = new ConfirmRecordingSlackMountClient({ [issuePath(47)]: issueFile(47) }) + const fleet = new FakeFleetClient() + const stateStore = new InMemoryStateStore({ batchSize: 2 }) + const factory = createFactory(config({ + slack: { ...slackConfig(), stakeholderUserIds: ['UOWNER'] }, + }), { + mount, + fleet, + stateStore, + triage: new StaticTriage(), + clock, + }) + + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(47), issueFile(47)))) + mount.files.set(issuePath(47), { content: issueFile(47, implementing) }) + fleet.emitAgentMessage({ + from: 'ar-47-impl-pear', + target: 'factory', + body: '[factory-needs-input] Which timeout policy applies?', + eventId: 'agent-question-47', + }) + await vi.waitFor(() => expect(factory.status().counters.agentQuestionTeamsReleased).toBe(1)) + clock.advance(7 * 24 * 60 * 60_000) + + await factory.runLoop({ maxIterations: 1 }) + + const clarificationMessages = slackReplyWrites(mount).map((write) => write.content.text) + expect(clarificationMessages).toContain( + '<@UOWNER>\nAR-47 has been parked for seven days without a reply.\n' + + 'Question from ar-47-impl-pear: Which timeout policy applies? Reply in this thread to wake the saved agent team, or move the issue out of Agent Implementing to cancel the wake.', + ) + expect(factory.status().counters.clarificationEscalationsPosted).toBe(1) + expect((await stateStore.listWaitingClarifications('factory-test'))[0]?.[1].escalatedAtMs).toBe(clock.now()) + }) + + it('retries a failed seven-day escalation after restart instead of consuming it', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-clarification-escalation-retry-')) + try { + const watchStatePath = join(root, 'factory-state.json') + const clock = new ManualClock() + const mount = new FailNextSlackReplyMountClient({ [issuePath(51)]: issueFile(51) }) + const firstFleet = new FakeFleetClient() + const factoryConfig = config({ + slack: { ...slackConfig(), stakeholderUserIds: ['UOWNER'] }, + }) + const firstFactory = createFactory(factoryConfig, { + mount, + fleet: firstFleet, + stateStore: new FileStateStore({ batchSize: 2, watchStatePath }), + triage: new StaticTriage(), + clock, + }) + + await firstFactory.dispatch(await firstFactory.triageIssue(parseLinearIssue(issuePath(51), issueFile(51)))) + mount.files.set(issuePath(51), { content: issueFile(51, implementing) }) + firstFleet.emitAgentMessage({ + from: 'ar-51-impl-pear', + target: 'factory', + body: '[factory-needs-input] Retry this escalation after restart.', + eventId: 'agent-question-51', + }) + await vi.waitFor(() => expect(firstFactory.status().counters.agentQuestionTeamsReleased).toBe(1)) + clock.advance(7 * 24 * 60 * 60_000) + await firstFactory.stop() + + mount.failNextReply = true + const failingFactory = createFactory(factoryConfig, { + mount, + fleet: new FakeFleetClient(), + stateStore: new FileStateStore({ batchSize: 2, watchStatePath }), + triage: new StaticTriage(), + clock, + }) + await failingFactory.start({ mode: 'live', liveSubscription: { transport: 'subscribe' } }) + await vi.waitFor(() => expect(failingFactory.status().counters.clarificationEscalationFailures).toBe(1)) + const afterFailure = (await new FileStateStore({ batchSize: 2, watchStatePath }) + .listWaitingClarifications('factory-test'))[0]?.[1] + expect(afterFailure?.escalatedAtMs).toBeUndefined() + expect(afterFailure?.escalation).toMatchObject({ owner: '', attempts: 1 }) + expect(mount.failedReplies).toBe(1) + await failingFactory.stop() + + const restartedState = new FileStateStore({ batchSize: 2, watchStatePath }) + const restartedFactory = createFactory(factoryConfig, { + mount, + fleet: new FakeFleetClient(), + stateStore: restartedState, + triage: new StaticTriage(), + clock, + }) + await restartedFactory.start({ mode: 'live', liveSubscription: { transport: 'subscribe' } }) + await vi.waitFor(() => expect(restartedFactory.status().counters.clarificationEscalationsPosted).toBe(1)) + const delivered = (await restartedState.listWaitingClarifications('factory-test'))[0]?.[1] + expect(delivered?.escalatedAtMs).toBe(clock.now()) + expect(delivered?.escalation).toBeUndefined() + await restartedFactory.stop() + } finally { + await rm(root, { recursive: true, force: true }) + } }) it('posts a mid-task agent question to the source GitHub issue when Slack is unconfigured', async () => { @@ -8403,6 +8810,145 @@ describe('FactoryLoop', () => { expect(factory.status().counters.githubAnswersInjected).toBe(1) }) + it('does not lose a human reply that arrives while the question post is completing', async () => { + const mount = new HumanReplyDuringQuestionMountClient({ [issuePath(39)]: issueFile(39) }) + const fleet = new FakeFleetClient() + fleet.setSessionRef('ar-39-impl-pear', 'session-ar-39-impl-pear') + fleet.setSessionRef('ar-39-review', 'session-ar-39-review') + const factory = createFactory(config({ slack: slackConfig() }), { + mount, + fleet, + triage: new StaticTriage(), + }) + + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(39), issueFile(39)))) + mount.files.set(issuePath(39), { content: issueFile(39, implementing) }) + await flush() + fleet.emitAgentMessage({ + from: 'ar-39-impl-pear', + target: 'factory', + body: '[factory-needs-input] Issue: AR-39\nQuestion: Can a fast reply race parking?', + eventId: 'agent-question-39', + }) + + await vi.waitFor(() => expect(factory.status().counters.clarificationTeamsWoken).toBe(1)) + expect(fleet.releases.map((release) => release.reason)).toEqual(['waiting-for-human', 'waiting-for-human']) + expect(fleet.resumes.map((resume) => resume.sessionRef)).toEqual([ + 'session-ar-39-impl-pear', + 'session-ar-39-review', + ]) + expect(slackAnswerInputs(fleet).map((input) => input.data)).toEqual([ + '\nHuman reply in the Slack thread:\nThe immediate answer must survive the post-to-park transition.\n\r', + '\nHuman reply in the Slack thread:\nThe immediate answer must survive the post-to-park transition.\n\r', + ]) + }) + + it('cold-starts with durable issue, question, and reply context when session refs are unavailable', async () => { + const mount = new ConfirmRecordingSlackMountClient({ [issuePath(37)]: issueFile(37) }) + const fleet = new FakeFleetClient() + const factory = createFactory(config({ slack: slackConfig() }), { + mount, + fleet, + triage: new StaticTriage(), + }) + + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(37), issueFile(37)))) + mount.files.set(issuePath(37), { content: issueFile(37, implementing) }) + fleet.emitAgentMessage({ + from: 'ar-37-impl-pear', + target: 'factory', + body: '[factory-needs-input] Issue: AR-37\nQuestion: Which retry helper should I use?', + eventId: 'agent-question-37', + }) + await flush() + await flush() + + emitSlackReply(mount, slackReplyFixturePath('C0FACTORY__factory-e2e', mount.threadTs, 'human-answer-37'), 'human-answer-37', { + text: 'Use retryOnTimeout from factory.ts.', + user: 'U123', + user_is_bot: false, + }) + await flush() + await flush() + + expect(fleet.resumes).toEqual([]) + expect(fleet.spawns).toHaveLength(4) + for (const spawn of fleet.spawns.slice(2)) { + expect(spawn.task).toContain('Factory released this team while waiting for human input') + expect(spawn.task).toContain('Which retry helper should I use?') + expect(spawn.task).toContain('Use retryOnTimeout from factory.ts.') + expect(spawn.task).toContain('Re-hydrate from the issue, branch, worktree, and any open PR') + } + expect(factory.status().counters.clarificationResumeFallbacks).toBe(2) + expect(factory.status().counters.clarificationTeamsWoken).toBe(1) + }) + + it('rearms a durable clarification after restart and wakes from a reply received while stopped', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-clarification-restart-')) + try { + const watchStatePath = join(root, 'factory-state.json') + const mount = new ConfirmRecordingSlackMountClient({ [issuePath(38)]: issueFile(38) }) + const firstFleet = new FakeFleetClient() + firstFleet.setSessionRef('ar-38-impl-pear', 'session-ar-38-impl-pear') + firstFleet.setSessionRef('ar-38-review', 'session-ar-38-review') + const firstFactory = createFactory(config({ slack: slackConfig() }), { + mount, + fleet: firstFleet, + stateStore: new FileStateStore({ batchSize: 2, watchStatePath }), + triage: new StaticTriage(), + }) + + await firstFactory.dispatch(await firstFactory.triageIssue(parseLinearIssue(issuePath(38), issueFile(38)))) + mount.files.set(issuePath(38), { content: issueFile(38, implementing) }) + firstFleet.emitAgentMessage({ + from: 'ar-38-impl-pear', + target: 'factory', + body: '[factory-needs-input] Issue: AR-38\nQuestion: Which durable path?', + eventId: 'agent-question-38', + }) + await vi.waitFor(() => expect(firstFactory.status().counters.agentQuestionTeamsReleased).toBe(1)) + await firstFactory.stop() + + emitSlackReply(mount, slackReplyFixturePath('C0FACTORY__factory-e2e', mount.threadTs, 'human-answer-38'), 'human-answer-38', { + text: 'Resume through the stored session refs.', + user: 'U123', + user_is_bot: false, + }) + + const restartedFleet = new FakeFleetClient() + const restartedFactory = createFactory(config({ slack: slackConfig() }), { + mount, + fleet: restartedFleet, + stateStore: new FileStateStore({ batchSize: 2, watchStatePath }), + triage: new StaticTriage(), + }) + await restartedFactory.start({ mode: 'live', liveSubscription: { transport: 'subscribe' } }) + + await vi.waitFor(() => expect(restartedFleet.resumes).toHaveLength(2)) + expect(restartedFleet.resumes.map((resume) => resume.sessionRef)).toEqual([ + 'session-ar-38-impl-pear', + 'session-ar-38-review', + ]) + expect(slackAnswerInputs(restartedFleet)).toHaveLength(2) + expect(restartedFactory.status().counters.clarificationTeamsWoken).toBe(1) + + restartedFleet.emitAgentMessage({ + from: 'ar-38-impl-pear', + target: 'factory', + body: '[factory-needs-input] Issue: AR-38\nQuestion: A second question after restart?', + eventId: 'agent-question-38-second', + }) + await vi.waitFor(() => expect(restartedFactory.status().counters.agentQuestionTeamsReleased).toBe(1)) + expect(restartedFleet.releases).toEqual([ + { name: 'ar-38-impl-pear', reason: 'waiting-for-human' }, + { name: 'ar-38-review', reason: 'waiting-for-human' }, + ]) + await restartedFactory.stop() + } finally { + await rm(root, { recursive: true, force: true }) + } + }) + it('rearms a persisted mid-task GitHub question and replays a reply received while stopped', async () => { const path = githubIssuePath('AgentWorkforce', 'pear', 63) const issue = githubIssueFile(63, { labels: ['factory'], author: 'reporter' }) diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 6076d91..95e63a0 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -1,3 +1,4 @@ +import { randomUUID } from 'node:crypto' import { mkdir, readFile, writeFile } from 'node:fs/promises' import { dirname, isAbsolute, resolve } from 'node:path' @@ -25,6 +26,7 @@ import type { GithubIssueCommentWatchState, RegistryHandoffAgent, StateStore, + WaitingClarification, } from '../ports/state' import type { Clock, Logger } from '../ports/system' import { InMemoryStateStore } from '../state/in-memory-state-store' @@ -118,6 +120,7 @@ type GithubIssueSource = { path: string raw: Record } +class ClarificationWakeLeaseLostError extends Error {} type SlackSyncStatusSeverity = 'soft' | 'hard' type SlackSyncStatusCheck = { known: boolean; degraded: boolean; reason?: string; severity?: SlackSyncStatusSeverity } type SlackEventWatermark = { known: boolean; lastEventAtMs?: number } @@ -169,6 +172,12 @@ const INJECTION_CONFIRMATION_TIMEOUT_MS = 90_000 const INJECTION_RETRY_DELAY_MS = 1_000 const INJECTION_RETRY_ATTEMPT_TIMEOUT_MS = 15_000 const INJECTION_MAX_ATTEMPTS = 6 +const CLARIFICATION_WAKE_LEASE_MS = 60_000 +const CLARIFICATION_WAKE_RETRY_MS = 1_000 +const CLARIFICATION_PARK_RETRY_MS = 5_000 +const CLARIFICATION_ESCALATION_LEASE_MS = 2 * 60_000 +const CLARIFICATION_ESCALATION_RETRY_MS = 5_000 +const CLARIFICATION_STALE_WARN_MS = 7 * 24 * 60 * 60_000 const STOP_TEARDOWN_TIMEOUT_MS = 2_500 const SLACK_EVENT_WATERMARK_CACHE_MS = 60_000 const MERGE_GATE_MAX_ATTEMPTS = 12 @@ -240,6 +249,12 @@ export class FactoryLoop implements Factory { readonly #labelDispatchFailures = new Map() readonly #pendingSlackClarifications = new Map() readonly #pendingGithubClarifications = new Map() + readonly #clarificationWakeInFlight = new Map>() + readonly #clarificationWakeRetryTimers = new Map>() + readonly #clarificationWakeOwner = `${process.pid}:${randomUUID()}` + #clarificationSweepTimer?: ReturnType + #clarificationSweepDueAtMs?: number + #clarificationSweepInFlight?: Promise readonly #postMergeDoneAdvances = new Set() #slackDegraded = false #slackDegradedReason: string | undefined @@ -447,6 +462,7 @@ export class FactoryLoop implements Factory { try { await this.#startLiveSubscription(opts.liveSubscription) await this.#rearmSlackReplyWatchers() + await this.#drainReadyClarificationWake() await this.#rearmGithubIssueCommentWatchers() this.#scheduleCompletionSweep(0) return @@ -479,6 +495,7 @@ export class FactoryLoop implements Factory { }) this.#started = true await this.#rearmSlackReplyWatchers() + await this.#drainReadyClarificationWake() await this.#rearmGithubIssueCommentWatchers() this.#scheduleCompletionSweep(0) } @@ -496,6 +513,11 @@ export class FactoryLoop implements Factory { this.#livePollInFlight = false this.#liveEventQueue.length = 0 this.#completionInFlight.clear() + for (const timer of this.#clarificationWakeRetryTimers.values()) clearTimeout(timer) + this.#clarificationWakeRetryTimers.clear() + if (this.#clarificationSweepTimer) clearTimeout(this.#clarificationSweepTimer) + this.#clarificationSweepTimer = undefined + this.#clarificationSweepDueAtMs = undefined this.#babysitterSpawned.clear() this.#babysitterPr.clear() const subscription = this.#subscription @@ -1343,9 +1365,13 @@ export class FactoryLoop implements Factory { // path never calls #start(), so without this a watcher only lives for the // process that originally dispatched — replies after a restart are dropped. await this.#rearmSlackReplyWatchers() + await this.#sweepWaitingClarifications() + await this.#drainReadyClarificationWake() for (let iteration = 0; iteration < maxIterations; iteration += 1) { await this.#writeLoopHeartbeat(heartbeatPath, registryPath, 'running', iteration, maxIterations) try { + await this.#sweepWaitingClarifications() + await this.#drainReadyClarificationWake() await this.#sweepPrStateCompletions('run-loop') reports.push(await this.runOnce({ dryRun: opts.dryRun })) consecutiveFailures = 0 @@ -2265,7 +2291,7 @@ export class FactoryLoop implements Factory { async #releaseAndTerminateAgents( agents: Array<[string, TrackedAgent]>, reason: string, - context: 'stop' | 'completion', + context: 'stop' | 'completion' | 'clarification', ): Promise { const protectedPids = await this.#protectedPids() for (const [agentName, tracked] of agents) { @@ -2882,6 +2908,7 @@ export class FactoryLoop implements Factory { } await this.#recordDispatchTerminal(record.issue) const next = (await this.#batch()).complete(record.issue) + await this.#drainReadyClarificationWake() await this.#stopSlackWatcher(record.issue) await this.#stopGithubIssueCommentWatcherForIssue(record.issue) await this.#writeInFlightRegistry() @@ -3009,18 +3036,25 @@ export class FactoryLoop implements Factory { }) } - await this.#postAgentQuestion(record, question) + const reserved = await this.#reserveHumanClarification(record, question) + if (reserved === false) return + const postedToSlack = await this.#postAgentQuestion(record, question) + if (postedToSlack && reserved) { + await this.#parkForHumanClarification(record, reserved) + } else if (reserved) { + await this.#state.clearWaitingClarification(this.#workspaceId, issueKey(record.issue)) + } } - async #postAgentQuestion(record: InFlightIssue, question: AgentQuestion): Promise { + async #postAgentQuestion(record: InFlightIssue, question: AgentQuestion): Promise { if (!this.#slack || !this.#config.slack) { await this.#postAgentQuestionToGithub(record, question) - return + return false } if (await this.#shouldSkipSlackWriteback('agent-question')) { this.#increment('agentQuestionsSkippedSlackDegraded') - return + return false } const key = issueKey(record.issue) @@ -3037,16 +3071,150 @@ export class FactoryLoop implements Factory { issue: record.issue, from: question.agentName, }) - return + return false } try { - await this.#slack.reply(threadId, agentQuestionSlackText(record.issue, question)) + await this.#slack.reply( + threadId, + agentQuestionSlackText(record.issue, question, this.#config.slack.stakeholderUserIds), + ) this.#increment('agentQuestionsPostedToSlack') this.#recordSlackWritebackSuccess('agent-question') + return true } catch (error) { this.#markSlackWritebackFailure('agent-question', error) this.#logger.warn?.(`[factory] failed to post agent question for ${record.issue.key}`, error) + return false + } + } + + async #reserveHumanClarification( + record: InFlightIssue, + question: AgentQuestion, + ): Promise { + if (!this.#slack || !this.#config.slack) { + return undefined + } + const key = issueKey(record.issue) + const threadId = await this.#state.getSlackThread(this.#workspaceId, key) + if (!threadId) { + this.#increment('agentQuestionReleaseSkippedMissingThread') + return undefined + } + + const agents = [...record.agents].map(([name, tracked]) => ({ + name, + tracked: structuredClone(tracked), + })) + if (agents.length === 0) { + this.#increment('agentQuestionReleaseSkippedNoAgents') + return undefined + } + + const waiting: WaitingClarification = { + issue: { ...record.issue }, + decision: structuredClone(record.decision), + dryRun: record.dryRun, + threadId, + askerName: question.agentName, + question: question.question, + askedAtMs: this.#clock.now(), + agents, + } + // Reserve before posting. A very fast human reply can now only become the + // durable wake trigger; it cannot be injected into a live agent and then + // lost while the team is parked moments later. + if (!await this.#state.reserveWaitingClarification(this.#workspaceId, key, waiting)) { + this.#increment('agentQuestionClarificationAlreadyReserved') + this.#logger.info?.('[factory] ignored a second agent question while clarification is already reserved', { + issue: record.issue.key, + asker: question.agentName, + }) + return false + } + this.#scheduleClarificationSweep(CLARIFICATION_STALE_WARN_MS) + return waiting + } + + async #parkForHumanClarification(record: InFlightIssue, waiting: WaitingClarification): Promise { + try { + await this.#finishClarificationPark(waiting, false) + } catch (error) { + // Keep the issue in the active batch until the fleet confirms that every + // team member is absent. The durable record remains release-pending and a + // maintenance sweep retries it without admitting replacement work early. + this.#increment('clarificationParkReleasePending') + this.#logger.warn?.('[factory] clarification park remains release-pending', { + issue: record.issue.key, + error, + }) + this.#scheduleClarificationSweep(CLARIFICATION_PARK_RETRY_MS) + } + } + + async #finishClarificationPark(waiting: WaitingClarification, recovered: boolean): Promise { + const key = issueKey(waiting.issue) + for (const { name } of waiting.agents) { + this.#fleet.markAgentTerminal?.(name, 'waiting-for-human') + } + await this.#releaseAgentsForClarification(key, waiting.agents.map(({ name, tracked }) => [name, tracked])) + + const batch = await this.#batch() + const next = batch.complete(waiting.issue) + // parkedAtMs is the durable wake gate. It is written only after roster + // confirmation proves every saved team member has relinquished its slot + // and after the old local batch record can no longer race the wake. + const parked = await this.#state.markClarificationParked(this.#workspaceId, key, this.#clock.now()) + if (!parked) { + if (next) batch.queue(next.decision, next.dryRun) + throw new Error(`durable clarification park refused for ${waiting.issue.key}`) + } + await this.#clearDispatchInFlight(waiting.issue) + await this.#writeInFlightRegistry() + this.#increment(recovered ? 'clarificationParksRecovered' : 'agentQuestionTeamsReleased') + this.#logger.info?.('[factory] released team while waiting for human clarification', { + issue: waiting.issue.key, + asker: waiting.askerName, + agents: waiting.agents.map(({ name }) => name), + recovered, + }) + this.#scheduleClarificationSweep(Math.max(1_000, CLARIFICATION_STALE_WARN_MS - (this.#clock.now() - waiting.askedAtMs))) + await this.#drainReadyClarificationWake() + if (next) await this.dispatch(next.decision, { dryRun: next.dryRun }) + } + + async #releaseAgentsForClarification(key: string, agents: Array<[string, TrackedAgent]>): Promise { + let waiting = await this.#state.getWaitingClarification(this.#workspaceId, key) + if (!waiting) return + for (const [name, tracked] of agents) { + const onlineBefore = new Set((await this.#fleet.roster()).agents.map((agent) => agent.name)) + if (waiting.releasedAgents?.includes(name) && !onlineBefore.has(name)) continue + try { + // Prefer broker release over process termination so the harness gets a + // graceful shutdown boundary and can flush its latest resumable state. + await this.#fleet.release(name, 'waiting-for-human') + } catch (error) { + this.#logger.warn?.('[factory] graceful clarification release failed; forcing local teardown', { + agentName: name, + error, + }) + await this.#releaseAndTerminateAgents([[name, tracked]], 'waiting-for-human', 'clarification') + } + const onlineAfter = new Set((await this.#fleet.roster()).agents.map((agent) => agent.name)) + if (onlineAfter.has(name)) { + throw new Error(`fleet still reports ${name} online after clarification release`) + } + waiting = await this.#state.markClarificationAgentReleased(this.#workspaceId, key, name) ?? waiting + } + + // Check the whole snapshot once more before opening the wake gate. This + // catches server-side restart policies that re-register a name between its + // individual release confirmation and the final parked transition. + const online = new Set((await this.#fleet.roster()).agents.map((agent) => agent.name)) + const stillOnline = agents.map(([name]) => name).filter((name) => online.has(name)) + if (stillOnline.length > 0) { + throw new Error(`clarification agents still online: ${stillOnline.join(', ')}`) } } @@ -4085,6 +4253,7 @@ export class FactoryLoop implements Factory { await this.#stopGithubIssueCommentWatcherForIssue(record.issue) await this.#recordDispatchTerminal(record.issue) const next = (await this.#batch()).complete(record.issue) + await this.#drainReadyClarificationWake() if (next) { await this.dispatch(next.decision, { dryRun: next.dryRun }) } @@ -4634,7 +4803,8 @@ export class FactoryLoop implements Factory { channelDir: string, preExistingPaths: readonly string[], ): Promise { - if (!isTriageEscalationWatchRecord(record)) { + const waiting = await this.#state.getWaitingClarification(this.#workspaceId, issueKey(record.issue)) + if (!isTriageEscalationWatchRecord(record) && !waiting) { return } @@ -4705,6 +4875,153 @@ export class FactoryLoop implements Factory { } await this.#rearmSlackWatcher(record, threadId) } + await this.#sweepWaitingClarifications() + for (const [, waiting] of await this.#state.listWaitingClarifications(this.#workspaceId)) { + const key = issueKey(waiting.issue) + // stop() clears ephemeral thread lookup state, but the durable + // clarification record owns the canonical thread while parked. Restore + // it so a resumed agent can ask a second question after a daemon restart. + await this.#state.setSlackThread(this.#workspaceId, key, waiting.threadId) + if (this.#slackWatchers.has(key) || this.#slackWatcherStarts.has(key)) { + continue + } + const record: InFlightIssue = { + issue: waiting.issue, + decision: waiting.decision, + dryRun: waiting.dryRun, + agents: new Map(), + invocationIds: new Set(), + } + await this.#rearmSlackWatcher(record, waiting.threadId) + } + } + + async #sweepWaitingClarifications(): Promise { + if (this.#clarificationSweepInFlight) { + await this.#clarificationSweepInFlight + return + } + const sweep = this.#performWaitingClarificationSweep() + .finally(() => { + if (this.#clarificationSweepInFlight === sweep) this.#clarificationSweepInFlight = undefined + }) + this.#clarificationSweepInFlight = sweep + await sweep + } + + async #performWaitingClarificationSweep(): Promise { + if (!this.#slack || !this.#config.slack || this.#stopping) return + if (this.#clarificationSweepTimer) clearTimeout(this.#clarificationSweepTimer) + this.#clarificationSweepTimer = undefined + this.#clarificationSweepDueAtMs = undefined + let nextDelayMs: number | undefined + + for (const [key, initial] of await this.#state.listWaitingClarifications(this.#workspaceId)) { + let waiting = initial + if (waiting.parkedAtMs === undefined) { + try { + await this.#finishClarificationPark(waiting, true) + waiting = await this.#state.getWaitingClarification(this.#workspaceId, key) ?? waiting + } catch (error) { + this.#increment('clarificationParkRetryFailures') + this.#logger.warn?.('[factory] could not finish release-pending clarification park', { + issue: waiting.issue.key, + error, + }) + nextDelayMs = Math.min(nextDelayMs ?? CLARIFICATION_PARK_RETRY_MS, CLARIFICATION_PARK_RETRY_MS) + continue + } + } + + if (waiting.reply || waiting.escalatedAtMs) continue + const waitingAgeMs = this.#clock.now() - waiting.askedAtMs + const untilEscalationMs = CLARIFICATION_STALE_WARN_MS - waitingAgeMs + if (untilEscalationMs > 0) { + nextDelayMs = Math.min(nextDelayMs ?? untilEscalationMs, untilEscalationMs) + continue + } + + const escalated = await this.#state.claimClarificationEscalation( + this.#workspaceId, + key, + this.#clarificationWakeOwner, + this.#clock.now(), + CLARIFICATION_ESCALATION_LEASE_MS, + ) + if (!escalated) { + // Another daemon may own the delivery attempt. Recheck so a crashed + // owner cannot strand the escalation after its durable lease expires. + nextDelayMs = Math.min( + nextDelayMs ?? CLARIFICATION_ESCALATION_RETRY_MS, + CLARIFICATION_ESCALATION_RETRY_MS, + ) + continue + } + this.#logger.warn?.('[factory] clarification remains parked without a human reply', { + issue: waiting.issue.key, + asker: waiting.askerName, + waitingAgeMs, + }) + try { + await this.#slack.reply( + escalated.threadId, + clarificationStaleSlackText(escalated, this.#config.slack.stakeholderUserIds), + ) + const completed = await this.#state.completeClarificationEscalation( + this.#workspaceId, + key, + this.#clarificationWakeOwner, + this.#clock.now(), + ) + if (!completed) { + this.#increment('clarificationEscalationOwnershipLost') + nextDelayMs = Math.min( + nextDelayMs ?? CLARIFICATION_ESCALATION_RETRY_MS, + CLARIFICATION_ESCALATION_RETRY_MS, + ) + continue + } + this.#increment('clarificationsParkedOverSevenDays') + this.#increment('clarificationEscalationsPosted') + } catch (error) { + await this.#state.releaseClarificationEscalation( + this.#workspaceId, + key, + this.#clarificationWakeOwner, + ) + this.#increment('clarificationEscalationFailures') + this.#logger.error?.('[factory] failed to post stale clarification escalation', { + issue: waiting.issue.key, + error, + }) + nextDelayMs = Math.min( + nextDelayMs ?? CLARIFICATION_ESCALATION_RETRY_MS, + CLARIFICATION_ESCALATION_RETRY_MS, + ) + } + } + + if (nextDelayMs !== undefined) this.#scheduleClarificationSweep(Math.max(1_000, nextDelayMs)) + } + + #scheduleClarificationSweep(delayMs: number): void { + if (this.#stopping) return + const dueAtMs = this.#clock.now() + Math.max(0, delayMs) + if (this.#clarificationSweepTimer && (this.#clarificationSweepDueAtMs ?? Number.MAX_SAFE_INTEGER) <= dueAtMs) return + if (this.#clarificationSweepTimer) clearTimeout(this.#clarificationSweepTimer) + const timer = setTimeout(() => { + this.#clarificationSweepTimer = undefined + this.#clarificationSweepDueAtMs = undefined + if (this.#stopping) return + void this.#sweepWaitingClarifications() + .catch((error) => { + this.#logger.warn?.('[factory] clarification maintenance sweep failed', error) + this.#scheduleClarificationSweep(CLARIFICATION_PARK_RETRY_MS) + }) + }, Math.max(0, delayMs)) + timer.unref?.() + this.#clarificationSweepTimer = timer + this.#clarificationSweepDueAtMs = dueAtMs } async #stopSlackWatcher(issue: IssueRef): Promise { @@ -4736,6 +5053,23 @@ export class FactoryLoop implements Factory { return } + const clarificationKey = issueKey(record.issue) + const waiting = await this.#state.getWaitingClarification(this.#workspaceId, clarificationKey) + if (waiting?.threadId === reply.threadTs) { + const claimed = await this.#state.claimClarificationReply(this.#workspaceId, clarificationKey, { + id: `${reply.threadTs}:${reply.messageTs}`, + text, + receivedAtMs: this.#clock.now(), + }) + if (!claimed) { + this.#increment('clarificationDuplicateWakesSuppressed') + return + } + this.#increment('clarificationRepliesClaimed') + await this.#wakeWaitingClarification(clarificationKey, claimed) + return + } + const liveRecord = (await this.#batch()).getIssue(record.issue) if (!liveRecord || liveRecord.dryRun) { if (isTriageEscalationWatchRecord(record)) { @@ -4767,6 +5101,302 @@ export class FactoryLoop implements Factory { } } + async #wakeWaitingClarification(key: string, waiting: WaitingClarification): Promise { + const existing = this.#clarificationWakeInFlight.get(key) + if (existing) { + await existing + return + } + const wake = this.#resumeWaitingClarification(key, waiting) + .finally(() => this.#clarificationWakeInFlight.delete(key)) + this.#clarificationWakeInFlight.set(key, wake) + await wake + } + + async #resumeWaitingClarification(key: string, waiting: WaitingClarification): Promise { + if (!waiting.reply) { + return + } + const claimed = await this.#state.claimClarificationWake( + this.#workspaceId, + key, + this.#clarificationWakeOwner, + this.#clock.now(), + CLARIFICATION_WAKE_LEASE_MS, + ) + if (!claimed) { + this.#increment('clarificationWakeClaimsSuppressed') + this.#scheduleClarificationWakeRetry(key) + return + } + waiting = claimed + const reply = waiting.reply + if (!reply) { + await this.#state.releaseClarificationWake(this.#workspaceId, key, this.#clarificationWakeOwner) + return + } + + let leaseLost = false + let renewalInFlight = false + const renewLease = async (): Promise => { + if (leaseLost) throw new ClarificationWakeLeaseLostError('clarification wake lease lost') + const renewed = await this.#state.renewClarificationWake( + this.#workspaceId, + key, + this.#clarificationWakeOwner, + this.#clock.now(), + ) + if (!renewed) { + leaseLost = true + throw new ClarificationWakeLeaseLostError('clarification wake lease lost') + } + } + const heartbeat = setInterval(() => { + if (renewalInFlight || leaseLost) return + renewalInFlight = true + void renewLease() + .catch(() => { leaseLost = true }) + .finally(() => { renewalInFlight = false }) + }, Math.max(1_000, Math.floor(CLARIFICATION_WAKE_LEASE_MS / 3))) + heartbeat.unref?.() + + try { + if (!await this.#clarificationIssueStillActive(waiting.issue)) { + await renewLease() + const completed = await this.#state.completeClarificationWake(this.#workspaceId, key, this.#clarificationWakeOwner) + if (!completed) { + this.#increment('clarificationWakeLeaseLosses') + return + } + await this.#stopSlackWatcher(waiting.issue) + this.#increment('clarificationWakesCancelledStaleIssue') + return + } + + const batch = await this.#batch() + if (!batch.canStart()) { + await this.#state.releaseClarificationWake(this.#workspaceId, key, this.#clarificationWakeOwner) + this.#increment('clarificationWakesQueuedForCapacity') + return + } + + const record = batch.start(waiting.decision, waiting.dryRun) + if (!record) { + await this.#state.releaseClarificationWake(this.#workspaceId, key, this.#clarificationWakeOwner) + this.#increment('clarificationWakesQueuedForCapacity') + return + } + + const resumed: Array<[string, TrackedAgent]> = [] + try { + await renewLease() + const onlineNames = new Set((await this.#fleet.roster()).agents.map((agent) => agent.name)) + for (const parked of waiting.agents) { + await renewLease() + const tracked = structuredClone(parked.tracked) + // A previous wake owner may have crashed after spawning but before + // clearing the durable record. Adopt an already-online deterministic + // name instead of duplicating the wake after the lease expires. + const result = onlineNames.has(parked.name) + ? { name: parked.name, sessionRef: tracked.sessionRef } + : await this.#resumeOrColdStartClarificationAgent(parked.name, tracked, waiting) + await renewLease() + const invocationId = batch.invocationIdFor(record.issue, tracked.spec) + batch.recordSpawn(record, tracked.spec, invocationId, result) + const live = record.agents.get(result.name) + if (live) resumed.push([result.name, live]) + } + + const event = slackReplyEvent(waiting.issue, reply.text) + for (const [name] of resumed) { + if (waiting.wake?.injectedAgents.includes(name)) continue + await renewLease() + if (this.#fleet.sendInput) { + await this.#fleet.sendInput(name, event) + } else { + await this.#fleet.sendMessage({ + to: name, + from: 'factory', + text: event.replace(/\r$/u, ''), + }) + } + const marked = await this.#state.markClarificationAgentInjected( + this.#workspaceId, + key, + this.#clarificationWakeOwner, + name, + ) + if (!marked) throw new ClarificationWakeLeaseLostError('clarification wake lease lost after injection') + this.#increment('clarificationReplyInjections') + } + + await renewLease() + await this.#writeInFlightRegistry() + await renewLease() + const completed = await this.#state.completeClarificationWake(this.#workspaceId, key, this.#clarificationWakeOwner) + if (!completed) throw new ClarificationWakeLeaseLostError('clarification wake completion lost ownership') + this.#increment('clarificationTeamsWoken') + this.#logger.info?.('[factory] woke team after human clarification', { + issue: waiting.issue.key, + agents: resumed.map(([name]) => name), + coldStarts: resumed.filter(([, tracked]) => !tracked.sessionRef).length, + }) + } catch (error) { + if (error instanceof ClarificationWakeLeaseLostError) { + batch.complete(waiting.issue) + await this.#writeInFlightRegistry() + this.#increment('clarificationWakeLeaseLosses') + this.#logger.warn?.('[factory] clarification wake ownership moved to another daemon', { + issue: waiting.issue.key, + }) + this.#scheduleClarificationWakeRetry(key) + return + } + for (const [name] of resumed) { + this.#fleet.markAgentTerminal?.(name, 'clarification-wake-failed') + } + await this.#releaseAndTerminateAgents(resumed, 'clarification-wake-failed', 'clarification') + batch.complete(waiting.issue) + await this.#state.releaseClarificationWake(this.#workspaceId, key, this.#clarificationWakeOwner) + await this.#writeInFlightRegistry() + this.#increment('clarificationWakeFailures') + this.#logger.error?.('[factory] failed to wake team after human clarification; wake remains durable for retry', { + issue: waiting.issue.key, + error, + }) + this.#scheduleClarificationWakeRetry(key) + } + } catch (error) { + if (error instanceof ClarificationWakeLeaseLostError) { + this.#increment('clarificationWakeLeaseLosses') + this.#logger.warn?.('[factory] clarification wake ownership moved to another daemon', { + issue: waiting.issue.key, + }) + this.#scheduleClarificationWakeRetry(key) + return + } + await this.#state.releaseClarificationWake(this.#workspaceId, key, this.#clarificationWakeOwner) + this.#increment('clarificationWakeFailures') + this.#logger.error?.('[factory] clarification wake preparation failed; wake remains durable for retry', { + issue: waiting.issue.key, + error, + }) + this.#scheduleClarificationWakeRetry(key) + } finally { + clearInterval(heartbeat) + } + } + + async #clarificationIssueStillActive(issueRef: IssueRef): Promise { + const issue = await this.#readIssue(issueRef.path) + if (!issue || !isInFactoryScope(issue, this.#config.safety) || !isDispatchableIssue(issue)) { + this.#logger.info?.('[factory] clarification wake cancelled because issue left factory scope', { + issue: issueRef.key, + exists: Boolean(issue), + inScope: issue ? isInFactoryScope(issue, this.#config.safety) : false, + dispatchable: issue ? isDispatchableIssue(issue) : false, + }) + return false + } + if (isGithubIssue(issue)) { + const state = issue.state?.name?.trim().toLowerCase() + const labels = new Set(issue.labels.map((label) => label.trim().toLowerCase())) + const required = this.#config.safety.requireLabel.trim().toLowerCase() + const active = state !== 'closed' && + Boolean(required) && + labels.has(required) && + labels.has('factory:in-progress') && + !labels.has('factory:human-review') + if (!active) this.#logger.info?.('[factory] clarification wake cancelled because GitHub issue is no longer active', { + issue: issueRef.key, + state, + labels: [...labels], + }) + return active + } + const role = this.#states.roleOf(issue.stateId) + if (role !== 'agentImplementing') this.#logger.info?.('[factory] clarification wake cancelled because Linear issue moved state', { + issue: issueRef.key, + stateId: issue.stateId, + stateName: issue.state?.name, + role, + }) + return role === 'agentImplementing' + } + + async #resumeOrColdStartClarificationAgent( + name: string, + tracked: TrackedAgent, + waiting: WaitingClarification, + ): Promise<{ name: string; sessionRef?: string }> { + if (tracked.sessionRef) { + try { + return await this.#fleet.resume({ + name, + sessionRef: tracked.sessionRef, + node: tracked.spec.node ?? 'self', + capability: tracked.spec.capability, + }) + } catch (error) { + this.#increment('clarificationResumeFallbacks') + this.#logger.warn?.('[factory] session resume failed; cold-starting from durable issue/question context', { + issue: waiting.issue.key, + agent: name, + sessionRef: tracked.sessionRef, + error: describeError(error).errorMessage, + }) + } + } else { + this.#increment('clarificationResumeFallbacks') + } + + const reply = waiting.reply?.text ?? '' + return await this.#fleet.spawn({ + name, + capability: tracked.spec.capability, + node: tracked.spec.node ?? 'self', + task: [ + tracked.spec.task, + '', + 'Factory released this team while waiting for human input and could not restore the prior harness session.', + `The blocked question was: ${waiting.question}`, + `The human replied: ${reply}`, + 'Re-hydrate from the issue, branch, worktree, and any open PR, then continue the task. Do not repeat completed work.', + ].join('\n'), + workflow: tracked.spec.workflow, + inputs: tracked.spec.inputs, + model: tracked.spec.model, + cwd: tracked.spec.clonePath, + repo: tracked.spec.repo, + restartPolicy: tracked.spec.restartPolicy ?? defaultRestartPolicy(tracked.spec), + channel: tracked.spec.channel, + }) + } + + async #drainReadyClarificationWake(): Promise { + const batch = await this.#batch() + if (!batch.canStart()) return + const ready = (await this.#state.listWaitingClarifications(this.#workspaceId)) + .filter(([, waiting]) => Boolean(waiting.reply)) + for (const [key, waiting] of ready) { + if (!batch.canStart()) break + await this.#wakeWaitingClarification(key, waiting) + } + } + + #scheduleClarificationWakeRetry(key: string): void { + if (this.#stopping || this.#clarificationWakeRetryTimers.has(key)) return + const timer = setTimeout(() => { + this.#clarificationWakeRetryTimers.delete(key) + if (this.#stopping) return + void this.#state.getWaitingClarification(this.#workspaceId, key) + .then((waiting) => waiting?.reply ? this.#wakeWaitingClarification(key, waiting) : undefined) + .catch((error) => this.#logger.warn?.('[factory] clarification wake retry failed', { key, error })) + }, CLARIFICATION_WAKE_RETRY_MS) + timer.unref?.() + this.#clarificationWakeRetryTimers.set(key, timer) + } + async #handleTriageEscalationSlackAnswer(record: InFlightIssue, text: string): Promise { const issue = await this.#readIssue(record.issue.path) if (!issue || !isInFactoryScope(issue, this.#config.safety) || !isDispatchableIssue(issue)) { @@ -6835,10 +7465,26 @@ const agentQuestionDedupeKey = (issue: IssueRef, question: AgentQuestion): strin question: question.question, }))}` -const agentQuestionSlackText = (issue: IssueRef, question: AgentQuestion): string => [ +const slackMentions = (userIds: string[]): string | undefined => { + const mentions = [...new Set(userIds.map((id) => id.trim()).filter(Boolean))] + .map((id) => `<@${id}>`) + return mentions.length > 0 ? mentions.join(' ') : undefined +} + +const agentQuestionSlackText = (issue: IssueRef, question: AgentQuestion, stakeholderUserIds: string[] = []): string => [ + slackMentions(stakeholderUserIds), `${issue.key}: ${question.agentName} needs input.`, `Question: ${question.question}`, -].join('\n') +].filter((line): line is string => Boolean(line)).join('\n') + +const clarificationStaleSlackText = ( + waiting: WaitingClarification, + stakeholderUserIds: string[] = [], +): string => [ + slackMentions(stakeholderUserIds), + `${waiting.issue.key} has been parked for seven days without a reply.`, + `Question from ${waiting.askerName}: ${waiting.question} Reply in this thread to wake the saved agent team, or move the issue out of Agent Implementing to cancel the wake.`, +].filter((line): line is string => Boolean(line)).join('\n') const escapeRegExp = (value: string): string => value.replace(/[.*+?^${}()|[\]\\]/gu, '\\$&') diff --git a/src/ports/state.ts b/src/ports/state.ts index b1db166..2ffd1aa 100644 --- a/src/ports/state.ts +++ b/src/ports/state.ts @@ -4,6 +4,43 @@ import type { IssueRef, TriageDecision } from '../types' export type CriticalRecord = { issue: IssueRef; input: SendInput } +export type ClarificationReply = { + id: string + text: string + receivedAtMs: number +} + +export type WaitingClarificationAgent = { + name: string + tracked: TrackedAgent +} + +export type WaitingClarification = { + issue: IssueRef + decision: TriageDecision + dryRun: boolean + threadId: string + askerName: string + question: string + askedAtMs: number + agents: WaitingClarificationAgent[] + releasedAgents?: string[] + parkedAtMs?: number + escalatedAtMs?: number + escalation?: { + owner: string + claimedAtMs: number + attempts: number + } + reply?: ClarificationReply + wake?: { + owner: string + claimedAtMs: number + attempts: number + injectedAgents: string[] + } +} + export type RegistryHandoffAgent = { issue: IssueRef name: string @@ -88,6 +125,22 @@ export interface StateStore { markAgentQuestion(workspaceId: string, key: string): Promise claimAgentQuestion(workspaceId: string, key: string): Promise + reserveWaitingClarification(workspaceId: string, issueKey: string, record: WaitingClarification): Promise + getWaitingClarification(workspaceId: string, issueKey: string): Promise + listWaitingClarifications(workspaceId: string): Promise> + claimClarificationReply(workspaceId: string, issueKey: string, reply: ClarificationReply): Promise + markClarificationAgentReleased(workspaceId: string, issueKey: string, agentName: string): Promise + markClarificationParked(workspaceId: string, issueKey: string, parkedAtMs: number): Promise + claimClarificationEscalation(workspaceId: string, issueKey: string, owner: string, nowMs: number, leaseMs: number): Promise + completeClarificationEscalation(workspaceId: string, issueKey: string, owner: string, escalatedAtMs: number): Promise + releaseClarificationEscalation(workspaceId: string, issueKey: string, owner: string): Promise + claimClarificationWake(workspaceId: string, issueKey: string, owner: string, nowMs: number, leaseMs: number): Promise + renewClarificationWake(workspaceId: string, issueKey: string, owner: string, nowMs: number): Promise + markClarificationAgentInjected(workspaceId: string, issueKey: string, owner: string, agentName: string): Promise + completeClarificationWake(workspaceId: string, issueKey: string, owner: string): Promise + releaseClarificationWake(workspaceId: string, issueKey: string, owner: string): Promise + clearWaitingClarification(workspaceId: string, issueKey: string): Promise + recordFailureHandoff(workspaceId: string, key: string, handoff: RegistryHandoffAgent): Promise getFailureHandoff(workspaceId: string, key: string): Promise listFailureHandoffs(workspaceId: string): Promise> diff --git a/src/state/file-state-store.test.ts b/src/state/file-state-store.test.ts index 5548b2f..35450c0 100644 --- a/src/state/file-state-store.test.ts +++ b/src/state/file-state-store.test.ts @@ -4,7 +4,7 @@ import { join } from 'node:path' import { setTimeout as delay } from 'node:timers/promises' import { describe, expect, it } from 'vitest' -import type { GithubIssueCommentWatchState } from '../ports/state' +import type { GithubIssueCommentWatchState, WaitingClarification } from '../ports/state' import { FileStateStore } from './file-state-store' describe('FileStateStore', () => { @@ -165,8 +165,106 @@ describe('FileStateStore', () => { await rm(root, { recursive: true, force: true }) } }) + + it('durably stores a parked team and atomically claims only the first human reply', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-file-state-clarification-')) + try { + const watchStatePath = join(root, 'factory-state.json') + const first = new FileStateStore({ batchSize: 2, watchStatePath }) + const waiting = waitingClarification(77) + expect(await first.reserveWaitingClarification('workspace-1', 'AR-77:uuid-77:path-77', waiting)).toBe(true) + expect(await first.reserveWaitingClarification('workspace-1', 'AR-77:uuid-77:path-77', { + ...waiting, + question: 'This duplicate must not overwrite the original question.', + })).toBe(false) + + const restarted = new FileStateStore({ batchSize: 2, watchStatePath }) + expect(await restarted.getWaitingClarification('workspace-1', 'AR-77:uuid-77:path-77')).toEqual(waiting) + + const [firstClaim, duplicateClaim] = await Promise.all([ + restarted.claimClarificationReply('workspace-1', 'AR-77:uuid-77:path-77', { + id: 'reply-1', text: 'Use the durable wake path.', receivedAtMs: 200, + }), + first.claimClarificationReply('workspace-1', 'AR-77:uuid-77:path-77', { + id: 'reply-2', text: 'Duplicate thread noise.', receivedAtMs: 201, + }), + ]) + expect([firstClaim, duplicateClaim].filter(Boolean)).toHaveLength(1) + const claimed = await restarted.getWaitingClarification('workspace-1', 'AR-77:uuid-77:path-77') + expect(claimed?.reply?.id).toMatch(/^reply-[12]$/u) + expect(await restarted.claimClarificationWake('workspace-1', 'AR-77:uuid-77:path-77', 'too-early', 250, 60_000)) + .toBeUndefined() + await restarted.markClarificationAgentReleased('workspace-1', 'AR-77:uuid-77:path-77', 'ar-77-impl') + expect((await first.getWaitingClarification('workspace-1', 'AR-77:uuid-77:path-77'))?.releasedAgents) + .toEqual(['ar-77-impl']) + expect(await restarted.markClarificationParked('workspace-1', 'AR-77:uuid-77:path-77', 274)).toBeUndefined() + await restarted.markClarificationAgentReleased('workspace-1', 'AR-77:uuid-77:path-77', 'ar-77-review') + await restarted.markClarificationParked('workspace-1', 'AR-77:uuid-77:path-77', 275) + + const [wakeA, wakeB] = await Promise.all([ + restarted.claimClarificationWake('workspace-1', 'AR-77:uuid-77:path-77', 'factory-a', 300, 60_000), + first.claimClarificationWake('workspace-1', 'AR-77:uuid-77:path-77', 'factory-b', 300, 60_000), + ]) + expect([wakeA, wakeB].filter(Boolean)).toHaveLength(1) + const wakeOwner = (wakeA ?? wakeB)?.wake?.owner + expect(wakeOwner).toMatch(/^factory-[ab]$/u) + expect(await restarted.renewClarificationWake('workspace-1', 'AR-77:uuid-77:path-77', 'wrong-owner', 301)).toBe(false) + expect(await restarted.renewClarificationWake('workspace-1', 'AR-77:uuid-77:path-77', wakeOwner!, 301)).toBe(true) + expect(await restarted.markClarificationAgentInjected('workspace-1', 'AR-77:uuid-77:path-77', wakeOwner!, 'ar-77-impl')).toBe(true) + expect(await restarted.completeClarificationWake('workspace-1', 'AR-77:uuid-77:path-77', 'wrong-owner')).toBe(false) + await restarted.releaseClarificationWake('workspace-1', 'AR-77:uuid-77:path-77', wakeOwner!) + expect((await first.getWaitingClarification('workspace-1', 'AR-77:uuid-77:path-77'))?.wake?.owner).toBe('') + const retry = await first.claimClarificationWake('workspace-1', 'AR-77:uuid-77:path-77', 'factory-retry', 302, 60_000) + expect(retry).toMatchObject({ wake: { owner: 'factory-retry', attempts: 2, injectedAgents: ['ar-77-impl'] } }) + + expect(await restarted.completeClarificationWake('workspace-1', 'AR-77:uuid-77:path-77', 'factory-retry')).toBe(true) + expect(await first.listWaitingClarifications('workspace-1')).toEqual([]) + } finally { + await rm(root, { recursive: true, force: true }) + } + }) }) +const waitingClarification = (number: number): WaitingClarification => { + const issue = { uuid: `uuid-${number}`, key: `AR-${number}`, path: `path-${number}` } + const implementer = { + name: `ar-${number}-impl`, + role: 'implementer' as const, + capability: 'spawn:codex' as const, + repo: 'AgentWorkforce/factory', + task: 'Implement the issue.', + } + const reviewer = { + name: `ar-${number}-review`, + role: 'reviewer' as const, + capability: 'spawn:codex' as const, + repo: 'AgentWorkforce/factory', + task: 'Review the implementation.', + } + return { + issue, + decision: { + issue, + routes: [{ repo: 'AgentWorkforce/factory', rationale: 'label' }], + scope: 'single', + implementers: [implementer], + reviewer, + thin: false, + confidence: 'high', + rationale: 'test', + }, + dryRun: false, + threadId: '1780000000.000077', + askerName: implementer.name, + question: 'Which wake path should I use?', + askedAtMs: 100, + agents: [ + { name: implementer.name, tracked: { spec: implementer, sessionRef: 'session-impl' } }, + { name: reviewer.name, tracked: { spec: reviewer, sessionRef: 'session-review' } }, + ], + } +} + const githubWatch = (number: number, claimedByCommentId?: string): GithubIssueCommentWatchState => ({ issue: { uuid: `uuid-${number}`, key: `AR-${number}`, path: `/linear/issues/AR-${number}__uuid-${number}.json` }, source: { diff --git a/src/state/file-state-store.ts b/src/state/file-state-store.ts index 6cbc6cf..57747cd 100644 --- a/src/state/file-state-store.ts +++ b/src/state/file-state-store.ts @@ -4,12 +4,17 @@ import { dirname, join } from 'node:path' import lockfile from 'proper-lockfile' -import type { GithubIssueCommentWatchState } from '../ports/state' +import type { ClarificationReply, GithubIssueCommentWatchState, WaitingClarification } from '../ports/state' import { InMemoryStateStore, type InMemoryStateStoreOptions } from './in-memory-state-store' +type PersistedWorkspaceState = { + githubIssueCommentWatches: Record + waitingClarifications: Record +} + type WatchStateDocument = { - version: 1 - workspaces: Record> + version: 2 + workspaces: Record } export type FileStateStoreOptions = InMemoryStateStoreOptions & { @@ -28,7 +33,8 @@ const WATCH_STATE_LOCK_STALE_MS = 60_000 /** * Keeps the factory's general runtime bookkeeping in memory while persisting - * GitHub escalation watches atomically so they survive a CLI process restart. + * GitHub escalation watches and parked clarification teams atomically so they + * survive a CLI process restart. * Mutations reload under an advisory lock so independent processes merge * updates instead of publishing divergent cached documents. */ @@ -49,8 +55,8 @@ export class FileStateStore extends InMemoryStateStore { await this.#exclusive(async () => { await this.#withMutationLock(async () => { const document = await this.#loadFromDisk() - const workspace = document.workspaces[workspaceId] ??= {} - workspace[key] = cloneWatch(watch) + const workspace = document.workspaces[workspaceId] ??= emptyWorkspaceState() + workspace.githubIssueCommentWatches[key] = cloneWatch(watch) await this.#persist(document) }) }) @@ -61,7 +67,7 @@ export class FileStateStore extends InMemoryStateStore { ): Promise> { return await this.#exclusive(async () => { const document = await this.#loadFromDisk() - return Object.entries(document.workspaces[workspaceId] ?? {}) + return Object.entries(document.workspaces[workspaceId]?.githubIssueCommentWatches ?? {}) .map(([key, watch]) => [key, cloneWatch(watch)]) }) } @@ -71,9 +77,273 @@ export class FileStateStore extends InMemoryStateStore { await this.#withMutationLock(async () => { const document = await this.#loadFromDisk() const workspace = document.workspaces[workspaceId] - if (!workspace || !(key in workspace)) return - delete workspace[key] - if (Object.keys(workspace).length === 0) { + if (!workspace || !(key in workspace.githubIssueCommentWatches)) return + delete workspace.githubIssueCommentWatches[key] + if (workspaceIsEmpty(workspace)) { + delete document.workspaces[workspaceId] + } + await this.#persist(document) + }) + }) + } + + override async reserveWaitingClarification( + workspaceId: string, + issueKey: string, + record: WaitingClarification, + ): Promise { + return await this.#exclusive(async () => { + return await this.#withMutationLock(async () => { + const document = await this.#loadFromDisk() + const workspace = document.workspaces[workspaceId] ??= emptyWorkspaceState() + if (workspace.waitingClarifications[issueKey]) return false + workspace.waitingClarifications[issueKey] = cloneClarification(record) + await this.#persist(document) + return true + }) + }) + } + + override async getWaitingClarification( + workspaceId: string, + issueKey: string, + ): Promise { + return await this.#exclusive(async () => { + const document = await this.#loadFromDisk() + const record = document.workspaces[workspaceId]?.waitingClarifications[issueKey] + return record ? cloneClarification(record) : undefined + }) + } + + override async listWaitingClarifications( + workspaceId: string, + ): Promise> { + return await this.#exclusive(async () => { + const document = await this.#loadFromDisk() + return Object.entries(document.workspaces[workspaceId]?.waitingClarifications ?? {}) + .map(([key, record]) => [key, cloneClarification(record)]) + }) + } + + override async claimClarificationReply( + workspaceId: string, + issueKey: string, + reply: ClarificationReply, + ): Promise { + return await this.#exclusive(async () => { + return await this.#withMutationLock(async () => { + const document = await this.#loadFromDisk() + const record = document.workspaces[workspaceId]?.waitingClarifications[issueKey] + if (!record || record.reply) { + return undefined + } + record.reply = { ...reply } + await this.#persist(document) + return cloneClarification(record) + }) + }) + } + + override async markClarificationAgentReleased( + workspaceId: string, + issueKey: string, + agentName: string, + ): Promise { + return await this.#mutateClarification(workspaceId, issueKey, (record) => { + record.releasedAgents ??= [] + if (!record.releasedAgents.includes(agentName)) record.releasedAgents.push(agentName) + }) + } + + override async claimClarificationWake( + workspaceId: string, + issueKey: string, + owner: string, + nowMs: number, + leaseMs: number, + ): Promise { + return await this.#exclusive(async () => { + return await this.#withMutationLock(async () => { + const document = await this.#loadFromDisk() + const record = document.workspaces[workspaceId]?.waitingClarifications[issueKey] + if (!record?.reply || record.parkedAtMs === undefined || (record.wake && record.wake.owner !== owner && nowMs - record.wake.claimedAtMs < leaseMs)) { + return undefined + } + record.wake = { + owner, + claimedAtMs: nowMs, + attempts: (record.wake?.attempts ?? 0) + 1, + injectedAgents: [...(record.wake?.injectedAgents ?? [])], + } + await this.#persist(document) + return cloneClarification(record) + }) + }) + } + + override async markClarificationParked( + workspaceId: string, + issueKey: string, + parkedAtMs: number, + ): Promise { + return await this.#exclusive(async () => { + return await this.#withMutationLock(async () => { + const document = await this.#loadFromDisk() + const record = document.workspaces[workspaceId]?.waitingClarifications[issueKey] + if (!record) return undefined + const released = new Set(record.releasedAgents ?? []) + if (record.agents.some(({ name }) => !released.has(name))) return undefined + record.parkedAtMs ??= parkedAtMs + await this.#persist(document) + return cloneClarification(record) + }) + }) + } + + override async claimClarificationEscalation( + workspaceId: string, + issueKey: string, + owner: string, + nowMs: number, + leaseMs: number, + ): Promise { + return await this.#exclusive(async () => { + return await this.#withMutationLock(async () => { + const document = await this.#loadFromDisk() + const record = document.workspaces[workspaceId]?.waitingClarifications[issueKey] + if (!record || record.reply || record.escalatedAtMs || ( + record.escalation && record.escalation.owner !== owner && nowMs - record.escalation.claimedAtMs < leaseMs + )) return undefined + record.escalation = { + owner, + claimedAtMs: nowMs, + attempts: (record.escalation?.attempts ?? 0) + 1, + } + await this.#persist(document) + return cloneClarification(record) + }) + }) + } + + override async completeClarificationEscalation( + workspaceId: string, + issueKey: string, + owner: string, + escalatedAtMs: number, + ): Promise { + return await this.#exclusive(async () => { + return await this.#withMutationLock(async () => { + const document = await this.#loadFromDisk() + const record = document.workspaces[workspaceId]?.waitingClarifications[issueKey] + if (record?.escalation?.owner !== owner) return false + record.escalatedAtMs = escalatedAtMs + delete record.escalation + await this.#persist(document) + return true + }) + }) + } + + override async releaseClarificationEscalation(workspaceId: string, issueKey: string, owner: string): Promise { + await this.#mutateClarification(workspaceId, issueKey, (record) => { + if (record.escalation?.owner !== owner) return + record.escalation.owner = '' + record.escalation.claimedAtMs = Number.MIN_SAFE_INTEGER + }) + } + + override async renewClarificationWake( + workspaceId: string, + issueKey: string, + owner: string, + nowMs: number, + ): Promise { + return await this.#mutateOwnedWake(workspaceId, issueKey, owner, (record) => { + record.wake!.claimedAtMs = nowMs + }) + } + + override async markClarificationAgentInjected( + workspaceId: string, + issueKey: string, + owner: string, + agentName: string, + ): Promise { + return await this.#mutateOwnedWake(workspaceId, issueKey, owner, (record) => { + if (!record.wake!.injectedAgents.includes(agentName)) record.wake!.injectedAgents.push(agentName) + }) + } + + override async completeClarificationWake(workspaceId: string, issueKey: string, owner: string): Promise { + return await this.#exclusive(async () => { + return await this.#withMutationLock(async () => { + const document = await this.#loadFromDisk() + const workspace = document.workspaces[workspaceId] + if (workspace?.waitingClarifications[issueKey]?.wake?.owner !== owner) return false + delete workspace.waitingClarifications[issueKey] + if (workspaceIsEmpty(workspace)) delete document.workspaces[workspaceId] + await this.#persist(document) + return true + }) + }) + } + + override async releaseClarificationWake(workspaceId: string, issueKey: string, owner: string): Promise { + await this.#exclusive(async () => { + await this.#withMutationLock(async () => { + const document = await this.#loadFromDisk() + const record = document.workspaces[workspaceId]?.waitingClarifications[issueKey] + if (record?.wake?.owner !== owner) return + record.wake.owner = '' + record.wake.claimedAtMs = Number.MIN_SAFE_INTEGER + await this.#persist(document) + }) + }) + } + + async #mutateOwnedWake( + workspaceId: string, + issueKey: string, + owner: string, + mutate: (record: WaitingClarification) => void, + ): Promise { + return await this.#exclusive(async () => { + return await this.#withMutationLock(async () => { + const document = await this.#loadFromDisk() + const record = document.workspaces[workspaceId]?.waitingClarifications[issueKey] + if (record?.wake?.owner !== owner) return false + mutate(record) + await this.#persist(document) + return true + }) + }) + } + + async #mutateClarification( + workspaceId: string, + issueKey: string, + mutate: (record: WaitingClarification) => void, + ): Promise { + return await this.#exclusive(async () => { + return await this.#withMutationLock(async () => { + const document = await this.#loadFromDisk() + const record = document.workspaces[workspaceId]?.waitingClarifications[issueKey] + if (!record) return undefined + mutate(record) + await this.#persist(document) + return cloneClarification(record) + }) + }) + } + + override async clearWaitingClarification(workspaceId: string, issueKey: string): Promise { + await this.#exclusive(async () => { + await this.#withMutationLock(async () => { + const document = await this.#loadFromDisk() + const workspace = document.workspaces[workspaceId] + if (!workspace || !(issueKey in workspace.waitingClarifications)) return + delete workspace.waitingClarifications[issueKey] + if (workspaceIsEmpty(workspace)) { delete document.workspaces[workspaceId] } await this.#persist(document) @@ -87,7 +357,7 @@ export class FileStateStore extends InMemoryStateStore { return parseDocument(parsed) } catch (error) { if (!isMissingFileError(error)) throw error - return { version: 1, workspaces: {} } + return { version: 2, workspaces: {} } } } @@ -138,15 +408,43 @@ export class FileStateStore extends InMemoryStateStore { } const parseDocument = (value: unknown): WatchStateDocument => { - if (!isRecord(value) || value.version !== 1 || !isRecord(value.workspaces)) { + if (!isRecord(value) || !isRecord(value.workspaces)) { throw new Error('Factory GitHub watch state file is invalid') } - return value as WatchStateDocument + if (value.version === 2) { + return value as WatchStateDocument + } + if (value.version === 1) { + const workspaces: Record = {} + for (const [workspaceId, watches] of Object.entries(value.workspaces)) { + if (!isRecord(watches)) { + throw new Error('Factory GitHub watch state file is invalid') + } + workspaces[workspaceId] = { + githubIssueCommentWatches: watches as Record, + waitingClarifications: {}, + } + } + return { version: 2, workspaces } + } + throw new Error('Factory GitHub watch state file is invalid') } const cloneWatch = (watch: GithubIssueCommentWatchState): GithubIssueCommentWatchState => structuredClone(watch) +const cloneClarification = (record: WaitingClarification): WaitingClarification => + structuredClone(record) + +const emptyWorkspaceState = (): PersistedWorkspaceState => ({ + githubIssueCommentWatches: {}, + waitingClarifications: {}, +}) + +const workspaceIsEmpty = (workspace: PersistedWorkspaceState): boolean => + Object.keys(workspace.githubIssueCommentWatches).length === 0 && + Object.keys(workspace.waitingClarifications).length === 0 + const syncParentDirectory = async (filePath: string): Promise => { const handle = await open(dirname(filePath), 'r') try { diff --git a/src/state/in-memory-state-store.ts b/src/state/in-memory-state-store.ts index a96642a..763682c 100644 --- a/src/state/in-memory-state-store.ts +++ b/src/state/in-memory-state-store.ts @@ -6,6 +6,8 @@ import type { GithubIssueCommentWatchState, RegistryHandoffAgent, StateStore, + WaitingClarification, + ClarificationReply, } from '../ports/state' type WorkspaceState = { @@ -16,6 +18,7 @@ type WorkspaceState = { githubIssueCommentWatches: Map seenAgentQuestionKeys: Set seenAgentQuestionOrder: string[] + waitingClarifications: Map dispatchAttempts: Map canonicalIssueStates: Map dispatchFailureReaperHandoffs: Map @@ -122,6 +125,149 @@ export class InMemoryStateStore implements StateStore { return true } + async reserveWaitingClarification(workspaceId: string, issueKey: string, record: WaitingClarification): Promise { + const waiting = this.#workspace(workspaceId).waitingClarifications + if (waiting.has(issueKey)) return false + waiting.set(issueKey, cloneWaitingClarification(record)) + return true + } + + async getWaitingClarification(workspaceId: string, issueKey: string): Promise { + const record = this.#workspace(workspaceId).waitingClarifications.get(issueKey) + return record ? cloneWaitingClarification(record) : undefined + } + + async listWaitingClarifications(workspaceId: string): Promise> { + return [...this.#workspace(workspaceId).waitingClarifications] + .map(([key, record]) => [key, cloneWaitingClarification(record)]) + } + + async claimClarificationReply( + workspaceId: string, + issueKey: string, + reply: ClarificationReply, + ): Promise { + const record = this.#workspace(workspaceId).waitingClarifications.get(issueKey) + if (!record || record.reply) { + return undefined + } + record.reply = { ...reply } + return cloneWaitingClarification(record) + } + + async markClarificationAgentReleased( + workspaceId: string, + issueKey: string, + agentName: string, + ): Promise { + const record = this.#workspace(workspaceId).waitingClarifications.get(issueKey) + if (!record) return undefined + record.releasedAgents ??= [] + if (!record.releasedAgents.includes(agentName)) record.releasedAgents.push(agentName) + return cloneWaitingClarification(record) + } + + async claimClarificationWake( + workspaceId: string, + issueKey: string, + owner: string, + nowMs: number, + leaseMs: number, + ): Promise { + const record = this.#workspace(workspaceId).waitingClarifications.get(issueKey) + if (!record?.reply || record.parkedAtMs === undefined || (record.wake && record.wake.owner !== owner && nowMs - record.wake.claimedAtMs < leaseMs)) { + return undefined + } + record.wake = { + owner, + claimedAtMs: nowMs, + attempts: (record.wake?.attempts ?? 0) + 1, + injectedAgents: [...(record.wake?.injectedAgents ?? [])], + } + return cloneWaitingClarification(record) + } + + async markClarificationParked(workspaceId: string, issueKey: string, parkedAtMs: number): Promise { + const record = this.#workspace(workspaceId).waitingClarifications.get(issueKey) + if (!record) return undefined + const released = new Set(record.releasedAgents ?? []) + if (record.agents.some(({ name }) => !released.has(name))) return undefined + record.parkedAtMs ??= parkedAtMs + return cloneWaitingClarification(record) + } + + async claimClarificationEscalation( + workspaceId: string, + issueKey: string, + owner: string, + nowMs: number, + leaseMs: number, + ): Promise { + const record = this.#workspace(workspaceId).waitingClarifications.get(issueKey) + if (!record || record.reply || record.escalatedAtMs || ( + record.escalation && record.escalation.owner !== owner && nowMs - record.escalation.claimedAtMs < leaseMs + )) return undefined + record.escalation = { + owner, + claimedAtMs: nowMs, + attempts: (record.escalation?.attempts ?? 0) + 1, + } + return cloneWaitingClarification(record) + } + + async completeClarificationEscalation( + workspaceId: string, + issueKey: string, + owner: string, + escalatedAtMs: number, + ): Promise { + const record = this.#workspace(workspaceId).waitingClarifications.get(issueKey) + if (record?.escalation?.owner !== owner) return false + record.escalatedAtMs = escalatedAtMs + delete record.escalation + return true + } + + async releaseClarificationEscalation(workspaceId: string, issueKey: string, owner: string): Promise { + const escalation = this.#workspace(workspaceId).waitingClarifications.get(issueKey)?.escalation + if (escalation?.owner !== owner) return + escalation.owner = '' + escalation.claimedAtMs = Number.MIN_SAFE_INTEGER + } + + async renewClarificationWake(workspaceId: string, issueKey: string, owner: string, nowMs: number): Promise { + const wake = this.#workspace(workspaceId).waitingClarifications.get(issueKey)?.wake + if (wake?.owner !== owner) return false + wake.claimedAtMs = nowMs + return true + } + + async markClarificationAgentInjected(workspaceId: string, issueKey: string, owner: string, agentName: string): Promise { + const wake = this.#workspace(workspaceId).waitingClarifications.get(issueKey)?.wake + if (wake?.owner !== owner) return false + if (!wake.injectedAgents.includes(agentName)) wake.injectedAgents.push(agentName) + return true + } + + async completeClarificationWake(workspaceId: string, issueKey: string, owner: string): Promise { + const state = this.#workspace(workspaceId) + if (state.waitingClarifications.get(issueKey)?.wake?.owner !== owner) return false + state.waitingClarifications.delete(issueKey) + return true + } + + async releaseClarificationWake(workspaceId: string, issueKey: string, owner: string): Promise { + const record = this.#workspace(workspaceId).waitingClarifications.get(issueKey) + if (record?.wake?.owner === owner) { + record.wake.owner = '' + record.wake.claimedAtMs = Number.MIN_SAFE_INTEGER + } + } + + async clearWaitingClarification(workspaceId: string, issueKey: string): Promise { + this.#workspace(workspaceId).waitingClarifications.delete(issueKey) + } + #rememberAgentQuestion(state: WorkspaceState, key: string): void { state.seenAgentQuestionKeys.add(key) state.seenAgentQuestionOrder.push(key) @@ -168,6 +314,7 @@ export class InMemoryStateStore implements StateStore { githubIssueCommentWatches: new Map(), seenAgentQuestionKeys: new Set(), seenAgentQuestionOrder: [], + waitingClarifications: new Map(), dispatchAttempts: new Map(), canonicalIssueStates: new Map(), dispatchFailureReaperHandoffs: new Map(), @@ -178,6 +325,9 @@ export class InMemoryStateStore implements StateStore { } } +const cloneWaitingClarification = (record: WaitingClarification): WaitingClarification => + structuredClone(record) + const cloneGithubIssueCommentWatch = (watch: GithubIssueCommentWatchState): GithubIssueCommentWatchState => ({ ...watch, issue: { ...watch.issue }, From 75db518f0829c6f5ea04b562d1161fb2dad5991d Mon Sep 17 00:00:00 2001 From: Miya Date: Thu, 16 Jul 2026 23:17:06 +0200 Subject: [PATCH 2/4] Harden clarification wake lease renewal --- src/orchestrator/factory.test.ts | 100 ++++++++++++++++++++++++++++++- src/orchestrator/factory.ts | 20 +++++-- 2 files changed, 112 insertions(+), 8 deletions(-) diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index bc21e13..954a828 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -262,6 +262,52 @@ class FailingSlackAnswerFleetClient extends FakeFleetClient { } } +class BlockingFirstClarificationResumeFleetClient extends FakeFleetClient { + readonly resumeStarted: Promise + #signalResumeStarted!: () => void + #releaseResume!: () => void + readonly #resumeReleased: Promise + #blocked = false + + constructor() { + super() + this.resumeStarted = new Promise((resolve) => { this.#signalResumeStarted = resolve }) + this.#resumeReleased = new Promise((resolve) => { this.#releaseResume = resolve }) + } + + releaseResume(): void { + this.#releaseResume() + } + + override async resume(input: Parameters[0]): Promise { + if (!this.#blocked) { + this.#blocked = true + this.#signalResumeStarted() + await this.#resumeReleased + } + return super.resume(input) + } +} + +class TransientClarificationRenewalStateStore extends InMemoryStateStore { + failNextRenewal = false + renewalFailures = 0 + + override async renewClarificationWake( + workspaceId: string, + issueKey: string, + owner: string, + nowMs: number, + ): Promise { + if (this.failNextRenewal) { + this.failNextRenewal = false + this.renewalFailures += 1 + throw new Error('transient state store outage') + } + return super.renewClarificationWake(workspaceId, issueKey, owner, nowMs) + } +} + class EscalatingTriage extends StaticTriage { readonly overrides: Partial> @@ -8843,6 +8889,56 @@ describe('FactoryLoop', () => { ]) }) + it('retries a transient background wake-lease renewal error without abandoning ownership', async () => { + const mount = new ConfirmRecordingSlackMountClient({ [issuePath(52)]: issueFile(52) }) + const fleet = new BlockingFirstClarificationResumeFleetClient() + fleet.setSessionRef('ar-52-impl-pear', 'session-ar-52-impl-pear') + fleet.setSessionRef('ar-52-review', 'session-ar-52-review') + const stateStore = new TransientClarificationRenewalStateStore({ batchSize: 2 }) + const warnings: unknown[][] = [] + const factory = createFactory(config({ slack: slackConfig() }), { + mount, + fleet, + stateStore, + triage: new StaticTriage(), + logger: { warn: (...args: unknown[]) => warnings.push(args) }, + }) + + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(52), issueFile(52)))) + mount.files.set(issuePath(52), { content: issueFile(52, implementing) }) + fleet.emitAgentMessage({ + from: 'ar-52-impl-pear', + target: 'factory', + body: '[factory-needs-input] Keep the wake lease through a transient outage.', + eventId: 'agent-question-52', + }) + await vi.waitFor(() => expect(factory.status().counters.agentQuestionTeamsReleased).toBe(1)) + + vi.useFakeTimers() + try { + emitSlackReply(mount, slackReplyFixturePath('C0FACTORY__factory-e2e', mount.threadTs, 'human-answer-52'), 'human-answer-52', { + text: 'Retry the lease heartbeat and continue.', + user: 'U123', + user_is_bot: false, + }) + await fleet.resumeStarted + + stateStore.failNextRenewal = true + await vi.advanceTimersByTimeAsync(20_000) + expect(stateStore.renewalFailures).toBe(1) + + fleet.releaseResume() + await vi.waitFor(() => expect(factory.status().counters.clarificationTeamsWoken).toBe(1)) + expect(factory.status().counters.clarificationWakeLeaseLosses).toBeUndefined() + expect(warnings).toContainEqual([ + '[factory] transient error renewing clarification wake lease; retrying', + expect.objectContaining({ issue: 'AR-52', error: expect.any(Error) }), + ]) + } finally { + vi.useRealTimers() + } + }) + it('cold-starts with durable issue, question, and reply context when session refs are unavailable', async () => { const mount = new ConfirmRecordingSlackMountClient({ [issuePath(37)]: issueFile(37) }) const fleet = new FakeFleetClient() @@ -8868,8 +8964,7 @@ describe('FactoryLoop', () => { user: 'U123', user_is_bot: false, }) - await flush() - await flush() + await vi.waitFor(() => expect(factory.status().counters.clarificationTeamsWoken).toBe(1)) expect(fleet.resumes).toEqual([]) expect(fleet.spawns).toHaveLength(4) @@ -8880,7 +8975,6 @@ describe('FactoryLoop', () => { expect(spawn.task).toContain('Re-hydrate from the issue, branch, worktree, and any open PR') } expect(factory.status().counters.clarificationResumeFallbacks).toBe(2) - expect(factory.status().counters.clarificationTeamsWoken).toBe(1) }) it('rearms a durable clarification after restart and wakes from a reply received while stopped', async () => { diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 95e63a0..b92f940 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -3187,9 +3187,9 @@ export class FactoryLoop implements Factory { async #releaseAgentsForClarification(key: string, agents: Array<[string, TrackedAgent]>): Promise { let waiting = await this.#state.getWaitingClarification(this.#workspaceId, key) if (!waiting) return + let online = new Set((await this.#fleet.roster()).agents.map((agent) => agent.name)) for (const [name, tracked] of agents) { - const onlineBefore = new Set((await this.#fleet.roster()).agents.map((agent) => agent.name)) - if (waiting.releasedAgents?.includes(name) && !onlineBefore.has(name)) continue + if (waiting.releasedAgents?.includes(name) && !online.has(name)) continue try { // Prefer broker release over process termination so the harness gets a // graceful shutdown boundary and can flush its latest resumable state. @@ -3205,14 +3205,15 @@ export class FactoryLoop implements Factory { if (onlineAfter.has(name)) { throw new Error(`fleet still reports ${name} online after clarification release`) } + online = onlineAfter waiting = await this.#state.markClarificationAgentReleased(this.#workspaceId, key, name) ?? waiting } // Check the whole snapshot once more before opening the wake gate. This // catches server-side restart policies that re-register a name between its // individual release confirmation and the final parked transition. - const online = new Set((await this.#fleet.roster()).agents.map((agent) => agent.name)) - const stillOnline = agents.map(([name]) => name).filter((name) => online.has(name)) + const finalOnline = new Set((await this.#fleet.roster()).agents.map((agent) => agent.name)) + const stillOnline = agents.map(([name]) => name).filter((name) => finalOnline.has(name)) if (stillOnline.length > 0) { throw new Error(`clarification agents still online: ${stillOnline.join(', ')}`) } @@ -5155,7 +5156,16 @@ export class FactoryLoop implements Factory { if (renewalInFlight || leaseLost) return renewalInFlight = true void renewLease() - .catch(() => { leaseLost = true }) + .catch((error: unknown) => { + if (error instanceof ClarificationWakeLeaseLostError) { + leaseLost = true + return + } + this.#logger.warn?.('[factory] transient error renewing clarification wake lease; retrying', { + issue: waiting.issue.key, + error, + }) + }) .finally(() => { renewalInFlight = false }) }, Math.max(1_000, Math.floor(CLARIFICATION_WAKE_LEASE_MS / 3))) heartbeat.unref?.() From 6844d1a88b7cb725c06dacff8f5bbc2ac6a80f5e Mon Sep 17 00:00:00 2001 From: Miya Date: Fri, 17 Jul 2026 00:02:33 +0200 Subject: [PATCH 3/4] Harden clarification delivery recovery --- src/dispatch/templates.ts | 2 +- src/orchestrator/factory.test.ts | 259 +++++++++++++++++++++++-- src/orchestrator/factory.ts | 292 ++++++++++++++++++++++++----- src/ports/state.ts | 9 + src/state/file-state-store.test.ts | 43 +++++ src/state/file-state-store.ts | 57 +++++- src/state/in-memory-state-store.ts | 44 ++++- 7 files changed, 646 insertions(+), 60 deletions(-) diff --git a/src/dispatch/templates.ts b/src/dispatch/templates.ts index 4cafae4..cb8e13d 100644 --- a/src/dispatch/templates.ts +++ b/src/dispatch/templates.ts @@ -101,7 +101,7 @@ export function renderAgentTask(input: RenderAgentTaskInput): string { 'If you are blocked or need a human answer mid-task, finish any safe reversible work first, then DM `factory` with `[factory-needs-input]`, the issue key, and one concrete question.', // Absolute path: the agent runs in its repo clone, not the daemon cwd // where .integrations lives, so a relative path would be unreachable. - `Factory will post the question to the Slack thread represented at ${input.slackDispatchThread.mountRoot}/slack/channels/${input.slackDispatchThread.channel}/messages/${input.slackDispatchThread.threadId.replaceAll('.', '_')}/replies/question.json.`, + `Factory will durably post and, if necessary, retry the question to the Slack thread represented at ${input.slackDispatchThread.mountRoot}/slack/channels/${input.slackDispatchThread.channel}/messages/${input.slackDispatchThread.threadId.replaceAll('.', '_')}/replies/question.json.`, 'After sending the marker, stop work and finish your session normally. Do not wait or poll: Factory will release the whole team and resume it from session memory only after the first human reply.', 'If session resume is unavailable, Factory will cold-start the team with the issue, question, reply, branch, and PR context so work can be re-hydrated explicitly.', ] diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 954a828..024e0ec 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -268,6 +268,7 @@ class BlockingFirstClarificationResumeFleetClient extends FakeFleetClient { #releaseResume!: () => void readonly #resumeReleased: Promise #blocked = false + disposeCalls = 0 constructor() { super() @@ -287,6 +288,10 @@ class BlockingFirstClarificationResumeFleetClient extends FakeFleetClient { } return super.resume(input) } + + override async dispose(): Promise { + this.disposeCalls += 1 + } } class TransientClarificationRenewalStateStore extends InMemoryStateStore { @@ -8502,6 +8507,130 @@ describe('FactoryLoop', () => { expect(fleet.resumes).toHaveLength(2) }) + it('parks an immediately exiting team and retries a failed durable question delivery after restart', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-question-delivery-retry-')) + try { + const watchStatePath = join(root, 'factory-state.json') + const mount = new FailNextSlackReplyMountClient({ [issuePath(54)]: issueFile(54) }) + const firstFleet = new FakeFleetClient() + const firstState = new FileStateStore({ batchSize: 2, watchStatePath }) + const factoryConfig = config({ slack: slackConfig() }) + const firstFactory = createFactory(factoryConfig, { + mount, + fleet: firstFleet, + stateStore: firstState, + triage: new StaticTriage(), + }) + + await firstFactory.dispatch(await firstFactory.triageIssue(parseLinearIssue(issuePath(54), issueFile(54)))) + mount.files.set(issuePath(54), { content: issueFile(54, implementing) }) + mount.failNextReply = true + firstFleet.emitAgentMessage({ + from: 'ar-54-impl-pear', + target: 'factory', + body: '[factory-needs-input] Preserve and retry this question.', + eventId: 'agent-question-54', + }) + // The prompt tells the asker to exit immediately. This callback is + // intentionally emitted in the same turn as the marker so it races the + // first durable state await. + firstFleet.emitAgentExit('ar-54-impl-pear', 'completed') + + await vi.waitFor(() => expect(firstFactory.status().counters.agentQuestionTeamsReleased).toBe(1)) + await vi.waitFor(() => expect(firstFactory.status().counters.clarificationQuestionDeliveryFailures).toBe(1)) + const pending = (await firstState.listWaitingClarifications('factory-test'))[0]?.[1] + expect(pending).toMatchObject({ + question: 'Preserve and retry this question.', + releasedAgents: ['ar-54-impl-pear', 'ar-54-review'], + questionDelivery: { owner: '', attempts: 1 }, + }) + expect(pending?.questionPostedAtMs).toBeUndefined() + expect(pending?.parkedAtMs).toBeTypeOf('number') + expect(firstFactory.status().counters.clarificationIntentExitsSuppressed).toBe(1) + expect(firstFactory.status().counters.clarificationQuestionDeliveryFailures).toBe(1) + expect(firstFleet.spawns).toHaveLength(2) + expect(firstFleet.releases).toEqual([ + { name: 'ar-54-impl-pear', reason: 'waiting-for-human' }, + { name: 'ar-54-review', reason: 'waiting-for-human' }, + ]) + expect(firstFactory.status().inFlight).toEqual([]) + await firstFactory.stop() + + const restartedState = new FileStateStore({ batchSize: 2, watchStatePath }) + const restartedFactory = createFactory(factoryConfig, { + mount, + fleet: new FakeFleetClient(), + stateStore: restartedState, + triage: new StaticTriage(), + }) + await restartedFactory.start({ mode: 'live', liveSubscription: { transport: 'subscribe' } }) + await vi.waitFor(() => expect(restartedFactory.status().counters.clarificationQuestionsDelivered).toBe(1)) + + const delivered = (await restartedState.listWaitingClarifications('factory-test'))[0]?.[1] + expect(delivered?.questionPostedAtMs).toBeTypeOf('number') + expect(delivered?.questionDelivery).toBeUndefined() + expect(slackReplyWrites(mount).filter((write) => + write.content.text.includes('Preserve and retry this question.'))).toHaveLength(2) + await restartedFactory.stop() + } finally { + await rm(root, { recursive: true, force: true }) + } + }) + + it('releases the whole team before waiting for slow Slack question confirmation', async () => { + const mount = new BlockingSlackQuestionMountClient({ [issuePath(57)]: issueFile(57) }) + const fleet = new FakeFleetClient() + fleet.setSessionRef('ar-57-impl-pear', 'session-ar-57-impl-pear') + fleet.setSessionRef('ar-57-review', 'session-ar-57-review') + const stateStore = new InMemoryStateStore({ batchSize: 2 }) + const factory = createFactory(config({ slack: slackConfig() }), { + mount, + fleet, + stateStore, + triage: new StaticTriage(), + }) + + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(57), issueFile(57)))) + mount.files.set(issuePath(57), { content: issueFile(57, implementing) }) + fleet.emitAgentMessage({ + from: 'ar-57-impl-pear', + target: 'factory', + body: '[factory-needs-input] Do not hold slots while Slack is slow.', + eventId: 'agent-question-57', + }) + await mount.questionWriteStarted + + expect(fleet.releases).toEqual([ + { name: 'ar-57-impl-pear', reason: 'waiting-for-human' }, + { name: 'ar-57-review', reason: 'waiting-for-human' }, + ]) + expect(factory.status().inFlight).toEqual([]) + const blocked = (await stateStore.listWaitingClarifications('factory-test'))[0]?.[1] + expect(blocked?.parkedAtMs).toBeTypeOf('number') + expect(blocked?.questionPostedAtMs).toBeUndefined() + expect(blocked?.questionDelivery?.owner).toBeTruthy() + + emitSlackReply(mount, slackReplyFixturePath('C0FACTORY__factory-e2e', mount.threadTs, 'human-answer-57'), 'human-answer-57', { + text: 'Persist this answer while confirmation is blocked.', + user: 'U123', + user_is_bot: false, + }) + await vi.waitFor(async () => { + const waiting = (await stateStore.listWaitingClarifications('factory-test'))[0]?.[1] + expect(waiting?.reply?.text).toBe('Persist this answer while confirmation is blocked.') + }) + expect(factory.status().counters.clarificationTeamsWoken).toBeUndefined() + expect(fleet.resumes).toEqual([]) + + mount.releaseQuestionWrite() + await vi.waitFor(() => expect(factory.status().counters.clarificationTeamsWoken).toBe(1)) + expect(factory.status().counters.clarificationQuestionsDelivered).toBe(1) + expect(fleet.resumes).toHaveLength(2) + expect(slackAnswerInputs(fleet)).toHaveLength(2) + expect(await stateStore.listWaitingClarifications('factory-test')).toEqual([]) + await factory.stop() + }) + it('tags configured Slack stakeholders when an agent needs input', async () => { const mount = new ConfirmRecordingSlackMountClient({ [issuePath(44)]: issueFile(44) }) const fleet = new FakeFleetClient() @@ -8528,7 +8657,7 @@ describe('FactoryLoop', () => { }) it('atomically keeps the first clarification when concurrent agent questions arrive', async () => { - const mount = new BlockingSlackQuestionMountClient({ [issuePath(49)]: issueFile(49) }) + const mount = new ConfirmRecordingSlackMountClient({ [issuePath(49)]: issueFile(49) }) const fleet = new FakeFleetClient() const stateStore = new InMemoryStateStore({ batchSize: 2 }) const factory = createFactory(config({ slack: slackConfig() }), { @@ -8545,16 +8674,14 @@ describe('FactoryLoop', () => { body: '[factory-needs-input] Preserve this original question.', eventId: 'agent-question-49-first', }) - await mount.questionWriteStarted - fleet.emitAgentMessage({ from: 'ar-49-review', target: 'factory', body: '[factory-needs-input] This concurrent question must not overwrite progress.', eventId: 'agent-question-49-second', }) + fleet.emitAgentExit('ar-49-review', 'completed') await vi.waitFor(() => expect(factory.status().counters.agentQuestionClarificationAlreadyReserved).toBe(1)) - mount.releaseQuestionWrite() await vi.waitFor(() => expect(factory.status().counters.agentQuestionTeamsReleased).toBe(1)) const saved = await stateStore.listWaitingClarifications('factory-test') @@ -8566,6 +8693,7 @@ describe('FactoryLoop', () => { }) expect(slackReplyWrites(mount)).toHaveLength(1) expect(fleet.releases).toHaveLength(2) + expect(factory.status().counters.clarificationIntentExitsSuppressed).toBe(1) }) it('retries a release-complete clarification when the durable parked transition is refused once', async () => { @@ -8777,6 +8905,7 @@ describe('FactoryLoop', () => { eventId: 'agent-question-51', }) await vi.waitFor(() => expect(firstFactory.status().counters.agentQuestionTeamsReleased).toBe(1)) + await vi.waitFor(() => expect(firstFactory.status().counters.clarificationQuestionsDelivered).toBe(1)) clock.advance(7 * 24 * 60 * 60_000) await firstFactory.stop() @@ -8939,6 +9068,51 @@ describe('FactoryLoop', () => { } }) + it('awaits and cancels an in-flight clarification wake before disposing the fleet', async () => { + const mount = new ConfirmRecordingSlackMountClient({ [issuePath(56)]: issueFile(56) }) + const fleet = new BlockingFirstClarificationResumeFleetClient() + fleet.setSessionRef('ar-56-impl-pear', 'session-ar-56-impl-pear') + fleet.setSessionRef('ar-56-review', 'session-ar-56-review') + const stateStore = new InMemoryStateStore({ batchSize: 2 }) + const factory = createFactory(config({ slack: slackConfig() }), { + mount, + fleet, + stateStore, + triage: new StaticTriage(), + }) + + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(56), issueFile(56)))) + mount.files.set(issuePath(56), { content: issueFile(56, implementing) }) + fleet.emitAgentMessage({ + from: 'ar-56-impl-pear', + target: 'factory', + body: '[factory-needs-input] Stop safely during my wake.', + eventId: 'agent-question-56', + }) + await vi.waitFor(() => expect(factory.status().counters.agentQuestionTeamsReleased).toBe(1)) + + emitSlackReply(mount, slackReplyFixturePath('C0FACTORY__factory-e2e', mount.threadTs, 'human-answer-56'), 'human-answer-56', { + text: 'This wake will overlap shutdown.', + user: 'U123', + user_is_bot: false, + }) + await fleet.resumeStarted + + let stopped = false + const stop = factory.stop().then(() => { stopped = true }) + await flush() + expect(stopped).toBe(false) + expect(fleet.disposeCalls).toBe(0) + + fleet.releaseResume() + await stop + expect(fleet.disposeCalls).toBe(1) + expect(slackAnswerInputs(fleet)).toEqual([]) + const durable = (await stateStore.listWaitingClarifications('factory-test'))[0]?.[1] + expect(durable?.reply?.text).toBe('This wake will overlap shutdown.') + expect(durable?.wake?.owner).toBe('') + }) + it('cold-starts with durable issue, question, and reply context when session refs are unavailable', async () => { const mount = new ConfirmRecordingSlackMountClient({ [issuePath(37)]: issueFile(37) }) const fleet = new FakeFleetClient() @@ -9001,6 +9175,7 @@ describe('FactoryLoop', () => { eventId: 'agent-question-38', }) await vi.waitFor(() => expect(firstFactory.status().counters.agentQuestionTeamsReleased).toBe(1)) + await vi.waitFor(() => expect(firstFactory.status().counters.clarificationQuestionsDelivered).toBe(1)) await firstFactory.stop() emitSlackReply(mount, slackReplyFixturePath('C0FACTORY__factory-e2e', mount.threadTs, 'human-answer-38'), 'human-answer-38', { @@ -9043,6 +9218,68 @@ describe('FactoryLoop', () => { } }) + it('startup drains a persisted clarification reply left before any wake lease was claimed', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-clarification-persisted-reply-')) + try { + const watchStatePath = join(root, 'factory-state.json') + const mount = new ConfirmRecordingSlackMountClient({ [issuePath(55)]: issueFile(55) }) + const firstFleet = new FakeFleetClient() + firstFleet.setSessionRef('ar-55-impl-pear', 'session-ar-55-impl-pear') + firstFleet.setSessionRef('ar-55-review', 'session-ar-55-review') + const firstState = new FileStateStore({ batchSize: 2, watchStatePath }) + const factoryConfig = config({ slack: slackConfig() }) + const firstFactory = createFactory(factoryConfig, { + mount, + fleet: firstFleet, + stateStore: firstState, + triage: new StaticTriage(), + }) + + await firstFactory.dispatch(await firstFactory.triageIssue(parseLinearIssue(issuePath(55), issueFile(55)))) + mount.files.set(issuePath(55), { content: issueFile(55, implementing) }) + firstFleet.emitAgentMessage({ + from: 'ar-55-impl-pear', + target: 'factory', + body: '[factory-needs-input] Prove startup drains my persisted answer.', + eventId: 'agent-question-55', + }) + await vi.waitFor(() => expect(firstFactory.status().counters.agentQuestionTeamsReleased).toBe(1)) + await vi.waitFor(() => expect(firstFactory.status().counters.clarificationQuestionsDelivered).toBe(1)) + await firstFactory.stop() + + const persisted = new FileStateStore({ batchSize: 2, watchStatePath }) + const [key] = (await persisted.listWaitingClarifications('factory-test'))[0]! + const claimed = await persisted.claimClarificationReply('factory-test', key, { + id: 'persisted-answer-55', + text: 'Resume this directly from startup drain.', + receivedAtMs: 500, + }) + expect(claimed?.reply?.id).toBe('persisted-answer-55') + expect(claimed?.wake).toBeUndefined() + + const restartedFleet = new FakeFleetClient() + const restartedState = new FileStateStore({ batchSize: 2, watchStatePath }) + const restartedFactory = createFactory(factoryConfig, { + mount, + fleet: restartedFleet, + stateStore: restartedState, + triage: new StaticTriage(), + }) + await restartedFactory.start({ mode: 'live', liveSubscription: { transport: 'subscribe' } }) + + await vi.waitFor(() => expect(restartedFactory.status().counters.clarificationTeamsWoken).toBe(1)) + expect(restartedFleet.resumes.map((resume) => resume.sessionRef)).toEqual([ + 'session-ar-55-impl-pear', + 'session-ar-55-review', + ]) + expect(slackAnswerInputs(restartedFleet)).toHaveLength(2) + expect(await restartedState.listWaitingClarifications('factory-test')).toEqual([]) + await restartedFactory.stop() + } finally { + await rm(root, { recursive: true, force: true }) + } + }) + it('rearms a persisted mid-task GitHub question and replays a reply received while stopped', async () => { const path = githubIssuePath('AgentWorkforce', 'pear', 63) const issue = githubIssueFile(63, { labels: ['factory'], author: 'reporter' }) @@ -9230,14 +9467,14 @@ describe('FactoryLoop', () => { await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(40), issueFile(40)))) fleet.emitAgentMessage(question) fleet.emitAgentMessage(question) - await flush() - await flush() - - expect(slackReplyWrites(mount).map((write) => write.content.text)).toEqual([ - 'AR-40: ar-40-impl-pear needs input.\nQuestion: Is this duplicate-safe?', - ]) - expect(factory.status().counters.agentQuestionsPostedToSlack).toBe(1) + await vi.waitFor(() => { + expect(slackReplyWrites(mount).map((write) => write.content.text)).toEqual([ + 'AR-40: ar-40-impl-pear needs input.\nQuestion: Is this duplicate-safe?', + ]) + expect(factory.status().counters.clarificationQuestionsDelivered).toBe(1) + }) expect(factory.status().counters.agentQuestionDuplicatesSuppressed).toBe(1) + await factory.stop() }) it('watches top-level inbound Slack thread replies keyed by real reply ts', async () => { diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index b92f940..cdc8170 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -121,6 +121,7 @@ type GithubIssueSource = { raw: Record } class ClarificationWakeLeaseLostError extends Error {} +class ClarificationWakeStoppedError extends Error {} type SlackSyncStatusSeverity = 'soft' | 'hard' type SlackSyncStatusCheck = { known: boolean; degraded: boolean; reason?: string; severity?: SlackSyncStatusSeverity } type SlackEventWatermark = { known: boolean; lastEventAtMs?: number } @@ -175,6 +176,8 @@ const INJECTION_MAX_ATTEMPTS = 6 const CLARIFICATION_WAKE_LEASE_MS = 60_000 const CLARIFICATION_WAKE_RETRY_MS = 1_000 const CLARIFICATION_PARK_RETRY_MS = 5_000 +const CLARIFICATION_QUESTION_DELIVERY_LEASE_MS = 2 * 60_000 +const CLARIFICATION_QUESTION_DELIVERY_RETRY_MS = 5_000 const CLARIFICATION_ESCALATION_LEASE_MS = 2 * 60_000 const CLARIFICATION_ESCALATION_RETRY_MS = 5_000 const CLARIFICATION_STALE_WARN_MS = 7 * 24 * 60 * 60_000 @@ -249,6 +252,8 @@ export class FactoryLoop implements Factory { readonly #labelDispatchFailures = new Map() readonly #pendingSlackClarifications = new Map() readonly #pendingGithubClarifications = new Map() + readonly #clarificationIntents = new Map() + readonly #clarificationQuestionDeliveryInFlight = new Map>() readonly #clarificationWakeInFlight = new Map>() readonly #clarificationWakeRetryTimers = new Map>() readonly #clarificationWakeOwner = `${process.pid}:${randomUUID()}` @@ -507,17 +512,25 @@ export class FactoryLoop implements Factory { this.#completionSweepTimer = undefined this.#stoppingHeartbeatRefreshActive = await this.#stopLiveHeartbeat('stopping') try { + // Fence every source of new clarification work before touching the fleet. + // A wake already past the fence is allowed to unwind, and is awaited + // without a timeout so it can never race fleet disposal. + for (const timer of this.#clarificationWakeRetryTimers.values()) clearTimeout(timer) + this.#clarificationWakeRetryTimers.clear() + if (this.#clarificationSweepTimer) clearTimeout(this.#clarificationSweepTimer) + this.#clarificationSweepTimer = undefined + this.#clarificationSweepDueAtMs = undefined + await this.#clarificationSweepInFlight + await this.#drainClarificationQuestionDeliveriesForStop() + await this.#drainClarificationWakesForStop() + this.#clarificationIntents.clear() + await this.#releaseInFlightAgents('factory-stopped') if (this.#livePollTimer) clearTimeout(this.#livePollTimer) this.#livePollTimer = undefined this.#livePollInFlight = false this.#liveEventQueue.length = 0 this.#completionInFlight.clear() - for (const timer of this.#clarificationWakeRetryTimers.values()) clearTimeout(timer) - this.#clarificationWakeRetryTimers.clear() - if (this.#clarificationSweepTimer) clearTimeout(this.#clarificationSweepTimer) - this.#clarificationSweepTimer = undefined - this.#clarificationSweepDueAtMs = undefined this.#babysitterSpawned.clear() this.#babysitterPr.clear() const subscription = this.#subscription @@ -543,6 +556,25 @@ export class FactoryLoop implements Factory { } } + async #drainClarificationWakesForStop(): Promise { + // A wake may add its promise just as the sweep that discovered it settles. + // Re-snapshot until the map is empty rather than assuming one await is a + // stable drain. + while (this.#clarificationWakeInFlight.size > 0) { + await Promise.allSettled([...this.#clarificationWakeInFlight.values()]) + } + } + + async #drainClarificationQuestionDeliveriesForStop(): Promise { + // Message handlers are fire-and-forget fleet callbacks. Track their Slack + // writes explicitly and re-snapshot until none remain, so shutdown cannot + // clear thread state (or let tests remove the state directory) underneath + // a late persistence step. + while (this.#clarificationQuestionDeliveryInFlight.size > 0) { + await Promise.allSettled([...this.#clarificationQuestionDeliveryInFlight.values()]) + } + } + async #boundedStopTeardown(label: string, teardown: () => Promise | void | undefined): Promise { let timer: ReturnType | undefined const action = Promise.resolve() @@ -2536,6 +2568,16 @@ export class FactoryLoop implements Factory { return } + // Agent messages and exits are separate fleet callbacks. A needs-input DM + // can therefore be followed by the instructed session exit before the + // first durable state await completes. The message handler installs this + // synchronous fence before yielding; the durable park path removes it only + // after the batch can no longer interpret that exit as ordinary completion. + if (this.#clarificationIntents.has(name)) { + this.#increment('clarificationIntentExitsSuppressed') + return + } + const batch = await this.#batch() const record = batch.getIssueByAgent(name) if (!record) { @@ -3001,48 +3043,75 @@ export class FactoryLoop implements Factory { if (!question || !isFactoryQuestionTarget(message.target)) { return } + if (this.#stopping) return - const record = (await this.#batch()).getIssueByAgent(question.agentName) - if (!record || record.dryRun) { - this.#increment('agentQuestionsIgnoredNoInFlight') - return - } + this.#clarificationIntents.set( + question.agentName, + (this.#clarificationIntents.get(question.agentName) ?? 0) + 1, + ) + let durableClarificationOwnsExit = false - if (question.issueKey && question.issueKey !== record.issue.key) { - this.#increment('agentQuestionsIgnoredIssueMismatch') - this.#logger.warn?.('[factory] ignored agent question for mismatched issue', { - from: question.agentName, - requestedIssue: question.issueKey, - activeIssue: record.issue.key, - }) - return - } + try { + const record = (await this.#batch()).getIssueByAgent(question.agentName) + if (this.#stopping) return + if (!record || record.dryRun) { + this.#increment('agentQuestionsIgnoredNoInFlight') + return + } - const dedupeKey = agentQuestionDedupeKey(record.issue, question) - if (!await this.#state.claimAgentQuestion(this.#workspaceId, dedupeKey)) { - this.#increment('agentQuestionDuplicatesSuppressed') - this.#logger.debug?.('[factory] suppressed duplicate agent question', { - from: question.agentName, - issue: record.issue.key, - }) - return - } + if (question.issueKey && question.issueKey !== record.issue.key) { + this.#increment('agentQuestionsIgnoredIssueMismatch') + this.#logger.warn?.('[factory] ignored agent question for mismatched issue', { + from: question.agentName, + requestedIssue: question.issueKey, + activeIssue: record.issue.key, + }) + return + } - if (!question.eventId) { - this.#increment('agentQuestionsMissingIdentity') - this.#logger.warn?.('[factory] agent question event missing stable identity; falling back to sender/content dedupe', { - from: question.agentName, - issue: record.issue.key, - }) - } + const dedupeKey = agentQuestionDedupeKey(record.issue, question) + if (!await this.#state.claimAgentQuestion(this.#workspaceId, dedupeKey)) { + this.#increment('agentQuestionDuplicatesSuppressed') + this.#logger.debug?.('[factory] suppressed duplicate agent question', { + from: question.agentName, + issue: record.issue.key, + }) + return + } - const reserved = await this.#reserveHumanClarification(record, question) - if (reserved === false) return - const postedToSlack = await this.#postAgentQuestion(record, question) - if (postedToSlack && reserved) { - await this.#parkForHumanClarification(record, reserved) - } else if (reserved) { - await this.#state.clearWaitingClarification(this.#workspaceId, issueKey(record.issue)) + if (!question.eventId) { + this.#increment('agentQuestionsMissingIdentity') + this.#logger.warn?.('[factory] agent question event missing stable identity; falling back to sender/content dedupe', { + from: question.agentName, + issue: record.issue.key, + }) + } + + if (this.#stopping) return + const reserved = await this.#reserveHumanClarification(record, question) + if (reserved === false) { + const existing = await this.#state.getWaitingClarification(this.#workspaceId, issueKey(record.issue)) + durableClarificationOwnsExit = Boolean( + existing?.agents.some(({ name }) => name === question.agentName), + ) + return + } + if (reserved) { + // The reservation, not Slack availability, owns the exit from here on. + // Always park immediately so a writeback outage cannot consume slots. + durableClarificationOwnsExit = true + await this.#parkForHumanClarification(record, reserved) + await this.#deliverClarificationQuestion(issueKey(record.issue), reserved) + } else { + if (this.#stopping) return + await this.#postAgentQuestion(record, question) + } + } finally { + if (!durableClarificationOwnsExit) { + const remaining = (this.#clarificationIntents.get(question.agentName) ?? 1) - 1 + if (remaining > 0) this.#clarificationIntents.set(question.agentName, remaining) + else this.#clarificationIntents.delete(question.agentName) + } } } @@ -3089,6 +3158,91 @@ export class FactoryLoop implements Factory { } } + async #deliverClarificationQuestion(key: string, waiting: WaitingClarification): Promise { + if (this.#stopping) return false + const existing = this.#clarificationQuestionDeliveryInFlight.get(key) + if (existing) return await existing + + const delivery = this.#performClarificationQuestionDelivery(key, waiting) + .finally(() => { + if (this.#clarificationQuestionDeliveryInFlight.get(key) === delivery) { + this.#clarificationQuestionDeliveryInFlight.delete(key) + } + }) + this.#clarificationQuestionDeliveryInFlight.set(key, delivery) + return await delivery + } + + async #performClarificationQuestionDelivery(key: string, waiting: WaitingClarification): Promise { + if (this.#stopping) return false + if (!this.#slack || !this.#config.slack || waiting.questionPostedAtMs !== undefined) { + return waiting.questionPostedAtMs !== undefined + } + + const claimed = await this.#state.claimClarificationQuestionDelivery( + this.#workspaceId, + key, + this.#clarificationWakeOwner, + this.#clock.now(), + CLARIFICATION_QUESTION_DELIVERY_LEASE_MS, + ) + if (!claimed) { + this.#increment('clarificationQuestionDeliveryClaimsSuppressed') + this.#scheduleClarificationSweep(CLARIFICATION_QUESTION_DELIVERY_RETRY_MS) + return false + } + + if (this.#stopping || await this.#shouldSkipSlackWriteback('agent-question')) { + await this.#state.releaseClarificationQuestionDelivery( + this.#workspaceId, + key, + this.#clarificationWakeOwner, + ) + if (!this.#stopping) this.#scheduleClarificationSweep(CLARIFICATION_QUESTION_DELIVERY_RETRY_MS) + return false + } + + try { + await this.#slack.reply( + claimed.threadId, + agentQuestionSlackText(claimed.issue, { + agentName: claimed.askerName, + question: claimed.question, + }, this.#config.slack.stakeholderUserIds), + ) + const completed = await this.#state.completeClarificationQuestionDelivery( + this.#workspaceId, + key, + this.#clarificationWakeOwner, + this.#clock.now(), + ) + if (!completed) { + this.#increment('clarificationQuestionDeliveryOwnershipLost') + this.#scheduleClarificationSweep(CLARIFICATION_QUESTION_DELIVERY_RETRY_MS) + return false + } + this.#increment('agentQuestionsPostedToSlack') + this.#increment('clarificationQuestionsDelivered') + this.#recordSlackWritebackSuccess('agent-question') + // A very fast human can reply while the Slack write is being confirmed. + // The reply is durable but wake-ineligible until questionPostedAtMs is + // committed above, so drain it immediately after opening that gate. + await this.#drainReadyClarificationWake() + return true + } catch (error) { + await this.#state.releaseClarificationQuestionDelivery( + this.#workspaceId, + key, + this.#clarificationWakeOwner, + ) + this.#markSlackWritebackFailure('agent-question', error) + this.#increment('clarificationQuestionDeliveryFailures') + this.#logger.warn?.(`[factory] failed to post agent question for ${claimed.issue.key}; keeping it durable for retry`, error) + this.#scheduleClarificationSweep(CLARIFICATION_QUESTION_DELIVERY_RETRY_MS) + return false + } + } + async #reserveHumanClarification( record: InFlightIssue, question: AgentQuestion, @@ -3138,6 +3292,7 @@ export class FactoryLoop implements Factory { } async #parkForHumanClarification(record: InFlightIssue, waiting: WaitingClarification): Promise { + if (this.#stopping) return try { await this.#finishClarificationPark(waiting, false) } catch (error) { @@ -3172,6 +3327,7 @@ export class FactoryLoop implements Factory { } await this.#clearDispatchInFlight(waiting.issue) await this.#writeInFlightRegistry() + for (const { name } of waiting.agents) this.#clarificationIntents.delete(name) this.#increment(recovered ? 'clarificationParksRecovered' : 'agentQuestionTeamsReleased') this.#logger.info?.('[factory] released team while waiting for human clarification', { issue: waiting.issue.key, @@ -4934,6 +5090,21 @@ export class FactoryLoop implements Factory { } } + if (waiting.questionPostedAtMs === undefined) { + await this.#deliverClarificationQuestion(key, waiting) + waiting = await this.#state.getWaitingClarification(this.#workspaceId, key) ?? waiting + if (waiting.questionPostedAtMs === undefined) { + nextDelayMs = Math.min( + nextDelayMs ?? CLARIFICATION_QUESTION_DELIVERY_RETRY_MS, + CLARIFICATION_QUESTION_DELIVERY_RETRY_MS, + ) + } + } + + // Do not accept arbitrary thread noise or escalate a question until its + // original Slack post is durably confirmed. Delivery retry is independent + // of parking so agents remain released throughout an outage. + if (waiting.questionPostedAtMs === undefined) continue if (waiting.reply || waiting.escalatedAtMs) continue const waitingAgeMs = this.#clock.now() - waiting.askedAtMs const untilEscalationMs = CLARIFICATION_STALE_WARN_MS - waitingAgeMs @@ -5115,7 +5286,7 @@ export class FactoryLoop implements Factory { } async #resumeWaitingClarification(key: string, waiting: WaitingClarification): Promise { - if (!waiting.reply) { + if (!waiting.reply || this.#stopping) { return } const claimed = await this.#state.claimClarificationWake( @@ -5131,6 +5302,10 @@ export class FactoryLoop implements Factory { return } waiting = claimed + if (this.#stopping) { + await this.#state.releaseClarificationWake(this.#workspaceId, key, this.#clarificationWakeOwner) + return + } const reply = waiting.reply if (!reply) { await this.#state.releaseClarificationWake(this.#workspaceId, key, this.#clarificationWakeOwner) @@ -5172,6 +5347,7 @@ export class FactoryLoop implements Factory { try { if (!await this.#clarificationIssueStillActive(waiting.issue)) { + this.#assertClarificationWakeRunning() await renewLease() const completed = await this.#state.completeClarificationWake(this.#workspaceId, key, this.#clarificationWakeOwner) if (!completed) { @@ -5183,7 +5359,10 @@ export class FactoryLoop implements Factory { return } + this.#assertClarificationWakeRunning() + const batch = await this.#batch() + this.#assertClarificationWakeRunning() if (!batch.canStart()) { await this.#state.releaseClarificationWake(this.#workspaceId, key, this.#clarificationWakeOwner) this.#increment('clarificationWakesQueuedForCapacity') @@ -5201,7 +5380,9 @@ export class FactoryLoop implements Factory { try { await renewLease() const onlineNames = new Set((await this.#fleet.roster()).agents.map((agent) => agent.name)) + this.#assertClarificationWakeRunning() for (const parked of waiting.agents) { + this.#assertClarificationWakeRunning() await renewLease() const tracked = structuredClone(parked.tracked) // A previous wake owner may have crashed after spawning but before @@ -5210,16 +5391,18 @@ export class FactoryLoop implements Factory { const result = onlineNames.has(parked.name) ? { name: parked.name, sessionRef: tracked.sessionRef } : await this.#resumeOrColdStartClarificationAgent(parked.name, tracked, waiting) - await renewLease() const invocationId = batch.invocationIdFor(record.issue, tracked.spec) batch.recordSpawn(record, tracked.spec, invocationId, result) const live = record.agents.get(result.name) if (live) resumed.push([result.name, live]) + this.#assertClarificationWakeRunning() + await renewLease() } const event = slackReplyEvent(waiting.issue, reply.text) for (const [name] of resumed) { if (waiting.wake?.injectedAgents.includes(name)) continue + this.#assertClarificationWakeRunning() await renewLease() if (this.#fleet.sendInput) { await this.#fleet.sendInput(name, event) @@ -5230,6 +5413,7 @@ export class FactoryLoop implements Factory { text: event.replace(/\r$/u, ''), }) } + this.#assertClarificationWakeRunning() const marked = await this.#state.markClarificationAgentInjected( this.#workspaceId, key, @@ -5252,6 +5436,16 @@ export class FactoryLoop implements Factory { coldStarts: resumed.filter(([, tracked]) => !tracked.sessionRef).length, }) } catch (error) { + if (error instanceof ClarificationWakeStoppedError) { + for (const [name] of resumed) { + this.#fleet.markAgentTerminal?.(name, 'factory-stopped') + } + await this.#releaseAndTerminateAgents(resumed, 'factory-stopped', 'clarification') + batch.complete(waiting.issue) + await this.#state.releaseClarificationWake(this.#workspaceId, key, this.#clarificationWakeOwner) + await this.#writeInFlightRegistry() + return + } if (error instanceof ClarificationWakeLeaseLostError) { batch.complete(waiting.issue) await this.#writeInFlightRegistry() @@ -5277,6 +5471,10 @@ export class FactoryLoop implements Factory { this.#scheduleClarificationWakeRetry(key) } } catch (error) { + if (error instanceof ClarificationWakeStoppedError) { + await this.#state.releaseClarificationWake(this.#workspaceId, key, this.#clarificationWakeOwner) + return + } if (error instanceof ClarificationWakeLeaseLostError) { this.#increment('clarificationWakeLeaseLosses') this.#logger.warn?.('[factory] clarification wake ownership moved to another daemon', { @@ -5297,6 +5495,10 @@ export class FactoryLoop implements Factory { } } + #assertClarificationWakeRunning(): void { + if (this.#stopping) throw new ClarificationWakeStoppedError('factory is stopping') + } + async #clarificationIssueStillActive(issueRef: IssueRef): Promise { const issue = await this.#readIssue(issueRef.path) if (!issue || !isInFactoryScope(issue, this.#config.safety) || !isDispatchableIssue(issue)) { @@ -5348,6 +5550,7 @@ export class FactoryLoop implements Factory { capability: tracked.spec.capability, }) } catch (error) { + this.#assertClarificationWakeRunning() this.#increment('clarificationResumeFallbacks') this.#logger.warn?.('[factory] session resume failed; cold-starting from durable issue/question context', { issue: waiting.issue.key, @@ -5360,6 +5563,7 @@ export class FactoryLoop implements Factory { this.#increment('clarificationResumeFallbacks') } + this.#assertClarificationWakeRunning() const reply = waiting.reply?.text ?? '' return await this.#fleet.spawn({ name, diff --git a/src/ports/state.ts b/src/ports/state.ts index 2ffd1aa..4d63ef1 100644 --- a/src/ports/state.ts +++ b/src/ports/state.ts @@ -24,6 +24,12 @@ export type WaitingClarification = { question: string askedAtMs: number agents: WaitingClarificationAgent[] + questionPostedAtMs?: number + questionDelivery?: { + owner: string + claimedAtMs: number + attempts: number + } releasedAgents?: string[] parkedAtMs?: number escalatedAtMs?: number @@ -128,6 +134,9 @@ export interface StateStore { reserveWaitingClarification(workspaceId: string, issueKey: string, record: WaitingClarification): Promise getWaitingClarification(workspaceId: string, issueKey: string): Promise listWaitingClarifications(workspaceId: string): Promise> + claimClarificationQuestionDelivery(workspaceId: string, issueKey: string, owner: string, nowMs: number, leaseMs: number): Promise + completeClarificationQuestionDelivery(workspaceId: string, issueKey: string, owner: string, postedAtMs: number): Promise + releaseClarificationQuestionDelivery(workspaceId: string, issueKey: string, owner: string): Promise claimClarificationReply(workspaceId: string, issueKey: string, reply: ClarificationReply): Promise markClarificationAgentReleased(workspaceId: string, issueKey: string, agentName: string): Promise markClarificationParked(workspaceId: string, issueKey: string, parkedAtMs: number): Promise diff --git a/src/state/file-state-store.test.ts b/src/state/file-state-store.test.ts index 35450c0..a7fe38a 100644 --- a/src/state/file-state-store.test.ts +++ b/src/state/file-state-store.test.ts @@ -223,6 +223,48 @@ describe('FileStateStore', () => { await rm(root, { recursive: true, force: true }) } }) + + it('leases delivery across daemons and persists fast replies without waking before confirmation', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-file-state-question-delivery-')) + try { + const watchStatePath = join(root, 'factory-state.json') + const first = new FileStateStore({ batchSize: 2, watchStatePath }) + const second = new FileStateStore({ batchSize: 2, watchStatePath }) + const waiting = { ...waitingClarification(78), questionPostedAtMs: undefined } + const key = 'AR-78:uuid-78:path-78' + await first.reserveWaitingClarification('workspace-1', key, waiting) + expect(await first.claimClarificationReply('workspace-1', key, { + id: 'thread-noise', text: 'Not an answer to a delivered question.', receivedAtMs: 199, + })).toBeUndefined() + await first.markClarificationAgentReleased('workspace-1', key, 'ar-78-impl') + await first.markClarificationAgentReleased('workspace-1', key, 'ar-78-review') + await first.markClarificationParked('workspace-1', key, 199) + + const [claimA, claimB] = await Promise.all([ + first.claimClarificationQuestionDelivery('workspace-1', key, 'factory-a', 200, 60_000), + second.claimClarificationQuestionDelivery('workspace-1', key, 'factory-b', 200, 60_000), + ]) + expect([claimA, claimB].filter(Boolean)).toHaveLength(1) + const owner = (claimA ?? claimB)?.questionDelivery?.owner + expect(owner).toMatch(/^factory-[ab]$/u) + expect(await first.claimClarificationQuestionDelivery('workspace-1', key, owner!, 201, 60_000)) + .toBeUndefined() + expect(await second.claimClarificationReply('workspace-1', key, { + id: 'fast-answer', text: 'Visible in Slack before confirmation.', receivedAtMs: 202, + })).toMatchObject({ reply: { id: 'fast-answer' } }) + expect(await first.claimClarificationWake('workspace-1', key, 'too-early', 202, 60_000)) + .toBeUndefined() + + await second.releaseClarificationQuestionDelivery('workspace-1', key, owner!) + const retry = await first.claimClarificationQuestionDelivery('workspace-1', key, 'factory-retry', 203, 60_000) + expect(retry?.questionDelivery).toMatchObject({ owner: 'factory-retry', attempts: 2 }) + expect(await second.completeClarificationQuestionDelivery('workspace-1', key, 'factory-retry', 204)).toBe(true) + expect(await first.claimClarificationWake('workspace-1', key, 'factory-wake', 205, 60_000)) + .toMatchObject({ reply: { id: 'fast-answer' }, wake: { owner: 'factory-wake' } }) + } finally { + await rm(root, { recursive: true, force: true }) + } + }) }) const waitingClarification = (number: number): WaitingClarification => { @@ -258,6 +300,7 @@ const waitingClarification = (number: number): WaitingClarification => { askerName: implementer.name, question: 'Which wake path should I use?', askedAtMs: 100, + questionPostedAtMs: 150, agents: [ { name: implementer.name, tracked: { spec: implementer, sessionRef: 'session-impl' } }, { name: reviewer.name, tracked: { spec: reviewer, sessionRef: 'session-review' } }, diff --git a/src/state/file-state-store.ts b/src/state/file-state-store.ts index 57747cd..a4d26e9 100644 --- a/src/state/file-state-store.ts +++ b/src/state/file-state-store.ts @@ -125,6 +125,59 @@ export class FileStateStore extends InMemoryStateStore { }) } + override async claimClarificationQuestionDelivery( + workspaceId: string, + issueKey: string, + owner: string, + nowMs: number, + leaseMs: number, + ): Promise { + return await this.#exclusive(async () => { + return await this.#withMutationLock(async () => { + const document = await this.#loadFromDisk() + const record = document.workspaces[workspaceId]?.waitingClarifications[issueKey] + if (!record || record.questionPostedAtMs !== undefined || ( + record.questionDelivery?.owner && + nowMs - record.questionDelivery.claimedAtMs < leaseMs + )) return undefined + record.questionDelivery = { + owner, + claimedAtMs: nowMs, + attempts: (record.questionDelivery?.attempts ?? 0) + 1, + } + await this.#persist(document) + return cloneClarification(record) + }) + }) + } + + override async completeClarificationQuestionDelivery( + workspaceId: string, + issueKey: string, + owner: string, + postedAtMs: number, + ): Promise { + return await this.#exclusive(async () => { + return await this.#withMutationLock(async () => { + const document = await this.#loadFromDisk() + const record = document.workspaces[workspaceId]?.waitingClarifications[issueKey] + if (record?.questionDelivery?.owner !== owner) return false + record.questionPostedAtMs = postedAtMs + delete record.questionDelivery + await this.#persist(document) + return true + }) + }) + } + + override async releaseClarificationQuestionDelivery(workspaceId: string, issueKey: string, owner: string): Promise { + await this.#mutateClarification(workspaceId, issueKey, (record) => { + if (record.questionDelivery?.owner !== owner) return + record.questionDelivery.owner = '' + record.questionDelivery.claimedAtMs = Number.MIN_SAFE_INTEGER + }) + } + override async claimClarificationReply( workspaceId: string, issueKey: string, @@ -134,7 +187,7 @@ export class FileStateStore extends InMemoryStateStore { return await this.#withMutationLock(async () => { const document = await this.#loadFromDisk() const record = document.workspaces[workspaceId]?.waitingClarifications[issueKey] - if (!record || record.reply) { + if (!record || (record.questionPostedAtMs === undefined && !record.questionDelivery?.owner) || record.reply) { return undefined } record.reply = { ...reply } @@ -166,7 +219,7 @@ export class FileStateStore extends InMemoryStateStore { return await this.#withMutationLock(async () => { const document = await this.#loadFromDisk() const record = document.workspaces[workspaceId]?.waitingClarifications[issueKey] - if (!record?.reply || record.parkedAtMs === undefined || (record.wake && record.wake.owner !== owner && nowMs - record.wake.claimedAtMs < leaseMs)) { + if (!record?.reply || record.questionPostedAtMs === undefined || record.parkedAtMs === undefined || (record.wake && record.wake.owner !== owner && nowMs - record.wake.claimedAtMs < leaseMs)) { return undefined } record.wake = { diff --git a/src/state/in-memory-state-store.ts b/src/state/in-memory-state-store.ts index 763682c..f7a5d8c 100644 --- a/src/state/in-memory-state-store.ts +++ b/src/state/in-memory-state-store.ts @@ -142,13 +142,53 @@ export class InMemoryStateStore implements StateStore { .map(([key, record]) => [key, cloneWaitingClarification(record)]) } + async claimClarificationQuestionDelivery( + workspaceId: string, + issueKey: string, + owner: string, + nowMs: number, + leaseMs: number, + ): Promise { + const record = this.#workspace(workspaceId).waitingClarifications.get(issueKey) + if (!record || record.questionPostedAtMs !== undefined || ( + record.questionDelivery?.owner && + nowMs - record.questionDelivery.claimedAtMs < leaseMs + )) return undefined + record.questionDelivery = { + owner, + claimedAtMs: nowMs, + attempts: (record.questionDelivery?.attempts ?? 0) + 1, + } + return cloneWaitingClarification(record) + } + + async completeClarificationQuestionDelivery( + workspaceId: string, + issueKey: string, + owner: string, + postedAtMs: number, + ): Promise { + const record = this.#workspace(workspaceId).waitingClarifications.get(issueKey) + if (record?.questionDelivery?.owner !== owner) return false + record.questionPostedAtMs = postedAtMs + delete record.questionDelivery + return true + } + + async releaseClarificationQuestionDelivery(workspaceId: string, issueKey: string, owner: string): Promise { + const delivery = this.#workspace(workspaceId).waitingClarifications.get(issueKey)?.questionDelivery + if (delivery?.owner !== owner) return + delivery.owner = '' + delivery.claimedAtMs = Number.MIN_SAFE_INTEGER + } + async claimClarificationReply( workspaceId: string, issueKey: string, reply: ClarificationReply, ): Promise { const record = this.#workspace(workspaceId).waitingClarifications.get(issueKey) - if (!record || record.reply) { + if (!record || (record.questionPostedAtMs === undefined && !record.questionDelivery?.owner) || record.reply) { return undefined } record.reply = { ...reply } @@ -175,7 +215,7 @@ export class InMemoryStateStore implements StateStore { leaseMs: number, ): Promise { const record = this.#workspace(workspaceId).waitingClarifications.get(issueKey) - if (!record?.reply || record.parkedAtMs === undefined || (record.wake && record.wake.owner !== owner && nowMs - record.wake.claimedAtMs < leaseMs)) { + if (!record?.reply || record.questionPostedAtMs === undefined || record.parkedAtMs === undefined || (record.wake && record.wake.owner !== owner && nowMs - record.wake.claimedAtMs < leaseMs)) { return undefined } record.wake = { From 76e5fe6f37d1e0f50c160159ec5dd2e390bd8b4a Mon Sep 17 00:00:00 2001 From: Miya Date: Fri, 17 Jul 2026 00:08:56 +0200 Subject: [PATCH 4/4] Stabilize async triage assertion --- src/orchestrator/factory.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 024e0ec..1a6565d 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -8149,7 +8149,7 @@ describe('FactoryLoop', () => { }) await vi.waitFor(() => expect(fleet.spawns.map((spawn) => spawn.name)).toEqual(['ar-65-impl-pear', 'ar-65-review'])) - expect(factory.status().counters.githubTriageAnswersDispatched).toBe(1) + await vi.waitFor(() => expect(factory.status().counters.githubTriageAnswersDispatched).toBe(1)) }) it('persists a pre-side-effect claim and suppresses the same reply after a crash restart', async () => {