fix(server): make issue report delivery durable

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

server/src/__tests__/issue-reports.test.ts

@@ -34,7 +34,7 @@ describeEmbeddedPostgres("issue reports", () => {
     await tempDb?.cleanup();
   });
 
-  it("deduplicates fingerprints and consumes pending reports in the target heartbeat", async () => {
+  it("updates duplicate deliveries and acknowledges reports only after delivery", async () => {
     const companyId = randomUUID();
     const originAgentId = randomUUID();
     const targetAgentId = randomUUID();
@@ -149,6 +149,35 @@ describeEmbeddedPostgres("issue reports", () => {
     expect(await db.select().from(issueReports).where(eq(issueReports.targetIssueId, targetIssueId)))
       .toHaveLength(1);
 
+    const correctedRunId = randomUUID();
+    await db.insert(heartbeatRuns).values({
+      id: correctedRunId,
+      companyId,
+      agentId: originAgentId,
+      invocationSource: "assignment",
+      triggerDetail: "system",
+      status: "running",
+      contextSnapshot: { issueId: originIssueId, taskId: originIssueId },
+    });
+    const corrected = await service.create({
+      ...input,
+      originRunId: correctedRunId,
+      report: {
+        ...input.report,
+        payload: {
+          type: "audit.result",
+          summary: "Phase 1 passed after correction",
+          data: { passed: true, checks: 8 },
+        },
+        requestWake: false,
+      },
+    });
+    expect(corrected.deduplicated).toBe(true);
+    expect(corrected.report.id).toBe(first.report.id);
+    expect(corrected.report.originRunId).toBe(correctedRunId);
+    expect(corrected.report.payload.summary).toBe("Phase 1 passed after correction");
+    expect(corrected.report.wakeRequested).toBe(false);
+
     const payload = await buildPaperclipWakePayload({
       db,
       companyId,
@@ -159,13 +188,71 @@ describeEmbeddedPostgres("issue reports", () => {
       expect.objectContaining({
         id: first.report.id,
         originIssueId,
-        originRunId,
+        originRunId: correctedRunId,
         originAgentId,
         fingerprint: "phase-1-audit",
-        payload: expect.objectContaining({ type: "audit.result" }),
+        payload: expect.objectContaining({
+          type: "audit.result",
+          summary: "Phase 1 passed after correction",
+        }),
       }),
     ]);
 
+    const pending = await db.select().from(issueReports).where(eq(issueReports.id, first.report.id));
+    expect(pending[0]?.consumedByRunId).toBeNull();
+    expect(pending[0]?.consumedAt).toBeNull();
+
+    const latestRunId = randomUUID();
+    await db.insert(heartbeatRuns).values({
+      id: latestRunId,
+      companyId,
+      agentId: originAgentId,
+      invocationSource: "assignment",
+      triggerDetail: "system",
+      status: "running",
+      contextSnapshot: { issueId: originIssueId, taskId: originIssueId },
+    });
+    await service.create({
+      ...input,
+      originRunId: latestRunId,
+      report: {
+        ...input.report,
+        payload: {
+          type: "audit.result",
+          summary: "Phase 1 final result",
+          data: { passed: true, checks: 9 },
+        },
+      },
+    });
+    await service.acknowledgePending({
+      companyId,
+      targetIssueId,
+      runId: targetRunId,
+      deliveries: [{ id: first.report.id, originRunId: correctedRunId }],
+    });
+    const stillPending = await db.select().from(issueReports).where(eq(issueReports.id, first.report.id));
+    expect(stillPending[0]?.consumedByRunId).toBeNull();
+    expect(stillPending[0]?.consumedAt).toBeNull();
+
+    const latestPayload = await buildPaperclipWakePayload({
+      db,
+      companyId,
+      runId: targetRunId,
+      contextSnapshot: { issueId: targetIssueId, taskId: targetIssueId, wakeReason: "assignment" },
+    });
+    expect(latestPayload?.reports).toEqual([
+      expect.objectContaining({
+        id: first.report.id,
+        originRunId: latestRunId,
+        payload: expect.objectContaining({ summary: "Phase 1 final result" }),
+      }),
+    ]);
+    await service.acknowledgePending({
+      companyId,
+      targetIssueId,
+      runId: targetRunId,
+      deliveries: [{ id: first.report.id, originRunId: latestRunId }],
+    });
     const consumed = await db.select().from(issueReports).where(eq(issueReports.id, first.report.id));
     expect(consumed[0]?.consumedByRunId).toBe(targetRunId);
     expect(consumed[0]?.consumedAt).not.toBeNull();

server/src/services/heartbeat.ts

@@ -4109,10 +4109,9 @@ export async function buildPaperclipWakePayload(input: {
           .then((rows) => rows[0] ?? null)
       : null);
   const pendingReports = issueId && input.runId
-    ? await issueReportService(input.db).consumePending({
+    ? await issueReportService(input.db).listPending({
         companyId: input.companyId,
         targetIssueId: issueId,
-        runId: input.runId,
       })
     : [];
   if (
@@ -11312,6 +11311,28 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
     } else {
       delete context[PAPERCLIP_WAKE_PAYLOAD_KEY];
     }
+    const pendingReportDeliveries = paperclipWakePayload?.reports.map((report) => ({
+      id: report.id,
+      originRunId: report.originRunId,
+    })) ?? [];
+    let pendingReportsAcknowledged = pendingReportDeliveries.length === 0;
+    const acknowledgePendingReports = async () => {
+      if (pendingReportsAcknowledged || !issueRef) return;
+      try {
+        await issueReportService(db).acknowledgePending({
+          companyId: agent.companyId,
+          targetIssueId: issueRef.id,
+          runId: run.id,
+          deliveries: pendingReportDeliveries,
+        });
+        pendingReportsAcknowledged = true;
+      } catch (error) {
+        logger.warn(
+          { err: error, runId: run.id, issueId: issueRef.id, reports: pendingReportDeliveries },
+          "failed to acknowledge delivered issue reports; reports will be retried",
+        );
+      }
+    };
     const taskMarkdown = buildPaperclipTaskMarkdown({
       issue: issueRef
         ? {
@@ -12809,6 +12830,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
                   : null,
               startedAt: meta.startedAt,
             });
+            await acknowledgePendingReports();
           },
           authToken: authToken ?? undefined,
         });
@@ -12896,6 +12918,9 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
       } else {
         outcome = "failed";
       }
+      if (outcome === "succeeded") {
+        await acknowledgePendingReports();
+      }
 
       const nextSessionState = resolveNextSessionState({
         adapterType: agent.adapterType,

server/src/services/issue-reports.ts

@@ -1,4 +1,5 @@
-import { and, asc, eq, inArray, isNull } from "drizzle-orm";
+import { and, asc, eq, isNull, or } from "drizzle-orm";
+import { isDeepStrictEqual } from "node:util";
 import type { Db } from "@paperclipai/db";
 import { heartbeatRuns, issueReports, issues } from "@paperclipai/db";
 import type { CreateIssueReport, IssueReport, IssueReportPayload } from "@paperclipai/shared";
@@ -103,6 +104,30 @@ export function issueReportService(db: Db) {
         ))
         .then((rows) => rows[0] ?? null);
       if (!existing) throw conflict("Issue report fingerprint already exists");
+      const deliveryChanged =
+        existing.originRunId !== input.originRunId
+        || existing.originAgentId !== input.originAgentId
+        || existing.wakeRequested !== input.report.requestWake
+        || !isDeepStrictEqual(existing.payload, input.report.payload);
+      if (deliveryChanged) {
+        const updatedAt = new Date();
+        const updated = await db
+          .update(issueReports)
+          .set({
+            originRunId: input.originRunId,
+            originAgentId: input.originAgentId,
+            payload: input.report.payload,
+            wakeRequested: input.report.requestWake,
+            consumedByRunId: null,
+            consumedAt: null,
+            updatedAt,
+          })
+          .where(eq(issueReports.id, existing.id))
+          .returning()
+          .then((rows) => rows[0] ?? null);
+        if (!updated) throw conflict("Issue report fingerprint already exists");
+        return { report: hydrateReport(updated), deduplicated: true };
+      }
       return { report: hydrateReport(existing), deduplicated: true };
     }
   }
@@ -116,35 +141,47 @@ export function issueReportService(db: Db) {
     return rows.map(hydrateReport);
   }
 
-  async function consumePending(input: {
+  async function listPending(input: {
+    companyId: string;
+    targetIssueId: string;
+    limit?: number;
+  }) {
+    const rows = await db
+      .select()
+      .from(issueReports)
+      .where(and(
+        eq(issueReports.companyId, input.companyId),
+        eq(issueReports.targetIssueId, input.targetIssueId),
+        isNull(issueReports.consumedAt),
+      ))
+      .orderBy(asc(issueReports.createdAt), asc(issueReports.id))
+      .limit(input.limit ?? DEFAULT_PENDING_REPORT_LIMIT);
+    return rows.map(hydrateReport);
+  }
+
+  async function acknowledgePending(input: {
     companyId: string;
     targetIssueId: string;
     runId: string;
-    limit?: number;
+    deliveries: Array<{ id: string; originRunId: string }>;
   }) {
-    return db.transaction(async (tx) => {
-      const pending = await tx
-        .select({ id: issueReports.id })
-        .from(issueReports)
-        .where(and(
-          eq(issueReports.companyId, input.companyId),
-          eq(issueReports.targetIssueId, input.targetIssueId),
-          isNull(issueReports.consumedAt),
-        ))
-        .orderBy(asc(issueReports.createdAt), asc(issueReports.id))
-        .limit(input.limit ?? DEFAULT_PENDING_REPORT_LIMIT)
-        .for("update", { skipLocked: true });
-      const ids = pending.map((row) => row.id);
-      if (ids.length === 0) return [];
-      const consumedAt = new Date();
-      const rows = await tx
-        .update(issueReports)
-        .set({ consumedByRunId: input.runId, consumedAt, updatedAt: consumedAt })
-        .where(and(inArray(issueReports.id, ids), isNull(issueReports.consumedAt)))
-        .returning();
-      return rows.map(hydrateReport);
-    });
+    if (input.deliveries.length === 0) return [];
+    const consumedAt = new Date();
+    const rows = await db
+      .update(issueReports)
+      .set({ consumedByRunId: input.runId, consumedAt, updatedAt: consumedAt })
+      .where(and(
+        eq(issueReports.companyId, input.companyId),
+        eq(issueReports.targetIssueId, input.targetIssueId),
+        or(...input.deliveries.map((delivery) => and(
+          eq(issueReports.id, delivery.id),
+          eq(issueReports.originRunId, delivery.originRunId),
+        ))),
+        isNull(issueReports.consumedAt),
+      ))
+      .returning();
+    return rows.map(hydrateReport);
   }
 
-  return { resolveOrigin, create, listForIssue, consumePending };
+  return { resolveOrigin, create, listForIssue, listPending, acknowledgePending };
 }