fix(feedback): claim delivery replay atomically

Use a partial unique index to serialize automatic and operator delivery retries, treat duplicate claims as already queued, and cover concurrent recovery sweeps.

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

packages/db/src/migrations/0219_vengeful_mandrill.sql

@@ -0,0 +1,2 @@
+-- paperclip:migration-safety-ignore large-create-index-not-concurrently: Drizzle migrations run transactionally, so CONCURRENTLY is unavailable because this partial index covers the new feedback-delivery key namespace and is required for atomic at-most-once delivery replay.
+CREATE UNIQUE INDEX IF NOT EXISTS "agent_wakeup_requests_feedback_delivery_idempotency_uq" ON "agent_wakeup_requests" USING btree ("company_id","idempotency_key") WHERE "agent_wakeup_requests"."idempotency_key" LIKE 'feedback_delivery:%' AND "agent_wakeup_requests"."status" <> 'skipped';

packages/db/src/migrations/meta/_journal.json

@@ -1520,6 +1520,13 @@
       "when": 1786711898729,
       "tag": "0218_mushy_jack_murdock",
       "breakpoints": true
+    },
+    {
+      "idx": 219,
+      "version": "7",
+      "when": 1786911207467,
+      "tag": "0219_vengeful_mandrill",
+      "breakpoints": true
     }
   ]
 }

packages/db/src/schema/agent_wakeup_requests.ts

@@ -40,6 +40,9 @@ export const agentWakeupRequests = pgTable(
     reviewPathRecoveryIdempotencyUq: uniqueIndex("agent_wakeup_requests_review_path_recovery_idempotency_uq")
       .on(table.companyId, table.idempotencyKey)
       .where(sql`${table.idempotencyKey} LIKE 'issue_review_path_lost:%' AND ${table.status} <> 'skipped'`),
+    feedbackDeliveryIdempotencyUq: uniqueIndex("agent_wakeup_requests_feedback_delivery_idempotency_uq")
+      .on(table.companyId, table.idempotencyKey)
+      .where(sql`${table.idempotencyKey} LIKE 'feedback_delivery:%' AND ${table.status} <> 'skipped'`),
     companyPayloadIssueIdx: index("agent_wakeup_requests_company_payload_issue_idx").on(
       table.companyId,
       sql`(${table.payload} ->> 'issueId')`,

server/src/__tests__/heartbeat-process-recovery.test.ts

@@ -6055,6 +6055,31 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
     expect(actionsAfter).toHaveLength(0);
   });
 
