fix(interactions): fence duplicate continuation wakes

packages/db/src/migrations/0228_violet_scorpion.sql

@@ -140,6 +140,8 @@ CREATE UNIQUE INDEX IF NOT EXISTS "issue_question_response_deliveries_interactio
 CREATE UNIQUE INDEX IF NOT EXISTS "issue_question_response_deliveries_correlation_uq" ON "issue_question_response_deliveries" USING btree ("correlation_id");--> statement-breakpoint
 CREATE INDEX IF NOT EXISTS "issue_question_response_deliveries_pending_idx" ON "issue_question_response_deliveries" USING btree ("status","created_at");--> statement-breakpoint
 CREATE INDEX IF NOT EXISTS "issue_question_response_deliveries_company_issue_idx" ON "issue_question_response_deliveries" USING btree ("company_id","issue_id","created_at");--> statement-breakpoint
+-- paperclip:migration-safety-ignore large-create-index-not-concurrently: This partial index covers a new question-response key prefix, so existing deployments have no matching rows; Drizzle applies migrations transactionally and cannot use CONCURRENTLY.
+CREATE UNIQUE INDEX IF NOT EXISTS "agent_wakeup_requests_question_response_delivery_idempotency_uq" ON "agent_wakeup_requests" USING btree ("company_id","idempotency_key") WHERE "agent_wakeup_requests"."idempotency_key" LIKE 'question-response:%' AND "agent_wakeup_requests"."status" NOT IN ('skipped', 'failed', 'cancelled');--> statement-breakpoint
 CREATE INDEX IF NOT EXISTS "managed_agent_profiles_company_idx" ON "managed_agent_profiles" USING btree ("company_id");--> statement-breakpoint
 CREATE UNIQUE INDEX IF NOT EXISTS "managed_agent_profiles_company_key_uq" ON "managed_agent_profiles" USING btree ("company_id","profile_key");--> statement-breakpoint
 CREATE UNIQUE INDEX IF NOT EXISTS "managed_agent_profiles_company_resource_uq" ON "managed_agent_profiles" USING btree ("company_id","anthropic_agent_id","agent_version","environment_id");--> statement-breakpoint

packages/db/src/migrations/meta/0228_snapshot.json

@@ -1605,6 +1605,28 @@
           "method": "btree",
           "with": {}
         },
