diff --git a/server/src/__tests__/issue-reports.test.ts b/server/src/__tests__/issue-reports.test.ts index 428643ea759..bacdc9ce23c 100644 --- a/server/src/__tests__/issue-reports.test.ts +++ b/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(); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 60312d9a731..063096cc660 100644 --- a/server/src/services/heartbeat.ts +++ b/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, diff --git a/server/src/services/issue-reports.ts b/server/src/services/issue-reports.ts index 3a5f305d006..8e551dba94c 100644 --- a/server/src/services/issue-reports.ts +++ b/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 }; }