fix(control-plane): make the hook redelivery sweep cover the window it claims

packages/control-plane/src/github/api.ts

@@ -62,8 +62,32 @@ export interface GithubRequestOpts {
   bigIdsAsStrings?: boolean
 }
 
+/** One page of a paginated GitHub list: the body plus the `rel="next"` cursor. */
+export interface GithubPage<T> {
+  data: T
+  /** Next page as a path under the SAME base — a `next` pointing anywhere else
+   *  is dropped rather than followed. */
+  nextPath?: string
+}
+
+/** `rel="next"` from a Link header, relative to `baseUrl`. */
+function nextPathFrom(link: string | null, baseUrl: string): string | undefined {
+  if (!link) return undefined
+  for (const part of link.split(',')) {
+    const url = /<([^>]+)>\s*;\s*rel="next"/.exec(part.trim())?.[1]
+    if (url?.startsWith(baseUrl)) return url.slice(baseUrl.length)
+  }
+  return undefined
+}
+
 /** One GitHub REST call → parsed JSON. Throws `GithubApiError` on any non-2xx. */
 export async function githubRequest<T>(path: string, opts: GithubRequestOpts): Promise<T> {
+  return (await githubRequestPage<T>(path, opts)).data
+}
+
+/** {@link githubRequest} keeping the pagination cursor — for the list endpoints
+ *  whose first page is not the whole answer. */
+export async function githubRequestPage<T>(path: string, opts: GithubRequestOpts): Promise<GithubPage<T>> {
   const fetchImpl = opts.fetchImpl ?? (fetch as FetchLike)
   let res: Response
   try {
@@ -83,11 +107,12 @@ export async function githubRequest<T>(path: string, opts: GithubRequestOpts): P
   }
   if (res.ok) {
     const text = await res.text()
-    if (!text) return undefined as T // 202-style empty success (e.g. redelivery accepted)
+    const nextPath = nextPathFrom(res.headers.get('link'), opts.baseUrl ?? API_BASE)
+    if (!text) return { data: undefined as T } // 202-style empty success (e.g. redelivery accepted)
     // Only `id` keys with ≥15 digits are re-quoted — small numeric ids
     // (repositories, installations) stay numbers for existing callers.
     const safe = opts.bigIdsAsStrings ? text.replace(/"id"\s*:\s*(\d{15,})/g, '"id":"$1"') : text
-    return JSON.parse(safe) as T
+    return { data: JSON.parse(safe) as T, ...(nextPath ? { nextPath } : {}) }
   }
 
   // Read the message for diagnostics; GitHub error bodies carry no secrets.

packages/control-plane/src/github/github.test.ts

@@ -10,7 +10,7 @@ import type { GitCredCapability } from '@agentconnect.md/protocol'
 import { FakeClock } from '../../test/fakes/fake-clock.js'
 import type { AgentRepoAuthorizationRecord, GithubInstallationRecord } from '../persistence/ports.js'
 import { githubAppBotIdentity, resolveGithubAppConfig, type GithubAppConfig } from './config.js'
-import { GithubApiError, githubRequest, mintAppJwt } from './api.js'
+import { GithubApiError, githubRequest, mintAppJwt, type FetchLike } from './api.js'
 import { InstallationTokenInvalidatedError, InstallationTokenService } from './installation-token.service.js'
 import { deriveInstallStateKey, mintInstallState, verifyInstallState, INSTALL_STATE_TTL_MS } from './install-state.js'
 import { GithubService } from './service.js'
@@ -152,6 +152,90 @@ describe('githubRequest', () => {
   })
 })
 
