fix: preserve partial feedback delivery progress

Co-Authored-By: Paperclip <noreply@paperclip.ing>

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<void>((resolve) => {
+      mockAdapterExecute.mockImplementationOnce(async () => {
+        resolve();
+        await new Promise<void>((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<never>((_, 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<void>((resolve) => {

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 };
     }

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<string, unknown> = {
         ...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: {

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", () => {

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: {

server/src/services/recovery/index.ts

@@ -44,6 +44,7 @@ export {
   hasCompleteFeedbackDispositionCoverage,
   isEligibleFeedbackDeliveryWake,
   isSuccessfulFeedbackDeliveryRecoveryMatch,
+  remainingFeedbackDispositionCommentIds,
   readFeedbackDeliveryRetryGeneration,
   readFeedbackDeliveryRunContext,
   readFeedbackDeliveryWakeCommentIds,