fix(server): cancel merge observation HTTP requests on shutdown

Bound GitHub object and repository probes, keep readiness independent of external reads, and page pending cards through the owning parser.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

doc/plans/2026-10-30-merged-pr-confirmations-sleep.md

@@ -13,10 +13,12 @@ it receives no GitHub payloads. No database migration is required.
 ## Durable truth and producers
 
 Pending `request_confirmation` interaction rows remain the durable work source.
-The candidate query excludes terminal and hidden issues, linked tool actions,
-and text without merge intent or a PR reference. The existing fail-closed parser
-then excludes additional actions, governed actions, and document/attachment targets.
-No accepted or cancelled history keeps a deadline alive.
+The candidate query pages pending cards in bounded groups of 100 and excludes
+terminal and hidden issues and linked tool actions. The existing fail-closed
+parser is the sole text eligibility source: it requires merge intent and a PR
+reference and excludes additional actions, governed actions, and document/attachment targets.
+No accepted or cancelled history keeps a deadline alive. A full page without any
+merge candidates still advances its keyset cursor; later candidates are not lost.
 
 Ordinary creation registers the topic before inserting a confirmation in its
 transaction. Existing confirmation-resolution notifications invalidate its issue
@@ -53,6 +55,14 @@ This is candidate-backed external observation. It does not eliminate all polling
 register a Cloud wake alarm, or prove that the host/database suspends. Existing
 sleep-safety blockers remain in force.
 
+Startup registers this observation without waiting for live GitHub requests.
+The internal resolver invocation accepts an optional host-owned abort signal;
+that signal is not part of external-object records or plugin/RPC payloads.
+The GitHub owner combines cancellation with a ten-second request budget and
+passes it to the actual object fetch and repository-visibility probes, including
+response body reads. Shutdown aborts the request and waits for its cancellation;
+no post-abort acceptance is committed. A timeout leaves the durable card pending.
+
 ## Acceptance evidence
 
 Focused tests use real PostgreSQL, the actual service producers, transaction
@@ -68,6 +78,22 @@ provider failure across drain; standby/resume; restart; tenant-scoped and duplic
 merge hints; latest/superseded cards during provider observation; hidden/terminal
 issues; URL targets; no-continuation policy; and shutdown during an in-flight read.
 Existing startup and normalized GitHub event tests exercise the changed consumers.
+Controlled local HTTP tests hold headers, partial response bodies, 404 probes,
+and public-visibility probes. They observe server-side response close after
+shutdown within one second, rather than only racing a promise. A shortened
+request budget times out a real request; the saved card then accepts after
+restart when that same HTTP fixture serves a merged response. Existing GitHub
+provider and external-object service tests preserve their authorization behavior.
+
+## Verification follow-up
+
+The original head passed full-workspace typecheck and build. A later required
+Linux CI shard failed the unchanged managed-skill maintenance CLI fixture in
+`managed-skill-content.test.ts`: it returned `unavailable_or_corrupt_package`
+where `local_storage` was expected. Both the isolated case and the complete
+unchanged 17-test file pass against fresh local PostgreSQL. The Linux cause is
+not established; this is retained as an unresolved CI failure, not called a flake.
+The current-head required CI result remains the completion gate.
 
 ## Existing limitations outside this migration
 

server/src/__tests__/external-objects-service.test.ts

@@ -1,3 +1,4 @@
+import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
 import { randomUUID } from "node:crypto";
 import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
 import { eq } from "drizzle-orm";
@@ -26,7 +27,8 @@ import type { PaperclipPluginManifestV1 } from "@paperclipai/shared";
 import type { PluginWorkerManager } from "../services/plugin-worker-manager.js";
 import { createGitHubExternalObjectProvider } from "../services/github-external-object-provider.js";
 
