fix(native): retry an unpublished accepted task reply durably instead of dropping it (#19200)

## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - A native run commits its status decision and its final reply comment in separate steps > - Background presentation recovery publishes an accepted reply when the live path does not record one > - That worker dropped the run after one refused attempt and only threw on an error, with no record on the run > - A task could then show done while the user never saw the answer, until an unrelated wake reloaded the queue > - This pull request makes the publication of an accepted ordinary task reply durable, bounded, and recorded > - The benefit is that a task cannot close silently without its answer ## Linked Issues or Issue Description Refs #19130 (verify requested publication at finalization, for documents), #19183 (deliver messages saved while a native run finalizes), #19186 (stale owner done). This change does not duplicate them. It covers the final reply comment. **What happened?** A native run accepted a result with disposition `done`. The status decision set the task to done. The accepted final reply was not published as a task comment. The run had no publication record (no `run.presentation.resolved` event and no `presentationDecision`). The reply was published by background recovery 58 minutes later, after an unrelated wake reloaded the presentation queue. **Expected behavior** The accepted reply is published promptly. A refused or failed attempt is retried with backoff and never posts twice. If publication fails permanently, the run and the task show a named failure with an owner. **Steps to reproduce** 1. Commit an accepted native task result with disposition `done`, with the final reply not yet visible to presentation recovery. 2. Run the native response presentation worker check once. The attempt is refused. 3. On `master`, the worker returns `null` and drops the run. Nothing records the attempt. Nothing retries it until another notification for that scope. **Paperclip version or commit** `master` at 3b17b5ce15. **Deployment mode** Cloud staging, native `paperclip_runner` agent. Root cause notes: the staging evidence shows that the live completion path did not reach its presentation step for the affected run (no `lifecycle run succeeded` event). Staging server logs for that window are not available, so the exact first refusal is not proven. The worker defect is proven: one refusal makes `check()` return `null`, which deletes the scheduled entry. One thrown error escapes with only in-memory backoff and no record. The fix closes that gap for every refusal or error cause. ## What Changed - `native-response-presentation-work.ts`: when an accepted ordinary task reply (`done` or `needs_review`, not an Agent Chat conversation) is still unpublished after a due attempt, the worker records the attempt and keeps the run scheduled. A thrown error for this class is recorded instead of escaping. When the reply is published, an open retry receipt is set to `resolved`. - New `native-response-publication-receipt.ts`: stores the receipt in `heartbeat_runs.result_json.responsePublication` (attempts, last reason, last error, next attempt). Backoff is 30 s doubling to at most 5 min, for 6 attempts. Each retry appends a `run.response_publication.retrying` run event. After the last attempt it records `native_response_publication_failed` with the run's agent as owner, a `run.response_publication.failed` run event, and an `issue.response_publication_failed` task activity. - `native-response-presentation-query.ts`: the candidate query exposes the saved retry deadline, so a restarted worker reconstructs the schedule. It excludes a run whose publication is recorded as failed, so retries stay bounded. - An open retry receipt is settled to `resolved` in the same write that records the reply's `commentId` (background recovery and the live `recordRunPresentationDecision`). A candidate that disappears for another reason leaves the receipt unchanged, so the bound cannot reset. - When the reply comment already exists but its decision was not recorded, recovery now records it as `reuse` with that `commentId`. It does not post again and does not count a refusal. - Under the run lock, an attempt is not counted again when another worker already saved a future deadline. The worker also honors that saved deadline after recovery. - Idempotency is unchanged: the comment is written only while the presentation decision has no `commentId`, under the existing coordinator lock. The receipt write never overwrites a run that already has a `commentId`. - `doc/execution-semantics.md`: documents the contract. ## Verification - `cd server && npx vitest run src/services/native-runtime/native-response-publication-retry.test.ts` - Red on `master` (3 failed): after one refusal `check()` returned `null` (dropped); one injected transaction error escaped with no record; the refusal loop never stopped (more than 10 attempts, no failure recorded). - Green with this change (5 passed, including an already posted reply and two concurrent workers): one refusal keeps the run scheduled and recorded, a reloaded worker keeps the same deadline, the reply posts exactly once after it becomes visible, and repeated checks do not double-post; one injected error is recorded and retried; permanent refusal stops after bounded attempts with the named, owned failure on the run event and task activity. - Related suites pass: `agent-chat-response-recovery.test.ts`, `presentation-ownership-read.test.ts`, `idle-native-finalization.test.ts`, `chat-monitor-continuation.test.ts`, `native-preserved-monitor-wait.test.ts`, `paperclip-control-plane-port.test.ts`, `chat-channels.integration.test.ts`, `photon/channel.integration.test.ts` (2047 of 2049 passed in one combined run). One local failure needs a bundled Slack skill file that is missing in this worktree. The other ("materializes direct external-chat finals once") failed once under load and then passed three times in isolation and in the combined set with this change. - Not run: a live staging run. The change has not been deployed. ## Risks - Behavior change: an ordinary task reply that recovery keeps refusing now gets a recorded failure and a task activity after about 12 minutes. Before, it was dropped silently. A persistent integrity decline (for example a digest mismatch) now becomes visible this way. This is intended. - The status decision still commits before the reply. This PR does not hold `done` on the comment. It makes a missing reply visible with an owner. - Agent Chat conversations and yielded chat responses keep their existing behavior and their completion outbox owner. - No schema migration. The receipt lives in the existing `result_json` column. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, with tool use and code execution in Claude Code. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-08 21:01:54 +01:00 · 2026-11-05 01:19:04 -06:00
1 parent 07a2941f0c
commit 87dfba6818
@@ -1,6 +1,7 @@
 import { preserveAcceptedRunUsageProjectionSql } from "../usage-receipts.js";
 import { providerFailureFromError } from "../../vendor/paperclip-runner/index.js";
 import { nativeTaskSessionPublicationHandled } from "../native-runtime/native-task-session-publication.js";
+import { settledResponsePublicationSql } from "../native-runtime/native-response-publication-receipt.js";
 import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js";
 import { measureRunPhase } from "../execution-preparation-timing.js";
 import { PROJECT_ACCESS_DENIED, ProjectAccessDeniedError } from "../project-access-denial.js";
@@ -359,7 +360,8 @@ export async function recordRunPresentationDecision(db: Db, input: {
  await db.transaction(async tx => {
    await notifyChatRunMilestones(tx, { companyId: input.companyId, runId: input.runId, issueId: input.issueId });
    await tx.update(heartbeatRuns).set({
-      resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({ presentationDecision: input.decision })}::jsonb`,
+      resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({ presentationDecision: input.decision })}::jsonb
+        || ${typeof input.decision.commentId === "string" ? settledResponsePublicationSql() : sql`'{}'::jsonb`}`,
      updatedAt: new Date(),
    }).where(eq(heartbeatRuns.id, input.runId));
  });
@@ -8,6 +8,11 @@ export function nativeResponsePresentationCandidates(scope?: NativeResponsePrese
  return db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId,
    issueId: nativeRunFinalizations.issueId, agentId: heartbeatRuns.agentId, nextAttemptAt: nativeRunFinalizations.nextAttemptAt,
    endpointId: chatActions.endpointId, principalId: chatActions.principalId,
+    // Durable publication retry schedule for an ordinary final task reply.
+    publicationNextAttemptAt: sql<string | null>`${heartbeatRuns.resultJson} #>> '{responsePublication,nextAttemptAt}'`.as("publication_next_attempt_at"),
+    // Agent Chat conversations keep their completion outbox as the bounded owner.
+    taskReply: sql<boolean>`coalesce(${nativeRunResults.resultJson} #>> '{result,reportedWorkDisposition}' in ('done', 'needs_review')
+      and (coalesce(${issues.conversationAgentId}::text, '') = '' or coalesce(${issues.conversationUserId}, '') = ''), false)`.as("task_reply"),
  }).from(nativeRunFinalizations)
    .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, nativeRunFinalizations.runId),
      eq(heartbeatRuns.companyId, nativeRunFinalizations.companyId), eq(heartbeatRuns.nativeIssueId, nativeRunFinalizations.issueId)))
