Merge current master native chat recovery into deferred receipts

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

* commit '11b49125db5ea1b274e69faf9d1abe057cdbb551':
  fix(chat): recover committed native Agent Chat replies (#18465)
  perf(heartbeat): drive retry promotion from durable deadlines (#18387)
  fix(ui): make the Coordinator composer footer opaque (#18455)
  fix(testops): verify unexpanded result job runs (#18462)
  fix(runner): preserve OpenCode written-answer defaults (#18456)
  fix(native): fence retries when replaying an audited Stop (#18457)
  test: await async outcomes and drive saved deadlines deterministically (#18416)

server/src/services/native-runtime/agent-chat-response-recovery.test.ts

@@ -0,0 +1,149 @@
+import { randomUUID } from "node:crypto";
+import { and, eq } from "drizzle-orm";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { agents, authUsers, companies, completionContracts, createDb, heartbeatRunEvents, heartbeatRuns, issueComments, issues,
+  nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db";
+import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
+import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js";
+import { NativeRunCoordinatorStore } from "./native-run-coordinator-store.js";
+import { finalizeNativeRun, recoverCommittedNativeChatResponse } from "./native-run-finalizer.js";
+import { nativeResponsePresentationWork } from "./native-response-presentation-work.js";
+
+const external = resolveExternalTestDatabaseUrl("PAPERCLIP_AGENT_CHAT_RESPONSE_TEST_DATABASE_URL");
+const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
+describe.skipIf(!support.supported)("committed Agent Chat reply recovery", () => {
+  let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
+  let db: Db;
+  beforeAll(async () => {
+    if (!external) database = await startEmbeddedPostgresTestDatabase("agent-chat-response-");
+    db = createDb(external ?? database!.connectionString);
+  }, 60_000);
+  afterAll(async () => { await db?.$client.end(); await database?.cleanup(); });
+
+  async function fixture(conversation = true) {
+    const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID();
+    const sessionId = randomUUID(), runnerId = randomUUID(), userId = `fixture-${randomUUID()}`, wakeId = randomUUID();
+    await db.insert(authUsers).values({ id: userId, name: "Conversation owner", email: `${userId}@example.test`,
+      createdAt: new Date("2020-01-01T00:00:00Z"), updatedAt: new Date("2020-01-01T00:00:00Z") });
+    await db.insert(companies).values({ id: companyId, name: "Reply recovery fixture", issuePrefix: companyId.slice(0, 8) });
+    await db.insert(agents).values({ id: agentId, companyId, name: "Conversation agent", adapterType: "paperclip_runner" });
+    await db.insert(issues).values({ id: issueId, companyId, title: "Conversation reply", status: "in_progress", assigneeAgentId: agentId,
+      ...(conversation ? { conversationAgentId: agentId, conversationUserId: userId, conversationState: "active", conversationSessionGeneration: 0 } : {}) });
+    await db.insert(issueComments).values({ id: wakeId, companyId, issueId, authorUserId: userId, body: "Hire one teammate and tell me when ready.",
+      createdAt: new Date("2020-01-01T00:00:00Z") });
+    const [contract] = await db.insert(completionContracts).values({ companyId, issueId, revision: 1,
+      schemaVersion: "paperclip.completion-contract.v1", policyVersion: "test", risk: "standard", completionAuthority: "server_arbiter",
+      incompleteCriteriaPolicy: "preserve_non_terminal", contractJson: { revision: "test", objective: "Answer the user", criteria: [] },
+      canonicalSha256: "fixture-conversation-contract", createdByActorType: "system", createdByActorId: "test" }).returning();
+    await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, nativeIssueId: issueId, nativeSessionId: sessionId,
+      runnerInstanceId: runnerId, completionContractId: contract.id, completionContractSha256: contract.canonicalSha256,
+      status: "running", runtimeMode: "native", responsibleUserId: userId,
+      contextSnapshot: { issueId, source: "issue.comment", wakeCommentId: wakeId, ...(conversation ? { conversationSessionGeneration: 0 } : {}) } });
+    await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, issueId));
+    const port = new NativeRunCoordinatorStore(db, { companyId, issueId, runId, agentId, normalizedSessionId: sessionId,
+      runnerSourceInstanceId: runnerId, completionContractId: contract.id, completionContractSha256: contract.canonicalSha256,
+      completionContractRevision: "test", completionContractCriterionIds: [] });
+    const reply = "The teammate is hired and ready for your next instruction.";
+    const prpEvent = { runId, normalizedSessionId: sessionId, turnId: "hire-turn", sourceEventId: "fixture-final-answer",
+      payload: { kind: "agentMessage", channel: "final", providerPhase: "final", text: reply } };
+    await db.insert(heartbeatRunEvents).values({ companyId, runId, agentId, seq: 170, eventType: "item.completed", payload: { prpEvent } });
+    await port.completeRun({ result: { schema: "paperclip.run_result.v1", reportedWorkDisposition: "done",
+      summary: "A completed hire, with its final message preserved separately.", completionClaim: { contractRevision: "test",
+        objectiveSatisfied: true, criteria: [], remainingWork: [] }, evidence: [], verification: [], attentionRequests: [], artifacts: [] },
+      terminal: { schema: "paperclip.prp.terminal.v1", runTerminalState: "succeeded", turnTerminalState: "completed", reportedWorkDisposition: "done" }, turnId: "hire-turn" });
+    await finalizeNativeRun({ db, runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: false });
+    await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date("2020-01-01T00:02:00Z") }).where(eq(heartbeatRuns.id, runId));
+    expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)))[0].phase).toBe("committed");
+    const input = { companyId, issueId, runId };
+    return { ...input, input, agentId, userId, wakeId, prpEvent, reply };
+  }
+  const comments = (runId: string) => db.select().from(issueComments).where(eq(issueComments.createdByRunId, runId));
+  const issue = async (issueId: string) => (await db.select().from(issues).where(eq(issues.id, issueId)))[0];
+
+  it("finds a committed final Agent Chat turn, persists the exact reply once and only then settles waiting", async () => {
+    const f = await fixture();
+    expect(await comments(f.runId)).toEqual([]);
+    expect(await issue(f.issueId)).toMatchObject({ status: "in_progress", conversationState: "active" });
+    const work = nativeResponsePresentationWork(db), entries = await work.load(f.input);
+    expect(entries).toEqual([expect.objectContaining({ id: f.runId })]);
+    await expect(work.check(entries[0])).resolves.toBeNull();
+    expect(await comments(f.runId)).toEqual([expect.objectContaining({ body: f.reply, authorAgentId: f.agentId, createdByRunId: f.runId })]);
+    expect(await issue(f.issueId)).toMatchObject({ status: "in_review", conversationState: "waiting" });
+    expect(await work.load(f.input)).toEqual([]);
+    await expect(recoverCommittedNativeChatResponse(db, f.input)).resolves.toBe(false);
+    expect(await comments(f.runId)).toHaveLength(1);
+  });
+  it("keeps a later user message actionable after recovering the earlier exact reply", async () => {
+    const f = await fixture();
+    await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, authorUserId: f.userId, body: "Another instruction",
+      createdAt: new Date("2020-01-01T00:01:00Z") });
+    await expect(recoverCommittedNativeChatResponse(db, f.input)).resolves.toBe(true);
+    expect(await comments(f.runId)).toHaveLength(1);
+    expect(await issue(f.issueId)).toMatchObject({ status: "in_progress", conversationState: "active" });
+  });
+  it("does not duplicate a reply under concurrent recovery", async () => {
+    const f = await fixture();
+    expect((await Promise.all([recoverCommittedNativeChatResponse(db, f.input), recoverCommittedNativeChatResponse(db, f.input)])).sort())
+      .toEqual([false, true]);
+    expect(await comments(f.runId)).toHaveLength(1);
+  });
+  it.each(["skip", "generation", "agent", "user", "execution", "waiting", "turn", "event-run", "session", "channel", "kind", "digest", "partial"])(
+    "keeps %s mismatches unchanged instead of publishing an unauthorized or incomplete reply", async fault => {
+      const f = await fixture();
+      if (["skip", "generation"].includes(fault)) await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: f.issueId,
+        source: "issue.comment", wakeCommentId: f.wakeId, conversationSessionGeneration: fault === "generation" ? 1 : 0,
+        ...(fault === "skip" ? { skipIssueComment: true } : {}) } }).where(eq(heartbeatRuns.id, f.runId));
+      if (fault === "user") await db.update(heartbeatRuns).set({ responsibleUserId: "another-user" }).where(eq(heartbeatRuns.id, f.runId));
+      if (fault === "agent") {
+        const other = randomUUID(); await db.insert(agents).values({ id: other, companyId: f.companyId, name: "Other agent" });
+        await db.update(issues).set({ conversationAgentId: other, assigneeAgentId: other }).where(eq(issues.id, f.issueId));
+      }
+      if (fault === "execution") {
+        const other = randomUUID(); await db.insert(heartbeatRuns).values({ id: other, companyId: f.companyId, agentId: f.agentId });
+        await db.update(issues).set({ executionRunId: other }).where(eq(issues.id, f.issueId));
+      }
+      if (fault === "waiting") await db.update(issues).set({ executionRunId: null, status: "in_review", conversationState: "waiting" }).where(eq(issues.id, f.issueId));
+      if (["turn", "event-run", "session", "channel", "kind"].includes(fault)) {
+        const event = structuredClone(f.prpEvent);
+        if (fault === "turn") event.turnId = "foreign-turn";
+        if (fault === "event-run") event.runId = randomUUID();
+        if (fault === "session") event.normalizedSessionId = randomUUID();
+        if (fault === "channel") event.payload.channel = "analysis";
+        if (fault === "kind") event.payload.kind = "reasoning";
+        await db.update(heartbeatRunEvents).set({ payload: { prpEvent: event } }).where(and(eq(heartbeatRunEvents.runId, f.runId), eq(heartbeatRunEvents.seq, 170)));
+      }
+      if (fault === "digest") await db.update(nativeRunResults).set({ canonicalSha256: "invalid-digest" }).where(eq(nativeRunResults.runId, f.runId));
+      if (fault === "partial") await db.update(nativeRunFinalizations).set({ phase: "copying_back" }).where(eq(nativeRunFinalizations.runId, f.runId));
+      const before = await issue(f.issueId);
+      await expect(recoverCommittedNativeChatResponse(db, f.input)).resolves.toBe(false);
+      expect(await comments(f.runId)).toEqual([]); expect(await issue(f.issueId)).toEqual(before);
+    },
+  );
+  it.each([false, true])("preserves a selected reply without replacing or resurrecting it (deleted=%s)", async deleted => {
+    const f = await fixture();
+    const [saved] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, authorAgentId: f.agentId,
+      createdByRunId: f.runId, body: "Already selected", ...(deleted ? { deletedAt: new Date("2020-01-01T00:03:00Z") } : {}) }).returning();
+    await db.update(heartbeatRuns).set({ resultJson: { presentationDecision: { commentId: saved.id, commentAction: "create" } } }).where(eq(heartbeatRuns.id, f.runId));
+    await expect(recoverCommittedNativeChatResponse(db, f.input)).resolves.toBe(false);
+    expect(await comments(f.runId)).toEqual([saved]);
+  });
+  it("never crosses company scope and keeps ordinary internal task response recovery intact", async () => {
+    const f = await fixture(false);
+    await expect(recoverCommittedNativeChatResponse(db, { ...f.input, companyId: randomUUID() })).resolves.toBe(false);
+    const work = nativeResponsePresentationWork(db), entries = await work.load(f.input);
+    expect(entries).toHaveLength(1); await expect(work.check(entries[0])).resolves.toBeNull();
+    expect(await comments(f.runId)).toEqual([expect.objectContaining({ body: f.reply })]);
+    expect((await issue(f.issueId)).conversationAgentId).toBeNull();
+  });
+  it("rolls back a partial reply write without changing the waiting state or consuming recovery", async () => {
+    const f = await fixture(), abort = new Error("fixture-commit-aborted");
+    const failing = new Proxy(db, { get(target, key) { return key === "transaction"
+      ? (callback: Parameters<Db["transaction"]>[0]) => db.transaction(async tx => { await callback(tx); throw abort; })
+      : Reflect.get(target, key); } });
+    await expect(recoverCommittedNativeChatResponse(failing, f.input)).rejects.toBe(abort);
+    expect(await comments(f.runId)).toEqual([]);
+    expect(await issue(f.issueId)).toMatchObject({ status: "in_progress", conversationState: "active" });
+    await expect(recoverCommittedNativeChatResponse(db, f.input)).resolves.toBe(true);
+    expect(await comments(f.runId)).toHaveLength(1);
+  });
+});