-const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
+const external = resolveExternalTestDatabaseUrl("PAPERCLIP_EXTERNAL_OBJECTS_TEST_DATABASE_URL");
+const embeddedPostgresSupport = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
 const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
 
 if (!embeddedPostgresSupport.supported) {
@@ -709,8 +711,8 @@ describeEmbeddedPostgres("externalObjectService", () => {
   let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
 
   beforeAll(async () => {
-    tempDb = await startEmbeddedPostgresTestDatabase("paperclip-external-objects-");
-    db = createDb(tempDb.connectionString);
+    if (!external) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-external-objects-");
+    db = createDb(external ?? tempDb!.connectionString, { maxConnections: 6 });
   }, 20_000);
 
   afterEach(async () => {
@@ -725,6 +727,7 @@ describeEmbeddedPostgres("externalObjectService", () => {
   });
 
   afterAll(async () => {
+    await db?.$client.end({ timeout: 1 });
     await tempDb?.cleanup();
   });
 

server/src/__tests__/merged-pr-confirmation-work.test.ts

@@ -1,3 +1,5 @@
+import { createServer } from "node:http";
+import { createPullRequestMergeStateResolver } from "../services/github-pull-request-merge.js";
 import { randomUUID } from "node:crypto";
 import { setTimeout as delay } from "node:timers/promises";
 import { eq, sql } from "drizzle-orm";
@@ -262,6 +264,92 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
     expect(wakeup).toHaveBeenCalledTimes(1); expect(errors).toEqual([]);
   });
 
+  it.each(["headers", "body", "not-found probe", "public visibility probe"] as const)("cancels an actual unresponsive HTTP %s read promptly at stop", async phase => {
+    vi.useRealTimers();
+    const card = await create();
+    let observed!: () => void, closed!: () => void;
+    const requested = new Promise<void>(resolve => { observed = resolve; });
+    const disconnected = new Promise<void>(resolve => { closed = resolve; });
+    const signals: AbortSignal[] = [];
+    const server = createServer((request, response) => {
+      if (phase === "not-found probe" && request.url?.includes("/pulls/")) { response.writeHead(404).end(); return; }
+      response.on("close", closed);
+      if (phase === "body") { response.writeHead(200, { "content-type": "application/json" }); response.write('{"merged":'); }
+      observed();
+    });
+    await new Promise<void>(resolve => { server.listen(0, "127.0.0.1", resolve); });
+    const address = server.address();
+    if (!address || typeof address === "string") throw new Error("Expected test HTTP address");
+    const resolveState = createPullRequestMergeStateResolver(db, {
+      requestTimeoutMs: 10_000,
+      credentialProvider: () => phase === "public visibility probe" ? { token: "fixture", source: "server_env", secretName: null } : null,
+      fetch: (url, init) => { signals.push(init!.signal!); return fetch(`http://127.0.0.1:${address.port}${new URL(url).pathname}`, init); },
+    });
+    const real = issueThreadInteractionService(db, { resolvePullRequestState: resolveState, wakeup });
+    coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
+    try {
+      const startupAt = performance.now();
+      const worker = registerMergeConfirmationWork(coordinator, real, () => true);
+      expect(performance.now() - startupAt).toBeLessThan(100);
+      await Promise.race([requested, delay(2000).then(() => { throw new Error("HTTP observation did not start"); })]);
+      const stoppingAt = performance.now();
+      await coordinator.stop(); coordinator = undefined;
+      expect(performance.now() - stoppingAt).toBeLessThan(1000);
+      await worker.ready;
+      await Promise.race([disconnected, delay(1000).then(() => { throw new Error("HTTP request remained open after stop"); })]);
+      expect(signals.at(-1)?.aborted).toBe(true);
+      expect((await stored(card.id)).status).toBe("pending"); expect(wakeup).not.toHaveBeenCalled(); expect(errors).toEqual([]);
+    } finally {
+      await coordinator?.stop(); coordinator = undefined;
+      server.closeAllConnections(); await new Promise<void>(resolve => { server.close(() => resolve()); });
+    }
+  });
+
+  it("times out a real unresponsive HTTP read and recovers its durable candidate on restart", async () => {
+    vi.useRealTimers(); const card = await create();
+    let serveMerge = false, closed!: () => void;
+    const disconnected = new Promise<void>(resolve => { closed = resolve; });
+    const server = createServer((_request, response) => {
+      if (serveMerge) { response.writeHead(200, { "content-type": "application/json" }).end(JSON.stringify({ merged: true, state: "closed" })); return; }
+      response.on("close", closed);
+    });
+    await new Promise<void>(resolve => { server.listen(0, "127.0.0.1", resolve); });
+    const address = server.address(); if (!address || typeof address === "string") throw new Error("Expected test HTTP address");
+    const real = issueThreadInteractionService(db, { resolvePullRequestState: createPullRequestMergeStateResolver(db, {
+      requestTimeoutMs: 100, tokenProvider: () => null,
+      fetch: (url, init) => fetch(`http://127.0.0.1:${address.port}${new URL(url).pathname}`, init),
+    }), wakeup });
+    coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
+    try {
+      const startedAt = performance.now();
+      await registerMergeConfirmationWork(coordinator, real, () => true).ready;
+      expect(performance.now() - startedAt).toBeLessThan(2000);
+      await Promise.race([disconnected, delay(1000).then(() => { throw new Error("Timed out HTTP request remained open"); })]);
+      expect(errors).toHaveLength(1); expect((await stored(card.id)).status).toBe("pending"); expect(wakeup).not.toHaveBeenCalled();
+      expect(coordinator.nextWakeAt()).not.toBeNull();
+      await coordinator.stop(); coordinator = undefined; serveMerge = true;
+      coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
+      await registerMergeConfirmationWork(coordinator, real, () => true).ready;
+      expect((await stored(card.id)).status).toBe("accepted"); expect(wakeup).toHaveBeenCalledTimes(1);
+    } finally {
+      await coordinator?.stop(); coordinator = undefined;
+      server.closeAllConnections(); await new Promise<void>(resolve => { server.close(() => resolve()); });
+    }
+  });
+
+  it("pages beyond a full pending-card page without duplicating the merge parser in SQL", async () => {
+    const card = await create();
+    const id = BigInt(`0x${card.id.replaceAll("-", "")}`);
+    await db.insert(issueThreadInteractions).values(Array.from({ length: 100 }, (_, index) => {
+      const hex = (id - BigInt(index + 1)).toString(16).padStart(32, "0");
+      return { id: `${hex.slice(0,8)}-${hex.slice(8,12)}-${hex.slice(12,16)}-${hex.slice(16,20)}-${hex.slice(20)}`,
+        companyId, issueId, kind: "request_confirmation", title: "Approve the plan?", payload: { version: 1 as const, prompt: "Approve the plan?" } };
+    }));
+    await start(); expect(provider).not.toHaveBeenCalled();
+    await waitFor(async () => expect((await stored(card.id)).status).toBe("accepted")); await quiet();
+    expect(provider).toHaveBeenCalledTimes(1); expect(wakeup).toHaveBeenCalledTimes(1); expect(errors).toEqual([]);
+  });
+
   it("limits normalized merge hints to the owning company and matching PR and tolerates duplicates", async () => {
     provider.mockResolvedValue("open"); const own = await create();
     const foreignCompany = randomUUID(), foreignIssue = randomUUID();

server/src/__tests__/server-startup-feedback-export.test.ts

@@ -180,6 +180,7 @@ const {
   const heartbeatServiceFactoryMock = vi.fn(() => heartbeatServiceMock);
   const issueThreadInteractionServiceMock = {
     sweepSupersededPendingRequestConfirmations: vi.fn(async () => ({ expired: 0 })),
+    mergedPullRequestConfirmationPage: vi.fn(async () => ({ rows: [], complete: true })),
     mergedPullRequestConfirmationCandidates: vi.fn(async () => []),
     sweepMergedPullRequestConfirmations: vi.fn(async () => ({
       checked: 0,
@@ -1335,7 +1336,7 @@ describe("startServer feedback export wiring", () => {
 
       expect(heartbeatServiceMock.tickAgentTimers).toHaveBeenCalledTimes(1);
       expect(externalObjectsServiceMock.refreshDueObjectsForActiveCompanies).toHaveBeenCalledTimes(1);
-      expect(issueThreadInteractionServiceMock.mergedPullRequestConfirmationCandidates).toHaveBeenCalledTimes(1);
+      expect(issueThreadInteractionServiceMock.mergedPullRequestConfirmationPage).toHaveBeenCalledTimes(1);
       expect(issueThreadInteractionServiceMock.sweepMergedPullRequestConfirmations).not.toHaveBeenCalled();
       expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.mergedPullRequestConfirmations)).toBe(true);
       expect(executionWorkspaceServiceMock.sweepTerminalWorkspaces).toHaveBeenCalledTimes(1);
@@ -1424,6 +1425,20 @@ describe("startServer feedback export wiring", () => {
     }
   });
 
+  it("becomes ready without waiting for the merge-confirmation observation", async () => {
+    loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: true }));
+    let release!: () => void;
+    const blocked = new Promise<{ rows: never[]; complete: boolean }>(resolve => { release = () => resolve({ rows: [], complete: true }); });
+    issueThreadInteractionServiceMock.mergedPullRequestConfirmationPage.mockReturnValueOnce(blocked);
+    const timer = vi.spyOn(globalThis, "setInterval").mockImplementation(() => 1 as unknown as ReturnType<typeof setInterval>);
+    let started = false;
+    const starting = startServer().then(server => { started = true; return server; });
+    try {
+      await vi.waitFor(() => expect(issueThreadInteractionServiceMock.mergedPullRequestConfirmationPage).toHaveBeenCalled());
+      await settleStartupHousekeeping(); expect(started).toBe(true); expect(getStartupRecoveryState().phase).toBe("ready");
+    } finally { release(); await starting; timer.mockRestore(); }
+  });
+
   it.each(["retained cleanup", "workspace finalization", "environment settlement", "native run projection", "native result commit", "native recovery notices", "external conversation settlement"] as const)("becomes ready while %s is held but joins it at shutdown", async owner => {
     loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: true }));
     let releaseCleanup!: () => void;

server/src/index.ts

@@ -1601,8 +1601,8 @@ async function startServerWithDatabaseTeardown(
         : config.workspaceCompactionDelayHours * 60 * 60 * 1000,
       terminalWorkspaceInspectionBackoffMs: TERMINAL_WORKSPACE_RETRY_MS,
     });
-    await registerMergeConfirmationWork(deliveryWork, mergedPullRequestConfirmations,
-      () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready;
+    registerMergeConfirmationWork(deliveryWork, mergedPullRequestConfirmations,
+      () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby());
     const gitRestorePointWorker = registerGitRestorePointWork(deliveryWork, gitRestorePointRetentionService(db as any, {
       storage: storageService,
       retentionDays: config.gitRestorePointRetentionDays,

server/src/services/external-objects.ts

@@ -84,6 +84,7 @@ export interface ExternalObjectResolver {
   resolve(input: {
     companyId: string;
     object: ExternalObjectRecord;
+    signal?: AbortSignal;
   }): Promise<ExternalObjectResolveResult>;
 }
 

server/src/services/github-external-object-provider.ts

@@ -15,6 +15,7 @@ type FetchLike = (url: string, init?: RequestInit) => Promise<Response>;
 
 export interface GitHubExternalObjectProviderOptions {
   fetch?: FetchLike;
+  requestTimeoutMs?: number;
   tokenProvider?: (companyId: string) => Promise<string | null> | string | null;
   /**
    * Resolve the credential and its source. Prefer this over `tokenProvider` when the
@@ -243,12 +244,12 @@ type RepoProbeOutcome = {
 
 async function probeRepoVisibility(
   identity: GitHubObjectIdentity,
-  opts: { fetch: FetchLike; headers?: Record<string, string> | undefined },
+  opts: { fetch: FetchLike; headers?: Record<string, string> | undefined; signal?: AbortSignal },
 ): Promise<RepoProbeOutcome> {
   try {
     const probe = await opts.fetch(
       `${gitHubApiBase(identity.host)}/repos/${encodeURIComponent(identity.owner)}/${encodeURIComponent(identity.repo)}`,
-      { headers: opts.headers },
+      { headers: opts.headers, signal: opts.signal },
     );
     if (probe.ok) {
       const body = await safeJson(probe);
@@ -427,12 +428,13 @@ export function createGitHubExternalObjectProvider(
     companyId: string,
     identity: GitHubObjectIdentity,
     headers: Record<string, string>,
+    signal: AbortSignal,
   ): Promise<RepoProbeOutcome> {
     const credentialFingerprint = createHash("sha256").update(headers.authorization ?? "anonymous").digest("hex");
     const key = `${companyId}\u0000${credentialFingerprint}\u0000${identity.owner.toLowerCase()}\u0000${identity.repo.toLowerCase()}`;
     const cached = probeCache.get(key);
     if (cached && cached.expiresAt > Date.now()) return cached.outcome;
-    const outcome = await probeRepoVisibility(identity, { fetch: fetchImpl, headers });
+    const outcome = await probeRepoVisibility(identity, { fetch: fetchImpl, headers, signal });
     if (outcome.visibility === "visible" || outcome.visibility === "hidden") {
       probeCache.set(key, { outcome, expiresAt: Date.now() + GITHUB_OBJECT_TTL_SECONDS * 1000 });
     }
@@ -464,7 +466,9 @@ export function createGitHubExternalObjectProvider(
     return {
       providerKey: "github",
       objectType,
-      async resolve({ companyId, object }) {
+      async resolve({ companyId, object, signal: ownerSignal }) {
+        const deadline = AbortSignal.timeout(opts.requestTimeoutMs ?? 10_000);
+        const signal = ownerSignal ? AbortSignal.any([ownerSignal, deadline]) : deadline;
         const identity = parseGitHubObject(object);
         if (!identity || identity.objectType !== objectType) {
           return {
@@ -509,7 +513,7 @@ export function createGitHubExternalObjectProvider(
           // "public" verdict must not authorize a later read of its private contents.
           // The cache is still used below to classify a 404, where a stale verdict can
           // only make the object look deleted — it can never expose private data.
-          const probe = await probeRepoVisibility(identity, { fetch: fetchImpl, headers });
+          const probe = await probeRepoVisibility(identity, { fetch: fetchImpl, headers, signal });
           if (probe.visibility === "hidden") {
             return {
               ok: false,
@@ -554,7 +558,7 @@ export function createGitHubExternalObjectProvider(
 
         let response: Response;
         try {
-          response = await fetchImpl(url, { headers });
+          response = await fetchImpl(url, { headers, signal });
         } catch {
           return {
             ok: false,
@@ -573,7 +577,7 @@ export function createGitHubExternalObjectProvider(
           // Probe the repository metadata to classify: if the repo itself is visible
           // the object truly is gone; if it is not, surface credentials-needed
           // instead of permanently freezing a reference as "Not found".
-          const probe = await cachedProbeRepoVisibility(companyId, identity, headers);
+          const probe = await cachedProbeRepoVisibility(companyId, identity, headers, signal);
           if (probe.visibility === "visible") {
             return { ok: true, snapshot: notFoundSnapshot(identity, etag) };
           }

server/src/services/github-pull-request-merge.ts

@@ -1,5 +1,5 @@
 import type { Db } from "@paperclipai/db";
-import { createGitHubExternalObjectProvider } from "./github-external-object-provider.js";
+import { createGitHubExternalObjectProvider, type GitHubExternalObjectProviderOptions } from "./github-external-object-provider.js";
 
 export type GitHubPullRequestReference = {
   host: "github.com";
@@ -25,11 +25,13 @@ export type PullRequestMergeDetails = {
 export type PullRequestMergeStateResolver = (
   companyId: string,
   reference: GitHubPullRequestReference,
+  signal?: AbortSignal,
 ) => Promise<PullRequestMergeState>;
 
 export type PullRequestMergeDetailsResolver = (
   companyId: string,
   reference: GitHubPullRequestReference,
+  signal?: AbortSignal,
 ) => Promise<PullRequestMergeDetails>;
 
 export const PULL_REQUEST_CACHE_MAX_ENTRIES = 1_000;
@@ -86,14 +88,15 @@ function readRecord(value: unknown): Record<string, unknown> | null {
     : null;
 }
 
-export function createPullRequestMergeDetailsResolver(db: Db): PullRequestMergeDetailsResolver {
-  const resolver = createGitHubExternalObjectProvider(db).resolvers
+export function createPullRequestMergeDetailsResolver(db: Db, options?: GitHubExternalObjectProviderOptions): PullRequestMergeDetailsResolver {
+  const resolver = createGitHubExternalObjectProvider(db, options).resolvers
     .find((candidate) => candidate.objectType === "pull_request") ?? null;
 
-  return async (companyId, reference) => {
+  return async (companyId, reference, signal) => {
     if (!resolver) return { state: "unknown", headRef: null, headSha: null };
     const result = await resolver.resolve({
       companyId,
+      signal,
       object: {
         externalId: `${reference.owner}/${reference.repo}#pull/${reference.number}`,
         sanitizedCanonicalUrl: `https://github.com/${reference.owner}/${reference.repo}/pull/${reference.number}`,
@@ -123,7 +126,7 @@ export function createPullRequestMergeDetailsResolver(db: Db): PullRequestMergeD
   };
 }
 
-export function createPullRequestMergeStateResolver(db: Db): PullRequestMergeStateResolver {
-  const resolveDetails = createPullRequestMergeDetailsResolver(db);
-  return async (companyId, reference) => (await resolveDetails(companyId, reference)).state;
+export function createPullRequestMergeStateResolver(db: Db, options?: GitHubExternalObjectProviderOptions): PullRequestMergeStateResolver {
+  const resolveDetails = createPullRequestMergeDetailsResolver(db, options);
+  return async (companyId, reference, signal) => (await resolveDetails(companyId, reference, signal)).state;
 }

server/src/services/issue-thread-interactions.ts

@@ -204,6 +204,7 @@ export type IssueThreadInteractionServiceOptions = {
   resolvePullRequestState?: (
     companyId: string,
     reference: GitHubPullRequestReference,
+    signal?: AbortSignal,
   ) => Promise<PullRequestMergeState>;
   wakeup?: InteractionWakeup;
   pullRequestCacheTtlMs?: number;
@@ -2045,10 +2046,11 @@ export function issueThreadInteractionService(
   async function resolvePullRequestState(
     companyId: string,
     reference: GitHubPullRequestReference,
+    signal?: AbortSignal,
   ): Promise<PullRequestMergeState> {
     if (opts.resolvePullRequestState)
-      return opts.resolvePullRequestState(companyId, reference);
-    return defaultPullRequestStateResolver?.(companyId, reference) ?? "unknown";
+      return opts.resolvePullRequestState(companyId, reference, signal);
+    return defaultPullRequestStateResolver?.(companyId, reference, signal) ?? "unknown";
   }
 
   async function resolvePullRequestStates(
@@ -2057,6 +2059,7 @@ export function issueThreadInteractionService(
       companyId: string;
       reference: GitHubPullRequestReference;
     }>,
+    signal?: AbortSignal,
   ) {
     const states = new Map<string, PullRequestMergeState>();
     const pending = entries.slice();
@@ -2064,11 +2067,13 @@ export function issueThreadInteractionService(
     await Promise.all(
       Array.from({ length: workerCount }, async () => {
         while (pending.length > 0) {
+          signal?.throwIfAborted();
           const entry = pending.shift();
           if (!entry) return;
           const state = await resolvePullRequestState(
             entry.companyId,
             entry.reference,
+            signal,
           );
           states.set(entry.key, state);
         }
@@ -2076,11 +2081,7 @@ export function issueThreadInteractionService(
     );
     return states;
   }
-  async function mergedPullRequestConfirmationCandidates(scope?: MergeConfirmationScope) {
-    const text = sql`concat_ws(' ', ${issueThreadInteractions.title}, ${issueThreadInteractions.summary},
-      ${issueThreadInteractions.payload}->>'prompt', ${issueThreadInteractions.payload}->>'acceptLabel',
-      ${issueThreadInteractions.payload}->>'detailsMarkdown', ${issueThreadInteractions.payload}->'target'->>'key',
-      ${issueThreadInteractions.payload}->'target'->>'label', ${issueThreadInteractions.payload}->'target'->>'href')`;
+  async function mergedPullRequestConfirmationPage(scope?: MergeConfirmationScope, after?: string) {
     const rows = await db.select({ interaction: issueThreadInteractions, issue: {
       id: issues.id, companyId: issues.companyId, projectId: issues.projectId, goalId: issues.goalId,
       status: issues.status, assigneeAgentId: issues.assigneeAgentId,
@@ -2092,14 +2093,24 @@ export function issueThreadInteractionService(
       scope ? eq(issueThreadInteractions.companyId, scope.companyId) : undefined,
       scope?.issueId ? eq(issueThreadInteractions.issueId, scope.issueId) : undefined,
       scope?.interactionId ? eq(issueThreadInteractions.id, scope.interactionId) : undefined,
-      sql`${text} ~* '(merge|merging)'`,
-      sql`${text} ~* '(github\\.com/[^ /]+/[^ /]+/pull/[1-9][0-9]*|[a-z0-9_.-]+/[a-z0-9_.-]+#[1-9][0-9]*)'`,
+      after ? sql`${issueThreadInteractions.id} > ${after}::uuid` : undefined,
       sql`NOT EXISTS (SELECT 1 FROM ${toolActionRequests} WHERE ${toolActionRequests.interactionId} = ${issueThreadInteractions.id})`,
-    ));
-    return rows.flatMap(row => {
+    )).orderBy(asc(issueThreadInteractions.id)).limit(100);
+    return { rows: rows.flatMap(row => {
       const references = getMergeConfirmationPullRequestReferences(row.interaction);
       return references.length ? [{ ...row, references }] : [];
-    });
+    }), after: rows.at(-1)?.interaction.id, complete: rows.length < 100 };
+  }
+
+  async function mergedPullRequestConfirmationCandidates(scope?: MergeConfirmationScope) {
+    const candidates: Awaited<ReturnType<typeof mergedPullRequestConfirmationPage>>["rows"] = [];
+    let after: string | undefined;
+    while (true) {
+      const page = await mergedPullRequestConfirmationPage(scope, after);
+      candidates.push(...page.rows);
+      if (page.complete) return candidates;
+      after = page.after;
+    }
   }
 
   async function getIdempotentInteraction(args: {
@@ -3012,6 +3023,7 @@ export function issueThreadInteractionService(
       await emitInteractionResolvedTelemetry(db, interaction);
       return interaction;
     },
+    mergedPullRequestConfirmationPage,
     mergedPullRequestConfirmationCandidates,
     sweepMergedPullRequestConfirmations: async (
       mergedHints: Array<{
@@ -3064,7 +3076,7 @@ export function issueThreadInteractionService(
 
       const refreshedStates = await resolvePullRequestStates([
         ...uniqueReferences.values(),
-      ]);
+      ], signal);
       if (scope && [...refreshedStates.values()].some(state => state === "unknown")) {
         throw new Error("Merge confirmation provider state is unavailable");
       }

server/src/services/merged-pr-confirmation-work.ts

@@ -9,16 +9,24 @@ export const MERGE_CONFIRMATION_RECONCILE_MS = 5 * 60_000;
 export function registerMergeConfirmationWork(
   coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
   service: Pick<ReturnType<typeof issueThreadInteractionService>,
-    "mergedPullRequestConfirmationCandidates" | "sweepMergedPullRequestConfirmations">,
+    "mergedPullRequestConfirmationCandidates" | "mergedPullRequestConfirmationPage" | "sweepMergedPullRequestConfirmations">,
   canRun: () => boolean,
 ) {
+  const cursors = new Map<string, string>();
   return registerScopedDeadlineWork(coordinator, {
     queue: DELIVERY_QUEUES.mergedPullRequestConfirmations,
     decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint) as MergeConfirmationScope,
     matches: (entry, scope?: MergeConfirmationScope) => !scope || (entry.companyId === scope.companyId &&
       (!scope.issueId || entry.issueId === scope.issueId) && (!scope.interactionId || entry.id === scope.interactionId)),
-    load: async (scope?: MergeConfirmationScope) => (await service.mergedPullRequestConfirmationCandidates(scope))
-      .map(({ interaction }) => ({ id: interaction.id, companyId: interaction.companyId, issueId: interaction.issueId, dueAt: Date.now() })),
+    load: async (scope?: MergeConfirmationScope) => {
+      const key = scope ? JSON.stringify(scope) : "*";
+      const page = await service.mergedPullRequestConfirmationPage(scope, cursors.get(key));
+      if (page.complete) cursors.delete(key);
+      else cursors.set(key, page.after!);
+      return { complete: page.complete, rows: page.rows.map(({ interaction }) => ({
+        id: interaction.id, companyId: interaction.companyId, issueId: interaction.issueId, dueAt: Date.now(),
+      })) };
+    },
     check: async (entry, signal) => {
       const scope = { companyId: entry.companyId, issueId: entry.issueId, interactionId: entry.id };
       await service.sweepMergedPullRequestConfirmations([], scope, signal);