diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 3da49a69..e287d7f7 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,246 @@ 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) + } +} + +/** + * Parks the *second* late reply inside its mount read so the route it registers + * lands in the middle of the terminal-fence drain rather than before it. The + * first reply still parks in its writeback, which is what holds the drain open + * long enough for the straddling registration to happen. + */ +class DrainStraddlingReplyMountClient extends BlockingUnroutableReplyMountClient { + readonly straddlingReadStarted: Promise + #signalStraddlingReadStarted!: () => void + #releaseStraddlingRead!: () => void + readonly #straddlingReadReleased: Promise + #blockedReadPath: string | undefined + + constructor(initialFiles: Record = {}) { + super(initialFiles) + this.straddlingReadStarted = new Promise((resolve) => { this.#signalStraddlingReadStarted = resolve }) + this.#straddlingReadReleased = new Promise((resolve) => { this.#releaseStraddlingRead = resolve }) + } + + blockReadsFor(path: string): void { + this.#blockedReadPath = path + } + + releaseStraddlingRead(): void { + this.#releaseStraddlingRead() + } + + override async readFile(path: string): Promise<{ content: unknown; revision?: string }> { + if (path === this.#blockedReadPath) { + this.#signalStraddlingReadStarted() + await this.#straddlingReadReleased + } + return await super.readFile(path) + } +} + +/** + * The same drain straddle as DrainStraddlingReplyMountClient, except the receipt + * the fence itself writes for the mid-drain reply fails. The first "no active + * agent" write still parks so the drain stays open; every later one throws until + * `failFencedReceipt` is cleared, standing in for Slack writeback being down for + * exactly the write nothing is waiting on. + */ +class FailingFencedReceiptMountClient extends ConfirmRecordingSlackMountClient { + readonly unroutableWriteStarted: Promise + readonly straddlingReadStarted: Promise + failFencedReceipt = true + fencedReceiptAttempts = 0 + #signalUnroutableWriteStarted!: () => void + #releaseUnroutableWrite!: () => void + readonly #unroutableWriteReleased: Promise + #signalStraddlingReadStarted!: () => void + #releaseStraddlingRead!: () => void + readonly #straddlingReadReleased: Promise + #parked = false + #blockedReadPath: string | undefined + + constructor(initialFiles: Record = {}) { + super(initialFiles) + this.unroutableWriteStarted = new Promise((resolve) => { this.#signalUnroutableWriteStarted = resolve }) + this.#unroutableWriteReleased = new Promise((resolve) => { this.#releaseUnroutableWrite = resolve }) + this.straddlingReadStarted = new Promise((resolve) => { this.#signalStraddlingReadStarted = resolve }) + this.#straddlingReadReleased = new Promise((resolve) => { this.#releaseStraddlingRead = resolve }) + } + + blockReadsFor(path: string): void { + this.#blockedReadPath = path + } + + releaseStraddlingRead(): void { + this.#releaseStraddlingRead() + } + + releaseUnroutableWrite(): void { + this.#releaseUnroutableWrite() + } + + override async readFile(path: string): Promise<{ content: unknown; revision?: string }> { + if (path === this.#blockedReadPath) { + this.#signalStraddlingReadStarted() + await this.#straddlingReadReleased + } + return await super.readFile(path) + } + + override async writeFile(path: string, content: unknown, opts?: { guarded?: boolean }): Promise { + if (slackWriteText(content).startsWith('Factory received this reply but could not route it')) { + if (!this.#parked) { + this.#parked = true + this.#signalUnroutableWriteStarted() + await this.#unroutableWriteReleased + } else if (this.failFencedReceipt) { + this.fencedReceiptAttempts += 1 + throw new Error('Slack writeback is unavailable') + } + } + await super.writeFile(path, content, opts) + } +} + +/** Fails the durable clear that follows an accepted terminal Slack receipt. */ +class FailingConversationClearStateStore extends InMemoryStateStore { + failClears = 1 + clearAttempts = 0 + + override async clearConversationSession(workspaceId: string, conversationId: string): Promise { + this.clearAttempts += 1 + if (this.failClears > 0) { + this.failClears -= 1 + throw new Error('conversation state write is unavailable') + } + await super.clearConversationSession(workspaceId, conversationId) + } +} + +/** Parks the terminal undelivered-replies receipt inside its provider write. */ +class ParkedTerminalReceiptMountClient extends ConfirmRecordingSlackMountClient { + receiptWriteEntered = false + #releaseReceiptWrite!: () => void + readonly #receiptWriteReleased: Promise + + constructor(initialFiles: Record = {}) { + super(initialFiles) + this.#receiptWriteReleased = new Promise((resolve) => { this.#releaseReceiptWrite = resolve }) + } + + releaseReceiptWrite(): void { + this.#releaseReceiptWrite() + } + + override async writeFile(path: string, content: unknown, opts?: { guarded?: boolean }): Promise { + if (!this.receiptWriteEntered && slackWriteText(content).startsWith('Factory could not deliver')) { + this.receiptWriteEntered = true + await this.#receiptWriteReleased + } + await super.writeFile(path, content, opts) + } +} + +/** Parks the per-reply "durably queued" receipt inside its provider write. */ +class ParkedReplyReceiptMountClient extends ConfirmRecordingSlackMountClient { + receiptWriteEntered = false + #releaseReceiptWrite!: () => void + readonly #receiptWriteReleased: Promise + + constructor(initialFiles: Record = {}) { + super(initialFiles) + this.#receiptWriteReleased = new Promise((resolve) => { this.#releaseReceiptWrite = resolve }) + } + + releaseReceiptWrite(): void { + this.#releaseReceiptWrite() + } + + override async writeFile(path: string, content: unknown, opts?: { guarded?: boolean }): Promise { + if (!this.receiptWriteEntered && slackWriteText(content).startsWith('Factory received this reply and durably queued it')) { + this.receiptWriteEntered = true + await this.#receiptWriteReleased + } + await super.writeFile(path, content, opts) + } +} + +/** Holds the terminal "unroutable" writeback open, then rejects it on release. */ +class RejectingUnroutableReplyMountClient extends ConfirmRecordingSlackMountClient { + readonly unroutableWriteStarted: Promise + #signalUnroutableWriteStarted!: () => void + #releaseUnroutableWrite!: () => void + readonly #unroutableWriteReleased: Promise + #blocked = false + failOnRelease = true + + 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 (slackWriteText(content).startsWith('Factory received this reply but could not route it')) { + if (!this.#blocked) { + this.#blocked = true + this.#signalUnroutableWriteStarted() + await this.#unroutableWriteReleased + } + if (this.failOnRelease) throw new Error('Slack writeback is unavailable') + } + await super.writeFile(path, content, opts) + } +} + class FailingGithubCommentReconciliationMountClient extends FailNextSlackReplyMountClient { failGithubCommentReconciliation: false | 'list' | 'read' | 'issue' = false @@ -15274,6 +15519,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,32 +16160,317 @@ 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('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('fences a Slack reply route that registers while the terminal drain is running', async () => { + const mount = new DrainStraddlingReplyMountClient({ [issuePath(415)]: issueFile(415) }) + 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 = 'reply that registered mid-drain and must not reach the reopened work unit' + + try { + const first = await factory.runOnce() + expect(first.dispatched.map((result) => result.issue.key)).toEqual(['AR-415']) + + fleet.emitAgentExit('ar-415-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)) + + // Reply 1 parks inside the "no active agent" writeback. Its route is the + // one the drain snapshots, and holding it open keeps the drain running. + emitSlackReply(mount, slackReplyFixturePath( + 'C0FACTORY__factory-e2e', retiredThreadTs, 'human-holds-route', + ), 'slack-human-holds-route', { + text: 'first late reply', + user: 'U415', + user_is_bot: false, + }) + await mount.unroutableWriteStarted + + // Reply 2 parks inside its mount read, BEFORE it has registered any route. + // It is therefore invisible to the drain's snapshot: this is the straddle. + const stalePath = slackReplyFixturePath('C0FACTORY__factory-e2e', retiredThreadTs, 'human-mid-drain') + mount.blockReadsFor(stalePath) + emitSlackReply(mount, stalePath, 'slack-human-mid-drain', { + text: staleText, + user: 'U415', + user_is_bot: false, + }) + await mount.straddlingReadStarted + + // Reopen. The reopened dispatch spawns before it touches the Slack fence, + // so four spawns means #stopSlackWatcher is now inside the drain. + await mount.writeFile(issuePath(415), issuePayload(415, ready)) + const reopening = factory.runOnce() + await vi.waitFor(() => expect(fleet.spawns).toHaveLength(4)) + for (let tick = 0; tick < 20; tick += 1) await flush() + + // Release reply 2 *while the drain is still awaiting reply 1's route*, so + // its route registration happens strictly between the drain's snapshot and + // the fence being cleared. + mount.releaseStraddlingRead() + 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-415']) + + for (let tick = 0; tick < 40; tick += 1) await flush() + + // Both replies belong to the retired thread, so both must get the "no + // active agent" writeback. A mid-drain route that falls through the + // cleared fence instead reaches #routeSlackConversationAnswerUnlocked and + // rebinds the retired thread to the reopened work unit. + await vi.waitFor(() => expect(factory.status().counters.slackAnswersUnroutableVisible).toBe(2)) + expect(factory.status().counters.slackConversationRepliesQueued ?? 0).toBe(0) + // ...and it must be the *drain* that fenced it, not the ordinary terminal + // fence: this counter is what proves the route registered mid-drain. + expect(factory.status().counters.slackReplyRoutesFencedDuringDrain).toBe(1) + 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.releaseStraddlingRead() + mount.releaseUnroutableWrite() + await factory.stop() + } }) + it('keeps the terminal drain waiting on the receipt the fence itself writes', async () => { + const mount = new FailingFencedReceiptMountClient({ [issuePath(416)]: issueFile(416) }) + 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 = 'reply whose fence receipt fails while the drain is still running' + + try { + const first = await factory.runOnce() + expect(first.dispatched.map((result) => result.issue.key)).toEqual(['AR-416']) + + fleet.emitAgentExit('ar-416-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)) + + // Reply 1 parks inside its "no active agent" writeback. Its route is the + // one the drain snapshots, and holding it open keeps the drain running. + emitSlackReply(mount, slackReplyFixturePath( + 'C0FACTORY__factory-e2e', retiredThreadTs, 'human-holds-route', + ), 'slack-human-holds-route', { + text: 'first late reply', + user: 'U416', + user_is_bot: false, + }) + await mount.unroutableWriteStarted + + // Reply 2 parks inside its mount read, before it has registered anything, + // so it reaches the fence-during-drain branch rather than the fence. + const stalePath = slackReplyFixturePath('C0FACTORY__factory-e2e', retiredThreadTs, 'human-mid-drain') + mount.blockReadsFor(stalePath) + emitSlackReply(mount, stalePath, 'slack-human-mid-drain', { + text: staleText, + user: 'U416', + user_is_bot: false, + }) + await mount.straddlingReadStarted + + await mount.writeFile(issuePath(416), issuePayload(416, ready)) + const reopening = factory.runOnce() + await vi.waitFor(() => expect(fleet.spawns).toHaveLength(4)) + for (let tick = 0; tick < 20; tick += 1) await flush() + + mount.releaseStraddlingRead() + 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-416']) + for (let tick = 0; tick < 40; tick += 1) await flush() + + // The receipt the fence writes is this work unit's own side effect, so the + // drain has to wait on it. It fails, so the drain fails closed and the + // watcher that owns this reply's retry timer is never torn down. + await vi.waitFor(() => expect(mount.fencedReceiptAttempts).toBeGreaterThanOrEqual(1)) + await vi.waitFor(() => expect( + factory.status().counters.slackReplyRouteDrainsFailed ?? 0, + ).toBeGreaterThanOrEqual(1)) + expect(factory.status().counters.slackDispatchThreadsDeferredUndrainedReply ?? 0) + .toBeGreaterThanOrEqual(1) + expect((await stateStore.listSlackThreadWatches('factory-test'))[0]?.[1]) + .toMatchObject({ kind: 'terminal-grace', threadId: retiredThreadTs }) + + // ...so once writeback recovers the human is still told, instead of the + // reply dying with a retry timer the stop() cleared out from under it. + mount.failFencedReceipt = false + await vi.waitFor( + () => expect(factory.status().counters.slackAnswersUnroutableVisible).toBe(2), + { timeout: 15_000, interval: 25 }, + ) + expect(factory.status().counters.slackConversationRepliesQueued ?? 0).toBe(0) + 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([]) + } finally { + mount.failFencedReceipt = false + mount.releaseStraddlingRead() + mount.releaseUnroutableWrite() + await factory.stop() + } + }, 30_000) + it('does not wire Slack answer injection when Slack is unconfigured', async () => { const mount = new CloudWritebackFakeMountClient({ [issuePath(22)]: issueFile(22) }) const fleet = new FakeFleetClient() @@ -15953,8 +16532,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 +18407,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 +18493,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 +18585,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 +18799,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,10 +18828,10 @@ describe('FactoryLoop', () => { expect(stateStore.failuresRemaining).toBe(0) }) - 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) }) + 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-42-impl-pear', 'session-ar-42-impl-pear') + fleet.setSessionRef('ar-44-impl-pear', 'session-ar-44-impl-pear') const slack = new RecordingSlack() const factory = createFactory(config({ slack: slackConfig() }), { mount, @@ -18257,19 +18839,53 @@ describe('FactoryLoop', () => { triage: new StaticTriage(), slack, }) - const messageTs = '1780751642.000001' - const path = slackTopLevelMessageFixturePath('C0FACTORY__factory-e2e', messageTs) - await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(42), issueFile(42)))) - mount.files.set(path, { - content: { - provider: 'slack', - objectType: 'message', - objectId: 'slack-human-reread', - payload: { - channel: 'C0FACTORY', - thread_ts: slack.threadId, - ts: messageTs, + 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.']) + // The unroutable notice also contains 'received', so a substring match here + // would pass for the wrong artifact. Only the exact receipt proves the retry + // re-sent the acknowledgement it was retrying. + await vi.waitFor(() => expect(slackReplyWrites(mount).filter((write) => + write.content.text === slackImplementerReceipt, + ).length).toBeGreaterThanOrEqual(2), { timeout: 4_000 }) + expect(mount.failedReplies).toBe(1) + expect(slackReplyWrites(mount).at(-1)?.content).toMatchObject({ + thread_ts: slack.threadId, + text: slackImplementerReceipt, + }) + }) + + 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() + fleet.setSessionRef('ar-42-impl-pear', 'session-ar-42-impl-pear') + const slack = new RecordingSlack() + const factory = createFactory(config({ slack: slackConfig() }), { + mount, + fleet, + triage: new StaticTriage(), + slack, + }) + const messageTs = '1780751642.000001' + const path = slackTopLevelMessageFixturePath('C0FACTORY__factory-e2e', messageTs) + + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(42), issueFile(42)))) + mount.files.set(path, { + content: { + provider: 'slack', + objectType: 'message', + objectId: 'slack-human-reread', + payload: { + channel: 'C0FACTORY', + thread_ts: slack.threadId, + ts: messageTs, text: 'status?', user: 'U123', user_is_bot: false, @@ -18279,7 +18895,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 +18994,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 +19041,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 +19778,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 +19802,903 @@ 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 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((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( + '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', + user_is_bot: false, + }) + + const restartedFleet = new FakeFleetClient() + restarted = createFactory(factoryConfig, { + mount, + fleet: restartedFleet, + triage: new StaticTriage(), + stateStore: state(), + clock, + }) + 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) + 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 { + await first.stop() + await restarted?.stop() + await rm(root, { recursive: true, force: true }) + } + }) + + 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('reschedules terminal Slack expiry when the watcher cleanup fails to drain', async () => { + const mount = new RejectingUnroutableReplyMountClient() + const fleet = new FakeFleetClient() + const stateStore = new InMemoryStateStore({ batchSize: 10 }) + const clock = new ManualClock() + clock.advance(10_000) + const decision = await new StaticTriage().triage(parseLinearIssue(issuePath(412), issueFile(412))) + const terminalIssue = { uuid: 'uuid-412', key: 'AR-412', path: issuePath(412) } + await stateStore.setSlackThreadWatch('factory-test', issueKey(terminalIssue), { + kind: 'terminal-grace', + issue: terminalIssue, + decision, + threadId: mount.threadTs, + retiredAtMs: clock.now(), + expiresAtMs: clock.now() + 5_000, + }) + + vi.useFakeTimers() + const factory = createFactory(config({ slack: slackConfig() }), { + mount, + fleet, + triage: new StaticTriage(), + stateStore, + clock, + }) + try { + await factory.start({ mode: 'dispatch-owner' }) + await vi.advanceTimersByTimeAsync(1) + + // Park a late reply inside the terminal "no active agent" writeback so the + // grace period expires with a reply route still in flight. + emitSlackReply(mount, slackReplyFixturePath( + 'C0FACTORY__factory-e2e', mount.threadTs, 'human-at-expiry', + ), 'slack-human-at-expiry', { + text: 'late reply held open across the grace expiry', + user: 'U412', + user_is_bot: false, + }) + await vi.advanceTimersByTimeAsync(1) + await mount.unroutableWriteStarted + + // The grace window closes while that route is still open. + clock.advance(6_000) + await vi.advanceTimersByTimeAsync(5_000) + + // The route then rejects, so #stopSlackWatcher fails closed and keeps the + // fence up. Expiry must not book that as a completed expiration. + mount.releaseUnroutableWrite() + await vi.advanceTimersByTimeAsync(1) + await vi.waitFor(() => expect(factory.status().counters.slackReplyRouteDrainsFailed).toBe(1)) + expect(factory.status().counters.slackTerminalWatchersExpired).toBeUndefined() + expect(factory.status().counters.slackTerminalWatchExpiriesDeferred).toBe(1) + expect((await stateStore.listSlackThreadWatches('factory-test'))[0]?.[1]) + .toMatchObject({ kind: 'terminal-grace', threadId: mount.threadTs }) + + // A deferred expiry has to come back. Once writeback recovers, the retried + // cleanup drains and only then does the watch actually retire. + mount.failOnRelease = false + await vi.advanceTimersByTimeAsync(5_000) + await vi.waitFor(() => expect(factory.status().counters.slackTerminalWatchersExpired).toBe(1)) + expect(await stateStore.listSlackThreadWatches('factory-test')).toEqual([]) + } finally { + mount.failOnRelease = false + mount.releaseUnroutableWrite() + await factory.stop() + vi.useRealTimers() + } + }) + + it('retries a deferred terminal Slack receipt while the grace watch is still alive', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-terminal-receipt-retry-')) + const watchStatePath = join(root, 'factory-state.json') + const issue = issueFile(411) + const mount = new FailingUndeliveredReceiptMountClient({ [issuePath(411)]: 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-411-impl-pear', 'session-ar-411-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(411), issue))) + emitSlackReply(mount, slackReplyFixturePath( + 'C0FACTORY__factory-e2e', mount.threadTs, 'human-undelivered-retry', + ), 'slack-human-undelivered-retry', { + text: 'Nobody has told me this reply went nowhere.', + user: 'U411', + user_is_bot: false, + }) + await vi.waitFor(async () => expect( + (await state().getConversationSession('factory-test', `slack:${mount.threadTs}`))?.pending, + ).toEqual([expect.objectContaining({ text: 'Nobody has told me this reply went nowhere.' })])) + + firstFleet.emitAgentExit('ar-411-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() + + const restartedFleet = new FakeFleetClient() + restarted = createFactory(factoryConfig, { + mount, + fleet: restartedFleet, + triage: new StaticTriage(), + stateStore: state(), + clock, + }) + await restarted.start({ mode: 'dispatch-owner' }) + await vi.waitFor(() => expect( + restarted?.status().counters.slackTerminalWatchReceiptsDeferred, + ).toBe(1)) + expect((await state().getConversationSession( + 'factory-test', `slack:${mount.threadTs}`, + ))?.pending).toHaveLength(1) + + // Writeback comes back inside the grace window. Settling the receipt is + // maintenance this daemon owns: leaving it for the next restart strands + // the queued reply behind a human who was never told it went nowhere. + mount.failReceipts = false + await vi.waitFor(() => expect( + restarted?.status().counters.slackTerminalWatchReceiptsRecovered, + ).toBe(1), { timeout: 15_000, interval: 25 }) + 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.', + ) + expect(await state().getConversationSession( + 'factory-test', `slack:${mount.threadTs}`, + )).toBeUndefined() + } finally { + await first.stop() + await restarted?.stop() + await rm(root, { recursive: true, force: true }) + } + }, 20_000) + + // Abandonment commits the terminal phase, records the dispatch terminal, and + // clears the pending abandon reason *before* it retires the Slack watcher. A + // receipt that throws out of retirement therefore takes the rest of abandon + // with it — the registry rewrite, the GitHub watcher stop, the queued next + // dispatch — and nothing re-runs them, because the reason that would have + // driven an in-process retry is already gone. + it('completes dispatch abandonment when the terminal Slack receipt fails, and retries the receipt', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-abandon-terminal-receipt-')) + const registryPath = join(root, 'registry.json') + const issue = issueFile(823) + const mount = new FailingUndeliveredReceiptMountClient({ [issuePath(823)]: issue }) + const fleet = new ResumeNameCollisionFleetClient() + fleet.setSessionRef('ar-823-impl-pear', 'session-ar-823-impl-pear') + const stateStore = new InMemoryStateStore({ batchSize: 10 }) + const factory = createFactory(config({ + slack: { ...slackConfig(), conversationCoalesceMs: 60_000 }, + loop: { registryPath }, + }), { + mount, + fleet, + triage: new StaticTriage(), + stateStore, + probePrResolver: async () => undefined, + }) + + try { + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(823), issue))) + emitSlackReply(mount, slackReplyFixturePath( + 'C0FACTORY__factory-e2e', mount.threadTs, 'human-abandon-receipt', + ), 'slack-human-abandon-receipt', { + text: 'Nobody has told me this reply went nowhere.', + user: 'U823', + user_is_bot: false, + }) + await vi.waitFor(async () => expect( + (await stateStore.getConversationSession('factory-test', `slack:${mount.threadTs}`))?.pending, + ).toEqual([expect.objectContaining({ text: 'Nobody has told me this reply went nowhere.' })])) + + fleet.emitAgentExit('ar-823-impl-pear', 'crash') + + // The receipt failed, so the retirement deferred it. Everything abandon + // still owed after that point must have run anyway. + await vi.waitFor(() => expect( + factory.status().counters.slackTerminalWatchReceiptsDeferred, + ).toBe(1), { timeout: 5_000 }) + expect(mount.receiptAttempts).toBeGreaterThanOrEqual(1) + await vi.waitFor(async () => expect( + (await readFactoryInFlightRegistry(registryPath))?.agents, + ).toEqual([]), { timeout: 5_000 }) + expect(factory.status().inFlight).toEqual([]) + + // Writeback returns inside the grace window: the deferred receipt is this + // daemon's to finish, not the next restart's. + mount.failReceipts = false + await vi.waitFor(() => expect( + factory.status().counters.slackTerminalWatchReceiptsRecovered, + ).toBe(1), { timeout: 15_000, interval: 25 }) + 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.', + ) + } finally { + await factory.stop() + await rm(root, { recursive: true, force: true }) + } + }, 25_000) + + it('holds the terminal Slack receipt lease for as long as the provider write runs', async () => { + const mount = new ParkedTerminalReceiptMountClient() + const fleet = new FakeFleetClient() + const stateStore = new InMemoryStateStore({ batchSize: 10 }) + const clock = new ManualClock() + clock.advance(10_000) + const terminalIssue = { uuid: 'uuid-418', key: 'AR-418', path: issuePath(418) } + const decision = await new StaticTriage().triage(parseLinearIssue(issuePath(418), issueFile(418))) + const conversationId = `slack:${mount.threadTs}` + await stateStore.setSlackThreadWatch('factory-test', issueKey(terminalIssue), { + kind: 'terminal-grace', + issue: terminalIssue, + decision, + threadId: mount.threadTs, + retiredAtMs: clock.now(), + expiresAtMs: clock.now() + 24 * 60 * 60_000, + }) + await stateStore.reserveConversationSession('factory-test', conversationId, { + provider: 'slack', + issue: terminalIssue, + externalId: mount.threadTs, + context: { channel: 'C0FACTORY' }, + history: [], + processedMessageIds: [], + acknowledgedMessageIds: [], + pending: [], + }) + await stateStore.appendConversationMessage('factory-test', conversationId, { + id: `${mount.threadTs}:1780751613.000200`, + text: 'Nobody has told me this reply went nowhere.', + receivedAtMs: clock.now(), + providerSequence: '1780751613.000200', + author: 'U418', + }) + + vi.useFakeTimers() + const factory = createFactory(config({ slack: slackConfig() }), { + mount, + fleet, + triage: new StaticTriage(), + stateStore, + clock, + }) + const terminalNotice = '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.' + const starting = factory.start({ mode: 'dispatch-owner' }) + try { + for (let tick = 0; tick < 200 && !mount.receiptWriteEntered; tick += 1) { + await vi.advanceTimersByTimeAsync(1) + } + expect(mount.receiptWriteEntered).toBe(true) + + // Slack writeback budgets 90s for the confirm alone, so a write that runs + // longer than the 60s lease is inside spec, not pathological. Push past + // the lease with the provider call still in flight. + for (let step = 0; step < 8; step += 1) { + clock.advance(10_000) + await vi.advanceTimersByTimeAsync(10_000) + } + + // A lease has to outlive the work it covers: while this write is running + // no second handler may take the receipt and post the notice again. + expect(await stateStore.claimConversationTerminalReceipt( + 'factory-test', conversationId, 'competing-handler', clock.now(), 60_000, + )).toBe(false) + + mount.releaseReceiptWrite() + await starting + await vi.advanceTimersByTimeAsync(1) + expect(slackReplyWrites(mount).filter((write) => write.content.text === terminalNotice)) + .toHaveLength(1) + expect(await stateStore.getConversationSession('factory-test', conversationId)).toBeUndefined() + } finally { + mount.releaseReceiptWrite() + await starting.catch(() => undefined) + await factory.stop() + vi.useRealTimers() + } + }) + + it('stops renewing the terminal Slack receipt lease once its provider write outruns the ceiling', async () => { + const mount = new ParkedTerminalReceiptMountClient() + const fleet = new FakeFleetClient() + const stateStore = new InMemoryStateStore({ batchSize: 10 }) + const clock = new ManualClock() + clock.advance(10_000) + const terminalIssue = { uuid: 'uuid-418', key: 'AR-418', path: issuePath(418) } + const decision = await new StaticTriage().triage(parseLinearIssue(issuePath(418), issueFile(418))) + const conversationId = `slack:${mount.threadTs}` + await stateStore.setSlackThreadWatch('factory-test', issueKey(terminalIssue), { + kind: 'terminal-grace', + issue: terminalIssue, + decision, + threadId: mount.threadTs, + retiredAtMs: clock.now(), + expiresAtMs: clock.now() + 24 * 60 * 60_000, + }) + await stateStore.reserveConversationSession('factory-test', conversationId, { + provider: 'slack', + issue: terminalIssue, + externalId: mount.threadTs, + context: { channel: 'C0FACTORY' }, + history: [], + processedMessageIds: [], + acknowledgedMessageIds: [], + pending: [], + }) + await stateStore.appendConversationMessage('factory-test', conversationId, { + id: `${mount.threadTs}:1780751613.000300`, + text: 'This reply must not be stranded behind a wedged write.', + receivedAtMs: clock.now(), + providerSequence: '1780751613.000300', + author: 'U418', + }) + + vi.useFakeTimers() + const factory = createFactory(config({ slack: slackConfig() }), { + mount, + fleet, + triage: new StaticTriage(), + stateStore, + clock, + }) + const starting = factory.start({ mode: 'dispatch-owner' }) + try { + for (let tick = 0; tick < 200 && !mount.receiptWriteEntered; tick += 1) { + await vi.advanceTimersByTimeAsync(1) + } + expect(mount.receiptWriteEntered).toBe(true) + + // Inside the renewal ceiling the claim still belongs to the write: a + // writeback that runs past the 60s idle lease is inside spec, and this is + // the steal the renewal exists to prevent. + for (let step = 0; step < 12; step += 1) { + clock.advance(10_000) + await vi.advanceTimersByTimeAsync(10_000) + } + expect(await stateStore.claimConversationTerminalReceipt( + 'factory-test', conversationId, 'competing-handler', clock.now(), 60_000, + )).toBe(false) + + // Past the ceiling the write is no longer "slow", it is wedged, and a + // heartbeat that keeps renewing through it is a lock with no owner: the + // human's queued reply would stay unclaimable until the process restarts. + // Renewal has to stop so the claim idles out and the retry can reclaim it. + for (let step = 0; step < 60; step += 1) { + clock.advance(10_000) + await vi.advanceTimersByTimeAsync(10_000) + } + expect(await stateStore.claimConversationTerminalReceipt( + 'factory-test', conversationId, 'competing-handler', clock.now(), 60_000, + )).toBe(true) + // And the expiry is reported, not silent. + expect(factory.status().counters.slackProviderReceiptLeaseRenewalsExpired).toBe(1) + // The queued reply survives for whoever holds the claim now. + expect((await stateStore.getConversationSession('factory-test', conversationId))?.pending) + .toHaveLength(1) + } finally { + mount.releaseReceiptWrite() + await starting.catch(() => undefined) + await factory.stop() + vi.useRealTimers() + } + }) + + it('stops renewing the terminal Slack receipt lease when the daemon is shutting down', async () => { + const mount = new ParkedTerminalReceiptMountClient() + const fleet = new FakeFleetClient() + const stateStore = new InMemoryStateStore({ batchSize: 10 }) + const clock = new ManualClock() + clock.advance(10_000) + const terminalIssue = { uuid: 'uuid-418', key: 'AR-418', path: issuePath(418) } + const decision = await new StaticTriage().triage(parseLinearIssue(issuePath(418), issueFile(418))) + const conversationId = `slack:${mount.threadTs}` + await stateStore.setSlackThreadWatch('factory-test', issueKey(terminalIssue), { + kind: 'terminal-grace', + issue: terminalIssue, + decision, + threadId: mount.threadTs, + retiredAtMs: clock.now(), + expiresAtMs: clock.now() + 24 * 60 * 60_000, + }) + await stateStore.reserveConversationSession('factory-test', conversationId, { + provider: 'slack', + issue: terminalIssue, + externalId: mount.threadTs, + context: { channel: 'C0FACTORY' }, + history: [], + processedMessageIds: [], + acknowledgedMessageIds: [], + pending: [], + }) + await stateStore.appendConversationMessage('factory-test', conversationId, { + id: `${mount.threadTs}:1780751613.000400`, + text: 'A restart must not be what frees this reply.', + receivedAtMs: clock.now(), + providerSequence: '1780751613.000400', + author: 'U418', + }) + + vi.useFakeTimers() + const factory = createFactory(config({ slack: slackConfig() }), { + mount, + fleet, + triage: new StaticTriage(), + stateStore, + clock, + }) + const starting = factory.start({ mode: 'dispatch-owner' }) + let stopping: Promise | undefined + try { + for (let tick = 0; tick < 200 && !mount.receiptWriteEntered; tick += 1) { + await vi.advanceTimersByTimeAsync(1) + } + expect(mount.receiptWriteEntered).toBe(true) + + // A stopping daemon has no standing to keep holding a claim on work it is + // walking away from; the successor must find the receipt free. + stopping = factory.stop() + for (let step = 0; step < 12; step += 1) { + clock.advance(10_000) + await vi.advanceTimersByTimeAsync(10_000) + } + // Well inside the renewal ceiling, so this is shutdown releasing the + // claim and not the ceiling expiring it. + expect(await stateStore.claimConversationTerminalReceipt( + 'factory-test', conversationId, 'competing-handler', clock.now(), 60_000, + )).toBe(true) + expect(factory.status().counters.slackProviderReceiptLeaseRenewalsStoppedForShutdown).toBe(1) + expect(factory.status().counters.slackProviderReceiptLeaseRenewalsExpired).toBeUndefined() + } finally { + mount.releaseReceiptWrite() + await starting.catch(() => undefined) + await (stopping ?? factory.stop()).catch(() => undefined) + vi.useRealTimers() + } + }) + + it('holds the Slack reply acknowledgement lease for as long as the provider write runs', async () => { + const issue = issueFile(419) + const mount = new ParkedReplyReceiptMountClient({ [issuePath(419)]: issue }) + const fleet = new FakeFleetClient() + fleet.setSessionRef('ar-419-impl-pear', 'session-ar-419-impl-pear') + const stateStore = new InMemoryStateStore({ batchSize: 10 }) + const clock = new ManualClock() + clock.advance(10_000) + const factory = createFactory(config({ + slack: { ...slackConfig(), conversationCoalesceMs: 60_000 }, + }), { + mount, + fleet, + triage: new StaticTriage(), + stateStore, + clock, + }) + const conversationId = `slack:${mount.threadTs}` + const replyId = `${mount.threadTs}:slack-human-slow-receipt` + try { + await factory.dispatch(await factory.triageIssue(parseLinearIssue(issuePath(419), issue))) + vi.useFakeTimers() + emitSlackReply(mount, slackReplyFixturePath( + 'C0FACTORY__factory-e2e', mount.threadTs, 'human-slow-receipt', + ), 'slack-human-slow-receipt', { + text: 'Acknowledge me before anyone else does.', + user: 'U419', + user_is_bot: false, + }) + for (let tick = 0; tick < 400 && !mount.receiptWriteEntered; tick += 1) { + await vi.advanceTimersByTimeAsync(1) + } + expect(mount.receiptWriteEntered).toBe(true) + + // The acknowledgement claim covers the same 90s-budgeted writeback as the + // terminal receipt, so it is the same lease-scope bug. + for (let step = 0; step < 8; step += 1) { + clock.advance(10_000) + await vi.advanceTimersByTimeAsync(10_000) + } + + expect(await stateStore.claimConversationMessageAcknowledgement( + 'factory-test', conversationId, replyId, 'competing-handler', clock.now(), 60_000, + )).toBe(false) + + mount.releaseReceiptWrite() + for (let tick = 0; tick < 40; tick += 1) await vi.advanceTimersByTimeAsync(1) + expect(slackReplyWrites(mount).filter((write) => write.content.text === slackImplementerReceipt)) + .toHaveLength(1) + expect((await stateStore.getConversationSession('factory-test', conversationId)) + ?.acknowledgedMessageIds).toEqual([replyId]) + } finally { + mount.releaseReceiptWrite() + vi.useRealTimers() + await factory.stop() + } + }) + + it('does not re-post a terminal Slack receipt when the durable clear behind it fails', async () => { + const mount = new ConfirmRecordingSlackMountClient() + const fleet = new FakeFleetClient() + const stateStore = new FailingConversationClearStateStore({ batchSize: 10 }) + const clock = new ManualClock() + clock.advance(10_000) + const terminalIssue = { uuid: 'uuid-417', key: 'AR-417', path: issuePath(417) } + const decision = await new StaticTriage().triage(parseLinearIssue(issuePath(417), issueFile(417))) + const conversationId = `slack:${mount.threadTs}` + await stateStore.setSlackThreadWatch('factory-test', issueKey(terminalIssue), { + kind: 'terminal-grace', + issue: terminalIssue, + decision, + threadId: mount.threadTs, + retiredAtMs: clock.now(), + expiresAtMs: clock.now() + 24 * 60 * 60_000, + }) + await stateStore.reserveConversationSession('factory-test', conversationId, { + provider: 'slack', + issue: terminalIssue, + externalId: mount.threadTs, + context: { channel: 'C0FACTORY' }, + history: [], + processedMessageIds: [], + acknowledgedMessageIds: [], + pending: [], + }) + await stateStore.appendConversationMessage('factory-test', conversationId, { + id: `${mount.threadTs}:1780751613.000100`, + text: 'Nobody has told me this reply went nowhere.', + receivedAtMs: clock.now(), + providerSequence: '1780751613.000100', + author: 'U417', + }) + + const factory = createFactory(config({ slack: slackConfig() }), { + mount, + fleet, + triage: new StaticTriage(), + stateStore, + clock, + }) + const terminalNotice = '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.' + try { + await factory.start({ mode: 'dispatch-owner' }) + + // Slack accepted the receipt; only the durable clear behind it failed. The + // retry owes the state store another clear, not the human a second notice. + await vi.waitFor(() => expect( + factory.status().counters.slackTerminalWatchReceiptsDeferred, + ).toBe(1)) + await vi.waitFor(() => expect( + factory.status().counters.slackTerminalWatchReceiptsRecovered, + ).toBe(1), { timeout: 15_000, interval: 25 }) + expect(stateStore.clearAttempts).toBeGreaterThanOrEqual(2) + expect(slackReplyWrites(mount).filter((write) => write.content.text === terminalNotice)) + .toHaveLength(1) + expect(await stateStore.getConversationSession('factory-test', conversationId)).toBeUndefined() + } finally { + await factory.stop() + } + }, 20_000) + + 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 () => { @@ -22318,6 +23834,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 758a79ba..4600f59a 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -394,8 +394,32 @@ 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 +// Both receipt leases guard an in-flight Slack writeback, and no fixed lease can +// cover one: MountSlackWriteback budgets 90s for the confirm alone, on top of an +// unbounded writeFile. Sizing them past that worst case would only trade a stolen +// claim for a stranded one — a lease long enough to survive the slowest write is +// equally long enough to hold the receipt hostage to a dead holder. So these +// bound the *idle* claim and #withRenewedProviderLease extends them for exactly +// as long as the write they cover is still running. +const SLACK_REPLY_ACKNOWLEDGEMENT_LEASE_MS = 60_000 +const SLACK_TERMINAL_RECEIPT_CLAIM_LEASE_MS = 60_000 +// Renewal without a ceiling is the same defect from the other side: a heartbeat +// that extends the claim for as long as the write runs also extends it forever +// when the write never returns, and nothing else can reclaim the receipt short +// of a restart. So renewal is bounded past the slowest write this daemon budgets +// for — MountSlackWriteback's 90s confirm on top of its writeFile — and beyond +// that the write is not slow, it is wedged: the heartbeat stops, the idle lease +// runs out, and the retry that owns the queued replies can take them back. +const SLACK_PROVIDER_LEASE_MAX_RENEWAL_MS = 5 * 60_000 const SLACK_CONVERSATION_TURN_RETRY_MS = 1_000 const SLACK_REPLY_ROUTE_RETRY_MS = 1_000 +// One pass drains the whole chain (#slackReplyRoutes holds only the newest +// route per key and every route awaits its predecessor). The extra passes only +// exist so the drain can prove quiescence rather than assume it. +const SLACK_REPLY_ROUTE_DRAIN_PASSES = 8 +const SLACK_TERMINAL_THREAD_GRACE_MS = 24 * 60 * 60_000 +const SLACK_TERMINAL_RECEIPT_RETRY_MS = 1_000 +const SLACK_TERMINAL_RECEIPT_RETRY_MAX_MS = 5 * 60_000 const MERGE_GATE_MAX_ATTEMPTS = 12 const MERGE_GATE_POLL_DELAY_MS = 10_000 const MAX_LABEL_IMPLEMENTERS = 4 @@ -516,6 +540,19 @@ export class FactoryLoop implements Factory { readonly #dispatchInFlight = new Map>() readonly #slackWatchers = new Map() readonly #slackWatcherStarts = new Map>() + readonly #slackTerminalWatchExpiryTimers = new Map>() + readonly #slackTerminalReceiptRetryTimers = new Map>() + readonly #terminalSlackWatchIssues = new Set() + /** + * The one in-memory record of "this work unit has an in-flight Slack side + * effect". Both ordinary reply routes and the writebacks the terminal fence + * issues on their behalf register here, because this map is what the terminal + * drain waits on: anything that touches Slack for a work unit without + * registering is invisible to the drain, and the watcher teardown that follows + * a successful drain then pulls that effect's retry timer out from under it. + */ + readonly #slackReplyRoutes = new Map>() + readonly #slackReplyRouteDrains = new Set() readonly #slackConversationTurns: CoalescedTaskQueue readonly #slackConversationOwner = `${process.pid}:${randomUUID()}` readonly #githubIssueCommentWatchers = new Map() @@ -1140,6 +1177,12 @@ 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() + for (const timer of this.#slackTerminalReceiptRetryTimers.values()) clearTimeout(timer) + this.#slackTerminalReceiptRetryTimers.clear() + this.#terminalSlackWatchIssues.clear() + this.#slackReplyRouteDrains.clear() await Promise.all([...this.#githubIssueCommentWatchers.values()].map((watcher) => watcher.stop())) this.#githubIssueCommentWatchers.clear() this.#githubIssueCommentWatchStates.clear() @@ -5521,7 +5564,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 +8525,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 +12808,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 +13332,24 @@ 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. + 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) if (existingThread || watcherStart) { @@ -13390,13 +13451,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 +13474,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 +13496,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 +13540,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 +13663,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 +14011,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 @@ -13948,7 +14027,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 @@ -14016,6 +14095,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)) { @@ -14149,7 +14235,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)) { @@ -14214,6 +14300,48 @@ 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 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) + 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') + this.#scheduleSlackTerminalReceiptRetry(watch.issue, watch.threadId, watch.expiresAtMs) + } + await this.#rearmSlackWatcher(watchRecord, watch.threadId, { + replayConversationReplies: true, + replayAfterMs: retiredAtMs, + }) + 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 @@ -14384,8 +14512,72 @@ 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. + async #drainSlackReplyRoutes(key: string): Promise { + // Snapshotting #slackReplyRoutes is not enough on its own. A reply handler + // that is still inside its mount read when the drain starts registers its + // route *after* the snapshot, so it would run once the fence is gone and + // bind the retired thread to the next dispatch — the same escape one level + // in. Bar *routing* for this key first (the bar and the registration are + // both synchronous, so nothing can slip between them), then drain whatever + // is already chained, then prove the set is empty before reporting success. + // A barred reply still answers the human, and that writeback registers here + // like any other effect, so the extra passes are what pick it up: quiescence + // means every effect this work unit started has settled, not merely the ones + // that existed when the drain began. + const nested = this.#slackReplyRouteDrains.has(key) + this.#slackReplyRouteDrains.add(key) + try { + for (let pass = 0; pass < SLACK_REPLY_ROUTE_DRAIN_PASSES; pass += 1) { + const route = this.#slackReplyRoutes.get(key) + if (!route) return true + try { + await route + } catch (error) { + // 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. + this.#logger.warn?.( + '[factory] in-flight Slack reply route did not drain; keeping terminal Slack fence', + { issue: key, error }, + ) + this.#increment('slackReplyRouteDrainsFailed') + return false + } + // The owner clears its own entry when it settles; retiring it here too + // keeps the loop monotonic if that finally has not run yet. + if (this.#slackReplyRoutes.get(key) === route) this.#slackReplyRoutes.delete(key) + } + // Not provably quiescent. Fail closed for the same reason as a rejection. + this.#logger.warn?.( + '[factory] Slack reply routes did not quiesce; keeping terminal Slack fence', + { issue: key }, + ) + this.#increment('slackReplyRouteDrainsFailed') + return false + } finally { + if (!nested) this.#slackReplyRouteDrains.delete(key) + } + } + + 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) + this.#slackTerminalWatchExpiryTimers.delete(key) + const receiptRetryTimer = this.#slackTerminalReceiptRetryTimers.get(key) + if (receiptRetryTimer) clearTimeout(receiptRetryTimer) + this.#slackTerminalReceiptRetryTimers.delete(key) const watcher = this.#slackWatchers.get(key) this.#slackWatchers.delete(key) const threadId = await this.#state.getSlackThread(this.#workspaceId, key) @@ -14396,6 +14588,301 @@ 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) + return true + } + + 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 retiredAtMs = existingWatch?.kind === 'terminal-grace' + ? terminalSlackWatchRetiredAtMs(existingWatch) + : this.#clock.now() + const expiresAtMs = existingWatch?.kind === 'terminal-grace' + ? existingWatch.expiresAtMs + : 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) + try { + await this.#surfaceUndeliveredSlackConversation(threadId) + await this.#state.clearConversationSession(this.#workspaceId, conversationId) + } catch (error) { + // The receipt fails whenever another handler holds the claim or Slack + // writeback is down — neither is a reason to abort retirement. Callers + // reach here having already committed the terminal phase and dropped the + // pending abandon reason, so a rejection escaping would strand the + // registry rewrite, the GitHub watcher stop, and the queued next dispatch + // with nothing left to re-run them. Keep the queued replies and let the + // retry that owns this receipt settle it inside the grace window. + this.#logger.warn?.( + '[factory] failed to settle undelivered Slack replies while retiring the watcher; will retry', + { issue: record.issue.key, error }, + ) + this.#increment('slackTerminalWatchReceiptsDeferred') + this.#scheduleSlackTerminalReceiptRetry(record.issue, threadId, expiresAtMs) + } + if (!this.#slackWatchers.has(key) && !this.#stopping) { + await this.#rearmSlackWatcher(record, threadId) + } + this.#scheduleSlackTerminalWatchExpiry(record.issue, expiresAtMs) + this.#increment('slackTerminalWatchersRetained') + } + + // The durable half of the same record. Every caller here follows the receipt + // with a state write (clearing the session), and those two cannot be one + // durable step: when the state write fails, the retry that owns it must not + // read "replies still queued" as "the human has not been told" and post the + // notice again. So the receipt is claimed before the provider write and marked + // posted after it, and a retry finds it already settled. + async #surfaceUndeliveredSlackConversation(threadId: string): Promise { + const conversationId = slackConversationId(threadId) + const session = await this.#state.getConversationSession(this.#workspaceId, conversationId) + const pendingCount = session + ? session.pending.length + (session.delivery?.messages.length ?? 0) + : 0 + if (pendingCount === 0) return + if (session?.terminalReceipt?.posted) { + this.#increment('slackTerminalReceiptsAlreadySettled') + return + } + if (!this.#slack) throw new Error(`Slack thread ${threadId} cannot surface undelivered replies without writeback`) + const claimId = randomUUID() + if (!await this.#state.claimConversationTerminalReceipt( + this.#workspaceId, + conversationId, + claimId, + this.#clock.now(), + SLACK_TERMINAL_RECEIPT_CLAIM_LEASE_MS, + )) { + const current = await this.#state.getConversationSession(this.#workspaceId, conversationId) + if (current?.terminalReceipt?.posted) { + this.#increment('slackTerminalReceiptsAlreadySettled') + return + } + // Another handler is mid-write. Fail closed so the queued replies survive + // for whoever settles them rather than racing a second notice onto the + // same thread. + throw new Error(`Slack thread ${threadId} terminal receipt is claimed by another handler; retrying`) + } + const noun = pendingCount === 1 ? 'reply' : 'replies' + const slack = this.#slack + try { + await this.#withRenewedProviderLease( + 'terminal Slack receipt', + SLACK_TERMINAL_RECEIPT_CLAIM_LEASE_MS, + () => this.#state.renewConversationTerminalReceipt( + this.#workspaceId, + conversationId, + claimId, + this.#clock.now(), + ), + () => 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.`, + ), + ) + } catch (error) { + await this.#state.releaseConversationTerminalReceipt(this.#workspaceId, conversationId, claimId) + throw error + } + if (!await this.#state.completeConversationTerminalReceipt( + this.#workspaceId, + conversationId, + claimId, + )) { + throw new Error(`Slack thread ${threadId} terminal receipt could not be recorded`) + } + this.#increment('slackConversationRepliesSurfacedTerminal') + } + + // A claim only means something for as long as it outlives the work it covers. + // A provider write can legitimately run past a fixed lease, at which point the + // claim stops protecting the write it was taken for and a second handler can + // post the same thing to the same human. Renewing on a heartbeat scopes the + // lease to the work instead of to a guessed duration, and leaves the idle + // timeout short enough that a holder that dies mid-write still frees it. + // + // The heartbeat is bounded on both ends, because a renewal loop that never + // stops is a lock with no owner check: a provider write that hangs would hold + // the receipt past every retry and past shutdown, and the human whose reply is + // queued behind it would be told nothing until the process restarts. So it + // stops at the ceiling and it stops when this daemon is stopping, and either + // way it says so — from there the claim ages out on its own idle lease and + // becomes reclaimable. The write may still land afterwards and duplicate the + // notice; a reply nobody can ever reclaim is the worse of the two. + async #withRenewedProviderLease( + label: string, + leaseMs: number, + renew: () => Promise, + run: () => Promise, + ): Promise { + const renewUntilMs = this.#clock.now() + SLACK_PROVIDER_LEASE_MAX_RENEWAL_MS + let renewalStopped = false + let renewalInFlight = false + const stopRenewing = (counter: string, reason: string): void => { + renewalStopped = true + this.#increment(counter) + this.#logger.warn?.( + `[factory] ${label} lease will not be renewed further (${reason}); ` + + 'its claim expires and the queued replies return to whoever retries them', + ) + } + const heartbeat = setInterval(() => { + if (renewalInFlight || renewalStopped) return + if (this.#stopping) { + stopRenewing('slackProviderReceiptLeaseRenewalsStoppedForShutdown', 'shutting down') + return + } + if (this.#clock.now() >= renewUntilMs) { + stopRenewing( + 'slackProviderReceiptLeaseRenewalsExpired', + `provider write exceeded ${SLACK_PROVIDER_LEASE_MAX_RENEWAL_MS}ms`, + ) + return + } + renewalInFlight = true + void renew() + .then((renewed) => { + if (renewed) return + // Losing the lease mid-write is not recoverable from in here: the + // write may already have landed. Stop renewing and let the caller's + // completion check fail closed, which keeps the queued replies for + // whoever holds the claim now. + renewalStopped = true + this.#increment('slackProviderReceiptLeasesLost') + this.#logger.warn?.(`[factory] ${label} lease was lost while its provider write was in flight`) + }) + .catch((error) => this.#logger.warn?.(`[factory] ${label} lease renewal failed`, { + error: describeError(error).errorMessage, + })) + .finally(() => { renewalInFlight = false }) + }, Math.max(1_000, Math.floor(leaseMs / 3))) + heartbeat.unref?.() + try { + return await run() + } finally { + clearInterval(heartbeat) + } + } + + // A terminal receipt that could not be written leaves the queued replies + // pending with the human who wrote them told nothing. That is retryable + // maintenance this daemon owns, not work to leave for the next restart: the + // grace watch is the only window in which the receipt can still land on the + // retired thread, so keep reattempting inside it and give up when it closes. + #scheduleSlackTerminalReceiptRetry( + issue: IssueRef, + threadId: string, + expiresAtMs: number, + attempt = 0, + ): void { + if (this.#stopping) return + const key = issueKey(issue) + const existing = this.#slackTerminalReceiptRetryTimers.get(key) + if (existing) clearTimeout(existing) + this.#slackTerminalReceiptRetryTimers.delete(key) + const remainingMs = expiresAtMs - this.#clock.now() + if (remainingMs <= 0) { + this.#increment('slackTerminalWatchReceiptsAbandoned') + return + } + const backoffMs = Math.min( + SLACK_TERMINAL_RECEIPT_RETRY_MAX_MS, + SLACK_TERMINAL_RECEIPT_RETRY_MS * 2 ** Math.min(attempt, 16), + ) + const timer = setTimeout(() => { + this.#slackTerminalReceiptRetryTimers.delete(key) + void this.#retrySlackTerminalReceipt(issue, threadId, attempt) + }, Math.max(0, Math.min(backoffMs, remainingMs))) + timer.unref?.() + this.#slackTerminalReceiptRetryTimers.set(key, timer) + } + + async #retrySlackTerminalReceipt(issue: IssueRef, threadId: string, attempt: number): Promise { + if (this.#stopping) return + const key = issueKey(issue) + const watch = (await this.#state.listSlackThreadWatches(this.#workspaceId)) + .find(([watchKey]) => watchKey === key)?.[1] + // The grace watch is gone (expired, or the work unit reopened): the thread + // this receipt would settle no longer exists, so there is nothing to say. + if (watch?.kind !== 'terminal-grace' || watch.threadId !== threadId) return + try { + await this.#surfaceUndeliveredSlackConversation(threadId) + await this.#state.clearConversationSession(this.#workspaceId, slackConversationId(threadId)) + this.#increment('slackTerminalWatchReceiptsRecovered') + } catch (error) { + this.#logger.warn?.('[factory] terminal Slack receipt retry failed; rescheduling', { + issue: issue.key, + error, + }) + this.#increment('slackTerminalWatchReceiptRetryFailures') + this.#scheduleSlackTerminalReceiptRetry(issue, threadId, watch.expiresAtMs, attempt + 1) + } + } + + #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 + } + // #stopSlackWatcher fails closed when an in-flight reply route will not + // drain, leaving the watch and its terminal fence in place. Counting that as + // an expiration retires the watch in the metrics while the real one lives + // on unwatched by any expiry timer, so reschedule and count only on success. + if (!await this.#stopSlackWatcher(issue)) { + this.#increment('slackTerminalWatchExpiriesDeferred') + this.#scheduleSlackTerminalWatchExpiry(issue, expiresAtMs, SLACK_REPLY_ROUTE_RETRY_MS) + return + } + this.#increment('slackTerminalWatchersExpired') } async #readSlackReply(path: string): Promise { @@ -14463,34 +14950,196 @@ 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 { + if (this.#slackReplyRouteDrains.has(clarificationKey)) { + // The terminal fence for this work unit is being drained right now. + // Registering an ordinary route here would put it past the drain's + // snapshot and run it once the fence is gone. Answer it the way the fence + // would have — but as a tracked effect, because this writeback is still a + // side effect of this work unit. Left untracked it is the same escape one + // level further in: the drain reports quiescence without it, the watcher + // stop clears the retry timer that owns this reply, and a slow or failed + // receipt leaves the human told nothing at all. + this.#increment('slackReplyRoutesFencedDuringDrain') + return await this.#trackSlackWorkUnitEffect(clarificationKey, async () => { + await this.#writeUnroutableSlackReply(reply.threadTs) + return undefined + }) + } + return await this.#trackSlackWorkUnitEffect(clarificationKey, () => + this.#routeSlackConversationAnswerUnlocked(record, reply, text, clarificationKey)) + } + + // Every Slack side effect a work unit makes on its own behalf runs through + // here, so #slackReplyRoutes stays the single record the terminal drain + // consults. Effects are chained per work unit: awaiting the newest one drains + // everything queued behind it, and a rejection propagates to the drain, which + // fails closed rather than tearing the effect's retry path down. + async #trackSlackWorkUnitEffect( + key: string, + run: () => Promise, + ): Promise { + const preceding = this.#slackReplyRoutes.get(key) + const effect = (async () => { + await preceding?.catch(() => undefined) + return await run() + })() + this.#slackReplyRoutes.set(key, effect) + try { + return await effect + } finally { + if (this.#slackReplyRoutes.get(key) === effect) this.#slackReplyRoutes.delete(key) + } + } + + 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) - 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.' + const slack = this.#slack + // Same lease scope as the terminal receipt: this claim covers a + // provider write that can outrun any fixed duration, so it is + // renewed for as long as that write is actually running. + await this.#withRenewedProviderLease( + 'Slack reply acknowledgement', + SLACK_REPLY_ACKNOWLEDGEMENT_LEASE_MS, + () => this.#state.renewConversationMessageAcknowledgement( + this.#workspaceId, + conversationId, + replyId, + acknowledgementClaimId, + this.#clock.now(), + ), + () => 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') + await this.#writeUnroutableSlackReply(reply.threadTs) 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 #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 { @@ -14885,6 +15534,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 +15552,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 +15569,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 } @@ -17776,6 +18434,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 0f22ed9e..de798417 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,20 @@ 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 + /** + * Write-ahead outbox record for the one-per-thread terminal receipt telling a + * human their queued replies were never delivered. The provider write and the + * session clear that follows it cannot be one durable step, so the claim is + * taken before the write and marked posted after it: a retry of the clear then + * finds the receipt already settled instead of telling the human a second + * time. The lease keeps a crash between claim and write from suppressing the + * receipt forever. + */ + terminalReceipt?: { claimId: string; claimedAtMs: number; posted?: boolean } /** New replies waiting for the short coalescing window. */ pending: ConversationMessage[] /** Claimed batch; new arrivals remain in pending while this resume runs. */ @@ -144,10 +158,26 @@ 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 + /** Provider-message cutoff preventing historical replies from replaying as terminal. */ + retiredAtMs?: number + expiresAtMs: number +} + export type DispatchAttemptState = { attempts: number inFlight: boolean @@ -441,11 +471,32 @@ 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 + /** + * Extend an acknowledgement claim while its provider write is still running. + * Returns false once the claim is gone, so the caller learns it was overtaken + * instead of writing on a lease it no longer holds. + */ + renewConversationMessageAcknowledgement(workspaceId: string, conversationId: string, messageId: string, claimId: string, nowMs: number): Promise + releaseConversationMessageAcknowledgement(workspaceId: string, conversationId: string, messageId: string, claimId: string): Promise + /** + * Reserve the right to write this conversation's terminal receipt. Returns + * false once the receipt is posted, and while another handler holds an + * unexpired claim. + */ + claimConversationTerminalReceipt(workspaceId: string, conversationId: string, claimId: string, nowMs: number, leaseMs: number): Promise + renewConversationTerminalReceipt(workspaceId: string, conversationId: string, claimId: string, nowMs: number): Promise + completeConversationTerminalReceipt(workspaceId: string, conversationId: string, claimId: string): Promise + releaseConversationTerminalReceipt(workspaceId: string, conversationId: 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 +507,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 3252fa3d..a805c6c6 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 4a180fa5..34aa73d0 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: {}, @@ -474,6 +475,7 @@ describe('FileStateStore', () => { }, history: [], processedMessageIds: [], + acknowledgedMessageIds: [], pending: [], } const first = new FileStateStore({ batchSize: 2, watchStatePath }) @@ -516,6 +518,253 @@ 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('durably records a posted terminal receipt so a restart cannot re-post it', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-file-state-terminal-receipt-')) + try { + const watchStatePath = join(root, 'factory-state.json') + const conversationId = 'slack:1780751612.176224' + const session = { + provider: 'slack', + issue: { uuid: 'uuid-135', key: 'AR-135', path: '/linear/issues/AR-135__uuid-135.json' }, + externalId: '1780751612.176224', + context: { channelDir: 'C0FACTORY__factory-e2e' }, + history: [], + processedMessageIds: [], + acknowledgedMessageIds: [], + pending: [], + } + const first = new FileStateStore({ batchSize: 2, watchStatePath }) + expect(await first.reserveConversationSession('workspace-1', conversationId, session)).toBe(true) + await first.appendConversationMessage('workspace-1', conversationId, { + id: '1780751613.000010', text: 'Nobody delivered this.', receivedAtMs: 1_000, author: 'U135', + }) + + // A second handler must not race a duplicate notice onto the thread while + // the first one's write is still in flight. + expect(await first.claimConversationTerminalReceipt( + 'workspace-1', conversationId, 'receipt-a', 1_001, 60_000, + )).toBe(true) + expect(await first.claimConversationTerminalReceipt( + 'workspace-1', conversationId, 'receipt-b', 1_002, 60_000, + )).toBe(false) + + // Slack writeback budgets 90s for the confirm alone, so the holder renews + // while its write runs: the idle lease stays short without letting a slow + // provider call outlive the claim taken for it. + expect(await first.renewConversationTerminalReceipt( + 'workspace-1', conversationId, 'receipt-a', 200_000, + )).toBe(true) + expect(await first.claimConversationTerminalReceipt( + 'workspace-1', conversationId, 'receipt-b', 220_000, 60_000, + )).toBe(false) + expect(await first.renewConversationTerminalReceipt( + 'workspace-1', conversationId, 'not-the-holder', 220_000, + )).toBe(false) + + // The per-message acknowledgement claim covers the same provider write and + // renews on the same terms. + expect(await first.claimConversationMessageAcknowledgement( + 'workspace-1', conversationId, '1780751613.000010', 'ack-a', 220_001, 60_000, + )).toBe(true) + expect(await first.renewConversationMessageAcknowledgement( + 'workspace-1', conversationId, '1780751613.000010', 'ack-a', 400_000, + )).toBe(true) + expect(await first.claimConversationMessageAcknowledgement( + 'workspace-1', conversationId, '1780751613.000010', 'ack-thief', 420_000, 60_000, + )).toBe(false) + expect(await first.renewConversationMessageAcknowledgement( + 'workspace-1', conversationId, '1780751613.000010', 'ack-thief', 420_000, + )).toBe(false) + await first.releaseConversationMessageAcknowledgement( + 'workspace-1', conversationId, '1780751613.000010', 'ack-a', + ) + + // A receipt that never reached Slack releases its claim, so the retry that + // owes the human the notice can still take it. + await first.releaseConversationTerminalReceipt('workspace-1', conversationId, 'receipt-a') + expect(await first.claimConversationTerminalReceipt( + 'workspace-1', conversationId, 'receipt-b', 1_003, 60_000, + )).toBe(true) + expect(await first.completeConversationTerminalReceipt( + 'workspace-1', conversationId, 'receipt-b', + )).toBe(true) + + // The replies are still queued because the clear behind the receipt failed. + // A restart therefore re-runs that maintenance and must find the receipt + // settled rather than telling the human a second time. + const restarted = new FileStateStore({ batchSize: 2, watchStatePath }) + expect((await restarted.getConversationSession('workspace-1', conversationId))) + .toMatchObject({ pending: [{ id: '1780751613.000010' }], terminalReceipt: { posted: true } }) + expect(await restarted.claimConversationTerminalReceipt( + 'workspace-1', conversationId, 'receipt-c', 1_004, 60_000, + )).toBe(false) + await restarted.releaseConversationTerminalReceipt('workspace-1', conversationId, 'receipt-b') + expect((await new FileStateStore({ batchSize: 2, watchStatePath }) + .getConversationSession('workspace-1', conversationId))?.terminalReceipt?.posted).toBe(true) + } finally { + await rm(root, { recursive: true, force: true }) + } + }) + + 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 { + 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', + retiredAtMs: 1_000, + 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 f3308daf..325eaed0 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,128 @@ 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 renewConversationMessageAcknowledgement( + workspaceId: string, + conversationId: string, + messageId: string, + claimId: string, + nowMs: number, + ): Promise { + const result = await this.#mutateConversation(workspaceId, conversationId, (session) => { + const claim = session.acknowledgementClaims?.[messageId] + if (claim?.claimId !== claimId) return false + claim.claimedAtMs = nowMs + 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 claimConversationTerminalReceipt( + workspaceId: string, + conversationId: string, + claimId: string, + nowMs: number, + leaseMs: number, + ): Promise { + const result = await this.#mutateConversation(workspaceId, conversationId, (session) => { + if (session.terminalReceipt?.posted) return false + const current = session.terminalReceipt + if (current && current.claimedAtMs + leaseMs > nowMs) return false + session.terminalReceipt = { claimId, claimedAtMs: nowMs } + return true + }) + return Boolean(result) + } + + override async renewConversationTerminalReceipt( + workspaceId: string, + conversationId: string, + claimId: string, + nowMs: number, + ): Promise { + const result = await this.#mutateConversation(workspaceId, conversationId, (session) => { + const receipt = session.terminalReceipt + if (receipt?.claimId !== claimId || receipt.posted) return false + receipt.claimedAtMs = nowMs + return true + }) + return Boolean(result) + } + + override async completeConversationTerminalReceipt( + workspaceId: string, + conversationId: string, + claimId: string, + ): Promise { + const result = await this.#mutateConversation(workspaceId, conversationId, (session) => { + if (session.terminalReceipt?.claimId !== claimId) return false + session.terminalReceipt = { ...session.terminalReceipt, posted: true } + return true + }) + return Boolean(result) + } + + override async releaseConversationTerminalReceipt( + workspaceId: string, + conversationId: string, + claimId: string, + ): Promise { + await this.#mutateConversation(workspaceId, conversationId, (session) => { + if (session.terminalReceipt?.claimId !== claimId || session.terminalReceipt.posted) return false + delete session.terminalReceipt + return true + }) + } + override async claimConversationTurn( workspaceId: string, conversationId: string, @@ -940,10 +1097,12 @@ 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) { + const hadDelivery = session.delivery !== undefined session.delivery = undefined - return false + return hadDelivery } + const agent = session.agent session.delivery = { claimId, owner, @@ -951,8 +1110,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 +1144,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 +1181,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 +1392,7 @@ const dispatchLifecycleHandedOffToBabysitters = (lifecycle: DispatchLifecycle): const emptyWorkspaceState = (): PersistedWorkspaceState => ({ githubIssueCommentWatches: {}, + slackThreadWatches: {}, waitingClarifications: {}, babysitterSessions: {}, babysitterGenerations: {}, @@ -1242,6 +1403,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 fa9c61f2..47d2f9e0 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,115 @@ 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 renewConversationMessageAcknowledgement( + workspaceId: string, + conversationId: string, + messageId: string, + claimId: string, + nowMs: number, + ): Promise { + const session = this.#workspace(workspaceId).conversationSessions.get(conversationId) + const claim = session?.acknowledgementClaims?.[messageId] + if (claim?.claimId !== claimId) return false + claim.claimedAtMs = nowMs + 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 claimConversationTerminalReceipt( + workspaceId: string, + conversationId: string, + claimId: string, + nowMs: number, + leaseMs: number, + ): Promise { + const session = this.#workspace(workspaceId).conversationSessions.get(conversationId) + if (!session) return false + if (session.terminalReceipt?.posted) return false + const current = session.terminalReceipt + if (current && current.claimedAtMs + leaseMs > nowMs) return false + session.terminalReceipt = { claimId, claimedAtMs: nowMs } + return true + } + + async renewConversationTerminalReceipt( + workspaceId: string, + conversationId: string, + claimId: string, + nowMs: number, + ): Promise { + const session = this.#workspace(workspaceId).conversationSessions.get(conversationId) + const receipt = session?.terminalReceipt + if (receipt?.claimId !== claimId || receipt.posted) return false + receipt.claimedAtMs = nowMs + return true + } + + async completeConversationTerminalReceipt( + workspaceId: string, + conversationId: string, + claimId: string, + ): Promise { + const session = this.#workspace(workspaceId).conversationSessions.get(conversationId) + if (session?.terminalReceipt?.claimId !== claimId) return false + session.terminalReceipt = { ...session.terminalReceipt, posted: true } + return true + } + + async releaseConversationTerminalReceipt( + workspaceId: string, + conversationId: string, + claimId: string, + ): Promise { + const session = this.#workspace(workspaceId).conversationSessions.get(conversationId) + if (session?.terminalReceipt?.claimId === claimId && !session.terminalReceipt.posted) { + delete session.terminalReceipt + } + } + async claimConversationTurn( workspaceId: string, conversationId: string, @@ -389,10 +513,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 +525,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 +556,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 +582,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 +944,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.test.ts b/src/state/watch-state-document.test.ts index d9a96abe..1751b510 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) => { @@ -18,6 +19,44 @@ describe('parseWatchStateDocument', () => { expect(parseWatchStateDocument(validDocument())).toEqual(validDocument()) }) + // Triage can label an issue `agent:swarm`, and every persisted decision — the + // Slack thread watch, the waiting clarification, the dispatch lifecycle — stores + // that scope verbatim. A validator that does not know the scope rejects the whole + // document, so one swarm issue costs every watch in the workspace on restart. + it('accepts a swarm-scope decision everywhere a triage decision is persisted', () => { + const document = validDocument() + const swarmDecision = { ...decision(), scope: 'swarm' } + document.workspaces.workspace.slackThreadWatches.watch = { + kind: 'terminal-grace', + issue: issue(), + decision: swarmDecision, + threadId: '1780751612.176224', + retiredAtMs: 1, + expiresAtMs: 2, + } + document.workspaces.workspace.waitingClarifications.clarification.decision = swarmDecision + document.workspaces.workspace.dispatchLifecycles.lifecycle.decision = swarmDecision + + expect(parseWatchStateDocument(document)).toEqual(document) + }) + + 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 = { @@ -77,6 +116,7 @@ const validDocument = (): Record => ({ }], }, }, + slackThreadWatches: {}, waitingClarifications: { clarification: { issue: issue(), diff --git a/src/state/watch-state-document.ts b/src/state/watch-state-document.ts index 1ac74d0d..79c53089 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,12 @@ 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)) || + (candidate.terminalReceipt !== undefined && !validConversationTerminalReceipt(candidate.terminalReceipt)) || (delivery !== undefined && !validConversationDelivery(delivery) && !validLegacyConversationDelivery(delivery)) ) throw invalidDocument() const session = structuredClone(candidate) as unknown as ConversationSessionState @@ -150,6 +159,15 @@ const parseConversationSessions = ( ...(session.delivery?.messages ?? []), ].map((message) => message.id))] : [...candidate.processedMessageIds 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'] + } + if (candidate.terminalReceipt !== undefined) { + session.terminalReceipt = structuredClone(candidate.terminalReceipt) as ConversationSessionState['terminalReceipt'] + } sessions[conversationId] = session } return sessions @@ -180,6 +198,14 @@ 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 validConversationTerminalReceipt = (value: unknown): boolean => + isRecord(value) && typeof value.claimId === 'string' && typeof value.claimedAtMs === 'number' && + (value.posted === undefined || typeof value.posted === 'boolean') + const parseBabysitterSessions = (value: Record): Record => { const sessions: Record = {} for (const [key, candidate] of Object.entries(value)) { @@ -288,6 +314,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') && @@ -429,9 +473,19 @@ const validPendingPullRequestWake = (value: unknown): boolean => isRecord(value) typeof value.repo === 'string' && Number.isSafeInteger(value.number) && (value.number as number) > 0 && Array.isArray(value.kinds) && value.kinds.every((kind) => typeof kind === 'string') +// A scope this validator does not know rejects the entire document, so one +// swarm-scoped issue would cost every watch in the workspace on restart. Keyed +// on the union itself, a new scope is a compile error here instead. +const triageScopes: Record = { + single: true, + workflow: true, + team: true, + swarm: true, +} + const validTriageDecision = (value: unknown): value is TriageDecision => isRecord(value) && validIssueRef(value.issue) && Array.isArray(value.routes) && value.routes.every(validRoute) && - (value.scope === 'single' || value.scope === 'workflow' || value.scope === 'team') && + (typeof value.scope === 'string' && Object.hasOwn(triageScopes, value.scope)) && Array.isArray(value.implementers) && value.implementers.every(validAgentSpec) && (value.workflow === undefined || validAgentSpec(value.workflow)) && validAgentSpec(value.reviewer) && typeof value.thin === 'boolean' && (value.confidence === 'high' || value.confidence === 'low') &&