+describe('GithubService.listHookDeliveries', () => {
+  const deliveryPage = (guids: string[], at: string): string =>
+    JSON.stringify(
+      guids.map((guid) => ({
+        id: '1234567890123456789',
+        guid,
+        delivered_at: at,
+        event: 'pull_request',
+        action: 'opened',
+        repository_id: 42,
+        installation_id: 7
+      }))
+    )
+  const svc = (fetchImpl: FetchLike) =>
+    new GithubService({
+      cfg: cfg(),
+      clock: new FakeClock(1_700_000_000_000),
+      installations: {} as never,
+      installState: { put: async () => {}, consume: async () => true },
+      pepper: 'p'.repeat(32),
+      fetchImpl
+    })
+
+  it('walks the cursor until a page reaches past the floor', async () => {
+    const paths: string[] = []
+    const fetchImpl = vi.fn(async (url: string) => {
+      paths.push(url)
+      const second = url.includes('cursor=next')
+      return new Response(
+        second ? deliveryPage(['old'], '2023-11-14T21:40:00.000Z') : deliveryPage(['new'], '2023-11-14T22:10:00.000Z'),
+        {
+          status: 200,
+          headers: {
+            'content-type': 'application/json',
+            link: '<https://api.github.com/app/hook/deliveries?per_page=100&cursor=next>; rel="next"'
+          }
+        }
+      )
+    })
+
+    const page = await svc(fetchImpl).listHookDeliveries({ deliveredSince: new Date('2023-11-14T21:50:00.000Z') })
+    expect(page.deliveries.map((d) => d.guid)).toEqual(['new', 'old'])
+    expect(page.truncated).toBe(false) // the second page reaches past the floor
+    expect(paths[1]).toContain('cursor=next')
+  })
+
+  it('reports truncation when the page budget runs out before the floor', async () => {
+    const fetchImpl = vi.fn(
+      async () =>
+        new Response(deliveryPage(['g'], '2023-11-14T22:10:00.000Z'), {
+          status: 200,
+          headers: {
+            'content-type': 'application/json',
+            link: '<https://api.github.com/app/hook/deliveries?per_page=100&cursor=next>; rel="next"'
+          }
+        })
+    )
+
+    const page = await svc(fetchImpl).listHookDeliveries({
+      maxPages: 3,
+      deliveredSince: new Date('2023-11-14T20:00:00.000Z')
+    })
+    expect(fetchImpl).toHaveBeenCalledTimes(3)
+    expect(page.truncated).toBe(true)
+  })
+
+  it('never follows a next cursor that points off the API base', async () => {
+    const fetchImpl = vi.fn(
+      async () =>
+        new Response(deliveryPage(['g'], '2023-11-14T22:10:00.000Z'), {
+          status: 200,
+          headers: {
+            'content-type': 'application/json',
+            link: '<https://evil.example/app/hook/deliveries?cursor=next>; rel="next"'
+          }
+        })
+    )
+
+    const page = await svc(fetchImpl).listHookDeliveries({ deliveredSince: new Date('2023-11-14T20:00:00.000Z') })
+    expect(fetchImpl).toHaveBeenCalledTimes(1)
+    expect(page.truncated).toBe(false)
+  })
+})
+
 describe('GithubService comment authorization lookups', () => {
   const installation: GithubInstallationRecord = {
     id: 'installation-row',

packages/control-plane/src/github/service.ts

@@ -23,7 +23,7 @@ import type {
   AgentRepoAuthorizationRepo,
   RepoAccess
 } from '../persistence/ports.js'
-import { githubRequest, mintAppJwt, GithubApiError, type FetchLike } from './api.js'
+import { githubRequest, githubRequestPage, mintAppJwt, GithubApiError, type FetchLike, type GithubPage } from './api.js'
 import { githubAppBotIdentity, type GithubAppConfig } from './config.js'
 import { InstallationTokenService, type CapabilityLevels, type MintedGitCred } from './installation-token.service.js'
 import { deriveInstallStateKey, mintInstallState, verifyInstallState } from './install-state.js'
@@ -76,6 +76,15 @@ export interface GhHookDelivery {
   installation_id: number | null
 }
 
+/** A delivery listing plus how honest it is about its own reach. */
+export interface GhHookDeliveryPage {
+  /** Newest first, across every page walked. */
+  deliveries: GhHookDelivery[]
+  /** The walk stopped on its page budget with a cursor left — the caller has
+   *  NOT seen everything back to `deliveredSince`, and must not claim it did. */
+  truncated: boolean
+}
+
 export interface GithubServiceDeps {
   cfg: GithubAppConfig
   clock: Clock
@@ -100,6 +109,8 @@ export interface GithubServiceDeps {
 const OUTDATED_INSTALLATIONS_CACHE_MS = 30_000
 const REPO_PAGE_CACHE_MS = 5 * 60_000
 const MAX_REPO_PAGE_CACHE_ENTRIES = 1_000
+/** Delivery-list pages one sweep may walk — the firehose bound, not the window. */
+const MAX_DELIVERY_PAGES = 10
 
 type RepoPageLookup = { ins: GithubInstallationRecord; page: number; perPage: number }
 
@@ -469,13 +480,29 @@ export class GithubService {
 
   /**
    * The App's recent webhook deliveries, newest first (P2.5 redelivery
-   * reconciliation). ONE page of `perPage` — the deliveries cursor rides a Link
-   * header `githubRequest` doesn't surface; the reconciler logs when the page
-   * is full so an under-covered window is visible, never silent.
+   * reconciliation). Follows the Link cursor until `deliveredSince` is reached
+   * — one page is a few minutes of traffic on a busy App, far less than the
+   * reconciler's look-back window — and stops at `maxPages` so a firehose
+   * bounds the sweep instead of the sweep bounding itself. The caller sees how
+   * far back the result actually reaches and reports the shortfall.
    */
-  async listHookDeliveries(perPage = 100): Promise<GhHookDelivery[]> {
+  async listHookDeliveries(
+    opts: { perPage?: number; maxPages?: number; deliveredSince?: Date } = {}
+  ): Promise<GhHookDeliveryPage> {
+    const floorMs = opts.deliveredSince?.getTime()
+    const deliveries: GhHookDelivery[] = []
     // bigIdsAsStrings: delivery ids overflow Number.MAX_SAFE_INTEGER — see GhHookDelivery.id.
-    return this.appRequest<GhHookDelivery[]>(`/app/hook/deliveries?per_page=${perPage}`, 'GET', true)
+    let path = `/app/hook/deliveries?per_page=${opts.perPage ?? 100}`
+    for (let page = 0; page < (opts.maxPages ?? MAX_DELIVERY_PAGES); page++) {
+      const rep: GithubPage<GhHookDelivery[]> = await this.appRequestPage<GhHookDelivery[]>(path, true)
+      deliveries.push(...rep.data)
+      const oldest = rep.data[rep.data.length - 1]
+      // No cursor left, or this page already reaches past the floor: complete.
+      if (!rep.nextPath || !oldest) return { deliveries, truncated: false }
+      if (floorMs !== undefined && Date.parse(oldest.delivered_at) <= floorMs) return { deliveries, truncated: false }
+      path = rep.nextPath
+    }
+    return { deliveries, truncated: true }
   }
 
   /** Ask GitHub to redeliver one delivery (202; lands on the relay pool again —
@@ -1063,8 +1090,17 @@ export class GithubService {
     method: 'GET' | 'POST' | 'DELETE' = 'GET',
     bigIdsAsStrings = false
   ): Promise<T> {
+    return (await this.appRequestPage<T>(path, bigIdsAsStrings, method)).data
+  }
+
+  /** {@link appRequest} keeping the `rel="next"` cursor (paginated list reads). */
+  private async appRequestPage<T>(
+    path: string,
+    bigIdsAsStrings = false,
+    method: 'GET' | 'POST' | 'DELETE' = 'GET'
+  ): Promise<GithubPage<T>> {
     const jwt = await mintAppJwt(this.deps.cfg)
-    return githubRequest<T>(path, {
+    return githubRequestPage<T>(path, {
       method,
       auth: jwt,
       fetchImpl: this.deps.fetchImpl,

packages/control-plane/src/orchestrator/hookRedeliveryReconciler.test.ts

@@ -73,6 +73,8 @@ function delivery(over: Partial<GhHookDelivery> = {}): GhHookDelivery {
 function make(opts: {
   hooks?: HookRecord[]
   deliveries?: GhHookDelivery[] | (() => GhHookDelivery[])
+  /** The listing walked its page budget without reaching `deliveredSince`. */
+  truncated?: boolean
   landed?: string[]
   relaysAlive?: boolean | (() => boolean)
   redeliverError?: boolean
@@ -99,11 +101,13 @@ function make(opts: {
   )
   const settleMock = vi.fn(async () => 0)
   const reviewFanoutClaimMock = vi.fn(async () => opts.reviewFanoutClaim ?? false)
+  const listMock = vi.fn(async (_opts?: { deliveredSince?: Date }) => ({
+    deliveries: typeof opts.deliveries === 'function' ? opts.deliveries() : (opts.deliveries ?? [delivery()]),
+    truncated: opts.truncated ?? false
+  }))
   const reconciler = new HookRedeliveryReconciler(
     {
-      listHookDeliveries: vi.fn(async () =>
-        typeof opts.deliveries === 'function' ? opts.deliveries() : (opts.deliveries ?? [delivery()])
-      ),
+      listHookDeliveries: listMock,
       redeliverHookDelivery: redeliverMock
     },
     {
@@ -120,9 +124,12 @@ function make(opts: {
   // Ticks stay MANUAL: tick()'s finally re-arms a timer, and clock.advance()
   // would fire those into async sweeps racing the test's own tick() calls.
   reconciler.stop()
-  return { reconciler, redelivered, redeliverMock, reviewFanoutClaimMock, claimMock, settleMock, clock }
+  return { reconciler, redelivered, redeliverMock, reviewFanoutClaimMock, claimMock, settleMock, listMock, clock }
 }
 
+/** Let a clock-fired sweep run to completion. */
+const flush = (): Promise<void> => new Promise((r) => setTimeout(r, 0))
+
 describe('HookRedeliveryReconciler', () => {
   it('redelivers a matching, unlanded GUID', async () => {
     const h = make({})
@@ -337,4 +344,65 @@ describe('HookRedeliveryReconciler', () => {
     expect(h.redelivered).toEqual([]) // never succeeded…
     expect(h.redeliverMock).toHaveBeenCalledTimes(3) // …retried next ticks, then the cap ended the calls
   })
+
+  it('asks the delivery listing to reach the whole look-back window', async () => {
+    const h = make({})
+    await h.reconciler.tick()
+    expect(h.listMock.mock.calls[0]?.[0]?.deliveredSince).toEqual(new Date(NOW - CFG.windowMs))
+  })
+
+  it('a truncated listing covers only what it listed, and the next sweep resumes there', async () => {
+    // The busy-App case: the page budget runs out long before the window does.
+    // Anything older than the oldest listed delivery was never looked at.
+    const listedFloor = NOW - 4 * 60 * 1000
+    const h = make({
+      truncated: true,
+      deliveries: () => [delivery({ delivered_at: new Date(listedFloor).toISOString() })]
+    })
+    await h.reconciler.tick()
+    h.clock.advance(CFG.intervalMs)
+    await h.reconciler.tick()
+
+    // Not `now − windowMs` (which would slide past the unlisted slice) and not
+    // `newest` — the second sweep starts exactly where the first one stopped.
+    expect(h.listMock.mock.calls[1]?.[0]?.deliveredSince).toEqual(new Date(listedFloor))
+  })
+
+  it('a listing truncated entirely inside the grace window covers no more than the ceiling it evaluated', async () => {
+    // Every listed delivery is younger than `graceMs`, so none of them were
+    // evaluated. Coverage must stop at that ceiling — carrying it up to the
+    // oldest listed delivery would skip everything in between once those
+    // deliveries age into eligibility.
+    const h = make({
+      truncated: true,
+      deliveries: () => [delivery({ delivered_at: new Date(NOW - 30_000).toISOString() })]
+    })
+    await h.reconciler.tick()
+    h.clock.advance(CFG.intervalMs)
+    await h.reconciler.tick()
+    expect(h.listMock.mock.calls[1]?.[0]?.deliveredSince).toEqual(new Date(NOW - CFG.graceMs))
+  })
+
+  it('a complete listing that is simply short still advances coverage', async () => {
+    const h = make({ deliveries: () => [delivery()] }) // truncated: false — quiet App
+    await h.reconciler.tick()
+    h.clock.advance(CFG.intervalMs)
+    await h.reconciler.tick()
+    // The first sweep covered its whole window, so the second resumes at that
+    // sweep's ceiling instead of re-listing an interval it already saw.
+    expect(h.listMock.mock.calls[1]?.[0]?.deliveredSince).toEqual(new Date(NOW - CFG.graceMs))
+  })
+
+  it('runs its first sweep early — a CP that restarts every few minutes still sweeps', async () => {
+    const h = make({})
+    h.reconciler.start()
+
+    h.clock.advance(30_000)
+    await flush()
+    expect(h.listMock).not.toHaveBeenCalled()
+    h.clock.advance(30_000) // 60s after boot, not a full interval
+    await flush()
+    expect(h.listMock).toHaveBeenCalledOnce()
+    h.reconciler.stop()
+  })
 })