+        "agent_wakeup_requests_question_response_delivery_idempotency_uq": {
+          "name": "agent_wakeup_requests_question_response_delivery_idempotency_uq",
+          "columns": [
+            {
+              "expression": "company_id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            },
+            {
+              "expression": "idempotency_key",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            }
+          ],
+          "isUnique": true,
+          "where": "\"agent_wakeup_requests\".\"idempotency_key\" LIKE 'question-response:%' AND \"agent_wakeup_requests\".\"status\" NOT IN ('skipped', 'failed', 'cancelled')",
+          "concurrently": false,
+          "method": "btree",
+          "with": {}
+        },
         "agent_wakeup_requests_company_payload_issue_idx": {
           "name": "agent_wakeup_requests_company_payload_issue_idx",
           "columns": [

packages/db/src/schema/agent_wakeup_requests.ts

@@ -43,6 +43,11 @@ export const agentWakeupRequests = pgTable(
     dispositionRepairIdempotencyUq: uniqueIndex("agent_wakeup_requests_disposition_repair_idempotency_uq")
       .on(table.companyId, table.idempotencyKey)
       .where(sql`${table.idempotencyKey} LIKE 'issue_disposition_repair:%' AND ${table.status} <> 'skipped'`),
+    questionResponseDeliveryIdempotencyUq: uniqueIndex(
+      "agent_wakeup_requests_question_response_delivery_idempotency_uq",
+    )
+      .on(table.companyId, table.idempotencyKey)
+      .where(sql`${table.idempotencyKey} LIKE 'question-response:%' AND ${table.status} NOT IN ('skipped', 'failed', 'cancelled')`),
     companyPayloadIssueIdx: index("agent_wakeup_requests_company_payload_issue_idx").on(
       table.companyId,
       sql`(${table.payload} ->> 'issueId')`,

server/src/__tests__/question-response-delivery.test.ts

@@ -244,7 +244,7 @@ describeEmbeddedPostgres("question response delivery", () => {
     expect(steer).not.toHaveBeenCalled();
     expect(wakeup).toHaveBeenCalledTimes(1);
     expect(wakeup.mock.calls[0]?.[1]).toMatchObject({
-      idempotencyKey: `interaction:${seeded.interaction.id}:answered`,
+      idempotencyKey: `question-response:${seeded.interaction.id}`,
       contextSnapshot: {
         interactionId: seeded.interaction.id,
         interactionStatus: "answered",
@@ -365,6 +365,95 @@ describeEmbeddedPostgres("question response delivery", () => {
     });
   });
 
+  it("enforces one durable wake per question-response idempotency key", async () => {
+    const seeded = await seed({ sourceStatus: "running" });
+    const idempotencyKey = `question-response:${seeded.interaction.id}`;
+    const request = {
+      companyId: seeded.companyId,
+      agentId: seeded.agentId,
+      source: "automation",
+      triggerDetail: "system",
+      reason: "issue_commented",
+      idempotencyKey,
+    } as const;
+
+    await db.insert(agentWakeupRequests).values({ ...request, status: "queued" });
+    await expect(db.insert(agentWakeupRequests).values({
+      ...request,
+      status: "coalesced",
+    })).rejects.toMatchObject({ cause: { code: "23505" } });
+
+    // Suppression receipts are intentionally outside the fence so the outbox
+    // can retry after scheduling is enabled again.
+    await expect(db.insert(agentWakeupRequests).values({
+      ...request,
+      status: "skipped",
+      finishedAt: new Date(),
+    })).resolves.toBeDefined();
+  });
+
+  it("reuses the winning wake when a concurrent insert hits the idempotency fence", async () => {
+    const seeded = await seed({ sourceStatus: "running" });
+    const fallbackRunId = randomUUID();
+    const wakeup = vi.fn().mockImplementation(async (
+      _agentId: string,
+      options: { idempotencyKey?: string | null },
+    ) => {
+      const request = {
+        companyId: seeded.companyId,
+        agentId: seeded.agentId,
+        source: "automation",
+        triggerDetail: "system",
+        reason: "issue_commented",
+        idempotencyKey: options.idempotencyKey,
+      } as const;
+      const [winner] = await db.insert(agentWakeupRequests).values({
+        ...request,
+        status: "queued",
+      }).returning();
+      await db.insert(heartbeatRuns).values({
+        id: fallbackRunId,
+        companyId: seeded.companyId,
+        agentId: seeded.agentId,
+        invocationSource: "automation",
+        status: "queued",
+        runtimeMode: "legacy",
+        driverKind: "codex",
+        wakeupRequestId: winner!.id,
+        contextSnapshot: { issueId: seeded.issueId },
+      });
+      await db.update(agentWakeupRequests).set({ runId: fallbackRunId })
+        .where(eq(agentWakeupRequests.id, winner!.id));
+
+      // Model the losing claimant reaching the same transactional insert after
+      // the winner commits. The service must recover the winner's receipt.
+      await db.insert(agentWakeupRequests).values({
+        ...request,
+        status: "coalesced",
+      });
+      throw new Error("unreachable");
+    });
+
+    const outcome = await questionResponseDeliveryService(db, {
+      heartbeat: { wakeup } as never,
+      steer: vi.fn(),
+    }).deliver(seeded.interaction.id);
+
+    expect(outcome).toMatchObject({
+      status: "fallback_queued",
+      mode: "wake_fallback",
+      targetRunId: fallbackRunId,
+    });
+    expect(wakeup).toHaveBeenCalledTimes(1);
+    const [delivery] = await db.select().from(issueQuestionResponseDeliveries);
+    expect(delivery).toMatchObject({
+      status: "fallback_queued",
+      attemptCount: 1,
+      errorCount: 0,
+      targetRunId: fallbackRunId,
+    });
+  });
+
   it("reuses a durable wake receipt instead of issuing a duplicate continuation", async () => {
     const seeded = await seed({ sourceStatus: "running" });
     const [wakeRequest] = await db.insert(agentWakeupRequests).values({
@@ -374,7 +463,7 @@ describeEmbeddedPostgres("question response delivery", () => {
       triggerDetail: "system",
       reason: "issue_commented",
       status: "queued",
-      idempotencyKey: `interaction:${seeded.interaction.id}:answered`,
+      idempotencyKey: `question-response:${seeded.interaction.id}`,
     }).returning();
     const [wakeRun] = await db.insert(heartbeatRuns).values({
       companyId: seeded.companyId,
@@ -518,7 +607,7 @@ describeEmbeddedPostgres("question response delivery", () => {
       expect(steer).not.toHaveBeenCalled();
       expect(wakeup).toHaveBeenCalledTimes(1);
       expect(wakeup.mock.calls[0]?.[1]).toMatchObject({
-        idempotencyKey: `interaction:${seeded.interaction.id}:answered`,
+        idempotencyKey: `question-response:${seeded.interaction.id}`,
         contextSnapshot: {
           issueId: seeded.issueId,
           interactionId: seeded.interaction.id,

server/src/services/question-response-delivery.ts

@@ -15,6 +15,7 @@ import type {
 import type {
   PaperclipQuestionResponse,
 } from "../vendor/paperclip-runner/index.js";
+import { isUniqueViolation } from "../db-errors.js";
 import { getTelemetryClient } from "../telemetry.js";
 import { logger } from "../middleware/logger.js";
 import { logActivity } from "./activity-log.js";
@@ -29,6 +30,8 @@ const DELIVERY_CLAIM_STALE_MS = 30_000;
 const DELIVERY_CLAIM_REFRESH_MS = 10_000;
 const MAX_DELIVERY_ATTEMPTS = 5;
 const DELIVERY_CORRELATION_PREFIX = "question-response:";
+const QUESTION_RESPONSE_WAKE_IDEMPOTENCY_CONSTRAINT =
+  "agent_wakeup_requests_question_response_delivery_idempotency_uq";
 
 class DeliveryClaimUnavailableError extends Error {
   constructor() {
@@ -38,6 +41,7 @@ class DeliveryClaimUnavailableError extends Error {
 }
 const DURABLE_WAKE_REQUEST_STATUSES = [
   "queued",
+  "claimed",
   "running",
   "succeeded",
   "coalesced",
@@ -590,7 +594,10 @@ export function questionResponseDeliveryService(
     }
 
     const actor = actorForInteraction(interaction);
-    const wakeIdempotencyKey = `interaction:${interaction.id}:${interaction.status}`;
+    // This is a new, migration-fenced namespace. The partial unique index on
+    // agent_wakeup_requests makes the wake transaction itself idempotent, so a
+    // reclaimed stale worker cannot create a second continuation run.
+    const wakeIdempotencyKey = `question-response:${interaction.id}`;
     try {
       const existingWake = await findDurableWakeRequest({
         companyId: interaction.companyId,
@@ -654,6 +661,29 @@ export function questionResponseDeliveryService(
       });
     } catch (error) {
       if (error instanceof DeliveryClaimUnavailableError) return terminalOutcome(interactionId);
+      if (isUniqueViolation(error, QUESTION_RESPONSE_WAKE_IDEMPOTENCY_CONSTRAINT)) {
+        // A concurrent claimant won the transactional wake fence after our
+        // preflight lookup. Reuse its committed receipt instead of consuming
+        // an error retry or issuing another continuation.
+        const durableWake = await findDurableWakeRequest({
+          companyId: interaction.companyId,
+          agentId: assigneeAgentId,
+          idempotencyKey: wakeIdempotencyKey,
+        });
+        if (durableWake) {
+          const targetRun = durableWake.run ?? queuedSuccessor ?? null;
+          const coalesced = Boolean(queuedSuccessor && targetRun?.id === queuedSuccessor.id);
+          return recordTerminal({
+            delivery: claimed,
+            interaction,
+            status: coalesced ? "delivered" : "fallback_queued",
+            mode: coalesced ? "coalesced" : "wake_fallback",
+            targetRunId: targetRun?.id ?? null,
+            adapter: targetRun?.driverKind ?? adapter,
+            errorCode: steeringErrorCode,
+          });
+        }
+      }
       const errorCode = error instanceof Error && compactLine(error.message)
         ? compactLine(error.message)!.slice(0, 160)
         : "question_response_wake_failed";