fix(server): schedule only outstanding merge confirmations

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

server/src/__tests__/merged-pr-confirmation-sweep.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 { inArray } from "drizzle-orm";
@@ -131,7 +132,8 @@ describe("merged pull-request confirmation extraction", () => {
   });
 });
 
-const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
+const external = resolveExternalTestDatabaseUrl("PAPERCLIP_MERGE_SWEEP_TEST_DATABASE_URL");
+const embeddedPostgresSupport = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
 const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
 
 describeEmbeddedPostgres("merged pull-request confirmation sweep", () => {
@@ -139,8 +141,8 @@ describeEmbeddedPostgres("merged pull-request confirmation sweep", () => {
   let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
 
   beforeAll(async () => {
-    tempDb = await startEmbeddedPostgresTestDatabase("paperclip-merged-pr-confirmations-");
-    db = createDb(tempDb.connectionString);
+    if (!external) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-merged-pr-confirmations-");
+    db = createDb(external ?? tempDb!.connectionString, { maxConnections: 6 });
   }, 20_000);
 
   afterEach(async () => {
@@ -153,6 +155,7 @@ describeEmbeddedPostgres("merged pull-request confirmation sweep", () => {
   });
 
   afterAll(async () => {
+    await db?.$client.end({ timeout: 1 });
     await tempDb?.cleanup();
   });
 
@@ -276,7 +279,7 @@ describeEmbeddedPostgres("merged pull-request confirmation sweep", () => {
       repo: "paperclip",
       number: 39,
     }])).resolves.toEqual({
-      checked: 6,
+      checked: 3,
       candidates: 3,
       accepted: 2,
       woken: 2,

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

@@ -0,0 +1,278 @@
+import { randomUUID } from "node:crypto";
+import { setTimeout as delay } from "node:timers/promises";
+import { eq, sql } from "drizzle-orm";
+import { activityLog, agents, chatActions, chatConversations, chatEndpoints, chatPublications, companies, createDb, issues, issueThreadInteractions, toolApplications, toolConnections, type Db } from "@paperclipai/db";
+import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
+import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
+import { issueThreadInteractionService } from "../services/issue-thread-interactions.js";
+import { issueService } from "../services/issues.js";
+import { registerMergeConfirmationWork, MERGE_CONFIRMATION_RECONCILE_MS } from "../services/merged-pr-confirmation-work.js";
+import { idleWorkSnapshot } from "../services/task-admission.js";
+import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
+import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
+
+const external = resolveExternalTestDatabaseUrl("PAPERCLIP_MERGE_WORK_TEST_DATABASE_URL");
+const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
+(support.supported ? describe : describe.skip)("merge confirmation work with PostgreSQL and simulated GitHub", () => {
+  let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
+  let db: ReturnType<typeof createDb>;
+  let companyId: string, issueId: string, agentId: string;
+  let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
+  let draining = false, standby = false;
+  const errors: unknown[] = [];
+  const provider = vi.fn(async (_companyId: string, _reference: { number: number }): Promise<"open" | "merged" | "unknown"> => "merged");
+  const wakeup = vi.fn(async () => ({ id: "simulated-durable-wake" }));
+  beforeAll(async () => {
+    if (!external) temporary = await startEmbeddedPostgresTestDatabase("merged-pr-work-");
+    db = createDb(external ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
+  }, 30_000);
+  beforeEach(async () => {
+    errors.length = 0; draining = false; standby = false;
+    provider.mockReset().mockResolvedValue("merged"); wakeup.mockClear();
+    companyId = randomUUID(); agentId = randomUUID(); issueId = randomUUID();
+    await db.insert(companies).values({ id: companyId, name: "Merge worker", issuePrefix: `M${companyId.slice(0, 6)}` });
+    await db.insert(agents).values({ id: agentId, companyId, name: "Coder", role: "engineer", status: "idle", adapterType: "codex_local" });
+    await db.insert(issues).values({ id: issueId, companyId, title: "Merge task", status: "todo", assigneeAgentId: agentId });
+    await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`)));
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: false });
+  });
+  afterEach(async () => {
+    await coordinator?.stop(); coordinator = undefined;
+    vi.useRealTimers(); vi.restoreAllMocks();
+    await db.delete(chatActions); await db.delete(chatPublications); await db.delete(chatConversations); await db.delete(chatEndpoints);
+    await db.delete(toolConnections); await db.delete(toolApplications);
+    await db.delete(activityLog); await db.delete(issueThreadInteractions); await db.delete(issues);
+    await db.delete(agents); await db.delete(companies);
+  });
+  afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); });
+  const service = (database: Db = db) => issueThreadInteractionService(database, { resolvePullRequestState: provider, wakeup });
+  const input = (number = 39) => ({ kind: "request_confirmation" as const, title: `Merge paperclipai/paperclip#${number}?`,
+    resolverPolicy: "human_only" as const, continuationPolicy: "wake_assignee_on_accept" as const,
+    payload: { version: 1 as const, prompt: "Merge the PR?" } });
+  const create = (database: Db = db, number = 39) => service(database).create({ id: issueId, companyId }, input(number), { agentId });
+  const stored = (id: string) => db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, id)).then(rows => rows[0]);
+  async function start() {
+    coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining && !standby,
+      canReconcile: () => !standby, onError: error => errors.push(error) });
+    const worker = registerMergeConfirmationWork(coordinator, service(), () => !draining && !standby);
+    await worker.ready;
+    return worker;
+  }
+  async function waitFor(check: () => Promise<void> | void) {
+    for (let attempt = 0; attempt < 100; attempt++) {
+      await vi.advanceTimersByTimeAsync(1000);
+      await delay(10);
+      try { await check(); return; } catch (error) { if (attempt === 99) throw error; }
+    }
+  }
+  const settle = () => waitFor(() => expect(idleWorkSnapshot().active).toBe(0));
+  async function quiet() {
+    await waitFor(() => { expect(idleWorkSnapshot().active).toBe(0); expect(coordinator!.nextWakeAt()).toBeNull(); });
+    const reads = vi.spyOn(db, "select"), transactions = vi.spyOn(db, "transaction"), executes = vi.spyOn(db, "execute");
+    const calls = provider.mock.calls.length;
+    await vi.advanceTimersByTimeAsync(3 * MERGE_CONFIRMATION_RECONCILE_MS);
+    expect(reads).not.toHaveBeenCalled(); expect(transactions).not.toHaveBeenCalled(); expect(executes).not.toHaveBeenCalled();
+    expect(provider).toHaveBeenCalledTimes(calls);
+    reads.mockRestore(); transactions.mockRestore(); executes.mockRestore();
+  }
+
+  it("settles empty and configured idle, then ordinary creation accepts, audits, wakes once and becomes quiet with retained history", async () => {
+    await start(); await quiet(); expect(provider).not.toHaveBeenCalled();
+    const card = await create();
+    await waitFor(async () => expect((await stored(card.id)).status).toBe("accepted"));
+    await quiet();
+    expect(wakeup).toHaveBeenCalledTimes(1);
+    expect(wakeup).toHaveBeenCalledWith(agentId, expect.objectContaining({ idempotencyKey: `interaction:${card.id}:accepted`, allowRunCoalescing: false }));
+    expect(await db.select().from(activityLog)).toEqual(expect.arrayContaining([expect.objectContaining({
+      action: "issue.thread_interaction_accepted", actorId: "system:pr-merged", details: expect.objectContaining({ interactionId: card.id }),
+    })]));
+    expect(errors).toEqual([]);
+  });
+
+  it("reconciles an open candidate at its deadline, never early, then stops after merge", async () => {
+    provider.mockResolvedValue("open"); await start(); const card = await create();
+    await waitFor(() => expect(provider).toHaveBeenCalledTimes(1)); await settle();
+    const dueAt = coordinator!.nextWakeAt()!;
+    expect(dueAt).toBeGreaterThan(Date.now());
+    await vi.advanceTimersByTimeAsync(dueAt - Date.now() - 1); expect(provider).toHaveBeenCalledTimes(1);
+    provider.mockResolvedValue("merged"); await vi.advanceTimersByTimeAsync(1);
+    await waitFor(async () => expect((await stored(card.id)).status).toBe("accepted")); await quiet();
+    expect(provider).toHaveBeenCalledTimes(2); expect(wakeup).toHaveBeenCalledTimes(1);
+  });
+
+  it("manual cancellation removes the candidate deadline without provider work", async () => {
+    provider.mockResolvedValue("open"); await start(); const card = await create();
+    await waitFor(() => expect(provider).toHaveBeenCalledTimes(1)); await settle();
+    await service().withdrawInteraction({ id: issueId, companyId }, card.id, {}, { agentId });
+    await quiet(); expect(provider).toHaveBeenCalledTimes(1); expect(wakeup).not.toHaveBeenCalled();
+    expect((await stored(card.id)).status).toBe("cancelled");
+  });
+
+  it("does not accept before caller-owned outer commit and forgets a rolled-back create", async () => {
+    await start(); await quiet();
+    let release!: () => void, created!: () => void;
+    const barrier = new Promise<void>(resolve => { release = resolve; });
+    const inserted = new Promise<void>(resolve => { created = resolve; });
+    let id = "";
+    const transaction = db.transaction(async tx => {
+      id = (await create(tx as unknown as Db)).id; created(); await barrier;
+    });
+    await inserted;
+    await vi.advanceTimersByTimeAsync(10_000); expect(provider).not.toHaveBeenCalled();
+    expect(await stored(id)).toBeUndefined(); release(); await transaction;
+    await waitFor(async () => expect((await stored(id)).status).toBe("accepted")); await quiet();
+    const prior = provider.mock.calls.length;
+    await expect(db.transaction(async tx => { await create(tx as unknown as Db, 41); throw new Error("rollback"); })).rejects.toThrow("rollback");
+    await quiet(); expect(provider).toHaveBeenCalledTimes(prior); expect(errors).toEqual([]);
+  });
+
+  it("retains provider failures across drain, resumes without overlap, and clears retry after success", async () => {
+    provider.mockRejectedValue(new Error("GitHub unavailable")); await start(); const card = await create();
+    await waitFor(() => expect(errors).toHaveLength(1)); await settle();
+    expect((await stored(card.id)).status).toBe("pending");
+    draining = true;
+    const reads = vi.spyOn(db, "select"), calls = provider.mock.calls.length;
+    await vi.advanceTimersByTimeAsync(120_000);
+    expect(reads).not.toHaveBeenCalled(); expect(provider).toHaveBeenCalledTimes(calls); reads.mockRestore();
+    draining = false; provider.mockResolvedValue("merged");
+    await waitFor(async () => expect((await stored(card.id)).status).toBe("accepted")); await quiet();
+    expect(wakeup).toHaveBeenCalledTimes(1);
+  });
+
+  it("keeps first work committed during drain and startup during standby until resume", async () => {
+    standby = true; await start(); const card = await create();
+    await vi.advanceTimersByTimeAsync(60_000); expect(provider).not.toHaveBeenCalled();
+    standby = false; draining = true;
+    await vi.advanceTimersByTimeAsync(60_000); expect(provider).not.toHaveBeenCalled();
+    draining = false;
+    await waitFor(async () => expect((await stored(card.id)).status).toBe("accepted")); await quiet();
+  });
+
+  it("reconstructs pending cards on restart and ignores terminal or hidden tasks", async () => {
+    provider.mockResolvedValue("open"); const card = await create(); await start();
+    await waitFor(() => expect(provider).toHaveBeenCalledTimes(1)); await settle();
+    await coordinator!.stop(); coordinator = undefined; provider.mockResolvedValue("merged");
+    await start(); await waitFor(async () => expect((await stored(card.id)).status).toBe("accepted")); await quiet();
+    const hidden = await db.transaction(async tx => {
+      const card = await create(tx as unknown as Db, 41);
+      await issueService(tx as unknown as Db).update(issueId, { hiddenAt: new Date() });
+      return card;
+    });
+    await quiet(); expect((await stored(hidden.id)).status).toBe("pending");
+    await issueService(db).update(issueId, { hiddenAt: null });
+    await waitFor(async () => expect((await stored(hidden.id)).status).toBe("accepted")); await quiet();
+  });
+
+  it("preserves a real chat binding's durable continuation outbox on automatic acceptance", async () => {
+    const applicationId = randomUUID(), connectionId = randomUUID(), endpointId = randomUUID();
+    await db.insert(toolApplications).values({ id: applicationId, companyId, applicationKey: `chat:slack:${endpointId}`, name: "Slack", type: "chat", status: "active" });
+    await db.insert(toolConnections).values({ id: connectionId, companyId, applicationId, name: "Slack", uid: endpointId,
+      connectionPurpose: "channel", transport: "chat_sdk", status: "active", enabled: true });
+    await db.insert(chatEndpoints).values({ id: endpointId, companyId, connectionId, provider: "slack", publicId: endpointId,
+      assignedAgentId: agentId, status: "active" });
+    const [conversation] = await db.insert(chatConversations).values({ companyId, endpointId, issueId,
+      externalConversationId: "merge-channel", externalThreadId: "merge-thread", externalLabel: "Merge thread", state: "active" }).returning();
+    await start(); await quiet(); const card = await create();
+    await waitFor(async () => expect((await stored(card.id)).status).toBe("accepted")); await quiet();
+    expect(await db.select().from(chatActions).where(eq(chatActions.kind, "interaction_wakeup"))).toEqual([
+      expect.objectContaining({ companyId, endpointId, conversationId: conversation.id, providerActionId: `interaction_wakeup:${card.id}`,
+        payload: expect.objectContaining({ interactionId: card.id, issueId, agentId, interactionStatus: "accepted" }) }),
+    ]);
+    expect(await db.select().from(chatPublications)).toEqual(expect.arrayContaining([
+      expect.objectContaining({ idempotencyKey: `interaction:${card.id}:${endpointId}`, state: "cancelled" }),
+    ]));
+    expect(errors).toEqual([]);
+  });
+
+  it("does not lose a newer card created during provider observation or wake the superseded card", async () => {
+    let release!: () => void, observed!: () => void;
+    const barrier = new Promise<void>(resolve => { release = resolve; });
+    const observing = new Promise<void>(resolve => { observed = resolve; });
+    provider.mockImplementationOnce(async () => { observed(); await barrier; return "merged"; });
+    await start(); const old = await create();
+    await vi.advanceTimersByTimeAsync(1000); await observing;
+    const current = await create(db, 41);
+    expect((await stored(old.id)).status).toBe("expired");
+    release();
+    await waitFor(async () => expect((await stored(current.id)).status).toBe("accepted")); await quiet();
+    expect(wakeup).toHaveBeenCalledTimes(1);
+    expect(wakeup).toHaveBeenCalledWith(agentId, expect.objectContaining({ idempotencyKey: `interaction:${current.id}:accepted` }));
+    expect(errors).toEqual([]);
+  });
+
+  it("keeps unknown provider results as retryable debt and does not act before retry", async () => {
+    provider.mockResolvedValue("unknown"); await start(); const card = await create();
+    await waitFor(() => expect(errors).toHaveLength(1));
+    const next = coordinator!.nextWakeAt()!;
+    await vi.advanceTimersByTimeAsync(next - Date.now() - 1); expect(provider).toHaveBeenCalledTimes(1);
+    provider.mockResolvedValue("merged"); await vi.advanceTimersByTimeAsync(1);
+    await waitFor(async () => expect((await stored(card.id)).status).toBe("accepted")); await quiet();
+    expect(provider).toHaveBeenCalledTimes(2);
+  });
+
+  it("honors terminal issues, pure merge URL targets and continuationPolicy none", async () => {
+    await start(); await quiet();
+    const target = await service().create({ id: issueId, companyId }, { ...input(), title: "Merge the linked PR?", continuationPolicy: "none",
+      payload: { version: 1, prompt: "Merge the PR?", target: { type: "custom", key: "github-pr-39", href: "https://github.com/paperclipai/paperclip/pull/39" } } }, { agentId });
+    await waitFor(async () => expect((await stored(target.id)).status).toBe("accepted")); await quiet();
+    expect(wakeup).not.toHaveBeenCalled();
+    provider.mockClear();
+    const terminal = await db.transaction(async tx => {
+      const card = await create(tx as unknown as Db, 41);
+      await issueService(tx as unknown as Db).update(issueId, { status: "cancelled" });
+      return card;
+    });
+    await quiet(); expect((await stored(terminal.id)).status).toBe("expired");
+    expect(provider).not.toHaveBeenCalled(); expect(wakeup).not.toHaveBeenCalled();
+  });
+
+  it("rechecks hidden state under the acceptance lock when visibility changes during a provider read", async () => {
+    let release!: () => void, observed!: () => void;
+    const barrier = new Promise<void>(resolve => { release = resolve; });
+    const observing = new Promise<void>(resolve => { observed = resolve; });
+    provider.mockImplementationOnce(async () => { observed(); await barrier; return "merged"; });
+    await start(); const card = await create(); await vi.advanceTimersByTimeAsync(1000); await observing;
+    await issueService(db).update(issueId, { hiddenAt: new Date() }); release(); await quiet();
+    expect((await stored(card.id)).status).toBe("pending"); expect(wakeup).not.toHaveBeenCalled();
+    await issueService(db).update(issueId, { hiddenAt: null });
+    await waitFor(async () => expect((await stored(card.id)).status).toBe("accepted")); await quiet();
+    expect(wakeup).toHaveBeenCalledTimes(1); expect(errors).toEqual([]);
+  });
+
+  it("unrelated pending confirmations do not retain worker SQL or provider work", async () => {
+    await start(); await quiet();
+    await service().create({ id: issueId, companyId }, { ...input(), title: "Approve the plan?", payload: { version: 1, prompt: "Approve the plan?" } }, { agentId });
+    await quiet(); expect(provider).not.toHaveBeenCalled(); expect(wakeup).not.toHaveBeenCalled();
+    expect(errors).toEqual([]);
+  });
+
+  it("awaits an in-flight observation at shutdown and reconstructs its unaccepted card on restart", async () => {
+    let release!: () => void, observed!: () => void;
+    const barrier = new Promise<void>(resolve => { release = resolve; });
+    const observing = new Promise<void>(resolve => { observed = resolve; });
+    provider.mockImplementationOnce(async () => { observed(); await barrier; return "merged"; });
+    await start(); const card = await create(); await vi.advanceTimersByTimeAsync(1000); await observing;
+    let stopped = false;
+    const stopping = coordinator!.stop().then(() => { stopped = true; });
+    await delay(10); expect(stopped).toBe(false); release(); await stopping; coordinator = undefined;
+    expect((await stored(card.id)).status).toBe("pending"); expect(wakeup).not.toHaveBeenCalled();
+    const reads = vi.spyOn(db, "select");
+    await vi.advanceTimersByTimeAsync(2 * MERGE_CONFIRMATION_RECONCILE_MS); expect(reads).not.toHaveBeenCalled(); reads.mockRestore();
+    await start(); await waitFor(async () => expect((await stored(card.id)).status).toBe("accepted")); await quiet();
+    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();
+    await db.insert(companies).values({ id: foreignCompany, name: "Other company", issuePrefix: "OTH" });
+    await db.insert(issues).values({ id: foreignIssue, companyId: foreignCompany, title: "Other", status: "todo" });
+    const foreign = await service().create({ id: foreignIssue, companyId: foreignCompany }, input(), { userId: "board" });
+    const local = service();
+    const hint = [{ companyId, owner: "PAPERCLIPAI", repo: "paperclip", number: 39 }];
+    expect((await local.sweepMergedPullRequestConfirmations(hint)).accepted).toBe(1);
+    expect((await stored(own.id)).status).toBe("accepted"); expect((await stored(foreign.id)).status).toBe("pending");
+    expect(provider).not.toHaveBeenCalled();
+    expect((await local.sweepMergedPullRequestConfirmations(hint)).accepted).toBe(0); expect(wakeup).toHaveBeenCalledTimes(1);
+  });
+});

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 })),
+    mergedPullRequestConfirmationCandidates: vi.fn(async () => []),
     sweepMergedPullRequestConfirmations: vi.fn(async () => ({
       checked: 0,
       candidates: 0,
@@ -1334,7 +1335,9 @@ describe("startServer feedback export wiring", () => {
 
       expect(heartbeatServiceMock.tickAgentTimers).toHaveBeenCalledTimes(1);
       expect(externalObjectsServiceMock.refreshDueObjectsForActiveCompanies).toHaveBeenCalledTimes(1);
-      expect(issueThreadInteractionServiceMock.sweepMergedPullRequestConfirmations).toHaveBeenCalledTimes(1);
+      expect(issueThreadInteractionServiceMock.mergedPullRequestConfirmationCandidates).toHaveBeenCalledTimes(1);
+      expect(issueThreadInteractionServiceMock.sweepMergedPullRequestConfirmations).not.toHaveBeenCalled();
+      expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.mergedPullRequestConfirmations)).toBe(true);
       expect(executionWorkspaceServiceMock.sweepTerminalWorkspaces).toHaveBeenCalledTimes(1);
       expect(routineServiceMock.tickScheduledTriggers).toHaveBeenCalledTimes(1);
       const routineWorkerIndex = deliveryWorkRegisterMock.mock.calls.findIndex(([queue]) => queue === DELIVERY_QUEUES.routineSchedules);