server/src/services/native-runtime/native-response-presentation-query.ts

@@ -1,6 +1,6 @@
 import { QueryBuilder } from "drizzle-orm/pg-core";
-import { and, asc, eq, gt, inArray, isNotNull, isNull, notExists, or, sql } from "drizzle-orm";
-import { heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, statusDecisions, statusDecisionEffects } from "@paperclipai/db";
+import { and, asc, eq, exists, gt, inArray, isNotNull, isNull, notExists, or, sql } from "drizzle-orm";
+import { heartbeatRunEvents, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, statusDecisions, statusDecisionEffects } from "@paperclipai/db";
 import type { NativeResponsePresentationScope } from "../native-response-presentation-notifications.js";
 
 export function nativeResponsePresentationCandidates(scope?: NativeResponsePresentationScope, after?: string, runId?: string) {
@@ -30,7 +30,24 @@ export function nativeResponsePresentationCandidates(scope?: NativeResponsePrese
       sql`${nativeRunResults.resultJson} #>> '{terminal,turnTerminalState}' = 'completed'`,
       or(
         and(eq(heartbeatRuns.status, "succeeded"),
-          sql`(coalesce(${issues.conversationAgentId}::text, '') = '' or coalesce(${issues.conversationUserId}, '') = '')`,
+          or(sql`(coalesce(${issues.conversationAgentId}::text, '') = '' or coalesce(${issues.conversationUserId}, '') = '')`,
+            and(eq(statusDecisions.reasonCode, "conversation_turn_finished"),
+              eq(issues.conversationAgentId, heartbeatRuns.agentId), eq(issues.conversationUserId, heartbeatRuns.responsibleUserId),
+              eq(issues.status, "in_progress"), eq(issues.conversationState, "active"),
+              or(isNull(issues.executionRunId), eq(issues.executionRunId, heartbeatRuns.id)),
+              sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issues.id}::text`,
+              sql`jsonb_typeof(${heartbeatRuns.contextSnapshot}->'conversationSessionGeneration') = 'number'`,
+              sql`${heartbeatRuns.contextSnapshot}->>'conversationSessionGeneration' = ${issues.conversationSessionGeneration}::text`,
+              exists(db.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents).where(and(
+                eq(heartbeatRunEvents.companyId, heartbeatRuns.companyId), eq(heartbeatRunEvents.runId, heartbeatRuns.id),
+                eq(heartbeatRunEvents.agentId, heartbeatRuns.agentId), eq(heartbeatRunEvents.eventType, "item.completed"),
+                sql`${heartbeatRunEvents.payload} #>> '{prpEvent,runId}' = ${heartbeatRuns.id}::text`,
+                sql`${heartbeatRunEvents.payload} #>> '{prpEvent,normalizedSessionId}' = ${heartbeatRuns.nativeSessionId}::text`,
+                sql`${heartbeatRunEvents.payload} #>> '{prpEvent,turnId}' = ${nativeRunResults.turnId}`,
+                sql`${heartbeatRunEvents.payload} #>> '{prpEvent,payload,kind}' = 'agentMessage'`,
+                sql`${heartbeatRunEvents.payload} #>> '{prpEvent,payload,channel}' = 'final'`,
+                sql`${heartbeatRunEvents.payload} #>> '{prpEvent,payload,text}' ~ '[^[:space:]]'`,
+              ))))),
           sql`${nativeRunResults.resultJson} #>> '{result,reportedWorkDisposition}' = 'done'`,
           sql`${nativeRunResults.resultJson} #>> '{terminal,reportedWorkDisposition}' = 'done'`),
         and(sql`${nativeRunResults.resultJson} #>> '{result,reportedWorkDisposition}' = 'yielded'`,

server/src/services/native-runtime/native-run-finalizer.ts

@@ -29,7 +29,7 @@ import { readPersistedNativeFailureEvidence } from "./native-provider-failure-ev
 import { dismissAutomaticCompletionReviews } from "./automatic-completion-reviews.js";
 import { findSupersededExternalReviewRequests } from "./superseded-external-review-requests.js";
 import { getNativeReviewAssignment, readNativeReviewAssignmentContext } from "./native-review-participant.js";
-import { conversationNativeDecision, isConversation } from "../agent-conversations.js";
+import { conversationNativeDecision, isConversation, settleConversationTurnForRun } from "../agent-conversations.js";
 import { issueTreeControlService } from "../issue-tree-control.js";
 import { randomUUID } from "node:crypto";
 import { preserveNativeWorkspaceExportLease } from "./native-workspace-export-resume.js";
@@ -747,6 +747,7 @@ async function recoverCommittedNativeChatResponseInScope(
 ): Promise<boolean> {
   let agentId: string | null = null;
   let materialized = false;
+  let conversationReply = false;
   const entered = beginRunPhase("response.transaction.entry");
   try {
     materialized = await measureRunPhase("response.transaction", () => db.transaction(async (transaction) => {
@@ -789,6 +790,7 @@ async function recoverCommittedNativeChatResponseInScope(
           id: heartbeatRuns.id,
           companyId: heartbeatRuns.companyId,
           agentId: heartbeatRuns.agentId,
+          responsibleUserId: heartbeatRuns.responsibleUserId,
           nativeIssueId: heartbeatRuns.nativeIssueId,
           nativeSessionId: heartbeatRuns.nativeSessionId,
           completionContractId: heartbeatRuns.completionContractId,
@@ -803,6 +805,8 @@ async function recoverCommittedNativeChatResponseInScope(
           contextSnapshot: sql<Record<string, unknown> | null>`case when ${heartbeatRuns.contextSnapshot} is null then null
             else jsonb_build_object(
               'skipIssueComment', ${heartbeatRuns.contextSnapshot}->'skipIssueComment',
+              'issueId', ${heartbeatRuns.contextSnapshot}->'issueId',
+              'conversationSessionGeneration', ${heartbeatRuns.contextSnapshot}->'conversationSessionGeneration',
               'source', ${heartbeatRuns.contextSnapshot}->'source',
               'externalChatContinuation', ${heartbeatRuns.contextSnapshot}->'externalChatContinuation',
               'paperclipWake', jsonb_build_object('externalInteractionContinuation',
@@ -909,12 +913,16 @@ async function recoverCommittedNativeChatResponseInScope(
         agentId = run.agentId;
         return true;
       }
-      // Copyback can be owned by reconciliation before the live heartbeat
-      // reaches its presentation step. Recover a completed internal task's
-      // exact final reply from the same accepted turn, without rerunning work
-      // or granting external-chat publication authority.
+      const conversation = isConversation(issue);
+      const currentConversation = conversation && decision.reasonCode === "conversation_turn_finished"
+        && issue.conversationAgentId === run.agentId && issue.conversationUserId === run.responsibleUserId
+        && issue.status === "in_progress" && issue.conversationState === "active"
+        && (!issue.executionRunId || issue.executionRunId === run.id)
+        && run.contextSnapshot?.issueId === issue.id
+        && typeof run.contextSnapshot.conversationSessionGeneration === "number"
+        && run.contextSnapshot.conversationSessionGeneration === issue.conversationSessionGeneration;
       if (
-        !isConversation(issue) &&
+        (!conversation || currentConversation) &&
         run.status === "succeeded" &&
         record(run.contextSnapshot).skipIssueComment !== true &&
         !isExternalChatPresentationContext(run.contextSnapshot) &&
@@ -929,6 +937,9 @@ async function recoverCommittedNativeChatResponseInScope(
             eq(heartbeatRunEvents.companyId, input.companyId),
             eq(heartbeatRunEvents.runId, run.id),
             eq(heartbeatRunEvents.eventType, "item.completed"),
+            conversation ? and(eq(heartbeatRunEvents.agentId, run.agentId),
+              sql`${heartbeatRunEvents.payload}->'prpEvent'->>'runId' = ${run.id}`,
+              sql`${heartbeatRunEvents.payload}->'prpEvent'->>'normalizedSessionId' = ${run.nativeSessionId}`) : undefined,
             sql`${heartbeatRunEvents.payload}->'prpEvent'->>'turnId' = ${accepted.turnId}`,
             sql`${heartbeatRunEvents.payload} #>> '{prpEvent,payload,kind}' = 'agentMessage'`,
             sql`${heartbeatRunEvents.payload} #>> '{prpEvent,payload,channel}' = 'final'`,
@@ -963,6 +974,7 @@ async function recoverCommittedNativeChatResponseInScope(
           })}::jsonb`, updatedAt: new Date(),
         }).where(eq(heartbeatRuns.id, run.id));
         agentId = run.agentId;
+        conversationReply = conversation;
         return true;
       }
       if (
@@ -1039,12 +1051,14 @@ async function recoverCommittedNativeChatResponseInScope(
   } finally {
     entered();
   }
-  if (materialized && agentId)
+  if (materialized && agentId) {
+    if (conversationReply) await settleConversationTurnForRun(db, input.companyId, input.runId);
     publishChatPublicationCommitSignal({
       ...input,
       agentId,
       eventType: "run.presentation.resolved",
     });
+  }
   return materialized;
 }
 

server/src/services/native-runtime/native-session-executor.test.ts

@@ -99,6 +99,7 @@ import { buildNativeHeartbeatPreparationSpans, createNativeRunTrace } from "./na
 import { NativeRunnerOwnershipUnverifiedError } from "./native-runner-ownership.js";
 import type { AdapterRuntimeEvent } from "../../adapters/index.js";
 import type { AdapterUsageCheckpoint } from "@paperclipai/adapter-utils";
+import { acknowledgedNativeStopExecutionHasStopped, isAcknowledgedNativeStop } from "../acknowledged-native-stop.js";
 
 vi.mock("node:fs", async (importOriginal) => {
   const actual = await importOriginal<typeof import("node:fs")>();
@@ -7852,6 +7853,27 @@ describe("native session cancellation", () => {
     expect(state.persistActivity).not.toHaveBeenCalled();
   });
 
+  it.each([
+    { companyId: "another-company" }, { runId: "another-run" }, { issueId: "another-issue" },
+    { scope: "issue" }, { effects: ["cancel_issue"] }, { intentAuditId: null },
+  ])("does not fence a failed retry for a foreign or stale Stop receipt %j", async changed => {
+    const intent = { schema: "paperclip.native-cancellation.v1", intentId: "native-cancellation:earlier-stop",
+      ...execution.binding, scope: "run", reasonCode: "cancellation_run_only", effects: ["release_run_resources"],
+      dispatchState: "acknowledged", dispatched: true, intentAuditId: "earlier-intent-audit",
+      acknowledgementAuditId: "earlier-ack-audit", ...changed };
+    const coordinator = { runId: execution.binding.runId, assessmentId: null, phase: "retryable_failure" };
+    const persistence = cancellationDb({ status: "failed", coordinator, resultJson: { nativeCancellation: intent } });
+    const before = structuredClone(persistence.getResultJson());
+    state.cancel.mockClear();
+    state.persistActivity.mockClear();
+    await expect(cancelNativeSession(execution.binding.runId, "retry Stop", { db: persistence.db, scope: "run" })).rejects.toThrow();
+    expect(persistence.updates).toEqual([]);
+    expect(coordinator.phase).toBe("retryable_failure");
+    expect(persistence.getResultJson()).toEqual(before);
+    expect(state.cancel).not.toHaveBeenCalled();
+    expect(state.persistActivity).not.toHaveBeenCalled();
+  });
+
   it("does not mistake a recovered running attempt for the failed retry that reserved Stop", async () => {
     const id = "11111111-1111-4111-8111-111111111111";
     const persistence = cancellationDb({ status: "running",
@@ -7877,6 +7899,39 @@ describe("native session cancellation", () => {
     expect(persistence.getResultJsonUpdateCount()).toBe(writes);
   });
 
+  it("fences a failed retry whose earlier Stop is already acknowledged without changing its receipts", async () => {
+    const intent = { schema: "paperclip.native-cancellation.v1", intentId: "native-cancellation:earlier-stop",
+      ...execution.binding, scope: "run", reasonCode: "cancellation_run_only",
+      effects: ["release_run_resources"], dispatchState: "acknowledged", dispatched: true,
+      intentAuditId: "earlier-intent-audit", acknowledgementAuditId: "earlier-ack-audit",
+      acknowledgedAt: "2026-01-01T00:00:00.000Z", decisionId: null, priorCoordinatorDecisionId: null };
+    const coordinator = { runId: execution.binding.runId, assessmentId: null, phase: "retryable_failure",
+      nextAttemptAt: new Date(Date.now() + 30_000) };
+    const persistence = cancellationDb({ status: "failed", coordinator,
+      resultJson: { nativeCancellation: intent, executionCancellation: { state: "requested", requestedAt: "later-stop" } } });
+    const before = structuredClone(persistence.getResultJson());
+    state.cancel.mockClear();
+    state.persistActivity.mockClear();
+    for (let attempt = 0; attempt < 2; attempt++) {
+      await expect(cancelNativeSession(execution.binding.runId, "retry Stop", { db: persistence.db, scope: "run" }))
+        .resolves.toMatchObject({ dispatched: true, auditId: intent.intentAuditId });
+      expect(coordinator).toMatchObject({ phase: "terminal_failure", failureCode: "native_retry_cancelled", nextAttemptAt: null });
+      expect(persistence.getResultJson()).toEqual(before);
+    }
+    expect(persistence.updates.filter(update => update.table === nativeRunFinalizations)).toHaveLength(1);
+    expect(persistence.getResultJsonUpdateCount()).toBe(0);
+    expect(state.cancel).not.toHaveBeenCalled();
+    expect(state.persistActivity).not.toHaveBeenCalled();
+    const stopped = { id: execution.binding.runId, companyId: execution.binding.companyId,
+      nativeIssueId: execution.binding.issueId, status: "cancelled", processPid: null, processGroupId: null,
+      resultJson: { ...persistence.getResultJson(), cancelledByActorType: "user", cancelledByUserId: "operator" } };
+    const lease = { provider: "daytona", providerLeaseId: "unconfirmed-provider-resource", status: "active",
+      releasedAt: null, cleanupStatus: null, metadata: {} };
+    const leaseDb = { select: () => ({ from: () => ({ where: async () => [lease] }) }) } as unknown as Db;
+    expect(isAcknowledgedNativeStop(stopped)).toBe(true);
+    await expect(acknowledgedNativeStopExecutionHasStopped(leaseDb, stopped)).resolves.toBe(false);
+  });
+
   it("uses the reserved caller UUID and permits same-intent and default retries", async () => {
     const id = "11111111-1111-4111-8111-111111111111";
     const persistence = cancellationDb({ resultJson: { startupCancellation: { cancellationRequestId: id } } });

server/src/services/native-runtime/native-session-executor.ts

@@ -7541,6 +7541,27 @@ export async function cancelNativeSession(
         runId, companyId: cancellationContext.companyId, issueId: cancellationContext.issueId,
         resultJson, requestId: requestedId, scope: options.scope ?? "run",
       }, coordinator)) throw conflict("Native retry is no longer cancellable");
+      const fenceRetryCancellation = async () => {
+        if (coordinator.phase !== "retryable_failure") return;
+        await tx
+          .update(nativeRunFinalizations)
+          .set({
+            phase: "terminal_failure",
+            nextAttemptAt: null,
+            leaseOwner: null,
+            leaseExpiresAt: null,
+            recoveryState: "blocked",
+            failureCode: "native_retry_cancelled",
+            updatedAt: new Date(),
+          })
+          .where(
+            and(
+              eq(nativeRunFinalizations.runId, runId),
+              eq(nativeRunFinalizations.companyId, cancellationContext.companyId),
+              eq(nativeRunFinalizations.phase, "retryable_failure"),
+            ),
+          );
+      };
       const existing = record(resultJson.nativeCancellation);
       const existingIntentId =
         typeof existing.intentId === "string" && existing.intentId.length > 0
@@ -7564,6 +7585,10 @@ export async function cancelNativeSession(
             : null;
         if (!existingAuditId)
           throw new Error("native_cancellation_intent_audit_missing");
+        if (retryCancellation && coordinator.phase === "retryable_failure") {
+          await notifyNativeResultCommit(tx, { companyId: cancellationContext.companyId, runId });
+          await fenceRetryCancellation();
+        }
         return {
           intentId: existingIntentId,
           auditId: existingAuditId,
@@ -7643,24 +7668,7 @@ export async function cancelNativeSession(
       // Cancellation is also an authority fence for a durable retry whose
       // preceding provider has already failed. No in-memory session is needed.
       await notifyNativeResultCommit(tx, { companyId: cancellationContext.companyId, runId });
-      await tx
-        .update(nativeRunFinalizations)
-        .set({
-          phase: "terminal_failure",
-          nextAttemptAt: null,
-          leaseOwner: null,
-          leaseExpiresAt: null,
-          recoveryState: "blocked",
-          failureCode: "native_retry_cancelled",
-          updatedAt: new Date(),
-        })
-        .where(
-          and(
-            eq(nativeRunFinalizations.runId, runId),
-            eq(nativeRunFinalizations.companyId, cancellationContext.companyId),
-            eq(nativeRunFinalizations.phase, "retryable_failure"),
-          ),
-        );
+      await fenceRetryCancellation();
       intentPublication = activity.publication;
       return {
         intentId,