fix(interactions): keep delivery claims retryable

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

@@ -3,6 +3,7 @@ import { eq } from "drizzle-orm";
 import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
 import {
   activityLog,
+  agentWakeupRequests,
   agents,
   companies,
   createDb,
@@ -41,6 +42,7 @@ describeEmbeddedPostgres("question response delivery", () => {
     await db.delete(issueThreadInteractions);
     await db.delete(activityLog);
     await db.delete(heartbeatRuns);
+    await db.delete(agentWakeupRequests);
     await db.delete(issues);
     await db.delete(goals);
     await db.delete(agents);
@@ -226,7 +228,7 @@ describeEmbeddedPostgres("question response delivery", () => {
     });
   });
 
-  it("never steers into the source run and records the single wake fallback", async () => {
+  it("never steers into the source run and keeps a skipped wake retryable", async () => {
     const seeded = await seed({ sourceStatus: "running" });
     const wakeup = vi.fn().mockResolvedValue(null);
     const steer = vi.fn();
@@ -235,11 +237,75 @@ describeEmbeddedPostgres("question response delivery", () => {
       steer,
     }).deliver(seeded.interaction.id);
 
-    expect(outcome).toMatchObject({ status: "fallback_queued", mode: "wake_fallback", targetRunId: null });
+    expect(outcome).toBeNull();
     expect(steer).not.toHaveBeenCalled();
     expect(wakeup).toHaveBeenCalledTimes(1);
     expect(JSON.stringify(wakeup.mock.calls[0])).not.toContain("Internal API");
     expect(JSON.stringify(wakeup.mock.calls[0])).not.toContain("Node.js");
+    const [delivery] = await db.select().from(issueQuestionResponseDeliveries);
+    expect(delivery).toMatchObject({
+      status: "pending",
+      attemptCount: 1,
+      lastErrorCode: "question_response_wake_skipped",
+    });
+  });
+
+  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({
+      companyId: seeded.companyId,
+      agentId: seeded.agentId,
+      source: "automation",
+      triggerDetail: "system",
+      reason: "issue_commented",
+      status: "queued",
+      idempotencyKey: `interaction:${seeded.interaction.id}:answered`,
+    }).returning();
+    const [wakeRun] = await db.insert(heartbeatRuns).values({
+      companyId: seeded.companyId,
+      agentId: seeded.agentId,
+      invocationSource: "automation",
+      status: "queued",
+      runtimeMode: "legacy",
+      wakeupRequestId: wakeRequest!.id,
+      contextSnapshot: { issueId: seeded.issueId },
+    }).returning();
+    await db.update(agentWakeupRequests).set({ runId: wakeRun!.id })
+      .where(eq(agentWakeupRequests.id, wakeRequest!.id));
+    const wakeup = vi.fn();
+
+    const outcome = await questionResponseDeliveryService(db, {
+      heartbeat: { wakeup } as never,
+      steer: vi.fn(),
+    }).deliver(seeded.interaction.id);
+
+    expect(outcome).toMatchObject({
+      status: "delivered",
+      mode: "coalesced",
+      targetRunId: wakeRun!.id,
+    });
+    expect(wakeup).not.toHaveBeenCalled();
+  });
+
+  it("keeps a long wake claim leased while the side effect is active", async () => {
+    const seeded = await seed({ sourceStatus: "running" });
+    let releaseWake!: (value: null) => void;
+    const wakeup = vi.fn(() => new Promise<null>((resolve) => {
+      releaseWake = resolve;
+    }));
+    const service = questionResponseDeliveryService(db, {
+      heartbeat: { wakeup } as never,
+      steer: vi.fn(),
+      claimStaleMs: 40,
+      claimRefreshMs: 5,
+    });
+
+    const deliveryPromise = service.deliver(seeded.interaction.id);
+    await vi.waitFor(() => expect(wakeup).toHaveBeenCalledTimes(1));
+    await new Promise((resolve) => setTimeout(resolve, 70));
+    await expect(service.sweepPending()).resolves.toMatchObject({ scanned: 0 });
+    releaseWake(null);
+    await expect(deliveryPromise).resolves.toBeNull();
   });
 
   it("falls back once when successor steering is unsupported", async () => {

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

@@ -1,6 +1,7 @@
-import { and, asc, eq, inArray, isNull, lte, or, sql } from "drizzle-orm";
+import { and, asc, desc, eq, inArray, isNull, lte, or, sql } from "drizzle-orm";
 import type { Db } from "@paperclipai/db";
 import {
+  agentWakeupRequests,
   agents,
   heartbeatRuns,
   issueQuestionResponseDeliveries,
@@ -25,8 +26,18 @@ import {
 } from "./native-runtime/native-session-executor.js";
 
 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 DURABLE_WAKE_REQUEST_STATUSES = [
+  "queued",
+  "running",
+  "succeeded",
+  "coalesced",
+  "deferred_issue_execution",
+  "retrying",
+  "scheduled_retry",
+] as const;
 
 type QuestionInteractionRow = typeof issueThreadInteractions.$inferSelect;
 type DeliveryRow = typeof issueQuestionResponseDeliveries.$inferSelect;
@@ -53,6 +64,9 @@ export interface QuestionResponseDeliveryServiceOptions {
   heartbeat: Heartbeat;
   steer?: typeof steerNativeSession;
   now?: () => Date;
+  /** Test-only lease timings. Production callers use the bounded defaults. */
+  claimStaleMs?: number;
+  claimRefreshMs?: number;
 }
 
 function record(value: unknown): Record<string, unknown> {
@@ -221,6 +235,11 @@ export function questionResponseDeliveryService(
 ) {
   const steer = options.steer ?? steerNativeSession;
   const now = options.now ?? (() => new Date());
+  const claimStaleMs = Math.max(2, options.claimStaleMs ?? DELIVERY_CLAIM_STALE_MS);
+  const claimRefreshMs = Math.max(
+    1,
+    Math.min(options.claimRefreshMs ?? DELIVERY_CLAIM_REFRESH_MS, Math.floor(claimStaleMs / 2)),
+  );
 
   async function claim(interactionId: string): Promise<DeliveryRow | null> {
     const claimAt = now();
@@ -235,7 +254,7 @@ export function questionResponseDeliveryService(
       if (
         current.status === "delivering"
         && current.lastAttemptAt
-        && current.lastAttemptAt.getTime() > claimAt.getTime() - DELIVERY_CLAIM_STALE_MS
+        && current.lastAttemptAt.getTime() > claimAt.getTime() - claimStaleMs
       ) return null;
       return tx.update(issueQuestionResponseDeliveries).set({
         status: "delivering",
@@ -354,6 +373,60 @@ export function questionResponseDeliveryService(
     return exhausted;
   }
 
+  async function withClaimLease<T>(delivery: DeliveryRow, operation: () => Promise<T>): Promise<T> {
+    let stopped = false;
+    let renewal = Promise.resolve();
+    const timer = setInterval(() => {
+      renewal = renewal.then(async () => {
+        if (stopped) return;
+        const renewedAt = now();
+        const renewed = await db.update(issueQuestionResponseDeliveries).set({
+          lastAttemptAt: renewedAt,
+          updatedAt: renewedAt,
+        }).where(and(
+          eq(issueQuestionResponseDeliveries.id, delivery.id),
+          eq(issueQuestionResponseDeliveries.status, "delivering"),
+        )).returning({ id: issueQuestionResponseDeliveries.id });
+        if (renewed.length === 0) stopped = true;
+      }).catch((error) => {
+        logger.warn({ err: error, deliveryId: delivery.id }, "question response claim lease renewal failed");
+      });
+    }, claimRefreshMs);
+    timer.unref?.();
+    try {
+      return await operation();
+    } finally {
+      stopped = true;
+      clearInterval(timer);
+      await renewal;
+    }
+  }
+
+  async function findDurableWakeRequest(input: {
+    companyId: string;
+    agentId: string;
+    idempotencyKey: string;
+  }) {
+    const request = await db.select({
+      id: agentWakeupRequests.id,
+      runId: agentWakeupRequests.runId,
+      status: agentWakeupRequests.status,
+    }).from(agentWakeupRequests).where(and(
+      eq(agentWakeupRequests.companyId, input.companyId),
+      eq(agentWakeupRequests.agentId, input.agentId),
+      eq(agentWakeupRequests.idempotencyKey, input.idempotencyKey),
+      inArray(agentWakeupRequests.status, [...DURABLE_WAKE_REQUEST_STATUSES]),
+    )).orderBy(desc(agentWakeupRequests.createdAt)).limit(1)
+      .then((rows) => rows[0] ?? null);
+    if (!request?.runId) return request ? { request, run: null } : null;
+    const run = await db.select().from(heartbeatRuns).where(and(
+      eq(heartbeatRuns.id, request.runId),
+      eq(heartbeatRuns.companyId, input.companyId),
+      eq(heartbeatRuns.agentId, input.agentId),
+    )).limit(1).then((rows) => rows[0] ?? null);
+    return { request, run };
+  }
+
   async function deliver(interactionId: string): Promise<QuestionResponseDeliveryOutcome | null> {
     const claimed = await claim(interactionId);
     if (!claimed) return terminalOutcome(interactionId);
@@ -407,10 +480,11 @@ export function questionResponseDeliveryService(
         errorCode: !issue ? "question_response_issue_missing" : "question_response_target_unavailable",
       });
     }
+    const assigneeAgentId = issue.assigneeAgentId;
 
     const liveRuns = await db.select().from(heartbeatRuns).where(and(
       eq(heartbeatRuns.companyId, interaction.companyId),
-      eq(heartbeatRuns.agentId, issue.assigneeAgentId),
+      eq(heartbeatRuns.agentId, assigneeAgentId),
       inArray(heartbeatRuns.status, ["queued", "running", "scheduled_retry"]),
     )).orderBy(asc(heartbeatRuns.createdAt));
     const issueRuns = liveRuns.filter((run) => issueIdFromRun(run) === interaction.issueId);
@@ -448,11 +522,11 @@ export function questionResponseDeliveryService(
     let steeringErrorCode: string | null = null;
     if (successorRunning?.runtimeMode === "native") {
       try {
-        const acknowledgement = await steer({
+        const acknowledgement = await withClaimLease(claimed, () => steer({
           runId: successorRunning.id,
           message: formatQuestionResponseSteeringMessage(envelope),
           correlationId: claimed.correlationId,
-        });
+        }));
         return recordTerminal({
           delivery: claimed,
           interaction,
@@ -472,43 +546,72 @@ export function questionResponseDeliveryService(
     }
 
     const actor = actorForInteraction(interaction);
+    const wakeIdempotencyKey = `interaction:${interaction.id}:${interaction.status}`;
     try {
-      const wakeRun = await options.heartbeat.wakeup(issue.assigneeAgentId, {
-        source: "automation",
-        triggerDetail: "system",
-        reason: "issue_commented",
-        payload: {
-          issueId: issue.id,
-          interactionId: interaction.id,
-          interactionKind: interaction.kind,
-          interactionStatus: interaction.status,
-          sourceCommentId: interaction.sourceCommentId,
-          sourceRunId: interaction.sourceRunId,
-          mutation: "interaction",
-        },
-        idempotencyKey: `interaction:${interaction.id}:${interaction.status}`,
-        requestedByActorType: actor.actorType,
-        requestedByActorId: actor.actorId,
-        contextSnapshot: {
-          issueId: issue.id,
-          taskId: issue.id,
-          interactionId: interaction.id,
-          interactionKind: interaction.kind,
-          interactionStatus: interaction.status,
-          sourceCommentId: interaction.sourceCommentId,
-          sourceRunId: interaction.sourceRunId,
-          wakeReason: "issue_commented",
-          source: "issue.interaction.respond",
-        },
+      const existingWake = await findDurableWakeRequest({
+        companyId: interaction.companyId,
+        agentId: assigneeAgentId,
+        idempotencyKey: wakeIdempotencyKey,
       });
-      const coalesced = Boolean(queuedSuccessor && wakeRun?.id === queuedSuccessor.id);
+      const wakeRun = existingWake?.run ?? (existingWake ? null : await withClaimLease(
+        claimed,
+        () => options.heartbeat.wakeup(assigneeAgentId, {
+          source: "automation",
+          triggerDetail: "system",
+          reason: "issue_commented",
+          payload: {
+            issueId: issue.id,
+            interactionId: interaction.id,
+            interactionKind: interaction.kind,
+            interactionStatus: interaction.status,
+            sourceCommentId: interaction.sourceCommentId,
+            sourceRunId: interaction.sourceRunId,
+            mutation: "interaction",
+          },
+          idempotencyKey: wakeIdempotencyKey,
+          requestedByActorType: actor.actorType,
+          requestedByActorId: actor.actorId,
+          contextSnapshot: {
+            issueId: issue.id,
+            taskId: issue.id,
+            interactionId: interaction.id,
+            interactionKind: interaction.kind,
+            interactionStatus: interaction.status,
+            sourceCommentId: interaction.sourceCommentId,
+            sourceRunId: interaction.sourceRunId,
+            wakeReason: "issue_commented",
+            source: "issue.interaction.respond",
+          },
+        }),
+      ));
+      const durableWake = existingWake ?? (wakeRun ? null : await findDurableWakeRequest({
+        companyId: interaction.companyId,
+        agentId: assigneeAgentId,
+        idempotencyKey: wakeIdempotencyKey,
+      }));
+      if (!wakeRun && !durableWake) {
+        const errorCode = "question_response_wake_skipped";
+        const exhausted = await releaseForRetry(claimed, errorCode);
+        if (!exhausted) return null;
+        return recordTerminal({
+          delivery: claimed,
+          interaction,
+          status: "failed",
+          mode: null,
+          targetRunId: null,
+          adapter,
+          errorCode,
+        });
+      }
+      const targetRun = wakeRun ?? 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: wakeRun?.id ?? queuedSuccessor?.id ?? null,
-        adapter: wakeRun?.driverKind ?? adapter,
+        targetRunId: targetRun?.id ?? null,
+        adapter: targetRun?.driverKind ?? adapter,
         errorCode: steeringErrorCode,
       });
     } catch (error) {
@@ -538,7 +641,7 @@ export function questionResponseDeliveryService(
 
   async function sweepPending(limit = 50) {
     const sweepAt = now();
-    const staleAt = new Date(sweepAt.getTime() - DELIVERY_CLAIM_STALE_MS);
+    const staleAt = new Date(sweepAt.getTime() - claimStaleMs);
     await db.update(issueQuestionResponseDeliveries).set({
       status: "pending",
       updatedAt: sweepAt,