@@ -26,6 +31,9 @@ export function nativeResponsePresentationCandidates(scope?: NativeResponsePrese
      eq(nativeRunResults.schemaStatus, "accepted"),
      sql`jsonb_typeof(${heartbeatRuns.resultJson} #> '{presentationDecision,commentId}') is distinct from 'string'`,
      sql`coalesce(${heartbeatRuns.resultJson} #> '{presentationDecision,commentAction}', '"create"'::jsonb) = '"create"'::jsonb`,
+      // A reply whose bounded publication retries are exhausted stays visible
+      // as a named failure instead of being retried forever.
+      sql`${heartbeatRuns.resultJson} #>> '{responsePublication,state}' is distinct from 'failed'`,
      sql`${heartbeatRuns.contextSnapshot}->'skipIssueComment' is distinct from 'true'::jsonb`,
      sql`coalesce(${statusDecisions.decisionJson}->'externalChatReviewPresentation', 'null'::jsonb) in ('null'::jsonb, 'false'::jsonb, '0'::jsonb, '""'::jsonb)`,
      sql`${nativeRunResults.resultJson} #>> '{result,schema}' = 'paperclip.run_result.v1'`,
@@ -8,6 +8,7 @@ import { onCommittedChatResponseAuthorityRegistered } from "../durable-chat-wake
 import { registerScopedDeadlineWork } from "../scoped-deadline-work.js";
 import { recoverCommittedNativeChatResponse } from "./native-run-finalizer.js";
 import { measureRunPhase, withRunResponsePresentationTiming } from "../execution-preparation-timing.js";
+import { recordNativeResponsePublicationAttempt } from "./native-response-publication-receipt.js";

 export function registerNativeResponsePresentationWork(db: Db,
  coordinator: ReturnType<typeof createDeliveryWorkCoordinator>, canRun: () => boolean) {
@@ -16,6 +17,12 @@ export function registerNativeResponsePresentationWork(db: Db,

 type Entry = { id: string; companyId: string; issueId: string; agentId: string; endpointId?: string | null; principalId?: string | null; dueAt: number };

+type Candidate = { nextAttemptAt: Date | null; publicationNextAttemptAt: string | null };
+function dueAt(row: Candidate) {
+  const published = row.publicationNextAttemptAt ? Date.parse(row.publicationNextAttemptAt) : 0;
+  return Math.max(row.nextAttemptAt?.getTime() ?? 0, Number.isFinite(published) ? published : 0);
+}
+
 export function nativeResponsePresentationWork(db: Db) {
  const candidates = (scope?: NativeResponsePresentationScope, after?: string, runId?: string) => {
    const query = nativeResponsePresentationCandidates(scope, after, runId).as("pending_native_response");
@@ -31,7 +38,7 @@ export function nativeResponsePresentationWork(db: Db) {
    let after: string | undefined;
    while (canRun() && !signal?.aborted) {
      const rows = await measureRunPhase("response.resolve", () => candidates(scope, after));
-      for (const row of rows) entries.push({ id: row.id, companyId: row.companyId, issueId: row.issueId, agentId: row.agentId, endpointId: row.endpointId, principalId: row.principalId, dueAt: row.nextAttemptAt?.getTime() ?? 0 });
+      for (const row of rows) entries.push({ id: row.id, companyId: row.companyId, issueId: row.issueId, agentId: row.agentId, endpointId: row.endpointId, principalId: row.principalId, dueAt: dueAt(row) });
      if (rows.length < 100) return entries;
      after = rows[rows.length - 1].id;
    }
@@ -45,13 +52,32 @@ export function nativeResponsePresentationWork(db: Db) {
    const scope = { companyId: entry.companyId, issueId: entry.issueId };
    const [current] = await measureRunPhase("response.resolve", () => candidates(scope, undefined, entry.id));
    if (!current) return null;
-    if (current.nextAttemptAt && current.nextAttemptAt.getTime() > Date.now()) return { ...entry, dueAt: current.nextAttemptAt.getTime() };
+    if (dueAt(current) > Date.now()) return { ...entry, dueAt: dueAt(current) };
    if (signal?.aborted) throw new Error("Native response presentation recovery stopped");
    const recoveryStartedAt = Date.now();
-    await recoverCommittedNativeChatResponse(db, { ...scope, runId: entry.id, onlyWhenDue: true });
+    let failure: unknown = null;
+    try {
+      await recoverCommittedNativeChatResponse(db, { ...scope, runId: entry.id, onlyWhenDue: true });
+    } catch (error) {
+      // Ordinary task replies own a durable, bounded retry below. Other
+      // presentation classes keep the coordinator's in-memory backoff.
+      if (!current.taskReply) throw error;
+      failure = error;
+    }
    const [remaining] = await measureRunPhase("response.resolve", () => candidates(scope, undefined, entry.id));
-    return remaining?.nextAttemptAt && remaining.nextAttemptAt.getTime() > recoveryStartedAt
-      ? { ...entry, dueAt: remaining.nextAttemptAt.getTime() } : null;
+    // A published reply settles its retry receipt in the same write that
+    // records the comment, so a vanished candidate needs no cleanup here.
+    if (!remaining) return null;
+    // The coordinator, or another worker's saved retry, already moved the deadline.
+    if (dueAt(remaining) > recoveryStartedAt) return { ...entry, dueAt: dueAt(remaining) };
+    if (!remaining.taskReply) return null;
+    // The accepted reply is still unpublished after a due attempt. Never drop
+    // it silently: schedule a bounded retry or record the named failure.
+    const attempt = await recordNativeResponsePublicationAttempt(db, {
+      companyId: entry.companyId, issueId: entry.issueId, runId: entry.id,
+      reason: failure ? "publication_error" : "publication_refused", error: failure,
+    });
+    return attempt?.state === "retrying" ? { ...entry, dueAt: attempt.nextAttemptAt } : null;
  }
  function register(coordinator: ReturnType<typeof createDeliveryWorkCoordinator>, canRun: () => boolean) {
    const worker = registerScopedDeadlineWork(coordinator, {
@@ -0,0 +1,136 @@
+import { and, eq, sql } from "drizzle-orm";
+import { heartbeatRuns, type Db } from "@paperclipai/db";
+import { logActivity } from "../activity-log.js";
+import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js";
+
+/**
+ * Durable, bounded publication of an accepted native task reply.
+ *
+ * The status decision for an accepted result commits independently of the
+ * final reply comment. When publication is refused or fails, the attempt is
+ * recorded on the run and retried with backoff. Exhausted retries leave a
+ * named failure with an owner on the run and the task instead of a silent
+ * done. The comment write itself stays idempotent through the presentation
+ * decision's `commentId`, so retries never double-post.
+ */
+export const NATIVE_RESPONSE_PUBLICATION_SCHEMA = "paperclip.native_response_publication.v1";
+export const NATIVE_RESPONSE_PUBLICATION_FAILED = "native_response_publication_failed";
+export const NATIVE_RESPONSE_PUBLICATION_MAX_ATTEMPTS = 6;
+const BASE_DELAY_MS = 30_000;
+const MAX_DELAY_MS = 300_000;
+
+export type NativeResponsePublicationReason = "publication_refused" | "publication_error";
+
+type Receipt = {
+  schema: typeof NATIVE_RESPONSE_PUBLICATION_SCHEMA;
+  state: "retrying" | "failed" | "resolved";
+  attempts: number;
+  firstAttemptAt: string;
+  lastAttemptAt: string;
+  lastReason: NativeResponsePublicationReason;
+  lastError?: string | null;
+  nextAttemptAt?: string | null;
+  reasonCode?: string;
+  owner?: { kind: "agent"; agentId: string };
+  failedAt?: string;
+  resolvedAt?: string;
+};
+
+export function nativeResponsePublicationDelayMs(attempts: number) {
+  return Math.min(MAX_DELAY_MS, BASE_DELAY_MS * 2 ** Math.max(0, attempts - 1));
+}
+
+function readReceipt(resultJson: unknown): Receipt | null {
+  if (!resultJson || typeof resultJson !== "object") return null;
+  const receipt = (resultJson as Record<string, unknown>).responsePublication;
+  return receipt && typeof receipt === "object" ? receipt as Receipt : null;
+}
+
+function errorText(error: unknown) {
+  if (!error) return null;
+  const text = error instanceof Error ? error.message : String(error);
+  return text.slice(0, 500);
+}
+
+function patchReceipt(tx: Db, runId: string, receipt: Receipt) {
+  // Never overwrite a materialized reply's run record with retry bookkeeping.
+  return tx.update(heartbeatRuns).set({
+    resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb)
+      || jsonb_build_object('responsePublication', ${JSON.stringify(receipt)}::jsonb)`,
+    updatedAt: new Date(),
+  }).where(and(eq(heartbeatRuns.id, runId),
+    sql`jsonb_typeof(${heartbeatRuns.resultJson} #> '{presentationDecision,commentId}') is distinct from 'string'`))
+    .returning({ id: heartbeatRuns.id });
+}
+
+/** Record one unsuccessful attempt. Returns the next due time, or null once
+ * the bounded retry is exhausted and the failure has been recorded. */
+export async function recordNativeResponsePublicationAttempt(db: Db, input: {
+  companyId: string; issueId: string; runId: string;
+  reason: NativeResponsePublicationReason; error?: unknown;
+}): Promise<{ state: "retrying"; nextAttemptAt: number } | { state: "failed" } | null> {
+  return db.transaction(async transaction => {
+    const tx = transaction as unknown as Db;
+    const [run] = await tx.select({ id: heartbeatRuns.id, agentId: heartbeatRuns.agentId, resultJson: heartbeatRuns.resultJson })
+      .from(heartbeatRuns)
+      .where(and(eq(heartbeatRuns.id, input.runId), eq(heartbeatRuns.companyId, input.companyId)))
+      .for("update");
+    if (!run) return null;
+    const previous = readReceipt(run.resultJson);
+    if (previous?.state === "failed") return { state: "failed" as const };
+    const now = new Date();
+    // Another worker already counted this due attempt and saved the next
+    // deadline. Keep its schedule instead of counting the attempt twice.
+    const scheduled = previous?.state === "retrying" && previous.nextAttemptAt ? Date.parse(previous.nextAttemptAt) : NaN;
+    if (Number.isFinite(scheduled) && scheduled > now.getTime()) return { state: "retrying" as const, nextAttemptAt: scheduled };
+    const attempts = (previous?.state === "retrying" ? previous.attempts : 0) + 1;
+    const base = {
+      schema: NATIVE_RESPONSE_PUBLICATION_SCHEMA as typeof NATIVE_RESPONSE_PUBLICATION_SCHEMA,
+      attempts,
+      firstAttemptAt: previous?.state === "retrying" ? previous.firstAttemptAt : now.toISOString(),
+      lastAttemptAt: now.toISOString(),
+      lastReason: input.reason,
+      lastError: errorText(input.error),
+    };
+    if (attempts < NATIVE_RESPONSE_PUBLICATION_MAX_ATTEMPTS) {
+      const nextAttemptAt = new Date(now.getTime() + nativeResponsePublicationDelayMs(attempts));
+      const [updated] = await patchReceipt(tx, run.id, { ...base, state: "retrying", nextAttemptAt: nextAttemptAt.toISOString() });
+      if (!updated) return null;
+      await appendHeartbeatRunEvent(tx, {
+        companyId: input.companyId, runId: run.id, agentId: run.agentId,
+        eventType: "run.response_publication.retrying", stream: "system", level: "warn",
+        message: `Final reply publication ${input.reason === "publication_error" ? "failed" : "was refused"}; retry ${attempts + 1} of ${NATIVE_RESPONSE_PUBLICATION_MAX_ATTEMPTS} is scheduled`,
+        payload: { attempts, reason: input.reason, nextAttemptAt: nextAttemptAt.toISOString() },
+      });
+      return { state: "retrying" as const, nextAttemptAt: nextAttemptAt.getTime() };
+    }
+    const owner = { kind: "agent" as const, agentId: run.agentId };
+    const [updated] = await patchReceipt(tx, run.id, { ...base, state: "failed", nextAttemptAt: null,
+      reasonCode: NATIVE_RESPONSE_PUBLICATION_FAILED, owner, failedAt: now.toISOString() });
+    if (!updated) return null;
+    await appendHeartbeatRunEvent(tx, {
+      companyId: input.companyId, runId: run.id, agentId: run.agentId,
+      eventType: "run.response_publication.failed", stream: "system", level: "error",
+      message: `The accepted final reply could not be published to the task after ${attempts} attempts`,
+      payload: { reasonCode: NATIVE_RESPONSE_PUBLICATION_FAILED, attempts, lastReason: input.reason, owner },
+    });
+    await logActivity(tx, {
+      companyId: input.companyId, actorType: "system", actorId: "native-response-publication",
+      action: "issue.response_publication_failed", entityType: "issue", entityId: input.issueId,
+      agentId: run.agentId, runId: run.id,
+      details: { runId: run.id, reasonCode: NATIVE_RESPONSE_PUBLICATION_FAILED, attempts, lastReason: input.reason,
+        lastError: base.lastError, owner },
+    });
+    return { state: "failed" as const };
+  });
+}
+
+/** JSON object to merge into `result_json` in the same write that records a
+ * published reply's `commentId`. It settles an open retry receipt atomically
+ * with publication, and is empty when no retry is open. */
+export function settledResponsePublicationSql() {
+  return sql`(case when ${heartbeatRuns.resultJson} #>> '{responsePublication,state}' = 'retrying'
+    then jsonb_build_object('responsePublication', ((${heartbeatRuns.resultJson} -> 'responsePublication') - 'nextAttemptAt')
+      || jsonb_build_object('state', 'resolved', 'resolvedAt', ${new Date().toISOString()}::text))
+    else '{}'::jsonb end)`;
+}
@@ -0,0 +1,178 @@
+import { randomUUID } from "node:crypto";
+import { and, eq, sql } from "drizzle-orm";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { activityLog, agents, companies, completionContracts, createDb, heartbeatRunEvents, heartbeatRuns, issueComments, issues,
+  nativeRunFinalizations, 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 } from "./native-run-finalizer.js";
+import { nativeResponsePresentationWork } from "./native-response-presentation-work.js";
+
+// An accepted native task result whose final reply was never published must
+// not be abandoned after one refused or failed publication attempt. Retries
+// are durable and bounded, never double-post, and an exhausted retry leaves a
+// named failure with an owner on the run and the task.
+const external = resolveExternalTestDatabaseUrl("PAPERCLIP_NATIVE_RESPONSE_PUBLICATION_TEST_DATABASE_URL");
+const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
+describe.skipIf(!support.supported)("native final reply publication retry", () => {
+  let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
+  let db: Db;
+  beforeAll(async () => {
+    if (!external) database = await startEmbeddedPostgresTestDatabase("native-response-publication-");
+    db = createDb(external ?? database!.connectionString);
+  }, 60_000);
+  afterAll(async () => { await db?.$client.end(); await database?.cleanup(); });
+
+  const reply = "Keyboard shortcuts are now available on every plan.";
+  async function fixture() {
+    const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID();
+    const sessionId = randomUUID(), runnerId = randomUUID();
+    await db.insert(companies).values({ id: companyId, name: "Reply publication fixture", issuePrefix: companyId.slice(0, 8) });
+    await db.insert(agents).values({ id: agentId, companyId, name: "Writer", adapterType: "paperclip_runner" });
+    await db.insert(issues).values({ id: issueId, companyId, title: "Short changelog blurb", status: "in_progress", assigneeAgentId: agentId });
+    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: "Write the blurb", criteria: [] },
+      canonicalSha256: "fixture-publication-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", contextSnapshot: { issueId, source: "assignment" } });
+    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: [] });
+    await port.completeRun({ result: { schema: "paperclip.run_result.v1", reportedWorkDisposition: "done",
+      summary: "Wrote the blurb.", completionClaim: { contractRevision: "test", objectiveSatisfied: true, criteria: [], remainingWork: [] },
+      evidence: [], verification: [], attentionRequests: [], artifacts: [] },
+      terminal: { schema: "paperclip.prp.terminal.v1", runTerminalState: "succeeded", turnTerminalState: "completed", reportedWorkDisposition: "done" },
+      turnId: "blurb-turn" });
+    await finalizeNativeRun({ db, runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: false });
+    expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)))[0].phase).toBe("committed");
+    await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, runId));
+    const prpEvent = { runId, normalizedSessionId: sessionId, turnId: "blurb-turn", sourceEventId: "fixture-final-reply",
+      payload: { kind: "agentMessage", channel: "final", providerPhase: "final", text: reply } };
+    const saveFinalReply = () => db.insert(heartbeatRunEvents).values({ companyId, runId, agentId, seq: 170, eventType: "item.completed",
+      payload: { prpEvent } });
+    return { companyId, issueId, runId, agentId, saveFinalReply };
+  }
+  const comments = (runId: string) => db.select().from(issueComments).where(eq(issueComments.createdByRunId, runId));
+  const run = async (runId: string) => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)))[0];
+  const receipt = async (runId: string) => ((await run(runId)).resultJson as Record<string, any> | null)?.responsePublication;
+  // Advance the saved deadline instead of the clock so the database and the
+  // worker agree on "due" without faking timers inside the driver.
+  const makeDue = (runId: string) => db.execute(sql`update heartbeat_runs set result_json = jsonb_set(result_json,
+    '{responsePublication,nextAttemptAt}', to_jsonb('2000-01-01T00:00:00.000Z'::text)) where id = ${runId}`);
+
+  it("keeps an accepted reply scheduled after a single refused publication and posts it exactly once", async () => {
+    const f = await fixture();
+    const work = nativeResponsePresentationWork(db);
+    const [entry] = await work.load({ companyId: f.companyId, issueId: f.issueId });
+    expect(entry).toMatchObject({ id: f.runId });
+
+    // The final reply is not visible yet, so publication is refused once.
+    const before = Date.now();
+    const rescheduled = await work.check(entry);
+    expect(await comments(f.runId)).toEqual([]);
+    expect(rescheduled).not.toBeNull();
+    expect(rescheduled!.dueAt).toBeGreaterThan(before);
+    expect(await receipt(f.runId)).toMatchObject({ state: "retrying", attempts: 1, lastReason: "publication_refused" });
+    // A restarted worker reconstructs the saved schedule instead of losing it.
+    const [reloaded] = await nativeResponsePresentationWork(db).load({ companyId: f.companyId, issueId: f.issueId });
+    expect(reloaded).toMatchObject({ id: f.runId, dueAt: rescheduled!.dueAt });
+    // Not yet due: no second attempt.
+    await expect(work.check(reloaded)).resolves.toMatchObject({ dueAt: rescheduled!.dueAt });
+    expect((await receipt(f.runId)).attempts).toBe(1);
+
+    await f.saveFinalReply();
+    await makeDue(f.runId);
+    const [due] = await work.load({ companyId: f.companyId, issueId: f.issueId });
+    await expect(work.check(due)).resolves.toBeNull();
+    expect(await comments(f.runId)).toEqual([expect.objectContaining({ body: reply, authorAgentId: f.agentId })]);
+    expect(await receipt(f.runId)).toMatchObject({ state: "resolved", attempts: 1 });
+    expect(((await run(f.runId)).resultJson as Record<string, any>).presentationDecision.commentId).toEqual(expect.any(String));
+    // Idempotent: nothing left to publish and a repeated check never double-posts.
+    expect(await work.load({ companyId: f.companyId, issueId: f.issueId })).toEqual([]);
+    await expect(work.check(due)).resolves.toBeNull();
+    expect(await comments(f.runId)).toHaveLength(1);
+  });
+
+  it("records a single publication error durably and retries it", async () => {
+    const f = await fixture();
+    await f.saveFinalReply();
+    const failure = new Error("fixture-comment-write-failed");
+    let injected = false;
+    const failing = new Proxy(db, { get(target, key) {
+      if (key === "transaction" && !injected) {
+        return (callback: Parameters<Db["transaction"]>[0]) => db.transaction(async tx => {
+          // The publication transaction is the only one that locks the
+          // finalization coordinator; fail exactly that one attempt.
+          injected = true; await callback(tx); throw failure;
+        });
+      }
+      return Reflect.get(target, key);
+    } });
+    const work = nativeResponsePresentationWork(failing as Db);
+    const [entry] = await work.load({ companyId: f.companyId, issueId: f.issueId });
+    const rescheduled = await work.check(entry);
+    expect(rescheduled).not.toBeNull();
+    expect(await comments(f.runId)).toEqual([]);
+    expect(await receipt(f.runId)).toMatchObject({ state: "retrying", attempts: 1, lastReason: "publication_error",
+      lastError: "fixture-comment-write-failed" });
+    await makeDue(f.runId);
+    await expect(work.check(entry)).resolves.toBeNull();
+    expect(await comments(f.runId)).toEqual([expect.objectContaining({ body: reply })]);
+  });
+
+  it("records an already posted reply without posting again or reporting a failure", async () => {
+    const f = await fixture();
+    await f.saveFinalReply();
+    // The live path wrote the reply but stopped before recording its decision.
+    const [posted] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, authorAgentId: f.agentId,
+      createdByRunId: f.runId, body: reply }).returning();
+    const work = nativeResponsePresentationWork(db);
+    const [entry] = await work.load({ companyId: f.companyId, issueId: f.issueId });
+    await expect(work.check(entry)).resolves.toBeNull();
+    expect(await comments(f.runId)).toEqual([posted]);
+    expect(((await run(f.runId)).resultJson as Record<string, any>).presentationDecision).toMatchObject({
+      commentAction: "reuse", commentId: posted.id });
+    expect(await receipt(f.runId)).toBeUndefined();
+    expect(await work.load({ companyId: f.companyId, issueId: f.issueId })).toEqual([]);
+  });
+
+  it("counts one due attempt once when two workers check it concurrently", async () => {
+    const f = await fixture();
+    const [entry] = await nativeResponsePresentationWork(db).load({ companyId: f.companyId, issueId: f.issueId });
+    const results = await Promise.all([nativeResponsePresentationWork(db).check(entry), nativeResponsePresentationWork(db).check(entry)]);
+    expect(results.every(result => result !== null)).toBe(true);
+    expect(await receipt(f.runId)).toMatchObject({ state: "retrying", attempts: 1 });
+    expect(await comments(f.runId)).toEqual([]);
+  });
+
+  it("stops after bounded attempts with a named, owned failure on the run and the task", async () => {
+    const f = await fixture();
+    const work = nativeResponsePresentationWork(db);
+    let attempts = 0;
+    for (;;) {
+      const [entry] = await work.load({ companyId: f.companyId, issueId: f.issueId });
+      if (!entry) break;
+      attempts += 1;
+      expect(attempts).toBeLessThanOrEqual(10);
+      await makeDue(f.runId).catch(() => undefined);
+      await work.check({ ...entry, dueAt: 0 });
+    }
+    expect(await comments(f.runId)).toEqual([]);
+    const failed = await receipt(f.runId);
+    expect(failed).toMatchObject({ state: "failed", reasonCode: "native_response_publication_failed",
+      owner: { kind: "agent", agentId: f.agentId } });
+    expect(failed.attempts).toBeGreaterThan(1);
+    const events = await db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, f.runId),
+      eq(heartbeatRunEvents.eventType, "run.response_publication.failed")));
+    expect(events).toEqual([expect.objectContaining({ level: "error" })]);
+    const activity = await db.select().from(activityLog).where(and(eq(activityLog.entityId, f.issueId),
+      eq(activityLog.action, "issue.response_publication_failed")));
+    expect(activity).toEqual([expect.objectContaining({ runId: f.runId, agentId: f.agentId })]);
+    // Bounded: the failed reply is no longer rescheduled.
+    expect(await work.load({ companyId: f.companyId, issueId: f.issueId })).toEqual([]);
+  });
+});
@@ -96,6 +96,7 @@ import {
  selectHeartbeatRunFinalAgentMessage,
 } from "../heartbeat-run-summary.js";
 import { resolveChatRunPresentationAuthorizationReason } from "../chat-run-publications.js";
