From e9c8419499d38d9ae867fed3c19f3111a76e2982 Mon Sep 17 00:00:00 2001 From: Dotta Date: Thu, 27 Aug 2026 09:03:56 -0500 Subject: [PATCH] fix(interactions): keep delivery claims retryable --- .../question-response-delivery.test.ts | 70 ++++++- .../services/question-response-delivery.ts | 175 ++++++++++++++---- 2 files changed, 207 insertions(+), 38 deletions(-) diff --git a/server/src/__tests__/question-response-delivery.test.ts b/server/src/__tests__/question-response-delivery.test.ts index e151d79f01b..8c5cc35b72b 100644 --- a/server/src/__tests__/question-response-delivery.test.ts +++ b/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((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 () => { diff --git a/server/src/services/question-response-delivery.ts b/server/src/services/question-response-delivery.ts index ec8b84ef3a5..fac667cb171 100644 --- a/server/src/services/question-response-delivery.ts +++ b/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 { @@ -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 { 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(delivery: DeliveryRow, operation: () => Promise): Promise { + 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 { 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,