From 2c081e4b47d31674668624ba4d35784ca7f419d3 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 16 Aug 2026 21:12:58 +0000 Subject: [PATCH] fix: preserve partial feedback delivery progress Co-Authored-By: Paperclip --- .../heartbeat-process-recovery.test.ts | 96 +++++++++++++++++++ server/src/routes/issues.ts | 8 +- server/src/services/heartbeat.ts | 87 +++++++++++------ .../recovery/feedback-delivery.test.ts | 18 ++++ .../services/recovery/feedback-delivery.ts | 24 ++++- server/src/services/recovery/index.ts | 1 + 6 files changed, 199 insertions(+), 35 deletions(-) diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index e44826b2f34..94e93534fee 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -1849,6 +1849,102 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { }); }); + it("retries only the unhandled comments after a partial feedback disposition", async () => { + let releaseAdapter: (() => void) | null = null; + const adapterStarted = new Promise((resolve) => { + mockAdapterExecute.mockImplementationOnce(async () => { + resolve(); + await new Promise((release) => { + releaseAdapter = release; + }); + return { + exitCode: 0, + signal: null, + timedOut: false, + errorMessage: null, + summary: "Handled the remaining feedback.", + provider: "test", + model: "test-model", + }; + }); + }); + const fixture = await seedRunFixture({ + agentStatus: "idle", + processPid: null, + processGroupId: null, + }); + const { commentId: handledCommentId } = await prepareFeedbackWake(fixture); + const outstandingCommentId = randomUUID(); + await db.insert(issueComments).values({ + id: outstandingCommentId, + companyId: fixture.companyId, + issueId: fixture.issueId, + authorType: "user", + authorUserId: "responsible-user", + body: "Please also handle this follow-up.", + }); + await db + .update(heartbeatRuns) + .set({ + contextSnapshot: { + issueId: fixture.issueId, + taskId: fixture.issueId, + wakeReason: "issue_commented", + commentId: outstandingCommentId, + wakeCommentId: outstandingCommentId, + wakeCommentIds: [handledCommentId, outstandingCommentId], + }, + }) + .where(eq(heartbeatRuns.id, fixture.runId)); + await db.insert(issueComments).values({ + companyId: fixture.companyId, + issueId: fixture.issueId, + authorType: "agent", + authorAgentId: fixture.agentId, + createdByRunId: fixture.runId, + body: "Handled the first part of the feedback batch.", + metadata: { + version: 1, + feedbackDisposition: { + kind: "feedback_delivery", + rootWakeupRequestId: fixture.wakeupRequestId, + handledCommentIds: [handledCommentId], + }, + sections: [{ + title: "Feedback disposition", + rows: [{ type: "key_value", label: "Handled comment", value: handledCommentId }], + }], + }, + }); + + const heartbeat = heartbeatService(db); + expect(await heartbeat.reapOrphanedRuns()).toEqual({ reaped: 1, runIds: [fixture.runId] }); + await Promise.race([ + adapterStarted, + new Promise((_, reject) => setTimeout(() => reject(new Error("feedback retry did not start")), 3_000)), + ]); + + const retryRun = await db + .select() + .from(heartbeatRuns) + .where(and( + eq(heartbeatRuns.agentId, fixture.agentId), + eq(heartbeatRuns.retryOfRunId, fixture.runId), + )) + .then((rows) => rows[0] ?? null); + expect(retryRun?.contextSnapshot).toMatchObject({ + feedbackDeliveryRootWakeupRequestId: fixture.wakeupRequestId, + feedbackDeliveryRetryGeneration: 1, + wakeCommentIds: [outstandingCommentId], + commentId: outstandingCommentId, + wakeCommentId: outstandingCommentId, + }); + + if (!releaseAdapter) throw new Error("feedback retry did not reach the adapter"); + releaseAdapter(); + if (retryRun) await heartbeat.waitForRunExecutionDrain(retryRun.id); + }); + it("keeps feedback exhausted when a successful replay posts an ordinary run comment", async () => { let releaseAdapter: (() => void) | null = null; const adapterStarted = new Promise((resolve) => { diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index ecf1e570d83..d38da557473 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -113,7 +113,7 @@ import { FEEDBACK_DELIVERY_RECOVERY_CAUSE, FEEDBACK_DELIVERY_RETRY_WAKE_REASON, FEEDBACK_DELIVERY_WAKE_COMMENT_IDS_KEY, - hasCompleteFeedbackDispositionCoverage, + remainingFeedbackDispositionCommentIds, readFeedbackDeliveryRunContext, } from "../services/recovery/feedback-delivery.js"; import { evaluateAgentInvokabilityFromDb } from "../services/agent-invokability.js"; @@ -5020,12 +5020,12 @@ export function issueRoutes( } const handledCommentIds = [...new Set(requested.handledCommentIds)]; - if (!hasCompleteFeedbackDispositionCoverage({ + if (remainingFeedbackDispositionCommentIds({ deliveryCommentIds: delivery.commentIds, handledCommentIds, - })) { + }) === null) { res.status(422).json({ - error: "Feedback disposition must cover every comment in the current run's feedback batch and no others", + error: "Feedback disposition must cover at least one comment in the current run's feedback batch and no others", }); return { ok: false }; } diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 415ef3a8aad..80ac0ca76ac 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -238,9 +238,9 @@ import { buildFeedbackDeliveryFingerprint, buildFeedbackDeliveryRetryIdempotencyKey, carriesExplicitFeedbackResume, - hasCompleteFeedbackDispositionCoverage, isEligibleFeedbackDeliveryWake, isSuccessfulFeedbackDeliveryRecoveryMatch, + remainingFeedbackDispositionCommentIds, readFeedbackDeliveryRetryGeneration, } from "./recovery/index.js"; import { isAutomaticRecoverySuppressedByPauseHold } from "./recovery/pause-hold-guard.js"; @@ -10227,10 +10227,10 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) disposition?.kind !== "feedback_delivery" || disposition.rootWakeupRequestId !== delivery.rootWakeupRequestId ) return false; - return hasCompleteFeedbackDispositionCoverage({ + return remainingFeedbackDispositionCommentIds({ deliveryCommentIds: delivery.commentIds, handledCommentIds: disposition.handledCommentIds, - }); + }) !== null; }) ?? null); if (!comment) return null; const disposition = comment.metadata?.feedbackDisposition; @@ -10313,13 +10313,11 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) } const existing = await recoveryActions.getActiveForIssue(input.run.companyId, issue.id); - if ( + const reusedExisting = Boolean( existing?.kind === FEEDBACK_DELIVERY_RECOVERY_ACTION_KIND && existing.cause === FEEDBACK_DELIVERY_RECOVERY_CAUSE - && existing.fingerprint === input.delivery.fingerprint - ) { - return { kind: "recovery" as const, action: existing, reusedExisting: true }; - } + && existing.fingerprint === input.delivery.fingerprint, + ); const action = await recoveryActions.upsertSourceScoped({ companyId: input.run.companyId, @@ -10374,7 +10372,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) gateErrorCode: input.gateErrorCode ?? null, }, }); - return { kind: "recovery" as const, action, reusedExisting: false }; + return { kind: "recovery" as const, action, reusedExisting }; } async function enqueueFeedbackDeliveryRetry(input: { @@ -10673,8 +10671,22 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) if (!delivery) return { kind: "not_applicable" as const }; const handlingEvidence = await findFeedbackHandlingEvidence(input.run, delivery); + const remainingCommentIds = handlingEvidence + ? remainingFeedbackDispositionCommentIds({ + deliveryCommentIds: delivery.commentIds, + handledCommentIds: handlingEvidence.sourceCommentIds, + }) + : delivery.commentIds; + // Persisted dispositions were validated when the comment was created. Be + // defensive if legacy or directly seeded metadata does not satisfy that + // contract: it must not suppress delivery. + const hasValidHandlingEvidence = Boolean(handlingEvidence && remainingCommentIds !== null); + const fullyHandled = hasValidHandlingEvidence && remainingCommentIds?.length === 0; + const pendingDelivery = hasValidHandlingEvidence && remainingCommentIds + ? { ...delivery, commentIds: remainingCommentIds } + : delivery; - if (input.run.status === "succeeded" && handlingEvidence) { + if (input.run.status === "succeeded" && handlingEvidence && fullyHandled) { const activeAction = await recoveryActions.getActiveForIssue( input.run.companyId, delivery.issueId, @@ -10749,7 +10761,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) return { kind: "handled" as const, delivery, handlingEvidence }; } - if (handlingEvidence) { + if (handlingEvidence && fullyHandled) { return { kind: "handled" as const, delivery, handlingEvidence }; } @@ -10757,17 +10769,22 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) input.run.errorCode === "process_lost" && !input.run.processPid && !input.run.processGroupId; - const automaticReplayExhausted = delivery.generation >= FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES; + const automaticReplayExhausted = pendingDelivery.generation >= FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES; if (automaticReplayExhausted) { const recovery = await ensureFeedbackDeliveryRecovery({ run: input.run, agent: input.agent, - delivery, + delivery: pendingDelivery, }); - return { kind: "recovery" as const, delivery, recovery }; + return { kind: "recovery" as const, delivery: pendingDelivery, recovery }; } const successfulWithoutHandlingEvidence = input.run.status === "succeeded"; - if (!preLaunchProcessLoss && !successfulWithoutHandlingEvidence && !input.existingRetryRun) { + if ( + !preLaunchProcessLoss + && !successfulWithoutHandlingEvidence + && !hasValidHandlingEvidence + && !input.existingRetryRun + ) { return { kind: "not_applicable" as const }; } @@ -10789,24 +10806,24 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) const recovery = await ensureFeedbackDeliveryRecovery({ run: input.run, agent: input.agent, - delivery, + delivery: pendingDelivery, gateReason: gate.reason, gateErrorCode: gate.errorCode, }); - return { kind: "recovery" as const, delivery, recovery }; + return { kind: "recovery" as const, delivery: pendingDelivery, recovery }; } if (input.existingRetryRun) { const retryContext: Record = { ...parseObject(input.existingRetryRun.contextSnapshot), - feedbackDeliveryRootWakeupRequestId: delivery.rootWakeupRequestId, + feedbackDeliveryRootWakeupRequestId: pendingDelivery.rootWakeupRequestId, feedbackDeliverySourceRunId: input.run.id, feedbackDeliveryRetryGeneration: 1, - [WAKE_COMMENT_IDS_KEY]: delivery.commentIds, + [WAKE_COMMENT_IDS_KEY]: pendingDelivery.commentIds, }; - if (delivery.commentIds.length > 0) { - retryContext.commentId = delivery.commentIds.at(-1); - retryContext.wakeCommentId = delivery.commentIds.at(-1); + if (pendingDelivery.commentIds.length > 0) { + retryContext.commentId = pendingDelivery.commentIds.at(-1); + retryContext.wakeCommentId = pendingDelivery.commentIds.at(-1); } const updatedRetry = await db .update(heartbeatRuns) @@ -10816,26 +10833,40 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) .then((rows) => rows[0] ?? input.existingRetryRun!); await logFeedbackRetryQueuedOnce({ run: input.run, - delivery, + delivery: pendingDelivery, retryRunId: updatedRetry.id, retryWakeupRequestId: updatedRetry.wakeupRequestId, disposition: "queued", }); - return { kind: "retry" as const, delivery, run: updatedRetry, disposition: "queued" as const }; + return { + kind: "retry" as const, + delivery: pendingDelivery, + run: updatedRetry, + disposition: "queued" as const, + }; } - const retry = await enqueueFeedbackDeliveryRetry({ run: input.run, agent: input.agent, delivery }); + const retry = await enqueueFeedbackDeliveryRetry({ + run: input.run, + agent: input.agent, + delivery: pendingDelivery, + }); if (retry.kind === "suppressed") { - return { kind: "suppressed" as const, delivery, reason: "state_changed" }; + return { kind: "suppressed" as const, delivery: pendingDelivery, reason: "state_changed" }; } await logFeedbackRetryQueuedOnce({ run: input.run, - delivery, + delivery: pendingDelivery, retryRunId: retry.run.id, retryWakeupRequestId: retry.wakeupRequestId, disposition: retry.kind, }); - return { kind: "retry" as const, delivery, run: retry.run, disposition: retry.kind }; + return { + kind: "retry" as const, + delivery: pendingDelivery, + run: retry.run, + disposition: retry.kind, + }; } async function applyPostRunDisposition(input: { diff --git a/server/src/services/recovery/feedback-delivery.test.ts b/server/src/services/recovery/feedback-delivery.test.ts index 36d9fdaff28..796e1460562 100644 --- a/server/src/services/recovery/feedback-delivery.test.ts +++ b/server/src/services/recovery/feedback-delivery.test.ts @@ -8,6 +8,7 @@ import { decideStrandedFeedbackDeliveryBackstop, hasCompleteFeedbackDispositionCoverage, isSuccessfulFeedbackDeliveryRecoveryMatch, + remainingFeedbackDispositionCommentIds, readFeedbackDeliveryRunContext, type StrandedFeedbackDeliveryBackstopProbe, } from "./feedback-delivery.js"; @@ -184,6 +185,23 @@ describe("feedback disposition coverage", () => { handledCommentIds: ["comment-1", "comment-2"], })).toBe(false); }); + + it("returns only unhandled comments for a valid partial disposition", () => { + expect(remainingFeedbackDispositionCommentIds({ + deliveryCommentIds: ["comment-1", "comment-2", "comment-3"], + handledCommentIds: ["comment-2"], + })).toEqual(["comment-1", "comment-3"]); + }); + + it.each([ + ["an empty receipt", []], + ["a receipt containing another delivery's comment", ["comment-1", "comment-3"]], + ])("rejects %s", (_label, handledCommentIds) => { + expect(remainingFeedbackDispositionCommentIds({ + deliveryCommentIds: ["comment-1", "comment-2"], + handledCommentIds, + })).toBeNull(); + }); }); describe("stranded feedback delivery backstop", () => { diff --git a/server/src/services/recovery/feedback-delivery.ts b/server/src/services/recovery/feedback-delivery.ts index 27a8fdaf1b7..ed6b12d06e3 100644 --- a/server/src/services/recovery/feedback-delivery.ts +++ b/server/src/services/recovery/feedback-delivery.ts @@ -81,15 +81,33 @@ export function hasCompleteFeedbackDispositionCoverage(input: { deliveryCommentIds: string[]; handledCommentIds: string[]; }) { + const remaining = remainingFeedbackDispositionCommentIds(input); + return remaining !== null && remaining.length === 0; +} + +/** + * Valid receipts acknowledge a non-empty subset of the batch and never IDs + * from another delivery. Returning the ordered remainder lets terminalization + * durably advance partially handled batches without replaying completed work. + */ +export function remainingFeedbackDispositionCommentIds(input: { + deliveryCommentIds: string[]; + handledCommentIds: string[]; +}): string[] | null { const deliveryIds = new Set( input.deliveryCommentIds.map(readNonEmptyString).filter((value): value is string => Boolean(value)), ); const handledIds = new Set( input.handledCommentIds.map(readNonEmptyString).filter((value): value is string => Boolean(value)), ); - return deliveryIds.size > 0 - && handledIds.size === deliveryIds.size - && [...deliveryIds].every((commentId) => handledIds.has(commentId)); + if ( + deliveryIds.size === 0 + || handledIds.size === 0 + || [...handledIds].some((commentId) => !deliveryIds.has(commentId)) + ) { + return null; + } + return [...deliveryIds].filter((commentId) => !handledIds.has(commentId)); } export function isSuccessfulFeedbackDeliveryRecoveryMatch(input: { diff --git a/server/src/services/recovery/index.ts b/server/src/services/recovery/index.ts index fe02abc4568..27ea85fcda0 100644 --- a/server/src/services/recovery/index.ts +++ b/server/src/services/recovery/index.ts @@ -44,6 +44,7 @@ export { hasCompleteFeedbackDispositionCoverage, isEligibleFeedbackDeliveryWake, isSuccessfulFeedbackDeliveryRecoveryMatch, + remainingFeedbackDispositionCommentIds, readFeedbackDeliveryRetryGeneration, readFeedbackDeliveryRunContext, readFeedbackDeliveryWakeCommentIds,