From 93bff79e156d8056bbfe845fa7d99a54858cbb78 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 14 Aug 2026 19:51:16 +0000 Subject: [PATCH] fix(feedback): resolve recovery after successful replay Co-Authored-By: Paperclip --- .../heartbeat-process-recovery.test.ts | 51 +++++++++++ server/src/services/heartbeat.ts | 87 ++++++++++++++++++- .../recovery/feedback-delivery.test.ts | 54 ++++++++++++ .../services/recovery/feedback-delivery.ts | 41 +++++++++ server/src/services/recovery/index.ts | 1 + 5 files changed, 233 insertions(+), 1 deletion(-) diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 75eea5673b7..c7e1bbeb4bb 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -1705,6 +1705,31 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { }); if (!retryRun || !releaseAdapter) throw new Error("feedback retry did not reach the adapter"); + const recoveryActionId = randomUUID(); + await db.insert(issueRecoveryActions).values({ + id: recoveryActionId, + companyId: fixture.companyId, + sourceIssueId: fixture.issueId, + kind: "feedback_delivery", + status: "active", + ownerType: "agent", + ownerAgentId: fixture.agentId, + previousOwnerAgentId: fixture.agentId, + returnOwnerAgentId: fixture.agentId, + cause: "feedback_delivery_exhausted", + fingerprint: buildFeedbackDeliveryFingerprint({ + companyId: fixture.companyId, + issueId: fixture.issueId, + agentId: fixture.agentId, + rootWakeupRequestId: fixture.wakeupRequestId, + }), + evidence: { + rootWakeupRequestId: fixture.wakeupRequestId, + outstandingCommentIds: [commentId], + }, + nextAction: "Retry feedback delivery manually.", + wakePolicy: { type: "manual_repair_required" }, + }); await db.insert(issueComments).values({ companyId: fixture.companyId, issueId: fixture.issueId, @@ -1723,6 +1748,32 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { .from(heartbeatRuns) .where(eq(heartbeatRuns.agentId, fixture.agentId)); expect(finalRuns).toHaveLength(2); + const resolvedAction = await db + .select() + .from(issueRecoveryActions) + .where(eq(issueRecoveryActions.id, recoveryActionId)) + .then((rows) => rows[0] ?? null); + expect(resolvedAction).toMatchObject({ + status: "resolved", + outcome: "restored", + }); + expect(resolvedAction?.resolvedAt).toBeTruthy(); + const resolutionActivity = await db + .select() + .from(activityLog) + .where(and( + eq(activityLog.entityId, fixture.issueId), + eq(activityLog.runId, retryRun.id), + eq(activityLog.action, "issue.recovery_action_resolved"), + )); + expect(resolutionActivity).toHaveLength(1); + expect(resolutionActivity[0]?.details).toMatchObject({ + source: "successful_feedback_delivery", + recoveryActionId, + rootWakeupRequestId: fixture.wakeupRequestId, + sourceRunId: retryRun.id, + outcome: "restored", + }); }); it("creates one source-scoped recovery action when the feedback replay is exhausted", async () => { diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index de81ea13203..da3c846fa2a 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -127,7 +127,13 @@ import { evaluateIssueRewakeThrottle, isThrottleCandidateIssueRewake, } from "./issue-rewake-throttle.js"; -import { logActivity, publishPluginDomainEvent, type LogActivityInput } from "./activity-log.js"; +import { + logActivity, + publishActivity, + publishPluginDomainEvent, + type ActivityPublication, + type LogActivityInput, +} from "./activity-log.js"; import { buildWorkspaceReadyComment, buildWorkspaceReadyMetadata, @@ -233,6 +239,7 @@ import { buildFeedbackDeliveryRetryIdempotencyKey, carriesExplicitFeedbackResume, isEligibleFeedbackDeliveryWake, + isSuccessfulFeedbackDeliveryRecoveryMatch, readFeedbackDeliveryRetryGeneration, } from "./recovery/index.js"; import { isAutomaticRecoverySuppressedByPauseHold } from "./recovery/pause-hold-guard.js"; @@ -10674,6 +10681,84 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) const delivery = await readFeedbackDeliveryContext(input.run); if (!delivery) return { kind: "not_applicable" as const }; + if (input.run.status === "succeeded") { + const activeAction = await recoveryActions.getActiveForIssue( + input.run.companyId, + delivery.issueId, + ); + if ( + activeAction + && isSuccessfulFeedbackDeliveryRecoveryMatch({ run: input.run, action: activeAction }) + ) { + const postCommitActivityPublications: ActivityPublication[] = []; + const resolvedAction = await db.transaction(async (tx) => { + const resolved = await recoveryActions.resolveActiveForIssue( + { + companyId: input.run.companyId, + sourceIssueId: delivery.issueId, + actionId: activeAction.id, + kind: FEEDBACK_DELIVERY_RECOVERY_ACTION_KIND, + cause: FEEDBACK_DELIVERY_RECOVERY_CAUSE, + status: "resolved", + outcome: "restored", + resolutionNote: "Automatically resolved after the matching feedback delivery run succeeded.", + }, + tx, + ); + if (!resolved) return null; + + await logActivity( + tx as unknown as Db, + { + companyId: input.run.companyId, + actorType: "system", + actorId: "heartbeat", + agentId: input.run.agentId, + runId: input.run.id, + action: "issue.recovery_action_resolved", + entityType: "issue", + entityId: delivery.issueId, + details: { + source: "successful_feedback_delivery", + recoveryActionId: resolved.id, + recoveryActionStatus: resolved.status, + outcome: resolved.outcome, + sourceIssueStatus: null, + resolutionNote: resolved.resolutionNote, + sourceRunId: input.run.id, + rootWakeupRequestId: delivery.rootWakeupRequestId, + }, + }, + postCommitActivityPublications, + ); + return resolved; + }); + for (const publication of postCommitActivityPublications) publishActivity(publication); + + if (resolvedAction) { + await appendRunEvent(input.run, await nextRunEventSeq(input.run.id), { + eventType: "lifecycle", + stream: "system", + level: "info", + message: "Successful feedback delivery resolved its recovery action", + payload: { + recoveryActionId: resolvedAction.id, + rootWakeupRequestId: delivery.rootWakeupRequestId, + }, + }); + } + } + + // A successful replay is the terminal delivery signal. It must not be + // followed by another exhaustion action merely because the agent chose a + // non-comment disposition (for example, a document revision). + return { + kind: "handled" as const, + delivery, + handlingEvidence: { kind: "successful_run" as const, id: input.run.id }, + }; + } + const handlingEvidence = await findFeedbackHandlingEvidence(input.run, delivery.issueId); if (handlingEvidence) { return { kind: "handled" as const, delivery, handlingEvidence }; diff --git a/server/src/services/recovery/feedback-delivery.test.ts b/server/src/services/recovery/feedback-delivery.test.ts index b3e65e5f7e0..7466bf83f50 100644 --- a/server/src/services/recovery/feedback-delivery.test.ts +++ b/server/src/services/recovery/feedback-delivery.test.ts @@ -6,6 +6,7 @@ import { buildFeedbackDeliveryFingerprint, buildFeedbackDeliveryRetryIdempotencyKey, decideStrandedFeedbackDeliveryBackstop, + isSuccessfulFeedbackDeliveryRecoveryMatch, readFeedbackDeliveryRunContext, type StrandedFeedbackDeliveryBackstopProbe, } from "./feedback-delivery.js"; @@ -89,6 +90,59 @@ const FINGERPRINT = buildFeedbackDeliveryFingerprint({ rootWakeupRequestId: ROOT_WAKE_ID, }); +describe("successful feedback delivery recovery matching", () => { + const matchingRun = { + companyId: COMPANY_ID, + status: "succeeded", + contextSnapshot: { + issueId: ISSUE_ID, + feedbackDeliveryRootWakeupRequestId: ROOT_WAKE_ID, + }, + }; + const matchingAction = { + companyId: COMPANY_ID, + sourceIssueId: ISSUE_ID, + kind: "feedback_delivery", + status: "active", + cause: "feedback_delivery_exhausted", + evidence: { rootWakeupRequestId: ROOT_WAKE_ID }, + }; + + it("matches only a successful run for the same company, issue, and root wake", () => { + expect(isSuccessfulFeedbackDeliveryRecoveryMatch({ + run: matchingRun, + action: matchingAction, + })).toBe(true); + }); + + it.each([ + ["non-successful run", { run: { ...matchingRun, status: "failed" } }], + ["different company", { run: { ...matchingRun, companyId: "company-2" } }], + ["different issue", { + run: { + ...matchingRun, + contextSnapshot: { ...matchingRun.contextSnapshot, issueId: "issue-2" }, + }, + }], + ["different root wake", { + run: { + ...matchingRun, + contextSnapshot: { + ...matchingRun.contextSnapshot, + feedbackDeliveryRootWakeupRequestId: "wake-root-2", + }, + }, + }], + ["non-feedback action", { action: { ...matchingAction, kind: "process_lost" } }], + ["resolved action", { action: { ...matchingAction, status: "resolved" } }], + ])("rejects a %s", (_label, overrides) => { + expect(isSuccessfulFeedbackDeliveryRecoveryMatch({ + run: "run" in overrides ? overrides.run : matchingRun, + action: "action" in overrides ? overrides.action : matchingAction, + })).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 cbb75aefa46..fe23bdc036e 100644 --- a/server/src/services/recovery/feedback-delivery.ts +++ b/server/src/services/recovery/feedback-delivery.ts @@ -77,6 +77,47 @@ export function buildFeedbackDeliveryManualRetryIdempotencyKey(input: { return `${input.fingerprint}:manual:${input.attempt}`; } +export function isSuccessfulFeedbackDeliveryRecoveryMatch(input: { + run: { + companyId: string; + status: string; + contextSnapshot: unknown; + }; + action: { + companyId: string; + sourceIssueId: string; + kind: string; + status: string; + cause: string; + evidence: unknown; + }; +}) { + if (input.run.status !== "succeeded") return false; + if ( + input.action.kind !== FEEDBACK_DELIVERY_RECOVERY_ACTION_KIND + || input.action.cause !== FEEDBACK_DELIVERY_RECOVERY_CAUSE + || (input.action.status !== "active" && input.action.status !== "escalated") + || input.action.companyId !== input.run.companyId + ) { + return false; + } + + const context = typeof input.run.contextSnapshot === "object" && input.run.contextSnapshot !== null + ? input.run.contextSnapshot as Record + : {}; + const evidence = typeof input.action.evidence === "object" && input.action.evidence !== null + ? input.action.evidence as Record + : {}; + const issueId = readNonEmptyString(context.issueId) ?? readNonEmptyString(context.taskId); + const rootWakeupRequestId = readNonEmptyString(context.feedbackDeliveryRootWakeupRequestId); + return Boolean( + issueId + && rootWakeupRequestId + && input.action.sourceIssueId === issueId + && readNonEmptyString(evidence.rootWakeupRequestId) === rootWakeupRequestId + ); +} + export function readFeedbackDeliveryWakeCommentIds( contextSnapshot: Record | null | undefined, ): string[] { diff --git a/server/src/services/recovery/index.ts b/server/src/services/recovery/index.ts index b989e3aba85..7e9a0173758 100644 --- a/server/src/services/recovery/index.ts +++ b/server/src/services/recovery/index.ts @@ -42,6 +42,7 @@ export { carriesExplicitFeedbackResume, decideStrandedFeedbackDeliveryBackstop, isEligibleFeedbackDeliveryWake, + isSuccessfulFeedbackDeliveryRecoveryMatch, readFeedbackDeliveryRetryGeneration, readFeedbackDeliveryRunContext, readFeedbackDeliveryWakeCommentIds,