From f620b6c9d14b6c3b81a02b4c171421194bce84ff Mon Sep 17 00:00:00 2001 From: Khaliq Date: Mon, 17 Aug 2026 14:56:54 +0200 Subject: [PATCH 1/6] fix: route durable Slack dispatch replies --- src/orchestrator/factory.test.ts | 301 +++++++++++++++++++++++++++-- src/orchestrator/factory.ts | 265 ++++++++++++++++++++++--- src/ports/state.ts | 30 ++- src/state/document-store.ts | 2 + src/state/file-state-store.test.ts | 104 ++++++++++ src/state/file-state-store.ts | 102 +++++++++- src/state/in-memory-state-store.ts | 70 ++++++- src/state/watch-state-document.ts | 40 +++- 8 files changed, 856 insertions(+), 58 deletions(-) diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 3da49a6..4e9eeca 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -15274,6 +15274,55 @@ describe('FactoryLoop', () => { expect(slack.roots).toEqual([]) }) + it('rearms a durable Slack triage escalation and replays a reply received while stopped', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-slack-triage-restart-')) + const watchStatePath = join(root, 'factory-state.json') + const mount = new CloudWritebackFakeMountClient({ [issuePath(131)]: issueFile(131) }) + const factoryConfig = config({ slack: slackConfig() }) + const state = () => new FileStateStore({ batchSize: 2, watchStatePath }) + const first = createFactory(factoryConfig, { + mount, + fleet: new FakeFleetClient(), + triage: new EscalatingTriage({ rationale: 'Matched repository from Linear label.' }), + stateStore: state(), + }) + let restarted: ReturnType | undefined + try { + await first.runOnce() + await first.stop() + + emitSlackReply(mount, slackReplyFixturePath( + 'C0FACTORY__factory-e2e', + mount.threadTs, + 'human-during-triage-restart', + ), 'slack-human-during-triage-restart', { + text: 'Use bounded retries and include the daemon restart acceptance case.', + user: 'U131', + user_name: 'human', + user_is_bot: false, + }) + + const restartedFleet = new FakeFleetClient() + restarted = createFactory(factoryConfig, { + mount, + fleet: restartedFleet, + triage: new StaticTriage(), + stateStore: state(), + }) + await restarted.start({ mode: 'dispatch-owner' }) + + await vi.waitFor(() => expect(restartedFleet.spawns.map((spawn) => spawn.name)) + .toEqual(['ar-131-impl-pear', 'ar-131-review']), { timeout: 4_000 }) + expect(restarted.status().counters.slackWatchersRearmed).toBe(1) + expect(restartedFleet.spawns.find((spawn) => spawn.name === 'ar-131-impl-pear')?.task) + .toContain('Human clarification from Slack:\nUse bounded retries and include the daemon restart acceptance case.') + } finally { + await first.stop() + await restarted?.stop() + await rm(root, { recursive: true, force: true }) + } + }) + it('posts low-confidence and thin GitHub triage escalation to the source issue when Slack is unconfigured', async () => { const path = githubIssuePath('AgentWorkforce', 'pear', 55) const mount = new CountingEventsMount({ [path]: githubIssueFile(55, { labels: ['factory'] }) }) @@ -15866,30 +15915,39 @@ describe('FactoryLoop', () => { expect(factory.status().counters.errors).toBeUndefined() }) - it('ignores a human Slack thread reply after the issue has no in-flight implementer', async () => { - const mount = new CloudWritebackFakeMountClient({ [issuePath(21)]: issueFile(21) }) + it('makes a late human Slack thread reply visibly unroutable after every agent exits', async () => { + const mount = new ConfirmRecordingSlackMountClient({ [issuePath(21)]: issueFile(21) }) const fleet = new FakeFleetClient() - const slack = new RecordingSlack() + const stateStore = new InMemoryStateStore({ batchSize: 10 }) const factory = createFactory(config({ slack: slackConfig() }), { mount, fleet, triage: new StaticTriage(), - slack, + stateStore, }) await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(21), issueFile(21)))) fleet.emitAgentExit('ar-21-impl-pear', 'issue-done') await vi.waitFor(() => expect(factory.status().inFlight).toEqual([])) - emitSlackReply(mount, slackReplyFixturePath('C0FACTORY__factory-e2e', slack.threadId, 'human-after-done'), 'slack-human-after-done', { + await vi.waitFor(async () => expect( + (await stateStore.listSlackThreadWatches('factory-test'))[0]?.[1], + ).toMatchObject({ kind: 'terminal-grace', threadId: mount.threadTs })) + await vi.waitFor(() => expect(factory.status().counters.slackTerminalWatchersRetained).toBe(1)) + emitSlackReply(mount, slackReplyFixturePath('C0FACTORY__factory-e2e', mount.threadTs, 'human-after-done'), 'slack-human-after-done', { text: 'please add one more test', user: 'U123', user_is_bot: false, }) - await flush() - await flush() + await vi.waitFor(() => expect(factory.status().counters.slackWebhookEventsObserved).toBe(1)) + await vi.waitFor(() => expect(factory.status().counters.slackAnswersIgnoredNoInFlight).toBe(1)) + await vi.waitFor(() => expect(factory.status().counters.slackAnswersUnroutableVisible).toBe(1)) + await vi.waitFor(() => expect(slackReplyWrites(mount).map((write) => write.content.text)).toContain( + 'Factory received this reply but could not route it because this work unit no longer has an active agent. Please continue on the linked issue or pull request.', + )) expect(factory.status().inFlight).toEqual([]) expect(slackAnswerInputs(fleet)).toEqual([]) + expect(factory.status().counters.slackAnswersUnroutableVisible).toBe(1) }) it('does not wire Slack answer injection when Slack is unconfigured', async () => { @@ -15953,8 +16011,11 @@ describe('FactoryLoop', () => { }) expect(factory.status().counters.slackConversationRepliesCoalesced).toBe(1) expect(slack.replies).toEqual([]) - expect(slackReplyWrites(mount)).toEqual([]) - expect(mount.confirmedPaths.filter((path) => path.includes('/replies/'))).toEqual([]) + expect(slackReplyWrites(mount).map((write) => write.content.text)).toEqual([ + slackImplementerReceipt, + slackImplementerReceipt, + ]) + expect(mount.confirmedPaths.filter((path) => path.includes('/replies/'))).toHaveLength(2) emitSlackReply(mount, slackReplyFixturePath('C0FACTORY__factory-e2e', slack.threadId, 'human-3'), 'slack-human-3', { text: 'What did you decide?', @@ -17825,7 +17886,7 @@ describe('FactoryLoop', () => { }) await expectSlackConversationResume(fleet, ['status?']) expect(slack.replies).toEqual([]) - expect(slackReplyWrites(mount)).toEqual([]) + expect(slackReplyWrites(mount).map((write) => write.content.text)).toEqual([slackImplementerReceipt]) }) it.each([ @@ -17911,7 +17972,7 @@ describe('FactoryLoop', () => { user_is_bot: false, }) await expectSlackConversationResume(fleet, ['status?']) - expect(slackReplyWrites(mount)).toEqual([]) + expect(slackReplyWrites(mount).map((write) => write.content.text)).toEqual([slackImplementerReceipt]) }) it('ignores the factory bot own Slack replies to avoid self-response loops', async () => { @@ -18003,7 +18064,7 @@ describe('FactoryLoop', () => { user_is_bot: false, }) await expectSlackConversationResume(fleet, ['new status?']) - expect(slackReplyWrites(mount)).toEqual([]) + expect(slackReplyWrites(mount).map((write) => write.content.text)).toEqual([slackImplementerReceipt]) }) it('re-arms the Slack reply watcher when a dispatch thread already persists (restart without a live watcher)', async () => { @@ -18217,7 +18278,7 @@ describe('FactoryLoop', () => { }) mount.emit(changeEvent(replyPath, 'slack-duplicate-human')) await expectSlackConversationResume(fleet, ['status?']) - expect(slackReplyWrites(mount)).toEqual([]) + expect(slackReplyWrites(mount).map((write) => write.content.text)).toEqual([slackImplementerReceipt]) }) it('retries a Slack reply after transient durable routing failure', async () => { @@ -18246,6 +18307,37 @@ describe('FactoryLoop', () => { expect(stateStore.failuresRemaining).toBe(0) }) + it('retries the visible Slack receipt after the durable reply was queued', async () => { + const mount = new FailNextSlackReplyMountClient({ [issuePath(44)]: issueFile(44) }) + const fleet = new FakeFleetClient() + fleet.setSessionRef('ar-44-impl-pear', 'session-ar-44-impl-pear') + const slack = new RecordingSlack() + const factory = createFactory(config({ slack: slackConfig() }), { + mount, + fleet, + triage: new StaticTriage(), + slack, + }) + + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(44), issueFile(44)))) + mount.failNextReply = true + emitSlackReply(mount, slackReplyFixturePath('C0FACTORY__factory-e2e', slack.threadId, 'human-ack-retry'), 'slack-human-ack-retry', { + text: 'Please retain this while the acknowledgement write retries.', + user: 'U123', + user_is_bot: false, + }) + + await expectSlackConversationResume(fleet, ['Please retain this while the acknowledgement write retries.']) + await vi.waitFor(() => expect(slackReplyWrites(mount).filter((write) => + write.content.text?.includes('received'), + ).length).toBeGreaterThanOrEqual(2), { timeout: 4_000 }) + expect(mount.failedReplies).toBe(1) + expect(slackReplyWrites(mount).at(-1)?.content).toMatchObject({ + thread_ts: slack.threadId, + text: expect.stringContaining('received'), + }) + }) + it('dedupes Slack conversation turns by human message ts across poll re-reads with fresh event ids', async () => { const mount = new CloudWritebackFakeMountClient({ [issuePath(42)]: issueFile(42) }) const fleet = new FakeFleetClient() @@ -18279,7 +18371,7 @@ describe('FactoryLoop', () => { mount.emit(changeEvent(path, 'slack-human-reread-1')) mount.emit(changeEvent(path, 'slack-human-reread-2')) await expectSlackConversationResume(fleet, ['status?']) - expect(slackReplyWrites(mount)).toEqual([]) + expect(slackReplyWrites(mount).map((write) => write.content.text)).toEqual([slackImplementerReceipt]) }) it('dispose unsubscribes Slack watchers and clears their polling timers', async () => { @@ -18378,7 +18470,10 @@ describe('FactoryLoop', () => { expect(factory.status().counters.slackConversationTurnResumeFailures).toBe(1) expect(factory.status().counters.slackConversationTurnsResumed).toBe(1) }) - expect(slackReplyWrites(mount)).toEqual([]) + expect(slackReplyWrites(mount).map((write) => write.content.text)).toEqual([ + slackImplementerReceipt, + slackImplementerReceipt, + ]) }) it('uses numeric Slack reply event ids without dropping fresh low-seq replies', async () => { @@ -18422,7 +18517,7 @@ describe('FactoryLoop', () => { }) mount.emit(changeEvent(replyPath, 1)) await expectSlackConversationResume(fleet, ['status?']) - expect(slackReplyWrites(mount)).toEqual([]) + expect(slackReplyWrites(mount).map((write) => write.content.text)).toEqual([slackImplementerReceipt]) expect(warnings.flat()).not.toContain('[factory] Slack reply event missing stable identity; falling back to path/content dedupe') }) }) @@ -19159,7 +19254,7 @@ describe('FactoryLoop PR babysitter', () => { await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(404), issue))) await vi.waitFor(async () => expect( - (await stateStore.getConversationSession('factory-test', `slack:${slack.threadId}`))?.agent.name, + (await stateStore.getConversationSession('factory-test', `slack:${slack.threadId}`))?.agent?.name, ).toBe('ar-404-impl-pear')) // The implementer hands off to a babysitter once its PR is ready; the @@ -19183,6 +19278,176 @@ describe('FactoryLoop PR babysitter', () => { name: 'ar-404-babysit', sessionRef: 'session-ar-404-babysit', }) + expect(slackReplyWrites(mount).map((write) => write.content.text)).toEqual([ + 'Factory received this reply and durably queued it for the PR babysitter.', + ]) + }) + + it('durably queues a long dispatch-thread reply until the babysitter becomes resumable', async () => { + const issue = realIssueFile(405, ready, { title: 'Real delayed babysitter Slack handoff' }) + const mount = new ConfirmRecordingSlackMountClient({ [issuePath(405)]: issue }) + const fleet = new FakeFleetClient() + const slack = new RecordingSlack() + const stateStore = new InMemoryStateStore({ batchSize: 10 }) + const factory = createFactory(babysitterConfig({ slack: slackConfig() }), { + mount, + fleet, + triage: new StaticTriage(), + slack, + stateStore, + probePrResolver: async () => ({ repo: 'AgentWorkforce/pear', prNumber: 405 }), + }) + const longReply = [ + 'There is unaddressed PR feedback and failing CI. Preserve this complete instruction beyond the Relay DM truncation boundary:', + 'fix the security finding, the correctness finding, and the bug, then rerun both failing checks.', + 'TAIL-MUST-REACH-THE-BABYSITTER', + ].join(' ') + + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(405), issue))) + emitSlackReply(mount, slackReplyFixturePath('C0FACTORY__factory-e2e', slack.threadId, 'human-before-owner'), 'slack-human-before-owner', { + text: longReply, + user: 'U123', + user_name: 'human', + user_is_bot: false, + }) + + await vi.waitFor(async () => expect( + (await stateStore.getConversationSession('factory-test', `slack:${slack.threadId}`))?.pending, + ).toEqual([expect.objectContaining({ text: longReply })])) + expect(slackConversationResumes(fleet)).toEqual([]) + expect(slackReplyWrites(mount)).toEqual([ + expect.objectContaining({ + content: expect.objectContaining({ + thread_ts: slack.threadId, + text: expect.stringMatching(/received.*stored.*agent/iu), + }), + }), + ]) + + fleet.setSessionRef('ar-405-babysit', 'session-ar-405-babysit') + fleet.emitAgentExit('ar-405-impl-pear', 'worker_exited') + + await vi.waitFor(() => expect(fleet.spawns.map((spawn) => spawn.name)).toContain('ar-405-babysit')) + await expectSlackConversationResume(fleet, [longReply, 'TAIL-MUST-REACH-THE-BABYSITTER']) + expect(slackConversationResumes(fleet)[0]).toMatchObject({ + name: 'ar-405-babysit', + sessionRef: 'session-ar-405-babysit', + }) + }) + + it('rearms a pre-existing dispatch thread onto its babysitter after a daemon restart', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-babysitter-slack-restart-')) + const watchStatePath = join(root, 'factory-state.json') + const issue = realIssueFile(406, ready, { title: 'Real babysitter Slack restart' }) + const mount = new ConfirmRecordingSlackMountClient({ [issuePath(406)]: issue }) + const factoryConfig = babysitterConfig({ slack: slackConfig() }) + const state = () => new FileStateStore({ batchSize: 10, watchStatePath }) + const firstFleet = new RemoteLifecycleFleetClient() + firstFleet.setSessionRef('ar-406-impl-pear', 'session-ar-406-impl-pear') + firstFleet.setSessionRef('ar-406-babysit', 'session-ar-406-babysit') + const first = createFactory(factoryConfig, { + mount, + fleet: firstFleet, + triage: new StaticTriage(), + stateStore: state(), + probePrResolver: async () => ({ repo: 'AgentWorkforce/pear', prNumber: 406 }), + }) + let restarted: ReturnType | undefined + try { + await first.dispatch(await first.triageIssue(parseLinearIssue(issuePath(406), issue))) + firstFleet.emitAgentExit('ar-406-impl-pear', 'worker_exited') + await vi.waitFor(async () => expect( + (await state().getConversationSession('factory-test', `slack:${mount.threadTs}`))?.agent, + ).toMatchObject({ name: 'ar-406-babysit', sessionRef: 'session-ar-406-babysit' })) + await first.stop() + + emitSlackReply(mount, slackReplyFixturePath( + 'C0FACTORY__factory-e2e', mount.threadTs, 'human-during-babysitter-restart', + ), 'slack-human-during-babysitter-restart', { + text: 'Recheck every unresolved review finding and both failing CI jobs.', + user: 'U406', + user_name: 'human', + user_is_bot: false, + }) + + const restartedFleet = new RemoteLifecycleFleetClient() + restarted = createFactory(factoryConfig, { + mount, + fleet: restartedFleet, + triage: new StaticTriage(), + stateStore: state(), + probePrResolver: async () => ({ repo: 'AgentWorkforce/pear', prNumber: 406 }), + }) + await restarted.start({ mode: 'dispatch-owner' }) + + await expectSlackConversationResume(restartedFleet, [ + 'Recheck every unresolved review finding and both failing CI jobs.', + ]) + expect(slackConversationResumes(restartedFleet)[0]).toMatchObject({ + name: 'ar-406-babysit', + sessionRef: 'session-ar-406-babysit', + }) + expect(restarted.status().counters.slackWatchersRearmed).toBe(1) + } finally { + await first.stop() + await restarted?.stop() + await rm(root, { recursive: true, force: true }) + } + }) + + it('rearms a terminal dispatch thread after restart so a late reply cannot disappear', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-terminal-slack-restart-')) + const watchStatePath = join(root, 'factory-state.json') + const issue = issueFile(407) + const mount = new ConfirmRecordingSlackMountClient({ [issuePath(407)]: issue }) + const factoryConfig = config({ slack: slackConfig() }) + const state = () => new FileStateStore({ batchSize: 10, watchStatePath }) + const firstFleet = new FakeFleetClient() + const first = createFactory(factoryConfig, { + mount, + fleet: firstFleet, + triage: new StaticTriage(), + stateStore: state(), + }) + let restarted: ReturnType | undefined + try { + await first.dispatch(await first.triageIssue(parseLinearIssue(issuePath(407), issue))) + firstFleet.emitAgentExit('ar-407-impl-pear', 'issue-done') + await vi.waitFor(() => expect(first.status().inFlight).toEqual([])) + await vi.waitFor(async () => expect( + (await state().listSlackThreadWatches('factory-test'))[0]?.[1], + ).toMatchObject({ kind: 'terminal-grace', threadId: mount.threadTs })) + await first.stop() + + emitSlackReply(mount, slackReplyFixturePath( + 'C0FACTORY__factory-e2e', mount.threadTs, 'human-after-terminal-restart', + ), 'slack-human-after-terminal-restart', { + text: 'There is still unaddressed review feedback.', + user: 'U407', + user_name: 'human', + user_is_bot: false, + }) + + const restartedFleet = new FakeFleetClient() + restarted = createFactory(factoryConfig, { + mount, + fleet: restartedFleet, + triage: new StaticTriage(), + stateStore: state(), + }) + await restarted.start({ mode: 'dispatch-owner' }) + + await vi.waitFor(() => expect(slackReplyWrites(mount).map((write) => write.content.text)).toContain( + 'Factory received this reply but could not route it because this work unit no longer has an active agent. Please continue on the linked issue or pull request.', + )) + expect(slackConversationResumes(restartedFleet)).toEqual([]) + expect(restarted.status().counters.slackAnswersUnroutableVisible).toBe(1) + expect(restarted.status().counters.slackWatchersRearmed).toBe(1) + } finally { + await first.stop() + await restarted?.stop() + await rm(root, { recursive: true, force: true }) + } }) it('does not attach a numeric GitHub issue to a merged PR whose body only contains a test count', async () => { @@ -22318,6 +22583,8 @@ const slackReplyWrites = (mount: FakeMountClient): Array<{ path: string; content .filter((write) => write.path.includes('/replies/')) .map((write) => ({ path: write.path, content: record(write.content) as { text?: string; thread_ts?: string } })) +const slackImplementerReceipt = 'Factory received this reply and durably queued it for the issue implementer.' + const slackAnswerInputs = (fleet: FakeFleetClient): Array<{ name: string; data: string }> => fleet.inputs.filter((input) => input.data !== '\r') diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 758a79b..c951ada 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -394,8 +394,10 @@ const STARTUP_AGENT_EXIT_DRAIN_TIMEOUT_MS = 30_000 const RECONCILED_AGENT_EXIT_CONCURRENCY = 4 const SLACK_EVENT_WATERMARK_CACHE_MS = 60_000 const SLACK_CONVERSATION_TURN_LEASE_MS = 60_000 +const SLACK_REPLY_ACKNOWLEDGEMENT_LEASE_MS = 60_000 const SLACK_CONVERSATION_TURN_RETRY_MS = 1_000 const SLACK_REPLY_ROUTE_RETRY_MS = 1_000 +const SLACK_TERMINAL_THREAD_GRACE_MS = 24 * 60 * 60_000 const MERGE_GATE_MAX_ATTEMPTS = 12 const MERGE_GATE_POLL_DELAY_MS = 10_000 const MAX_LABEL_IMPLEMENTERS = 4 @@ -516,6 +518,7 @@ export class FactoryLoop implements Factory { readonly #dispatchInFlight = new Map>() readonly #slackWatchers = new Map() readonly #slackWatcherStarts = new Map>() + readonly #slackTerminalWatchExpiryTimers = new Map>() readonly #slackConversationTurns: CoalescedTaskQueue readonly #slackConversationOwner = `${process.pid}:${randomUUID()}` readonly #githubIssueCommentWatchers = new Map() @@ -1140,6 +1143,8 @@ export class FactoryLoop implements Factory { await this.#boundedStopTeardown('factory subscription unsubscribe', () => subscription?.unsubscribe()) await Promise.all([...this.#slackWatchers.values()].map((watcher) => watcher.stop())) this.#slackWatchers.clear() + for (const timer of this.#slackTerminalWatchExpiryTimers.values()) clearTimeout(timer) + this.#slackTerminalWatchExpiryTimers.clear() await Promise.all([...this.#githubIssueCommentWatchers.values()].map((watcher) => watcher.stop())) this.#githubIssueCommentWatchers.clear() this.#githubIssueCommentWatchStates.clear() @@ -5521,7 +5526,7 @@ export class FactoryLoop implements Factory { for (const [name] of record.agents) { this.#fleet.markAgentTerminal?.(name, 'durable-dispatch-abandoned') } - await this.#stopSlackWatcher(record.issue) + await this.#retireSlackWatcher(record) await this.#stopGithubIssueCommentWatcherForIssue(record.issue) await this.#writeInFlightRegistry() this.#increment('dispatchLifecycleStaleIssuesAbandoned') @@ -8482,7 +8487,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.#retireSlackWatcher(record) await this.#stopGithubIssueCommentWatcherForIssue(record.issue) await this.#writeInFlightRegistry() if (next) { @@ -12765,7 +12770,7 @@ export class FactoryLoop implements Factory { batch.complete(record.issue) } if (!await this.#saveDispatchLifecycle(record, 'releasing', undefined, releaseReason)) return - await this.#stopSlackWatcher(record.issue) + await this.#retireSlackWatcher(record) await this.#stopGithubIssueCommentWatcherForIssue(record.issue) await this.#recordDispatchTerminal(record.issue) await this.#finishDurableRelease(record, releaseReason) @@ -13289,6 +13294,14 @@ export class FactoryLoop implements Factory { } const key = issueKey(record.issue) + const previousWatch = (await this.#state.listSlackThreadWatches(this.#workspaceId)) + .find(([watchKey]) => watchKey === key)?.[1] + if (previousWatch?.kind === 'terminal-grace') { + // A reopened work unit needs a fresh dispatch notification and a fresh + // conversation. Do not let the old grace-period watcher (or its expiry + // timer) capture and later tear down the new dispatch. + await this.#stopSlackWatcher(record.issue) + } const existingThread = await this.#persistedSlackThread(key) const watcherStart = this.#slackWatcherStarts.get(key) if (existingThread || watcherStart) { @@ -13390,13 +13403,14 @@ export class FactoryLoop implements Factory { if (existing) { const sessionRef = owned?.tracked.sessionRef const agentName = owned ? (owned.tracked.result?.name ?? owned.name) : undefined + let rebound = false if ( owned && sessionRef && - (agentName !== existing.agent.name || ( + (!existing.agent || agentName !== existing.agent.name || ( options.forceAgentRebind === true && sessionRef !== existing.agent.sessionRef )) ) { - const rebound = await this.#state.rebindConversationSession(this.#workspaceId, conversationId, { + rebound = await this.#state.rebindConversationSession(this.#workspaceId, conversationId, { name: agentName!, sessionRef, role: owned.tracked.spec.role, @@ -13412,26 +13426,21 @@ export class FactoryLoop implements Factory { this.#workspaceId, issueKey(existing.issue), ) - if (!waiting) this.#slackConversationTurns.schedule(conversationId) + if (!waiting && (existing.agent || rebound)) this.#slackConversationTurns.schedule(conversationId) } return } const sessionRef = owned?.tracked.sessionRef - if (!owned || !sessionRef) { - this.#increment('slackConversationSessionsSkippedMissingSession') - return - } - const channelDir = await this.#slackChannelDir() ?? this.#config.slack?.channel if (!channelDir) return - const agentName = owned.tracked.result?.name ?? owned.name + const agentName = owned ? (owned.tracked.result?.name ?? owned.name) : undefined const reserved = await this.#state.reserveConversationSession(this.#workspaceId, conversationId, { provider: 'slack', issue: { ...record.issue }, externalId: threadId, context: { channelDir }, - agent: { + ...(owned && sessionRef && agentName ? { agent: { name: agentName, sessionRef, role: owned.tracked.spec.role, @@ -13439,12 +13448,16 @@ export class FactoryLoop implements Factory { capability: owned.tracked.spec.capability, repo: owned.tracked.spec.repo, clonePath: owned.tracked.spec.clonePath, - }, + } } : {}), history: [], processedMessageIds: [], + acknowledgedMessageIds: [], + acknowledgementClaims: {}, pending: [], }) - if (reserved) this.#increment('slackConversationSessionsOwned') + if (reserved) { + this.#increment(owned && sessionRef ? 'slackConversationSessionsOwned' : 'slackConversationSessionsReservedUnowned') + } } // Called right after a babysitter is spawned/reattached for an issue's PR so @@ -13479,11 +13492,20 @@ export class FactoryLoop implements Factory { ) if (!claimed?.delivery) { const current = await this.#state.getConversationSession(this.#workspaceId, conversationId) - if (current && (current.pending.length > 0 || current.delivery)) { + if (current?.agent && (current.pending.length > 0 || current.delivery)) { this.#slackConversationTurns.schedule(conversationId, SLACK_CONVERSATION_TURN_RETRY_MS) } return } + if (!claimed.agent) { + await this.#state.releaseConversationTurn( + this.#workspaceId, + conversationId, + this.#slackConversationOwner, + claimId, + ) + return + } if (!await this.#ownsActiveSlackConversationIssue(claimed.issue)) { await this.#state.releaseConversationTurn( @@ -13593,10 +13615,12 @@ export class FactoryLoop implements Factory { session: ConversationSessionState, result: SpawnResult, ): Promise { + const sessionAgent = session.agent + if (!sessionAgent) return const record = (await this.#batch()).getIssue(session.issue) if (!record) return const entry = [...record.agents.entries()].find(([name, tracked]) => - name === session.agent.name || tracked.result?.name === session.agent.name) + name === sessionAgent.name || tracked.result?.name === sessionAgent.name) if (!entry) return const [previousName, tracked] = entry tracked.result = { @@ -13939,7 +13963,14 @@ export class FactoryLoop implements Factory { `Question: ${triageEscalationQuestion(decision, issue)}`, ].join('\n'), }) - await this.#state.setSlackThread(this.#workspaceId, issueKey(decision.issue), root.threadId) + const key = issueKey(decision.issue) + await this.#state.setSlackThread(this.#workspaceId, key, root.threadId) + await this.#state.setSlackThreadWatch(this.#workspaceId, key, { + kind: 'triage', + issue: { ...decision.issue }, + decision: structuredClone(decision), + threadId: root.threadId, + }) const replayedResult = await this.#watchSlackThread(escalationWatchRecord(decision), root.threadId) this.#recordSlackWritebackSuccess('triage-escalation') return replayedResult @@ -14214,6 +14245,24 @@ export class FactoryLoop implements Factory { this.#slackConversationTurns.schedule(conversationId) } } + for (const [key, watch] of await this.#state.listSlackThreadWatches(this.#workspaceId)) { + if (this.#slackWatchers.has(key) || this.#slackWatcherStarts.has(key)) continue + if (watch.kind === 'terminal-grace' && watch.expiresAtMs <= this.#clock.now()) { + await this.#stopSlackWatcher(watch.issue) + continue + } + await this.#state.setSlackThread(this.#workspaceId, key, watch.threadId) + const watchRecord = escalationWatchRecord(watch.decision) + if (watch.kind === 'terminal-grace') { + const conversationId = slackConversationId(watch.threadId) + await this.#slackConversationTurns.cancel(conversationId) + await this.#state.clearConversationSession(this.#workspaceId, conversationId) + await this.#rearmSlackWatcher(watchRecord, watch.threadId, { replayConversationReplies: true }) + this.#scheduleSlackTerminalWatchExpiry(watch.issue, watch.expiresAtMs) + continue + } + await this.#rearmSlackWatcher(watchRecord, watch.threadId, { replayConversationReplies: true }) + } await this.#sweepWaitingClarifications() for (const [, waiting] of await this.#state.listWaitingClarifications(this.#workspaceId)) { if (!waiting.threadId) continue @@ -14386,6 +14435,9 @@ export class FactoryLoop implements Factory { async #stopSlackWatcher(issue: IssueRef): Promise { const key = issueKey(issue) + const expiryTimer = this.#slackTerminalWatchExpiryTimers.get(key) + if (expiryTimer) clearTimeout(expiryTimer) + this.#slackTerminalWatchExpiryTimers.delete(key) const watcher = this.#slackWatchers.get(key) this.#slackWatchers.delete(key) const threadId = await this.#state.getSlackThread(this.#workspaceId, key) @@ -14396,6 +14448,77 @@ export class FactoryLoop implements Factory { await this.#state.clearConversationSession(this.#workspaceId, conversationId) } await this.#state.clearSlackThread(this.#workspaceId, key) + await this.#state.clearSlackThreadWatch(this.#workspaceId, key) + } + + async #retireSlackWatcher(record: InFlightIssue): Promise { + const key = issueKey(record.issue) + const threadId = await this.#state.getSlackThread(this.#workspaceId, key) + if (!threadId) { + await this.#stopSlackWatcher(record.issue) + return + } + + const existingWatch = (await this.#state.listSlackThreadWatches(this.#workspaceId)) + .find(([watchKey]) => watchKey === key)?.[1] + const expiresAtMs = existingWatch?.kind === 'terminal-grace' + ? existingWatch.expiresAtMs + : this.#clock.now() + SLACK_TERMINAL_THREAD_GRACE_MS + await this.#state.setSlackThreadWatch(this.#workspaceId, key, { + kind: 'terminal-grace', + issue: { ...record.issue }, + decision: structuredClone(record.decision), + threadId, + expiresAtMs, + }) + + // A terminal thread must never retain a resumable session for an agent that + // has already exited. Keep only the exact-thread listener so a late human + // reply receives the explicit no-active-agent writeback below. + const conversationId = slackConversationId(threadId) + await this.#slackConversationTurns.cancel(conversationId) + await this.#state.clearConversationSession(this.#workspaceId, conversationId) + if (!this.#slackWatchers.has(key) && !this.#stopping) { + await this.#rearmSlackWatcher(record, threadId) + } + this.#scheduleSlackTerminalWatchExpiry(record.issue, expiresAtMs) + this.#increment('slackTerminalWatchersRetained') + } + + #scheduleSlackTerminalWatchExpiry( + issue: IssueRef, + expiresAtMs: number, + retryDelayMs?: number, + ): void { + if (this.#stopping) return + const key = issueKey(issue) + const existing = this.#slackTerminalWatchExpiryTimers.get(key) + if (existing) clearTimeout(existing) + const timer = setTimeout(() => { + this.#slackTerminalWatchExpiryTimers.delete(key) + void this.#expireSlackTerminalWatcher(issue, expiresAtMs).catch((error) => { + this.#logger.warn?.('[factory] failed to expire terminal Slack reply watcher; retrying', { + issue: issue.key, + error, + }) + this.#scheduleSlackTerminalWatchExpiry(issue, expiresAtMs, SLACK_REPLY_ROUTE_RETRY_MS) + }) + }, retryDelayMs ?? Math.max(0, expiresAtMs - this.#clock.now())) + timer.unref?.() + this.#slackTerminalWatchExpiryTimers.set(key, timer) + } + + async #expireSlackTerminalWatcher(issue: IssueRef, expiresAtMs: number): Promise { + const key = issueKey(issue) + const watch = (await this.#state.listSlackThreadWatches(this.#workspaceId)) + .find(([watchKey]) => watchKey === key)?.[1] + if (watch?.kind !== 'terminal-grace' || watch.expiresAtMs !== expiresAtMs) return + if (watch.expiresAtMs > this.#clock.now()) { + this.#scheduleSlackTerminalWatchExpiry(issue, watch.expiresAtMs) + return + } + await this.#stopSlackWatcher(issue) + this.#increment('slackTerminalWatchersExpired') } async #readSlackReply(path: string): Promise { @@ -14464,33 +14587,116 @@ export class FactoryLoop implements Factory { } const conversationId = slackConversationId(reply.threadTs) - const conversation = await this.#state.getConversationSession(this.#workspaceId, conversationId) + let conversation = await this.#state.getConversationSession(this.#workspaceId, conversationId) + let liveRecord: InFlightIssue | undefined + if (!conversation) { + liveRecord = (await this.#batch()).getIssue(record.issue) + if (liveRecord && !liveRecord.dryRun) { + await this.#ensureSlackConversationSession(liveRecord, reply.threadTs) + conversation = await this.#state.getConversationSession(this.#workspaceId, conversationId) + } + } if (conversation && issueKey(conversation.issue) === clarificationKey) { + const replyId = `${reply.threadTs}:${reply.messageTs}` const queued = await this.#state.appendConversationMessage(this.#workspaceId, conversationId, { - id: `${reply.threadTs}:${reply.messageTs}`, + id: replyId, text, receivedAtMs: slackMessageReceivedAtMs(reply.messageTs, this.#clock.now()), providerSequence: reply.messageTs, author: reply.author, }) - if (!queued) { + const durable = queued ?? await this.#state.getConversationSession(this.#workspaceId, conversationId) + if (!durable || !durable.processedMessageIds.includes(replyId)) { + throw new Error(`Slack reply ${replyId} was not durably queued`) + } + if (!(durable.acknowledgedMessageIds ?? []).includes(replyId)) { + const acknowledgementClaimId = randomUUID() + const acknowledgementClaimed = await this.#state.claimConversationMessageAcknowledgement( + this.#workspaceId, + conversationId, + replyId, + acknowledgementClaimId, + this.#clock.now(), + SLACK_REPLY_ACKNOWLEDGEMENT_LEASE_MS, + ) + if (acknowledgementClaimed) { + try { + if (!this.#slack) throw new Error(`Slack reply ${replyId} cannot be acknowledged without writeback`) + const owner = durable.agent?.role === 'babysitter' + ? 'the PR babysitter' + : durable.agent + ? 'the issue implementer' + : 'an issue agent' + const receipt = durable.agent + ? `Factory received this reply and durably queued it for ${owner}.` + : 'Factory received and durably stored this reply; it will route when an issue agent is resumable.' + await this.#slack.reply(reply.threadTs, receipt) + if (!await this.#state.completeConversationMessageAcknowledgement( + this.#workspaceId, + conversationId, + replyId, + acknowledgementClaimId, + )) { + throw new Error(`Slack reply ${replyId} receipt could not be recorded`) + } + this.#increment('slackConversationRepliesAcknowledged') + } catch (error) { + await this.#state.releaseConversationMessageAcknowledgement( + this.#workspaceId, + conversationId, + replyId, + acknowledgementClaimId, + ) + throw error + } + } else { + const acknowledgementState = await this.#state.getConversationSession( + this.#workspaceId, + conversationId, + ) + if (!(acknowledgementState?.acknowledgedMessageIds ?? []).includes(replyId)) { + throw new Error(`Slack reply ${replyId} receipt is claimed by another handler; retrying`) + } + } + } + if (queued) { + this.#increment('slackConversationRepliesQueued') + } else { this.#increment('slackConversationDuplicateRepliesSuppressed') - return } - this.#increment('slackConversationRepliesQueued') - this.#slackConversationTurns.schedule(conversationId) + const pending = durable.pending.some((message) => message.id === replyId) || + Boolean(durable.delivery?.messages.some((message) => message.id === replyId)) + if (pending && durable.agent) { + this.#slackConversationTurns.schedule(conversationId) + } else if (pending) { + this.#increment('slackConversationRepliesWaitingForOwner') + } return } - const liveRecord = (await this.#batch()).getIssue(record.issue) + liveRecord ??= (await this.#batch()).getIssue(record.issue) if (!liveRecord || liveRecord.dryRun) { if (isTriageEscalationWatchRecord(record)) { return await this.#handleTriageEscalationSlackAnswer(record, text) } this.#increment('slackAnswersIgnoredNoInFlight') + if (this.#slack) { + await this.#slack.reply( + reply.threadTs, + 'Factory received this reply but could not route it because this work unit no longer has an active agent. Please continue on the linked issue or pull request.', + ) + this.#increment('slackAnswersUnroutableVisible') + } return } this.#increment('slackAnswersIgnoredNoConversationSession') + if (this.#slack) { + await this.#slack.reply( + reply.threadTs, + 'Factory received this reply but could not create a durable agent route. It will remain replayable; please also continue on the linked issue or pull request.', + ) + this.#increment('slackAnswersUnroutableVisible') + } } async #wakeWaitingClarification(key: string, waiting: WaitingClarification): Promise { @@ -14885,6 +15091,7 @@ export class FactoryLoop implements Factory { const batch = await this.#batch() if (batch.isInFlight(record.issue) || batch.isQueued(record.issue)) { this.#increment('slackTriageAnswersIgnoredAlreadyActive') + await this.#state.clearSlackThreadWatch(this.#workspaceId, issueKey(record.issue)) return } if (await this.#dispatchBlockReason(record.issue)) { @@ -14902,6 +15109,10 @@ export class FactoryLoop implements Factory { if (hasDispatchableRoute(decision)) { this.#pendingSlackClarifications.set(issueKey(decision.issue), text) const result = await this.#startOrQueueSlackClarifiedDecision(dispatchAfterSlackClarification(decision, escalationReason)) + const active = await this.#batch() + if (result || active.isInFlight(decision.issue) || active.isQueued(decision.issue)) { + await this.#state.clearSlackThreadWatch(this.#workspaceId, issueKey(record.issue)) + } this.#increment('slackTriageAnswersDispatchedWithRemainingEscalation') return result } @@ -14915,6 +15126,10 @@ export class FactoryLoop implements Factory { this.#pendingSlackClarifications.set(issueKey(decision.issue), text) const result = await this.#startOrQueueSlackClarifiedDecision(decision) + const active = await this.#batch() + if (result || active.isInFlight(decision.issue) || active.isQueued(decision.issue)) { + await this.#state.clearSlackThreadWatch(this.#workspaceId, issueKey(record.issue)) + } this.#increment('slackTriageAnswersDispatched') return result } diff --git a/src/ports/state.ts b/src/ports/state.ts index 0f22ed9..569c87b 100644 --- a/src/ports/state.ts +++ b/src/ports/state.ts @@ -115,7 +115,7 @@ export type ConversationSessionState = { externalId: string /** Provider-specific routing metadata; continuity itself stays provider-neutral. */ context: Record - agent: { + agent?: { name: string sessionRef: string /** @@ -134,6 +134,10 @@ export type ConversationSessionState = { history: ConversationMessage[] /** Durable dedupe ledger; unlike rendered history, this is never context-trimmed. */ processedMessageIds: string[] + /** Human replies whose visible provider receipt has been acknowledged. */ + acknowledgedMessageIds?: string[] + /** Short durable claims preventing duplicate concurrent provider receipts. */ + acknowledgementClaims?: Record /** New replies waiting for the short coalescing window. */ pending: ConversationMessage[] /** Claimed batch; new arrivals remain in pending while this resume runs. */ @@ -144,10 +148,24 @@ export type ConversationSessionState = { attempts: number messages: ConversationMessage[] /** Binding captured at claim time so a later handoff cannot be overwritten. */ - agent: Pick + agent: Pick, 'name' | 'sessionRef'> } } +/** Durable metadata required to reconstruct a pre-dispatch Slack watcher. */ +export type SlackThreadWatchState = { + kind: 'triage' + issue: IssueRef + decision: TriageDecision + threadId: string +} | { + kind: 'terminal-grace' + issue: IssueRef + decision: TriageDecision + threadId: string + expiresAtMs: number +} + export type DispatchAttemptState = { attempts: number inFlight: boolean @@ -441,11 +459,17 @@ export interface StateStore { getSlackThread(workspaceId: string, issueKey: string): Promise clearSlackThread(workspaceId: string, issueKey: string): Promise clearSlackThreads(workspaceId: string): Promise + setSlackThreadWatch(workspaceId: string, issueKey: string, watch: SlackThreadWatchState): Promise + listSlackThreadWatches(workspaceId: string): Promise> + clearSlackThreadWatch(workspaceId: string, issueKey: string): Promise reserveConversationSession(workspaceId: string, conversationId: string, session: ConversationSessionState): Promise getConversationSession(workspaceId: string, conversationId: string): Promise listConversationSessions(workspaceId: string): Promise> appendConversationMessage(workspaceId: string, conversationId: string, message: ConversationMessage): Promise + claimConversationMessageAcknowledgement(workspaceId: string, conversationId: string, messageId: string, claimId: string, nowMs: number, leaseMs: number): Promise + completeConversationMessageAcknowledgement(workspaceId: string, conversationId: string, messageId: string, claimId: string): Promise + releaseConversationMessageAcknowledgement(workspaceId: string, conversationId: string, messageId: string, claimId: string): Promise claimConversationTurn(workspaceId: string, conversationId: string, owner: string, claimId: string, nowMs: number, leaseMs: number): Promise renewConversationTurn(workspaceId: string, conversationId: string, owner: string, claimId: string, nowMs: number): Promise completeConversationTurn(workspaceId: string, conversationId: string, owner: string, claimId: string, agent: { name: string; sessionRef?: string }): Promise @@ -456,7 +480,7 @@ export interface StateStore { * once a babysitter takes over an issue whose Slack thread was reserved by the * implementer) without disturbing accumulated history/pending turns. */ - rebindConversationSession(workspaceId: string, conversationId: string, agent: ConversationSessionState['agent']): Promise + rebindConversationSession(workspaceId: string, conversationId: string, agent: NonNullable): Promise setGithubIssueCommentWatch(workspaceId: string, key: string, watch: GithubIssueCommentWatchState): Promise listGithubIssueCommentWatches(workspaceId: string): Promise> diff --git a/src/state/document-store.ts b/src/state/document-store.ts index 3252fa3..a805c6c 100644 --- a/src/state/document-store.ts +++ b/src/state/document-store.ts @@ -5,11 +5,13 @@ import type { DiscoverySweepState, DispatchLifecycle, GithubIssueCommentWatchState, + SlackThreadWatchState, WaitingClarification, } from '../ports/state' export type PersistedWorkspaceState = { githubIssueCommentWatches: Record + slackThreadWatches: Record waitingClarifications: Record babysitterSessions: Record babysitterGenerations: Record diff --git a/src/state/file-state-store.test.ts b/src/state/file-state-store.test.ts index 4a180fa..ed10c08 100644 --- a/src/state/file-state-store.test.ts +++ b/src/state/file-state-store.test.ts @@ -516,6 +516,110 @@ describe('FileStateStore', () => { } }) + it('persists an unowned Slack turn, fences its visible receipt, and delivers after owner rebind', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-file-state-slack-unowned-')) + try { + const watchStatePath = join(root, 'factory-state.json') + const conversationId = 'slack:1780751612.176220' + const first = new FileStateStore({ batchSize: 2, watchStatePath }) + await first.reserveConversationSession('workspace-1', conversationId, { + provider: 'slack', + issue: { uuid: 'uuid-131', key: 'AR-131', path: '/linear/issues/AR-131__uuid-131.json' }, + externalId: '1780751612.176220', + context: { channelDir: 'C0FACTORY__factory-e2e' }, + history: [], + processedMessageIds: [], + acknowledgedMessageIds: [], + acknowledgementClaims: {}, + pending: [], + }) + await first.appendConversationMessage('workspace-1', conversationId, { + id: 'message-1', text: 'Keep the complete long instruction.', receivedAtMs: 1_000, + }) + expect(await first.claimConversationTurn( + 'workspace-1', conversationId, 'turn-owner', 'turn-claim', 1_001, 60_000, + )).toBeUndefined() + + const restarted = new FileStateStore({ batchSize: 2, watchStatePath }) + expect(await restarted.claimConversationMessageAcknowledgement( + 'workspace-1', conversationId, 'message-1', 'ack-a', 1_002, 60_000, + )).toBe(true) + expect(await first.claimConversationMessageAcknowledgement( + 'workspace-1', conversationId, 'message-1', 'ack-b', 1_003, 60_000, + )).toBe(false) + await restarted.releaseConversationMessageAcknowledgement('workspace-1', conversationId, 'message-1', 'ack-a') + expect(await first.claimConversationMessageAcknowledgement( + 'workspace-1', conversationId, 'message-1', 'ack-b', 1_004, 60_000, + )).toBe(true) + expect(await first.completeConversationMessageAcknowledgement( + 'workspace-1', conversationId, 'message-1', 'ack-b', + )).toBe(true) + await restarted.rebindConversationSession('workspace-1', conversationId, { + name: 'ar-131-babysit-factory', sessionRef: 'session-babysitter', role: 'babysitter', + }) + + expect(await new FileStateStore({ batchSize: 2, watchStatePath }).claimConversationTurn( + 'workspace-1', conversationId, 'turn-owner', 'turn-claim', 1_005, 60_000, + )).toMatchObject({ + agent: { name: 'ar-131-babysit-factory', sessionRef: 'session-babysitter' }, + acknowledgedMessageIds: ['message-1'], + delivery: { messages: [{ id: 'message-1', text: 'Keep the complete long instruction.' }] }, + }) + } finally { + await rm(root, { recursive: true, force: true }) + } + }) + + it('persists and clears the compact pre-dispatch Slack triage watch', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-file-state-slack-watch-')) + try { + const watchStatePath = join(root, 'factory-state.json') + const lifecycle = dispatchLifecycle(132) + const watch = { + kind: 'triage' as const, + issue: lifecycle.issue, + decision: lifecycle.decision, + threadId: '1780751612.176221', + } + const first = new FileStateStore({ batchSize: 2, watchStatePath }) + await first.setSlackThreadWatch('workspace-1', 'AR-132:uuid-132', watch) + + const restarted = new FileStateStore({ batchSize: 2, watchStatePath }) + expect(await restarted.listSlackThreadWatches('workspace-1')).toEqual([ + ['AR-132:uuid-132', watch], + ]) + await restarted.clearSlackThreadWatch('workspace-1', 'AR-132:uuid-132') + expect(await new FileStateStore({ batchSize: 2, watchStatePath }) + .listSlackThreadWatches('workspace-1')).toEqual([]) + } finally { + await rm(root, { recursive: true, force: true }) + } + }) + + it('persists the bounded terminal Slack watch used for restart replay', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-file-state-terminal-slack-watch-')) + try { + const watchStatePath = join(root, 'factory-state.json') + const lifecycle = dispatchLifecycle(133) + const watch = { + kind: 'terminal-grace' as const, + issue: lifecycle.issue, + decision: lifecycle.decision, + threadId: '1780751612.176222', + expiresAtMs: 86_401_000, + } + const first = new FileStateStore({ batchSize: 2, watchStatePath }) + await first.setSlackThreadWatch('workspace-1', 'AR-133:uuid-133', watch) + + expect(await new FileStateStore({ batchSize: 2, watchStatePath }) + .listSlackThreadWatches('workspace-1')).toEqual([ + ['AR-133:uuid-133', watch], + ]) + } finally { + await rm(root, { recursive: true, force: true }) + } + }) + it('fences stale claim completion and preserves a conversation owner rebound during resume', async () => { const root = await mkdtemp(join(tmpdir(), 'factory-file-state-slack-fencing-')) try { diff --git a/src/state/file-state-store.ts b/src/state/file-state-store.ts index f3308da..33299f9 100644 --- a/src/state/file-state-store.ts +++ b/src/state/file-state-store.ts @@ -19,6 +19,7 @@ import type { DiscoverySweepLease, DiscoverySweepRenewal, DiscoverySweepState, + SlackThreadWatchState, WaitingClarification, } from '../ports/state' import { InMemoryStateStore, type InMemoryStateStoreOptions } from './in-memory-state-store' @@ -56,9 +57,9 @@ const WATCH_STATE_LOCK_STALE_MS = 60_000 /** * Keeps the factory's general runtime bookkeeping in memory while persisting - * GitHub escalation watches, parked clarification teams, exact babysitter PR - * ownership, and thread-owned conversation turns atomically so they survive a - * CLI process restart. + * GitHub/Slack escalation watches, parked clarification teams, exact + * babysitter PR ownership, and thread-owned conversation turns 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. */ @@ -373,6 +374,40 @@ export class DocumentStateStore extends InMemoryStateStore { })) } + override async setSlackThreadWatch( + workspaceId: string, + key: string, + watch: SlackThreadWatchState, + ): Promise { + await this.#exclusive(async () => this.#withMutationLock(async () => { + const document = await this.#loadFromDisk() + const workspace = document.workspaces[workspaceId] ??= emptyWorkspaceState() + workspace.slackThreadWatches[key] = structuredClone(watch) + await this.#persist(document) + })) + } + + override async listSlackThreadWatches( + workspaceId: string, + ): Promise> { + return await this.#exclusive(async () => { + const document = await this.#loadFromDisk() + return Object.entries(document.workspaces[workspaceId]?.slackThreadWatches ?? {}) + .map(([key, watch]) => [key, structuredClone(watch)]) + }) + } + + override async clearSlackThreadWatch(workspaceId: string, key: string): Promise { + await this.#exclusive(async () => this.#withMutationLock(async () => { + const document = await this.#loadFromDisk() + const workspace = document.workspaces[workspaceId] + if (!workspace || !(key in workspace.slackThreadWatches)) return + delete workspace.slackThreadWatches[key] + if (workspaceIsEmpty(workspace)) delete document.workspaces[workspaceId] + await this.#persist(document) + })) + } + override async setGithubIssueCommentWatch( workspaceId: string, key: string, @@ -925,6 +960,55 @@ export class DocumentStateStore extends InMemoryStateStore { }) } + override async claimConversationMessageAcknowledgement( + workspaceId: string, + conversationId: string, + messageId: string, + claimId: string, + nowMs: number, + leaseMs: number, + ): Promise { + const result = await this.#mutateConversation(workspaceId, conversationId, (session) => { + if (!conversationHasMessage(session, messageId)) return false + if ((session.acknowledgedMessageIds ?? []).includes(messageId)) return false + session.acknowledgementClaims ??= {} + const current = session.acknowledgementClaims[messageId] + if (current && current.claimedAtMs + leaseMs > nowMs) return false + session.acknowledgementClaims[messageId] = { claimId, claimedAtMs: nowMs } + return true + }) + return Boolean(result) + } + + override async completeConversationMessageAcknowledgement( + workspaceId: string, + conversationId: string, + messageId: string, + claimId: string, + ): Promise { + const result = await this.#mutateConversation(workspaceId, conversationId, (session) => { + if (session.acknowledgementClaims?.[messageId]?.claimId !== claimId) return false + session.acknowledgedMessageIds ??= [] + if (!session.acknowledgedMessageIds.includes(messageId)) session.acknowledgedMessageIds.push(messageId) + delete session.acknowledgementClaims[messageId] + return true + }) + return Boolean(result) + } + + override async releaseConversationMessageAcknowledgement( + workspaceId: string, + conversationId: string, + messageId: string, + claimId: string, + ): Promise { + await this.#mutateConversation(workspaceId, conversationId, (session) => { + if (session.acknowledgementClaims?.[messageId]?.claimId !== claimId) return false + delete session.acknowledgementClaims[messageId] + return true + }) + } + override async claimConversationTurn( workspaceId: string, conversationId: string, @@ -940,10 +1024,11 @@ export class DocumentStateStore extends InMemoryStateStore { const attempts = session.delivery?.attempts ?? 0 if (session.delivery) session.pending.unshift(...session.delivery.messages) session.pending.sort(compareConversationMessages) - if (session.pending.length === 0) { + if (!session.agent || session.pending.length === 0) { session.delivery = undefined return false } + const agent = session.agent session.delivery = { claimId, owner, @@ -951,8 +1036,8 @@ export class DocumentStateStore extends InMemoryStateStore { attempts: attempts + 1, messages: session.pending.splice(0), agent: { - name: session.agent.name, - sessionRef: session.agent.sessionRef, + name: agent.name, + sessionRef: agent.sessionRef, }, } return true @@ -985,6 +1070,7 @@ export class DocumentStateStore extends InMemoryStateStore { if (!session.delivery || session.delivery.owner !== owner || session.delivery.claimId !== claimId) return false session.history = [...session.history, ...session.delivery.messages].slice(-CONVERSATION_HISTORY_LIMIT) if ( + session.agent && session.agent.name === session.delivery.agent.name && session.agent.sessionRef === session.delivery.agent.sessionRef ) { @@ -1021,7 +1107,7 @@ export class DocumentStateStore extends InMemoryStateStore { override async rebindConversationSession( workspaceId: string, conversationId: string, - agent: ConversationSessionState['agent'], + agent: NonNullable, ): Promise { const result = await this.#mutateConversation(workspaceId, conversationId, (session) => { session.agent = structuredClone(agent) @@ -1232,6 +1318,7 @@ const dispatchLifecycleHandedOffToBabysitters = (lifecycle: DispatchLifecycle): const emptyWorkspaceState = (): PersistedWorkspaceState => ({ githubIssueCommentWatches: {}, + slackThreadWatches: {}, waitingClarifications: {}, babysitterSessions: {}, babysitterGenerations: {}, @@ -1242,6 +1329,7 @@ const emptyWorkspaceState = (): PersistedWorkspaceState => ({ const workspaceIsEmpty = (workspace: PersistedWorkspaceState): boolean => Object.keys(workspace.githubIssueCommentWatches).length === 0 && + Object.keys(workspace.slackThreadWatches).length === 0 && Object.keys(workspace.waitingClarifications).length === 0 && Object.keys(workspace.babysitterSessions).length === 0 && Object.keys(workspace.babysitterGenerations).length === 0 && diff --git a/src/state/in-memory-state-store.ts b/src/state/in-memory-state-store.ts index fa9c61f..f2ca689 100644 --- a/src/state/in-memory-state-store.ts +++ b/src/state/in-memory-state-store.ts @@ -12,6 +12,7 @@ import type { DispatchAttemptState, GithubIssueCommentWatchState, RegistryHandoffAgent, + SlackThreadWatchState, ConversationMessage, ConversationSessionState, DiscoveryCheckpoint, @@ -29,6 +30,7 @@ type WorkspaceState = { criticalMessages: Map resumedExitKeys: Set slackThreadIds: Map + slackThreadWatches: Map conversationSessions: Map githubIssueCommentWatches: Map seenAgentQuestionKeys: Set @@ -334,6 +336,19 @@ export class InMemoryStateStore implements StateStore { this.#workspace(workspaceId).slackThreadIds.clear() } + async setSlackThreadWatch(workspaceId: string, issueKey: string, watch: SlackThreadWatchState): Promise { + this.#workspace(workspaceId).slackThreadWatches.set(issueKey, structuredClone(watch)) + } + + async listSlackThreadWatches(workspaceId: string): Promise> { + return [...this.#workspace(workspaceId).slackThreadWatches] + .map(([key, watch]) => [key, structuredClone(watch)]) + } + + async clearSlackThreadWatch(workspaceId: string, issueKey: string): Promise { + this.#workspace(workspaceId).slackThreadWatches.delete(issueKey) + } + async reserveConversationSession( workspaceId: string, conversationId: string, @@ -372,6 +387,50 @@ export class InMemoryStateStore implements StateStore { return cloneConversationSession(session) } + async claimConversationMessageAcknowledgement( + workspaceId: string, + conversationId: string, + messageId: string, + claimId: string, + nowMs: number, + leaseMs: number, + ): Promise { + const session = this.#workspace(workspaceId).conversationSessions.get(conversationId) + if (!session || !conversationHasMessage(session, messageId)) return false + if ((session.acknowledgedMessageIds ?? []).includes(messageId)) return false + session.acknowledgementClaims ??= {} + const current = session.acknowledgementClaims[messageId] + if (current && current.claimedAtMs + leaseMs > nowMs) return false + session.acknowledgementClaims[messageId] = { claimId, claimedAtMs: nowMs } + return true + } + + async completeConversationMessageAcknowledgement( + workspaceId: string, + conversationId: string, + messageId: string, + claimId: string, + ): Promise { + const session = this.#workspace(workspaceId).conversationSessions.get(conversationId) + if (session?.acknowledgementClaims?.[messageId]?.claimId !== claimId) return false + session.acknowledgedMessageIds ??= [] + if (!session.acknowledgedMessageIds.includes(messageId)) session.acknowledgedMessageIds.push(messageId) + delete session.acknowledgementClaims[messageId] + return true + } + + async releaseConversationMessageAcknowledgement( + workspaceId: string, + conversationId: string, + messageId: string, + claimId: string, + ): Promise { + const session = this.#workspace(workspaceId).conversationSessions.get(conversationId) + if (session?.acknowledgementClaims?.[messageId]?.claimId === claimId) { + delete session.acknowledgementClaims[messageId] + } + } + async claimConversationTurn( workspaceId: string, conversationId: string, @@ -389,10 +448,11 @@ export class InMemoryStateStore implements StateStore { session.pending.unshift(...session.delivery.messages) } session.pending.sort(compareConversationMessages) - if (session.pending.length === 0) { + if (!session.agent || session.pending.length === 0) { session.delivery = undefined return undefined } + const agent = session.agent session.delivery = { claimId, owner, @@ -400,8 +460,8 @@ export class InMemoryStateStore implements StateStore { attempts: (session.delivery?.attempts ?? 0) + 1, messages: session.pending.splice(0), agent: { - name: session.agent.name, - sessionRef: session.agent.sessionRef, + name: agent.name, + sessionRef: agent.sessionRef, }, } return cloneConversationSession(session) @@ -431,6 +491,7 @@ export class InMemoryStateStore implements StateStore { if (!session?.delivery || session.delivery.owner !== owner || session.delivery.claimId !== claimId) return false session.history = [...session.history, ...session.delivery.messages].slice(-CONVERSATION_HISTORY_LIMIT) if ( + session.agent && session.agent.name === session.delivery.agent.name && session.agent.sessionRef === session.delivery.agent.sessionRef ) { @@ -456,7 +517,7 @@ export class InMemoryStateStore implements StateStore { async rebindConversationSession( workspaceId: string, conversationId: string, - agent: ConversationSessionState['agent'], + agent: NonNullable, ): Promise { const session = this.#workspace(workspaceId).conversationSessions.get(conversationId) if (!session) return false @@ -818,6 +879,7 @@ export class InMemoryStateStore implements StateStore { criticalMessages: new Map(), resumedExitKeys: new Set(), slackThreadIds: new Map(), + slackThreadWatches: new Map(), conversationSessions: new Map(), githubIssueCommentWatches: new Map(), seenAgentQuestionKeys: new Set(), diff --git a/src/state/watch-state-document.ts b/src/state/watch-state-document.ts index 1ac74d0..395306e 100644 --- a/src/state/watch-state-document.ts +++ b/src/state/watch-state-document.ts @@ -7,6 +7,7 @@ import type { DiscoverySweepState, DispatchLifecycle, GithubIssueCommentWatchState, + SlackThreadWatchState, WaitingClarification, } from '../ports/state' import type { AgentSpec, SpawnResult } from '../ports/fleet' @@ -23,6 +24,7 @@ export const parseWatchStateDocument = (value: unknown): WatchStateDocument => { for (const [workspaceId, rawWorkspace] of Object.entries(value.workspaces)) { if (!isRecord(rawWorkspace)) throw invalidDocument() const watches = rawWorkspace.githubIssueCommentWatches + const slackWatches = rawWorkspace.slackThreadWatches const clarifications = rawWorkspace.waitingClarifications const babysitters = rawWorkspace.babysitterSessions const generations = rawWorkspace.babysitterGenerations @@ -31,6 +33,7 @@ export const parseWatchStateDocument = (value: unknown): WatchStateDocument => { const discoverySweep = rawWorkspace.discoverySweep if ( !isRecord(watches) || + (slackWatches !== undefined && !isRecord(slackWatches)) || !isRecord(clarifications) || (babysitters !== undefined && !isRecord(babysitters)) || (generations !== undefined && !isRecord(generations)) || @@ -40,6 +43,7 @@ export const parseWatchStateDocument = (value: unknown): WatchStateDocument => { ) throw invalidDocument() workspaces[workspaceId] = { githubIssueCommentWatches: parseGithubIssueCommentWatches(watches), + slackThreadWatches: parseSlackThreadWatches(slackWatches ?? {}), waitingClarifications: parseWaitingClarifications(clarifications), babysitterSessions: parseBabysitterSessions(babysitters ?? {}), babysitterGenerations: parseBabysitterGenerations(generations ?? {}), @@ -62,6 +66,7 @@ export const parseWatchStateDocument = (value: unknown): WatchStateDocument => { } workspaces[workspaceId] = { githubIssueCommentWatches: parseGithubIssueCommentWatches(watches), + slackThreadWatches: {}, waitingClarifications: parseWaitingClarifications(clarifications), babysitterSessions: parseBabysitterSessions(babysitters ?? {}), babysitterGenerations: {}, @@ -78,6 +83,7 @@ export const parseWatchStateDocument = (value: unknown): WatchStateDocument => { if (!isRecord(watches)) throw invalidDocument() workspaces[workspaceId] = { githubIssueCommentWatches: parseGithubIssueCommentWatches(watches), + slackThreadWatches: {}, waitingClarifications: {}, babysitterSessions: {}, babysitterGenerations: {}, @@ -123,7 +129,7 @@ const parseConversationSessions = ( ): Record => { const sessions: Record = {} for (const [conversationId, candidate] of Object.entries(value)) { - if (!isRecord(candidate) || !isRecord(candidate.issue) || !isRecord(candidate.agent) || !isRecord(candidate.context)) { + if (!isRecord(candidate) || !isRecord(candidate.issue) || !isRecord(candidate.context)) { throw invalidDocument() } const issue = candidate.issue @@ -133,9 +139,11 @@ const parseConversationSessions = ( typeof issue.uuid !== 'string' || typeof issue.key !== 'string' || typeof issue.path !== 'string' || typeof candidate.provider !== 'string' || typeof candidate.externalId !== 'string' || !Object.values(candidate.context).every((entry) => typeof entry === 'string') || - typeof agent.name !== 'string' || typeof agent.sessionRef !== 'string' || + (agent !== undefined && (!isRecord(agent) || typeof agent.name !== 'string' || typeof agent.sessionRef !== 'string')) || !validConversationMessages(candidate.history) || !validConversationMessages(candidate.pending) || (candidate.processedMessageIds !== undefined && !validConversationMessageIds(candidate.processedMessageIds)) || + (candidate.acknowledgedMessageIds !== undefined && !validConversationMessageIds(candidate.acknowledgedMessageIds)) || + (candidate.acknowledgementClaims !== undefined && !validConversationAcknowledgementClaims(candidate.acknowledgementClaims)) || (delivery !== undefined && !validConversationDelivery(delivery) && !validLegacyConversationDelivery(delivery)) ) throw invalidDocument() const session = structuredClone(candidate) as unknown as ConversationSessionState @@ -150,6 +158,12 @@ const parseConversationSessions = ( ...(session.delivery?.messages ?? []), ].map((message) => message.id))] : [...candidate.processedMessageIds as string[]] + if (candidate.acknowledgedMessageIds !== undefined) { + session.acknowledgedMessageIds = [...candidate.acknowledgedMessageIds as string[]] + } + if (candidate.acknowledgementClaims !== undefined) { + session.acknowledgementClaims = structuredClone(candidate.acknowledgementClaims) as ConversationSessionState['acknowledgementClaims'] + } sessions[conversationId] = session } return sessions @@ -180,6 +194,10 @@ const validLegacyConversationDelivery = (value: unknown): value is { const validConversationMessageIds = (value: unknown): value is string[] => Array.isArray(value) && value.every((id) => typeof id === 'string') +const validConversationAcknowledgementClaims = (value: unknown): boolean => + isRecord(value) && Object.values(value).every((claim) => isRecord(claim) && + typeof claim.claimId === 'string' && typeof claim.claimedAtMs === 'number') + const parseBabysitterSessions = (value: Record): Record => { const sessions: Record = {} for (const [key, candidate] of Object.entries(value)) { @@ -288,6 +306,24 @@ const parseGithubIssueCommentWatches = ( return watches } +const parseSlackThreadWatches = ( + value: Record, +): Record => { + const watches: Record = {} + for (const [key, candidate] of Object.entries(value)) { + if ( + !isRecord(candidate) || !validIssueRef(candidate.issue) || !validTriageDecision(candidate.decision) || + typeof candidate.threadId !== 'string' || + (candidate.kind !== 'triage' && candidate.kind !== 'terminal-grace') || + (candidate.kind === 'terminal-grace' && ( + !validNumber(candidate.expiresAtMs) || !validOptionalNumber(candidate.retiredAtMs) + )) + ) throw invalidDocument() + watches[key] = structuredClone(candidate) as unknown as SlackThreadWatchState + } + return watches +} + const validGithubWatchPending = (value: unknown): boolean => isRecord(value) && typeof value.correlationId === 'string' && (value.kind === 'triage' || value.kind === 'agent-question') && From a49d160da7fe310ce4d581465bac0d5e797235eb Mon Sep 17 00:00:00 2001 From: Khaliq Date: Mon, 17 Aug 2026 15:14:23 +0200 Subject: [PATCH 2/6] fix: fence terminal Slack reply cleanup --- src/orchestrator/factory.test.ts | 64 ++++++++++++++++++ src/orchestrator/factory.ts | 101 +++++++++++++++++++++++++---- src/ports/state.ts | 2 + src/state/file-state-store.test.ts | 1 + 4 files changed, 156 insertions(+), 12 deletions(-) diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 4e9eeca..f3fb4d4 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -19402,16 +19402,30 @@ describe('FactoryLoop PR babysitter', () => { const mount = new ConfirmRecordingSlackMountClient({ [issuePath(407)]: issue }) const factoryConfig = config({ slack: slackConfig() }) const state = () => new FileStateStore({ batchSize: 10, watchStatePath }) + const clock = new ManualClock() + clock.advance(10_000) const firstFleet = new FakeFleetClient() + firstFleet.setSessionRef('ar-407-impl-pear', 'session-ar-407-impl-pear') const first = createFactory(factoryConfig, { mount, fleet: firstFleet, triage: new StaticTriage(), stateStore: state(), + clock, }) let restarted: ReturnType | undefined try { await first.dispatch(await first.triageIssue(parseLinearIssue(issuePath(407), issue))) + emitSlackReply(mount, slackReplyFixturePath( + 'C0FACTORY__factory-e2e', mount.threadTs, 'human-before-terminal', + ), 'slack-human-before-terminal', { + ts: '9.000', + text: 'This reply was already routed before completion.', + user: 'U407', + user_name: 'human', + user_is_bot: false, + }) + await expectSlackConversationResume(firstFleet, ['This reply was already routed before completion.']) firstFleet.emitAgentExit('ar-407-impl-pear', 'issue-done') await vi.waitFor(() => expect(first.status().inFlight).toEqual([])) await vi.waitFor(async () => expect( @@ -19422,6 +19436,7 @@ describe('FactoryLoop PR babysitter', () => { emitSlackReply(mount, slackReplyFixturePath( 'C0FACTORY__factory-e2e', mount.threadTs, 'human-after-terminal-restart', ), 'slack-human-after-terminal-restart', { + ts: '11.000', text: 'There is still unaddressed review feedback.', user: 'U407', user_name: 'human', @@ -19434,6 +19449,7 @@ describe('FactoryLoop PR babysitter', () => { fleet: restartedFleet, triage: new StaticTriage(), stateStore: state(), + clock, }) await restarted.start({ mode: 'dispatch-owner' }) @@ -19443,6 +19459,8 @@ describe('FactoryLoop PR babysitter', () => { expect(slackConversationResumes(restartedFleet)).toEqual([]) expect(restarted.status().counters.slackAnswersUnroutableVisible).toBe(1) expect(restarted.status().counters.slackWatchersRearmed).toBe(1) + expect(slackReplyWrites(mount).filter((write) => + write.content.text?.includes('no longer has an active agent'))).toHaveLength(1) } finally { await first.stop() await restarted?.stop() @@ -19450,6 +19468,52 @@ describe('FactoryLoop PR babysitter', () => { } }) + it('surfaces an acknowledged reply if the work unit terminates during coalescing', async () => { + const issue = issueFile(408) + const mount = new ConfirmRecordingSlackMountClient({ [issuePath(408)]: issue }) + const fleet = new FakeFleetClient() + fleet.setSessionRef('ar-408-impl-pear', 'session-ar-408-impl-pear') + const stateStore = new InMemoryStateStore({ batchSize: 10 }) + const factory = createFactory(config({ + slack: { ...slackConfig(), conversationCoalesceMs: 60_000 }, + }), { + mount, + fleet, + triage: new StaticTriage(), + stateStore, + }) + try { + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(408), issue))) + emitSlackReply(mount, slackReplyFixturePath( + 'C0FACTORY__factory-e2e', mount.threadTs, 'human-during-coalesce', + ), 'slack-human-during-coalesce', { + text: 'Please do not lose this acknowledged instruction.', + user: 'U408', + user_name: 'human', + user_is_bot: false, + }) + await vi.waitFor(async () => expect( + (await stateStore.getConversationSession('factory-test', `slack:${mount.threadTs}`))?.pending, + ).toEqual([expect.objectContaining({ text: 'Please do not lose this acknowledged instruction.' })])) + await vi.waitFor(() => expect(slackReplyWrites(mount).map((write) => write.content.text)).toContain( + slackImplementerReceipt, + )) + + fleet.emitAgentExit('ar-408-impl-pear', 'issue-done') + + await vi.waitFor(() => expect(slackReplyWrites(mount).map((write) => write.content.text)).toContain( + 'Factory could not deliver 1 queued reply because this work unit no longer has an active agent. Please continue on the linked issue or pull request.', + )) + await expect(stateStore.getConversationSession( + 'factory-test', `slack:${mount.threadTs}`, + )).resolves.toBeUndefined() + expect(slackConversationResumes(fleet)).toEqual([]) + expect(factory.status().counters.slackConversationRepliesSurfacedTerminal).toBe(1) + } finally { + await factory.stop() + } + }) + it('does not attach a numeric GitHub issue to a merged PR whose body only contains a test count', async () => { const path = githubIssuePath('AgentWorkforce', 'pear', 52) const issueFile = githubIssueFile(52, { diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index c951ada..37a56bd 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -519,6 +519,8 @@ export class FactoryLoop implements Factory { readonly #slackWatchers = new Map() readonly #slackWatcherStarts = new Map>() readonly #slackTerminalWatchExpiryTimers = new Map>() + readonly #terminalSlackWatchIssues = new Set() + readonly #slackReplyRoutes = new Map>() readonly #slackConversationTurns: CoalescedTaskQueue readonly #slackConversationOwner = `${process.pid}:${randomUUID()}` readonly #githubIssueCommentWatchers = new Map() @@ -1145,6 +1147,7 @@ export class FactoryLoop implements Factory { this.#slackWatchers.clear() for (const timer of this.#slackTerminalWatchExpiryTimers.values()) clearTimeout(timer) this.#slackTerminalWatchExpiryTimers.clear() + this.#terminalSlackWatchIssues.clear() await Promise.all([...this.#githubIssueCommentWatchers.values()].map((watcher) => watcher.stop())) this.#githubIssueCommentWatchers.clear() this.#githubIssueCommentWatchStates.clear() @@ -13979,7 +13982,7 @@ export class FactoryLoop implements Factory { async #watchSlackThread( record: InFlightIssue, threadId: string, - options: { replayConversationReplies?: boolean } = {}, + options: { replayConversationReplies?: boolean; replayAfterMs?: number } = {}, ): Promise { if (!this.#config.slack) { return @@ -14047,6 +14050,13 @@ export class FactoryLoop implements Factory { if (!reply || !reply.isThreadReply || reply.threadTs !== threadId || reply.channelDir !== channelDir) { return } + if ( + allowPreExisting && + options.replayAfterMs !== undefined && + slackMessageReceivedAtMs(reply.messageTs, Number.MAX_SAFE_INTEGER) < options.replayAfterMs + ) { + return + } const replyMessageKey = `${reply.threadTs}:${reply.messageTs}` if (seenReplyMessages.has(replyMessageKey)) { @@ -14180,7 +14190,7 @@ export class FactoryLoop implements Factory { async #rearmSlackWatcher( record: InFlightIssue, threadId: string, - options: { replayConversationReplies?: boolean } = {}, + options: { replayConversationReplies?: boolean; replayAfterMs?: number } = {}, ): Promise { const key = issueKey(record.issue) if (this.#slackWatchers.has(key) || this.#slackWatcherStarts.has(key)) { @@ -14254,10 +14264,15 @@ export class FactoryLoop implements Factory { await this.#state.setSlackThread(this.#workspaceId, key, watch.threadId) const watchRecord = escalationWatchRecord(watch.decision) if (watch.kind === 'terminal-grace') { + this.#terminalSlackWatchIssues.add(key) const conversationId = slackConversationId(watch.threadId) await this.#slackConversationTurns.cancel(conversationId) + await this.#surfaceUndeliveredSlackConversation(watch.threadId) await this.#state.clearConversationSession(this.#workspaceId, conversationId) - await this.#rearmSlackWatcher(watchRecord, watch.threadId, { replayConversationReplies: true }) + await this.#rearmSlackWatcher(watchRecord, watch.threadId, { + replayConversationReplies: true, + replayAfterMs: watch.retiredAtMs, + }) this.#scheduleSlackTerminalWatchExpiry(watch.issue, watch.expiresAtMs) continue } @@ -14435,6 +14450,7 @@ export class FactoryLoop implements Factory { async #stopSlackWatcher(issue: IssueRef): Promise { const key = issueKey(issue) + this.#terminalSlackWatchIssues.delete(key) const expiryTimer = this.#slackTerminalWatchExpiryTimers.get(key) if (expiryTimer) clearTimeout(expiryTimer) this.#slackTerminalWatchExpiryTimers.delete(key) @@ -14461,22 +14477,29 @@ export class FactoryLoop implements Factory { const existingWatch = (await this.#state.listSlackThreadWatches(this.#workspaceId)) .find(([watchKey]) => watchKey === key)?.[1] + const retiredAtMs = existingWatch?.kind === 'terminal-grace' + ? existingWatch.retiredAtMs + : this.#clock.now() const expiresAtMs = existingWatch?.kind === 'terminal-grace' ? existingWatch.expiresAtMs - : this.#clock.now() + SLACK_TERMINAL_THREAD_GRACE_MS + : retiredAtMs + SLACK_TERMINAL_THREAD_GRACE_MS await this.#state.setSlackThreadWatch(this.#workspaceId, key, { kind: 'terminal-grace', issue: { ...record.issue }, decision: structuredClone(record.decision), threadId, + retiredAtMs, expiresAtMs, }) // A terminal thread must never retain a resumable session for an agent that // has already exited. Keep only the exact-thread listener so a late human // reply receives the explicit no-active-agent writeback below. + this.#terminalSlackWatchIssues.add(key) + await this.#slackReplyRoutes.get(key)?.catch(() => undefined) const conversationId = slackConversationId(threadId) await this.#slackConversationTurns.cancel(conversationId) + await this.#surfaceUndeliveredSlackConversation(threadId) await this.#state.clearConversationSession(this.#workspaceId, conversationId) if (!this.#slackWatchers.has(key) && !this.#stopping) { await this.#rearmSlackWatcher(record, threadId) @@ -14485,6 +14508,24 @@ export class FactoryLoop implements Factory { this.#increment('slackTerminalWatchersRetained') } + async #surfaceUndeliveredSlackConversation(threadId: string): Promise { + const session = await this.#state.getConversationSession( + this.#workspaceId, + slackConversationId(threadId), + ) + const pendingCount = session + ? session.pending.length + (session.delivery?.messages.length ?? 0) + : 0 + if (pendingCount === 0) return + if (!this.#slack) throw new Error(`Slack thread ${threadId} cannot surface undelivered replies without writeback`) + const noun = pendingCount === 1 ? 'reply' : 'replies' + await this.#slack.reply( + threadId, + `Factory could not deliver ${pendingCount} queued ${noun} because this work unit no longer has an active agent. Please continue on the linked issue or pull request.`, + ) + this.#increment('slackConversationRepliesSurfacedTerminal') + } + #scheduleSlackTerminalWatchExpiry( issue: IssueRef, expiresAtMs: number, @@ -14586,6 +14627,39 @@ export class FactoryLoop implements Factory { return } + return await this.#routeSlackConversationAnswer(record, reply, text, clarificationKey) + } + + async #routeSlackConversationAnswer( + record: InFlightIssue, + reply: SlackThreadReply, + text: string, + clarificationKey: string, + ): Promise { + const preceding = this.#slackReplyRoutes.get(clarificationKey) + const route = (async () => { + await preceding?.catch(() => undefined) + return await this.#routeSlackConversationAnswerUnlocked(record, reply, text, clarificationKey) + })() + this.#slackReplyRoutes.set(clarificationKey, route) + try { + return await route + } finally { + if (this.#slackReplyRoutes.get(clarificationKey) === route) this.#slackReplyRoutes.delete(clarificationKey) + } + } + + async #routeSlackConversationAnswerUnlocked( + record: InFlightIssue, + reply: SlackThreadReply, + text: string, + clarificationKey: string, + ): Promise { + if (this.#terminalSlackWatchIssues.has(clarificationKey)) { + await this.#writeUnroutableSlackReply(reply.threadTs) + return + } + const conversationId = slackConversationId(reply.threadTs) let conversation = await this.#state.getConversationSession(this.#workspaceId, conversationId) let liveRecord: InFlightIssue | undefined @@ -14679,14 +14753,7 @@ export class FactoryLoop implements Factory { if (isTriageEscalationWatchRecord(record)) { return await this.#handleTriageEscalationSlackAnswer(record, text) } - this.#increment('slackAnswersIgnoredNoInFlight') - if (this.#slack) { - await this.#slack.reply( - reply.threadTs, - 'Factory received this reply but could not route it because this work unit no longer has an active agent. Please continue on the linked issue or pull request.', - ) - this.#increment('slackAnswersUnroutableVisible') - } + await this.#writeUnroutableSlackReply(reply.threadTs) return } this.#increment('slackAnswersIgnoredNoConversationSession') @@ -14699,6 +14766,16 @@ export class FactoryLoop implements Factory { } } + async #writeUnroutableSlackReply(threadId: string): Promise { + this.#increment('slackAnswersIgnoredNoInFlight') + if (!this.#slack) return + await this.#slack.reply( + threadId, + 'Factory received this reply but could not route it because this work unit no longer has an active agent. Please continue on the linked issue or pull request.', + ) + this.#increment('slackAnswersUnroutableVisible') + } + async #wakeWaitingClarification(key: string, waiting: WaitingClarification): Promise { const existing = this.#clarificationWakeInFlight.get(key) if (existing) { diff --git a/src/ports/state.ts b/src/ports/state.ts index 569c87b..4c2a6e2 100644 --- a/src/ports/state.ts +++ b/src/ports/state.ts @@ -163,6 +163,8 @@ export type SlackThreadWatchState = { issue: IssueRef decision: TriageDecision threadId: string + /** Provider-message cutoff preventing historical replies from replaying as terminal. */ + retiredAtMs: number expiresAtMs: number } diff --git a/src/state/file-state-store.test.ts b/src/state/file-state-store.test.ts index ed10c08..1269646 100644 --- a/src/state/file-state-store.test.ts +++ b/src/state/file-state-store.test.ts @@ -606,6 +606,7 @@ describe('FileStateStore', () => { issue: lifecycle.issue, decision: lifecycle.decision, threadId: '1780751612.176222', + retiredAtMs: 1_000, expiresAtMs: 86_401_000, } const first = new FileStateStore({ batchSize: 2, watchStatePath }) From dd9158ddc8e36d1933a12172a648115333b99eb7 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Mon, 17 Aug 2026 15:20:16 +0200 Subject: [PATCH 3/6] fix: migrate terminal Slack watch watermarks --- src/orchestrator/factory.test.ts | 11 ++++++++--- src/orchestrator/factory.ts | 13 +++++++++++-- src/ports/state.ts | 2 +- 3 files changed, 20 insertions(+), 6 deletions(-) diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index f3fb4d4..35bb733 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -19428,9 +19428,12 @@ describe('FactoryLoop PR babysitter', () => { await expectSlackConversationResume(firstFleet, ['This reply was already routed before completion.']) firstFleet.emitAgentExit('ar-407-impl-pear', 'issue-done') await vi.waitFor(() => expect(first.status().inFlight).toEqual([])) - await vi.waitFor(async () => expect( - (await state().listSlackThreadWatches('factory-test'))[0]?.[1], - ).toMatchObject({ kind: 'terminal-grace', threadId: mount.threadTs })) + await vi.waitFor(async () => expect((await state().listSlackThreadWatches('factory-test'))[0]?.[1]) + .toMatchObject({ kind: 'terminal-grace', threadId: mount.threadTs, retiredAtMs: 10_000 })) + const [[watchKey, terminalWatch]] = await state().listSlackThreadWatches('factory-test') + if (terminalWatch?.kind !== 'terminal-grace') throw new Error('expected terminal Slack watch') + const { retiredAtMs: _legacyMissingWatermark, ...legacyTerminalWatch } = terminalWatch + await state().setSlackThreadWatch('factory-test', watchKey, legacyTerminalWatch) await first.stop() emitSlackReply(mount, slackReplyFixturePath( @@ -19459,6 +19462,8 @@ describe('FactoryLoop PR babysitter', () => { expect(slackConversationResumes(restartedFleet)).toEqual([]) expect(restarted.status().counters.slackAnswersUnroutableVisible).toBe(1) expect(restarted.status().counters.slackWatchersRearmed).toBe(1) + expect((await state().listSlackThreadWatches('factory-test'))[0]?.[1]) + .toMatchObject({ kind: 'terminal-grace', retiredAtMs: 10_000 }) expect(slackReplyWrites(mount).filter((write) => write.content.text?.includes('no longer has an active agent'))).toHaveLength(1) } finally { diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 37a56bd..f5ffa66 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -14264,6 +14264,10 @@ export class FactoryLoop implements Factory { await this.#state.setSlackThread(this.#workspaceId, key, watch.threadId) const watchRecord = escalationWatchRecord(watch.decision) if (watch.kind === 'terminal-grace') { + const retiredAtMs = terminalSlackWatchRetiredAtMs(watch) + if (watch.retiredAtMs !== retiredAtMs) { + await this.#state.setSlackThreadWatch(this.#workspaceId, key, { ...watch, retiredAtMs }) + } this.#terminalSlackWatchIssues.add(key) const conversationId = slackConversationId(watch.threadId) await this.#slackConversationTurns.cancel(conversationId) @@ -14271,7 +14275,7 @@ export class FactoryLoop implements Factory { await this.#state.clearConversationSession(this.#workspaceId, conversationId) await this.#rearmSlackWatcher(watchRecord, watch.threadId, { replayConversationReplies: true, - replayAfterMs: watch.retiredAtMs, + replayAfterMs: retiredAtMs, }) this.#scheduleSlackTerminalWatchExpiry(watch.issue, watch.expiresAtMs) continue @@ -14478,7 +14482,7 @@ export class FactoryLoop implements Factory { const existingWatch = (await this.#state.listSlackThreadWatches(this.#workspaceId)) .find(([watchKey]) => watchKey === key)?.[1] const retiredAtMs = existingWatch?.kind === 'terminal-grace' - ? existingWatch.retiredAtMs + ? terminalSlackWatchRetiredAtMs(existingWatch) : this.#clock.now() const expiresAtMs = existingWatch?.kind === 'terminal-grace' ? existingWatch.expiresAtMs @@ -18068,6 +18072,11 @@ const slackMessageReceivedAtMs = (messageTs: string, fallback: number): number = return Number.isFinite(seconds) && seconds > 0 ? Math.floor(seconds * 1_000) : fallback } +const terminalSlackWatchRetiredAtMs = (watch: { retiredAtMs?: number; expiresAtMs: number }): number => + typeof watch.retiredAtMs === 'number' && Number.isFinite(watch.retiredAtMs) + ? watch.retiredAtMs + : Math.max(0, watch.expiresAtMs - SLACK_TERMINAL_THREAD_GRACE_MS) + const eventIdentity = (event: ChangeEvent): string | undefined => { const record = event as unknown as Record const rawId = record.id ?? record.event_id ?? record.seq diff --git a/src/ports/state.ts b/src/ports/state.ts index 4c2a6e2..7bbd33d 100644 --- a/src/ports/state.ts +++ b/src/ports/state.ts @@ -164,7 +164,7 @@ export type SlackThreadWatchState = { decision: TriageDecision threadId: string /** Provider-message cutoff preventing historical replies from replaying as terminal. */ - retiredAtMs: number + retiredAtMs?: number expiresAtMs: number } From 3f047fef9de115393163b7016a9b2d49cc44a507 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Mon, 17 Aug 2026 15:30:54 +0200 Subject: [PATCH 4/6] test: cover Slack document state shape --- src/state/file-state-store.test.ts | 1 + src/state/watch-state-document.test.ts | 2 ++ 2 files changed, 3 insertions(+) diff --git a/src/state/file-state-store.test.ts b/src/state/file-state-store.test.ts index 1269646..de5e24c 100644 --- a/src/state/file-state-store.test.ts +++ b/src/state/file-state-store.test.ts @@ -38,6 +38,7 @@ describe('FileStateStore', () => { workspaces: { 'workspace-1': { githubIssueCommentWatches: {}, + slackThreadWatches: {}, waitingClarifications: {}, babysitterSessions: {}, babysitterGenerations: {}, diff --git a/src/state/watch-state-document.test.ts b/src/state/watch-state-document.test.ts index d9a96ab..748265a 100644 --- a/src/state/watch-state-document.test.ts +++ b/src/state/watch-state-document.test.ts @@ -5,6 +5,7 @@ import { parseWatchStateDocument } from './watch-state-document' describe('parseWatchStateDocument', () => { it.each([ ['GitHub watch', 'githubIssueCommentWatches'], + ['Slack thread watch', 'slackThreadWatches'], ['waiting clarification', 'waitingClarifications'], ['dispatch lifecycle', 'dispatchLifecycles'], ])('rejects a malformed %s record during readiness parsing', (_label, collection) => { @@ -77,6 +78,7 @@ const validDocument = (): Record => ({ }], }, }, + slackThreadWatches: {}, waitingClarifications: { clarification: { issue: issue(), From 4cecf6c9ea794096f78126206b24546e50b83c3b Mon Sep 17 00:00:00 2001 From: Khaliq Date: Mon, 17 Aug 2026 15:38:32 +0200 Subject: [PATCH 5/6] fix: persist legacy Slack conversation recovery --- src/state/file-state-store.test.ts | 55 ++++++++++++++++++++++++++ src/state/file-state-store.ts | 3 +- src/state/watch-state-document.test.ts | 17 ++++++++ src/state/watch-state-document.ts | 6 +-- 4 files changed, 77 insertions(+), 4 deletions(-) diff --git a/src/state/file-state-store.test.ts b/src/state/file-state-store.test.ts index de5e24c..1326ad4 100644 --- a/src/state/file-state-store.test.ts +++ b/src/state/file-state-store.test.ts @@ -475,6 +475,7 @@ describe('FileStateStore', () => { }, history: [], processedMessageIds: [], + acknowledgedMessageIds: [], pending: [], } const first = new FileStateStore({ batchSize: 2, watchStatePath }) @@ -571,6 +572,60 @@ describe('FileStateStore', () => { } }) + it('durably requeues an expired delivery when its conversation has no owner', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-file-state-slack-expired-unowned-')) + try { + const watchStatePath = join(root, 'factory-state.json') + const conversationId = 'slack:1780751612.176223' + const message = { id: 'message-expired', text: 'Keep me pending.', receivedAtMs: 1_000 } + await writeFile(watchStatePath, JSON.stringify({ + version: 3, + workspaces: { + 'workspace-1': { + githubIssueCommentWatches: {}, + slackThreadWatches: {}, + waitingClarifications: {}, + babysitterSessions: {}, + babysitterGenerations: {}, + conversationSessions: { + [conversationId]: { + provider: 'slack', + issue: { uuid: 'uuid-134', key: 'AR-134', path: '/linear/issues/AR-134__uuid-134.json' }, + externalId: '1780751612.176223', + context: { channelDir: 'C0FACTORY__factory-e2e' }, + history: [], + processedMessageIds: [message.id], + pending: [], + delivery: { + claimId: 'expired-claim', + owner: 'stopped-owner', + claimedAtMs: 1_000, + attempts: 1, + messages: [message], + agent: { name: 'ar-134-impl-factory', sessionRef: 'expired-session' }, + }, + }, + }, + dispatchLifecycles: {}, + discoverySweep: { consecutiveOverloads: 0, backoffUntilMs: 0, lastEpoch: 0 }, + }, + }, + })) + + const requeued = await new FileStateStore({ batchSize: 2, watchStatePath }).claimConversationTurn( + 'workspace-1', conversationId, 'replacement-owner', 'replacement-claim', 62_000, 60_000, + ) + expect(requeued).toMatchObject({ pending: [message] }) + expect(requeued?.delivery).toBeUndefined() + const restored = await new FileStateStore({ batchSize: 2, watchStatePath }) + .getConversationSession('workspace-1', conversationId) + expect(restored).toMatchObject({ pending: [message] }) + expect(restored?.delivery).toBeUndefined() + } finally { + await rm(root, { recursive: true, force: true }) + } + }) + it('persists and clears the compact pre-dispatch Slack triage watch', async () => { const root = await mkdtemp(join(tmpdir(), 'factory-file-state-slack-watch-')) try { diff --git a/src/state/file-state-store.ts b/src/state/file-state-store.ts index 33299f9..137e4c9 100644 --- a/src/state/file-state-store.ts +++ b/src/state/file-state-store.ts @@ -1025,8 +1025,9 @@ export class DocumentStateStore extends InMemoryStateStore { if (session.delivery) session.pending.unshift(...session.delivery.messages) session.pending.sort(compareConversationMessages) if (!session.agent || session.pending.length === 0) { + const hadDelivery = session.delivery !== undefined session.delivery = undefined - return false + return hadDelivery } const agent = session.agent session.delivery = { diff --git a/src/state/watch-state-document.test.ts b/src/state/watch-state-document.test.ts index 748265a..bd8d711 100644 --- a/src/state/watch-state-document.test.ts +++ b/src/state/watch-state-document.test.ts @@ -19,6 +19,23 @@ describe('parseWatchStateDocument', () => { expect(parseWatchStateDocument(validDocument())).toEqual(validDocument()) }) + it('migrates legacy conversation history as already acknowledged without acknowledging pending work', () => { + const document = validDocument() + document.workspaces.workspace.conversationSessions.legacy = { + provider: 'slack', + issue: issue(), + externalId: '1780751612.176224', + context: { channelDir: 'factory' }, + agent: { name: 'implementer', sessionRef: 'session-implementer' }, + history: [{ id: 'delivered', text: 'Already delivered.', receivedAtMs: 1_000 }], + processedMessageIds: ['delivered', 'pending'], + pending: [{ id: 'pending', text: 'Still pending.', receivedAtMs: 1_001 }], + } + + expect(parseWatchStateDocument(document).workspaces.workspace?.conversationSessions.legacy) + .toMatchObject({ acknowledgedMessageIds: ['delivered'] }) + }) + it.each([ ['preview reference', (document: Record) => { document.workspaces.workspace.waitingClarifications.clarification.decision.implementers[0].preview = { diff --git a/src/state/watch-state-document.ts b/src/state/watch-state-document.ts index 395306e..15facc7 100644 --- a/src/state/watch-state-document.ts +++ b/src/state/watch-state-document.ts @@ -158,9 +158,9 @@ const parseConversationSessions = ( ...(session.delivery?.messages ?? []), ].map((message) => message.id))] : [...candidate.processedMessageIds as string[]] - if (candidate.acknowledgedMessageIds !== undefined) { - session.acknowledgedMessageIds = [...candidate.acknowledgedMessageIds as string[]] - } + session.acknowledgedMessageIds = candidate.acknowledgedMessageIds === undefined + ? session.history.map((message) => message.id) + : [...candidate.acknowledgedMessageIds as string[]] if (candidate.acknowledgementClaims !== undefined) { session.acknowledgementClaims = structuredClone(candidate.acknowledgementClaims) as ConversationSessionState['acknowledgementClaims'] } From 313063f6de2a818d113306ed7df67cae867b5f29 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Wed, 19 Aug 2026 23:14:33 +0200 Subject: [PATCH 6/6] fix: drain in-flight Slack reply routes before clearing the terminal fence MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reopening a work unit called #stopSlackWatcher, which dropped the terminal fence synchronously. A reply route already in flight for the retired thread — or one chained behind it — then passed the fence check with the fence gone, resolved the live record by issue key, and bound the retired thread to the freshly dispatched work unit, delivering a stale human reply to a new agent. Drain the per-work-unit route chain before clearing the fence, and fail closed: a route that rejects is replayed by the watcher after SLACK_REPLY_ROUTE_RETRY_MS, so the fence stays up and the reopen defers to the next reconcile rather than letting that replay land on the new dispatch. Also stop one undeliverable terminal receipt from aborting watcher rehydration. #surfaceUndeliveredSlackConversation needs Slack writeback; when it was down at startup the throw escaped #rearmSlackReplyWatchers and left every remaining thread watched by nobody. Treat it as retryable per-thread maintenance, keep the queued replies rather than clearing replies nobody was told about, and continue. Co-Authored-By: Claude Opus 5 Session-Id: 5f4a448f-5d6a-4187-856f-6dbf5647562b --- src/orchestrator/factory.test.ts | 223 +++++++++++++++++++++++++++++++ src/orchestrator/factory.ts | 61 ++++++++- 2 files changed, 280 insertions(+), 4 deletions(-) diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 35bb733..ea8a785 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -1659,6 +1659,11 @@ class HumanReplyDuringQuestionMountClient extends CloudWritebackFakeMountClient } } +const slackWriteText = (content: unknown): string => + typeof content === 'object' && content !== null && 'text' in content + ? String((content as { text?: unknown }).text ?? '') + : '' + class ConfirmRecordingSlackMountClient extends CloudWritebackFakeMountClient { readonly confirmedPaths: string[] = [] @@ -1723,6 +1728,51 @@ class FailNextSlackReplyMountClient extends CloudWritebackFakeMountClient { } } +/** + * Parks the first "no active agent" writeback so a Slack reply route can be held + * mid-flight while the work unit reopens underneath it. + */ +class BlockingUnroutableReplyMountClient extends ConfirmRecordingSlackMountClient { + readonly unroutableWriteStarted: Promise + #signalUnroutableWriteStarted!: () => void + #releaseUnroutableWrite!: () => void + readonly #unroutableWriteReleased: Promise + #blocked = false + + constructor(initialFiles: Record = {}) { + super(initialFiles) + this.unroutableWriteStarted = new Promise((resolve) => { this.#signalUnroutableWriteStarted = resolve }) + this.#unroutableWriteReleased = new Promise((resolve) => { this.#releaseUnroutableWrite = resolve }) + } + + releaseUnroutableWrite(): void { + this.#releaseUnroutableWrite() + } + + override async writeFile(path: string, content: unknown, opts?: { guarded?: boolean }): Promise { + if (!this.#blocked && slackWriteText(content).startsWith('Factory received this reply but could not route it')) { + this.#blocked = true + this.#signalUnroutableWriteStarted() + await this.#unroutableWriteReleased + } + await super.writeFile(path, content, opts) + } +} + +/** Fails every undelivered-reply receipt, standing in for Slack writeback being down. */ +class FailingUndeliveredReceiptMountClient extends ConfirmRecordingSlackMountClient { + failReceipts = true + receiptAttempts = 0 + + override async writeFile(path: string, content: unknown, opts?: { guarded?: boolean }): Promise { + if (this.failReceipts && slackWriteText(content).startsWith('Factory could not deliver')) { + this.receiptAttempts += 1 + throw new Error('Slack writeback is unavailable') + } + await super.writeFile(path, content, opts) + } +} + class FailingGithubCommentReconciliationMountClient extends FailNextSlackReplyMountClient { failGithubCommentReconciliation: false | 'list' | 'read' | 'issue' = false @@ -15950,6 +16000,94 @@ describe('FactoryLoop', () => { expect(factory.status().counters.slackAnswersUnroutableVisible).toBe(1) }) + it('drains an in-flight Slack reply route before a reopen clears the terminal fence', async () => { + const mount = new BlockingUnroutableReplyMountClient({ [issuePath(414)]: issueFile(414) }) + const fleet = new RemoteLifecycleFleetClient() + const stateStore = new InMemoryStateStore({ batchSize: 10 }) + const factory = createFactory(config({ slack: slackConfig() }), { + mount, + fleet, + triage: new StaticTriage(), + stateStore, + }) + const retiredThreadTs = mount.threadTs + const staleText = 'stale reply that must not reach the reopened work unit' + + try { + const first = await factory.runOnce() + expect(first.dispatched.map((result) => result.issue.key)).toEqual(['AR-414']) + + fleet.emitAgentExit('ar-414-impl-pear', 'issue-done') + await vi.waitFor(() => expect(factory.status().inFlight).toEqual([])) + await vi.waitFor(async () => expect( + (await stateStore.listSlackThreadWatches('factory-test'))[0]?.[1], + ).toMatchObject({ kind: 'terminal-grace', threadId: retiredThreadTs })) + await vi.waitFor(() => expect(factory.status().counters.slackTerminalWatchersRetained).toBe(1)) + + // The first late reply parks inside the "no active agent" writeback, so the + // route chain for this work unit stays open. + emitSlackReply(mount, slackReplyFixturePath( + 'C0FACTORY__factory-e2e', retiredThreadTs, 'human-holds-route', + ), 'slack-human-holds-route', { + text: 'first late reply', + user: 'U414', + user_is_bot: false, + }) + await mount.unroutableWriteStarted + + // The second reply queues behind it and is therefore still in flight at the + // exact moment the work unit reopens. + const stalePath = slackReplyFixturePath('C0FACTORY__factory-e2e', retiredThreadTs, 'human-stale') + emitSlackReply(mount, stalePath, 'slack-human-stale', { + text: staleText, + user: 'U414', + user_is_bot: false, + }) + await vi.waitFor(() => expect(mount.reads).toContain(stalePath)) + await flush() + await flush() + await flush() + + // Reopen with both routes undrained. The reopened dispatch spawns before it + // touches the Slack fence, so waiting on the second generation of agents + // lands us at the boundary in both the fenced and unfenced code paths. + await mount.writeFile(issuePath(414), issuePayload(414, ready)) + const reopening = factory.runOnce() + await vi.waitFor(() => expect(fleet.spawns).toHaveLength(4)) + for (let tick = 0; tick < 20; tick += 1) await flush() + mount.releaseUnroutableWrite() + const reopened = await reopening + expect(reopened.dispatched.map((result) => result.issue.key)).toEqual(['AR-414']) + + // Wait for the second route to settle either way: fenced (unroutable) or + // fallen through onto the reopened dispatch (queued). + await vi.waitFor(() => expect( + (factory.status().counters.slackAnswersUnroutableVisible ?? 0) + + (factory.status().counters.slackConversationRepliesQueued ?? 0), + ).toBe(2)) + + // Both replies belong to the retired thread and must be answered as + // unroutable, never queued onto the fresh dispatch. + expect(factory.status().counters.slackConversationRepliesQueued ?? 0).toBe(0) + expect(factory.status().counters.slackAnswersUnroutableVisible).toBe(2) + const conversation = await stateStore.getConversationSession( + 'factory-test', `slack:${retiredThreadTs}`, + ) + const carried = [ + ...(conversation?.pending ?? []), + ...(conversation?.history ?? []), + ...(conversation?.delivery?.messages ?? []), + ].map((message) => message.text) + expect(carried).not.toContain(staleText) + expect(slackConversationResumes(fleet)).toEqual([]) + expect(slackAnswerInputs(fleet).map((input) => input.data).join('\n')).not.toContain(staleText) + expect(fleet.spawns.map((spawn) => spawn.task ?? '').join('\n')).not.toContain(staleText) + } finally { + mount.releaseUnroutableWrite() + await factory.stop() + } + }) + it('does not wire Slack answer injection when Slack is unconfigured', async () => { const mount = new CloudWritebackFakeMountClient({ [issuePath(22)]: issueFile(22) }) const fleet = new FakeFleetClient() @@ -19473,6 +19611,91 @@ describe('FactoryLoop PR babysitter', () => { } }) + it('keeps rehydrating Slack watchers when one terminal receipt cannot be written', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-terminal-receipt-rearm-')) + const watchStatePath = join(root, 'factory-state.json') + const issue = issueFile(409) + const mount = new FailingUndeliveredReceiptMountClient({ [issuePath(409)]: issue }) + const factoryConfig = config({ + slack: { ...slackConfig(), conversationCoalesceMs: 60_000 }, + }) + const state = () => new FileStateStore({ batchSize: 10, watchStatePath }) + const clock = new ManualClock() + clock.advance(10_000) + const firstFleet = new FakeFleetClient() + firstFleet.setSessionRef('ar-409-impl-pear', 'session-ar-409-impl-pear') + const first = createFactory(factoryConfig, { + mount, + fleet: firstFleet, + triage: new StaticTriage(), + stateStore: state(), + clock, + }) + let restarted: ReturnType | undefined + try { + await first.dispatch(await first.triageIssue(parseLinearIssue(issuePath(409), issue))) + emitSlackReply(mount, slackReplyFixturePath( + 'C0FACTORY__factory-e2e', mount.threadTs, 'human-undelivered', + ), 'slack-human-undelivered', { + text: 'This reply was never delivered to an agent.', + user: 'U409', + user_is_bot: false, + }) + await vi.waitFor(async () => expect( + (await state().getConversationSession('factory-test', `slack:${mount.threadTs}`))?.pending, + ).toEqual([expect.objectContaining({ text: 'This reply was never delivered to an agent.' })])) + + // Terminating with writeback down persists the terminal-grace watch and + // leaves the reply queued: the receipt is what fails, not the state write. + firstFleet.emitAgentExit('ar-409-impl-pear', 'issue-done') + await vi.waitFor(async () => expect( + (await state().listSlackThreadWatches('factory-test'))[0]?.[1], + ).toMatchObject({ kind: 'terminal-grace', threadId: mount.threadTs })) + await vi.waitFor(() => expect(mount.receiptAttempts).toBeGreaterThanOrEqual(1)) + await first.stop() + + // A second, independent thread whose watcher must survive the first one's + // receipt failure. + const [[, terminalWatch]] = await state().listSlackThreadWatches('factory-test') + if (terminalWatch?.kind !== 'terminal-grace') throw new Error('expected terminal Slack watch') + const siblingThreadTs = '1780751612.409409' + const siblingIssue = { uuid: 'uuid-410', key: 'AR-410', path: issuePath(410) } + await state().setSlackThreadWatch('factory-test', 'AR-410', { + ...terminalWatch, + issue: siblingIssue, + // escalationWatchRecord() rebuilds the watched record from the decision, + // so the sibling needs its own issue there too. + decision: { ...terminalWatch.decision, issue: siblingIssue }, + threadId: siblingThreadTs, + }) + + const restartedFleet = new FakeFleetClient() + restarted = createFactory(factoryConfig, { + mount, + fleet: restartedFleet, + triage: new StaticTriage(), + stateStore: state(), + clock, + }) + await restarted.start({ mode: 'dispatch-owner' }) + + // The undeliverable receipt is retryable maintenance for its own thread and + // must not take the sibling watcher down with it. + await vi.waitFor(() => expect(restarted?.status().counters.slackWatchersRearmed).toBe(2)) + expect(restarted.status().counters.slackTerminalWatchReceiptsDeferred).toBe(1) + // The replies nobody was told about are still queued for a later retry. + expect((await state().getConversationSession( + 'factory-test', `slack:${mount.threadTs}`, + ))?.pending).toEqual([ + expect.objectContaining({ text: 'This reply was never delivered to an agent.' }), + ]) + } finally { + await first.stop() + await restarted?.stop() + await rm(root, { recursive: true, force: true }) + } + }) + it('surfaces an acknowledged reply if the work unit terminates during coalescing', async () => { const issue = issueFile(408) const mount = new ConfirmRecordingSlackMountClient({ [issuePath(408)]: issue }) diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index f5ffa66..8c9fb8d 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -13303,7 +13303,17 @@ export class FactoryLoop implements Factory { // A reopened work unit needs a fresh dispatch notification and a fresh // conversation. Do not let the old grace-period watcher (or its expiry // timer) capture and later tear down the new dispatch. - await this.#stopSlackWatcher(record.issue) + if (!await this.#stopSlackWatcher(record.issue)) { + // Fail closed. An undrained reply route still holds the retired thread + // and would bind it to this dispatch, delivering a stale human reply to + // fresh work. Leave the fence up; the next reconcile retries the drain. + this.#logger.warn?.( + '[factory] deferring Slack dispatch thread for reopened work unit; in-flight reply route not drained', + { issue: record.issue.key }, + ) + this.#increment('slackDispatchThreadsDeferredUndrainedReply') + return + } } const existingThread = await this.#persistedSlackThread(key) const watcherStart = this.#slackWatcherStarts.get(key) @@ -14271,8 +14281,22 @@ export class FactoryLoop implements Factory { this.#terminalSlackWatchIssues.add(key) const conversationId = slackConversationId(watch.threadId) await this.#slackConversationTurns.cancel(conversationId) - await this.#surfaceUndeliveredSlackConversation(watch.threadId) - await this.#state.clearConversationSession(this.#workspaceId, conversationId) + try { + await this.#surfaceUndeliveredSlackConversation(watch.threadId) + await this.#state.clearConversationSession(this.#workspaceId, conversationId) + } catch (error) { + // The undelivered-reply receipt needs Slack writeback, which may be + // unavailable at startup. That is retryable state maintenance for this + // one thread, not a reason to abandon rehydration: aborting here would + // leave every remaining thread watched by nobody. Keep the queued + // replies (clearing them now would drop replies nobody was told about) + // and carry on re-arming. + this.#logger.warn?.( + '[factory] failed to settle undelivered Slack replies for terminal watch; will retry', + { issue: watch.issue.key, error }, + ) + this.#increment('slackTerminalWatchReceiptsDeferred') + } await this.#rearmSlackWatcher(watchRecord, watch.threadId, { replayConversationReplies: true, replayAfterMs: retiredAtMs, @@ -14452,8 +14476,36 @@ export class FactoryLoop implements Factory { this.#clarificationSweepDueAtMs = dueAtMs } - async #stopSlackWatcher(issue: IssueRef): Promise { + // The terminal fence is the only thing that makes an in-flight reply route + // answer "no active agent" instead of binding the retired thread to whatever + // dispatch owns this key. Routes are chained per work unit, so awaiting the + // newest one drains every reply queued behind it. A route that *rejects* is + // not drained: the watcher replays it after SLACK_REPLY_ROUTE_RETRY_MS, and + // that replay would land on the next dispatch. Fail closed and let the caller + // keep the fence up rather than leak a stale human reply onto fresh work. + async #drainSlackReplyRoutes(key: string): Promise { + const route = this.#slackReplyRoutes.get(key) + if (!route) return true + try { + await route + return true + } catch (error) { + this.#logger.warn?.( + '[factory] in-flight Slack reply route did not drain; keeping terminal Slack fence', + { issue: key, error }, + ) + this.#increment('slackReplyRouteDrainsFailed') + return false + } + } + + async #stopSlackWatcher(issue: IssueRef): Promise { const key = issueKey(issue) + // Drain before clearing the fence. Clearing it first lets a reply that is + // already mid-route — or one queued behind it — fall through the fence check + // in #routeSlackConversationAnswerUnlocked and rebind the retired thread to + // the next dispatch of this work unit. + if (!await this.#drainSlackReplyRoutes(key)) return false this.#terminalSlackWatchIssues.delete(key) const expiryTimer = this.#slackTerminalWatchExpiryTimers.get(key) if (expiryTimer) clearTimeout(expiryTimer) @@ -14469,6 +14521,7 @@ export class FactoryLoop implements Factory { } await this.#state.clearSlackThread(this.#workspaceId, key) await this.#state.clearSlackThreadWatch(this.#workspaceId, key) + return true } async #retireSlackWatcher(record: InFlightIssue): Promise {