From 7e3f512ac07ca69d17426fc7c99b3a3fc42d4b32 Mon Sep 17 00:00:00 2001 From: Dotta Date: Wed, 21 Oct 2026 10:09:29 -0500 Subject: [PATCH] test: observe recovered delivery through persisted reply receipts Co-Authored-By: Paperclip --- .../runner-e2e/finalization-delivery.test.ts | 59 +++++++++++++++++++ tests/runner-e2e/finalization-delivery.ts | 37 ++++++++++++ .../runner-e2e/finalization-recovery-flow.ts | 24 ++++---- tests/runner-e2e/finalization-recovery.ts | 2 +- 4 files changed, 111 insertions(+), 11 deletions(-) create mode 100644 tests/runner-e2e/finalization-delivery.test.ts create mode 100644 tests/runner-e2e/finalization-delivery.ts diff --git a/tests/runner-e2e/finalization-delivery.test.ts b/tests/runner-e2e/finalization-delivery.test.ts new file mode 100644 index 00000000000..746501de07b --- /dev/null +++ b/tests/runner-e2e/finalization-delivery.test.ts @@ -0,0 +1,59 @@ +import { describe, expect, it, vi } from "vitest"; +import { inspectFinalizationDelivery, waitForFinalizationDelivery } from "./finalization-delivery.js"; + +function fixture() { + const binding = { companyId: "company", issueId: "task", runId: "run" }; + return { binding, agentId: "agent", runCount: 1, + run: { id: "run", agentId: "agent", companyId: "company", nativeIssueId: "task", status: "succeeded", issueCommentStatus: "satisfied", + issueCommentSatisfiedByCommentId: "reply", resultJson: { nativeResult: { summary: "Internal saved result." }, + presentationDecision: { schema: "paperclip.run_presentation_decision.v1", chosenSource: "final_agent_message", commentAction: "create", commentId: "reply" } } }, + comments: [{ id: "reply", companyId: "company", issueId: "task", authorAgentId: "agent", createdByRunId: "run", + body: "Here is your downloadable proof.", deletedAt: null }] }; +} + +describe("finalization final delivery observation", () => { + it("accepts the persisted final message through its receipt and presentation identity", () => { + const observed = inspectFinalizationDelivery(fixture()); + expect(observed.finalDeliveryCount).toBe(1); + expect(observed.deliveryReceiptMatches).toBe(true); + expect(JSON.stringify(observed)).not.toContain("downloadable proof"); + }); + it("counts an extra current-run reply with different text instead of hiding it", () => { + const state = fixture(); + state.comments[0]!.body = state.run.resultJson.nativeResult.summary; + state.comments.push({ ...state.comments[0]!, id: "duplicate", body: "A second different final reply." }); + const observed = inspectFinalizationDelivery(state); + expect(observed.finalDeliveryCount).toBe(2); + expect(observed.deliveryReceiptMatches).toBe(false); + }); + it.each(["company", "task", "run", "presentation", "receipt", "deleted", "blank"] as const)("rejects a mismatched %s final delivery", kind => { + const state = fixture(); + if (kind === "company") state.run.companyId = "foreign"; + if (kind === "task") state.comments[0]!.issueId = "foreign"; + if (kind === "run") state.run.id = "foreign"; + if (kind === "presentation") state.run.resultJson.presentationDecision.commentId = "foreign"; + if (kind === "receipt") state.run.issueCommentSatisfiedByCommentId = "foreign"; + if (kind === "deleted") state.comments[0]!.deletedAt = "2026-10-21T12:00:00Z"; + if (kind === "blank") state.comments[0]!.body = " "; + expect(inspectFinalizationDelivery(state).deliveryReceiptMatches).toBe(false); + }); + it("waits across commit, comment creation, and the later receipt write", async () => { + const state = fixture(); + const load = vi.fn().mockResolvedValueOnce({ ...state, comments: [], run: { ...state.run, issueCommentStatus: "pending", issueCommentSatisfiedByCommentId: null } }) + .mockResolvedValueOnce({ ...state, run: { ...state.run, issueCommentStatus: "pending", issueCommentSatisfiedByCommentId: null } }) + .mockResolvedValue(state); + const result = await waitForFinalizationDelivery({ ...state, deadlineAt: Date.now() + 1000, intervalMs: 0, load }); + expect(load).toHaveBeenCalledTimes(3); + expect(result.delivery.deliveryReceiptMatches).toBe(true); + expect(result.run.issueCommentSatisfiedByCommentId).toBe("reply"); + }); + it.each(["duplicate", "new-admission"] as const)("fails promptly on %s during the delivery wait", async kind => { + const state = fixture(); + if (kind === "duplicate") state.comments.push({ ...state.comments[0]!, id: "duplicate", body: "Second reply" }); + else state.runCount++; + const load = vi.fn().mockResolvedValue(state); + await expect(waitForFinalizationDelivery({ ...state, deadlineAt: Date.now() + 1000, intervalMs: 0, load })) + .rejects.toThrow(kind === "duplicate" ? "duplicate final replies" : "repeated provider admission"); + expect(load).toHaveBeenCalledOnce(); + }); +}); diff --git a/tests/runner-e2e/finalization-delivery.ts b/tests/runner-e2e/finalization-delivery.ts new file mode 100644 index 00000000000..6899ed71da9 --- /dev/null +++ b/tests/runner-e2e/finalization-delivery.ts @@ -0,0 +1,37 @@ +import { createHash } from "node:crypto"; +import { pollUntil } from "./api.js"; +import type { FinalizationCrashBinding } from "./finalization-crash-boundary.js"; + +type Row = Record; +export function inspectFinalizationDelivery(input: { binding: FinalizationCrashBinding; agentId: string; run: Row; comments: Row[] }) { + const { run, binding } = input, presentation = run.resultJson?.presentationDecision; + const replies = input.comments.filter(comment => comment.companyId === binding.companyId && comment.issueId === binding.issueId && + comment.authorAgentId === input.agentId && comment.createdByRunId === binding.runId && !comment.deletedAt); + const receiptMatches = run.id === binding.runId && run.companyId === binding.companyId && run.nativeIssueId === binding.issueId && + run.agentId === input.agentId && run.issueCommentStatus === "satisfied" && replies.length === 1 && + typeof replies[0]!.body === "string" && Boolean(replies[0]!.body.trim()) && run.issueCommentSatisfiedByCommentId === replies[0]!.id && + presentation?.schema === "paperclip.run_presentation_decision.v1" && presentation.commentId === replies[0]!.id && + ["create", "reuse"].includes(presentation.commentAction) && + ["existing_issue_comment", "final_agent_message", "semantic_result_summary", "adapter_final_response"].includes(presentation.chosenSource); + return { finalDeliveryCount: replies.length, deliveryReceiptMatches: receiptMatches, + presentation: { commentId: typeof presentation?.commentId === "string" ? presentation.commentId : null, + chosenSource: typeof presentation?.chosenSource === "string" ? presentation.chosenSource : null }, + replies: replies.map(comment => ({ id: comment.id, authorAgentId: comment.authorAgentId, createdByRunId: comment.createdByRunId, + bodySha256: createHash("sha256").update(String(comment.body ?? "").trim()).digest("hex") })) }; +} + +export async function waitForFinalizationDelivery(input: { + binding: FinalizationCrashBinding; agentId: string; deadlineAt: number; intervalMs?: number; + load(): Promise<{ run: Row; comments: Row[]; runCount: number }>; +}) { + return pollUntil({ label: "one persisted final reply and its matching delivery receipt", deadlineAt: input.deadlineAt, + intervalMs: input.intervalMs ?? 500, + load: async () => { + const state = await input.load(); + return { ...state, delivery: inspectFinalizationDelivery({ ...input, ...state }) }; + }, + accept: state => state.runCount === 1 && state.delivery.deliveryReceiptMatches, + reject: state => state.runCount !== 1 ? "Final delivery repeated provider admission" : + state.delivery.finalDeliveryCount > 1 ? "Recovery produced duplicate final replies" : + ["failed", "cancelled", "timed_out"].includes(state.run.status) ? "Recovered run failed before delivery" : undefined }); +} diff --git a/tests/runner-e2e/finalization-recovery-flow.ts b/tests/runner-e2e/finalization-recovery-flow.ts index 97a0bd6d421..23c994b76fe 100644 --- a/tests/runner-e2e/finalization-recovery-flow.ts +++ b/tests/runner-e2e/finalization-recovery-flow.ts @@ -1,3 +1,4 @@ +import { waitForFinalizationDelivery } from "./finalization-delivery.js"; import { captureStockHarness, gradeStockHire } from "./stock-harness.js"; import { createHash } from "node:crypto"; import { createReadStream } from "node:fs"; @@ -105,13 +106,13 @@ export async function runFinalizationRecoveryFlow(input: { accept: run => run?.id === observation!.binding.runId && run.status === "succeeded" && run.nativePhase === "committed" && run.resultJson?.finalizationPhase === "committed" && issue!.status === "done", reject: run => runs.length > 1 ? "Recovery repeated provider admission" : run && ["failed", "cancelled", "timed_out"].includes(run.status) ? "Original run did not recover" : undefined }); - const run = runs[0]!; + let run = runs[0]!; const provider = run.runnerProfileJson?.nativeExecutionInput?.provider; if (provider?.model !== execution.profile.model || (provider.kind === "acpx" ? `acpx:${provider.agent}` : provider.kind) !== (execution.profile.provider === "codex" ? "codex" : "acpx:claude")) throw new Error("Recovered run did not retain the configured provider and model"); - const [rows, operations, comments, interactions, view, artifacts] = await Promise.all([ + const [rows, operations, interactions, view, artifacts] = await Promise.all([ events(run.id), api.get(`/api/heartbeat-runs/${run.id}/workspace-operations`), - api.get(`/api/issues/${issue.id}/comments`), api.get(`/api/issues/${issue.id}/interactions`), + api.get(`/api/issues/${issue.id}/interactions`), api.get(`/api/issues/${issue.id}/workspace`), readRegisteredArtifacts(api, issue.id, run.id, [files.proof]), ]); if (!proof) throw new Error("Missing original proof before crash"); @@ -122,8 +123,13 @@ export async function runFinalizationRecoveryFlow(input: { throw new Error("Restored binary is missing, linked, or the wrong size"); const hash = createHash("sha256"); for await (const chunk of createReadStream(file)) hash.update(chunk); - const finalReply = comments.filter(comment => comment.authorAgentId === fixtures.agent.id && comment.createdByRunId === run.id && - comment.body?.trim() === run.resultJson.nativeResult?.summary?.trim()); + const delivered = await waitForFinalizationDelivery({ binding: observation!.binding, agentId: fixtures.agent.id, deadlineAt: input.deadlineAt, + load: async () => { + await refresh(); + return { run: runs[0] ?? {}, runCount: runs.length, comments: await api.get(`/api/issues/${issue!.id}/comments`) }; + } }); + run = delivered.run; + const delivery = delivered.delivery; await page.goto(`/${encodeURIComponent(fixtures.company.issuePrefix!)}/issues/${encodeURIComponent(issue.identifier ?? issue.id)}`); await expect(page.getByTestId("task-chat-agent-bubble").locator(`a[href*="${artifacts[0]!.attachmentId}"]`).last()).toBeVisible({ timeout: 30_000 }); observation!.final = { runIds: runs.map(candidate => candidate.id), committed: run.runtimeMode === "native" && run.nativePhase === "committed", @@ -132,8 +138,8 @@ export async function runFinalizationRecoveryFlow(input: { operation.heartbeatRunId === run.id && operation.phase === "workspace_finalize" && operation.status === "succeeded" && operation.metadata?.resumedFromOperationId) .map(operation => ({ resumedFromOperationId: operation.metadata.resumedFromOperationId, startedAt: operation.startedAt })), artifactId: artifacts[0]!.attachmentId, artifactSha256: artifacts[0]!.sha256, byteSize: stat.size, - actualSha256: hash.digest("hex"), expectedSha256: proof.sha256, proofReference: proof.reference, finalDeliveryCount: finalReply.length, - deliveryReceiptMatches: run.issueCommentStatus === "satisfied" && finalReply.length === 1 && run.issueCommentSatisfiedByCommentId === finalReply[0]!.id, + actualSha256: hash.digest("hex"), expectedSha256: proof.sha256, proofReference: proof.reference, finalDeliveryCount: delivery.finalDeliveryCount, + deliveryReceiptMatches: delivery.deliveryReceiptMatches, visibleDelivery: true, pendingInteractions: interactions.filter(interaction => interaction.status === "pending").length }; await input.evidence("finalization-after.json", { binding: observation!.binding, events: eventReceipts(rows), operations: operations.map(operation => ({ id: operation.id, companyId: operation.companyId, issueId: operation.issueId, @@ -141,9 +147,7 @@ export async function runFinalizationRecoveryFlow(input: { finishedAt: operation.finishedAt, resumedFromOperationId: operation.metadata?.resumedFromOperationId })), restoredFile: { path: files.data, byteSize: stat.size, sha256: observation!.final.actualSha256 }, delivery: { runId: run.id, status: run.issueCommentStatus, satisfiedByCommentId: run.issueCommentSatisfiedByCommentId, - summarySha256: createHash("sha256").update(String(run.resultJson.nativeResult?.summary ?? "").trim()).digest("hex"), - replies: finalReply.map(comment => ({ id: comment.id, authorAgentId: comment.authorAgentId, createdByRunId: comment.createdByRunId, - bodySha256: createHash("sha256").update(String(comment.body).trim()).digest("hex") })) } }); + presentation: delivery.presentation, replies: delivery.replies } }); const checks = gradeFinalizationRecovery(observation!); input.observe(issue, runs, checks); await save("completed"); diff --git a/tests/runner-e2e/finalization-recovery.ts b/tests/runner-e2e/finalization-recovery.ts index f5bc6188ac2..5bf1eb08582 100644 --- a/tests/runner-e2e/finalization-recovery.ts +++ b/tests/runner-e2e/finalization-recovery.ts @@ -72,7 +72,7 @@ export function gradeFinalizationRecovery(value: FinalizationRecoveryObservation export function finalizationRecoveryDigest() { const hash = createHash("sha256"); - for (const file of ["finalization-recovery.ts", "finalization-recovery-flow.ts", "finalization-crash-boundary.ts", "server.ts", "server-stop.ts", "registered-artifact.ts", "runner.spec.ts", "stock-harness.ts"]) + for (const file of ["finalization-recovery.ts", "finalization-recovery-flow.ts", "finalization-delivery.ts", "finalization-crash-boundary.ts", "server.ts", "server-stop.ts", "registered-artifact.ts", "runner.spec.ts", "stock-harness.ts"]) hash.update(file).update(readFileSync(new URL(file, import.meta.url))); return hash.digest("hex"); }