+import { settledResponsePublicationSql } from "./native-response-publication-receipt.js";
 import { hasChatRunOwnedProviderInteraction } from "../chat-interaction-arbitration.js";
 import {
  authorizeCommittedChatResponse,
@@ -1036,6 +1037,17 @@ async function recoverCommittedNativeChatResponseInScope(
          existingComment: findHeartbeatRunCompletionComment(comments, run.resultJson),
          finalAgentMessage,
        });
+        if (resolved.decision.commentAction === "reuse" && typeof resolved.decision.commentId === "string") {
+          // The reply is already on the task (for example, the live path wrote
+          // it but did not record its decision). Record it; never post again.
+          await tx.update(heartbeatRuns).set({
+            resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({
+              presentationDecision: { ...resolved.decision,
+                reasonCodes: [...resolved.decision.reasonCodes, "committed_task_response_reused"] },
+            })}::jsonb || ${settledResponsePublicationSql()}`, updatedAt: new Date(),
+          }).where(eq(heartbeatRuns.id, run.id));
+          return false;
+        }
        if (!resolved.text || resolved.decision.commentAction !== "create") return false;
        await notifyChatRunMilestones(tx, input);
        const comment = await issueService(db).addComment(input.issueId, resolved.text,
@@ -1045,7 +1057,7 @@ async function recoverCommittedNativeChatResponseInScope(
          resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({
            presentationDecision: { ...resolved.decision, commentId: comment.id,
              reasonCodes: [...resolved.decision.reasonCodes, "committed_task_response_recovered"] },
-          })}::jsonb`, updatedAt: new Date(),
+          })}::jsonb || ${settledResponsePublicationSql()}`, updatedAt: new Date(),
        }).where(eq(heartbeatRuns.id, run.id));
        agentId = run.agentId;
        conversationReply = conversation;