feat(webhooks): accepted PR events ride the durable signal queue

Closes the last lossy event path: a crash between the webhook 202 and the
review session spawning no longer loses the PR event.

- The filter chain (signature, action, draft, allowlist, dedup, cap) stays
  inline — response semantics unchanged. On acceptance the handler
  enqueues a pr-review signal (deduped on the existing idempotency key,
  1h TTL) instead of fire-and-forgetting processPrReviewAsync
- processQueuedPrReview consumes from the queue: redelivery-safe via the
  pre-allocated session id (existing session = spawn already happened),
  malformed signals dropped with a log, transient failures rethrown so
  the queue retries them (3 attempts)
- Publisher injected by ws-server; without one (tests, engine down or
  enqueue failure) the legacy inline path runs unchanged

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: alari76

server/webhook-handler.test.ts

@@ -352,6 +352,72 @@ describe('WebhookHandler', () => {
     })
   })
 
+  describe('pr-review durable signal path', () => {
+    function makePrBody(overrides: Record<string, unknown> = {}): Buffer {
+      return Buffer.from(JSON.stringify({
+        action: 'opened',
+        pull_request: {
+          number: 7,
+          title: 'Add feature',
+          body: 'desc',
+          draft: false,
+          html_url: 'https://github.com/acme/widget/pull/7',
+          user: { login: 'alice' },
+          head: { sha: 'abc123', ref: 'feat/x' },
+          base: { ref: 'main' },
+        },
+        repository: { full_name: 'acme/widget', name: 'widget' },
+        sender: { login: 'alice' },
+        ...overrides,
+      }))
+    }
+
+    it('enqueues a pr-review signal on acceptance instead of processing inline', async () => {
+      await handler.checkHealth()
+      const publish = vi.fn()
+      handler.setSignalPublisher(publish)
+
+      const body = makePrBody()
+      const result = await handler.handleWebhook(body, makeHeaders(body, { event: 'pull_request' }))
+
+      expect(result.statusCode).toBe(202)
+      expect(result.body.accepted).toBe(true)
+      expect(publish).toHaveBeenCalledTimes(1)
+      const input = publish.mock.calls[0][0] as { kind: string; dedupeKey: string; payload: Record<string, unknown> }
+      expect(input.kind).toBe('pr-review')
+      expect(input.dedupeKey).toMatch(/^pr-review::/)
+      expect(input.payload.sessionId).toBe(result.body.sessionId)
+      // Inline processing did not run — no session was created at accept time.
+      expect(sessions.create).not.toHaveBeenCalled()
+    })
+
+    it('falls back to inline processing when the publisher throws', async () => {
+      await handler.checkHealth()
+      handler.setSignalPublisher(() => { throw new Error('engine down') })
+
+      const body = makePrBody()
+      const result = await handler.handleWebhook(body, makeHeaders(body, { event: 'pull_request' }))
+
+      // Accepted either way; the inline path runs async and is not asserted
+      // further here (it exercises the gh mocks like the legacy path).
+      expect(result.statusCode).toBe(202)
+    })
+
+    it('processQueuedPrReview is a no-op when the pre-allocated session already exists (redelivery)', async () => {
+      ;(sessions.get as ReturnType<typeof vi.fn>).mockReturnValue({ id: 's-1' })
+      await handler.processQueuedPrReview({
+        payload: { pull_request: { number: 7 }, repository: { full_name: 'acme/widget', name: 'widget' } },
+        webhookEvent: { id: 'e1' },
+        sessionId: 's-1',
+      })
+      expect(sessions.create).not.toHaveBeenCalled()
+    })
+
+    it('processQueuedPrReview drops malformed signals without throwing', async () => {
+      await expect(handler.processQueuedPrReview({})).resolves.toBeUndefined()
+    })
+  })
+
   describe('event history', () => {
     it('getEvents returns a copy of events', async () => {
       await handler.checkHealth()

server/webhook-handler.ts

@@ -45,6 +45,18 @@ export class WebhookHandler extends WebhookHandlerBase<WebhookEvent, WebhookEven
   private sessions: SessionManager
   private dedup: WebhookDedup
   private ghHealthy = false
+  /**
+   * Durable-queue publisher (the trigger engine's enqueueSignal, injected).
+   * When set, accepted PR events are enqueued as `pr-review` signals instead
+   * of being processed fire-and-forget — a crash between the 202 and the
+   * review session spawning no longer loses the event. Without it (tests,
+   * engine unavailable) the legacy inline path runs unchanged.
+   */
+  private signalPublisher: ((input: { kind: string; payload?: Record<string, unknown>; dedupeKey?: string; ttlMs?: number }) => void) | null = null
+
+  setSignalPublisher(publish: ((input: { kind: string; payload?: Record<string, unknown>; dedupeKey?: string; ttlMs?: number }) => void) | null): void {
+    this.signalPublisher = publish
+  }
 
   constructor(config: FullWebhookConfig, sessions: SessionManager) {
     super('webhook', PROCESSING_TIMEOUT_MS)
@@ -538,18 +550,64 @@ export class WebhookHandler extends WebhookHandlerBase<WebhookEvent, WebhookEven
     this.recordEvent(webhookEvent)
     this.dedup.recordProcessed(eventId, idempotencyKey)
 
-    // Process asynchronously
-    this.processPrReviewAsync(payload, webhookEvent, sessionId).catch(err => {
-      console.error('[webhook] PR review async processing error:', err)
-      this.updateEventStatus(eventId, 'error', String(err))
-    })
+    // Durable path: enqueue and let the dispatcher deliver (at-least-once,
+    // survives a crash before the session spawns). Falls back to the legacy
+    // inline path when no publisher is wired or the enqueue itself fails.
+    let queued = false
+    if (this.signalPublisher) {
+      try {
+        this.signalPublisher({
+          kind: 'pr-review',
+          payload: { payload, webhookEvent, sessionId } as unknown as Record<string, unknown>,
+          dedupeKey: `pr-review::${idempotencyKey}`,
+          ttlMs: 60 * 60 * 1000,
+        })
+        queued = true
+      } catch (err) {
+        console.error('[webhook] Failed to enqueue pr-review signal, processing inline:', err)
+      }
+    }
+    if (!queued) {
+      this.processPrReviewAsync(payload, webhookEvent, sessionId).catch(err => {
+        console.error('[webhook] PR review async processing error:', err)
+        this.updateEventStatus(eventId, 'error', String(err))
+      })
+    }
 
     return {
       statusCode: 202,
       body: { accepted: true, eventId, status: 'processing', sessionId },
     }
   }
 
+  /**
+   * Consume a queued `pr-review` signal. Redelivery-safe: the session id was
+   * pre-allocated at acceptance, so an existing session means the spawn
+   * already happened and the redelivery is a no-op. A rejection (transient gh
+   * failure before any session exists) propagates so the queue retries it.
+   */
+  async processQueuedPrReview(raw: Record<string, unknown>): Promise<void> {
+    const { payload, webhookEvent, sessionId } = raw as unknown as {
+      payload: PullRequestPayload
+      webhookEvent: WebhookEvent
+      sessionId: string
+    }
+    if (!payload || !webhookEvent || !sessionId) {
+      // Malformed signal — acking (returning) is correct; retrying can't fix it.
+      console.error('[webhook] Malformed pr-review signal payload, dropping')
+      return
+    }
+    if (this.sessions.get(sessionId)) return
+
+    try {
+      await this.processPrReviewAsync(payload, webhookEvent, sessionId)
+    } catch (err) {
+      console.error('[webhook] Queued PR review processing error:', err)
+      this.updateEventStatus(webhookEvent.id, 'error', String(err))
+      throw err
+    }
+  }
+
   /**
    * Handle PR closed/merged — archive or delete the review cache and kill any
    * active review sessions for this PR.

server/ws-server.ts

@@ -725,6 +725,12 @@ server.listen(port, '0.0.0.0', () => {
       if (!handler) throw new Error('Commit event handler not available')
       await handler.handle(payload as unknown as import('./commit-event-handler.js').CommitEvent)
     })
+
+    // Accepted PR webhook events ride the same durable queue: the webhook
+    // handler enqueues after its filter chain, and the spawn happens here —
+    // redelivery-safe via the pre-allocated session id.
+    webhookHandler.setSignalPublisher((input) => { engine.enqueueSignal(input) })
+    engine.registerSignalHandler('pr-review', (payload) => webhookHandler.processQueuedPrReview(payload))
     if (authToken) {
       const serverUrl = `http://localhost:${port}`
       ensureHookConfig(authToken, serverUrl)