diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 663633ceafb..39a02164c3b 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -237,12 +237,14 @@ import { FEEDBACK_DELIVERY_RETRY_WAKE_REASON, buildFeedbackDeliveryFingerprint, buildFeedbackDeliveryRetryIdempotencyKey, + canCoalesceFeedbackDeliveryRetry, carriesExplicitFeedbackResume, collectFeedbackDispositionHandledCommentIds, isEligibleFeedbackDeliveryWake, isSuccessfulFeedbackDeliveryRecoveryMatch, remainingFeedbackDispositionCommentIds, readFeedbackDeliveryRetryGeneration, + readFeedbackDeliveryRunContext, } from "./recovery/index.js"; import { isAutomaticRecoverySuppressedByPauseHold } from "./recovery/pause-hold-guard.js"; import { @@ -10483,6 +10485,9 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) }; const liveContext = parseObject(liveRun?.contextSnapshot); + const liveDelivery = liveRun + ? readFeedbackDeliveryRunContext({ companyId: input.run.companyId, run: liveRun }) + : null; if ( liveRun && readNonEmptyString(liveContext.feedbackDeliveryRootWakeupRequestId) === @@ -10498,8 +10503,13 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) if ( liveRun && - liveRun.agentId === input.run.agentId && - liveRun.status !== "running" + canCoalesceFeedbackDeliveryRetry({ + sourceAgentId: input.run.agentId, + liveAgentId: liveRun.agentId, + liveStatus: liveRun.status, + sourceRootWakeupRequestId: input.delivery.rootWakeupRequestId, + liveRootWakeupRequestId: liveDelivery?.rootWakeupRequestId ?? null, + }) ) { const mergedContext = mergeCoalescedContextSnapshot(retryContext, parseObject(liveRun.contextSnapshot)); const orderedCommentIds = await orderCommentIds(mergeWakeCommentIds(retryContext, liveRun.contextSnapshot)); @@ -10532,6 +10542,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) eq(agentWakeupRequests.agentId, input.run.agentId), eq(agentWakeupRequests.status, "deferred_issue_execution"), sql`${agentWakeupRequests.payload} ->> 'issueId' = ${input.delivery.issueId}`, + sql`${agentWakeupRequests.payload} -> '_paperclipWakeContext' ->> 'feedbackDeliveryRootWakeupRequestId' = ${input.delivery.rootWakeupRequestId}`, )) .orderBy(asc(agentWakeupRequests.requestedAt), asc(agentWakeupRequests.id)) .limit(1) diff --git a/server/src/services/recovery/feedback-delivery.test.ts b/server/src/services/recovery/feedback-delivery.test.ts index e23b11f5b46..c9a05c7d8ac 100644 --- a/server/src/services/recovery/feedback-delivery.test.ts +++ b/server/src/services/recovery/feedback-delivery.test.ts @@ -5,6 +5,7 @@ import { STRANDED_FEEDBACK_DELIVERY_BACKSTOP_SOURCE, buildFeedbackDeliveryFingerprint, buildFeedbackDeliveryRetryIdempotencyKey, + canCoalesceFeedbackDeliveryRetry, collectFeedbackDispositionHandledCommentIds, decideStrandedFeedbackDeliveryBackstop, hasCompleteFeedbackDispositionCoverage, @@ -243,6 +244,27 @@ describe("manual feedback retry attempts", () => { }); }); +describe("feedback retry coalescing", () => { + const base = { + sourceAgentId: "agent-1", + liveAgentId: "agent-1", + liveStatus: "queued", + sourceRootWakeupRequestId: "wake-root-1", + liveRootWakeupRequestId: "wake-root-1" as string | null, + }; + + it("coalesces queued work for the same delivery root", () => { + expect(canCoalesceFeedbackDeliveryRetry(base)).toBe(true); + }); + + it("keeps a different user wake in a separate delivery lane", () => { + expect(canCoalesceFeedbackDeliveryRetry({ + ...base, + liveRootWakeupRequestId: "wake-root-2", + })).toBe(false); + }); +}); + describe("stranded feedback delivery backstop", () => { it("requeues exactly one normal-model recovery for the original assignee", () => { const decision = decide(); diff --git a/server/src/services/recovery/feedback-delivery.ts b/server/src/services/recovery/feedback-delivery.ts index 0a0832f8c0a..ac2ef880884 100644 --- a/server/src/services/recovery/feedback-delivery.ts +++ b/server/src/services/recovery/feedback-delivery.ts @@ -97,6 +97,21 @@ export function nextFeedbackDeliveryManualRetryAttempt(input: { return maxAttempt + 1; } +export function canCoalesceFeedbackDeliveryRetry(input: { + sourceAgentId: string; + liveAgentId: string; + liveStatus: string; + sourceRootWakeupRequestId: string; + liveRootWakeupRequestId: string | null; +}) { + return input.liveAgentId === input.sourceAgentId + && input.liveStatus !== "running" + && ( + input.liveRootWakeupRequestId === null + || input.liveRootWakeupRequestId === input.sourceRootWakeupRequestId + ); +} + export function hasCompleteFeedbackDispositionCoverage(input: { deliveryCommentIds: string[]; handledCommentIds: string[]; diff --git a/server/src/services/recovery/index.ts b/server/src/services/recovery/index.ts index dd65a335a15..36f9b77d713 100644 --- a/server/src/services/recovery/index.ts +++ b/server/src/services/recovery/index.ts @@ -39,6 +39,7 @@ export { STRANDED_FEEDBACK_DELIVERY_BACKSTOP_SOURCE, buildFeedbackDeliveryFingerprint, buildFeedbackDeliveryRetryIdempotencyKey, + canCoalesceFeedbackDeliveryRetry, carriesExplicitFeedbackResume, collectFeedbackDispositionHandledCommentIds, decideStrandedFeedbackDeliveryBackstop,