+  it("atomically claims one stranded-feedback replay across concurrent recovery sweeps", async () => {
+    const { companyId, issueId } = await seedStrandedFeedbackReviewFixture({
+      interactionStatus: "expired",
+    });
+    const firstHeartbeat = heartbeatService(db);
+    const secondHeartbeat = heartbeatService(db);
+
+    const results = await Promise.all([
+      firstHeartbeat.reconcileStrandedAssignedIssues(),
+      secondHeartbeat.reconcileStrandedAssignedIssues(),
+    ]);
+
+    expect(results.reduce((sum, result) => sum + result.strandedFeedbackRequeued, 0)).toBe(1);
+    const replayWakes = await db
+      .select()
+      .from(agentWakeupRequests)
+      .where(and(
+        eq(agentWakeupRequests.companyId, companyId),
+        eq(agentWakeupRequests.reason, FEEDBACK_DELIVERY_RETRY_WAKE_REASON),
+        sql`${agentWakeupRequests.payload} ->> 'issueId' = ${issueId}`,
+      ));
+    expect(replayWakes).toHaveLength(1);
+    expect(replayWakes[0]?.idempotencyKey).toMatch(/^feedback_delivery:.*:generation:1$/);
+  });
+
   it("opens one explicit recovery action when stranded feedback replay is exhausted", async () => {
     const { companyId, agentId, issueId, runId, rootWakeupRequestId, commentId } =
       await seedStrandedFeedbackReviewFixture({ retryGeneration: 1, interactionStatus: "expired" });

server/src/routes/issues.ts

@@ -112,6 +112,7 @@ import {
   FEEDBACK_DELIVERY_RECOVERY_CAUSE,
   FEEDBACK_DELIVERY_RETRY_WAKE_REASON,
   FEEDBACK_DELIVERY_WAKE_COMMENT_IDS_KEY,
+  isFeedbackDeliveryIdempotencyConflict,
 } from "../services/recovery/feedback-delivery.js";
 import { evaluateAgentInvokabilityFromDb } from "../services/agent-invokability.js";
 import { getTelemetryClient } from "../telemetry.js";
@@ -10914,6 +10915,7 @@ export function issueRoutes(
       [FEEDBACK_DELIVERY_WAKE_COMMENT_IDS_KEY]: outstandingCommentIds,
       ...(latestCommentId ? { commentId: latestCommentId, wakeCommentId: latestCommentId } : {}),
     };
+    let duplicateRetryClaim = false;
     await heartbeat.wakeup(targetAgentId, {
       source: "on_demand",
       triggerDetail: "manual",
@@ -10933,8 +10935,25 @@ export function issueRoutes(
           "An operator asked Paperclip to deliver human feedback on this task again. Handle the outstanding "
           + "comment(s) now and record a valid disposition.",
       },
+    }).catch((error: unknown) => {
+      if (isFeedbackDeliveryIdempotencyConflict(error)) {
+        duplicateRetryClaim = true;
+        return null;
+      }
+      throw error;
     });
 
+    if (duplicateRetryClaim) {
+      const states = await readStates();
+      res.json({
+        outcome: "already_queued",
+        reason: null,
+        message: "A delivery retry is already queued.",
+        feedbackDelivery: states.get(outstandingCommentIds[0] ?? "") ?? null,
+      } satisfies IssueFeedbackDeliveryRetryResponse);
+      return;
+    }
+
     await logActivity(db, {
       companyId: issue.companyId,
       actorType: actor.actorType,

server/src/services/recovery/feedback-delivery.test.ts

@@ -6,6 +6,7 @@ import {
   buildFeedbackDeliveryFingerprint,
   buildFeedbackDeliveryRetryIdempotencyKey,
   decideStrandedFeedbackDeliveryBackstop,
+  isFeedbackDeliveryIdempotencyConflict,
   isSuccessfulFeedbackDeliveryRecoveryMatch,
   readFeedbackDeliveryRunContext,
   type StrandedFeedbackDeliveryBackstopProbe,
@@ -17,6 +18,19 @@ const AGENT_ID = "agent-1";
 const ROOT_WAKE_ID = "wake-root-1";
 const COMMENT_ID = "comment-1";
 
+it("recognizes only feedback-delivery atomic claim conflicts", () => {
+  expect(isFeedbackDeliveryIdempotencyConflict({
+    cause: {
+      code: "23505",
+      constraint_name: "agent_wakeup_requests_feedback_delivery_idempotency_uq",
+    },
+  })).toBe(true);
+  expect(isFeedbackDeliveryIdempotencyConflict({
+    code: "23505",
+    constraint_name: "some_other_unique_index",
+  })).toBe(false);
+});
+
 const openProbe: StrandedFeedbackDeliveryBackstopProbe = {
   hasActiveExecutionPath: false,
   hasPendingWakeInteraction: false,

server/src/services/recovery/feedback-delivery.ts

@@ -6,6 +6,8 @@ export const FEEDBACK_DELIVERY_RECOVERY_CAUSE = "feedback_delivery_exhausted";
 export const FEEDBACK_DELIVERY_RECOVERY_ACTION_KIND = "feedback_delivery" as const;
 export const FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES = 1;
 export const FEEDBACK_DELIVERY_WAKE_COMMENT_IDS_KEY = "wakeCommentIds";
+export const FEEDBACK_DELIVERY_IDEMPOTENCY_INDEX =
+  "agent_wakeup_requests_feedback_delivery_idempotency_uq";
 
 export const STRANDED_FEEDBACK_DELIVERY_BACKSTOP_SOURCE = "issue.stranded_feedback_delivery_backstop";
 export const STRANDED_FEEDBACK_DELIVERY_BACKSTOP_INSTRUCTION =
@@ -77,6 +79,34 @@ export function buildFeedbackDeliveryManualRetryIdempotencyKey(input: {
   return `${input.fingerprint}:manual:${input.attempt}`;
 }
 
+export function isFeedbackDeliveryIdempotencyConflict(error: unknown): boolean {
+  let current: unknown = error;
+  for (let depth = 0; depth < 4 && current && typeof current === "object"; depth += 1) {
+    const candidate = current as {
+      code?: unknown;
+      constraint?: unknown;
+      constraint_name?: unknown;
+      message?: unknown;
+      cause?: unknown;
+    };
+    const constraint = candidate.constraint ?? candidate.constraint_name;
+    if (
+      candidate.code === "23505"
+      && (
+        constraint === FEEDBACK_DELIVERY_IDEMPOTENCY_INDEX
+        || (
+          typeof candidate.message === "string"
+          && candidate.message.includes(FEEDBACK_DELIVERY_IDEMPOTENCY_INDEX)
+        )
+      )
+    ) {
+      return true;
+    }
+    current = candidate.cause;
+  }
+  return false;
+}
+
 export function isSuccessfulFeedbackDeliveryRecoveryMatch(input: {
   run: {
     companyId: string;

server/src/services/recovery/index.ts

@@ -34,6 +34,7 @@ export {
   FEEDBACK_DELIVERY_RECOVERY_CAUSE,
   FEEDBACK_DELIVERY_RETRY_REASON,
   FEEDBACK_DELIVERY_RETRY_WAKE_REASON,
+  FEEDBACK_DELIVERY_IDEMPOTENCY_INDEX,
   STRANDED_FEEDBACK_DELIVERY_BACKSTOP_INSTRUCTION,
   STRANDED_FEEDBACK_DELIVERY_BACKSTOP_NEXT_ACTION,
   STRANDED_FEEDBACK_DELIVERY_BACKSTOP_SOURCE,
@@ -42,6 +43,7 @@ export {
   carriesExplicitFeedbackResume,
   decideStrandedFeedbackDeliveryBackstop,
   isEligibleFeedbackDeliveryWake,
+  isFeedbackDeliveryIdempotencyConflict,
   isSuccessfulFeedbackDeliveryRecoveryMatch,
   readFeedbackDeliveryRetryGeneration,
   readFeedbackDeliveryRunContext,

server/src/services/recovery/service.ts

@@ -91,6 +91,7 @@ import {
   STRANDED_FEEDBACK_DELIVERY_BACKSTOP_SOURCE,
   buildFeedbackDeliveryRetryIdempotencyKey,
   decideStrandedFeedbackDeliveryBackstop,
+  isFeedbackDeliveryIdempotencyConflict,
   readFeedbackDeliveryRunContext,
   type StrandedFeedbackDeliveryBackstopDecision,
 } from "./feedback-delivery.js";
@@ -3966,6 +3967,12 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
       requestedByActorType: "system",
       requestedByActorId: null,
       contextSnapshot: decision.contextSnapshot,
+    }).catch((error: unknown) => {
+      // Startup, periodic, and terminal-run recovery can race after their
+      // read-side probes. The partial unique index is the durable claim; the
+      // loser treats that conflict as an already-queued generation.
+      if (isFeedbackDeliveryIdempotencyConflict(error)) return null;
+      throw error;
     });
     if (!queued) {
       return { outcome: "skipped", reason: "replay_wake_not_queued" };