test: observe recovered delivery through persisted reply receipts

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

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();
+  });
+});

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<string, any>;
+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 });
+}

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<Row[]>(`/api/heartbeat-runs/${run.id}/workspace-operations`),
-      api.get<Row[]>(`/api/issues/${issue.id}/comments`), api.get<Row[]>(`/api/issues/${issue.id}/interactions`),
+      api.get<Row[]>(`/api/issues/${issue.id}/interactions`),
       api.get<Row>(`/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<Row[]>(`/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");

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");
 }