fix: route Slack dispatch replies to babysitters

src/ports/state.ts

@@ -115,7 +115,7 @@ export type ConversationSessionState = {
   externalId: string
   /** Provider-specific routing metadata; continuity itself stays provider-neutral. */
   context: Record<string, string>
-  agent: {
+  agent?: {
     name: string
     sessionRef: string
     /**
@@ -134,6 +134,10 @@ export type ConversationSessionState = {
   history: ConversationMessage[]
   /** Durable dedupe ledger; unlike rendered history, this is never context-trimmed. */
   processedMessageIds: string[]
+  /** Human replies whose visible provider receipt has been acknowledged. */
+  acknowledgedMessageIds?: string[]
+  /** Short durable claims preventing duplicate concurrent provider receipts. */
+  acknowledgementClaims?: Record<string, { claimId: string; claimedAtMs: number }>
   /** New replies waiting for the short coalescing window. */
   pending: ConversationMessage[]
   /** Claimed batch; new arrivals remain in pending while this resume runs. */
@@ -144,10 +148,26 @@ export type ConversationSessionState = {
     attempts: number
     messages: ConversationMessage[]
     /** Binding captured at claim time so a later handoff cannot be overwritten. */
-    agent: Pick<ConversationSessionState['agent'], 'name' | 'sessionRef'>
+    agent: Pick<NonNullable<ConversationSessionState['agent']>, '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 +461,17 @@ export interface StateStore {
   getSlackThread(workspaceId: string, issueKey: string): Promise<string | undefined>
   clearSlackThread(workspaceId: string, issueKey: string): Promise<void>
   clearSlackThreads(workspaceId: string): Promise<void>
+  setSlackThreadWatch(workspaceId: string, issueKey: string, watch: SlackThreadWatchState): Promise<void>
+  listSlackThreadWatches(workspaceId: string): Promise<Array<[string, SlackThreadWatchState]>>
+  clearSlackThreadWatch(workspaceId: string, issueKey: string): Promise<void>
 
   reserveConversationSession(workspaceId: string, conversationId: string, session: ConversationSessionState): Promise<boolean>
   getConversationSession(workspaceId: string, conversationId: string): Promise<ConversationSessionState | undefined>
   listConversationSessions(workspaceId: string): Promise<Array<[string, ConversationSessionState]>>
   appendConversationMessage(workspaceId: string, conversationId: string, message: ConversationMessage): Promise<ConversationSessionState | undefined>
+  claimConversationMessageAcknowledgement(workspaceId: string, conversationId: string, messageId: string, claimId: string, nowMs: number, leaseMs: number): Promise<boolean>
+  completeConversationMessageAcknowledgement(workspaceId: string, conversationId: string, messageId: string, claimId: string): Promise<boolean>
+  releaseConversationMessageAcknowledgement(workspaceId: string, conversationId: string, messageId: string, claimId: string): Promise<void>
   claimConversationTurn(workspaceId: string, conversationId: string, owner: string, claimId: string, nowMs: number, leaseMs: number): Promise<ConversationSessionState | undefined>
   renewConversationTurn(workspaceId: string, conversationId: string, owner: string, claimId: string, nowMs: number): Promise<boolean>
   completeConversationTurn(workspaceId: string, conversationId: string, owner: string, claimId: string, agent: { name: string; sessionRef?: string }): Promise<boolean>
@@ -456,7 +482,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<boolean>
+  rebindConversationSession(workspaceId: string, conversationId: string, agent: NonNullable<ConversationSessionState['agent']>): Promise<boolean>
 
   setGithubIssueCommentWatch(workspaceId: string, key: string, watch: GithubIssueCommentWatchState): Promise<void>
   listGithubIssueCommentWatches(workspaceId: string): Promise<Array<[string, GithubIssueCommentWatchState]>>

src/state/document-store.ts

@@ -5,11 +5,13 @@ import type {
   DiscoverySweepState,
   DispatchLifecycle,
   GithubIssueCommentWatchState,
+  SlackThreadWatchState,
   WaitingClarification,
 } from '../ports/state'
 
 export type PersistedWorkspaceState = {
   githubIssueCommentWatches: Record<string, GithubIssueCommentWatchState>
+  slackThreadWatches: Record<string, SlackThreadWatchState>
   waitingClarifications: Record<string, WaitingClarification>
   babysitterSessions: Record<string, BabysitterSessionState>
   babysitterGenerations: Record<string, BabysitterGenerationRecord>

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,165 @@ 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 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 {